#![doc = include_str!("../Documentation.md")]
use axum::body::{Body, to_bytes};
use axum::extract::Request;
use axum::http::header::{CACHE_CONTROL, CONTENT_LENGTH, CONTENT_TYPE, HOST};
use axum::http::{HeaderValue, StatusCode};
use axum::middleware::{self, Next};
use axum::response::Response;
use axum::routing::get;
use axum::{Json, Router};
use kcode_k1_access_profiles::K1AccessProfiles;
use kcode_k1_accounts::K1Accounts;
use kcode_k1_daemon_files::DaemonFiles;
use kcode_k1_groups::K1Groups;
use kcode_k1_http::{Config, K1Http};
use kcode_k1_http_accounts::K1HttpAccounts;
use kcode_k1_http_people::K1HttpPeople;
use kcode_k1_http_replay::{ReplayConfig, ReplayWindow};
use kcode_k1_invites::K1Invites;
use kcode_k1_peering::K1Peering;
use kcode_k1_persons::K1Persons;
use kcode_k1_txn_ordering::K1TxnOrdering;
use kcode_k1_users::K1Users;
use serde::Serialize;
use serde_json::Value;
use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::process::ExitCode;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::net::TcpListener;
use tokio::signal::unix::{Signal, SignalKind, signal};
const LISTEN_ADDRESS: &str = "127.0.0.1:4450";
const PUBLIC_ORIGIN: &str = "http://localhost:4450";
const INVITE_LINK_URL: &str = "http://localhost:4321/lib/kcode-k1-ui/*/account.html";
const AUTHORITY: &str = "localhost:4450";
const STARTUP_BOUND: Duration = Duration::from_millis(100);
const API_OPERATION: &str = "serve API request";
#[derive(Clone, Serialize)]
struct PublicConfig {
protocol: &'static str,
server_id: String,
public_origin: &'static str,
}
#[derive(Serialize)]
struct Ready {
event: &'static str,
public_origin: &'static str,
unused_invites: usize,
}
struct Prepared {
app: Router,
listener: TcpListener,
signals: Signals,
unused_invites: usize,
}
struct Signals {
interrupt: Signal,
terminate: Signal,
}
pub fn run(k1_root: PathBuf) -> ExitCode {
let runtime = match tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
{
Ok(runtime) => runtime,
Err(_) => {
eprintln!("kcode-k1-daemon: startup failed");
return ExitCode::from(1);
}
};
runtime.block_on(run_async(k1_root))
}
async fn run_async(k1_root: PathBuf) -> ExitCode {
let started = Instant::now();
let prepared = match startup(k1_root).await {
Ok(prepared) => prepared,
Err(()) => {
warn_if_slow(started.elapsed(), "error");
eprintln!("kcode-k1-daemon: startup failed");
return ExitCode::from(1);
}
};
let elapsed = started.elapsed();
if write_readiness(prepared.unused_invites).is_err() {
warn_if_slow(elapsed, "error");
eprintln!("kcode-k1-daemon: startup failed");
return ExitCode::from(1);
}
warn_if_slow(elapsed, "ready");
let Prepared {
app,
listener,
signals,
..
} = prepared;
match axum::serve(listener, app)
.with_graceful_shutdown(signals.wait())
.await
{
Ok(()) => ExitCode::SUCCESS,
Err(_) => {
eprintln!("kcode-k1-daemon: listener failed");
ExitCode::from(1)
}
}
}
async fn startup(k1_root: PathBuf) -> Result<Prepared, ()> {
let state_root = state_root(&k1_root);
let files = DaemonFiles::open(&state_root).map_err(|_| ())?;
let ordering = Arc::new(K1TxnOrdering::open(&state_root.join("ordering")).map_err(|_| ())?);
let peering = Arc::new(
K1Peering::open(&state_root.join("peering"), Arc::clone(&ordering)).map_err(|_| ())?,
);
let persons = Arc::new(
K1Persons::open(
&state_root.join("persons"),
Arc::clone(&ordering),
Arc::clone(&peering),
)
.map_err(|_| ())?,
);
let invites = Arc::new(
K1Invites::open(
&state_root.join("invites"),
Arc::clone(&ordering),
Arc::clone(&peering),
)
.map_err(|_| ())?,
);
let accounts = Arc::new(K1Accounts::open(Arc::clone(&invites)).map_err(|_| ())?);
let users = Arc::new(K1Users::new(Arc::clone(&accounts), Arc::clone(&persons)));
let groups = Arc::new(
K1Groups::open(
&state_root.join("groups"),
Arc::clone(&ordering),
Arc::clone(&peering),
)
.map_err(|_| ())?,
);
let profiles = Arc::new(
K1AccessProfiles::open(
&state_root.join("access-profiles"),
Arc::clone(&ordering),
Arc::clone(&peering),
)
.map_err(|_| ())?,
);
let replay = ReplayWindow::open(ReplayConfig {
epoch_file: files.replay_epoch_path().to_owned(),
max_nonces_per_epoch: usize::MAX,
})
.await
.map_err(|_| ())?;
let unused_invites = kcode_k1_daemon_invite_stock::reconcile(
&invites,
files.invite_links_path(),
INVITE_LINK_URL,
)
.map_err(|_| ())?;
if unused_invites < 100 {
return Err(());
}
let adapter = K1HttpAccounts::new(
Arc::clone(&accounts),
Arc::clone(&invites),
Arc::clone(&users),
);
let people = K1HttpPeople::new(accounts, users, groups, profiles);
let http = K1Http::new(
Config {
server_id: files.server_id().to_owned(),
public_origin: PUBLIC_ORIGIN.to_owned(),
max_body_bytes: usize::MAX,
},
replay,
adapter.identity_provider(),
)
.map_err(|_| ())?;
let authenticated = adapter
.authenticated_routes()
.merge(people.authenticated_routes())
.fallback(api_not_found);
let api = http
.router(
adapter.registration_endpoint(),
kcode_k1_terms::endpoint(),
authenticated,
)
.layer(middleware::from_fn(contextualize_api_error));
let config = PublicConfig {
protocol: "K1-HTTP-1",
server_id: files.server_id().to_owned(),
public_origin: PUBLIC_ORIGIN,
};
let config_route = get(move || {
let config = config.clone();
async move { ([(CACHE_CONTROL, "no-store")], Json(config)) }
});
let app = Router::new()
.route("/config.json", config_route)
.merge(api)
.layer(middleware::from_fn(require_authority));
Ok(Prepared {
app,
listener: TcpListener::bind(LISTEN_ADDRESS).await.map_err(|_| ())?,
signals: Signals::install()?,
unused_invites,
})
}
fn state_root(k1_root: &Path) -> PathBuf {
k1_root.join("state")
}
async fn api_not_found() -> Response {
json_error(
StatusCode::NOT_FOUND,
"not_found",
"authenticated API route not found",
)
}
async fn contextualize_api_error(request: Request, next: Next) -> Response {
let response = next.run(request).await;
if !(response.status().is_client_error() || response.status().is_server_error()) {
return response;
}
let (mut parts, body) = response.into_parts();
let bytes = match to_bytes(body, usize::MAX).await {
Ok(bytes) => bytes,
Err(_) => return Response::from_parts(parts, Body::empty()),
};
let Some(contextualized) = contextualize_error_body(&bytes) else {
return Response::from_parts(parts, Body::from(bytes));
};
parts.headers.remove(CONTENT_LENGTH);
Response::from_parts(parts, Body::from(contextualized))
}
fn contextualize_error_body(bytes: &[u8]) -> Option<Vec<u8>> {
let mut payload: Value = serde_json::from_slice(bytes).ok()?;
let object = payload.as_object_mut()?;
let code = object.get("error")?.as_str()?.to_owned();
let source = object
.get("message")
.and_then(Value::as_str)
.map(str::to_owned)
.unwrap_or_else(|| format!("error code {code}"));
object.insert(
"message".to_owned(),
Value::String(format!("{API_OPERATION}: {source}")),
);
Some(payload.to_string().into_bytes())
}
async fn require_authority(request: Request, next: Next) -> Response {
let mut values = request.headers().get_all(HOST).iter();
if values
.next()
.is_some_and(|value| value.as_bytes() == AUTHORITY.as_bytes())
&& values.next().is_none()
{
next.run(request).await
} else {
json_error(
StatusCode::MISDIRECTED_REQUEST,
"invalid_request_authority",
"validate request authority: request authority is invalid",
)
}
}
fn json_error(status: StatusCode, code: &'static str, message: &'static str) -> Response {
let mut response = Response::new(Body::from(
serde_json::json!({"error": code, "message": message}).to_string(),
));
*response.status_mut() = status;
response
.headers_mut()
.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
response
.headers_mut()
.insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
response
}
fn write_readiness(unused_invites: usize) -> Result<(), ()> {
let stdout = std::io::stdout();
let mut output = stdout.lock();
serde_json::to_writer(
&mut output,
&Ready {
event: "ready",
public_origin: PUBLIC_ORIGIN,
unused_invites,
},
)
.map_err(|_| ())?;
output.write_all(b"\n").map_err(|_| ())?;
output.flush().map_err(|_| ())
}
fn warn_if_slow(elapsed: Duration, outcome: &'static str) {
if elapsed > STARTUP_BOUND {
eprintln!(
"{{\"module\":\"kcode-k1-daemon\",\"operation\":\"startup\",\"elapsed_us\":{},\"outcome\":\"{outcome}\"}}",
elapsed.as_micros()
);
}
}
impl Signals {
fn install() -> Result<Self, ()> {
Ok(Self {
interrupt: signal(SignalKind::interrupt()).map_err(|_| ())?,
terminate: signal(SignalKind::terminate()).map_err(|_| ())?,
})
}
async fn wait(mut self) {
tokio::select! {
_ = self.interrupt.recv() => {}
_ = self.terminate.recv() => {}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn public_operation_accepts_only_the_state_root() {
let _: fn(PathBuf) -> ExitCode = run;
}
#[test]
fn invite_link_and_backend_origins_remain_distinct() {
assert_eq!(
INVITE_LINK_URL,
"http://localhost:4321/lib/kcode-k1-ui/*/account.html"
);
assert_eq!(PUBLIC_ORIGIN, "http://localhost:4450");
assert_ne!(INVITE_LINK_URL, PUBLIC_ORIGIN);
}
#[test]
fn existing_child_message_is_preserved_under_daemon_context() {
let body = contextualize_error_body(
br#"{"error":"group_failed","message":"load group: child failure","detail":7}"#,
)
.unwrap();
let payload: Value = serde_json::from_slice(&body).unwrap();
assert_eq!(payload["error"], "group_failed");
assert_eq!(payload["detail"], 7);
assert_eq!(
payload["message"],
"serve API request: load group: child failure"
);
}
#[test]
fn missing_child_message_is_derived_from_stable_code() {
let body = contextualize_error_body(br#"{"error":"invalid_signature"}"#).unwrap();
let payload: Value = serde_json::from_slice(&body).unwrap();
assert_eq!(payload["error"], "invalid_signature");
assert_eq!(
payload["message"],
"serve API request: error code invalid_signature"
);
}
#[test]
fn supplied_root_maps_only_to_state() {
assert_eq!(
state_root(Path::new("/trusted/k1")),
PathBuf::from("/trusted/k1/state")
);
}
}