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

kcode_k1_daemon_lib/
lib.rs
