kcode_k1_daemon_lib/
lib.rs

1#![doc = include_str!("../Documentation.md")]
2
3use axum::body::{Body, to_bytes};
4use axum::extract::Request;
5use axum::http::header::{CACHE_CONTROL, CONTENT_LENGTH, CONTENT_TYPE, HOST};
6use axum::http::{HeaderValue, StatusCode};
7use axum::middleware::{self, Next};
8use axum::response::Response;
9use axum::routing::get;
10use axum::{Json, Router};
11use kcode_k1_access_profiles::K1AccessProfiles;
12use kcode_k1_accounts::K1Accounts;
13use kcode_k1_daemon_files::DaemonFiles;
14use kcode_k1_groups::K1Groups;
15use kcode_k1_http::{Config, K1Http};
16use kcode_k1_http_accounts::K1HttpAccounts;
17use kcode_k1_http_people::K1HttpPeople;
18use kcode_k1_http_replay::{ReplayConfig, ReplayWindow};
19use kcode_k1_invites::K1Invites;
20use kcode_k1_peering::K1Peering;
21use kcode_k1_txn_ordering::K1TxnOrdering;
22use serde::Serialize;
23use serde_json::Value;
24use std::io::Write as _;
25use std::path::{Path, PathBuf};
26use std::process::ExitCode;
27use std::sync::Arc;
28use std::time::{Duration, Instant};
29use tokio::net::TcpListener;
30use tokio::signal::unix::{Signal, SignalKind, signal};
31
32const LISTEN_ADDRESS: &str = "127.0.0.1:4450";
33const PUBLIC_ORIGIN: &str = "http://localhost:4450";
34const AUTHORITY: &str = "localhost:4450";
35const STARTUP_BOUND: Duration = Duration::from_millis(100);
36const API_OPERATION: &str = "serve API request";
37
38#[derive(Clone, Serialize)]
39struct PublicConfig {
40    protocol: &'static str,
41    server_id: String,
42    public_origin: &'static str,
43}
44
45#[derive(Serialize)]
46struct Ready {
47    event: &'static str,
48    public_origin: &'static str,
49    unused_invites: usize,
50}
51
52struct Prepared {
53    app: Router,
54    listener: TcpListener,
55    signals: Signals,
56    unused_invites: usize,
57}
58
59struct Signals {
60    interrupt: Signal,
61    terminate: Signal,
62}
63
64/// Runs one K1 daemon rooted at `k1_root` until it receives SIGINT or SIGTERM.
65pub fn run(k1_root: PathBuf) -> ExitCode {
66    let runtime = match tokio::runtime::Builder::new_multi_thread()
67        .enable_all()
68        .build()
69    {
70        Ok(runtime) => runtime,
71        Err(_) => {
72            eprintln!("kcode-k1-daemon: startup failed");
73            return ExitCode::from(1);
74        }
75    };
76    runtime.block_on(run_async(k1_root))
77}
78
79async fn run_async(k1_root: PathBuf) -> ExitCode {
80    let started = Instant::now();
81    let prepared = match startup(k1_root).await {
82        Ok(prepared) => prepared,
83        Err(()) => {
84            warn_if_slow(started.elapsed(), "error");
85            eprintln!("kcode-k1-daemon: startup failed");
86            return ExitCode::from(1);
87        }
88    };
89    let elapsed = started.elapsed();
90    if write_readiness(prepared.unused_invites).is_err() {
91        warn_if_slow(elapsed, "error");
92        eprintln!("kcode-k1-daemon: startup failed");
93        return ExitCode::from(1);
94    }
95    warn_if_slow(elapsed, "ready");
96    let Prepared {
97        app,
98        listener,
99        signals,
100        ..
101    } = prepared;
102    match axum::serve(listener, app)
103        .with_graceful_shutdown(signals.wait())
104        .await
105    {
106        Ok(()) => ExitCode::SUCCESS,
107        Err(_) => {
108            eprintln!("kcode-k1-daemon: listener failed");
109            ExitCode::from(1)
110        }
111    }
112}
113
114async fn startup(k1_root: PathBuf) -> Result<Prepared, ()> {
115    let state_root = state_root(&k1_root);
116    let files = DaemonFiles::open(&state_root).map_err(|_| ())?;
117    let ordering = Arc::new(K1TxnOrdering::open(&state_root.join("ordering")).map_err(|_| ())?);
118    let peering = Arc::new(
119        K1Peering::open(&state_root.join("peering"), Arc::clone(&ordering)).map_err(|_| ())?,
120    );
121    let invites = Arc::new(
122        K1Invites::open(
123            &state_root.join("invites"),
124            Arc::clone(&ordering),
125            Arc::clone(&peering),
126        )
127        .map_err(|_| ())?,
128    );
129    let accounts = Arc::new(K1Accounts::open(Arc::clone(&invites)).map_err(|_| ())?);
130    let groups = Arc::new(
131        K1Groups::open(
132            &state_root.join("groups"),
133            Arc::clone(&ordering),
134            Arc::clone(&peering),
135        )
136        .map_err(|_| ())?,
137    );
138    let profiles = Arc::new(
139        K1AccessProfiles::open(
140            &state_root.join("access-profiles"),
141            Arc::clone(&ordering),
142            Arc::clone(&peering),
143        )
144        .map_err(|_| ())?,
145    );
146    let replay = ReplayWindow::open(ReplayConfig {
147        epoch_file: files.replay_epoch_path().to_owned(),
148        max_nonces_per_epoch: usize::MAX,
149    })
150    .await
151    .map_err(|_| ())?;
152    let unused_invites =
153        kcode_k1_daemon_invite_stock::reconcile(&invites, files.invite_links_path(), PUBLIC_ORIGIN)
154            .map_err(|_| ())?;
155    if unused_invites < 100 {
156        return Err(());
157    }
158    let accounts_for_people = Arc::clone(&accounts);
159    let adapter = K1HttpAccounts::new(accounts, Arc::clone(&invites));
160    let people = K1HttpPeople::new(accounts_for_people, groups, profiles);
161    let http = K1Http::new(
162        Config {
163            server_id: files.server_id().to_owned(),
164            public_origin: PUBLIC_ORIGIN.to_owned(),
165            max_body_bytes: usize::MAX,
166        },
167        replay,
168        adapter.identity_provider(),
169    )
170    .map_err(|_| ())?;
171    let authenticated = adapter
172        .authenticated_routes()
173        .merge(people.authenticated_routes())
174        .fallback(api_not_found);
175    let api = http
176        .router(
177            adapter.registration_endpoint(),
178            kcode_k1_terms::endpoint(),
179            authenticated,
180        )
181        .layer(middleware::from_fn(contextualize_api_error));
182    let config = PublicConfig {
183        protocol: "K1-HTTP-1",
184        server_id: files.server_id().to_owned(),
185        public_origin: PUBLIC_ORIGIN,
186    };
187    let config_route = get(move || {
188        let config = config.clone();
189        async move { ([(CACHE_CONTROL, "no-store")], Json(config)) }
190    });
191    let app = Router::new()
192        .route("/config.json", config_route)
193        .merge(api)
194        .layer(middleware::from_fn(require_authority));
195    Ok(Prepared {
196        app,
197        listener: TcpListener::bind(LISTEN_ADDRESS).await.map_err(|_| ())?,
198        signals: Signals::install()?,
199        unused_invites,
200    })
201}
202
203fn state_root(k1_root: &Path) -> PathBuf {
204    k1_root.join("state")
205}
206
207async fn api_not_found() -> Response {
208    json_error(
209        StatusCode::NOT_FOUND,
210        "not_found",
211        "authenticated API route not found",
212    )
213}
214
215async fn contextualize_api_error(request: Request, next: Next) -> Response {
216    let response = next.run(request).await;
217    if !(response.status().is_client_error() || response.status().is_server_error()) {
218        return response;
219    }
220    let (mut parts, body) = response.into_parts();
221    let bytes = match to_bytes(body, usize::MAX).await {
222        Ok(bytes) => bytes,
223        Err(_) => return Response::from_parts(parts, Body::empty()),
224    };
225    let Some(contextualized) = contextualize_error_body(&bytes) else {
226        return Response::from_parts(parts, Body::from(bytes));
227    };
228    parts.headers.remove(CONTENT_LENGTH);
229    Response::from_parts(parts, Body::from(contextualized))
230}
231
232fn contextualize_error_body(bytes: &[u8]) -> Option<Vec<u8>> {
233    let mut payload: Value = serde_json::from_slice(bytes).ok()?;
234    let object = payload.as_object_mut()?;
235    let code = object.get("error")?.as_str()?.to_owned();
236    let source = object
237        .get("message")
238        .and_then(Value::as_str)
239        .map(str::to_owned)
240        .unwrap_or_else(|| format!("error code {code}"));
241    object.insert(
242        "message".to_owned(),
243        Value::String(format!("{API_OPERATION}: {source}")),
244    );
245    Some(payload.to_string().into_bytes())
246}
247
248async fn require_authority(request: Request, next: Next) -> Response {
249    let mut values = request.headers().get_all(HOST).iter();
250    if values
251        .next()
252        .is_some_and(|value| value.as_bytes() == AUTHORITY.as_bytes())
253        && values.next().is_none()
254    {
255        next.run(request).await
256    } else {
257        json_error(
258            StatusCode::MISDIRECTED_REQUEST,
259            "invalid_request_authority",
260            "validate request authority: request authority is invalid",
261        )
262    }
263}
264
265fn json_error(status: StatusCode, code: &'static str, message: &'static str) -> Response {
266    let mut response = Response::new(Body::from(
267        serde_json::json!({"error": code, "message": message}).to_string(),
268    ));
269    *response.status_mut() = status;
270    response
271        .headers_mut()
272        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
273    response
274        .headers_mut()
275        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
276    response
277}
278
279fn write_readiness(unused_invites: usize) -> Result<(), ()> {
280    let stdout = std::io::stdout();
281    let mut output = stdout.lock();
282    serde_json::to_writer(
283        &mut output,
284        &Ready {
285            event: "ready",
286            public_origin: PUBLIC_ORIGIN,
287            unused_invites,
288        },
289    )
290    .map_err(|_| ())?;
291    output.write_all(b"\n").map_err(|_| ())?;
292    output.flush().map_err(|_| ())
293}
294
295fn warn_if_slow(elapsed: Duration, outcome: &'static str) {
296    if elapsed > STARTUP_BOUND {
297        eprintln!(
298            "{{\"module\":\"kcode-k1-daemon\",\"operation\":\"startup\",\"elapsed_us\":{},\"outcome\":\"{outcome}\"}}",
299            elapsed.as_micros()
300        );
301    }
302}
303
304impl Signals {
305    fn install() -> Result<Self, ()> {
306        Ok(Self {
307            interrupt: signal(SignalKind::interrupt()).map_err(|_| ())?,
308            terminate: signal(SignalKind::terminate()).map_err(|_| ())?,
309        })
310    }
311
312    async fn wait(mut self) {
313        tokio::select! {
314            _ = self.interrupt.recv() => {}
315            _ = self.terminate.recv() => {}
316        }
317    }
318}
319
320#[cfg(test)]
321mod tests {
322    use super::*;
323
324    #[test]
325    fn existing_child_message_is_preserved_under_daemon_context() {
326        let body = contextualize_error_body(
327            br#"{"error":"group_failed","message":"load group: child failure","detail":7}"#,
328        )
329        .unwrap();
330        let payload: Value = serde_json::from_slice(&body).unwrap();
331        assert_eq!(payload["error"], "group_failed");
332        assert_eq!(payload["detail"], 7);
333        assert_eq!(
334            payload["message"],
335            "serve API request: load group: child failure"
336        );
337    }
338
339    #[test]
340    fn missing_child_message_is_derived_from_stable_code() {
341        let body = contextualize_error_body(br#"{"error":"invalid_signature"}"#).unwrap();
342        let payload: Value = serde_json::from_slice(&body).unwrap();
343        assert_eq!(payload["error"], "invalid_signature");
344        assert_eq!(
345            payload["message"],
346            "serve API request: error code invalid_signature"
347        );
348    }
349
350    #[test]
351    fn supplied_root_maps_only_to_state() {
352        assert_eq!(
353            state_root(Path::new("/trusted/k1")),
354            PathBuf::from("/trusted/k1/state")
355        );
356    }
357}
kcode_k1_daemon_lib/lib.rs

kcode_k1_daemon_lib/
lib.rs
