diff --git a/Cargo.lock b/Cargo.lock index 39adee8b384..844e138b05d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1474,7 +1474,9 @@ dependencies = [ "httparse", "nix 0.31.3", "nostr 0.44.7", + "pty-process", "reqwest 0.13.4", + "rustix", "rustls", "serde", "serde_json", @@ -8761,6 +8763,16 @@ dependencies = [ "thiserror 1.0.69", ] +[[package]] +name = "pty-process" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71cec9e2670207c5ebb9e477763c74436af3b9091dd550b9fb3c1bec7f3ea266" +dependencies = [ + "rustix", + "tokio", +] + [[package]] name = "pulldown-cmark" version = "0.13.6" diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index dbb5835968d..a14b4aa98d6 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -82,7 +82,11 @@ evalexpr = { workspace = true } # Process-group kill (safe wrapper around killpg) — Unix-only; kill_process_group # has a #[cfg(not(unix))] fallback in acp.rs. [target.'cfg(unix)'.dependencies] -nix = { version = "0.31", default-features = false, features = ["signal", "user"] } +nix = { version = "0.31", default-features = false, features = ["signal", "user", "process"] } + +[target.'cfg(target_os = "linux")'.dependencies] +pty-process = { version = "=0.5.3", features = ["async"] } +rustix = { version = "1.1.4", features = ["process"] } [dev-dependencies] axum = { workspace = true } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 702e14f78e9..b99f6675f9c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -23,6 +23,7 @@ mod recovery_wake; mod relay; mod run_task; mod runtime; +mod sandbox_terminal; /// Shared channel task admission through the existing encrypted control plane. pub mod sharing; use runtime::{AgentRuntime, PoolStartup, SessionMode}; @@ -2898,6 +2899,9 @@ pub fn run() -> Result<()> { Some("git-sign-nostr") => std::process::exit(git_sign_nostr::run()), _ => {} } + if is_subcommand("sandbox-terminal-worker") { + std::process::exit(sandbox_terminal::worker_entry()); + } config::propagate_legacy_env_vars(); if is_subcommand("run") { let runtime = tokio::runtime::Runtime::new()?; diff --git a/crates/buzz-acp/src/sandbox_terminal.rs b/crates/buzz-acp/src/sandbox_terminal.rs new file mode 100644 index 00000000000..e8dfc347627 --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal.rs @@ -0,0 +1,22 @@ +//! Local terminal process boundary. Relay authorization is applied by the +//! harness before this worker is started. The worker uses local IPC only. + +#[cfg(target_os = "linux")] +mod process_tree; +#[cfg(target_os = "linux")] +mod worker; + +/// Dispatch the isolated worker before ordinary harness configuration or Tokio +/// threads exist. Exit 2 means containment could not be confirmed: the caller +/// must terminate the harness instead of leaving an uncontained sandbox alive. +pub(crate) fn worker_entry() -> i32 { + #[cfg(target_os = "linux")] + { + worker::entry() + } + #[cfg(not(target_os = "linux"))] + { + eprintln!("sandbox terminals require Linux process containment"); + 2 + } +} diff --git a/crates/buzz-acp/src/sandbox_terminal/process_tree.rs b/crates/buzz-acp/src/sandbox_terminal/process_tree.rs new file mode 100644 index 00000000000..90bd8697272 --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal/process_tree.rs @@ -0,0 +1,217 @@ +//! Linux subreaper teardown scoped to a dedicated terminal worker. + +use std::collections::{HashMap, HashSet}; +use std::io::Read; +use std::time::{Duration, Instant}; + +use anyhow::{bail, ensure, Context, Result}; +use nix::sys::wait::{waitpid, WaitPidFlag, WaitStatus}; +use rustix::process::{pidfd_open, pidfd_send_signal, Pid, PidfdFlags, Signal}; + +const MAX_PROCESSES: usize = 4096; +const MAX_DESCENDANTS: usize = 256; +const MAX_STAT_BYTES: u64 = 4096; +const CLEANUP_TIMEOUT: Duration = Duration::from_secs(5); + +#[derive(Clone, Copy)] +struct Identity { + pid: i32, + parent: i32, + started: u64, + zombie: bool, +} + +/// Establish and verify containment before any shell can exist. This process +/// owns only one terminal, so adopted children can never be harness siblings. +pub(super) fn establish() -> Result<()> { + nix::sys::prctl::set_child_subreaper(true).context("terminal subreaper setup failed")?; + ensure!( + nix::sys::prctl::get_child_subreaper()?, + "terminal subreaper unavailable" + ); + nix::sys::prctl::set_pdeathsig(Some(nix::sys::signal::Signal::SIGTERM))?; + ensure!( + nix::unistd::getppid().as_raw() > 1, + "terminal parent disappeared" + ); + // Refuse old kernels or seccomp profiles before creating a shell. pidfds + // plus start-time verification keep PID reuse from targeting another task. + let own = pidfd_open(rustix::process::getpid(), PidfdFlags::empty())?; + pidfd_send_signal(&own, Signal::CONT)?; + ensure!( + descendants()?.is_empty(), + "terminal worker already owns children" + ); + Ok(()) +} + +fn identity(pid: i32) -> Result> { + let file = match std::fs::File::open(format!("/proc/{pid}/stat")) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None), + Err(error) => return Err(error.into()), + }; + let mut text = String::new(); + file.take(MAX_STAT_BYTES + 1).read_to_string(&mut text)?; + ensure!( + text.len() <= MAX_STAT_BYTES as usize, + "terminal process metadata exceeds bound" + ); + let (_, fields) = text + .rsplit_once(')') + .context("invalid terminal process metadata")?; + let fields: Vec<_> = fields.split_whitespace().collect(); + ensure!(fields.len() >= 20, "incomplete terminal process metadata"); + Ok(Some(Identity { + pid, + parent: fields[1].parse()?, + started: fields[19].parse()?, + zombie: matches!(fields[0], "Z" | "X"), + })) +} + +fn descendants() -> Result> { + let mut processes = HashMap::new(); + let mut scanned = 0; + for entry in std::fs::read_dir("/proc")? { + let entry = entry?; + let Some(pid) = entry + .file_name() + .to_str() + .and_then(|name| name.parse::().ok()) + else { + continue; + }; + if pid <= 0 { + continue; + } + scanned += 1; + ensure!( + scanned <= MAX_PROCESSES, + "terminal process scan exceeds bound" + ); + if let Some(process) = identity(pid)? { + processes.insert(pid, process); + } + } + select_descendants(processes, std::process::id() as i32) +} + +fn select_descendants(processes: HashMap, own: i32) -> Result> { + let mut selected = HashSet::from([own]); + loop { + let before = selected.len(); + for process in processes.values() { + if selected.contains(&process.parent) { + selected.insert(process.pid); + } + } + ensure!( + selected.len() <= MAX_DESCENDANTS + 1, + "terminal descendant limit exceeded" + ); + if before == selected.len() { + break; + } + } + Ok(processes + .into_values() + .filter(|process| process.pid != own && selected.contains(&process.pid)) + .collect()) +} + +fn signal(process: Identity, signal: Signal) -> Result<()> { + if process.zombie { + return Ok(()); + } + let pid = Pid::from_raw(process.pid).context("invalid terminal process id")?; + let fd = match pidfd_open(pid, PidfdFlags::empty()) { + Ok(fd) => fd, + Err(rustix::io::Errno::SRCH) => return Ok(()), + Err(error) => return Err(error.into()), + }; + if !identity(process.pid)?.is_some_and(|current| current.started == process.started) { + return Ok(()); + } + match pidfd_send_signal(&fd, signal) { + Ok(()) | Err(rustix::io::Errno::SRCH) => Ok(()), + Err(error) => Err(error.into()), + } +} + +/// Stop, kill and reap the complete terminal tree, including reparented +/// setsid/double-fork children. A containment failure is always an error. +pub(super) fn cleanup() -> Result<()> { + let deadline = Instant::now() + CLEANUP_TIMEOUT; + loop { + let children = descendants()?; + for child in &children { + signal(*child, Signal::STOP)?; + } + for child in children { + signal(child, Signal::KILL)?; + } + for _ in 0..=MAX_DESCENDANTS { + match waitpid(None, Some(WaitPidFlag::WNOHANG)) { + Ok(WaitStatus::StillAlive) | Err(nix::errno::Errno::ECHILD) => break, + Ok(_) | Err(nix::errno::Errno::EINTR) => {} + Err(error) => return Err(error.into()), + } + } + if descendants()?.is_empty() { + return Ok(()); + } + if Instant::now() >= deadline { + bail!("terminal process teardown timed out"); + } + std::thread::sleep(Duration::from_millis(20)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn terminal_process_selection_excludes_worker_and_siblings() { + let processes = [(10, 1), (20, 10), (30, 20), (40, 1), (50, 10)] + .into_iter() + .map(|(pid, parent)| { + ( + pid, + Identity { + pid, + parent, + started: 1, + zombie: false, + }, + ) + }) + .collect(); + let mut selected: Vec<_> = select_descendants(processes, 10) + .unwrap() + .into_iter() + .map(|process| process.pid) + .collect(); + selected.sort(); + assert_eq!(selected, [20, 30, 50]); + } + + #[test] + fn terminal_process_selection_refuses_excess_descendants() { + let processes = (20..20 + MAX_DESCENDANTS as i32 + 1) + .map(|pid| { + ( + pid, + Identity { + pid, + parent: 10, + started: 1, + zombie: false, + }, + ) + }) + .collect(); + assert!(select_descendants(processes, 10).is_err()); + } +} diff --git a/crates/buzz-acp/src/sandbox_terminal/worker.rs b/crates/buzz-acp/src/sandbox_terminal/worker.rs new file mode 100644 index 00000000000..4903acde20c --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal/worker.rs @@ -0,0 +1,216 @@ +//! A single PTY under a dedicated Linux subreaper. Relay authorization +//! belongs to its parent; this process handles bounded local PTY operations. + +use std::path::PathBuf; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anyhow::{bail, ensure, Context, Result}; +use buzz_sdk::sandbox_attach::{Packet, MAX_DATA_BYTES, MAX_GRANT_SECS, MAX_IN_FLIGHT_BYTES}; +use futures_util::StreamExt; +use serde::Serialize; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio_util::codec::{FramedRead, LinesCodec}; + +use super::process_tree; + +const MAX_LINE_BYTES: usize = 32 * 1024; +const IO_TIMEOUT: Duration = Duration::from_secs(5); +const IDLE_TIMEOUT: Duration = Duration::from_secs(300); + +#[derive(Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum Output<'a> { + Ready, + Packet { packet: &'a Packet }, + Ended { failed: bool }, +} + +/// This mode is intentionally internal and runs before harness configuration. +/// Exit 1 is an operation error after confirmed cleanup; exit 2 is a fatal +/// containment error, which the parent must never treat as a normal close. +pub(super) fn entry() -> i32 { + let arguments: Vec<_> = std::env::args_os().skip(2).collect(); + let result = (|| -> Result<_> { + ensure!( + arguments.len() == 2, + "terminal worker requires workspace and expiry" + ); + let workspace = PathBuf::from(&arguments[0]).canonicalize()?; + ensure!(workspace.is_dir(), "terminal workspace is not a directory"); + let expiry = arguments[1] + .to_str() + .context("invalid terminal expiry")? + .parse::()?; + let now = SystemTime::now().duration_since(UNIX_EPOCH)?.as_secs(); + ensure!( + expiry > now && expiry - now <= MAX_GRANT_SECS, + "invalid terminal lifetime" + ); + process_tree::establish()?; + Ok((workspace, Duration::from_secs(expiry - now))) + })(); + let Ok((workspace, lifetime)) = result else { + eprintln!("terminal worker setup refused"); + return 2; + }; + let runtime = match tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + { + Ok(runtime) => runtime, + Err(_) => { + eprintln!("terminal worker runtime unavailable"); + return 2; + } + }; + let result = runtime.block_on(run(workspace, lifetime)); + // Dropping the PTY wakes its readers, but only subreaper teardown proves + // that all jobs and detached descendants have actually stopped. + if process_tree::cleanup().is_err() { + runtime.shutdown_timeout(Duration::from_millis(100)); + eprintln!("terminal containment failed; harness must terminate"); + return 2; + } + let failed = result.is_err(); + let receipt = runtime.block_on(output(&Output::Ended { failed })); + // Tokio stdin may have an in-flight blocking read after a lease/signal + // cancels it. Process exit retires that worker; never await it forever. + runtime.shutdown_timeout(Duration::from_millis(100)); + if failed || receipt.is_err() { + eprintln!("terminal worker operation failed"); + 1 + } else { + 0 + } +} + +async fn output(message: &Output<'_>) -> Result<()> { + let mut bytes = serde_json::to_vec(message)?; + ensure!( + bytes.len() < MAX_LINE_BYTES, + "terminal output exceeds bound" + ); + bytes.push(b'\n'); + tokio::time::timeout(IO_TIMEOUT, async { + let mut stdout = tokio::io::stdout(); + stdout.write_all(&bytes).await?; + stdout.flush().await + }) + .await + .context("terminal output timed out")??; + Ok(()) +} + +/// Reap jobs that could retain the slave, then preserve the shell's buffered +/// final output. A byte or time limit is an error, never a successful truncation. +async fn drain_tail(pty: &mut pty_process::Pty) -> Result<()> { + process_tree::cleanup()?; + tokio::time::timeout(IO_TIMEOUT, async { + let mut remaining = MAX_IN_FLIGHT_BYTES; + let mut bytes = [0_u8; MAX_DATA_BYTES]; + loop { + match pty.read(&mut bytes).await { + Ok(0) => return Ok(()), + Ok(length) => { + ensure!( + length <= remaining, + "terminal final output exceeds drain bound" + ); + remaining -= length; + let packet = Packet::data(&bytes[..length])?; + output(&Output::Packet { packet: &packet }).await?; + } + Err(error) if error.raw_os_error() == Some(nix::libc::EIO) => return Ok(()), + Err(error) => return Err(anyhow::Error::from(error)), + } + } + }) + .await + .context("terminal final output drain timed out")? +} + +async fn run(workspace: PathBuf, lifetime: Duration) -> Result<()> { + use tokio::signal::unix::{signal, SignalKind}; + let (mut term, mut hangup, mut interrupt) = ( + signal(SignalKind::terminate())?, + signal(SignalKind::hangup())?, + signal(SignalKind::interrupt())?, + ); + let deadline = tokio::time::Instant::now() + lifetime; + let mut lines = FramedRead::new( + tokio::io::stdin(), + LinesCodec::new_with_max_length(MAX_LINE_BYTES), + ); + let first = tokio::time::timeout(IO_TIMEOUT.min(lifetime), lines.next()) + .await? + .context("terminal open was not supplied")??; + let Packet::Open { rows, cols } = serde_json::from_str::(&first)? else { + bail!("first terminal command must open the PTY"); + }; + Packet::Open { rows, cols }.bytes()?; + let (mut pty, pts) = pty_process::open()?; + pty.resize(pty_process::Size::new(rows, cols))?; + let environment = ["PATH", "HOME", "LANG", "LC_ALL", "TMPDIR"] + .into_iter() + .filter_map(|name| std::env::var_os(name).map(|value| (name, value))); + let mut child = pty_process::Command::new("/bin/sh") + .arg("-i") + .current_dir(workspace) + .env_clear() + .envs(environment) + .env("TERM", "xterm-256color") + .kill_on_drop(true) + .spawn(pts)?; + output(&Output::Ready).await?; + let mut idle = tokio::time::Instant::now(); + let mut tick = tokio::time::interval(Duration::from_millis(100)); + let mut buffer = [0_u8; MAX_DATA_BYTES]; + loop { + tokio::select! { + biased; + _ = term.recv() => break, + _ = hangup.recv() => break, + _ = interrupt.recv() => break, + _ = tick.tick() => { + if tokio::time::Instant::now() >= deadline || idle.elapsed() >= IDLE_TIMEOUT { break; } + if child.try_wait()?.is_some() { + drain_tail(&mut pty).await?; + break; + } + } + line = lines.next() => { + let Some(line) = line else { break; }; + let packet: Packet = serde_json::from_str(&line?)?; + let bytes = packet.bytes()?; + match packet { + Packet::Data { .. } => { + if !bytes.is_empty() { + idle = tokio::time::Instant::now(); + tokio::time::timeout(IO_TIMEOUT, pty.write_all(&bytes)).await??; + } + } + Packet::Resize { rows, cols } => { + pty.resize(pty_process::Size::new(rows, cols))?; + idle = tokio::time::Instant::now(); + } + Packet::Close => break, + _ => bail!("invalid terminal worker operation"), + } + } + read = pty.read(&mut buffer) => { + match read { + Ok(0) => break, + Ok(length) => { + let packet = Packet::data(&buffer[..length])?; + output(&Output::Packet { packet: &packet }).await?; + } + // Linux PTYs report EIO when the slave closes normally. + Err(error) if error.raw_os_error() == Some(nix::libc::EIO) => break, + Err(error) => return Err(error.into()), + } + } + } + } + drop(pty); + Ok(()) +} diff --git a/crates/buzz-acp/tests/sandbox_terminal_worker.rs b/crates/buzz-acp/tests/sandbox_terminal_worker.rs new file mode 100644 index 00000000000..e5772f360bc --- /dev/null +++ b/crates/buzz-acp/tests/sandbox_terminal_worker.rs @@ -0,0 +1,314 @@ +#![cfg(target_os = "linux")] + +//! Drive the production binary and a real controlling PTY. Descendant cleanup +//! is checked independently through /proc, including double-fork + setsid. + +use std::process::Stdio; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use buzz_sdk::sandbox_attach::Packet; +use futures_util::StreamExt; +use serde_json::{json, Value}; +use tokio::io::AsyncWriteExt; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; +use tokio_util::codec::{FramedRead, LinesCodec}; + +const BOUND: Duration = Duration::from_secs(8); + +struct Worker { + child: Child, + input: Option, + output: FramedRead, +} + +impl Worker { + fn start(workspace: &std::path::Path, seconds: u64) -> Self { + let expiry = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() + + seconds; + let mut child = Command::new(env!("CARGO_BIN_EXE_buzz-acp")) + .args([ + "sandbox-terminal-worker", + workspace.to_str().unwrap(), + &expiry.to_string(), + ]) + .env_clear() + .env("PATH", std::env::var_os("PATH").unwrap()) + .env("HOME", workspace) + .env("BUZZ_PRIVATE_KEY", "must-not-reach-shell") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + Self { + input: child.stdin.take(), + output: FramedRead::new( + child.stdout.take().unwrap(), + LinesCodec::new_with_max_length(32768), + ), + child, + } + } + + async fn send(&mut self, packet: Packet) { + let mut bytes = serde_json::to_vec(&packet).unwrap(); + bytes.push(b'\n'); + self.input + .as_mut() + .unwrap() + .write_all(&bytes) + .await + .unwrap(); + } + + async fn next(&mut self) -> Value { + let line = tokio::time::timeout(BOUND, self.output.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + serde_json::from_str(&line).unwrap() + } + + async fn open(&mut self) { + self.send(Packet::Open { rows: 24, cols: 80 }).await; + assert_eq!(self.next().await, json!({"type":"ready"})); + } + + async fn until(&mut self, marker: &str) -> String { + let mut text = String::new(); + while !text.contains(marker) { + assert!(text.len() < 65536, "terminal test output exceeded bound"); + let next = self.next().await; + assert_eq!(next["type"], "packet"); + let packet: Packet = serde_json::from_value(next["packet"].clone()).unwrap(); + text.push_str(&String::from_utf8_lossy(&packet.bytes().unwrap())); + } + text + } + + async fn close(&mut self) { + self.input.take(); + let status = tokio::time::timeout(BOUND, self.child.wait()) + .await + .unwrap() + .unwrap(); + assert!(status.success(), "worker failed after EOF: {status}"); + } +} + +impl Drop for Worker { + fn drop(&mut self) { + if let Some(pid) = self.child.id() { + let _ = nix::sys::signal::kill( + nix::unistd::Pid::from_raw(pid as i32), + nix::sys::signal::Signal::SIGCONT, + ); + let _ = nix::sys::signal::kill( + nix::unistd::Pid::from_raw(pid as i32), + nix::sys::signal::Signal::SIGTERM, + ); + let deadline = std::time::Instant::now() + BOUND; + while std::time::Instant::now() < deadline { + if self.child.try_wait().ok().flatten().is_some() { + return; + } + std::thread::sleep(Duration::from_millis(20)); + } + let _ = self.child.start_kill(); + } + } +} + +/// Stop only the worker until its shell has exited with unread PTY output. +/// A ready timer must not discard that tail, even with a background job still +/// holding the slave open. This fails against the original early-exit branch. +#[tokio::test] +async fn terminal_worker_drains_final_output_after_shell_exit() { + use nix::sys::signal::{kill, Signal}; + use nix::unistd::Pid; + let workspace = tempfile::tempdir().unwrap(); + let mut worker = Worker::start(workspace.path(), 60); + worker.open().await; + let command = b"sh -c 'trap \"\" HUP; echo $$ > hold.pid; sleep 60' & echo $$ > shell.pid; while [ ! -f release ]; do sleep 0.01; done; printf '%s%s\\n' FINAL_ OUTPUT; exit\n"; + worker.send(Packet::data(command).unwrap()).await; + let shell = tokio::time::timeout(BOUND, async { + loop { + if let (Ok(shell), Ok(_)) = ( + std::fs::read_to_string(workspace.path().join("shell.pid")), + std::fs::read_to_string(workspace.path().join("hold.pid")), + ) { + if let Ok(pid) = shell.trim().parse::() { + break pid; + } + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + let worker_pid = Pid::from_raw(worker.child.id().unwrap() as i32); + kill(worker_pid, Signal::SIGSTOP).unwrap(); + std::fs::write(workspace.path().join("release"), b"").unwrap(); + tokio::time::timeout(BOUND, async { + loop { + let stat = std::fs::read_to_string(format!("/proc/{shell}/stat")).unwrap(); + if stat.rsplit_once(')').unwrap().1.split_whitespace().next() == Some("Z") { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + kill(worker_pid, Signal::SIGCONT).unwrap(); + worker.until("FINAL_OUTPUT").await; + for _ in 0..8 { + let event = worker.next().await; + if event["type"] == "ended" { + assert_eq!(event["failed"], false); + assert!(tokio::time::timeout(BOUND, worker.child.wait()) + .await + .unwrap() + .unwrap() + .success()); + return; + } + } + panic!("terminal did not end after its bounded tail drain"); +} + +#[tokio::test] +async fn terminal_worker_has_real_pty_resize_and_a_clean_environment() { + let workspace = tempfile::tempdir().unwrap(); + let mut worker = Worker::start(workspace.path(), 60); + worker.open().await; + worker.send(Packet::Resize { rows: 37, cols: 91 }).await; + worker.send(Packet::data(b"stty size; printf '%s%s\\n' TTY_ OK; test -z \"$BUZZ_PRIVATE_KEY\" && printf '%s%s\\n' CLEAN_ ENV\n").unwrap()).await; + let output = worker.until("CLEAN_ENV").await; + assert!(output.contains("37 91")); + assert!(output.contains("TTY_OK")); + assert!(!output.contains("must-not-reach-shell")); + worker.close().await; +} + +struct Detached(i32, String); +impl Drop for Detached { + fn drop(&mut self) { + let stat = std::fs::read_to_string(format!("/proc/{}/stat", self.0)).ok(); + if stat + .as_deref() + .and_then(|stat| stat.rsplit_once(')')) + .and_then(|(_, stat)| stat.split_whitespace().nth(19)) + == Some(self.1.as_str()) + { + let _ = nix::sys::signal::kill( + nix::unistd::Pid::from_raw(self.0), + nix::sys::signal::Signal::SIGKILL, + ); + } + } +} + +#[tokio::test] +async fn terminal_worker_reaps_setsid_double_fork_children_and_preserves_siblings() { + let workspace = tempfile::tempdir().unwrap(); + let pid_file = workspace.path().join("detached.pid"); + let mut sibling = Command::new("sleep") + .arg("60") + .kill_on_drop(true) + .spawn() + .unwrap(); + let mut worker = Worker::start(workspace.path(), 60); + worker.open().await; + let script = r#"import os,time; p=os.fork(); p and os._exit(0); os.setsid(); p=os.fork(); p and os._exit(0); open("detached.pid","w").write(str(os.getpid())); time.sleep(60)"#; + worker + .send(Packet::data(format!("python3 -c '{script}'\n").as_bytes()).unwrap()) + .await; + let pid = tokio::time::timeout(BOUND, async { + loop { + if let Some(pid) = std::fs::read_to_string(&pid_file) + .ok() + .and_then(|value| value.parse::().ok()) + { + break pid; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .unwrap(); + let stat = std::fs::read_to_string(format!("/proc/{pid}/stat")).unwrap(); + let started = stat + .rsplit_once(')') + .unwrap() + .1 + .split_whitespace() + .nth(19) + .unwrap() + .to_owned(); + let _cleanup = Detached(pid, started); + worker.close().await; + assert!( + !std::path::Path::new(&format!("/proc/{pid}")).exists(), + "detached terminal process survived" + ); + assert!( + sibling.try_wait().unwrap().is_none(), + "harness sibling was killed" + ); + sibling.kill().await.unwrap(); + sibling.wait().await.unwrap(); +} + +#[tokio::test] +async fn terminal_worker_expiry_ends_a_live_shell_without_parent_eof() { + let workspace = tempfile::tempdir().unwrap(); + let mut worker = Worker::start(workspace.path(), 2); + worker.open().await; + assert!(worker.input.is_some()); + let status = tokio::time::timeout(Duration::from_secs(4), worker.child.wait()) + .await + .unwrap() + .unwrap(); + assert!(status.success()); +} + +#[tokio::test] +async fn terminal_worker_rejects_invalid_dimensions_and_bounded_input() { + let workspace = tempfile::tempdir().unwrap(); + let mut invalid = Worker::start(workspace.path(), 60); + invalid.send(Packet::Open { rows: 0, cols: 80 }).await; + assert_eq!( + tokio::time::timeout(BOUND, invalid.child.wait()) + .await + .unwrap() + .unwrap() + .code(), + Some(1) + ); + let mut worker = Worker::start(workspace.path(), 60); + worker.open().await; + let mut oversized = serde_json::to_vec(&Packet::Resize { rows: 24, cols: 80 }).unwrap(); + oversized.extend_from_slice(&vec![b' '; 32769]); + oversized.push(b'\n'); + worker + .input + .as_mut() + .unwrap() + .write_all(&oversized) + .await + .unwrap(); + assert_eq!( + tokio::time::timeout(BOUND, worker.child.wait()) + .await + .unwrap() + .unwrap() + .code(), + Some(1) + ); +} diff --git a/docs/sandbox-terminal-worker.md b/docs/sandbox-terminal-worker.md new file mode 100644 index 00000000000..9145e12afe4 --- /dev/null +++ b/docs/sandbox-terminal-worker.md @@ -0,0 +1,77 @@ +# Contained sandbox terminal worker + +This layer provides a local PTY process boundary for the relay terminal runtime. +It does not subscribe to relay events or accept an attach grant. The runtime +consumer must verify the owner-approved protocol from [NIP-TA](nips/NIP-TA.md) +before starting a worker, and must stop it when that authority ends. + +The harness starts a separate `buzz-acp sandbox-terminal-worker` process with +the workspace and grant expiry. The worker establishes and verifies a Linux +child subreaper before opening a PTY. Reparented jobs, including `setsid` and +double-fork children, therefore remain its descendants. Harness and agent +siblings are outside that process tree. + +The implementation uses the safe async API from +[pty-process 0.5.3](https://docs.rs/pty-process/0.5.3/pty_process/). +It retains the workspace's existing Tokio and Rustix versions. There is no +Python helper, image copy, tunnel or additional network listener. + +## Local protocol and limits + +Input is newline-delimited JSON using the NIP-TA `open`, `data`, `resize` and +`close` packet shapes. Output is `ready`, a `packet` containing bounded data, +or `ended` with a failure flag. The first input must open the PTY. The runtime +must wait for `ready` before acknowledging the remote client's open packet. + +| Resource | Bound | +| --- | --- | +| JSON input or output line | 32 KiB | +| Decoded data packet | 16 KiB | +| Terminal dimensions | 1 through 1000 rows and columns | +| Initial open and each I/O operation | 5 seconds | +| Grant lifetime | At most one hour, enforced with a monotonic deadline | +| Idle without input or resize | Five minutes; empty heartbeats do not reset it | +| Descendants | 256 | +| Process scan | 4096 processes; 4096 bytes per stat record | +| Complete process cleanup | Five seconds | +| Final output after natural shell exit | 64 KiB and one five-second drain deadline | +| Tokio shutdown after a cancelled stdin read | 100 milliseconds | + +The shell is `/bin/sh -i`, running in the supplied workspace as the agent user. +Its environment contains only PATH, HOME, LANG, LC_ALL, TMPDIR and TERM. This +removes ambient relay and model credentials; it does not isolate files or +privileges already available to the sandbox user. Per-session owner approval +must convey that shell access includes those sandbox privileges. + +## Shutdown contract + +EOF, close, termination signals, idle, expiry and I/O failure all lead to +process cleanup. On natural shell exit, the worker first stops background jobs +that could retain the slave, then drains buffered output before reporting the +end. Exceeding the drain byte or time bound reports failure. The worker stops and kills its descendants, then reaps adopted +children until the tree is empty. A pidfd plus process start-time verification +prevents PID reuse from directing a signal at an unrelated process. + +Exit 0 means the operation ended with confirmed containment. Exit 1 means an +operation or receipt failed after containment was confirmed. Exit 2 means +setup or complete containment could not be confirmed. The runtime consumer +must treat exit 2 as fatal and terminate the harness so the container lifecycle +can remove remaining processes. It must never report an ordinary successful +close for that outcome. + +Linux subreaper and pidfd support are required. Other platforms refuse this +worker mode. The existing Buzz app flows do not invoke it. + +## Verification + +```sh +cargo test --locked -p buzz-acp --test sandbox_terminal_worker +cargo test --locked -p buzz-acp --lib terminal_process_selection +``` + +The first suite runs the production binary with a real PTY and checks terminal +resize, environment filtering, EOF, expiry, malformed input, final output after +shell exit with a background holder, and a detached +double-fork child. It also verifies that a sibling process survives cleanup. +The second checks the production process-selection boundary and descendant cap +without sending signals to host processes.