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