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

kcode_k1_daemon_lib/
lib.rs
