diff --git a/Cargo.lock b/Cargo.lock index 131a645..d853cf6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -59,6 +59,7 @@ dependencies = [ "dm-web", "dm-wslc-api", "futures-util", + "nix 0.31.3", "serde", "serde_json", "tempfile", @@ -959,6 +960,7 @@ version = "0.1.0" dependencies = [ "dm-server", "fuser", + "nix 0.31.3", "serde_json", "tempfile", "thiserror 2.0.21", diff --git a/crates/memory/Cargo.toml b/crates/memory/Cargo.toml index 944a48b..33cbae4 100644 --- a/crates/memory/Cargo.toml +++ b/crates/memory/Cargo.toml @@ -39,3 +39,6 @@ tokio-tungstenite = "0.30" tempfile = "3.27.0" tokio-postgres = "0.7.18" tower = { version = "0.5.3", features = ["util"] } + +[target.'cfg(unix)'.dev-dependencies] +nix = { version = "0.31", features = ["signal"] } diff --git a/crates/memory/src/main.rs b/crates/memory/src/main.rs index 9e96d2f..bcc9a50 100644 --- a/crates/memory/src/main.rs +++ b/crates/memory/src/main.rs @@ -39,6 +39,7 @@ mod review; mod savings; mod sources; mod status; +mod stops; mod taps; mod tool_server; mod transcripts; @@ -1925,8 +1926,7 @@ async fn prepare_launch( }); deputy.iter().for_each(|d| memory.set_deputy(d)); memory.set_tools(taps::TapTools::with_egress(root, egress.clone())); - let run_dir = root.join("state/run"); - std::fs::create_dir_all(&run_dir)?; + let run_dir = run_dir(root)?; Ok(Prepared { memory, egress, @@ -1947,6 +1947,57 @@ async fn prepare_launch( }) } +/// Where launchers keep their egress sockets, made if need be, with what killed +/// launchers left there swept. +fn run_dir(root: &Path) -> std::io::Result { + let dir = root.join("state/run"); + std::fs::create_dir_all(&dir)?; + sweep_sockets(&dir); + Ok(dir) +} + +/// Remove the egress sockets killed launchers left in `run_dir`. One goes only when it is +/// named `egress-.sock`, as [`prepare_launch`] names it, it is a socket and not a +/// link, and the launcher its name carries is not a running process. Returns the +/// sockets removed. +fn sweep_sockets(run_dir: &Path) -> Vec { + let Ok(entries) = std::fs::read_dir(run_dir) else { + return Vec::new(); + }; + let own = std::process::id(); + entries + .flatten() + .filter_map(|e| { + let name = e.file_name(); + let digits = name + .to_str()? + .strip_prefix("egress-")? + .strip_suffix(".sock")?; + if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) { + return None; + } + let pid: u32 = digits.parse().ok()?; + if pid == 0 || pid == own || dm_sandbox::stop::process_runs(pid) { + return None; + } + #[cfg(unix)] + { + use std::os::unix::fs::FileTypeExt; + if !e.file_type().ok()?.is_socket() { + return None; + } + // a pid names a process in this pid namespace only: a launcher in + // another, sharing the root, still listens + if !dm_sandbox::stop::socket_refuses(&e.path()) { + return None; + } + } + std::fs::remove_file(e.path()).ok()?; + Some(e.path()) + }) + .collect() +} + /// What a tunnelled harness reported about itself, read from the transcript it left and /// recorded beside what the egress observed, marked as the harness's own count. fn record_reported_usage(p: &Prepared) { @@ -2047,6 +2098,8 @@ async fn run_confined( capture: bool, runner: &Runner, ) -> Result> { + // counted until the mount below is gone, so a launcher told to stop waits for it + let _in_flight = stops::InFlight::enter()?; let host_home = home_dir(); let rootfs = match runner { Runner::Image(rootfs) => Some(rootfs.as_path()), @@ -2117,11 +2170,11 @@ async fn run_confined( /// The session's render served as a live mount, where this host can: every open of an /// entry inside is then a `get` the memory serves, whatever tool made it, and the hook /// is told it need not make one. Where it cannot, the render is mounted as files, as -/// before, and the launch says why. Mounts killed launchers left under the harness's -/// mount directory are cleared first. +/// before, and the launch says why. Mounts killed launchers left under the root's mount +/// directory, of any harness, are cleared first. fn live_mount(p: &Prepared, harness: Harness) -> Option { + dm_mount::sweep_root(&p.root); let base = p.root.join(dm_mount::MOUNTS).join(harness.binary()); - dm_mount::sweep(&base); let at = base.join(p.projection.file_name().unwrap_or_default()); match dm_mount::mount(p.memory.clone(), &p.projection, &at) { Ok(m) => { @@ -2898,6 +2951,7 @@ async fn cmd_launch(root: &Path, command: Command) -> Fallible { (None, None) => None, }; let (program, args) = split_program(harness.into(), args); + let stops = stops::Stops::listen()?; let session = Session::start( root, harness.into(), @@ -2912,13 +2966,21 @@ async fn cmd_launch(root: &Path, command: Command) -> Fallible { RunOptions::new(model, subagents, program, None).empty_memory(empty_memory), ) .await?; - let code = match prompt { - Some(prompt) => work::prompted(&session, &prompt, &args, no_handoff).await?, - None => work::interactive(&session, &args).await?, - }; - if mine_after { - session.mine(mine_model.as_deref(), mine_via).await?; - } + let code = stops + .around( + async { + let code = match prompt { + Some(prompt) => work::prompted(&session, &prompt, &args, no_handoff).await?, + None => work::interactive(&session, &args).await?, + }; + if mine_after { + session.mine(mine_model.as_deref(), mine_via).await?; + } + Ok(code) + }, + session.abandon(), + ) + .await?; session.finish(code).await; std::process::exit(code); } @@ -2974,6 +3036,7 @@ async fn cmd_work(root: &Path, command: Command) -> Fallible { }; let workdir = resolve_workdir(root, workdir)?; refuse_browser_unconfined(unconfined, backend)?; + let stops = stops::Stops::listen()?; let session = Session::start( root, harness.into(), @@ -3000,7 +3063,9 @@ async fn cmd_work(root: &Path, command: Command) -> Fallible { prefetch: !no_prefetch, task, }; - let code = work::run(&session, &options).await?; + let code = stops + .around(work::run(&session, &options), session.abandon()) + .await?; session.finish(code).await; std::process::exit(code); } @@ -3686,15 +3751,18 @@ impl Session { let started = std::time::Instant::now(); let (user, sys) = dm_sandbox::measure::own_cpu_ms(); let (code, out, used) = match &self.runner { - Runner::Unconfined(Some((addr, _))) => run_unconfined( - self.harness, - self.auth, - &self.workdir, - &self.p.home, - &format!("http://{addr}"), - args, - capture, - ), + Runner::Unconfined(Some((addr, _))) => { + run_unconfined( + self.harness, + self.auth, + &self.workdir, + &self.p.home, + &format!("http://{addr}"), + args, + capture, + ) + .await + } Runner::Unconfined(None) => Err("the unconfined runner was not started".into()), Runner::Namespaces | Runner::Image(_) | Runner::Seatbelt => { run_confined( @@ -3807,6 +3875,18 @@ impl Session { } } + /// What a launcher told twice to stop does before it exits at once: remove the + /// socket and the render, which no sweep removes, and put the task a worker holds + /// back in the pool. The run in flight, its mount and its scratch directory, end + /// with the launcher and are swept by the next one. + pub async fn abandon(&self) { + std::fs::remove_file(&self.p.socket).ok(); + std::fs::remove_dir_all(&self.p.projection).ok(); + if let Some(id) = work::holding() { + self.p.memory.release(&id, None).await.ok(); + } + } + /// Stop serving, report usage, take the transcripts the record now keeps out of the /// private home, and tell the host the final code where there is one. After mining, /// which reads the home's copy. @@ -3819,14 +3899,15 @@ impl Session { serve.abort(); std::fs::remove_file(&self.p.socket).ok(); } + // the render was this session's alone, and goes with it; here as well as in the + // guard, since a launch ends in process::exit, which runs no destructors. Before + // anything is said, which a terminal that hung up refuses. + std::fs::remove_dir_all(&self.p.projection).ok(); if !self.p.metered { record_reported_usage(&self.p); } report_usage(&self.p.egress); transcripts::remove_kept(&self.p.memory, &self.p.home).await; - // the render was this session's alone, and goes with it; here as well as in the - // guard, since a launch ends in process::exit, which runs no destructors - std::fs::remove_dir_all(&self.p.projection).ok(); if let Runner::Requests(r) = &self.runner { std::fs::write(r.dir.join(requests::DONE), code.to_string()).ok(); } @@ -3851,7 +3932,7 @@ fn home_dir() -> Option { /// Run the harness on the host with the egress on a TCP port. The same configuration as /// the confined path, minus the namespaces. #[allow(clippy::too_many_arguments)] // the host-side twin of dm_sandbox::Request -fn run_unconfined( +async fn run_unconfined( harness: Harness, auth: AuthMode, workdir: &Path, @@ -3860,6 +3941,9 @@ fn run_unconfined( args: &[String], capture: bool, ) -> Result> { + // counted until the harness is reaped and its end said, so a launcher told to stop + // waits for that rather than exiting first + let _in_flight = stops::InFlight::enter()?; std::fs::create_dir_all(home)?; let mut cmd = std::process::Command::new(harness.binary()); cmd.current_dir(workdir); @@ -3912,7 +3996,17 @@ fn run_unconfined( } } cmd.args(args); - let (code, out, r) = dm_sandbox::measure::run(&mut cmd, capture)?; + // on a thread of its own, so a launcher told to stop hears the signal while it runs; + // a headless run in a process group of its own, so a stop kills what it left in the + // background too, and an interactive one in the terminal's foreground group + let (code, out, r) = tokio::task::spawn_blocking(move || { + if capture { + dm_sandbox::measure::run_grouped(&mut cmd, capture) + } else { + dm_sandbox::measure::run(&mut cmd, capture) + } + }) + .await??; Ok(( code, capture.then(|| String::from_utf8_lossy(&out).into_owned()), @@ -5281,3 +5375,67 @@ mod backend_tests { ); } } + +#[cfg(all(test, unix))] +mod socket_sweep_tests { + use super::*; + + #[test] + fn a_killed_launcher_s_socket_is_swept_and_nothing_a_live_one_holds() { + let dir = tempfile::tempdir().unwrap(); + let run_dir = dir.path(); + let mut gone = std::process::Command::new("/bin/sh") + .args(["-c", ":"]) + .spawn() + .unwrap(); + let dead = gone.id(); + gone.wait().unwrap(); + let mut live_child = std::process::Command::new("/bin/sh") + .args(["-c", "read x"]) + .stdin(std::process::Stdio::piped()) + .spawn() + .unwrap(); + let live = live_child.id(); + let stale = run_dir.join(format!("egress-{dead}.sock")); + // the file a listener leaves when its process dies: nothing answers there + drop(std::os::unix::net::UnixListener::bind(&stale).unwrap()); + // a dead pid's name on a socket something still listens on: a launcher in + // another pid namespace, whose pid means nothing here + let elsewhere = run_dir.join(format!("egress-{dead}2.sock")); + let _elsewhere = std::os::unix::net::UnixListener::bind(&elsewhere).unwrap(); + let held = run_dir.join(format!("egress-{live}.sock")); + let _held = std::os::unix::net::UnixListener::bind(&held).unwrap(); + let own = run_dir.join(format!("egress-{}.sock", std::process::id())); + let _own = std::os::unix::net::UnixListener::bind(&own).unwrap(); + // a dead pid's name on a file that is not a socket, and on a link + let file = run_dir.join(format!("egress-{dead}0.sock")); + std::fs::write(&file, "").unwrap(); + let outside = tempfile::tempdir().unwrap(); + let target = outside.path().join("kept.sock"); + let _target = std::os::unix::net::UnixListener::bind(&target).unwrap(); + let link = run_dir.join(format!("egress-{dead}1.sock")); + std::os::unix::fs::symlink(&target, &link).unwrap(); + let other = run_dir.join(format!("other-{dead}.sock")); + let _other = std::os::unix::net::UnixListener::bind(&other).unwrap(); + + // a few times: a process another test forks holds a copy of the closed + // listener's descriptor until it execs, and a connection to it meanwhile is taken + let mut swept = Vec::new(); + for _ in 0..20 { + swept.extend(sweep_sockets(run_dir)); + if !stale.exists() { + break; + } + std::thread::sleep(std::time::Duration::from_millis(10)); + } + + live_child.kill().unwrap(); + live_child.wait().unwrap(); + assert_eq!(swept, vec![stale.clone()]); + assert!(!stale.exists()); + for kept in [&held, &own, &file, &other, &target, &elsewhere] { + assert!(kept.exists(), "{}", kept.display()); + } + assert!(link.is_symlink()); + } +} diff --git a/crates/memory/src/stops.rs b/crates/memory/src/stops.rs new file mode 100644 index 0000000..9c7752d --- /dev/null +++ b/crates/memory/src/stops.rs @@ -0,0 +1,212 @@ +//! A launcher told to stop. SIGTERM, SIGINT and SIGHUP end a session the way a harness +//! that exits does: the sandbox is killed, the run in flight unmounts the projection and +//! removes its scratch directory as a run that ended does, and the session finishes, +//! removing its socket and its render. The launcher then exits with 128 plus the +//! signal's number, as a shell reports a process a signal ended. +//! +//! A headless run without a terminal is in a process group of its own, which a stop +//! kills whole, with what the harness left in the background. On a terminal the harness +//! stays in the terminal's foreground group, so it may use the terminal: a signal sent +//! to the launcher alone then ends the harness and not what it left in the background, +//! which the launcher stops waiting for after a short grace. Ctrl-C, which the terminal +//! sends to the whole foreground group, reaches those too. +//! +//! SIGKILL cannot be caught. What a launcher killed by it leaves, the next launcher to +//! start a run sweeps: `dm_sandbox` its scratch directory, `dm_mount` its mount, and the +//! launcher its egress socket, each named for or locked by the launcher that made it. +//! Its render stays: nothing records which launcher a render belongs to. + +use std::future::Future; +use std::sync::atomic::{AtomicUsize, Ordering}; + +type Fallible = Result>; + +/// Runs between the start of their setup and the end of their cleanup. +static IN_FLIGHT: AtomicUsize = AtomicUsize::new(0); + +/// A run in flight, from before it mounts anything until it has undone all of it. +pub struct InFlight(()); + +impl InFlight { + /// Count a run in flight, or refuse it once the launcher is stopping. + /// + /// # Errors + /// + /// When a signal has told the launcher to stop. + pub fn enter() -> Fallible { + IN_FLIGHT.fetch_add(1, Ordering::SeqCst); + // counted before the check: a stop either sees this run or this run sees the stop + let run = InFlight(()); + if dm_sandbox::stop::stopping() { + return Err("the launcher is stopping".into()); + } + Ok(run) + } +} + +impl Drop for InFlight { + fn drop(&mut self) { + IN_FLIGHT.fetch_sub(1, Ordering::SeqCst); + } +} + +/// The signals that stop a launcher, listened for from when this is made, so one that +/// arrives while the session starts is not the process's death. A signal the launcher +/// was started to ignore, as under `nohup`, is not listened for and stays ignored. +pub struct Stops { + #[cfg(unix)] + signals: [(i32, Option); 3], +} + +/// The first of the signals listened for, by number. +#[cfg(unix)] +async fn first(signals: &mut [(i32, Option); 3]) -> i32 { + async fn one(s: &mut Option) { + match s { + Some(s) => { + s.recv().await; + } + None => std::future::pending().await, + } + } + let [(n1, s1), (n2, s2), (n3, s3)] = signals; + tokio::select! { + () = one(s1) => *n1, + () = one(s2) => *n2, + () = one(s3) => *n3, + } +} + +impl Stops { + /// Listen for SIGTERM, SIGINT and SIGHUP from now on, each that is not ignored. + /// + /// # Errors + /// + /// When a signal handler cannot be installed. + pub fn listen() -> std::io::Result { + #[cfg(unix)] + { + use tokio::signal::unix::{signal, SignalKind}; + let listen = |n: i32, kind: SignalKind| -> std::io::Result<_> { + Ok(( + n, + if dm_sandbox::stop::ignored(n) { + None + } else { + Some(signal(kind)?) + }, + )) + }; + Ok(Stops { + signals: [ + listen(15, SignalKind::terminate())?, + listen(2, SignalKind::interrupt())?, + listen(1, SignalKind::hangup())?, + ], + }) + } + #[cfg(not(unix))] + { + Ok(Stops {}) + } + } + + /// Run `flow`, a session's work, to its end, or until a signal stops it. On a signal + /// every run is killed; the run in flight is let finish its cleanup, and whatever the + /// flow would do after it is not done. Returns the flow's code, or 128 plus the + /// signal's number. A second signal while the cleanup runs ends the wait: `hard`, + /// quick cleanup that must not be left out, runs, and the launcher exits. + /// + /// # Errors + /// + /// The flow's, when it ends of itself. + pub async fn around( + self, + flow: impl Future>, + hard: impl Future, + ) -> Fallible { + #[cfg(unix)] + { + let mut signals = self.signals; + let mut flow = std::pin::pin!(flow); + // the signals first: a harness the same hangup ended returns into a flow that + // would print to the terminal that hung up + let signal = tokio::select! { + biased; + n = first(&mut signals) => n, + r = &mut flow => return r, + }; + dm_sandbox::stop::stop(); + let said = say(signal); + if !said || (signal == 1 && !both_to_files()) { + // nowhere to say anything: what the flow and the finish print goes nowhere, + // rather than ending the launcher in a panic before it cleans up + dm_sandbox::stop::quiet_standard_streams().ok(); + } + while IN_FLIGHT.load(Ordering::SeqCst) > 0 { + tokio::select! { + biased; + n = first(&mut signals) => { + // told twice: stop waiting, but leave nothing the next launch + // cannot sweep. The cap bounds the waits in it only: a removal or + // a ledger statement that runs synchronously runs to its end + // (the ledger's own busy timeout bounds a statement). + let quick = std::time::Duration::from_secs(5); + tokio::time::timeout(quick, hard).await.ok(); + std::process::exit(128 + n); + } + _ = &mut flow => break, + () = tokio::time::sleep(std::time::Duration::from_millis(20)) => {} + } + } + Ok(128 + signal) + } + #[cfg(not(unix))] + { + let _ = hard; + flow.await + } + } +} + +/// Whether standard output and standard error both go to regular files, as under +/// `nohup`. Otherwise a SIGHUP may mean they have gone: a terminal that hung up, which +/// then no longer answers as a terminal, or a pipe to a reader on it. +#[cfg(unix)] +fn both_to_files() -> bool { + use std::os::fd::AsFd; + let to_a_file = |fd: std::os::fd::BorrowedFd<'_>| { + fd.try_clone_to_owned() + .map(std::fs::File::from) + .and_then(|f| f.metadata()) + .is_ok_and(|m| m.is_file()) + }; + to_a_file(std::io::stdout().as_fd()) && to_a_file(std::io::stderr().as_fd()) +} + +/// Say on standard error that a signal stops the launcher. Whether it was said: a +/// terminal that hung up refuses the write. +#[cfg(unix)] +fn say(signal: i32) -> bool { + use std::io::Write; + let name = match signal { + 1 => "SIGHUP", + 2 => "SIGINT", + _ => "SIGTERM", + }; + let line = format!("memory: {name}: ending the run and cleaning up\n"); + std::io::stderr().write_all(line.as_bytes()).is_ok() +} + +/// Write `text` to standard output, and a line end after it when it has none. A reader +/// that has gone, a closed pipe, is not an error here: the launcher goes on to settle and +/// clean up, where `print!` would panic. +pub fn answer(text: &str) { + use std::io::Write; + let mut out = std::io::stdout().lock(); + let _ = out.write_all(text.as_bytes()); + if !text.ends_with('\n') { + let _ = out.write_all(b"\n"); + } + let _ = out.flush(); +} diff --git a/crates/memory/src/work.rs b/crates/memory/src/work.rs index f7892ae..af9af55 100644 --- a/crates/memory/src/work.rs +++ b/crates/memory/src/work.rs @@ -185,10 +185,7 @@ pub async fn prompted( let (code, output) = session.run(&argv, true).await?; let output = output.map(|o| answer_of(session, o)); if let Some(out) = &output { - print!("{out}"); - if !out.ends_with('\n') { - println!(); - } + crate::stops::answer(out); } let mut after = handoff_ids(&client, None).await?; let wrote = after.difference(&before).next().is_some(); @@ -259,6 +256,12 @@ pub async fn run(session: &Session, options: &Options) -> Fallible { let client = Arc::new(client(session.memory()).await?); let mut last = NOTHING_TO_DO; loop { + // counted from before the claim until the task is settled: a launcher told to + // stop waits, and the task goes back to the pool with a handoff, as when the + // harness stops of itself. One told to stop claims nothing more. + let Ok(in_flight) = crate::stops::InFlight::enter() else { + return Ok(last); + }; let claimed = match &options.task { Some(id) => client.claim(id, options.lease_seconds).await?, // a worker only claims: rank is the ledger's, moved along dependency edges @@ -270,6 +273,7 @@ pub async fn run(session: &Session, options: &Options) -> Fallible { } }; let Some(task) = claimed else { + drop(in_flight); if !options.keep_going { eprintln!("memory: nothing claimable"); return Ok(if last == NOTHING_TO_DO { @@ -281,11 +285,21 @@ pub async fn run(session: &Session, options: &Options) -> Fallible { tokio::time::sleep(options.idle).await; continue; }; + if dm_sandbox::stop::stopping() { + // claimed while the stop came: not started, so back with no handoff + client.release(&task.id, None).await.ok(); + return Ok(last); + } last = one(session, &client, &task, options).await?; + drop(in_flight); if options.task.is_some() { return Ok(last); } - if let Some(plan) = &options.mine { + if let Some(plan) = options + .mine + .as_ref() + .filter(|_| !dm_sandbox::stop::stopping()) + { let via = if plan.via_harness { crate::MineVia::Harness } else { @@ -309,6 +323,7 @@ async fn one( ) -> Fallible { let lease_seconds = options.lease_seconds; eprintln!("memory: claimed {}: {}", task.id, task.title); + let holding = Holding::of(&task.id); let memory = session.memory(); let platform = Platform::for_harness(&memory, session.harness()).await; let prompt = prompt_for( @@ -341,7 +356,16 @@ async fn one( .run_task(&session.harness().headless(&prompt), true, Some(&task.id)) .await; heartbeat.abort(); - let (code, output) = run?; + let (code, output) = match run { + Ok(ran) => ran, + Err(e) => { + // a run refused because a stop came first leaves nothing to settle + if dm_sandbox::stop::stopping() { + client.release(&task.id, None).await.ok(); + } + return Err(e); + } + }; let output = output.map(|o| answer_of(session, o)); let now = client.get_task(&task.id).await?; file_outbox( @@ -353,14 +377,46 @@ async fn one( .await?; let output = output.unwrap_or_default(); if !output.trim().is_empty() { - print!("{output}"); - if !output.ends_with('\n') { - println!(); - } + crate::stops::answer(&output); } + // settled from here: a launcher told twice to stop no longer releases it, as the + // release or completion below may already be done, and the task another's + drop(holding); settle(session.harness(), client, task, code, output).await } +/// The task this worker holds, if any: what a launcher told twice to stop releases +/// before it exits. +static HOLDING: std::sync::Mutex> = std::sync::Mutex::new(None); + +/// The task this worker holds, while it holds one. +pub fn holding() -> Option { + HOLDING + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() +} + +/// A task held, from its claim until it is settled. +struct Holding; + +impl Holding { + fn of(id: &str) -> Holding { + *HOLDING + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(id.to_string()); + Holding + } +} + +impl Drop for Holding { + fn drop(&mut self) { + *HOLDING + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) = None; + } +} + /// After the harness exits: a completed task stays complete; a task still held goes /// back to the pool with the harness's output filed as a handoff. async fn settle( diff --git a/crates/memory/tests/launch.rs b/crates/memory/tests/launch.rs index 722d87d..66262b6 100644 --- a/crates/memory/tests/launch.rs +++ b/crates/memory/tests/launch.rs @@ -856,6 +856,1093 @@ fn a_prompted_launch_runs_headless_and_insists_on_a_handoff() { ); } +/// A confined prompted run captures the harness's standard output through a pipe, which +/// holds 64 KiB. A harness that writes more than that finishes, and all of what it wrote +/// reaches the launcher's own output. Bounded in time: a launch that never returns is +/// this test failing, and its process group is killed so nothing outlives the test. +#[cfg(target_os = "linux")] +#[test] +fn a_confined_prompted_run_captures_more_than_a_pipe_holds() { + use std::os::unix::process::CommandExt; + use std::time::{Duration, Instant}; + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let _reap = Reap(dir.path().to_path_buf()); + let root = init_root(dir.path()); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + let bins = fake_harness( + &work, + "printf 'begin\\n'; head -c 200000 /dev/zero | tr '\\0' a; printf '\\nend\\n'; exit 0", + ); + let stdout = dir.path().join("stdout"); + let stderr = dir.path().join("stderr"); + let mut cmd = memory(); + cmd.args([ + "--root", + root.to_str().unwrap(), + "launch", + "--harness", + "claude", + "--auth", + "key", + ]) + .args(["--workdir", work.to_str().unwrap()]) + .args(["--prompt", "hello", "--no-handoff"]) + .env("PATH", path_with(&bins)) + .env("ANTHROPIC_API_KEY", "sk-ant-placeholder") + .stdin(Stdio::null()) + .stdout(std::fs::File::create(&stdout).unwrap()) + .stderr(std::fs::File::create(&stderr).unwrap()) + .process_group(0); + let mut child = cmd.spawn().unwrap(); + let deadline = Instant::now() + Duration::from_secs(90); + let status = loop { + if let Some(s) = child.try_wait().unwrap() { + break s; + } + if Instant::now() > deadline { + Command::new("kill") + .args(["-KILL", "--", &format!("-{}", child.id())]) + .status() + .ok(); + child.wait().ok(); + panic!( + "the launch had not returned after 90 s, with {} bytes out: {}", + std::fs::metadata(&stdout).map(|m| m.len()).unwrap_or(0), + std::fs::read_to_string(&stderr).unwrap_or_default() + ); + } + std::thread::sleep(Duration::from_millis(100)); + }; + let err = std::fs::read_to_string(&stderr).unwrap(); + assert_eq!(status.code(), Some(0), "{err}"); + assert!(err.contains("confined to"), "{err}"); + let out = std::fs::read_to_string(&stdout).unwrap(); + let want = format!("begin\n{}\nend\n", "a".repeat(200_000)); + assert!( + out.contains(&want), + "the harness's output, whole: {} bytes out", + out.len() + ); +} + +// ---- a launcher told to stop, and one killed ---- + +/// A confined prompted launch, with its temporary +/// directory at `tmp`, so the scratch directory it makes is where the test looks. Its +/// output goes to `.out` and `.err`. +#[cfg(target_os = "linux")] +fn start_confined(root: &Path, work: &Path, tmp: &Path) -> std::process::Child { + start_amem( + root, + work, + tmp, + &["launch", "--prompt", "hello", "--no-handoff"], + ) +} + +/// `amem ` for the claude harness on a key, in `work`, as [`start_confined`] runs it. +#[cfg(target_os = "linux")] +fn start_amem(root: &Path, work: &Path, tmp: &Path, args: &[&str]) -> std::process::Child { + amem_in(root, work, tmp, args) + .stdout(std::fs::File::create(work.with_extension("out")).unwrap()) + .spawn() + .unwrap() +} + +/// As [`start_amem`], with standard output a pipe whose reader is already gone: a +/// launcher piped into a program that has exited. +#[cfg(target_os = "linux")] +fn start_amem_into_a_closed_pipe( + root: &Path, + work: &Path, + tmp: &Path, + args: &[&str], +) -> std::process::Child { + let mut child = amem_in(root, work, tmp, args) + .stdout(Stdio::piped()) + .spawn() + .unwrap(); + drop(child.stdout.take()); + child +} + +/// The command [`start_amem`] runs, its standard output not yet set. +#[cfg(target_os = "linux")] +fn amem_in(root: &Path, work: &Path, tmp: &Path, args: &[&str]) -> Command { + let mut cmd = memory(); + cmd.args(["--root", root.to_str().unwrap()]) + .args(args) + .args(["--harness", "claude", "--auth", "key"]) + .args(["--workdir", work.to_str().unwrap()]); + cmd.env("PATH", path_with(&work.join("bins"))) + .env("ANTHROPIC_API_KEY", "sk-ant-placeholder") + .env("TMPDIR", tmp) + .stdin(Stdio::null()) + .stderr(std::fs::File::create(work.with_extension("err")).unwrap()); + cmd +} + +/// A working tree whose harness says it started, then runs until it is killed. +#[cfg(target_os = "linux")] +fn sleeper(dir: &Path, name: &str) -> PathBuf { + let work = dir.join(name); + std::fs::create_dir_all(&work).unwrap(); + fake_harness(&work, "touch ./started\nsleep 300"); + work +} + +/// A launch whose harness has started and runs until it is killed, and its working tree. +#[cfg(target_os = "linux")] +fn running_sleeper( + root: &Path, + dir: &Path, + name: &str, + tmp: &Path, +) -> (std::process::Child, PathBuf) { + let work = sleeper(dir, name); + let launch = start_confined(root, &work, tmp); + assert!( + started(&work, 60), + "{}", + std::fs::read_to_string(work.with_extension("err")).unwrap() + ); + (launch, work) +} + +/// Send `child` the signal named `sig`, as `kill` names it. +#[cfg(target_os = "linux")] +fn signal(child: &std::process::Child, sig: &str) { + let ok = Command::new("kill") + .args([&format!("-{sig}"), &child.id().to_string()]) + .status() + .unwrap(); + assert!(ok.success()); +} + +/// Wait for `child` to exit, `secs` at most. +#[cfg(target_os = "linux")] +fn exited(child: &mut std::process::Child, secs: u64) -> Option { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs); + while std::time::Instant::now() < deadline { + if let Some(s) = child.try_wait().unwrap() { + return Some(s); + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + None +} + +/// Wait for the harness in `work` to have started, `secs` at most. +#[cfg(target_os = "linux")] +fn started(work: &Path, secs: u64) -> bool { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs); + while std::time::Instant::now() < deadline { + if work.join("started").exists() { + return true; + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + false +} + +/// The processes whose command line names something under `dir`, or whose working +/// directory is under it, as pid and command line. +#[cfg(target_os = "linux")] +fn processes_under(dir: &Path) -> Vec<(String, String)> { + let needle = dir.to_string_lossy().into_owned(); + std::fs::read_dir("/proc") + .unwrap() + .flatten() + .map(|e| e.file_name().to_string_lossy().into_owned()) + .filter(|pid| pid.bytes().all(|b| b.is_ascii_digit())) + .filter_map(|pid| { + let at = Path::new("/proc").join(&pid); + let c = std::fs::read(at.join("cmdline")).ok()?; + let c = String::from_utf8_lossy(&c).replace('\0', " "); + // or working there: a harness's child whose command line names nothing + let cwd = std::fs::read_link(at.join("cwd")).ok(); + (c.contains(&needle) || cwd.is_some_and(|d| d.starts_with(dir))).then_some((pid, c)) + }) + .collect() +} + +/// Whether no process names anything under `dir` within `secs`. +#[cfg(target_os = "linux")] +fn gone_within(dir: &Path, secs: u64) -> bool { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs); + while std::time::Instant::now() < deadline { + if processes_under(dir).is_empty() { + return true; + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + false +} + +/// The mounts under `dir`, from this process's mount table: no mountpoint is touched, +/// since one a launcher left may not answer. +#[cfg(target_os = "linux")] +fn mounts_under(dir: &Path) -> Vec { + let needle = format!("{}/", dir.display()); + std::fs::read_to_string("/proc/self/mountinfo") + .unwrap() + .lines() + .filter_map(|l| l.split(' ').nth(4)) + .filter(|m| m.starts_with(&needle)) + .map(str::to_string) + .collect() +} + +/// The names in `dir`, listed without a stat of any of them. +#[cfg(target_os = "linux")] +fn names_in(dir: &Path) -> Vec { + let mut names: Vec = std::fs::read_dir(dir) + .map(|d| { + d.flatten() + .map(|e| e.file_name().to_string_lossy().into_owned()) + .collect() + }) + .unwrap_or_default(); + names.sort(); + names +} + +/// Kills, when dropped, every process whose command line names something under its +/// directory, and lazily unmounts what is left there: a test that fails leaves nothing +/// running or mounted. A run's harness goes with its relay, pid 1 of its namespace. +#[cfg(target_os = "linux")] +struct Reap(PathBuf); + +#[cfg(target_os = "linux")] +impl Drop for Reap { + fn drop(&mut self) { + for (pid, _) in processes_under(&self.0) { + Command::new("kill") + .args(["-KILL", &pid]) + .stderr(Stdio::null()) + .status() + .ok(); + } + std::thread::sleep(std::time::Duration::from_millis(500)); + for m in mounts_under(&self.0) { + Command::new("fusermount3").args(["-uz", &m]).status().ok(); + } + } +} + +/// What a launch leaves: the processes of its run, its mount, its scratch directory, +/// its socket and its render. A SIGTERM ends the run and removes every one, as a run +/// that ends of itself does, and the launcher exits as a process the signal ended. +#[cfg(target_os = "linux")] +#[test] +fn a_launcher_told_to_stop_ends_its_run_and_leaves_nothing() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let _reap = Reap(dir.path().to_path_buf()); + let (mut launch, work) = running_sleeper(&root, dir.path(), "work", &tmp); + assert!( + !mounts_under(&root).is_empty(), + "the projection is a live mount" + ); + assert_eq!(names_in(&tmp).len(), 1, "the run has its scratch directory"); + + signal(&launch, "TERM"); + let status = exited(&mut launch, 30); + + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 15), "{err}"); + // said once, as the stop; the kill that ends the sandbox is not a death to report + assert_eq!(err.matches("SIGTERM").count(), 1, "{err}"); + assert!(!err.contains("killed by"), "{err}"); + assert_eq!(processes_under(dir.path()), Vec::new()); + assert_eq!(mounts_under(&root), Vec::::new()); + assert_eq!( + names_in(&tmp), + Vec::::new(), + "the scratch directory" + ); + assert_eq!(names_in(&root.join("state/run")), Vec::::new()); + assert_eq!( + names_in(&root.join("state/mount/claude")), + Vec::::new() + ); + assert_eq!( + names_in(&root.join("state/projection/claude")), + Vec::::new() + ); +} + +/// SIGKILL `killed`, a running launch in `work`, and check what it leaves: no process of +/// its run, which dies with it, and its scratch directory, socket and mount. +#[cfg(target_os = "linux")] +fn killed_by_sigkill(root: &Path, tmp: &Path, killed: &mut std::process::Child, work: &Path) { + signal(killed, "KILL"); + killed.wait().unwrap(); + // the sandbox dies with its launcher: the intermediate, the relay, the harness + assert!(gone_within(work, 5), "{:?}", processes_under(work)); + let killed_socket = format!("egress-{}.sock", killed.id()); + assert!(names_in(&root.join("state/run")).contains(&killed_socket)); + + let killed_scratch = format!("memory-sandbox-{}", killed.id()); + assert!(names_in(tmp).contains(&killed_scratch)); + assert_eq!( + mounts_under(root).len(), + 2, + "the killed launch's mount is left" + ); +} + +/// SIGKILL cannot be caught: a launcher killed by it leaves its scratch directory and its +/// mount, with nothing running to serve it. The next launch sweeps both, and what a +/// killed launcher of long ago left, and keeps what a live launcher holds. +#[cfg(target_os = "linux")] +#[test] +fn the_next_launch_sweeps_what_a_killed_one_left_and_keeps_a_live_one_s() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + // left by a launcher long gone: its pid is no process now + let mut gone = Command::new("/bin/sh").args(["-c", ":"]).spawn().unwrap(); + let gone_pid = gone.id(); + gone.wait().unwrap(); + std::fs::create_dir(tmp.join(format!("memory-sandbox-{gone_pid}"))).unwrap(); + let _reap = Reap(dir.path().to_path_buf()); + let (mut live, _) = running_sleeper(&root, dir.path(), "live", &tmp); + let live_mounts = mounts_under(&root); + assert_eq!(live_mounts.len(), 1); + let (mut killed, killed_work) = running_sleeper(&root, dir.path(), "killed", &tmp); + assert_eq!(mounts_under(&root).len(), 2); + killed_by_sigkill(&root, &tmp, &mut killed, &killed_work); + + let next_work = dir.path().join("next"); + std::fs::create_dir_all(&next_work).unwrap(); + fake_harness(&next_work, "exit 0"); + let mut next = start_confined(&root, &next_work, &tmp); + let status = exited(&mut next, 60); + let err = std::fs::read_to_string(next_work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(0), "{err}"); + + assert_eq!( + names_in(&tmp), + vec![format!("memory-sandbox-{}", live.id())], + "the killed launch's scratch directory and the old one are swept" + ); + assert_eq!( + mounts_under(&root), + live_mounts, + "the killed launch's mount is swept and the live launch's stays" + ); + let live_session = Path::new(&live_mounts[0]).file_name().unwrap(); + let live_session = live_session.to_string_lossy().into_owned(); + assert_eq!( + names_in(&root.join("state/mount/claude")), + vec![live_session.clone(), format!("{live_session}.lock")], + "the live launch's mountpoint and its record" + ); + assert_eq!( + names_in(&root.join("state/run")), + vec![format!("egress-{}.sock", live.id())], + "the killed launch's socket is swept and the live launch's stays" + ); + assert!( + live.try_wait().unwrap().is_none(), + "the live launch runs on" + ); + // and it still ends cleanly + signal(&live, "TERM"); + assert_eq!(exited(&mut live, 30).and_then(|s| s.code()), Some(128 + 15)); + assert_eq!(mounts_under(&root), Vec::::new()); + assert_eq!(names_in(&tmp), Vec::::new()); +} + +/// A terminal that hangs up sends SIGHUP to its session's leader, and refuses every write +/// after it. The launcher still ends the run and cleans up: no panic on a write to the +/// terminal that is gone stops it first. +#[cfg(target_os = "linux")] +#[test] +fn a_launcher_whose_terminal_hangs_up_still_cleans_up() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + if !script_available() { + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let work = sleeper(dir.path(), "work"); + let _reap = Reap(dir.path().to_path_buf()); + // the launcher as the session leader of a terminal of its own, which script makes + let line = format!( + "exec {} --root {} launch --prompt hello --no-handoff --harness claude --auth key --workdir {}", + env!("CARGO_BIN_EXE_amem"), + root.display(), + work.display() + ); + let mut terminal = Command::new("script") + .args(["-qfec", &line, "/dev/null"]) + .env("PATH", path_with(&work.join("bins"))) + .env("ANTHROPIC_API_KEY", "sk-ant-placeholder") + .env("TMPDIR", &tmp) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + assert!(started(&work, 60)); + assert!(!mounts_under(&root).is_empty()); + + // the terminal goes: its other end closes, and the kernel hangs it up + signal(&terminal, "KILL"); + terminal.wait().unwrap(); + + assert!( + gone_within(dir.path(), 30), + "{:?}", + processes_under(dir.path()) + ); + assert_eq!(mounts_under(&root), Vec::::new()); + assert_eq!( + names_in(&tmp), + Vec::::new(), + "the scratch directory" + ); + assert_eq!( + names_in(&root.join("state/run")), + Vec::::new(), + "the socket" + ); + assert_eq!( + names_in(&root.join("state/mount/claude")), + Vec::::new() + ); + assert_eq!( + names_in(&root.join("state/projection/claude")), + Vec::::new(), + "the render" + ); +} + +/// An unconfined run is told to stop as a confined one is: at once, with 143. +#[cfg(target_os = "linux")] +#[test] +fn an_unconfined_launcher_told_to_stop_ends_its_run_at_once() { + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + // exec, so the process the launcher spawned is the one that sleeps + fake_harness(&work, "touch ./started\nexec sleep 300"); + let _reap = Reap(dir.path().to_path_buf()); + let mut launch = start_amem( + &root, + &work, + &tmp, + &[ + "launch", + "--unconfined", + "--prompt", + "hello", + "--no-handoff", + ], + ); + assert!( + started(&work, 60), + "{}", + std::fs::read_to_string(work.with_extension("err")).unwrap() + ); + signal(&launch, "TERM"); + let status = exited(&mut launch, 10); + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 15), "{err}"); + assert_eq!(names_in(&root.join("state/run")), Vec::::new()); +} + +/// A launcher that is stopped (Ctrl-Z) does not answer for its mount. Another launch +/// does not ask it anything: it starts, runs and ends, and the stopped launcher's mount +/// is kept for it. +#[cfg(target_os = "linux")] +#[test] +fn a_stopped_launcher_s_mount_is_kept_and_not_asked() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let _reap = Reap(dir.path().to_path_buf()); + let (mut stopped, _) = running_sleeper(&root, dir.path(), "stopped", &tmp); + let its_mounts = mounts_under(&root); + assert_eq!(its_mounts.len(), 1); + signal(&stopped, "STOP"); + + let next_work = dir.path().join("next"); + std::fs::create_dir_all(&next_work).unwrap(); + fake_harness(&next_work, "exit 0"); + let mut next = start_confined(&root, &next_work, &tmp); + let status = exited(&mut next, 60); + signal(&stopped, "CONT"); + let err = std::fs::read_to_string(next_work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(0), "{err}"); + assert_eq!( + mounts_under(&root), + its_mounts, + "the stopped launcher's mount stays" + ); + + signal(&stopped, "TERM"); + assert_eq!( + exited(&mut stopped, 30).and_then(|s| s.code()), + Some(128 + 15) + ); + assert_eq!(mounts_under(&root), Vec::::new()); +} + +/// `amem work --loop` told to stop mid-run settles the task it holds, as when the harness +/// stops of itself: back in the pool with a handoff. It claims nothing more. +#[cfg(target_os = "linux")] +#[test] +fn a_worker_told_to_stop_releases_its_task_with_a_handoff_and_claims_no_other() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let ids = [file_task(&root, "First"), file_task(&root, "Second")]; + let work = sleeper(dir.path(), "work"); + let _reap = Reap(dir.path().to_path_buf()); + let mut worker = start_amem( + &root, + &work, + &tmp, + &["work", "--loop", "--lease-seconds", "60"], + ); + assert!( + started(&work, 60), + "{}", + std::fs::read_to_string(work.with_extension("err")).unwrap() + ); + signal(&worker, "TERM"); + let status = exited(&mut worker, 30); + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 15), "{err}"); + + let claimed: Vec<&String> = ids + .iter() + .filter(|id| err.contains(&format!("claimed {id}"))) + .collect(); + assert_eq!(claimed.len(), 1, "one task claimed: {err}"); + for id in &ids { + let t = cli(&root, &["task", "get", id]); + assert_eq!(t["status"], "admitted", "in the pool: {t}"); + if claimed.contains(&id) { + assert!(t["handoff"].as_str().is_some(), "with a handoff: {t}"); + } else { + assert!(t["handoff"].is_null(), "never claimed: {t}"); + } + } +} + +/// A worker in `dir/work` whose harness prints, says it started, then runs until killed; +/// one task filed for it. +#[cfg(target_os = "linux")] +fn a_task_and_a_printing_sleeper(root: &Path, dir: &Path) -> (String, PathBuf) { + let id = file_task(root, "First"); + let work = dir.join("work"); + std::fs::create_dir_all(&work).unwrap(); + fake_harness(&work, "echo working\ntouch ./started\nsleep 300"); + (id, work) +} + +/// Whether task `id` is back in the pool, and with a handoff. +#[cfg(target_os = "linux")] +fn released(root: &Path, id: &str) -> (bool, bool) { + let t = cli(root, &["task", "get", id]); + (t["status"] == "admitted", t["handoff"].as_str().is_some()) +} + +/// A launcher piped into a reader that has gone: the answer it prints after a stop +/// goes nowhere, and the worker settles its task and the launch finishes all the same. +#[cfg(target_os = "linux")] +#[test] +fn a_launcher_whose_output_reader_has_gone_still_settles_and_cleans_up() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let (id, work) = a_task_and_a_printing_sleeper(&root, dir.path()); + let _reap = Reap(dir.path().to_path_buf()); + for args in [ + &["work", "--loop", "--lease-seconds", "60"][..], + &["launch", "--prompt", "hello", "--no-handoff"][..], + ] { + std::fs::remove_file(work.join("started")).ok(); + let mut launch = start_amem_into_a_closed_pipe(&root, &work, &tmp, args); + assert!(started(&work, 60), "{args:?}"); + signal(&launch, "TERM"); + let status = exited(&mut launch, 30); + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert_eq!( + status.and_then(|s| s.code()), + Some(128 + 15), + "{args:?}: {err}" + ); + assert_eq!( + names_in(&root.join("state/run")), + Vec::::new(), + "{err}" + ); + assert_eq!( + names_in(&root.join("state/projection/claude")), + Vec::::new() + ); + } + assert_eq!(released(&root, &id), (true, true), "back with a handoff"); +} + +/// A launcher started to ignore SIGHUP, under `nohup` or by a shell without job control, +/// goes on ignoring it: the run is not stopped. +#[cfg(target_os = "linux")] +#[test] +fn a_launcher_started_to_ignore_sighup_is_not_stopped_by_it() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let work = sleeper(dir.path(), "work"); + let _reap = Reap(dir.path().to_path_buf()); + let line = format!( + "trap '' HUP; exec {} --root {} launch --prompt hello --no-handoff --harness claude --auth key --workdir {}", + env!("CARGO_BIN_EXE_amem"), + root.display(), + work.display() + ); + let mut launch = Command::new("/bin/sh") + .args(["-c", &line]) + .env("PATH", path_with(&work.join("bins"))) + .env("ANTHROPIC_API_KEY", "sk-ant-placeholder") + .env("TMPDIR", &tmp) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(std::fs::File::create(work.with_extension("err")).unwrap()) + .spawn() + .unwrap(); + assert!(started(&work, 60)); + + signal(&launch, "HUP"); + std::thread::sleep(std::time::Duration::from_secs(1)); + + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert!( + launch.try_wait().unwrap().is_none(), + "the run goes on: {err}" + ); + assert_eq!(mounts_under(&root).len(), 1); + signal(&launch, "TERM"); + assert_eq!( + exited(&mut launch, 30).and_then(|s| s.code()), + Some(128 + 15) + ); +} + +/// A headless unconfined run that left a child in the background, holding its output +/// open: a stop kills the child too, and the worker settles its task and exits. +#[cfg(target_os = "linux")] +#[test] +fn an_unconfined_worker_told_to_stop_ends_what_its_harness_left_running() { + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let id = file_task(&root, "First"); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + fake_harness( + &work, + "touch ./started\nsh -c \"sleep 300; echo $PWD\" &\nwait", + ); + let _reap = Reap(dir.path().to_path_buf()); + let mut worker = start_amem( + &root, + &work, + &tmp, + &["work", "--unconfined", "--loop", "--lease-seconds", "60"], + ); + assert!( + started(&work, 60), + "{}", + std::fs::read_to_string(work.with_extension("err")).unwrap() + ); + signal(&worker, "TERM"); + let status = exited(&mut worker, 15); + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 15), "{err}"); + assert!(gone_within(&work, 5), "{:?}", processes_under(&work)); + assert_eq!(released(&root, &id), (true, true), "back with a handoff"); +} + +/// Whether script(1) is here to make a terminal with; on CI it must be. +#[cfg(target_os = "linux")] +fn script_available() -> bool { + let here = Command::new("script").arg("--version").output().is_ok(); + assert!( + here || std::env::var_os("CI").is_none(), + "util-linux script makes the terminal here, and on CI it must be there" + ); + if !here { + eprintln!("skipping: no script(1) to make a terminal with"); + } + here +} + +/// `amem ` for the claude harness in `work`, as the session leader of a terminal +/// of its own that script(1) makes, its standard error to `.err`. script exits +/// with the launcher's code, and keeps what the terminal showed in `.tty`. +#[cfg(target_os = "linux")] +fn on_a_terminal(root: &Path, work: &Path, tmp: &Path, args: &str) -> std::process::Child { + let line = format!( + "exec {} --root {} {args} --harness claude --auth key --workdir {} 2>{}", + env!("CARGO_BIN_EXE_amem"), + root.display(), + work.display(), + work.with_extension("err").display() + ); + Command::new("script") + .args(["-qfec", &line]) + .arg(work.with_extension("tty")) + .env("PATH", path_with(&work.join("bins"))) + .env("ANTHROPIC_API_KEY", "sk-ant-placeholder") + .env("TMPDIR", tmp) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .unwrap() +} + +/// The pid of the launcher running on `root`. +#[cfg(target_os = "linux")] +fn launcher_of(root: &Path) -> String { + let bin = env!("CARGO_BIN_EXE_amem"); + let found: Vec = processes_under(root) + .into_iter() + .filter(|(_, c)| c.starts_with(bin) && c.contains(" --root ")) + .map(|(pid, _)| pid) + .collect(); + assert_eq!(found.len(), 1, "{found:?}"); + found[0].clone() +} + +/// A headless unconfined run on a terminal stays in the terminal's foreground group: a +/// harness that sets the terminal, as a password prompt does, is not stopped for it. +#[cfg(target_os = "linux")] +#[test] +fn a_headless_unconfined_run_on_a_terminal_may_use_the_terminal() { + if !script_available() { + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + fake_harness( + &work, + "stty -echo (String, PathBuf, std::process::Child) { + let id = file_task(root, "First"); + let work = dir.join("work"); + std::fs::create_dir_all(&work).unwrap(); + fake_harness( + &work, + "touch ./started\nsh -c \"sleep 300; echo $PWD\" &\nwait", + ); + let terminal = on_a_terminal( + root, + &work, + tmp, + "work --unconfined --loop --lease-seconds 60", + ); + assert!( + started(&work, 60), + "{}", + std::fs::read_to_string(work.with_extension("err")).unwrap_or_default() + ); + (id, work, terminal) +} + +/// On a terminal the harness is in the terminal's group, which a stop does not kill +/// whole; what it left in the background, holding the output open, does not keep the +/// worker from settling its task and exiting. +#[cfg(target_os = "linux")] +#[test] +fn a_worker_on_a_terminal_told_to_stop_ends_though_its_harness_left_a_child() { + if !script_available() { + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let _reap = Reap(dir.path().to_path_buf()); + let (id, work, mut terminal) = + a_worker_on_a_terminal_with_a_background_child(&root, dir.path(), &tmp); + let launcher = launcher_of(&root); + Command::new("kill") + .args(["-TERM", &launcher]) + .status() + .unwrap(); + let status = exited(&mut terminal, 15); + let err = std::fs::read_to_string(work.with_extension("err")).unwrap_or_default(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 15), "{err}"); + assert_eq!(released(&root, &id), (true, true), "back with a handoff"); +} + +/// A headless unconfined harness that exits at once, leaving a child that holds its +/// output and ignores it: the run is still in flight while its output is read, so a stop +/// then kills the child with the harness's process group. +#[cfg(target_os = "linux")] +#[test] +fn a_stop_kills_what_an_exited_harness_left_holding_its_output() { + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let id = file_task(&root, "First"); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + fake_harness( + &work, + "sh -c \"sleep 300; echo $PWD\" &\ntouch ./started\nexit 0", + ); + let _reap = Reap(dir.path().to_path_buf()); + let mut worker = start_amem( + &root, + &work, + &tmp, + &["work", "--unconfined", "--loop", "--lease-seconds", "60"], + ); + assert!(started(&work, 60)); + // the harness has gone; its child holds the output + std::thread::sleep(std::time::Duration::from_millis(300)); + let child = |dir: &Path| { + processes_under(dir) + .into_iter() + .any(|(_, c)| c.contains("sleep 300; echo")) + }; + assert!(child(&work), "{:?}", processes_under(&work)); + signal(&worker, "TERM"); + let status = exited(&mut worker, 15); + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 15), "{err}"); + assert!(gone_within(&work, 5), "{:?}", processes_under(&work)); + assert_eq!(released(&root, &id), (true, true), "back with a handoff"); +} + +/// Whether the harness in `work`, which wrote its pid to `./pid`, has exited within +/// `secs`: gone, or a zombie its parent has not reaped. +#[cfg(target_os = "linux")] +fn exited_unreaped(work: &Path, secs: u64) -> bool { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(secs); + while std::time::Instant::now() < deadline { + if let Ok(pid) = std::fs::read_to_string(work.join("pid")) { + let stat = std::fs::read_to_string(format!("/proc/{}/stat", pid.trim())); + // the state follows the command name, which ends at the last ')' + let state = stat + .as_deref() + .ok() + .and_then(|s| s.rsplit_once(')')) + .and_then(|(_, rest)| rest.trim_start().chars().next()); + if matches!(state, None | Some('Z')) { + return true; + } + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + false +} + +/// A harness killed by SIGKILL before any stop, an out-of-memory kill say, is said to +/// have been, though a stop comes while what it left still holds its output. +#[cfg(target_os = "linux")] +#[test] +fn a_harness_killed_before_a_stop_is_said_to_have_been_killed() { + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + fake_harness( + &work, + "sh -c \"sleep 300; echo $PWD\" &\necho $$ > ./pid\ntouch ./started\nkill -9 $$", + ); + let _reap = Reap(dir.path().to_path_buf()); + let mut launch = start_amem( + &root, + &work, + &tmp, + &[ + "launch", + "--unconfined", + "--prompt", + "hello", + "--no-handoff", + ], + ); + assert!(started(&work, 60)); + // the stop after the harness's death, which leaves it a zombie until it is reaped + assert!(exited_unreaped(&work, 30), "the harness did not die"); + signal(&launch, "TERM"); + let status = exited(&mut launch, 15); + let err = std::fs::read_to_string(work.with_extension("err")).unwrap(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 15), "{err}"); + assert!(err.contains("was killed by SIGKILL"), "{err}"); +} + +/// Told twice to stop, a worker does not wait for the run's cleanup, but it leaves +/// nothing no sweep removes: its task back in the pool, its render and its socket gone. +/// The second signal comes as soon as the first is said, before the task is settled: the +/// task comes back with no handoff, which only this path does. +#[cfg(target_os = "linux")] +#[test] +fn a_worker_told_twice_to_stop_releases_its_task_and_removes_its_render() { + if !script_available() { + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let _reap = Reap(dir.path().to_path_buf()); + let (id, work, mut terminal) = + a_worker_on_a_terminal_with_a_background_child(&root, dir.path(), &tmp); + use nix::sys::signal::{kill, Signal}; + let launcher = nix::unistd::Pid::from_raw(launcher_of(&root).parse().unwrap()); + kill(launcher, Signal::SIGINT).unwrap(); + // the second from the loop that sees the first in progress, with no process to + // start between: it comes a millisecond or so after the stop is said, before the + // cleanup that follows it, the reap and the settling of the task, is done. The read + // of the output that what the harness left holds open can only make that longer. + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10); + let read_err = || std::fs::read_to_string(work.with_extension("err")).unwrap_or_default(); + while !read_err().contains("ending the run and cleaning up") { + assert!(std::time::Instant::now() < deadline, "{}", read_err()); + std::thread::sleep(std::time::Duration::from_millis(1)); + } + kill(launcher, Signal::SIGINT).unwrap(); + let status = exited(&mut terminal, 30); + let err = read_err(); + assert_eq!(status.and_then(|s| s.code()), Some(128 + 2), "{err}"); + assert_eq!( + released(&root, &id), + (true, false), + "back, with no handoff: {err}" + ); + assert_eq!( + names_in(&root.join("state/projection/claude")), + Vec::::new(), + "the render" + ); + assert_eq!( + names_in(&root.join("state/run")), + Vec::::new(), + "the socket" + ); +} + +/// Under umask 000 the launcher's own scratch directory is still its own: a confined run +/// is not refused for it. +#[cfg(target_os = "linux")] +#[test] +fn a_confined_run_under_umask_000_runs() { + if !namespaces_available() { + eprintln!("skipping: unprivileged namespaces are not available here"); + return; + } + let dir = tempfile::tempdir().unwrap(); + let root = init_root(dir.path()); + let tmp = dir.path().join("tmp"); + std::fs::create_dir_all(&tmp).unwrap(); + let work = dir.path().join("work"); + std::fs::create_dir_all(&work).unwrap(); + fake_harness(&work, "exit 0"); + let _reap = Reap(dir.path().to_path_buf()); + let line = format!( + "umask 000; exec {} --root {} launch --prompt hello --no-handoff --harness claude --auth key --workdir {}", + env!("CARGO_BIN_EXE_amem"), + root.display(), + work.display() + ); + let out = Command::new("/bin/sh") + .args(["-c", &line]) + .env("PATH", path_with(&work.join("bins"))) + .env("ANTHROPIC_API_KEY", "sk-ant-placeholder") + .env("TMPDIR", &tmp) + .stdin(Stdio::null()) + .output() + .unwrap(); + let err = String::from_utf8_lossy(&out.stderr); + assert_eq!(out.status.code(), Some(0), "{err}"); + assert!(err.contains("confined to"), "{err}"); +} + // ---- the wslc backend's inside half, driven by a fake host ---- /// `memory inside` writes run requests for the host and waits for exit codes. This fake diff --git a/crates/mount/Cargo.toml b/crates/mount/Cargo.toml index 2aeec49..2f7d5c1 100644 --- a/crates/mount/Cargo.toml +++ b/crates/mount/Cargo.toml @@ -15,6 +15,7 @@ tokio = { workspace = true, features = ["rt"] } [target.'cfg(target_os = "linux")'.dependencies] fuser = { version = "0.18.0", default-features = false } +nix = { version = "0.31", features = ["fs"] } [dev-dependencies] tempfile = "3.27.0" diff --git a/crates/mount/src/lib.rs b/crates/mount/src/lib.rs index c6a0bd5..08d706f 100644 --- a/crates/mount/src/lib.rs +++ b/crates/mount/src/lib.rs @@ -149,6 +149,9 @@ pub struct Mounted { at: PathBuf, #[cfg(target_os = "linux")] session: Option, + /// Dropped after the mount and its mountpoint are gone. + #[cfg(target_os = "linux")] + _owner: linux::Owner, } impl Mounted { @@ -186,12 +189,15 @@ pub fn mount(memory: Arc, render: &Path, at: &Path) -> Result, render: &Path, at: &Path) -> Result Vec { let mut cleared = Vec::new(); + if is_link(base) { + return cleared; + } let Ok(entries) = std::fs::read_dir(base) else { return cleared; }; for entry in entries.flatten() { + // the kind as the directory lists it, without a stat of a mountpoint + if entry.file_type().is_ok_and(|t| t.is_symlink()) { + continue; + } let path = entry.path(); #[cfg(target_os = "linux")] - linux::clear_dead(&path); + if let Some(swept) = sweep_recorded(&path) { + cleared.extend(swept); + continue; + } if std::fs::remove_dir(&path).is_ok() { cleared.push(path); } @@ -220,6 +238,63 @@ pub fn sweep(base: &Path) -> Vec { cleared } +/// Sweep `path` by its record: `None` for a mountpoint from before records were kept, +/// which the caller removes when it is empty, as then. +#[cfg(target_os = "linux")] +fn sweep_recorded(path: &Path) -> Option> { + let name = path.file_name()?.to_str()?; + if let Some(at) = name.strip_suffix(".lock") { + // a record whose launcher died before it made the mountpoint + let at = path.with_file_name(at); + if let linux::Held::Gone(lock) = linux::held(&at) { + let no_mountpoint = std::fs::symlink_metadata(&at) + .is_err_and(|e| e.kind() == std::io::ErrorKind::NotFound); + if no_mountpoint { + std::fs::remove_file(path).ok(); + } + drop(lock); + } + return Some(None); + } + match linux::held(path) { + linux::Held::Live => Some(None), + linux::Held::Gone(lock) => { + linux::clear_dead(path); + let removed = std::fs::remove_dir(path).is_ok(); + std::fs::remove_file(linux::lock_path(path)).ok(); + drop(lock); + Some(removed.then(|| path.to_path_buf())) + } + linux::Held::Unrecorded => { + linux::clear_dead(path); + None + } + } +} + +/// [`sweep`] every harness's mount directory under `root`'s [`MOUNTS`], so a launch of +/// one harness clears what killed launches of another left. Returns the directories +/// cleared. +pub fn sweep_root(root: &Path) -> Vec { + let mounts = root.join(MOUNTS); + if is_link(&mounts) { + return Vec::new(); + } + let Ok(entries) = std::fs::read_dir(&mounts) else { + return Vec::new(); + }; + entries + .flatten() + .filter(|e| e.file_type().is_ok_and(|t| t.is_dir())) + .flat_map(|e| sweep(&e.path())) + .collect() +} + +/// Whether `path` is a symbolic link, which a sweep does not follow. +fn is_link(path: &Path) -> bool { + std::fs::symlink_metadata(path).is_ok_and(|m| m.file_type().is_symlink()) +} + #[cfg(test)] mod tests { use super::*; @@ -273,4 +348,78 @@ mod tests { "no base, nothing to do" ); } + + #[cfg(unix)] + #[test] + fn a_sweep_reaches_every_harness_and_follows_no_link() { + let root = tempfile::tempdir().unwrap(); + let outside = tempfile::tempdir().unwrap(); + let empty_outside = outside.path().join("empty"); + std::fs::create_dir(&empty_outside).unwrap(); + let session_outside = outside.path().join("h/sess-x"); + std::fs::create_dir_all(&session_outside).unwrap(); + let mounts = root.path().join(MOUNTS); + for harness in ["claude", "codex"] { + std::fs::create_dir_all(mounts.join(harness).join("sess-gone")).unwrap(); + } + // links in a harness's directory, and a harness directory that is a link, to + // directories outside the root that a followed link would clear + std::os::unix::fs::symlink(&empty_outside, mounts.join("claude/sess-link")).unwrap(); + std::os::unix::fs::symlink(outside.path(), mounts.join("gemini")).unwrap(); + + let mut cleared = sweep_root(root.path()); + cleared.sort(); + + assert_eq!( + cleared, + vec![ + mounts.join("claude/sess-gone"), + mounts.join("codex/sess-gone") + ] + ); + assert!(empty_outside.is_dir(), "a link is not followed"); + assert!(mounts.join("claude/sess-link").is_symlink()); + // a root whose mount directory is a link is not swept at all + let linked_root = tempfile::tempdir().unwrap(); + std::fs::create_dir(linked_root.path().join("state")).unwrap(); + std::os::unix::fs::symlink(outside.path(), linked_root.path().join(MOUNTS)).unwrap(); + assert!(sweep_root(linked_root.path()).is_empty()); + assert!(empty_outside.is_dir()); + assert!(session_outside.is_dir()); + } + + #[cfg(target_os = "linux")] + #[test] + fn a_sweep_keeps_what_a_live_launcher_s_record_holds_and_clears_a_dead_one_s() { + let dir = tempfile::tempdir().unwrap(); + let base = dir.path().join("claude"); + std::fs::create_dir_all(&base).unwrap(); + // a live launcher: it holds its record, its mountpoint may not answer + let live = base.join("sess-live"); + let owner = linux::own(&live).unwrap(); + std::fs::create_dir(&live).unwrap(); + // one about to make its mountpoint: the record is there, the directory not yet + let coming = linux::own(&base.join("sess-coming")).unwrap(); + // a dead launcher's record and mountpoint, and a record without a mountpoint + let dead = base.join("sess-dead"); + std::fs::write(linux::lock_path(&dead), "").unwrap(); + std::fs::create_dir(&dead).unwrap(); + let stray = linux::lock_path(&base.join("sess-stray")); + std::fs::write(&stray, "").unwrap(); + + let cleared = sweep(&base); + + assert_eq!(cleared, vec![dead.clone()]); + assert!(!linux::lock_path(&dead).exists()); + assert!(!stray.exists()); + assert!(live.is_dir(), "a live launcher's mountpoint is kept"); + assert!(linux::lock_path(&live).exists()); + assert!(linux::lock_path(&base.join("sess-coming")).exists()); + drop(owner); + drop(coming); + assert!( + !linux::lock_path(&live).exists(), + "the record goes with its owner" + ); + } } diff --git a/crates/mount/src/linux.rs b/crates/mount/src/linux.rs index 66341f9..0cc2df4 100644 --- a/crates/mount/src/linux.rs +++ b/crates/mount/src/linux.rs @@ -10,6 +10,7 @@ use fuser::{ LockOwner, MountOption, OpenAccMode, OpenFlags, ReplyAttr, ReplyData, ReplyDirectory, ReplyEntry, ReplyOpen, Request, }; +use nix::fcntl::{Flock, FlockArg}; use serde_json::json; use std::collections::HashMap; use std::ffi::OsStr; @@ -54,10 +55,113 @@ pub(crate) fn availability() -> Result<(), String> { } } +/// The record that a launcher serves the mount at `at`: an exclusive lock on the file +/// beside it, `.lock`, held from before the mountpoint is made until after it is +/// gone. The kernel lets the lock go when the launcher dies, however it dies, so a +/// sweep can tell a dead launcher's mount from a live one's without asking the mount, +/// which a live launcher that is stopped does not answer. +pub(crate) struct Owner { + path: PathBuf, + _lock: Flock, +} + +impl Drop for Owner { + fn drop(&mut self) { + // removed while still locked, so no sweep takes a lock on a file going away + std::fs::remove_file(&self.path).ok(); + } +} + +/// The lock file of the mount at `at`. +pub(crate) fn lock_path(at: &Path) -> PathBuf { + let mut name = at.as_os_str().to_owned(); + name.push(".lock"); + PathBuf::from(name) +} + +fn open_lock(path: &Path, create: bool) -> std::io::Result { + use std::os::unix::fs::OpenOptionsExt; + std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(create) + .truncate(false) + .custom_flags(nix::fcntl::OFlag::O_NOFOLLOW.bits()) + .open(path) +} + +/// Whether the file at `path` is still the one `file` has open. +fn still_at(path: &Path, file: &std::fs::File) -> bool { + match (std::fs::symlink_metadata(path), file.metadata()) { + (Ok(at), Ok(held)) => at.dev() == held.dev() && at.ino() == held.ino(), + _ => false, + } +} + +/// Take the record for a mount at `at`, before its mountpoint is made. +pub(crate) fn own(at: &Path) -> Result { + let path = lock_path(at); + let failed = |source| Error::Io { + path: path.display().to_string(), + source, + }; + if let Some(dir) = path.parent() { + std::fs::create_dir_all(dir).map_err(|e| Error::Io { + path: dir.display().to_string(), + source: e, + })?; + } + // twice: a sweep can remove a free record between its open here and its lock + for _ in 0..2 { + let file = open_lock(&path, true).map_err(failed)?; + let lock = Flock::lock(file, FlockArg::LockExclusiveNonblock) + .map_err(|(_, e)| failed(std::io::Error::from(e)))?; + if still_at(&path, &lock) { + return Ok(Owner { path, _lock: lock }); + } + } + Err(failed(std::io::Error::other( + "removed while it was being locked", + ))) +} + +/// What the record of the mount at `at` says. +pub(crate) enum Held { + /// A launcher holds it: its mount is not to be touched, nor asked anything. + Live, + /// No launcher holds it; the lock is taken here, for as long as this lives. + Gone(Flock), + /// There is none: a mount from before records were kept. + Unrecorded, +} + +/// Read the record of the mount at `at`. +pub(crate) fn held(at: &Path) -> Held { + let path = lock_path(at); + let Ok(file) = open_lock(&path, false) else { + return if path.exists() || path.is_symlink() { + Held::Live + } else { + Held::Unrecorded + }; + }; + match Flock::lock(file, FlockArg::LockExclusiveNonblock) { + Ok(lock) if still_at(&path, &lock) => Held::Gone(lock), + _ => Held::Live, + } +} + /// Whether `at` is a mount whose server is gone: a killed launcher leaves one, and every -/// call on it fails with "transport endpoint is not connected". +/// call on it fails with "transport endpoint is not connected". The server is the +/// launcher, so this is the record of whether its launcher lives. A link at `at` is not +/// followed, and is not a dead mount. fn is_dead(at: &Path) -> bool { - std::fs::metadata(at) + if std::fs::symlink_metadata(at).is_ok_and(|m| m.file_type().is_symlink()) { + return false; + } + // an open, which always reaches the server: the kernel answers a stat from its + // cache for as long as the attributes' TTL after the server has gone + std::fs::read_dir(at) .err() .is_some_and(|e| e.raw_os_error() == Some(libc_enotconn())) } diff --git a/crates/sandbox/src/lib.rs b/crates/sandbox/src/lib.rs index 64a1ec7..9a21382 100644 --- a/crates/sandbox/src/lib.rs +++ b/crates/sandbox/src/lib.rs @@ -41,7 +41,10 @@ use std::path::{Path, PathBuf}; #[cfg(target_os = "linux")] pub mod linux; pub mod measure; +#[cfg(target_os = "linux")] +mod scratch; pub mod seatbelt; +pub mod stop; pub use measure::Resources; diff --git a/crates/sandbox/src/linux.rs b/crates/sandbox/src/linux.rs index 1724717..0ca89e3 100644 --- a/crates/sandbox/src/linux.rs +++ b/crates/sandbox/src/linux.rs @@ -447,20 +447,23 @@ pub fn run_captured(plan: &Plan) -> Result<(i32, Vec, Resources), Error> { } fn measured(plan: &Plan, capture: bool) -> Result<(i32, Vec, Resources), Error> { - let (mut cmd, root) = command(plan)?; + let scratch = crate::scratch::Scratch::make(&std::env::temp_dir())?; + let (mut cmd, _) = command(plan)?; let ran = crate::measure::run(&mut cmd, capture) .map_err(|e| confine(format!("spawn {}: {e}", plan.argv[0]))); - std::fs::remove_dir_all(&root).ok(); + drop(scratch); ran } /// The command that builds the sandbox in `pre_exec` and execs the relay, and the -/// scratch directory it pivots into, for the caller to remove. +/// scratch directory it pivots into, which the caller holds as a [`crate::scratch::Scratch`]. fn command(plan: &Plan) -> Result<(Command, PathBuf), Error> { let uid = getuid().as_raw(); let gid = getgid().as_raw(); - let root = std::env::temp_dir().join(format!("memory-sandbox-{}", std::process::id())); - std::fs::create_dir_all(&root).map_err(|e| confine(format!("{}: {e}", root.display())))?; + // made and locked by the caller's Scratch; not made again here, where it would + // not be locked + let root = crate::scratch::dir(&std::env::temp_dir()); + let launcher = nix::unistd::getpid(); let plan2 = plan.clone(); let root2 = root.clone(); @@ -482,10 +485,17 @@ fn command(plan: &Plan) -> Result<(Command, PathBuf), Error> { enter_user_namespace(uid, gid)?; sethostname(&plan2.hostname).map_err(|e| confine(format!("hostname: {e}")))?; loopback_up()?; + // the run ends with the launcher, however the launcher ends + die_with_parent(launcher)?; + // the intermediate, as pid 1 can still tell it after the fork: its + // parent is outside its namespace, and getppid reads 0 there + let intermediate = pidfd_of_self(); // before build_root: it mounts this namespace's procfs, which must be // made after the fork that enters the namespace and before pivot_root // takes the host's procfs out of view become_pid_one()?; + // pid 1: the namespace, and the harness in it, ends with the intermediate + die_with_intermediate(intermediate)?; build_root(&plan2, &root2)?; Ok(()) }; @@ -575,7 +585,8 @@ fn cloexec_from_three(limit: libc::c_int) -> std::io::Result<()> { fn become_pid_one() -> Result<(), Error> { unshare(CloneFlags::CLONE_NEWPID).map_err(|e| confine(format!("unshare pid: {e}")))?; // SAFETY: fork between fork and exec in an already single-threaded child; the parent - // branch only waits and exits, the child branch only mounts and returns to exec. + // branch only closes descriptors, waits and exits, the child branch only mounts and + // returns to exec. let pid = unsafe { libc::fork() }; if pid < 0 { return Err(confine(format!( @@ -584,6 +595,11 @@ fn become_pid_one() -> Result<(), Error> { ))); } if pid > 0 { + // The intermediate never execs, so its copies of the launcher's descriptors + // stay open until it exits: among them std's exec-error channel, whose end of + // file is what lets `spawn` return. The child, pid 1, keeps its own copies and + // reports through that channel as before. + close_from_three(); let mut status: libc::c_int = 0; // SAFETY: waitpid on our own child with a valid status pointer. let r = unsafe { libc::waitpid(pid, &mut status, 0) }; @@ -650,6 +666,71 @@ fn say_the_relay_died(signal: i32, core_dumped: bool) { unsafe { libc::write(2, buf.as_ptr().cast(), n) }; } +/// Close every descriptor from 3 up. Runs between fork and exec, so it allocates nothing: +/// `close_range` (Linux 5.9) in one call, or else `close` on each number below the +/// open-files limit. +fn close_from_three() { + // SAFETY: close_range with integer arguments only. + let r = unsafe { libc::syscall(libc::SYS_close_range, 3, libc::c_uint::MAX, 0) }; + if r == 0 { + return; + } + for fd in 3..open_files_limit() { + // SAFETY: close of a descriptor number; one that is not open fails with EBADF. + unsafe { libc::close(fd) }; + } +} + +/// Have the kernel SIGKILL this process when the thread that forked it ends, and fail if +/// `parent` is no longer its parent: gone before the request took hold. Runs between fork +/// and exec; the setting outlives the exec. Allocates only to say why it failed. +fn die_with_parent(parent: nix::unistd::Pid) -> Result<(), Error> { + nix::sys::prctl::set_pdeathsig(nix::sys::signal::Signal::SIGKILL) + .map_err(|e| confine(format!("parent-death signal: {e}")))?; + if nix::unistd::getppid() == parent { + Ok(()) + } else { + Err(confine("the launcher has gone")) + } +} + +/// A pidfd for this process, close-on-exec, or -1 where the kernel has none (before 5.3). +fn pidfd_of_self() -> libc::c_int { + // SAFETY: pidfd_open with this process's pid and no flags; it returns a new + // descriptor or -1. + let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, libc::getpid(), 0) }; + libc::c_int::try_from(fd).unwrap_or(-1) +} + +/// As [`die_with_parent`] for pid 1, whose parent is the intermediate behind `pidfd`: +/// fail if the intermediate has already exited, which makes the pidfd readable. Without +/// a pidfd the check is not made, and an intermediate that dies between its fork and +/// the request leaves pid 1 to run until the harness ends. +fn die_with_intermediate(pidfd: libc::c_int) -> Result<(), Error> { + nix::sys::prctl::set_pdeathsig(nix::sys::signal::Signal::SIGKILL) + .map_err(|e| confine(format!("parent-death signal: {e}")))?; + if pidfd < 0 { + return Ok(()); + } + let mut poll = libc::pollfd { + fd: pidfd, + events: libc::POLLIN, + revents: 0, + }; + // SAFETY: poll on one valid pollfd, without waiting; then close of the pidfd this + // process owns. + let gone = unsafe { + let ready = libc::poll(&raw mut poll, 1, 0); + libc::close(pidfd); + ready > 0 + }; + if gone { + Err(confine("the intermediate has gone")) + } else { + Ok(()) + } +} + /// What `memory relay` does inside: spawn the harness and wait for it. The relay is /// already pid 1 of the harness's namespace, courtesy of [`run`]. /// diff --git a/crates/sandbox/src/measure.rs b/crates/sandbox/src/measure.rs index e982a51..e026814 100644 --- a/crates/sandbox/src/measure.rs +++ b/crates/sandbox/src/measure.rs @@ -39,34 +39,154 @@ pub struct Resources { /// /// When the process cannot be spawned, its output read, or it cannot be waited for. pub fn run(cmd: &mut Command, capture: bool) -> std::io::Result<(i32, Vec, Resources)> { + run_as(cmd, capture, false) +} + +/// As [`run`], with the process in a process group of its own, which +/// [`crate::stop::stop`] kills whole with what the run left in the background, when this +/// process has no controlling terminal. With one, the process stays in the terminal's +/// foreground group: in a group of its own it would be stopped by the first touch of +/// the terminal, a password prompt or `stty`. +/// +/// # Errors +/// +/// As [`run`]. +pub fn run_grouped(cmd: &mut Command, capture: bool) -> std::io::Result<(i32, Vec, Resources)> { + #[cfg(unix)] + if !crate::stop::has_a_terminal() { + use std::os::unix::process::CommandExt; + cmd.process_group(0); + return run_as(cmd, capture, true); + } + run_as(cmd, capture, false) +} + +/// How long a stopped run's output is read for once the run itself has gone: what it +/// left in the background, and did not die with it, may hold the output open. +const GRACE: Duration = Duration::from_secs(2); + +fn run_as( + cmd: &mut Command, + capture: bool, + group: bool, +) -> std::io::Result<(i32, Vec, Resources)> { let started = Instant::now(); if capture { cmd.stdin(Stdio::null()).stdout(Stdio::piped()); } let mut child = cmd.spawn()?; + let stoppable = crate::stop::register(&child, group); + // read on a thread of its own, so the run is waited for while its output is read + let read = child.stdout.take().map(|mut s| { + let out = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let (done, ended) = std::sync::mpsc::channel(); + let into = out.clone(); + std::thread::spawn(move || { + let mut buf = [0u8; 8192]; + let r = loop { + match s.read(&mut buf) { + Ok(0) => break Ok(()), + Ok(n) => lock(&into).extend_from_slice(&buf[..n]), + Err(e) if e.kind() == std::io::ErrorKind::Interrupted => {} + Err(e) => break Err(e), + } + }; + done.send(r).ok(); + }); + (out, ended) + }); + exited(&child); + // read to the end while the exited child is not yet reaped: a stop meanwhile still + // kills its process group, what it left in the background, and its number is + // given to no other process or group while it is unreaped let mut out = Vec::new(); - if let Some(mut s) = child.stdout.take() { - s.read_to_end(&mut out)?; + let mut read_failed = Ok(()); + if let Some((buf, ended)) = read { + read_failed = read_until_ended(&ended); + out = std::mem::take(&mut *lock(&buf)); } - let (code, mut r) = wait(child, cmd.get_program())?; + // whether a stop killed it while it ran: then a SIGKILL it ended with was the stop's + let stopped = stoppable.killed(); + // no longer stopped once it is reaped, after which its numbers may be given again + drop(stoppable); + let (code, mut r) = wait(child, cmd.get_program(), stopped)?; + read_failed?; r.wall_ms = millis(started.elapsed()); Ok((code, out, r)) } +fn lock(m: &std::sync::Mutex>) -> std::sync::MutexGuard<'_, Vec> { + m.lock().unwrap_or_else(std::sync::PoisonError::into_inner) +} + +/// Wait for the reader to reach the end of the output, or, once the launcher is told +/// to stop, for [`GRACE`] at most: what the run left in the background may hold it open. +fn read_until_ended(ended: &std::sync::mpsc::Receiver>) -> std::io::Result<()> { + let mut since: Option = None; + loop { + match ended.recv_timeout(Duration::from_millis(50)) { + Ok(r) => return r, + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => return Ok(()), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + if crate::stop::stopping() { + let at = *since.get_or_insert_with(Instant::now); + if at.elapsed() > GRACE { + return Ok(()); + } + } + } + } + } +} + +/// Wait for `child` to exit, without reaping it. +#[cfg(unix)] +fn exited(child: &Child) { + let pid: libc::id_t = child.id(); + loop { + // SAFETY: siginfo_t is plain data, for which all zeroes is a valid value. + let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() }; + // SAFETY: waitid on our own child, not yet reaped, leaving it unreaped (WNOWAIT), + // with a valid pointer to write into. + let r = unsafe { + libc::waitid( + libc::P_PID, + pid, + &raw mut info, + libc::WEXITED | libc::WNOWAIT, + ) + }; + if r == 0 || std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted { + return; + } + } +} + +#[cfg(not(unix))] +fn exited(_child: &Child) {} + /// The code a process ended with, as a shell gives it: its own exit code, or 128 plus /// the number of the signal that killed it. A death by signal is said on standard error, /// naming `program`, since the code alone does not tell it from an exit; SIGINT and -/// SIGPIPE are not, as a shell does not say them. +/// SIGPIPE are not, as a shell does not say them, nor a SIGKILL once +/// [`crate::stop::stop`] has sent it. #[must_use] pub fn exit_code(status: ExitStatus, program: &OsStr) -> i32 { + exit_code_as(status, program, crate::stop::stopping()) +} + +/// As [`exit_code`], with `stopped` whether a stop killed the process while it ran: a +/// SIGKILL from before the stop, an out-of-memory kill say, is said. +fn exit_code_as(status: ExitStatus, program: &OsStr, stopped: bool) -> i32 { #[cfg(not(unix))] - let _ = program; + let _ = (program, stopped); #[cfg(unix)] { use std::io::Write; use std::os::unix::process::ExitStatusExt; if let Some(signal) = status.signal() { - if is_said(signal) { + let stopped = signal == libc::SIGKILL && stopped; + if is_said(signal) && !stopped { let core = if status.core_dumped() { ", core dumped" } else { @@ -142,7 +262,7 @@ pub fn own_cpu_ms() -> (u64, u64) { } #[cfg(unix)] -fn wait(child: Child, program: &OsStr) -> std::io::Result<(i32, Resources)> { +fn wait(child: Child, program: &OsStr, stopped: bool) -> std::io::Result<(i32, Resources)> { use std::os::unix::process::ExitStatusExt; let pid = libc::pid_t::try_from(child.id()).map_err(std::io::Error::other)?; let mut status: libc::c_int = 0; @@ -162,7 +282,7 @@ fn wait(child: Child, program: &OsStr) -> std::io::Result<(i32, Resources)> { // reaped above; dropping the handle does not wait again drop(child); Ok(( - exit_code(ExitStatus::from_raw(status), program), + exit_code_as(ExitStatus::from_raw(status), program, stopped), Resources { wall_ms: 0, user_ms: timeval_ms(usage.ru_utime), @@ -174,9 +294,9 @@ fn wait(child: Child, program: &OsStr) -> std::io::Result<(i32, Resources)> { } #[cfg(not(unix))] -fn wait(mut child: Child, program: &OsStr) -> std::io::Result<(i32, Resources)> { +fn wait(mut child: Child, program: &OsStr, stopped: bool) -> std::io::Result<(i32, Resources)> { let status = child.wait()?; - Ok((exit_code(status, program), Resources::default())) + Ok((exit_code_as(status, program, stopped), Resources::default())) } fn millis(d: Duration) -> u64 { diff --git a/crates/sandbox/src/scratch.rs b/crates/sandbox/src/scratch.rs new file mode 100644 index 0000000..35e1709 --- /dev/null +++ b/crates/sandbox/src/scratch.rs @@ -0,0 +1,267 @@ +//! The directory a Linux sandbox builds its root in: `memory-sandbox-` in the +//! temporary directory, named for the launcher and locked by it for as long as the run +//! lasts. The launcher removes its own when the run ends; one a killed launcher left is +//! removed by the next launcher to start a run, once nothing holds it. + +use crate::Error; +use nix::errno::Errno; +use nix::fcntl::{Flock, FlockArg}; +use nix::sys::signal::kill; +use nix::unistd::{geteuid, Pid}; +use std::fs::File; +use std::os::unix::fs::{DirBuilderExt, MetadataExt, OpenOptionsExt}; +use std::path::{Path, PathBuf}; + +/// What every scratch directory's name starts with; the launcher's pid follows. +const PREFIX: &str = "memory-sandbox-"; + +/// This launcher's scratch directory, locked while it is held and removed when it is +/// dropped. +pub(crate) struct Scratch { + path: PathBuf, + _lock: Flock, +} + +impl Scratch { + /// Sweep what killed launchers left in `tmp`, then make and lock this launcher's own. + /// + /// # Errors + /// + /// When the directory cannot be made or opened, or another process holds its lock. + pub(crate) fn make(tmp: &Path) -> Result { + sweep(tmp); + let path = dir(tmp); + let failed = |e: &dyn std::fmt::Display| Error::Confine(format!("{}: {e}", path.display())); + std::fs::create_dir_all(tmp).map_err(|e| failed(&e))?; + // made with a mode of its own, whatever the umask: this user's alone + match std::fs::DirBuilder::new().mode(0o700).create(&path) { + Err(e) if e.kind() != std::io::ErrorKind::AlreadyExists => return Err(failed(&e)), + _ => {} + } + let dir = open_dir(&path).map_err(|e| failed(&e))?; + // one that was there already, another user's or one anyone may write into, is + // not built in + let meta = dir.metadata().map_err(|e| failed(&e))?; + if meta.uid() != geteuid().as_raw() || meta.mode() & 0o002 != 0 { + return Err(failed(&"not this user's own directory")); + } + let lock = Flock::lock(dir, FlockArg::LockExclusiveNonblock) + .map_err(|(_, e)| failed(&format!("held by another process: {e}")))?; + if !still_at(&path, &lock) { + return Err(failed(&"removed or replaced while it was being locked")); + } + Ok(Scratch { path, _lock: lock }) + } + + /// Where the root is built. + #[cfg(test)] + fn path(&self) -> &Path { + &self.path + } +} + +impl Drop for Scratch { + fn drop(&mut self) { + std::fs::remove_dir_all(&self.path).ok(); + } +} + +/// This launcher's scratch directory in `tmp`. +pub(crate) fn dir(tmp: &Path) -> PathBuf { + tmp.join(format!("{PREFIX}{}", std::process::id())) +} + +/// The directory at `path` itself, never what a link there points to. +fn open_dir(path: &Path) -> std::io::Result { + std::fs::OpenOptions::new() + .read(true) + .custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW) + .open(path) +} + +/// Whether the directory at `path` is still the one `dir` has open: a sweep may remove +/// it, and another make it again, between the open and the lock. +fn still_at(path: &Path, dir: &File) -> bool { + match (std::fs::symlink_metadata(path), dir.metadata()) { + (Ok(at), Ok(held)) => at.dev() == held.dev() && at.ino() == held.ino(), + _ => false, + } +} + +/// The pid a scratch directory's name carries, when the name is one this module makes. +fn pid_of(name: &str) -> Option { + let digits = name.strip_prefix(PREFIX)?; + if digits.is_empty() || !digits.bytes().all(|b| b.is_ascii_digit()) { + return None; + } + digits.parse().ok().filter(|p| *p > 0) +} + +/// Whether a process with this pid is running. One that is, whoever's, or one that +/// cannot be told, counts as running: a live launcher's directory is never removed. +fn running(pid: i32) -> bool { + !matches!(kill(Pid::from_raw(pid), None), Err(Errno::ESRCH)) +} + +/// Remove the scratch directories killed launchers left in `tmp`. One goes only when its +/// name is one this module makes, it is a directory of this user's and not a link, the +/// launcher its name carries is not running, nothing holds its lock, and it is empty, as +/// the host sees a sandbox's root once its mount namespace is gone. Anything else in +/// `tmp` is left alone. Returns the directories removed. +pub(crate) fn sweep(tmp: &Path) -> Vec { + let Ok(entries) = std::fs::read_dir(tmp) else { + return Vec::new(); + }; + let own = i32::try_from(std::process::id()).unwrap_or(0); + let me = geteuid().as_raw(); + entries + .flatten() + .filter_map(|e| { + let pid = pid_of(e.file_name().to_str()?)?; + if pid == own || running(pid) { + return None; + } + let path = e.path(); + let dir = open_dir(&path).ok()?; + if dir.metadata().ok()?.uid() != me { + return None; + } + let held = Flock::lock(dir, FlockArg::LockExclusiveNonblock).ok()?; + if !still_at(&path, &held) { + return None; + } + std::fs::remove_dir(&path).ok()?; + Some(path) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + + /// A process that runs until it is killed. By absolute path: other tests change PATH. + fn waiting() -> std::process::Command { + let mut cmd = std::process::Command::new("/bin/sh"); + cmd.args(["-c", "read x"]) + .stdin(std::process::Stdio::piped()); + cmd + } + + /// A pid no process has: one that has been reaped. + fn dead_pid() -> i32 { + let mut child = std::process::Command::new("/bin/sh") + .args(["-c", ":"]) + .spawn() + .unwrap(); + let pid = i32::try_from(child.id()).unwrap(); + child.wait().unwrap(); + assert!(!running(pid)); + pid + } + + #[test] + fn a_name_is_a_scratch_directory_s_only_when_this_module_makes_it() { + assert_eq!(pid_of("memory-sandbox-4242"), Some(4242)); + for name in [ + "memory-sandbox-", + "memory-sandbox-0", + "memory-sandbox-+42", + "memory-sandbox-42x", + "memory-sandbox-42/..", + "memory-sandbox-99999999999", + "other-4242", + "xmemory-sandbox-4242", + ] { + assert_eq!(pid_of(name), None, "{name}"); + } + } + + #[test] + fn a_killed_launcher_s_directory_is_swept_and_nothing_a_live_one_holds() { + let tmp = tempfile::tempdir().unwrap(); + let outside = tempfile::tempdir().unwrap(); + let dead = dead_pid(); + let stale = tmp.path().join(format!("{PREFIX}{dead}")); + std::fs::create_dir(&stale).unwrap(); + // a live launcher: its pid runs + let mut live_child = waiting().spawn().unwrap(); + let live = tmp.path().join(format!("{PREFIX}{}", live_child.id())); + std::fs::create_dir(&live).unwrap(); + // a dead pid's name, but its lock is held: a launcher in another pid namespace + let held_pid = dead_pid(); + let held = tmp.path().join(format!("{PREFIX}{held_pid}")); + std::fs::create_dir(&held).unwrap(); + let _lock = Flock::lock(open_dir(&held).unwrap(), FlockArg::LockExclusiveNonblock) + .map_err(|(_, e)| e) + .unwrap(); + // a dead pid's name on a link to a directory elsewhere, and on a directory with + // something in it + let linked_pid = dead_pid(); + let linked = tmp.path().join(format!("{PREFIX}{linked_pid}")); + std::os::unix::fs::symlink(outside.path(), &linked).unwrap(); + let full_pid = dead_pid(); + let full = tmp.path().join(format!("{PREFIX}{full_pid}")); + std::fs::create_dir(&full).unwrap(); + std::fs::write(full.join("kept"), "x").unwrap(); + // a name of another shape + let other = tmp.path().join(format!("memory-sandbox-{dead}-x")); + std::fs::create_dir(&other).unwrap(); + + let swept = sweep(tmp.path()); + + live_child.kill().unwrap(); + live_child.wait().unwrap(); + assert_eq!(swept, vec![stale.clone()]); + assert!(!stale.exists()); + for kept in [&live, &held, &full, &other] { + assert!(kept.is_dir(), "{}", kept.display()); + } + assert!(linked.is_symlink()); + assert!(outside.path().is_dir()); + assert!(full.join("kept").is_file()); + } + + #[test] + fn a_directory_anyone_may_write_into_is_not_built_in() { + use std::os::unix::fs::PermissionsExt; + let tmp = tempfile::tempdir().unwrap(); + let path = dir(tmp.path()); + std::fs::create_dir(&path).unwrap(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o777)).unwrap(); + assert!(Scratch::make(tmp.path()).is_err()); + assert!(path.is_dir(), "left as it was"); + } + + #[test] + fn a_directory_made_again_after_it_was_opened_is_not_the_one_held() { + let tmp = tempfile::tempdir().unwrap(); + let path = tmp.path().join(format!("{PREFIX}1")); + std::fs::create_dir(&path).unwrap(); + let held = open_dir(&path).unwrap(); + assert!(still_at(&path, &held)); + // made while the first still exists, so it cannot be given the same inode + let other = tmp.path().join("other"); + std::fs::create_dir(&other).unwrap(); + std::fs::remove_dir(&path).unwrap(); + std::fs::rename(&other, &path).unwrap(); + assert!( + !still_at(&path, &held), + "a sweep and another launcher came between" + ); + } + + #[test] + fn a_launcher_s_own_directory_is_locked_while_held_and_gone_when_dropped() { + let tmp = tempfile::tempdir().unwrap(); + let scratch = Scratch::make(tmp.path()).unwrap(); + let path = scratch.path().to_path_buf(); + assert_eq!( + path.file_name().unwrap().to_str().unwrap(), + format!("{PREFIX}{}", std::process::id()) + ); + assert!(Flock::lock(open_dir(&path).unwrap(), FlockArg::LockExclusiveNonblock).is_err()); + drop(scratch); + assert!(!path.exists()); + } +} diff --git a/crates/sandbox/src/stop.rs b/crates/sandbox/src/stop.rs new file mode 100644 index 0000000..7c16518 --- /dev/null +++ b/crates/sandbox/src/stop.rs @@ -0,0 +1,586 @@ +//! Stopping the runs this process has started, for a launcher told to stop by a signal. +//! [`stop`] kills every run [`crate::measure::run`] has spawned and not yet waited for, +//! and every run spawned after it, so each returns as a run that ended does: the +//! launcher then undoes what the run set up and exits, rather than dying with the +//! sandbox's scratch directory, its mount and its socket left behind. +//! +//! A run is killed with SIGKILL. On Linux the process spawned is the sandbox's +//! intermediate, which never execs and so still has the launcher's own signal handlers; +//! a signal it could catch would not end it. Its death ends the sandbox's pid namespace, +//! and the harness in it, through the parent-death signal its pid 1 holds. + +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Mutex, MutexGuard, PoisonError}; + +/// The runs of this process. +static RUNS: Runs = Runs::new(); + +/// Kill every run this process has spawned and not yet waited for, and every run it +/// spawns from now on. +pub fn stop() { + RUNS.stop(); +} + +/// Whether [`stop`] has been called: a launcher starts no further run once it has. +#[must_use] +pub fn stopping() -> bool { + RUNS.stopping() +} + +/// Whether a process with this pid runs. One that does, whoever's, or one that cannot be +/// told, counts as running, so that what a live process owns is never swept. +#[must_use] +pub fn process_runs(pid: u32) -> bool { + #[cfg(unix)] + { + let Ok(pid) = libc::pid_t::try_from(pid) else { + return true; + }; + if pid <= 0 { + return true; + } + // SAFETY: kill with signal 0 sends nothing; it only checks that the pid exists. + let r = unsafe { libc::kill(pid, 0) }; + r == 0 || std::io::Error::last_os_error().raw_os_error() != Some(libc::ESRCH) + } + #[cfg(not(unix))] + { + let _ = pid; + true + } +} + +/// Point standard output and standard error at the null device, for a launcher whose +/// terminal has hung up: a write to it fails, and `print!` and `eprintln!` panic on a +/// failed write, which would end the launcher before it cleans up. +/// +/// # Errors +/// +/// When the null device cannot be opened or put in place. +pub fn quiet_standard_streams() -> std::io::Result<()> { + #[cfg(unix)] + { + use std::os::fd::AsRawFd; + let null = std::fs::OpenOptions::new().write(true).open("/dev/null")?; + for fd in [1, 2] { + // SAFETY: dup2 of a descriptor this function holds open onto a standard one. + if unsafe { libc::dup2(null.as_raw_fd(), fd) } < 0 { + return Err(std::io::Error::last_os_error()); + } + } + } + Ok(()) +} + +/// Count `child` among the runs [`stop`] kills, until the returned value is dropped; +/// killed at once when [`stop`] has already been called. Called before the child is +/// waited for, while its pid is still its own. +/// +/// With `group`, the child leads a process group of its own, and the whole group is +/// killed: what the run left running in the background holds its output open. +pub(crate) fn register(child: &std::process::Child, group: bool) -> Registered<'static> { + RUNS.register(child, group) +} + +/// Whether this process has a controlling terminal, which a process group of a run's +/// own would be stopped for touching. +#[must_use] +pub fn has_a_terminal() -> bool { + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + // without waiting: a serial line without CLOCAL would hold the open + std::fs::OpenOptions::new() + .read(true) + .custom_flags(libc::O_NONBLOCK | libc::O_NOCTTY) + .open("/dev/tty") + .is_ok() + } + #[cfg(not(unix))] + { + false + } +} + +/// Whether nothing listens on the unix socket at `path`: a connection to it is refused. +/// The connection does not wait. On Linux a listener whose backlog is full, one that is +/// stopped and takes no more, counts as one that listens. Elsewhere, macOS say, a full +/// backlog refuses as well; the sweep asks only after the pid the socket is named for +/// has been found not running, and without pid namespaces there that pid is the +/// listener's, so a full backlog there is not a live launcher's. +#[must_use] +pub fn socket_refuses(path: &std::path::Path) -> bool { + #[cfg(unix)] + { + use std::os::unix::ffi::OsStrExt; + let bytes = path.as_os_str().as_bytes(); + // SAFETY: sockaddr_un is plain data, for which all zeroes is a valid value. + let mut addr: libc::sockaddr_un = unsafe { std::mem::zeroed() }; + if bytes.len() >= addr.sun_path.len() { + return false; + } + addr.sun_family = libc::AF_UNIX as libc::sa_family_t; + for (to, from) in addr.sun_path.iter_mut().zip(bytes) { + *to = *from as libc::c_char; + } + let fd = nonblocking_unix_stream(); + if fd < 0 { + return false; + } + let len = libc::socklen_t::try_from(std::mem::size_of::()).unwrap_or(0); + // SAFETY: connect of the socket made above to an address of the size given, then + // close of that socket. + unsafe { + let r = libc::connect(fd, (&raw const addr).cast(), len); + let refused = + r < 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ECONNREFUSED); + libc::close(fd); + refused + } + } + #[cfg(not(unix))] + { + let _ = path; + false + } +} + +/// A non-blocking, close-on-exec unix stream socket, or -1. +#[cfg(any(target_os = "linux", target_os = "android", target_os = "freebsd"))] +fn nonblocking_unix_stream() -> libc::c_int { + // SAFETY: a socket call with constant arguments; it returns a descriptor or -1. + unsafe { + libc::socket( + libc::AF_UNIX, + libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + 0, + ) + } +} + +/// A non-blocking, close-on-exec unix stream socket, or -1: by fcntl after the socket +/// call, where the socket call cannot set either. +#[cfg(all( + unix, + not(any(target_os = "linux", target_os = "android", target_os = "freebsd")) +))] +fn nonblocking_unix_stream() -> libc::c_int { + // SAFETY: a socket call with constant arguments, then fcntl on the descriptor it + // returned, and its close when fcntl fails. + unsafe { + let fd = libc::socket(libc::AF_UNIX, libc::SOCK_STREAM, 0); + if fd < 0 { + return -1; + } + let fl = libc::fcntl(fd, libc::F_GETFL); + let set = fl >= 0 + && libc::fcntl(fd, libc::F_SETFL, fl | libc::O_NONBLOCK) == 0 + && libc::fcntl(fd, libc::F_SETFD, libc::FD_CLOEXEC) == 0; + if set { + fd + } else { + libc::close(fd); + -1 + } + } +} + +/// Whether this process ignores `signal`, as one started under `nohup`, or in the +/// background by a shell without job control, ignores SIGHUP or SIGINT. A launcher keeps +/// that: it does not stop on a signal it was started to ignore. +#[must_use] +pub fn ignored(signal: i32) -> bool { + #[cfg(unix)] + { + // SAFETY: sigaction is plain data, for which all zeroes is a valid value. + let mut old: libc::sigaction = unsafe { std::mem::zeroed() }; + // SAFETY: sigaction with a null new action only reads the current one into `old`. + let r = unsafe { libc::sigaction(signal, std::ptr::null(), &raw mut old) }; + r == 0 && old.sa_sigaction == libc::SIG_IGN + } + #[cfg(not(unix))] + { + let _ = signal; + false + } +} + +/// Runs spawned and not yet waited for, and whether they are being stopped. +struct Runs { + stopping: AtomicBool, + running: Mutex>, +} + +impl Runs { + const fn new() -> Runs { + Runs { + stopping: AtomicBool::new(false), + running: Mutex::new(Vec::new()), + } + } + + fn running(&self) -> MutexGuard<'_, Vec> { + self.running.lock().unwrap_or_else(PoisonError::into_inner) + } + + fn stop(&self) { + self.stopping.store(true, Ordering::SeqCst); + for h in self.running().iter() { + h.kill(); + } + } + + fn stopping(&self) -> bool { + self.stopping.load(Ordering::SeqCst) + } + + fn register(&self, child: &std::process::Child, group: bool) -> Registered<'_> { + let h = Handle::of(child.id(), group); + let mut list = self.running(); + // read under the lock that stop takes after setting it: a run registered while + // stop goes through its list is either in the list or sees the flag + if self.stopping() { + h.kill(); + } + let killed = h.killed.clone(); + list.push(h); + Registered { + runs: self, + pid: child.id(), + killed, + } + } +} + +/// A spawned run, until it has been waited for. +pub(crate) struct Registered<'a> { + runs: &'a Runs, + pid: u32, + killed: std::sync::Arc, +} + +impl Registered<'_> { + /// Whether a stop killed the run while it was still running: then a SIGKILL it ended + /// with was the stop's, and not a death to report. + pub(crate) fn killed(&self) -> bool { + self.killed.load(Ordering::SeqCst) + } +} + +impl Drop for Registered<'_> { + fn drop(&mut self) { + self.runs.running().retain(|h| h.pid != self.pid); + } +} + +/// What a run is killed through: on Linux a pidfd, which names the process itself and +/// not a number a later process may be given once it is reaped. +struct Handle { + pid: u32, + /// Set when a stop kills the run before it has exited. + killed: std::sync::Arc, + /// Whether the pid leads a process group of the run's own, killed with it. + #[cfg(unix)] + group: bool, + #[cfg(target_os = "linux")] + pidfd: Option, +} + +impl Handle { + fn of(pid: u32, group: bool) -> Handle { + #[cfg(target_os = "linux")] + { + use std::os::fd::FromRawFd; + let raw = libc::pid_t::try_from(pid).unwrap_or(0); + // SAFETY: pidfd_open takes a pid and flags and returns a new descriptor or -1. + let fd = unsafe { libc::syscall(libc::SYS_pidfd_open, raw, 0) }; + let pidfd = libc::c_int::try_from(fd) + .ok() + .filter(|fd| *fd >= 0) + // SAFETY: a descriptor pidfd_open just returned, owned by nothing else. + .map(|fd| unsafe { std::os::fd::OwnedFd::from_raw_fd(fd) }); + Handle { + pid, + killed: std::sync::Arc::default(), + group, + pidfd, + } + } + #[cfg(not(target_os = "linux"))] + { + #[cfg(not(unix))] + let _ = group; + Handle { + pid, + killed: std::sync::Arc::default(), + #[cfg(unix)] + group, + } + } + } + + /// Whether the run has already exited, though not yet reaped. + fn exited(&self) -> bool { + #[cfg(unix)] + { + let pid: libc::id_t = self.pid; + // SAFETY: siginfo_t is plain data, for which all zeroes is a valid value. + let mut info: libc::siginfo_t = unsafe { std::mem::zeroed() }; + // SAFETY: waitid on this process's own child, without waiting and leaving it + // unreaped, with a valid pointer to write into. + let r = unsafe { + libc::waitid( + libc::P_PID, + pid, + &raw mut info, + libc::WEXITED | libc::WNOHANG | libc::WNOWAIT, + ) + }; + // a child that has exited fills the siginfo; one still running leaves it zero + r == 0 && info.si_signo != 0 + } + #[cfg(not(unix))] + { + false + } + } + + fn kill(&self) { + if !self.exited() { + self.killed.store(true, Ordering::SeqCst); + } + #[cfg(unix)] + if self.group { + if let Ok(pid) = libc::pid_t::try_from(self.pid) { + // SAFETY: kill of the process group this run's child leads. The handle is + // dropped before the leader is reaped, so while it is here no other + // process or group can have the leader's number. + unsafe { libc::kill(-pid, libc::SIGKILL) }; + } + } + #[cfg(target_os = "linux")] + if let Some(fd) = &self.pidfd { + use std::os::fd::AsRawFd; + // SAFETY: pidfd_send_signal on a pidfd this handle owns, with no siginfo. + unsafe { + libc::syscall( + libc::SYS_pidfd_send_signal, + fd.as_raw_fd(), + libc::SIGKILL, + std::ptr::null::(), + 0, + ) + }; + return; + } + #[cfg(unix)] + if let Ok(pid) = libc::pid_t::try_from(self.pid) { + // SAFETY: kill with a positive pid of a child not yet waited for. + unsafe { libc::kill(pid, libc::SIGKILL) }; + } + } +} + +#[cfg(test)] +#[cfg(unix)] +mod tests { + use super::*; + use std::process::{Child, Command, Stdio}; + + /// A process that runs until it is killed. By absolute path: other tests change PATH. + fn waiting() -> Child { + Command::new("/bin/sh") + .args(["-c", "read x"]) + .stdin(Stdio::piped()) + .spawn() + .unwrap() + } + + /// Whether `child` was killed within ten seconds; killed here if not, so a failing + /// test does not hang. + fn killed(mut child: Child) -> bool { + for _ in 0..100 { + if let Some(status) = child.try_wait().unwrap() { + return !status.success(); + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + child.kill().ok(); + child.wait().ok(); + false + } + + #[test] + fn a_stop_kills_the_runs_in_flight_and_every_run_after_it() { + let runs = Runs::new(); + let before = waiting(); + let kept = runs.register(&before, false); + let mut done = Command::new("/bin/sh").args(["-c", ":"]).spawn().unwrap(); + let gone = runs.register(&done, false); + done.wait().unwrap(); + drop(gone); + assert_eq!(runs.running().len(), 1); + assert!(!runs.stopping()); + + runs.stop(); + + assert!(runs.stopping()); + assert!(killed(before), "a run in flight is killed"); + drop(kept); + assert!(runs.running().is_empty()); + let after = waiting(); + let _r = runs.register(&after, false); + assert!(killed(after), "a run spawned after the stop is killed"); + } + + #[test] + fn a_stop_kills_a_grouped_run_and_what_it_left_in_the_background() { + use std::io::BufRead; + use std::os::unix::process::CommandExt; + let runs = Runs::new(); + let mut leader = Command::new("/bin/sh"); + leader + .args(["-c", "/bin/sleep 300 & echo $!; wait"]) + .stdout(Stdio::piped()) + .process_group(0); + let mut leader = leader.spawn().unwrap(); + let mut line = String::new(); + std::io::BufReader::new(leader.stdout.take().unwrap()) + .read_line(&mut line) + .unwrap(); + let background: u32 = line.trim().parse().unwrap(); + let _r = runs.register(&leader, true); + assert!(process_runs(background)); + + runs.stop(); + + assert!(killed(leader)); + let gone = (0..100).any(|_| { + std::thread::sleep(std::time::Duration::from_millis(50)); + !process_runs(background) + }); + if !gone { + Command::new("kill") + .args(["-KILL", &background.to_string()]) + .status() + .ok(); + } + assert!(gone, "the background child goes with its group"); + } + + #[test] + fn a_signal_this_process_ignores_is_said_to_be_ignored() { + assert!(!ignored(libc::SIGUSR2)); + // SAFETY: setting SIGUSR2, which nothing in this test binary uses, to ignored. + unsafe { libc::signal(libc::SIGUSR2, libc::SIG_IGN) }; + assert!(ignored(libc::SIGUSR2)); + // SAFETY: as above, back to the default. + unsafe { libc::signal(libc::SIGUSR2, libc::SIG_DFL) }; + assert!(!ignored(libc::SIGUSR2)); + } + + #[test] + fn a_socket_refuses_only_when_nothing_listens_and_a_full_one_does_not_hang() { + let dir = tempfile::tempdir().unwrap(); + // bound and never listening: refused, whatever a fork by another test copies + let gone = dir.path().join("gone.sock"); + bound_only(&gone); + assert!(socket_refuses(&gone)); + let live = dir.path().join("live.sock"); + let _live = std::os::unix::net::UnixListener::bind(&live).unwrap(); + assert!(!socket_refuses(&live)); + assert!( + !socket_refuses(&dir.path().join("none.sock")), + "not there: not refused" + ); + + full_backlog_is_a_listener(dir.path()); + } + + /// Bind a unix stream socket at `path` and close it, without listening. + fn bound_only(path: &std::path::Path) { + use std::os::unix::ffi::OsStrExt; + // SAFETY: a zeroed address filled with a path that fits, a socket, its bind and + // its close. + unsafe { + let mut addr: libc::sockaddr_un = std::mem::zeroed(); + addr.sun_family = libc::AF_UNIX as libc::sa_family_t; + for (to, from) in addr.sun_path.iter_mut().zip(path.as_os_str().as_bytes()) { + *to = *from as libc::c_char; + } + let fd = libc::socket(libc::AF_UNIX, libc::SOCK_STREAM, 0); + assert!(fd >= 0); + let len = std::mem::size_of::() as libc::socklen_t; + assert_eq!(libc::bind(fd, (&raw const addr).cast(), len), 0); + libc::close(fd); + } + } + + /// On Linux a full backlog is a listener, said without waiting. macOS refuses a + /// connection to a full backlog. + #[cfg(not(target_os = "linux"))] + fn full_backlog_is_a_listener(_dir: &std::path::Path) {} + + #[cfg(target_os = "linux")] + fn full_backlog_is_a_listener(dir: &std::path::Path) { + // a listener that accepts nothing, its backlog filled + let full = dir.join("full.sock"); + let listener = std::os::unix::net::UnixListener::bind(&full).unwrap(); + { + use std::os::fd::AsRawFd; + // SAFETY: listen on a socket this test holds, with the smallest backlog. + assert_eq!(unsafe { libc::listen(listener.as_raw_fd(), 0) }, 0); + } + let mut held = Vec::new(); + loop { + let s = socket2_like_connect(&full); + match s { + Some(fd) => held.push(fd), + None => break, + } + assert!(held.len() < 1000, "the backlog never filled"); + } + // on a thread of its own, so a probe that waits fails here rather than hangs + let (said, answer) = std::sync::mpsc::channel(); + let probed = full.clone(); + std::thread::spawn(move || said.send(socket_refuses(&probed)).ok()); + let refused = answer.recv_timeout(std::time::Duration::from_secs(2)); + assert_eq!( + refused, + Ok(false), + "a full backlog is a listener, said at once" + ); + for fd in held { + // SAFETY: close of a socket this test opened. + unsafe { libc::close(fd) }; + } + } + + /// A non-blocking connection to `path` that is kept open; `None` once one would wait. + #[cfg(target_os = "linux")] + fn socket2_like_connect(path: &std::path::Path) -> Option { + use std::os::unix::ffi::OsStrExt; + // SAFETY: as in socket_refuses: a zeroed address filled with a path that fits, a + // non-blocking socket, a connect of it. + unsafe { + let mut addr: libc::sockaddr_un = std::mem::zeroed(); + addr.sun_family = libc::AF_UNIX as libc::sa_family_t; + for (to, from) in addr.sun_path.iter_mut().zip(path.as_os_str().as_bytes()) { + *to = *from as libc::c_char; + } + let fd = libc::socket(libc::AF_UNIX, libc::SOCK_STREAM, 0); + libc::fcntl( + fd, + libc::F_SETFL, + libc::fcntl(fd, libc::F_GETFL) | libc::O_NONBLOCK, + ); + let len = std::mem::size_of::() as libc::socklen_t; + if libc::connect(fd, (&raw const addr).cast(), len) == 0 { + Some(fd) + } else { + libc::close(fd); + None + } + } + } +}