From b39bbd4f18b54b61344f671bf40513d19042e471 Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 6 Nov 2026 19:49:19 +0000 Subject: [PATCH 1/4] Run world service policy in WASI init with pollable process observation --- host/src/harness/mod.rs | 3 +- host/src/process/mod.rs | 19 ++ host/src/process/services.rs | 211 +++++--------- host/src/process/watcher.rs | 110 ++++++++ host/tests/init.rs | 455 +++++++++++++++++++++++++++++++ host/tests/process.rs | 64 +---- host/wit/process.wit | 9 + userland/Cargo.lock | 11 + userland/Cargo.toml | 2 +- userland/init/Cargo.toml | 31 +++ userland/init/README.md | 75 +++++ userland/init/src/journalctl.rs | 66 +++++ userland/init/src/lib.rs | 188 +++++++++++++ userland/init/src/main.rs | 421 ++++++++++++++++++++++++++++ userland/init/src/probe.rs | 96 +++++++ userland/init/src/systemctl.rs | 123 +++++++++ userland/init/src/systemd-run.rs | 82 ++++++ userland/spawn/src/lib.rs | 6 +- vfs/src/lib.rs | 4 + vfs/src/mounts.rs | 18 +- vfs/src/observe.rs | 51 +++- 21 files changed, 1821 insertions(+), 224 deletions(-) create mode 100644 host/src/process/watcher.rs create mode 100644 host/tests/init.rs create mode 100644 userland/init/Cargo.toml create mode 100644 userland/init/README.md create mode 100644 userland/init/src/journalctl.rs create mode 100644 userland/init/src/lib.rs create mode 100644 userland/init/src/main.rs create mode 100644 userland/init/src/probe.rs create mode 100644 userland/init/src/systemctl.rs create mode 100644 userland/init/src/systemd-run.rs diff --git a/host/src/harness/mod.rs b/host/src/harness/mod.rs index b95d1cf..357b9e4 100644 --- a/host/src/harness/mod.rs +++ b/host/src/harness/mod.rs @@ -208,7 +208,6 @@ pub async fn run_with_userland(path: &Path, userland: Option<&Path>) -> anyhow:: })?; let services = crate::process::supervise(runtime.clone()); let result = run_with_exec(world, Arc::new(runtime)).await; - services.abort(); - let _ = services.await; + services.shutdown().await; result } diff --git a/host/src/process/mod.rs b/host/src/process/mod.rs index 187c423..04858c5 100644 --- a/host/src/process/mod.rs +++ b/host/src/process/mod.rs @@ -41,11 +41,14 @@ mod bindings { with: { "wasi:io": wasmtime_wasi::p2::bindings::io, "sausage:process/host.child": super::Child, + "sausage:process/host.watcher": super::Watcher, }, }); } use bindings::sausage::process::host::{Host, HostChild, Request, Stdio}; mod services; +mod watcher; +pub use watcher::Watcher; mod stdio; pub use services::supervise; use stdio::Io; @@ -470,7 +473,23 @@ impl Host for State { Ok(self.runtime.kill_pid(pid).map_err(|e| e.to_string())) } } +#[async_trait::async_trait] +impl wasmtime_wasi::p2::Pollable for Child { + async fn ready(&mut self) { + while self.code.borrow_and_update().is_none() { + if self.code.changed().await.is_err() { + break; + } + } + } +} impl HostChild for State { + async fn subscribe( + &mut self, + child: Resource, + ) -> wasmtime::Result> { + wasmtime_wasi::p2::subscribe(&mut self.table, child) + } async fn pid(&mut self, child: Resource) -> wasmtime::Result { Ok(self.table.get(&child)?.pid) } diff --git a/host/src/process/services.rs b/host/src/process/services.rs index f61e065..147ae8c 100644 --- a/host/src/process/services.rs +++ b/host/src/process/services.rs @@ -1,49 +1,28 @@ -//! Supervises the world's services list, live edits, crash limits, and graceful stops. -use super::{GroupGuard, Request, Runtime, Started, Stdio}; -use serde::Deserialize; -use std::{ - collections::{BTreeMap, VecDeque}, - sync::Arc, - time::Duration, -}; -use tokio::time::Instant; - -#[derive(Clone, Deserialize, PartialEq, Eq)] -struct Spec { - name: String, - argv: Vec, - #[serde(default = "root")] - cwd: String, - #[serde(default)] - env: BTreeMap, +//! Starts world init, reaps its command tree, and applies host restart and shutdown deadlines. +use super::{GroupGuard, Request, Runtime, Stdio}; +use std::{sync::Arc, time::Duration}; +pub struct Supervisor { + shutdown: tokio::sync::oneshot::Sender<()>, + task: tokio::task::JoinHandle<()>, } -fn root() -> String { - "/".into() +impl Supervisor { + pub async fn shutdown(self) { + let _ = self.shutdown.send(()); + let _ = self.task.await; + } } -struct Live { - child: Started, - _guard: GroupGuard, -} -struct Service { - spec: Spec, - live: Option, - crashes: VecDeque, - stopped: bool, -} -impl Service { - async fn start(&mut self, runtime: &Arc) { - match runtime - .start( +/// The host knows only init. Unit policy belongs to the WASI program. +pub fn supervise(runtime: Arc) -> Supervisor { + let (shutdown, mut stop) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + let mut backoff = Duration::from_secs(1); + loop { + let start = runtime.start( Request { - program: self.spec.argv.first().cloned().unwrap_or_default(), - argv: self.spec.argv.clone(), - cwd: self.spec.cwd.clone(), - env: self - .spec - .env - .iter() - .map(|(k, v)| (k.clone(), v.clone())) - .collect(), + program: "/bin/init.wasm".into(), + argv: vec!["/bin/init.wasm".into()], + cwd: "/".into(), + env: Vec::new(), group: None, stdin: Stdio::Pipe, stdout: Stdio::Null, @@ -54,116 +33,56 @@ impl Service { 0, None, None, - ) - .await - { - Ok(child) => { + ); + let child = tokio::select! { _ = &mut stop => return, child = start => child }; + if let Ok(mut child) = child { let guard = GroupGuard { runtime: runtime.clone(), group: child.pid, }; - self.live = Some(Live { - child, - _guard: guard, - }); + let exiting = async { + while child.code.borrow_and_update().is_none() { + if child.code.changed().await.is_err() { + break; + } + } + }; + let shutdown = tokio::select! { _ = &mut stop => true, _ = exiting => false }; + if shutdown { + drop(child.input.take()); + let _ = tokio::time::timeout(Duration::from_secs(15), async { + while child.code.borrow_and_update().is_none() { + if child.code.changed().await.is_err() { + break; + } + } + }) + .await; + drop(guard); + reaped(&runtime, child.pid).await; + return; + } + // This guard kills all descendants, including separate process groups. + drop(guard); + reaped(&runtime, child.pid).await; } - Err(_) => self.crashed(), + tokio::select! { _ = &mut stop => return, _ = tokio::time::sleep(backoff) => () } + backoff = (backoff * 2).min(Duration::from_secs(30)); } - } - fn crashed(&mut self) { - let now = Instant::now(); - while self - .crashes - .front() - .is_some_and(|t| now.duration_since(*t) >= Duration::from_secs(10)) + }); + Supervisor { shutdown, task } +} +async fn reaped(runtime: &Runtime, command: u32) { + loop { + if !runtime + .entries + .lock() + .unwrap() + .values() + .any(|entry| entry.command == command) { - self.crashes.pop_front(); - } - self.crashes.push_back(now); - self.stopped = self.crashes.len() > 5; - } - async fn stop(&mut self) { - if let Some(mut live) = self.live.take() { - drop(live.child.input.take()); - let _ = tokio::time::timeout(Duration::from_secs(5), async { - loop { - if live.child.code.borrow().is_some() { - break; - } - if live.child.code.changed().await.is_err() { - break; - } - } - }) - .await; - // The guard stops remaining descendants, or the process after its grace period. + return; } + tokio::task::yield_now().await; } } -/// Start the supervisor. Aborting its task stops its service processes. -/// The YAML file contains name, argv, optional cwd, and an optional map of environment variables. -pub fn supervise(runtime: Arc) -> tokio::task::JoinHandle<()> { - tokio::spawn(async move { - let mut services: BTreeMap = BTreeMap::new(); - let mut previous = Vec::new(); - let mut interval = tokio::time::interval(Duration::from_millis(100)); - loop { - interval.tick().await; - let content = match runtime - .world - .root - .open("/config/services.yaml", Default::default()) - .await - { - Ok(file) => match file.stat().await { - Ok(stat) => file.read(0, stat.size as usize).await.ok(), - Err(_) => None, - }, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Some(b"[]".to_vec()), - Err(_) => None, - }; - if let Some(content) = content.filter(|c| *c != previous) - && let Ok(specs) = serde_yaml_ng::from_slice::>(&content) - { - let specs: BTreeMap<_, _> = - specs.into_iter().map(|s| (s.name.clone(), s)).collect(); - let removed: Vec<_> = services - .keys() - .filter(|name| !specs.contains_key(*name)) - .cloned() - .collect(); - for name in removed { - services.remove(&name).unwrap().stop().await; - } - for (name, spec) in specs { - let service = services.entry(name).or_insert_with(|| Service { - spec: spec.clone(), - live: None, - crashes: VecDeque::new(), - stopped: false, - }); - if service.spec != spec { - service.stop().await; - } - service.spec = spec; - service.crashes.clear(); - service.stopped = false; - } - previous = content; - } - for service in services.values_mut() { - if service - .live - .as_ref() - .is_some_and(|live| live.child.code.borrow().is_some()) - { - service.live = None; - service.crashed(); - } - if service.live.is_none() && !service.stopped { - service.start(&runtime).await; - } - } - } - }) -} diff --git a/host/src/process/watcher.rs b/host/src/process/watcher.rs new file mode 100644 index 0000000..5ec2408 --- /dev/null +++ b/host/src/process/watcher.rs @@ -0,0 +1,110 @@ +//! Adapts VFS observation to bounded, pollable world-path change queues. +use super::{ + State, + bindings::sausage::process::host::{Change, ChangeKind, HostWatcher}, +}; +use sausage_vfs::observe::{ChangeKind as Kind, under}; +use tokio::sync::broadcast::{Receiver, error::TryRecvError}; +use wasmtime::component::Resource; +use wasmtime_wasi::p2::{DynPollable, Pollable, subscribe}; + +pub struct Watcher { + paths: Vec, + events: Receiver<(String, Kind)>, + pending: Vec, +} +impl Watcher { + fn push(&mut self, path: String, kind: Kind) { + if self.paths.iter().any(|base| under(&path, base)) { + if self + .pending + .iter() + .any(|c| matches!(c.kind, ChangeKind::Overflow)) + { + return; + } + if self.pending.len() >= 256 { + self.overflow(); + return; + } + self.pending.push(Change { + path, + kind: match kind { + Kind::Created => ChangeKind::Created, + Kind::Modified => ChangeKind::Modified, + Kind::Removed => ChangeKind::Removed, + }, + }); + } + } + fn overflow(&mut self) { + self.pending.clear(); + self.pending.push(Change { + path: "/".into(), + kind: ChangeKind::Overflow, + }); + } + fn drain(&mut self) { + loop { + match self.events.try_recv() { + Ok((path, kind)) => self.push(path, kind), + Err(TryRecvError::Lagged(_)) => self.overflow(), + Err(_) => break, + } + } + } +} +#[async_trait::async_trait] +impl Pollable for Watcher { + async fn ready(&mut self) { + loop { + self.drain(); + if !self.pending.is_empty() { + return; + } + match self.events.recv().await { + Ok((path, kind)) => self.push(path, kind), + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => self.overflow(), + Err(_) => return, + } + } + } +} +impl HostWatcher for State { + async fn new(&mut self, paths: Vec) -> wasmtime::Result> { + let observer = self + .runtime + .world + .root + .observer() + .ok_or_else(|| wasmtime::format_err!("filesystem has no observer"))?; + Ok(self.table.push(Watcher { + paths: paths + .iter() + .map(|p| { + sausage_vfs::resolve("/", p) + .trim_end_matches('/') + .to_owned() + }) + .map(|p| if p.is_empty() { "/".into() } else { p }) + .collect(), + events: observer.subscribe(), + pending: Vec::new(), + })?) + } + async fn subscribe( + &mut self, + watcher: Resource, + ) -> wasmtime::Result> { + subscribe(&mut self.table, watcher) + } + async fn changes(&mut self, watcher: Resource) -> wasmtime::Result> { + let watcher = self.table.get_mut(&watcher)?; + watcher.drain(); + Ok(std::mem::take(&mut watcher.pending)) + } + async fn drop(&mut self, watcher: Resource) -> wasmtime::Result<()> { + self.table.delete(watcher)?; + Ok(()) + } +} diff --git a/host/tests/init.rs b/host/tests/init.rs new file mode 100644 index 0000000..db9b111 --- /dev/null +++ b/host/tests/init.rs @@ -0,0 +1,455 @@ +//! Init acceptance tests use real WASI programs, a persistent guest cache, and temporary worlds. +use sausage::{ + exec::{Command, Exec, Options, Stream}, + process::{Runtime, supervise}, + world::World, +}; +use sausage_vfs::{Layer, Open}; +use std::{ + path::{Path, PathBuf}, + sync::{Arc, OnceLock}, + time::Duration, +}; +static SERIAL: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); +fn guests() -> &'static Path { + static BUILT: OnceLock = OnceLock::new(); + BUILT.get_or_init(|| { + let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("init-guests"); + let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../userland/Cargo.toml"); + assert!( + std::process::Command::new("cargo") + .args([ + "build", + "--release", + "--target", + "wasm32-wasip2", + "-p", + "sausage-init", + "--manifest-path" + ]) + .arg(manifest) + .env("CARGO_TARGET_DIR", &target) + .status() + .unwrap() + .success() + ); + target.join("wasm32-wasip2/release") + }) +} +fn world() -> (tempfile::TempDir, Arc, Arc) { + let temp = tempfile::tempdir().unwrap(); + let bin = temp.path().join("world/bin"); + std::fs::create_dir_all(&bin).unwrap(); + for (source, name) in [ + ("init", "init"), + ("systemctl", "systemctl"), + ("journalctl", "journalctl"), + ("systemd-run", "systemd-run"), + ("init-probe", "probe"), + ] { + std::fs::copy( + guests().join(format!("{source}.wasm")), + bin.join(format!("{name}.wasm")), + ) + .unwrap(); + } + std::fs::create_dir_all( + temp.path() + .join("world/etc/systemd/system/default.target.wants"), + ) + .unwrap(); + let cache = Path::new(env!("CARGO_TARGET_TMPDIR")).join("init-components"); + std::fs::create_dir_all(&cache).unwrap(); + std::fs::create_dir_all(temp.path().join("host")).unwrap(); + std::os::unix::fs::symlink(cache, temp.path().join("host/compile-cache")).unwrap(); + let world = World::open(temp.path()).unwrap(); + let root = world.root.clone(); + (temp, Runtime::new(world).unwrap(), root) +} +async fn write(root: &Arc, path: &str, text: &[u8]) { + let file = root + .open( + path, + Open { + create: true, + write: true, + truncate: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.write(0, text.to_vec()).await.unwrap(); +} +async fn unit(root: &Arc, name: &str, body: &str) { + write( + root, + &format!("/etc/systemd/system/default.target.wants/{name}.service"), + b"", + ) + .await; + write( + root, + &format!("/etc/systemd/system/{name}.service"), + body.as_bytes(), + ) + .await; +} +async fn run(runtime: &Arc, argv: &[&str]) -> (i32, String) { + let mut out = Vec::new(); + let code = runtime + .exec( + Command::Argv(argv.iter().map(|a| (*a).into()).collect()), + Options { + timeout: Some(Duration::from_secs(30)), + cwd: "/".into(), + ..Default::default() + }, + &mut |stream, bytes| { + if stream == Stream::Stdout { + out.extend_from_slice(bytes); + } else { + eprint!("{}", String::from_utf8_lossy(bytes)); + } + }, + ) + .await + .unwrap(); + (code, String::from_utf8_lossy(&out).into_owned()) +} +async fn wait(mut check: impl FnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(30), async { + while !check() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} +async fn state(root: &Arc, unit: &str, expected: &str) { + tokio::time::timeout(Duration::from_secs(30), async { + loop { + if let Ok(file) = root + .open( + &format!("/run/systemd/state/{unit}.service"), + Default::default(), + ) + .await + && let Ok(bytes) = file.read(0, 65536).await + && serde_json::from_slice::(&bytes) + .is_ok_and(|s| s["state"] == expected) + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); +} +fn lines(path: &Path) -> Vec { + std::fs::read_to_string(path) + .unwrap_or_default() + .lines() + .map(str::to_owned) + .collect() +} +#[tokio::test(flavor = "multi_thread")] +async fn watcher_and_child_exit_pollable() { + let _serial = SERIAL.lock().await; + let (_temp, runtime, _) = world(); + let result = run(&runtime, &["/bin/probe.wasm", "watch"]).await; + assert_eq!(result.0, 0, "{}", result.1); +} +#[tokio::test(flavor = "multi_thread")] +async fn dependencies_restart_limit_and_journal() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + unit( + &root, + "prepare", + "[Service]\nType=oneshot\nExecStart=/bin/probe.wasm oneshot /order.log\n", + ) + .await; + unit(&root, "after", "[Unit]\nAfter=prepare.service\nRequires=prepare.service\n[Service]\nExecStart=/bin/probe.wasm after /after.log\n").await; + unit( + &root, + "fail", + "[Service]\nType=oneshot\nExecStart=/bin/probe.wasm exit 2\n", + ) + .await; + unit(&root, "blocked", "[Unit]\nAfter=fail.service\nRequires=fail.service\n[Service]\nExecStart=/bin/probe.wasm oneshot /blocked.log\n").await; + unit(&root, "crash", "[Service]\nExecStart=/bin/probe.wasm crash /crash.log\nRestart=on-failure\nRestartSec=0.01\n").await; + unit(&root, "always", "[Service]\nExecStart=/bin/probe.wasm oneshot /always.log\nRestart=always\nRestartSec=0.01\nStartLimitBurst=2\n").await; + unit( + &root, + "success", + "[Service]\nExecStart=/bin/probe.wasm oneshot /success.log\nRestart=on-failure\n", + ) + .await; + unit( + &root, + "no", + "[Service]\nExecStart=/bin/probe.wasm crash /no.log\nRestart=no\n", + ) + .await; + unit( + &root, + "linger", + "[Service]\nExecStart=/bin/probe.wasm linger /linger.log\nTimeoutStopSec=0.02\n", + ) + .await; + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + wait(|| { + lines(&path.join("after.log")) == ["start"] && lines(&path.join("crash.log")).len() == 5 + }) + .await; + state(&root, "always", "failed").await; + assert_eq!(lines(&path.join("always.log")).len(), 2); + assert_eq!(lines(&path.join("success.log")).len(), 1); + assert_eq!(lines(&path.join("no.log")).len(), 1); + state(&root, "crash", "failed").await; + state(&root, "after", "active").await; + assert_eq!(lines(&path.join("order.log")), ["start"]); + assert!(!path.join("blocked.log").exists()); + wait(|| { + std::fs::read_to_string(path.join("var/log/journal/after.service.log")) + .unwrap_or_default() + .contains("line 10") + }) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "is-active", "after"]) + .await + .0, + 0 + ); + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "is-active", "crash"]) + .await + .0, + 3 + ); + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "is-failed", "crash"]) + .await + .0, + 0 + ); + let (code, output) = run( + &runtime, + &["/bin/journalctl.wasm", "-u", "after", "-n", "5"], + ) + .await; + assert_eq!(code, 0); + assert_eq!(output, "line 6\nline 7\nline 8\nline 9\nline 10\n"); + state(&root, "linger", "active").await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "stop", "linger"]) + .await + .0, + 0 + ); + state(&root, "linger", "inactive").await; + assert_eq!(lines(&path.join("linger.log")), ["start"]); + supervisor.shutdown().await; + assert_eq!(lines(&path.join("after.log")), ["start", "stop"]); +} +#[tokio::test(flavor = "multi_thread")] +async fn live_unit_changes_and_binary_replacement() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + unit( + &root, + "live", + "[Service]\nExecStart=/bin/probe.wasm server /live.log\n", + ) + .await; + wait(|| lines(&path.join("live.log")) == ["start"]).await; + write( + &root, + "/etc/systemd/system/live.service", + b"[Service]\nExecStart=/bin/probe.wasm server /live.log\nEnvironment=REV=2\n", + ) + .await; + wait(|| lines(&path.join("live.log")) == ["start", "stop", "start"]).await; + let bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + write(&root, "/bin/replacement.wasm", &bytes).await; + root.rename("/bin/replacement.wasm", "/bin/probe.wasm") + .await + .unwrap(); + wait(|| lines(&path.join("live.log")).len() >= 5).await; + root.remove_file("/etc/systemd/system/live.service") + .await + .unwrap(); + wait(|| { + lines(&path.join("live.log")) + .last() + .is_some_and(|l| l == "stop") + }) + .await; + supervisor.shutdown().await; +} +#[tokio::test(flavor = "multi_thread")] +async fn enable_persists_but_transient_units_do_not() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + write(&root, "/etc/systemd/system/persistent.service", b"[Service]\nExecStart=/bin/probe.wasm server /persistent.log\n[Install]\nWantedBy=default.target\n").await; + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + wait(|| path.join("var/log/journal").exists()).await; + assert_eq!( + run( + &runtime, + &["/bin/systemctl.wasm", "enable", "--now", "persistent"] + ) + .await + .0, + 0 + ); + wait(|| lines(&path.join("persistent.log")) == ["start"]).await; + assert_eq!( + run( + &runtime, + &["/bin/systemctl.wasm", "is-enabled", "persistent"] + ) + .await + .0, + 0 + ); + assert_eq!( + run( + &runtime, + &[ + "/bin/systemd-run.wasm", + "--unit", + "transient", + "--wait", + "/bin/probe.wasm", + "oneshot", + "/transient.log" + ] + ) + .await + .0, + 0 + ); + supervisor.shutdown().await; + drop(root); + drop(runtime); + let world = World::open(temp.path()).unwrap(); + assert!( + world + .root + .stat("/run/systemd/transient/transient.service", true) + .await + .is_err() + ); + let runtime = Runtime::new(world).unwrap(); + let supervisor = supervise(runtime.clone()); + wait(|| lines(&path.join("persistent.log")) == ["start", "stop", "start"]).await; + assert_eq!(lines(&path.join("transient.log")), ["start"]); + supervisor.shutdown().await; +} +#[tokio::test(flavor = "multi_thread")] +async fn init_crash_reaps_children_and_restarts_enabled_units() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + unit( + &root, + "server", + "[Service]\nExecStart=/bin/probe.wasm server /server.log\n", + ) + .await; + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world/server.log"); + wait(|| lines(&path) == ["start"]).await; + assert_eq!(run(&runtime, &["/bin/probe.wasm", "kill-init"]).await.0, 0); + wait(|| { + lines(&path) + .iter() + .filter(|line| line.as_str() == "start") + .count() + == 2 + }) + .await; + let table = root + .open("/proc/processes", Default::default()) + .await + .unwrap() + .read(0, 65536) + .await + .unwrap(); + let table = String::from_utf8(table).unwrap(); + assert_eq!( + table + .lines() + .filter(|l| l.contains("/bin/probe.wasm server")) + .count(), + 1 + ); + supervisor.shutdown().await; +} +#[tokio::test(flavor = "multi_thread")] +async fn journal_follow_observes_new_lines() { + let _serial = SERIAL.lock().await; + let (_temp, runtime, root) = world(); + // Init creates the public journal directory. + let supervisor = supervise(runtime.clone()); + tokio::time::timeout(Duration::from_secs(30), async { + while root.stat("/var/log/journal", true).await.is_err() { + tokio::task::yield_now().await; + } + }) + .await + .unwrap(); + write(&root, "/var/log/journal/x.service.log", b"old\n").await; + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); + let executor = runtime.clone(); + let task = tokio::spawn(async move { + executor + .exec( + Command::Argv(vec![ + "/bin/journalctl.wasm".into(), + "-u".into(), + "x".into(), + "-f".into(), + ]), + Options { + cwd: "/".into(), + ..Default::default() + }, + &mut |_, bytes| { + let _ = tx.send(bytes.to_vec()); + }, + ) + .await + }); + let first = tokio::time::timeout(Duration::from_secs(30), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(first, b"old\n"); + let file = root + .open( + "/var/log/journal/x.service.log", + Open { + write: true, + ..Default::default() + }, + ) + .await + .unwrap(); + file.append(b"new\n".to_vec()).await.unwrap(); + let next = tokio::time::timeout(Duration::from_secs(30), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(next, b"new\n"); + task.abort(); + let _ = task.await; + supervisor.shutdown().await; +} diff --git a/host/tests/process.rs b/host/tests/process.rs index 3332f75..1a9d6d1 100644 --- a/host/tests/process.rs +++ b/host/tests/process.rs @@ -12,7 +12,7 @@ use std::{ mod features { //! Acceptance coverage for preview1, identity, inherited IO, services, HTTP, and bundle aliases. use super::*; - use std::{collections::BTreeMap, time::Instant}; + use std::collections::BTreeMap; async fn poll(mut condition: impl FnMut() -> bool) { tokio::time::timeout(Duration::from_secs(10), async { @@ -23,13 +23,6 @@ mod features { .await .unwrap(); } - fn lines(path: &Path) -> Vec { - std::fs::read_to_string(path) - .unwrap_or_default() - .lines() - .map(str::to_owned) - .collect() - } async fn prepare(runtime: &Arc, argv: Vec) { assert_eq!( runtime @@ -285,61 +278,6 @@ mod features { assert_eq!(runtime.compiled_programs(), 1); } #[tokio::test(flavor = "multi_thread")] - async fn services_restart_edit_limit_and_eof() { - let _serial = SERIAL.lock().await; - let (temp, runtime, processes) = world(); - prepare(&runtime, vec!["/bin/launcher".into()]).await; - let root = temp.path().join("world"); - std::fs::create_dir_all(root.join("config")).unwrap(); - let spec = |name: &str, mode: &str, log: &str| serde_json::json!({"name": name, "argv": ["/bin/launcher", mode, log]}); - let write = |specs: Vec| { - std::fs::write( - root.join("config/services.yaml"), - serde_yaml_ng::to_string(&specs).unwrap(), - ) - .unwrap() - }; - write(vec![ - spec("a", "server", "/a.log"), - spec("crash", "crash", "/crash.log"), - ]); - let supervisor = sausage::process::supervise(runtime.clone()); - poll(|| { - lines(&root.join("a.log")) == ["start"] && lines(&root.join("crash.log")).len() == 6 - }) - .await; - prepare(&runtime, vec!["/bin/launcher".into()]).await; - assert_eq!(lines(&root.join("crash.log")).len(), 6); - assert!( - processes - .lock() - .unwrap() - .values() - .all(|p| p.conversation.is_none()) - ); - write(vec![spec("b", "server", "/b.log")]); - poll(|| { - lines(&root.join("a.log")) == ["start", "stop"] - && lines(&root.join("b.log")) == ["start"] - }) - .await; - let mut changed = spec("b", "server", "/b.log"); - changed["env"] = serde_json::json!({"REV": "new"}); - write(vec![changed]); - poll(|| lines(&root.join("b.log")) == ["start", "stop", "start"]).await; - write(vec![]); - poll(|| lines(&root.join("b.log")) == ["start", "stop", "start", "stop"]).await; - write(vec![spec("linger", "linger", "/linger.log")]); - poll(|| lines(&root.join("linger.log")) == ["start"]).await; - let stop = Instant::now(); - write(vec![]); - poll(|| processes.lock().unwrap().is_empty()).await; - assert!(stop.elapsed() >= Duration::from_secs(5)); - supervisor.abort(); - let _ = supervisor.await; - empty(&processes).await; - } - #[tokio::test(flavor = "multi_thread")] async fn git_local_and_anonymous_https_clone() { let _serial = SERIAL.lock().await; let (temp, runtime, _) = world(); diff --git a/host/wit/process.wit b/host/wit/process.wit index c38a856..f993026 100644 --- a/host/wit/process.wit +++ b/host/wit/process.wit @@ -3,6 +3,14 @@ package sausage:process@0.1.0; /// Programs share one world. These resources provide lifecycle and bounded stdio transport. interface host { use wasi:io/streams@0.2.12.{input-stream, output-stream}; + use wasi:io/poll@0.2.12.{pollable}; + enum change-kind { created, modified, removed, overflow } + record change { path: string, kind: change-kind } + resource watcher { + constructor(paths: list); + subscribe: func() -> pollable; + changes: func() -> list; + } enum stdio { pipe, inherit, null } record request { program: string, @@ -21,6 +29,7 @@ interface host { stdin: func() -> output-stream; stdout: func() -> input-stream; stderr: func() -> input-stream; + subscribe: func() -> pollable; status: func() -> option; wait: func() -> s32; kill: func(); diff --git a/userland/Cargo.lock b/userland/Cargo.lock index 6840a5c..f06325f 100644 --- a/userland/Cargo.lock +++ b/userland/Cargo.lock @@ -4045,6 +4045,17 @@ dependencies = [ "walkdir", ] +[[package]] +name = "sausage-init" +version = "0.1.0" +dependencies = [ + "sausage-spawn", + "serde", + "serde_json", + "shlex 1.3.0", + "wasi 0.14.7+wasi-0.2.4", +] + [[package]] name = "sausage-procps" version = "0.1.0" diff --git a/userland/Cargo.toml b/userland/Cargo.toml index 271c59c..b774fb9 100644 --- a/userland/Cargo.toml +++ b/userland/Cargo.toml @@ -1,7 +1,7 @@ # The default CLIs of a world, built for wasm32-wasip2. Its own workspace, outside the host's. [workspace] resolver = "3" -members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools"] +members = ["git", "coreutils", "procps", "cwd-init/probe", "ripgrep", "diffutils", "jaq", "sed", "grep", "fd", "spawn", "findutils", "exec-tools", "init"] # WASI compatibility patches: git/vendor/PATCHES.md. [patch.crates-io] diff --git a/userland/init/Cargo.toml b/userland/init/Cargo.toml new file mode 100644 index 0000000..971e321 --- /dev/null +++ b/userland/init/Cargo.toml @@ -0,0 +1,31 @@ +[package] +name = "sausage-init" +version = "0.1.0" +edition = "2024" + +[dependencies] +sausage-spawn = { path = "../spawn" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" +wasi = "0.14" +shlex = "1" + +[[bin]] +name = "init-probe" +path = "src/probe.rs" + +[[bin]] +name = "init" +path = "src/main.rs" + +[[bin]] +name = "systemctl" +path = "src/systemctl.rs" + +[[bin]] +name = "journalctl" +path = "src/journalctl.rs" + +[[bin]] +name = "systemd-run" +path = "src/systemd-run.rs" diff --git a/userland/init/README.md b/userland/init/README.md new file mode 100644 index 0000000..be2a090 --- /dev/null +++ b/userland/init/README.md @@ -0,0 +1,75 @@ +# World init + +`/bin/init.wasm` owns unit policy. The host starts init without a conversation. +The host restarts init after an exit and kills its descendants before the next start. + +Init reads persistent units from `/etc/systemd/system/*.service`. +It reads transient units from `/run/systemd/transient/*.service`. +The host mounts `/run` in memory, so transient units disappear after a host restart. + +Unit files use `[Unit]`, `[Service]`, and `[Install]` sections. +Init supports dependency ordering, simple and oneshot services, environment variables, and restart policies. +The default start limit permits five starts within ten seconds. +A stop closes stdin. Init kills the process group after `TimeoutStopSec`, which defaults to ten seconds. +When init receives EOF, it stops dependents before prerequisites. + +Enable records are files under `default.target.wants` or `multi-user.target.wants` in the persistent unit directory. +The `[Install]` section supplies `WantedBy` for `systemctl enable`. +Init observes VFS changes and applies unit edits without `daemon-reload`. +A replacement of a running unit's executable restarts that unit. + +## Control protocol + +The protocol uses UTF-8 JSON files. Writers create a temporary file, then rename it to the final path. +Each request uses a unique identifier. + +1. The client watches `/run/systemd/replies/` before it submits the request. +2. The client writes `/run/systemd/requests/.json`. +3. Init reads the request and applies the action. +4. Init writes `/run/systemd/replies/` and removes the request. +5. The client reads and removes the reply. + +Request example: + +```json +{"action":"start","unit":"example.service"} +``` + +Actions are `start`, `stop`, and `restart`. +A reply acknowledges the action. It does not imply that the service completed its start or stop. +The control tools then wait for the requested state. A failed start returns code 1. + +Reply example: + +```json +{"code":0,"text":""} +``` + +A missing unit returns code 5. An unknown action returns code 1. +Clients read `/run/systemd/state/` for the current state: + +```json +{"state":"active","pid":42,"code":null} +``` + +States are `inactive`, `activating`, `active`, `deactivating`, and `failed`. +The `code` field records the most recent exit status. +`systemctl is-active` returns 3 for an inactive unit. +`systemctl is-failed` returns 0 for a failed unit and 1 otherwise. +`daemon-reload` returns 0 without an action. + +## Journal + +Init drains each unit's stdout and stderr into `/var/log/journal/.log`. +The cap is 1 MiB. Init keeps one rotated file with the `.1` suffix. +`journalctl -u -n ` prints the last lines. +`journalctl -f` observes the journal file and prints appended bytes. + +## Process observation + +`sausage-spawn` exposes `Watcher`, `ChangeKind`, and child exit pollables. +A watcher matches each supplied path and all its descendants. +Its pollable remains ready while changes remain pending. +`changes()` returns and clears the pending changes. +The queue holds at most 256 changes. An `overflow` event requires a rescan. +Observers report successful VFS mutations, not direct host filesystem writes outside the VFS. diff --git a/userland/init/src/journalctl.rs b/userland/init/src/journalctl.rs new file mode 100644 index 0000000..c2c5c64 --- /dev/null +++ b/userland/init/src/journalctl.rs @@ -0,0 +1,66 @@ +//! Reads a unit journal, with optional live output through the world watcher. +use sausage_init::name; +use std::{ + fs, + io::{self, Write}, +}; +fn run() -> io::Result<()> { + let mut args = std::env::args().skip(1); + let (mut unit, mut lines, mut follow) = (String::new(), 10usize, false); + while let Some(arg) = args.next() { + match arg.as_str() { + "-u" => { + unit = name( + &args + .next() + .ok_or_else(|| io::Error::other("missing unit"))?, + ) + } + "-n" => { + lines = args + .next() + .and_then(|n| n.parse().ok()) + .ok_or_else(|| io::Error::other("missing count"))? + } + "-f" => follow = true, + "--no-pager" => (), + _ => return Err(io::Error::other("unknown argument")), + } + } + let path = format!("/var/log/journal/{unit}.log"); + let watcher = sausage_spawn::Watcher::new(&[path.clone()]); + let bytes = fs::read(&path).unwrap_or_default(); + let mut starts = vec![0]; + for (i, b) in bytes.iter().enumerate() { + if *b == b'\n' && i + 1 < bytes.len() { + starts.push(i + 1); + } + } + let start = if lines == 0 { + bytes.len() + } else { + starts[starts.len().saturating_sub(lines)] + }; + io::stdout().write_all(&bytes[start..])?; + io::stdout().flush()?; + let mut offset = bytes.len(); + while follow { + let pollable = watcher.subscribe(); + pollable.block(); + drop(pollable); + watcher.changes(); + let bytes = fs::read(&path).unwrap_or_default(); + if bytes.len() < offset { + offset = 0; + } + io::stdout().write_all(&bytes[offset..])?; + io::stdout().flush()?; + offset = bytes.len(); + } + Ok(()) +} +fn main() { + if run().is_err() { + std::process::exit(1); + } +} diff --git a/userland/init/src/lib.rs b/userland/init/src/lib.rs new file mode 100644 index 0000000..8c53313 --- /dev/null +++ b/userland/init/src/lib.rs @@ -0,0 +1,188 @@ +//! Unit definitions and the file protocol shared by init and its control tools. +use serde::{Deserialize, Serialize}; +use std::{collections::BTreeMap, fs, io, time::Duration}; +/// WASI 0.2's legacy exit call loses numeric error codes. Use the standard extended call. +pub fn exit(code: i32) -> ! { + wasi::cli::exit::exit_with_code(code as u8); + unreachable!("exit returned") +} +pub const RUN: &str = "/run/systemd"; +pub const UNITS: &str = "/etc/systemd/system"; +pub const TRANSIENT: &str = "/run/systemd/transient"; +#[derive(Clone, Debug, PartialEq)] +pub struct Unit { + pub values: BTreeMap, + pub source: String, +} +impl Unit { + pub fn parse(source: String) -> io::Result { + let mut values = BTreeMap::new(); + let mut section = ""; + for line in source.lines().map(str::trim) { + if line.is_empty() || line.starts_with(['#', ';']) { + continue; + } + if line.starts_with('[') && line.ends_with(']') { + section = &line[1..line.len() - 1]; + continue; + } + let (key, value) = line + .split_once('=') + .ok_or_else(|| io::Error::other("expected Key=Value"))?; + let key = format!("{section}.{}", key.trim()); + if matches!( + key.as_str(), + "Unit.After" + | "Unit.Before" + | "Unit.Requires" + | "Unit.Wants" + | "Service.Environment" + ) { + values + .entry(key) + .and_modify(|v: &mut String| { + v.push(' '); + v.push_str(value.trim()); + }) + .or_insert(value.trim().into()); + } else { + values.insert(key, value.trim().into()); + } + } + if !values.contains_key("Service.ExecStart") { + return Err(io::Error::other("missing ExecStart")); + } + Ok(Self { values, source }) + } + pub fn get(&self, key: &str) -> &str { + self.values.get(key).map(String::as_str).unwrap_or("") + } + pub fn words(&self, key: &str) -> Vec { + self.get(key).split_whitespace().map(name).collect() + } + pub fn seconds(&self, key: &str, default: f64) -> Duration { + Duration::from_secs_f64( + self.get(key) + .trim_end_matches('s') + .parse::() + .ok() + .filter(|n| n.is_finite() && *n >= 0.0) + .unwrap_or(default), + ) + } +} +pub fn name(s: &str) -> String { + if s.ends_with(".service") || s.ends_with(".target") { + s.into() + } else { + format!("{s}.service") + } +} +pub fn unit_path(name: &str) -> String { + let persistent = format!("{UNITS}/{name}"); + if std::path::Path::new(&persistent).exists() { + persistent + } else { + format!("{TRANSIENT}/{name}") + } +} +pub fn enabled(name: &str) -> bool { + ["default.target", "multi-user.target"] + .iter() + .any(|target| std::path::Path::new(&format!("{UNITS}/{target}.wants/{name}")).exists()) +} +#[derive(Serialize, Deserialize, Default)] +pub struct Status { + pub state: String, + pub pid: Option, + pub code: Option, +} +#[derive(Serialize, Deserialize)] +pub struct Request { + pub action: String, + pub unit: String, +} +#[derive(Serialize, Deserialize)] +pub struct Reply { + pub code: i32, + pub text: String, +} +pub fn atomic(path: &str, bytes: &[u8]) -> io::Result<()> { + let tmp = format!("{path}.tmp"); + fs::write(&tmp, bytes)?; + fs::rename(tmp, path) +} +pub fn request(action: &str, unit: &str) -> io::Result { + let id = format!( + "{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let reply = format!("{RUN}/replies/{id}"); + let state = format!("{RUN}/state/{}", name(unit)); + let watcher = sausage_spawn::Watcher::new(&[reply.clone(), state.clone()]); + atomic( + &format!("{RUN}/requests/{id}.json"), + &serde_json::to_vec(&Request { + action: action.into(), + unit: name(unit), + })?, + )?; + let deadline = std::time::Instant::now() + Duration::from_secs(60); + let mut acknowledged = false; + loop { + if !acknowledged && let Ok(bytes) = fs::read(&reply) { + fs::remove_file(&reply)?; + let reply: Reply = serde_json::from_slice(&bytes)?; + if reply.code != 0 { + return Ok(reply); + } + acknowledged = true; + } + if acknowledged && let Ok(bytes) = fs::read(&state) { + let status: Status = serde_json::from_slice(&bytes)?; + let done = if action == "stop" { + status.state == "inactive" + } else { + status.state == "active" + || status.state == "failed" + || status.state == "inactive" && status.code.is_some() + }; + if done { + return Ok(Reply { + code: if action != "stop" && status.state == "failed" { + 1 + } else { + 0 + }, + text: String::new(), + }); + } + } + if std::time::Instant::now() >= deadline { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "init did not reply", + )); + } + // The short timer also handles a missing or failed init. + let _ = watcher.changes(); + std::thread::sleep(Duration::from_millis(20)); + } +} +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn unit_sections_and_defaults() { + let unit = Unit::parse("[Unit]\nAfter=prepare\nRequires=prepare\n[Service]\nType=oneshot\nExecStart=/bin/task.wasm\n".into()).unwrap(); + assert_eq!(unit.words("Unit.Requires"), ["prepare.service"]); + assert_eq!( + unit.seconds("Service.TimeoutStopSec", 10.), + Duration::from_secs(10) + ); + assert!(Unit::parse("[Unit]\nDescription=missing command".into()).is_err()); + } +} diff --git a/userland/init/src/main.rs b/userland/init/src/main.rs new file mode 100644 index 0000000..7a49b80 --- /dev/null +++ b/userland/init/src/main.rs @@ -0,0 +1,421 @@ +//! Runs the world's unit state machine, journal drains, and file-based control requests. +use sausage_init::*; +use sausage_spawn::{Child, Command, Stdio}; +use std::{ + collections::{BTreeMap, BTreeSet, VecDeque}, + fs, + io::{self, Write}, + time::{Duration, Instant}, +}; +struct Service { + unit: Unit, + child: Option, + state: String, + code: Option, + desired: bool, + starts: VecDeque, + next: Instant, + stopping: Option, +} +impl Service { + fn new(unit: Unit) -> Self { + Self { + unit, + child: None, + state: "inactive".into(), + code: None, + desired: false, + starts: VecDeque::new(), + next: Instant::now(), + stopping: None, + } + } + fn stop(&mut self) { + self.desired = false; + if let Some(child) = &mut self.child { + child.stdin.take(); + self.stopping + .get_or_insert(Instant::now() + self.unit.seconds("Service.TimeoutStopSec", 10.)); + self.state = "deactivating".into(); + } else { + self.state = "inactive".into(); + } + } + fn start(&mut self) { + let now = Instant::now(); + let interval = self.unit.seconds("Service.StartLimitIntervalSec", 10.); + while self + .starts + .front() + .is_some_and(|t| now.duration_since(*t) >= interval) + { + self.starts.pop_front(); + } + let burst = self + .unit + .get("Service.StartLimitBurst") + .parse() + .unwrap_or(5); + if burst != 0 && self.starts.len() >= burst { + self.state = "failed".into(); + self.desired = false; + return; + } + self.starts.push_back(now); + let Some(argv) = shlex::split(self.unit.get("Service.ExecStart")).filter(|a| !a.is_empty()) + else { + self.state = "failed".into(); + self.desired = false; + return; + }; + let mut command = Command::new(&argv[0]); + command + .args(&argv[1..]) + .new_group() + .stdin(Stdio::Pipe) + .stdout(Stdio::Pipe) + .stderr(Stdio::Pipe); + let cwd = self.unit.get("Service.WorkingDirectory"); + command.current_dir(if cwd.is_empty() { "/" } else { cwd }); + for assignment in shlex::split(self.unit.get("Service.Environment")).unwrap_or_default() { + if let Some((key, value)) = assignment.split_once('=') { + command.env(key, value); + } + } + match command.spawn() { + Ok(child) => { + self.child = Some(child); + self.state = if self.unit.get("Service.Type") == "oneshot" { + "activating" + } else { + "active" + } + .into(); + self.code = None; + } + Err(_) => { + self.code = Some(127); + self.exited(127); + } + } + } + fn exited(&mut self, code: i32) { + self.code = Some(code); + if self.stopping.take().is_some() { + self.state = if self.desired { + "activating" + } else { + "inactive" + } + .into(); + if self.desired { + self.code = None; + } + return; + } + self.state = if code != 0 { + "failed" + } else if matches!( + self.unit.get("Service.RemainAfterExit"), + "yes" | "true" | "1" + ) { + "active" + } else { + "inactive" + } + .into(); + let restart = self.unit.get("Service.Restart"); + if restart == "always" || restart == "on-failure" && code != 0 { + self.next = Instant::now() + self.unit.seconds("Service.RestartSec", 0.1); + } else { + self.desired = false; + } + } + fn tick(&mut self, name: &str) { + if let Some(child) = &mut self.child { + for slot in [&mut child.stdout, &mut child.stderr] { + if let Some(stream) = slot { + loop { + match stream.read(8192) { + Ok(bytes) if !bytes.is_empty() => journal(name, &bytes), + Ok(_) => break, + Err(_) => { + *slot = None; + break; + } + } + } + } + } + if child.try_wait().is_none() && self.stopping.is_some_and(|t| Instant::now() >= t) { + let _ = sausage_spawn::kill(child.id()); + } + if let Some(code) = child.try_wait() { + if child.stdout.is_none() && child.stderr.is_none() { + self.child = None; + self.exited(code); + } + } + } + } + fn publish(&self, name: &str) { + let status = Status { + state: self.state.clone(), + pid: self.child.as_ref().map(Child::id), + code: self.code, + }; + let bytes = serde_json::to_vec(&status).unwrap(); + let path = format!("{RUN}/state/{name}"); + if fs::read(&path).ok().as_deref() != Some(bytes.as_slice()) { + let _ = atomic(&path, &bytes); + } + } +} +fn journal(name: &str, bytes: &[u8]) { + let path = format!("/var/log/journal/{name}.log"); + if fs::metadata(&path).is_ok_and(|m| m.len() + bytes.len() as u64 > 1024 * 1024) { + let _ = fs::rename(&path, format!("{path}.1")); + } + if let Ok(mut f) = fs::OpenOptions::new().create(true).append(true).open(path) { + let _ = f.write_all(bytes); + } +} +fn activate(name: &str, services: &mut BTreeMap, seen: &mut BTreeSet) { + if !seen.insert(name.into()) { + return; + } + let Some(service) = services.get(name) else { + return; + }; + let deps = [ + service.unit.words("Unit.Requires"), + service.unit.words("Unit.Wants"), + ] + .concat(); + for dep in deps { + activate(&dep, services, seen); + } + let service = services.get_mut(name).unwrap(); + service.desired = true; + if service.child.is_none() { + service.state = "activating".into(); + service.code = None; + } + service.next = Instant::now(); +} +fn predecessors(name: &str, services: &BTreeMap) -> Vec { + let mut deps = services[name].unit.words("Unit.After"); + deps.extend(services[name].unit.words("Unit.Requires")); + for (other, service) in services { + if service.unit.words("Unit.Before").iter().any(|s| s == name) { + deps.push(other.clone()); + } + } + deps +} +fn scan(services: &mut BTreeMap) { + let mut found = BTreeMap::new(); + let mut present = BTreeSet::new(); + for dir in [TRANSIENT, UNITS] { + for entry in fs::read_dir(dir).into_iter().flatten().flatten() { + let name = entry.file_name().to_string_lossy().into_owned(); + if !name.ends_with(".service") { + continue; + } + present.insert(name.clone()); + if let Ok(source) = fs::read_to_string(entry.path()) { + if let Ok(unit) = Unit::parse(source) { + found.insert(name, unit); + } + } + } + } + for (name, service) in services.iter_mut() { + if !present.contains(name) { + service.stop(); + } + } + services.retain(|name, service| present.contains(name) || service.child.is_some()); + let mut new_enabled = Vec::new(); + for (name, unit) in found { + if !services.contains_key(&name) && enabled(&name) { + new_enabled.push(name.clone()); + } + let service = services + .entry(name.clone()) + .or_insert_with(|| Service::new(unit.clone())); + if service.unit != unit { + let desired = service.desired || service.state == "active"; + service.stop(); + service.unit = unit; + service.desired = desired; + service.starts.clear(); + service.next = Instant::now(); + } + } + for name in new_enabled { + activate(&name, services, &mut BTreeSet::new()); + } +} +fn controls(services: &mut BTreeMap) { + for entry in fs::read_dir(format!("{RUN}/requests")) + .into_iter() + .flatten() + .flatten() + { + let filename = entry.file_name().to_string_lossy().into_owned(); + if !filename.ends_with(".json") { + continue; + } + let Ok(bytes) = fs::read(entry.path()) else { + continue; + }; + let Ok(request) = serde_json::from_slice::(&bytes) else { + continue; + }; + let mut reply = Reply { + code: 0, + text: String::new(), + }; + if !services.contains_key(&request.unit) { + reply.code = 5; + reply.text = "unit not found".into(); + } else { + match request.action.as_str() { + "start" => { + services.get_mut(&request.unit).unwrap().starts.clear(); + activate(&request.unit, services, &mut BTreeSet::new()); + } + "stop" => services.get_mut(&request.unit).unwrap().stop(), + "restart" => { + let service = services.get_mut(&request.unit).unwrap(); + service.stop(); + service.desired = true; + if service.child.is_none() { + service.state = "activating".into(); + service.code = None; + } + service.starts.clear(); + service.next = Instant::now(); + } + _ => { + reply.code = 1; + reply.text = "unknown action".into(); + } + } + } + if let Some(service) = services.get(&request.unit) { + service.publish(&request.unit); + } + let id = filename.trim_end_matches(".json"); + let _ = atomic( + &format!("{RUN}/replies/{id}"), + &serde_json::to_vec(&reply).unwrap(), + ); + let _ = fs::remove_file(entry.path()); + } +} +fn main() -> io::Result<()> { + for dir in [ + UNITS, + TRANSIENT, + "/run/systemd/requests", + "/run/systemd/replies", + "/run/systemd/state", + "/var/log/journal", + ] { + fs::create_dir_all(dir)?; + } + let watcher = sausage_spawn::Watcher::new(&["/".into()]); + let stdin = wasi::cli::stdin::get_stdin(); + let mut services = BTreeMap::new(); + scan(&mut services); + loop { + if stdin.read(1).is_err() { + break; + } + let changes = watcher.changes(); + if !changes.is_empty() { + for service in services.values_mut() { + let program = shlex::split(service.unit.get("Service.ExecStart")) + .and_then(|a| a.first().cloned()) + .unwrap_or_default(); + if service.child.is_some() + && changes.iter().any(|c| { + c.path == program || matches!(c.kind, sausage_spawn::ChangeKind::Overflow) + }) + { + service.stop(); + service.desired = true; + service.next = Instant::now(); + } + } + scan(&mut services); + } + controls(&mut services); + let names: Vec<_> = services.keys().cloned().collect(); + for name in &names { + services.get_mut(name).unwrap().tick(name); + } + for name in &names { + let service = &services[name]; + if !service.desired + || service.child.is_some() + || Instant::now() < service.next + || service.stopping.is_some() + { + continue; + } + let required = service.unit.words("Unit.Requires"); + if required + .iter() + .any(|dep| !services.contains_key(dep) || services[dep].state == "failed") + { + let service = services.get_mut(name).unwrap(); + service.state = "failed".into(); + service.desired = false; + continue; + } + if predecessors(name, &services).iter().any(|dep| { + services.get(dep).is_some_and(|s| { + s.state == "activating" || s.desired && s.child.is_none() && s.code.is_none() + }) + }) { + continue; + } + services.get_mut(name).unwrap().start(); + } + for (name, service) in &services { + service.publish(name); + } + std::thread::sleep(Duration::from_millis(20)); + } + // Stop dependents before their prerequisites. A cycle stops together. + while services.values().any(|s| s.child.is_some()) { + let live: Vec<_> = services + .iter() + .filter(|(_, s)| s.child.is_some()) + .map(|(n, _)| n.clone()) + .collect(); + let mut leaves: Vec<_> = live + .iter() + .filter(|name| { + !live + .iter() + .any(|other| other != *name && predecessors(other, &services).contains(name)) + }) + .cloned() + .collect(); + if leaves.is_empty() { + leaves = live; + } + for name in leaves { + let service = services.get_mut(&name).unwrap(); + service.stop(); + service.tick(&name); + service.publish(&name); + } + std::thread::sleep(Duration::from_millis(20)); + } + Ok(()) +} diff --git a/userland/init/src/probe.rs b/userland/init/src/probe.rs new file mode 100644 index 0000000..bba6356 --- /dev/null +++ b/userland/init/src/probe.rs @@ -0,0 +1,96 @@ +//! Acceptance guest for unit lifecycle and process-interface readiness tests. +use std::{ + fs, + io::{Read, Write}, +}; +fn main() { + let args: Vec<_> = std::env::args().collect(); + match args.get(1).map(String::as_str).unwrap_or("") { + "kill-init" => { + let table = fs::read_to_string("/proc/processes").unwrap(); + let row = table + .lines() + .find(|line| line.ends_with("/bin/init.wasm")) + .unwrap(); + sausage_spawn::kill(row.split('\t').next().unwrap().parse().unwrap()).unwrap(); + } + "watch" => { + use sausage_spawn::{ChangeKind, Watcher}; + let watcher = Watcher::new(&["/run/watched".into()]); + let poll = watcher.subscribe(); + assert!(!poll.ready()); + fs::create_dir_all("/run/watched").unwrap(); + poll.block(); + assert!( + watcher + .changes() + .iter() + .any(|c| matches!(c.kind, ChangeKind::Created)) + ); + fs::write("/run/watched/a", "a").unwrap(); + watcher.changes(); + fs::write("/run/watched/a", "b").unwrap(); + assert!( + watcher + .changes() + .iter() + .any(|c| c.path == "/run/watched/a" && matches!(c.kind, ChangeKind::Modified)) + ); + fs::remove_file("/run/watched/a").unwrap(); + assert!( + watcher + .changes() + .iter() + .any(|c| matches!(c.kind, ChangeKind::Removed)) + ); + for i in 0..300 { + fs::write(format!("/run/watched/{i}"), "").unwrap(); + } + assert!( + watcher + .changes() + .iter() + .any(|c| matches!(c.kind, ChangeKind::Overflow)) + ); + drop(poll); + let mut cmd = sausage_spawn::Command::new("/bin/probe.wasm"); + cmd.arg("exit").arg("7"); + let child = cmd.spawn().unwrap(); + let exit = child.subscribe(); + exit.block(); + assert_eq!(child.try_wait(), Some(7)); + } + "exit" => sausage_init::exit(args.get(2).and_then(|a| a.parse().ok()).unwrap_or(0)), + "oneshot" | "server" | "crash" | "after" | "linger" => { + if args[1] == "after" { + assert!( + !fs::read_to_string("/proc/processes") + .unwrap() + .contains("oneshot /order.log") + ); + } + let mut log = fs::OpenOptions::new() + .create(true) + .append(true) + .open(&args[2]) + .unwrap(); + writeln!(log, "start").unwrap(); + for n in 1..=10 { + println!("line {n}"); + } + if args[1] == "linger" { + loop { + std::thread::sleep(std::time::Duration::from_secs(1)); + } + } + if args[1] == "crash" { + std::process::exit(1); + } + if args[1] == "server" || args[1] == "after" { + let _ = std::io::stdin().read_to_end(&mut Vec::new()); + writeln!(log, "stop").unwrap(); + } + } + _ => panic!("unknown probe mode"), + } +} diff --git a/userland/init/src/systemctl.rs b/userland/init/src/systemctl.rs new file mode 100644 index 0000000..2a44a3c --- /dev/null +++ b/userland/init/src/systemctl.rs @@ -0,0 +1,123 @@ +//! Controls init and reads its public unit state files. +use sausage_init::*; +use std::{fs, io}; +fn run() -> io::Result { + let args: Vec<_> = std::env::args().skip(1).collect(); + let action = args.first().map(String::as_str).unwrap_or("list-units"); + if action == "daemon-reload" { + return Ok(0); + } + if action == "list-units" { + for entry in fs::read_dir(format!("{RUN}/state"))? { + let entry = entry?; + let status: Status = serde_json::from_slice(&fs::read(entry.path())?)?; + println!("{} {}", entry.file_name().to_string_lossy(), status.state); + } + return Ok(0); + } + let now = args.iter().any(|a| a == "--now"); + let units: Vec<_> = args + .iter() + .skip(1) + .filter(|a| !a.starts_with('-')) + .map(|a| name(a)) + .collect(); + if units.is_empty() { + return Ok(1); + } + let mut result = 0; + for unit in units { + let status = fs::read(format!("{RUN}/state/{unit}")) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + .unwrap_or_default(); + let code = match action { + "is-active" => { + println!( + "{}", + if status.state.is_empty() { + "unknown" + } else { + &status.state + } + ); + if status.state == "active" { 0 } else { 3 } + } + "is-failed" => { + println!( + "{}", + if status.state.is_empty() { + "unknown" + } else { + &status.state + } + ); + if status.state == "failed" { 0 } else { 1 } + } + "is-enabled" => { + println!( + "{}", + if enabled(&unit) { + "enabled" + } else { + "disabled" + } + ); + if enabled(&unit) { 0 } else { 1 } + } + "status" => { + println!("{unit}: {}", status.state); + if status.state == "active" { + 0 + } else if status.state.is_empty() { + 4 + } else { + 3 + } + } + "cat" => { + print!("{}", fs::read_to_string(unit_path(&unit))?); + 0 + } + "enable" | "disable" => { + let spec = Unit::parse(fs::read_to_string(unit_path(&unit))?)?; + let targets: Vec<_> = spec + .get("Install.WantedBy") + .split_whitespace() + .filter(|t| matches!(*t, "default.target" | "multi-user.target")) + .collect(); + if action == "enable" && targets.is_empty() { + return Ok(1); + } + for target in ["default.target", "multi-user.target"] { + let dir = format!("{UNITS}/{target}.wants"); + let path = format!("{dir}/{unit}"); + if action == "enable" && targets.contains(&target) { + fs::create_dir_all(dir)?; + fs::write(path, b"")?; + } else if action == "disable" { + let _ = fs::remove_file(path); + } + } + if now { + request(if action == "enable" { "start" } else { "stop" }, &unit)?.code + } else { + 0 + } + } + "start" | "stop" | "restart" => { + let reply = request(action, &unit)?; + if !reply.text.is_empty() { + println!("{}", reply.text); + } + reply.code + } + _ => 1, + }; + result = result.max(code); + } + Ok(result) +} +fn main() { + sausage_init::exit(run().unwrap_or(1)); +} diff --git a/userland/init/src/systemd-run.rs b/userland/init/src/systemd-run.rs new file mode 100644 index 0000000..968face --- /dev/null +++ b/userland/init/src/systemd-run.rs @@ -0,0 +1,82 @@ +//! Creates transient unit files and optionally waits for their final state. +use sausage_init::*; +use std::{fs, io, time::Duration}; +fn run() -> io::Result { + let mut args = std::env::args().skip(1).peekable(); + let mut unit = format!( + "run-{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + let (mut wait, mut properties, mut command) = (false, Vec::new(), Vec::new()); + while let Some(arg) = args.next() { + match arg.as_str() { + "--unit" => { + unit = args + .next() + .ok_or_else(|| io::Error::other("missing unit"))? + } + "--wait" => wait = true, + "-p" => properties.push( + args.next() + .ok_or_else(|| io::Error::other("missing property"))?, + ), + "--working-directory" => properties.push(format!( + "WorkingDirectory={}", + args.next() + .ok_or_else(|| io::Error::other("missing directory"))? + )), + "-E" => properties.push(format!( + "Environment={}", + args.next() + .ok_or_else(|| io::Error::other("missing environment"))? + )), + "--" => { + command.extend(args); + break; + } + _ if arg.starts_with("--unit=") => unit = arg[7..].into(), + _ if arg.starts_with('-') => return Ok(1), + _ => { + command.push(arg); + command.extend(args); + break; + } + } + } + if command.is_empty() { + return Ok(1); + } + let unit = name(&unit); + let mut text = format!( + "[Service]\nExecStart={}\n", + shlex::try_join(command.iter().map(String::as_str)).map_err(io::Error::other)? + ); + for property in properties { + text.push_str(&property); + text.push('\n'); + } + fs::create_dir_all(TRANSIENT)?; + atomic(&format!("{TRANSIENT}/{unit}"), text.as_bytes())?; + let reply = request("start", &unit)?; + if reply.code != 0 { + return Ok(reply.code); + } + if wait { + loop { + if let Ok(bytes) = fs::read(format!("{RUN}/state/{unit}")) { + let status: Status = serde_json::from_slice(&bytes)?; + if let Some(code) = status.code { + return Ok(code); + } + } + std::thread::sleep(Duration::from_millis(20)); + } + } + Ok(0) +} +fn main() { + sausage_init::exit(run().unwrap_or(1)); +} diff --git a/userland/spawn/src/lib.rs b/userland/spawn/src/lib.rs index 004ee87..a8b4020 100644 --- a/userland/spawn/src/lib.rs +++ b/userland/spawn/src/lib.rs @@ -7,7 +7,10 @@ mod bindings { } use bindings::sausage::process::host; pub use bindings::wasi::io::streams::{InputStream, OutputStream}; -pub use host::Stdio; +pub use host::{Stdio, Watcher, Change, ChangeKind}; +pub use bindings::wasi::io::poll::{Pollable, poll}; +/// Hard-stop a process group by a live process ID. +pub fn kill(pid: u32) -> io::Result<()> { host::kill(pid).map_err(io::Error::other) } /// A prepared child process. pub struct Command { @@ -63,6 +66,7 @@ pub struct Child { } impl Child { pub fn id(&self) -> u32 { self.child.pid() } + pub fn subscribe(&self) -> Pollable { self.child.subscribe() } pub fn try_wait(&self) -> Option { self.child.status() } pub fn wait(&self) -> i32 { self.child.wait() } /// Stop every process in this child's group. diff --git a/vfs/src/lib.rs b/vfs/src/lib.rs index 03d816c..0902c87 100644 --- a/vfs/src/lib.rs +++ b/vfs/src/lib.rs @@ -67,6 +67,10 @@ pub trait File: Send + Sync { /// Implementations must retain open file identity and return portable IO errors. #[async_trait] pub trait Layer: Send + Sync { + /// Shared observation of mutations in this namespace, when available. + fn observer(&self) -> Option { + None + } async fn stat(&self, path: &str, follow: bool) -> Result; /// Return the target of a symbolic link, or InvalidInput for another node. async fn readlink(&self, _path: &str) -> Result { diff --git a/vfs/src/mounts.rs b/vfs/src/mounts.rs index c31207c..d062cc9 100644 --- a/vfs/src/mounts.rs +++ b/vfs/src/mounts.rs @@ -6,15 +6,24 @@ struct Mount { readonly: bool, } /// A per-process mount namespace. Mount the root before use. -#[derive(Default)] pub struct Mounts { mounts: BTreeMap, + observer: observe::Observer, +} +impl Default for Mounts { + fn default() -> Self { + Self { + mounts: BTreeMap::new(), + observer: observe::Observer::new(|_| {}), + } + } } impl Mounts { /// Add or replace a mount. Read-only mounts reject every mutation. pub fn mount(&mut self, path: &str, layer: Arc, readonly: bool) { - self.mounts - .insert(resolve("/", path), Mount { layer, readonly }); + let path = resolve("/", path); + let layer = self.observer.layer(layer, &path, Process::default()); + self.mounts.insert(path, Mount { layer, readonly }); } pub(crate) async fn metadata_file(&self, path: &str, follow: bool) -> Result> { let (m, p) = self.route(path, true)?; @@ -49,6 +58,9 @@ impl Mounts { } #[async_trait] impl Layer for Mounts { + fn observer(&self) -> Option { + Some(self.observer.clone()) + } async fn stat(&self, path: &str, follow: bool) -> Result { let (m, p) = self.route(path, false)?; m.layer.stat(&p, follow).await diff --git a/vfs/src/observe.rs b/vfs/src/observe.rs index e7f67d0..1738164 100644 --- a/vfs/src/observe.rs +++ b/vfs/src/observe.rs @@ -6,6 +6,14 @@ use std::{ }; use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock}; +/// The successful namespace or content mutation observed at a world path. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ChangeKind { + Created, + Modified, + Removed, +} + type Namespace = Mutex>>>; enum Held { Read { _guard: OwnedRwLockReadGuard<()> }, @@ -20,6 +28,7 @@ type Callback = dyn Fn(&str) + Send + Sync; #[derive(Clone)] pub struct Observer { changed: Arc, + events: tokio::sync::broadcast::Sender<(String, ChangeKind)>, paths: Arc>>>, // Only open-registration versus rename/remove needs ordering. Lock the affected // subtrees, not all IO: a blocked open must not stall unrelated namespaces. @@ -29,10 +38,19 @@ impl Observer { pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self { Self { changed: Arc::new(changed), + events: tokio::sync::broadcast::channel(256).0, paths: Arc::default(), namespace: Arc::default(), } } + /// Subscribe before the first mutation of interest. Lag means callers must rescan. + pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> { + self.events.subscribe() + } + fn notify(&self, path: &str, kind: ChangeKind) { + (self.changed)(path); + let _ = self.events.send((path.to_owned(), kind)); + } /// Wrap a layer at its canonical namespace prefix (e.g. `/run`). pub fn layer(&self, layer: Arc, prefix: &str, process: Process) -> Arc { Arc::new(Observed { @@ -149,7 +167,7 @@ impl Handle { let result = f.await?; let path = path.lock().unwrap().clone(); if let Some(path) = path { - (observer.changed)(&path); + observer.notify(&path, ChangeKind::Modified); } Ok(result) }) @@ -178,6 +196,7 @@ impl Layer for Observed { self.process .operation(async move { let order = observer.order(&[(&path, false)]).await; + let existed = !options.create || layer.stat(&local, options.follow).await.is_ok(); let file = layer.open(&local, options).await?; let handle = Arc::new(Handle { file, @@ -187,7 +206,14 @@ impl Layer for Observed { }); drop(order); if options.create || options.truncate { - (observer.changed)(&path); + observer.notify( + &path, + if existed { + ChangeKind::Modified + } else { + ChangeKind::Created + }, + ); } Ok(handle as Arc) }) @@ -203,7 +229,7 @@ impl Layer for Observed { self.process .operation(async move { layer.mkdir(&local).await?; - (observer.changed)(&path); + observer.notify(&path, ChangeKind::Created); Ok(()) }) .await @@ -226,13 +252,22 @@ impl Layer for Observed { self.process .operation(async move { let order = observer.order(&[(&from, true), (&to, true)]).await; + let replaced = layer.stat(&b, false).await.is_ok(); layer.rename(&a, &b).await?; - if from != to { - observer.moved(&from, Some(&to)); + if from == to { + return Ok(()); } + observer.moved(&from, Some(&to)); drop(order); - (observer.changed)(&from); - (observer.changed)(&to); + observer.notify(&from, ChangeKind::Removed); + observer.notify( + &to, + if replaced { + ChangeKind::Modified + } else { + ChangeKind::Created + }, + ); Ok(()) }) .await @@ -256,7 +291,7 @@ impl Observed { } observer.moved(&path, None); drop(order); - (observer.changed)(&path); + observer.notify(&path, ChangeKind::Removed); Ok(()) }) .await From 457a966068b6d86f3e1b27c0a9910bf4e00c7a68 Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 6 Nov 2026 20:00:29 +0000 Subject: [PATCH 2/4] Isolate watcher queues and selectively reload resolved service binaries --- host/src/process/watcher.rs | 21 +++--- host/tests/init.rs | 72 ++++++++++++++++++- userland/Cargo.lock | 1 + userland/init/README.md | 12 +++- userland/init/src/lib.rs | 7 +- userland/init/src/main.rs | 114 +++++++++++++++++++++++++------ userland/init/src/systemd-run.rs | 7 +- userland/spawn/Cargo.toml | 1 + userland/spawn/src/lib.rs | 4 +- vfs/src/observe.rs | 52 ++++++++++++-- 10 files changed, 247 insertions(+), 44 deletions(-) diff --git a/host/src/process/watcher.rs b/host/src/process/watcher.rs index 5ec2408..88c8963 100644 --- a/host/src/process/watcher.rs +++ b/host/src/process/watcher.rs @@ -78,17 +78,18 @@ impl HostWatcher for State { .root .observer() .ok_or_else(|| wasmtime::format_err!("filesystem has no observer"))?; + let paths: Vec = paths + .iter() + .map(|p| { + sausage_vfs::resolve("/", p) + .trim_end_matches('/') + .to_owned() + }) + .map(|p| if p.is_empty() { "/".into() } else { p }) + .collect(); Ok(self.table.push(Watcher { - paths: paths - .iter() - .map(|p| { - sausage_vfs::resolve("/", p) - .trim_end_matches('/') - .to_owned() - }) - .map(|p| if p.is_empty() { "/".into() } else { p }) - .collect(), - events: observer.subscribe(), + events: observer.subscribe(paths.clone()), + paths, pending: Vec::new(), })?) } diff --git a/host/tests/init.rs b/host/tests/init.rs index db9b111..844320d 100644 --- a/host/tests/init.rs +++ b/host/tests/init.rs @@ -154,6 +154,74 @@ fn lines(path: &Path) -> Vec { .map(str::to_owned) .collect() } +#[tokio::test] +async fn unrelated_bursts_and_watched_overflow_only_reload_changed_units() { + let _serial = SERIAL.lock().await; + let (temp, runtime, root) = world(); + let supervisor = supervise(runtime.clone()); + let path = temp.path().join("world"); + wait(|| path.join("var/log/journal").exists()).await; + for name in ["changed", "unchanged"] { + write( + &root, + &format!("/run/systemd/transient/{name}.service"), + format!("[Service]\nExecStart=probe server /{name}.log\n").as_bytes(), + ) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", name]) + .await + .0, + 0 + ); + wait(|| lines(&path.join(format!("{name}.log"))) == ["start"]).await; + } + tokio::task::unconstrained(async { + for _ in 0..4000 { + write(&root, "/run/unrelated", b"noise").await; + } + }) + .await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", "unchanged"]) + .await + .0, + 0 + ); + assert_eq!(lines(&path.join("changed.log")), ["start"]); + assert_eq!(lines(&path.join("unchanged.log")), ["start"]); + // This single-thread executor cannot schedule init during ready tmpfs writes. + // The watched queue therefore overflows before init can consume the edits. + tokio::task::unconstrained(async { + write( + &root, + "/run/systemd/transient/changed.service", + b"[Service]\nExecStart=probe server /changed.log\nEnvironment=REV=2\n", + ) + .await; + for _ in 0..1000 { + write(&root, "/run/systemd/transient/noise", b"noise").await; + } + }) + .await; + wait(|| lines(&path.join("changed.log")) == ["start", "stop", "start"]).await; + assert_eq!( + run(&runtime, &["/bin/systemctl.wasm", "start", "unchanged"]) + .await + .0, + 0 + ); + assert_eq!(lines(&path.join("unchanged.log")), ["start"]); + // Bare names also watch the resolved executable, not the literal name. + let mut bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + bytes.extend_from_slice(&[0, 2, 1, b'x']); + write(&root, "/bin/replacement.wasm", &bytes).await; + root.rename("/bin/replacement.wasm", "/bin/probe.wasm") + .await + .unwrap(); + wait(|| lines(&path.join("unchanged.log")) == ["start", "stop", "start"]).await; + supervisor.shutdown().await; +} #[tokio::test(flavor = "multi_thread")] async fn watcher_and_child_exit_pollable() { let _serial = SERIAL.lock().await; @@ -276,7 +344,9 @@ async fn live_unit_changes_and_binary_replacement() { ) .await; wait(|| lines(&path.join("live.log")) == ["start", "stop", "start"]).await; - let bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + let mut bytes = std::fs::read(guests().join("init-probe.wasm")).unwrap(); + // Add a valid custom section, so the replacement differs in content. + bytes.extend_from_slice(&[0, 2, 1, b'x']); write(&root, "/bin/replacement.wasm", &bytes).await; root.rename("/bin/replacement.wasm", "/bin/probe.wasm") .await diff --git a/userland/Cargo.lock b/userland/Cargo.lock index f06325f..1ec69bc 100644 --- a/userland/Cargo.lock +++ b/userland/Cargo.lock @@ -4074,6 +4074,7 @@ dependencies = [ name = "sausage-spawn" version = "0.1.0" dependencies = [ + "wasi 0.14.7+wasi-0.2.4", "wit-bindgen 0.46.0", ] diff --git a/userland/init/README.md b/userland/init/README.md index be2a090..6cabd5e 100644 --- a/userland/init/README.md +++ b/userland/init/README.md @@ -16,7 +16,14 @@ When init receives EOF, it stops dependents before prerequisites. Enable records are files under `default.target.wants` or `multi-user.target.wants` in the persistent unit directory. The `[Install]` section supplies `WantedBy` for `systemctl enable`. Init observes VFS changes and applies unit edits without `daemon-reload`. -A replacement of a running unit's executable restarts that unit. +A change to a running unit's executable content restarts that unit. +Init resolves bare executable names through `PATH`, which defaults to `/usr/local/bin:/bin`. +Each directory tries the bare name before the `.wasm` suffix. +Init watches those candidate paths and compares the resolved executable's content hash. +After an overflow, init rescans definitions and executable hashes. +It restarts only units with a changed definition or executable. +Unrelated world writes do not enter these watcher queues. +Init and request clients block on pollables, with timers for lifecycle deadlines. ## Control protocol @@ -71,5 +78,6 @@ The cap is 1 MiB. Init keeps one rotated file with the `.1` suffix. A watcher matches each supplied path and all its descendants. Its pollable remains ready while changes remain pending. `changes()` returns and clears the pending changes. -The queue holds at most 256 changes. An `overflow` event requires a rescan. +Each watcher filters paths before its independent queue holds at most 256 changes. +An `overflow` event requires a rescan. Observers report successful VFS mutations, not direct host filesystem writes outside the VFS. diff --git a/userland/init/src/lib.rs b/userland/init/src/lib.rs index 8c53313..2478668 100644 --- a/userland/init/src/lib.rs +++ b/userland/init/src/lib.rs @@ -131,6 +131,10 @@ pub fn request(action: &str, unit: &str) -> io::Result { })?, )?; let deadline = std::time::Instant::now() + Duration::from_secs(60); + let timeout = wasi::clocks::monotonic_clock::subscribe_duration( + Duration::from_secs(60).as_nanos() as u64, + ); + let changed = watcher.subscribe(); let mut acknowledged = false; loop { if !acknowledged && let Ok(bytes) = fs::read(&reply) { @@ -167,9 +171,8 @@ pub fn request(action: &str, unit: &str) -> io::Result { "init did not reply", )); } - // The short timer also handles a missing or failed init. + sausage_spawn::poll(&[&changed, &timeout]); let _ = watcher.changes(); - std::thread::sleep(Duration::from_millis(20)); } } #[cfg(test)] diff --git a/userland/init/src/main.rs b/userland/init/src/main.rs index 7a49b80..4a63394 100644 --- a/userland/init/src/main.rs +++ b/userland/init/src/main.rs @@ -5,9 +5,67 @@ use std::{ collections::{BTreeMap, BTreeSet, VecDeque}, fs, io::{self, Write}, - time::{Duration, Instant}, + time::Instant, }; +fn program_paths(unit: &Unit) -> Vec { + let program = shlex::split(unit.get("Service.ExecStart")) + .and_then(|args| args.first().cloned()) + .unwrap_or_default(); + if program.contains('/') { + return vec![program]; + } + let path = shlex::split(unit.get("Service.Environment")) + .unwrap_or_default() + .into_iter() + .find_map(|v| v.strip_prefix("PATH=").map(str::to_owned)) + .unwrap_or_else(|| "/usr/local/bin:/bin".into()); + path.split(':') + .flat_map(|dir| [format!("{dir}/{program}"), format!("{dir}/{program}.wasm")]) + .collect() +} +fn binary(unit: &Unit) -> Option<(String, u64)> { + use std::hash::{Hash, Hasher}; + program_paths(unit).into_iter().find_map(|path| { + let bytes = fs::read(&path).ok()?; + let mut hash = std::collections::hash_map::DefaultHasher::new(); + bytes.hash(&mut hash); + Some((path, hash.finish())) + }) +} +fn wait_events( + services: &BTreeMap, + mut pollables: Vec, + tick_at: Instant, +) { + let now = Instant::now(); + let mut deadline = None; + for service in services.values() { + if let Some(child) = &service.child { + if child.try_wait().is_none() { + pollables.push(child.subscribe()); + } + for stream in [&child.stdout, &child.stderr].into_iter().flatten() { + pollables.push(stream.subscribe()); + } + } + let time = service + .stopping + .or_else(|| (service.desired && service.child.is_none()).then_some(service.next)); + if let Some(time) = time.filter(|time| *time > tick_at) { + deadline = Some(deadline.map_or(time, |old: Instant| old.min(time))); + } + } + if let Some(deadline) = deadline { + pollables.push(wasi::clocks::monotonic_clock::subscribe_duration( + deadline.saturating_duration_since(now).as_nanos() as u64, + )); + } + if !pollables.is_empty() { + sausage_spawn::poll(&pollables.iter().collect::>()); + } +} struct Service { + binary: Option<(String, u64)>, unit: Unit, child: Option, state: String, @@ -20,6 +78,7 @@ struct Service { impl Service { fn new(unit: Unit) -> Self { Self { + binary: binary(&unit), unit, child: None, state: "inactive".into(), @@ -68,7 +127,12 @@ impl Service { self.desired = false; return; }; - let mut command = Command::new(&argv[0]); + let program = self + .binary + .as_ref() + .map(|(path, _)| path.as_str()) + .unwrap_or(&argv[0]); + let mut command = Command::new(program); command .args(&argv[1..]) .new_group() @@ -244,7 +308,9 @@ fn scan(services: &mut BTreeMap) { let service = services .entry(name.clone()) .or_insert_with(|| Service::new(unit.clone())); - if service.unit != unit { + let binary = binary(&unit); + if service.unit != unit || service.binary != binary { + service.binary = binary; let desired = service.desired || service.state == "active"; service.stop(); service.unit = unit; @@ -326,7 +392,8 @@ fn main() -> io::Result<()> { ] { fs::create_dir_all(dir)?; } - let watcher = sausage_spawn::Watcher::new(&["/".into()]); + let mut paths = vec![UNITS.into(), TRANSIENT.into(), format!("{RUN}/requests")]; + let mut watcher = sausage_spawn::Watcher::new(&paths); let stdin = wasi::cli::stdin::get_stdin(); let mut services = BTreeMap::new(); scan(&mut services); @@ -336,23 +403,10 @@ fn main() -> io::Result<()> { } let changes = watcher.changes(); if !changes.is_empty() { - for service in services.values_mut() { - let program = shlex::split(service.unit.get("Service.ExecStart")) - .and_then(|a| a.first().cloned()) - .unwrap_or_default(); - if service.child.is_some() - && changes.iter().any(|c| { - c.path == program || matches!(c.kind, sausage_spawn::ChangeKind::Overflow) - }) - { - service.stop(); - service.desired = true; - service.next = Instant::now(); - } - } scan(&mut services); } controls(&mut services); + let tick_at = Instant::now(); let names: Vec<_> = services.keys().cloned().collect(); for name in &names { services.get_mut(name).unwrap().tick(name); @@ -388,7 +442,24 @@ fn main() -> io::Result<()> { for (name, service) in &services { service.publish(name); } - std::thread::sleep(Duration::from_millis(20)); + let mut next_paths = vec![UNITS.into(), TRANSIENT.into(), format!("{RUN}/requests")]; + for service in services.values() { + next_paths.extend(program_paths(&service.unit)); + } + next_paths.sort(); + next_paths.dedup(); + if paths != next_paths { + paths = next_paths; + watcher = sausage_spawn::Watcher::new(&paths); + // Close the snapshot/subscription race before blocking. + scan(&mut services); + continue; + } + wait_events( + &services, + vec![watcher.subscribe(), stdin.subscribe()], + tick_at, + ); } // Stop dependents before their prerequisites. A cycle stops together. while services.values().any(|s| s.child.is_some()) { @@ -409,13 +480,16 @@ fn main() -> io::Result<()> { if leaves.is_empty() { leaves = live; } + let tick_at = Instant::now(); for name in leaves { let service = services.get_mut(&name).unwrap(); service.stop(); service.tick(&name); service.publish(&name); } - std::thread::sleep(Duration::from_millis(20)); + if services.values().any(|s| s.child.is_some()) { + wait_events(&services, Vec::new(), tick_at); + } } Ok(()) } diff --git a/userland/init/src/systemd-run.rs b/userland/init/src/systemd-run.rs index 968face..172d8a8 100644 --- a/userland/init/src/systemd-run.rs +++ b/userland/init/src/systemd-run.rs @@ -1,6 +1,6 @@ //! Creates transient unit files and optionally waits for their final state. use sausage_init::*; -use std::{fs, io, time::Duration}; +use std::{fs, io}; fn run() -> io::Result { let mut args = std::env::args().skip(1).peekable(); let mut unit = format!( @@ -65,6 +65,8 @@ fn run() -> io::Result { return Ok(reply.code); } if wait { + let watcher = sausage_spawn::Watcher::new(&[format!("{RUN}/state/{unit}")]); + let changed = watcher.subscribe(); loop { if let Ok(bytes) = fs::read(format!("{RUN}/state/{unit}")) { let status: Status = serde_json::from_slice(&bytes)?; @@ -72,7 +74,8 @@ fn run() -> io::Result { return Ok(code); } } - std::thread::sleep(Duration::from_millis(20)); + changed.block(); + let _ = watcher.changes(); } } Ok(0) diff --git a/userland/spawn/Cargo.toml b/userland/spawn/Cargo.toml index 794a6a2..92e83db 100644 --- a/userland/spawn/Cargo.toml +++ b/userland/spawn/Cargo.toml @@ -5,3 +5,4 @@ edition = "2024" [dependencies] wit-bindgen = "=0.46.0" +wasi = "0.14" diff --git a/userland/spawn/src/lib.rs b/userland/spawn/src/lib.rs index a8b4020..772cb72 100644 --- a/userland/spawn/src/lib.rs +++ b/userland/spawn/src/lib.rs @@ -3,12 +3,12 @@ //! Pipe callers must drain stdout and stderr before wait to avoid bounded-pipe deadlocks. use std::{collections::BTreeMap, io, path::{Path, PathBuf}}; mod bindings { - wit_bindgen::generate!({ path: "../../host/wit", world: "program", generate_all }); + wit_bindgen::generate!({ path: "../../host/wit", world: "program", generate_all, with: { "wasi:io/poll@0.2.12": ::wasi::io::poll } }); } use bindings::sausage::process::host; pub use bindings::wasi::io::streams::{InputStream, OutputStream}; pub use host::{Stdio, Watcher, Change, ChangeKind}; -pub use bindings::wasi::io::poll::{Pollable, poll}; +pub use wasi::io::poll::{Pollable, poll}; /// Hard-stop a process group by a live process ID. pub fn kill(pid: u32) -> io::Result<()> { host::kill(pid).map_err(io::Error::other) } diff --git a/vfs/src/observe.rs b/vfs/src/observe.rs index 1738164..4a5008b 100644 --- a/vfs/src/observe.rs +++ b/vfs/src/observe.rs @@ -23,12 +23,16 @@ enum Held { type TrackedPath = Mutex>; type Path = Arc; type Callback = dyn Fn(&str) + Send + Sync; +type Subscription = ( + Vec, + tokio::sync::broadcast::Sender<(String, ChangeKind)>, +); /// Shared observation for mount namespaces. Paths supplied to the callback are virtual paths. /// Callbacks run after filesystem locks have been released; they must not block. #[derive(Clone)] pub struct Observer { changed: Arc, - events: tokio::sync::broadcast::Sender<(String, ChangeKind)>, + events: Arc>>, paths: Arc>>>, // Only open-registration versus rename/remove needs ordering. Lock the affected // subtrees, not all IO: a blocked open must not stall unrelated namespaces. @@ -38,18 +42,31 @@ impl Observer { pub fn new(changed: impl Fn(&str) + Send + Sync + 'static) -> Self { Self { changed: Arc::new(changed), - events: tokio::sync::broadcast::channel(256).0, + events: Arc::default(), paths: Arc::default(), namespace: Arc::default(), } } /// Subscribe before the first mutation of interest. Lag means callers must rescan. - pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> { - self.events.subscribe() + pub fn subscribe( + &self, + paths: Vec, + ) -> tokio::sync::broadcast::Receiver<(String, ChangeKind)> { + let (sender, receiver) = tokio::sync::broadcast::channel(256); + self.events.lock().unwrap().push((paths, sender)); + receiver } fn notify(&self, path: &str, kind: ChangeKind) { (self.changed)(path); - let _ = self.events.send((path.to_owned(), kind)); + self.events.lock().unwrap().retain(|(paths, sender)| { + if sender.receiver_count() == 0 { + return false; + } + if paths.iter().any(|base| under(path, base)) { + let _ = sender.send((path.to_owned(), kind)); + } + true + }); } /// Wrap a layer at its canonical namespace prefix (e.g. `/run`). pub fn layer(&self, layer: Arc, prefix: &str, process: Process) -> Arc { @@ -334,6 +351,31 @@ impl File for Handle { #[cfg(test)] mod tests { + #[test] + fn queues_filter_before_the_bound_and_overflow_independently() { + let observer = super::Observer::new(|_| {}); + let mut a = observer.subscribe(vec!["/watched/a".into()]); + let mut b = observer.subscribe(vec!["/watched/b".into()]); + for _ in 0..4000 { + observer.notify("/elsewhere", super::ChangeKind::Modified); + } + assert!(matches!( + a.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + for _ in 0..300 { + observer.notify("/watched/a", super::ChangeKind::Modified); + } + assert!(matches!( + a.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Lagged(_)) + )); + assert!(matches!( + b.try_recv(), + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + )); + } + use super::*; use tokio::sync::mpsc; #[test] From c70948efe21022b519c2f5c311b775a20779549e Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 6 Nov 2026 21:05:11 +0000 Subject: [PATCH 3/4] api: add public credential routes and SSH signing oracle --- Cargo.lock | 1 + host/Cargo.toml | 1 + host/src/api/README.md | 71 ++++++++++++ host/src/api/mod.rs | 207 +++++++++++++++++++++++++++++++++++ host/src/api/ssh.rs | 96 +++++++++++++++++ host/src/cred/grants.rs | 19 +++- host/src/lib.rs | 1 + host/tests/api.rs | 233 ++++++++++++++++++++++++++++++++++++++++ 8 files changed, 625 insertions(+), 4 deletions(-) create mode 100644 host/src/api/README.md create mode 100644 host/src/api/mod.rs create mode 100644 host/src/api/ssh.rs create mode 100644 host/tests/api.rs diff --git a/Cargo.lock b/Cargo.lock index 7d50be6..dd5aa83 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4995,6 +4995,7 @@ dependencies = [ "anyhow", "async-trait", "base64 0.22.1", + "bytes", "chacha20poly1305", "clap", "deno_ast", diff --git a/host/Cargo.toml b/host/Cargo.toml index 88863a7..983031f 100644 --- a/host/Cargo.toml +++ b/host/Cargo.toml @@ -51,6 +51,7 @@ ssh-key = { version = "0.6.7", features = ["encryption", "ed25519", "ecdsa", "rs uuid = { version = "1.29.0", features = ["v4"] } webauthn-rs = "0.5" openssl = { version = "0.10.82", features = ["vendored"] } +bytes = "1" [dev-dependencies] tempfile = "3" diff --git a/host/src/api/README.md b/host/src/api/README.md new file mode 100644 index 0000000..1e3ce76 --- /dev/null +++ b/host/src/api/README.md @@ -0,0 +1,71 @@ +# Host API + +`handle(&Api, Caller, Request)` answers JSON requests for `host.sausage`. +The transport supplies the caller from the process table, never from request JSON or headers. +The transport hook is outside this module. +Use one `Cred` per host process. +SQLite and script work must run on a blocking thread in the transport. + +T25 Note 1 pauses authorization routes for security decisions C1 and H2. +T27 will change the mutation methods to require typed proofs. +This module currently exposes only reads, request filing, and SSH operations. +It does not expose decisions, enrollment, revocation, or secret entry. +Those routes return 404 until the orchestrator authorizes further work. +The API retains the passkey manager and entry key for the later routes. +The host supplies their configuration and host-only paths. +Tests use RP ID `localhost` and origin `http://localhost:8080`. + +Reads are open to every caller. +Successful responses use status 200 and JSON. +Errors use `{"error":"kind","message":"kind"}` with static, secret-free text. +Unknown routes return 404. +Invalid JSON or parameters return 400. +Conflicts return 409. +Request filing permits 32 attempts per process within a 60-second window. +The next attempt returns 429 until that window expires. +Explicit requests and automatic SSH requests share this budget. +Reads and granted SSH operations do not spend the budget. + +## Current routes + +- `GET /whoami` returns the trusted caller. +- `GET /cred/items` returns public item snapshots. +- `GET /cred/items/{id}` returns one public snapshot. +- `GET /cred/requests?state=pending` returns requests with diffs and summaries. + Omit `state` to return every request. +- `GET /cred/requests/{id}` returns one request with its diff and summary. +- `POST /cred/requests` accepts `{"kind":"grant","key":"stable-key","request":{...}}`. + The request contains `items` and optional `duration_secs`, `max_uses`, and `validator`. + Kind `edit` accepts an `OperationRequest` with `item`, `base_version`, and `change`. + This route returns `{"request":"id"}` without approving the request. + The key prefixes `inject:` and `ssh:` belong to automatic requests. + An explicit request with either prefix returns 400. +- `GET /cred/grants` returns the grant ledger. +- `POST /ssh/identity` accepts `{"host":"example.com"}`. + It returns `{"item":"id","public_key":"base64 SSH public key blob","user":null}`. + A public `user` field supplies the optional default user. +- `POST /ssh/sign` accepts `{"host":"example.com","item":"id","payload":"base64"}`. + It returns `{"signature":"base64 SSH wire signature"}`. + +SSH operations match SSH rules and enforce free use or a caller grant. +Both operations evaluate item and grant validators. +Identity discovery checks the grant without spending a use. +Signing reserves one grant use. +An ungranted caller receives 403 with `error`, `message`, `item`, and `request`. +The host files that request before it responds. +Two matching items produce 409. +The signing item must match the selected host rule. +The host never returns a private key or decrypted field value. + +## Held routes + +The original brief also requests these route groups: + +- Request challenges and decisions. +- Passkey enrollment and listing. +- Grant revocation through an approved request. +- The entry public key, entry challenges, and encrypted field writes. + +Their final contracts depend on the security decisions and T27's typed proof interface. +The host must not trust world-served pages to describe the approved action. +No public route accepts a caller identity, a raw binding, or an approver string. diff --git a/host/src/api/mod.rs b/host/src/api/mod.rs new file mode 100644 index 0000000..57d0191 --- /dev/null +++ b/host/src/api/mod.rs @@ -0,0 +1,207 @@ +//! JSON routes for host.sausage. The transport supplies the trusted caller. +//! Authorization routes remain on hold until the host-origin and proof decisions land. +mod ssh; + +use crate::cred::{self, Caller, Cred, entry::EntryKey, webauthn::Passkeys}; +use bytes::Bytes; +use http::{Request, Response}; +use serde::de::DeserializeOwned; +use serde_json::{Value, json}; +use std::{ + collections::HashMap, + sync::Mutex, + time::{Duration, Instant}, +}; + +pub struct Api<'a> { + pub cred: Cred, + pub passkeys: Mutex>, + pub entry: EntryKey, + requests: Mutex>, +} +impl<'a> Api<'a> { + /// Passkeys owns the RP configuration. Its database must remain host-only. + pub fn new(cred: Cred, passkeys: Passkeys<'a>, entry: EntryKey) -> Self { + Self { + cred, + passkeys: Mutex::new(passkeys), + entry, + requests: Mutex::new(HashMap::new()), + } + } + fn rate_limit(&self, caller: &Caller) -> Result<()> { + let now = Instant::now(); + let mut requests = self.requests.lock().map_err(|_| Error(500, "storage"))?; + requests.retain(|_, (start, _)| now.duration_since(*start) < Duration::from_secs(60)); + let (_, count) = requests.entry(caller.pid).or_insert((now, 0)); + if *count >= 32 { + return Err(Error(429, "rate_limited")); + } + *count += 1; + Ok(()) + } +} + +struct Error(u16, &'static str); +type Result = std::result::Result; +impl From for Error { + fn from(error: cred::Error) -> Self { + let status = match error.0 { + "item_deleted" | "request_missing" | "grant_missing" => 404, + "ambiguous_item" | "stale" | "already_decided" | "key_conflict" => 409, + "storage" => 500, + _ => 400, + }; + Self(status, error.0) + } +} +fn parse(value: Value) -> Result { + serde_json::from_value(value).map_err(|_| Error(400, "invalid_json")) +} +fn string<'a>(value: &'a Value, key: &str) -> Result<&'a str> { + value + .get(key) + .and_then(Value::as_str) + .ok_or(Error(400, "invalid_params")) +} +fn value(data: T) -> Result { + serde_json::to_value(data).map_err(|_| Error(500, "serialization")) +} +fn request_view(request: cred::requests::Request) -> Result { + let summary = if let Some(requested) = request.payload.get("requested") { + let requested: cred::edits::OperationRequest = parse(requested.clone())?; + requested + .change + .iter() + .map(cred::summary::operation_summary) + .collect::>() + .join(" ") + } else if request.payload.get("grant_revoke").is_some() { + "Revoke the grant. All access through this grant will stop.".into() + } else if request.payload.get("grant_create").is_some() { + "Create a grant. Review the subject and limits in Details.".into() + } else if request.payload.get("delete").is_some() { + "Delete the item and permanently discard its field values.".into() + } else if request.payload.get("revert").is_some() { + "Restore the public configuration. Discarded secret values cannot be recovered.".into() + } else { + let grant: cred::requests::GrantRequest = parse(request.payload.clone())?; + format!( + "Give access to {} credential items for {}.", + grant.items.len(), + cred::summary::human_duration(grant.duration_secs.unwrap_or(3600)) + ) + }; + let mut out = value(&request)?; + out["diff"] = request.payload; + out["summary"] = json!(summary); + Ok(out) +} + +/// Handle a request without a network transport. Errors contain only static error kinds. +pub async fn handle(ctx: &Api<'_>, caller: Caller, request: Request) -> Response { + let result = route(ctx, &caller, request); + let (status, body) = match result { + Ok(body) => ( + if body.get("error").and_then(Value::as_str) == Some("grant_required") { + 403 + } else { + 200 + }, + body, + ), + Err(Error(status, kind)) => (status, json!({"error":kind,"message":kind})), + }; + Response::builder() + .status(status) + .header("content-type", "application/json") + .body(Bytes::from(serde_json::to_vec(&body).expect("JSON value"))) + .expect("valid response") +} +fn route(ctx: &Api<'_>, caller: &Caller, request: Request) -> Result { + let path: Vec<_> = request.uri().path().trim_matches('/').split('/').collect(); + let method = request.method().as_str(); + let body: Value = if request.body().is_empty() { + json!({}) + } else { + serde_json::from_slice(request.body()).map_err(|_| Error(400, "invalid_json"))? + }; + match (method, path.as_slice()) { + ("GET", ["whoami"]) => value(caller), + ("GET", ["cred", "items"]) => value(ctx.cred.item_list()?), + ("GET", ["cred", "items", id]) => value(ctx.cred.item_get(id)?), + ("GET", ["cred", "requests"]) => { + let state = url::form_urlencoded::parse(request.uri().query().unwrap_or("").as_bytes()) + .find(|(key, _)| key == "state") + .map(|(_, state)| state.into_owned()); + ctx.cred + .request_list()? + .into_iter() + .filter(|r| state.as_ref().is_none_or(|s| s == &r.state)) + .map(request_view) + .collect::>>() + .map(Value::Array) + } + ("GET", ["cred", "requests", id]) => request_view( + ctx.cred + .request_list()? + .into_iter() + .find(|r| r.id == *id) + .ok_or(Error(404, "request_missing"))?, + ), + ("POST", ["cred", "requests"]) => { + let key = string(&body, "key")?; + if key.starts_with("inject:") || key.starts_with("ssh:") { + return Err(Error(400, "reserved_key")); + } + ctx.rate_limit(caller)?; + let id = match string(&body, "kind")? { + "grant" => ctx + .cred + .file_request(caller, key, parse(body["request"].clone())?)?, + "edit" => { + ctx.cred + .file_operation_request(caller, key, parse(body["request"].clone())?)? + } + _ => return Err(Error(400, "invalid_kind")), + }; + Ok(json!({"request":id})) + } + ("GET", ["cred", "grants"]) => value(ctx.cred.grant_list()?), + ("POST", ["ssh", action @ ("identity" | "sign")]) => { + ssh::handle(ctx, caller, action, &body) + } + _ => Err(Error(404, "route_missing")), + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn request_budget_is_per_caller_and_bounded() -> anyhow::Result<()> { + let dir = tempfile::tempdir()?; + let db = rusqlite::Connection::open_in_memory()?; + let api = Api::new( + Cred::open(dir.path())?, + Passkeys::new(&db, "localhost", "http://localhost:8080")?, + EntryKey::open(&dir.path().join("entry.key"))?, + ); + let caller = Caller::default(); + for _ in 0..32 { + assert!(api.rate_limit(&caller).is_ok()); + } + assert!(matches!( + api.rate_limit(&caller), + Err(Error(429, "rate_limited")) + )); + assert!( + api.rate_limit(&Caller { + pid: 1, + ..Default::default() + }) + .is_ok() + ); + Ok(()) + } +} diff --git a/host/src/api/ssh.rs b/host/src/api/ssh.rs new file mode 100644 index 0000000..adf80f6 --- /dev/null +++ b/host/src/api/ssh.rs @@ -0,0 +1,96 @@ +//! SSH signing oracle. Rules and grants precede every key operation. +use super::{Api, Error, Result, string}; +use crate::cred::{self, Caller, script::Script, types::FieldType}; +use base64::{Engine, engine::general_purpose::STANDARD}; +use serde_json::{Value, json}; + +pub(super) fn handle(ctx: &Api<'_>, caller: &Caller, action: &str, body: &Value) -> Result { + let cred = &ctx.cred; + let host = string(body, "host")?; + let mut items = cred.item_list()?.into_iter().filter(|item| { + item.rules + .iter() + .any(|rule| cred::items::rule_matches(rule, "ssh", host, "/", "POST")) + }); + let item = items.next().ok_or(Error(404, "item_missing"))?; + if items.next().is_some() { + return Err(Error(409, "ambiguous_item")); + } + if action == "sign" && string(body, "item")? != item.id { + return Err(Error(403, "item_mismatch")); + } + let prepared = if item.free_use { + Some((None, cred.credential(&item, true)?)) + } else { + cred.prepare_use(&item, caller, action == "sign")? + .map(|(grant, values)| (Some(grant), values)) + }; + let Some((grant, values)) = prepared else { + ctx.rate_limit(caller)?; + let request = cred.file_request( + caller, + &format!("ssh:{}", item.id), + cred::requests::GrantRequest { + items: vec![item.id.clone()], + duration_secs: None, + max_uses: None, + validator: None, + }, + )?; + return Ok( + json!({"error":"grant_required","message":"grant_required","item":item.id,"request":request}), + ); + }; + let public = cred.credential(&item, false)?; + let context = json!({"request":{"method":"POST","url":format!("ssh://{host}/"),"headers":[]},"credential":public,"item":{"id":item.id,"name":item.name},"grant":grant.as_ref().map(|g|json!({"conversation":match &g.subject {cred::types::Subject::Conversation(id)=>Some(id),cred::types::Subject::Program(_)=>None},"subject":g.subject,"uses_left":g.max_uses.map(|m|m-g.uses),"exp":g.exp}))}); + for source in [ + item.validate_js.as_deref(), + grant.as_ref().and_then(|g| g.validator.as_deref()), + ] + .into_iter() + .flatten() + { + let output = cred::v8::V8 + .evaluate(source, &context.to_string(), None, Default::default()) + .map_err(|_| Error(403, "validator_denied"))?; + let output: Value = + serde_json::from_str(&output).map_err(|_| Error(403, "validator_denied"))?; + if output + .as_bool() + .or_else(|| output.get("allow").and_then(Value::as_bool)) + != Some(true) + { + return Err(Error(403, "validator_denied")); + } + } + let field = item + .fields + .iter() + .find(|field| field.field_type == FieldType::SshPrivateKey) + .ok_or(Error(400, "signing_key_missing"))?; + let secret = values + .get(&field.name) + .and_then(Value::as_str) + .ok_or(Error(400, "signing_key_missing"))?; + let format = field.key_type.as_deref().ok_or(Error(400, "invalid_key"))?; + if action == "identity" { + let public_key = cred::sign::public_key(secret, None, format, "ssh") + .map_err(|_| Error(400, "invalid_key"))?; + Ok( + json!({"item":item.id,"public_key":public_key,"user":public.get("user").and_then(Value::as_str)}), + ) + } else { + let payload = STANDARD + .decode(string(body, "payload")?) + .map_err(|_| Error(400, "invalid_payload"))?; + let (padding, hash) = match format { + "rs256" | "rsa-sha2-256" => (Some("pkcs1v15"), Some("sha256")), + "rsa-sha2-512" => (Some("pkcs1v15"), Some("sha512")), + _ => (None, None), + }; + let signature = cred::sign::sign(secret, None, format, &payload, "ssh", padding, hash) + .map_err(|_| Error(400, "sign_failed"))?; + cred.audit_sign(&item.id, caller, &payload)?; + Ok(json!({"signature":signature.signature})) + } +} diff --git a/host/src/cred/grants.rs b/host/src/cred/grants.rs index 3d04654..caa2f5d 100644 --- a/host/src/cred/grants.rs +++ b/host/src/cred/grants.rs @@ -139,6 +139,15 @@ impl Cred { &self, item: &super::items::Item, caller: &Caller, + ) -> Result> { + self.prepare_use(item, caller, true) + } + /// Public-key discovery checks grants without spending a signing use. + pub(crate) fn prepare_use( + &self, + item: &super::items::Item, + caller: &Caller, + consume: bool, ) -> Result> { let mut s = self.access()?; let master = s.master.clone(); @@ -156,11 +165,13 @@ impl Cred { return Err(Error("stale")); } let credential = super::items::credential(&tx, &master, item, true)?; - let changed=tx.execute("UPDATE grants SET uses=uses+1 WHERE id=?1 AND revoked=0 AND (exp IS NULL OR exp>?2) AND (max_uses IS NULL OR uses?2) AND (max_uses IS NULL OR uses Caller { + Caller { + pid: 17, + program: "/bin/agent".into(), + conversations: vec!["conversation".into()], + } +} +async fn send(api: &Api<'_>, method: &str, path: &str, body: Value) -> Result<(u16, Value)> { + let request = http::Request::builder() + .method(method) + .uri(path) + .header("x-sausage-pid", "999") + .body(Bytes::from(serde_json::to_vec(&body)?))?; + let response = handle(api, caller(), request).await; + let text = std::str::from_utf8(response.body())?; + assert!(!text.contains("browser-only-value")); + assert!(!text.contains(&STANDARD.encode([7u8; 32]))); + assert_eq!(response.headers()["content-type"], "application/json"); + Ok(( + response.status().as_u16(), + serde_json::from_slice(response.body())?, + )) +} +async fn ok(api: &Api<'_>, method: &str, path: &str, body: Value) -> Result { + let (status, value) = send(api, method, path, body).await?; + assert_eq!(status, 200, "unexpected API status"); + Ok(value) +} +fn item(cred: &Cred, name: &str) -> Result { + Ok(cred.create_item(serde_json::from_value(json!({"name":name,"rules":[{"scheme":"ssh","hosts":["example.com"],"path_prefixes":["/"]}],"transform_js":"","fields":[{"name":"secret","public":false,"field_type":"ssh_private_key","key_type":"ed25519-raw"}]}))?,&caller(),"fixture")?) +} + +#[tokio::test(flavor = "current_thread")] +async fn public_routes_and_ssh() -> Result<()> { + let dir = tempfile::tempdir()?; + let db = rusqlite::Connection::open(dir.path().join("cred.sqlite"))?; + let cred = Cred::open(dir.path())?; + let item = item(&cred, "ssh")?; + cred.set_field( + &item.id, + &item.fields[0].field_id, + &STANDARD.encode([7u8; 32]), + &caller(), + "fixture", + )?; + let api = Api::new( + cred, + Passkeys::new(&db, "localhost", "http://localhost:8080")?, + EntryKey::open(&dir.path().join("entry.key"))?, + ); + let who = ok( + &api, + "GET", + "/whoami", + json!({"pid":999,"conversations":["forged"]}), + ) + .await?; + assert_eq!(who["pid"], 17); + assert_eq!(who["conversations"], json!(["conversation"])); + let mut pending = String::new(); + for action in ["identity", "sign"] { + let (status, denied) = send( + &api, + "POST", + &format!("/ssh/{action}"), + json!({"host":"example.com","item":item.id,"payload":"YWJj"}), + ) + .await?; + assert_eq!(status, 403); + assert_eq!(denied["item"], item.id); + let request = denied["request"].as_str().unwrap(); + if pending.is_empty() { + pending = request.into(); + } else { + assert_eq!(pending, request); + } + } + for prefix in ["inject:", "ssh:"] { + assert_eq!( + send( + &api, + "POST", + "/cred/requests", + json!({"kind":"grant","key":format!("{prefix}spoof"),"request":{"items":[item.id]}}) + ) + .await? + .1["error"], + "reserved_key" + ); + } + let filed = ok( + &api, + "POST", + "/cred/requests", + json!({"kind":"grant","key":"explicit","request":{"items":[item.id]},"caller":{"pid":999}}), + ) + .await?; + let id = filed["request"].as_str().unwrap(); + let request = ok(&api, "GET", &format!("/cred/requests/{id}"), json!({})).await?; + assert_eq!(request["pid"], 17); + assert_eq!( + request["subject"], + json!({"kind":"conversation","id":"conversation"}) + ); + assert!(request["diff"].is_object()); + assert!(request["summary"].is_string()); + assert_eq!( + ok(&api, "GET", "/cred/requests?state=approved", json!({})).await?, + json!([]) + ); + assert_eq!( + ok(&api, "GET", "/cred/requests?state=pending", json!({})) + .await? + .as_array() + .unwrap() + .len(), + 2 + ); + let edit = ok(&api,"POST","/cred/requests",json!({"kind":"edit","key":"rename","request":{"item":item.id,"base_version":api.cred.item_get(&item.id)?.version,"change":[{"op":"set_name","name":"renamed"}]}})).await?; + let view = ok( + &api, + "GET", + &format!("/cred/requests/{}", edit["request"].as_str().unwrap()), + json!({}), + ) + .await?; + assert!(view["summary"].as_str().unwrap().contains("Rename")); + assert_eq!(api.cred.item_get(&item.id)?.name, "ssh"); + // Trusted fixture setup only. No world route creates this grant. + api.cred.grant_create( + cred::grants::Grant { + id: String::new(), + items: vec![item.id.clone()], + subject: cred::types::Subject::Conversation("conversation".into()), + exp: None, + max_uses: Some(1), + uses: 0, + revoked: false, + validator: None, + note: None, + }, + &caller(), + Some("fixture"), + )?; + let identity = ok(&api, "POST", "/ssh/identity", json!({"host":"example.com"})).await?; + assert_eq!(identity["item"], item.id); + assert_eq!(api.cred.grant_list()?[0].uses, 0); + let public = STANDARD.decode(identity["public_key"].as_str().unwrap())?; + let expected = ed25519_dalek::SigningKey::from_bytes(&[7u8; 32]).verifying_key(); + assert!(public.ends_with(expected.as_bytes())); + assert_ne!( + send( + &api, + "POST", + "/ssh/sign", + json!({"host":"example.com","item":"different","payload":"YWJj"}) + ) + .await? + .0, + 200 + ); + let signed = ok( + &api, + "POST", + "/ssh/sign", + json!({"host":"example.com","item":item.id,"payload":STANDARD.encode(b"payload")}), + ) + .await?; + assert_eq!(api.cred.grant_list()?[0].uses, 1); + let signature = STANDARD.decode(signed["signature"].as_str().unwrap())?; + use ed25519_dalek::Verifier; + assert!(signature.starts_with(b"\0\0\0\x0bssh-ed25519\0\0\0\x40")); + expected.verify( + b"payload", + &ed25519_dalek::Signature::from_slice(&signature[19..])?, + )?; + assert_eq!( + send( + &api, + "POST", + "/ssh/sign", + json!({"host":"example.com","item":item.id,"payload":"YWJj"}) + ) + .await? + .0, + 403 + ); + ok(&api, "GET", "/cred/items", json!({})).await?; + ok(&api, "GET", &format!("/cred/items/{}", item.id), json!({})).await?; + ok(&api, "GET", "/cred/grants", json!({})).await?; + for path in [ + "/cred/passkeys/register/start", + "/cred/items/x/fields/y", + "/cred/requests/x/decide", + "/cred/grants/x/revoke", + ] { + assert_eq!(send(&api, "POST", path, json!({})).await?.0, 404); + } + Ok(()) +} + +#[tokio::test(flavor = "current_thread")] +async fn ssh_rejects_ambiguous_items() -> Result<()> { + let dir = tempfile::tempdir()?; + let db = rusqlite::Connection::open_in_memory()?; + let cred = Cred::open(dir.path())?; + item(&cred, "one")?; + item(&cred, "two")?; + let api = Api::new( + cred, + Passkeys::new(&db, "localhost", "http://localhost:8080")?, + EntryKey::open(&dir.path().join("entry.key"))?, + ); + assert_eq!( + send(&api, "POST", "/ssh/identity", json!({"host":"example.com"})) + .await? + .0, + 409 + ); + assert!(api.cred.request_list()?.is_empty()); + Ok(()) +} From 9517515efef89049f48ad6fbd0b2f34d834240ac Mon Sep 17 00:00:00 2001 From: Surma Date: Fri, 6 Nov 2026 23:43:49 +0000 Subject: [PATCH 4/4] api: shut the init supervisor down gracefully after the console (merge fix for T22 + T25) --- host/src/api/server.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/host/src/api/server.rs b/host/src/api/server.rs index bf8d3b9..3adec06 100644 --- a/host/src/api/server.rs +++ b/host/src/api/server.rs @@ -145,7 +145,6 @@ pub async fn run_with_userland( } anyhow::anyhow!(text) }); - services.abort(); - let _ = services.await; + services.shutdown().await; result }