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_persons::K1Persons;
22use kcode_k1_txn_ordering::K1TxnOrdering;
23use kcode_k1_users::K1Users;
24use kcode_k1_vault::{K1Vault, SecretString};
25use serde::Serialize;
26use serde_json::Value;
27use std::io::Write as _;
28use std::path::{Path, PathBuf};
29use std::process::ExitCode;
30use std::sync::Arc;
31use std::time::{Duration, Instant};
32use tokio::net::TcpListener;
33use tokio::signal::unix::{Signal, SignalKind, signal};
34
35const LISTEN_ADDRESS: &str = "127.0.0.1:4450";
36const PUBLIC_ORIGIN: &str = "http://localhost:4450";
37const INVITE_LINK_URL: &str = "http://localhost:4321/lib/kcode-k1-ui/*/account.html";
38const AUTHORITY: &str = "localhost:4450";
39const STARTUP_BOUND: Duration = Duration::from_millis(100);
40const API_OPERATION: &str = "serve API request";
41
42#[derive(Clone, Serialize)]
43struct PublicConfig {
44    protocol: &'static str,
45    server_id: String,
46    public_origin: &'static str,
47}
48
49#[derive(Serialize)]
50struct Ready {
51    event: &'static str,
52    public_origin: &'static str,
53    unused_invites: usize,
54}
55
56struct Prepared {
57    app: Router,
58    listener: TcpListener,
59    signals: Signals,
60    unused_invites: usize,
61    vault: Arc<K1Vault>,
62}
63
64struct Signals {
65    interrupt: Signal,
66    terminate: Signal,
67}
68
69pub fn run(k1_root: PathBuf) -> ExitCode {
70    let runtime = match tokio::runtime::Builder::new_multi_thread()
71        .enable_all()
72        .build()
73    {
74        Ok(runtime) => runtime,
75        Err(_) => {
76            eprintln!("kcode-k1-daemon: startup failed");
77            return ExitCode::from(1);
78        }
79    };
80    let passphrase = match rpassword::prompt_password("Unlock K1 vault: ") {
81        Ok(passphrase) => match protect_passphrase(passphrase) {
82            Ok(passphrase) => passphrase,
83            Err(()) => {
84                eprintln!("kcode-k1-daemon: startup failed");
85                return ExitCode::from(1);
86            }
87        },
88        Err(_) => {
89            eprintln!("kcode-k1-daemon: startup failed");
90            return ExitCode::from(1);
91        }
92    };
93    runtime.block_on(run_async(k1_root, passphrase))
94}
95
96fn protect_passphrase(passphrase: String) -> Result<SecretString, ()> {
97    if passphrase.is_empty() {
98        Err(())
99    } else {
100        Ok(SecretString::from(passphrase))
101    }
102}
103
104async fn run_async(k1_root: PathBuf, passphrase: SecretString) -> ExitCode {
105    let started = Instant::now();
106    let prepared = match startup(k1_root, passphrase).await {
107        Ok(prepared) => prepared,
108        Err(()) => {
109            warn_if_slow(started.elapsed(), "error");
110            eprintln!("kcode-k1-daemon: startup failed");
111            return ExitCode::from(1);
112        }
113    };
114    let elapsed = started.elapsed();
115    if write_readiness(prepared.unused_invites).is_err() {
116        warn_if_slow(elapsed, "error");
117        eprintln!("kcode-k1-daemon: startup failed");
118        return ExitCode::from(1);
119    }
120    warn_if_slow(elapsed, "ready");
121    let Prepared {
122        app,
123        listener,
124        signals,
125        vault,
126        ..
127    } = prepared;
128    let result = axum::serve(listener, app)
129        .with_graceful_shutdown(signals.wait())
130        .await;
131    drop(vault);
132    match result {
133        Ok(()) => ExitCode::SUCCESS,
134        Err(_) => {
135            eprintln!("kcode-k1-daemon: listener failed");
136            ExitCode::from(1)
137        }
138    }
139}
140
141async fn startup(k1_root: PathBuf, passphrase: SecretString) -> Result<Prepared, ()> {
142    let state_root = state_root(&k1_root);
143    let files = DaemonFiles::open(&state_root).map_err(|_| ())?;
144    let ordering = Arc::new(K1TxnOrdering::open(&state_root.join("ordering")).map_err(|_| ())?);
145    let peering = Arc::new(
146        K1Peering::open(&state_root.join("peering"), Arc::clone(&ordering)).map_err(|_| ())?,
147    );
148    let vault = open_vault(
149        &state_root,
150        passphrase,
151        Arc::clone(&ordering),
152        Arc::clone(&peering),
153    )?;
154    let persons = Arc::new(
155        K1Persons::open(
156            &state_root.join("persons"),
157            Arc::clone(&ordering),
158            Arc::clone(&peering),
159        )
160        .map_err(|_| ())?,
161    );
162    let invites = Arc::new(
163        K1Invites::open(
164            &state_root.join("invites"),
165            Arc::clone(&ordering),
166            Arc::clone(&peering),
167        )
168        .map_err(|_| ())?,
169    );
170    let accounts = Arc::new(K1Accounts::open(Arc::clone(&invites)).map_err(|_| ())?);
171    let users = Arc::new(K1Users::new(Arc::clone(&accounts), Arc::clone(&persons)));
172    let groups = Arc::new(
173        K1Groups::open(
174            &state_root.join("groups"),
175            Arc::clone(&ordering),
176            Arc::clone(&peering),
177        )
178        .map_err(|_| ())?,
179    );
180    let profiles = Arc::new(
181        K1AccessProfiles::open(
182            &state_root.join("access-profiles"),
183            Arc::clone(&ordering),
184            Arc::clone(&peering),
185        )
186        .map_err(|_| ())?,
187    );
188    let replay = ReplayWindow::open(ReplayConfig {
189        epoch_file: files.replay_epoch_path().to_owned(),
190        max_nonces_per_epoch: usize::MAX,
191    })
192    .await
193    .map_err(|_| ())?;
194    let unused_invites = kcode_k1_daemon_invite_stock::reconcile(
195        &invites,
196        files.invite_links_path(),
197        INVITE_LINK_URL,
198    )
199    .map_err(|_| ())?;
200    if unused_invites < 100 {
201        return Err(());
202    }
203    let adapter = K1HttpAccounts::new(
204        Arc::clone(&accounts),
205        Arc::clone(&invites),
206        Arc::clone(&users),
207    );
208    let people = K1HttpPeople::new(accounts, users, groups, profiles);
209    let http = K1Http::new(
210        Config {
211            server_id: files.server_id().to_owned(),
212            public_origin: PUBLIC_ORIGIN.to_owned(),
213            max_body_bytes: usize::MAX,
214        },
215        replay,
216        adapter.identity_provider(),
217    )
218    .map_err(|_| ())?;
219    let authenticated = adapter
220        .authenticated_routes()
221        .merge(people.authenticated_routes())
222        .fallback(api_not_found);
223    let api = http
224        .router(
225            adapter.registration_endpoint(),
226            kcode_k1_terms::endpoint(),
227            authenticated,
228        )
229        .layer(middleware::from_fn(contextualize_api_error));
230    let config = PublicConfig {
231        protocol: "K1-HTTP-1",
232        server_id: files.server_id().to_owned(),
233        public_origin: PUBLIC_ORIGIN,
234    };
235    let config_route = get(move || {
236        let config = config.clone();
237        async move { ([(CACHE_CONTROL, "no-store")], Json(config)) }
238    });
239    let app = Router::new()
240        .route("/config.json", config_route)
241        .merge(api)
242        .layer(middleware::from_fn(require_authority));
243    Ok(Prepared {
244        app,
245        listener: TcpListener::bind(LISTEN_ADDRESS).await.map_err(|_| ())?,
246        signals: Signals::install()?,
247        unused_invites,
248        vault,
249    })
250}
251
252fn open_vault(
253    state_root: &Path,
254    passphrase: SecretString,
255    ordering: Arc<K1TxnOrdering>,
256    peering: Arc<K1Peering>,
257) -> Result<Arc<K1Vault>, ()> {
258    K1Vault::open(&state_root.join("vault"), passphrase, ordering, peering)
259        .map(Arc::new)
260        .map_err(|_| ())
261}
262
263fn state_root(k1_root: &Path) -> PathBuf {
264    k1_root.join("state")
265}
266
267async fn api_not_found() -> Response {
268    json_error(
269        StatusCode::NOT_FOUND,
270        "not_found",
271        "authenticated API route not found",
272    )
273}
274
275async fn contextualize_api_error(request: Request, next: Next) -> Response {
276    let response = next.run(request).await;
277    if !(response.status().is_client_error() || response.status().is_server_error()) {
278        return response;
279    }
280    let (mut parts, body) = response.into_parts();
281    let bytes = match to_bytes(body, usize::MAX).await {
282        Ok(bytes) => bytes,
283        Err(_) => return Response::from_parts(parts, Body::empty()),
284    };
285    let Some(contextualized) = contextualize_error_body(&bytes) else {
286        return Response::from_parts(parts, Body::from(bytes));
287    };
288    parts.headers.remove(CONTENT_LENGTH);
289    Response::from_parts(parts, Body::from(contextualized))
290}
291
292fn contextualize_error_body(bytes: &[u8]) -> Option<Vec<u8>> {
293    let mut payload: Value = serde_json::from_slice(bytes).ok()?;
294    let object = payload.as_object_mut()?;
295    let code = object.get("error")?.as_str()?.to_owned();
296    let source = object
297        .get("message")
298        .and_then(Value::as_str)
299        .map(str::to_owned)
300        .unwrap_or_else(|| format!("error code {code}"));
301    object.insert(
302        "message".to_owned(),
303        Value::String(format!("{API_OPERATION}: {source}")),
304    );
305    Some(payload.to_string().into_bytes())
306}
307
308async fn require_authority(request: Request, next: Next) -> Response {
309    let mut values = request.headers().get_all(HOST).iter();
310    if values
311        .next()
312        .is_some_and(|value| value.as_bytes() == AUTHORITY.as_bytes())
313        && values.next().is_none()
314    {
315        next.run(request).await
316    } else {
317        json_error(
318            StatusCode::MISDIRECTED_REQUEST,
319            "invalid_request_authority",
320            "validate request authority: request authority is invalid",
321        )
322    }
323}
324
325fn json_error(status: StatusCode, code: &'static str, message: &'static str) -> Response {
326    let mut response = Response::new(Body::from(
327        serde_json::json!({"error": code, "message": message}).to_string(),
328    ));
329    *response.status_mut() = status;
330    response
331        .headers_mut()
332        .insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
333    response
334        .headers_mut()
335        .insert(CACHE_CONTROL, HeaderValue::from_static("no-store"));
336    response
337}
338
339fn write_readiness(unused_invites: usize) -> Result<(), ()> {
340    let stdout = std::io::stdout();
341    let mut output = stdout.lock();
342    serde_json::to_writer(
343        &mut output,
344        &Ready {
345            event: "ready",
346            public_origin: PUBLIC_ORIGIN,
347            unused_invites,
348        },
349    )
350    .map_err(|_| ())?;
351    output.write_all(b"\n").map_err(|_| ())?;
352    output.flush().map_err(|_| ())
353}
354
355fn warn_if_slow(elapsed: Duration, outcome: &'static str) {
356    if elapsed > STARTUP_BOUND {
357        eprintln!(
358            "{{\"module\":\"kcode-k1-daemon\",\"operation\":\"startup\",\"elapsed_us\":{},\"outcome\":\"{outcome}\"}}",
359            elapsed.as_micros()
360        );
361    }
362}
363
364impl Signals {
365    fn install() -> Result<Self, ()> {
366        Ok(Self {
367            interrupt: signal(SignalKind::interrupt()).map_err(|_| ())?,
368            terminate: signal(SignalKind::terminate()).map_err(|_| ())?,
369        })
370    }
371
372    async fn wait(mut self) {
373        tokio::select! {
374            _ = self.interrupt.recv() => {}
375            _ = self.terminate.recv() => {}
376        }
377    }
378}
379
380#[cfg(test)]
381mod tests {
382    use super::*;
383
384    #[test]
385    fn public_operation_accepts_only_the_state_root() {
386        let _: fn(PathBuf) -> ExitCode = run;
387    }
388
389    #[test]
390    fn accepted_passphrase_boundary_is_strict_and_protected() {
391        assert!(protect_passphrase(String::new()).is_err());
392        let text = "conspicuous-fake-passphrase-never-real";
393        let protected = protect_passphrase(text.to_owned()).unwrap();
394        assert!(!format!("{protected:?}").contains(text));
395    }
396
397    #[test]
398    fn vault_composition_persists_at_the_fixed_path() {
399        let root =
400            std::env::temp_dir().join(format!("kcode-k1-daemon-vault-test-{}", std::process::id()));
401        let _ = std::fs::remove_dir_all(&root);
402        let state = state_root(&root);
403        assert_eq!(state.join("vault"), root.join("state/vault"));
404        let parts = || {
405            let ordering = Arc::new(K1TxnOrdering::open(&state.join("ordering")).unwrap());
406            let peering =
407                Arc::new(K1Peering::open(&state.join("peering"), ordering.clone()).unwrap());
408            (ordering, peering)
409        };
410        let password = || SecretString::from("fake-test-password-never-real");
411        let (ordering, peering) = parts();
412        let vault = open_vault(&state, password(), ordering.clone(), peering.clone()).unwrap();
413        vault
414            .set(
415                "fake-provider-secret",
416                SecretString::from("conspicuous-fake-value-never-real"),
417            )
418            .unwrap();
419        drop((vault, peering, ordering));
420        let (ordering, peering) = parts();
421        let vault = open_vault(&state, password(), ordering.clone(), peering.clone()).unwrap();
422        drop((vault, peering, ordering));
423        let (ordering, peering) = parts();
424        assert!(
425            open_vault(
426                &state,
427                SecretString::from("wrong-fake-password-never-real"),
428                ordering,
429                peering
430            )
431            .is_err()
432        );
433        std::fs::remove_dir_all(root).unwrap();
434    }
435
436    #[test]
437    fn invite_link_and_backend_origins_remain_distinct() {
438        assert_eq!(
439            INVITE_LINK_URL,
440            "http://localhost:4321/lib/kcode-k1-ui/*/account.html"
441        );
442        assert_eq!(PUBLIC_ORIGIN, "http://localhost:4450");
443        assert_ne!(INVITE_LINK_URL, PUBLIC_ORIGIN);
444    }
445
446    #[test]
447    fn existing_child_message_is_preserved_under_daemon_context() {
448        let body = contextualize_error_body(
449            br#"{"error":"group_failed","message":"load group: child failure","detail":7}"#,
450        )
451        .unwrap();
452        let payload: Value = serde_json::from_slice(&body).unwrap();
453        assert_eq!(payload["error"], "group_failed");
454        assert_eq!(payload["detail"], 7);
455        assert_eq!(
456            payload["message"],
457            "serve API request: load group: child failure"
458        );
459    }
460
461    #[test]
462    fn missing_child_message_is_derived_from_stable_code() {
463        let body = contextualize_error_body(br#"{"error":"invalid_signature"}"#).unwrap();
464        let payload: Value = serde_json::from_slice(&body).unwrap();
465        assert_eq!(payload["error"], "invalid_signature");
466        assert_eq!(
467            payload["message"],
468            "serve API request: error code invalid_signature"
469        );
470    }
471
472    #[test]
473    fn supplied_root_maps_only_to_state() {
474        assert_eq!(
475            state_root(Path::new("/trusted/k1")),
476            PathBuf::from("/trusted/k1/state")
477        );
478    }
479}
kcode_k1_daemon_lib/lib.rs

kcode_k1_daemon_lib/
lib.rs
