diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index a14b4aa98d6..1f2775992d2 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -86,7 +86,7 @@ nix = { version = "0.31", default-features = false, features = ["signal", "user" [target.'cfg(target_os = "linux")'.dependencies] pty-process = { version = "=0.5.3", features = ["async"] } -rustix = { version = "1.1.4", features = ["process"] } +rustix = { version = "1.1.4", features = ["process", "fs"] } [dev-dependencies] axum = { workspace = true } diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index b99f6675f9c..7058ba19e30 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -3411,6 +3411,9 @@ async fn run_harness( // `IN_FLIGHT_DEADLINE_SECS` expires. let (steer_ack_tx, mut steer_ack_rx) = mpsc::unbounded_channel::(); + let mut terminal_task = + sandbox_terminal::spawn(config, relay.rest_client(), shutdown_rx.clone())?; + // Startup is complete; the main loop now owns graceful shutdown. let _ = startup_ready.send(()); @@ -3616,6 +3619,25 @@ async fn run_harness( tracing::info!("shutting down"); break; } + result = async { + match terminal_task.as_mut() { + Some(task) => task.await, + None => std::future::pending().await, + } + }, if terminal_task.is_some() => { + terminal_task = None; + match result { + Ok(Ok(())) => {}, + Ok(Err(error)) if !sandbox_terminal::containment_failure(&error) => { + tracing::warn!("terminal service unavailable; restart the sandbox to retry"); + } + Ok(Err(error)) => return Err(error), + Err(_) => { + anyhow::bail!("terminal task failed without confirming containment"); + } + } + None + } wake = recovery_wake::wait(&mut respawn_rx, retry_at, maintenance_at) => { Some(PoolEvent::Recovery(wake)) } @@ -4595,6 +4617,9 @@ async fn run_harness( // just as promptly. Timeout is a backstop for a slot stuck outside the // select (e.g. in spawn); only then do we fall back to aborting. let _ = shutdown_tx.send(()); + // PTY teardown precedes checkpoint scans. A failed containment boundary is + // fatal even when ordinary model work could otherwise continue. + sandbox_terminal::stop(terminal_task).await?; let wake_drain = tokio::time::timeout(Duration::from_secs(30), async { while wake_tasks.join_next().await.is_some() {} }) diff --git a/crates/buzz-acp/src/sandbox_terminal.rs b/crates/buzz-acp/src/sandbox_terminal.rs index e8dfc347627..d7e160e4eb1 100644 --- a/crates/buzz-acp/src/sandbox_terminal.rs +++ b/crates/buzz-acp/src/sandbox_terminal.rs @@ -1,11 +1,73 @@ //! 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 authority; +#[cfg(target_os = "linux")] +mod client; +#[cfg(target_os = "linux")] +mod journal; #[cfg(target_os = "linux")] mod process_tree; #[cfg(target_os = "linux")] +mod server; +#[cfg(target_os = "linux")] mod worker; +/// Start terminal service only for a Linux owner-gated, immutable channel thread. +/// Unsupported transports and scopes never subscribe to terminal requests. +pub(crate) fn spawn( + config: &crate::config::Config, + rest: crate::relay::RestClient, + shutdown: tokio::sync::watch::Receiver<()>, +) -> anyhow::Result>>> { + #[cfg(target_os = "linux")] + { + Ok(server::Config::from_harness(config)? + .map(|config| tokio::spawn(server::serve(config, rest, shutdown)))) + } + #[cfg(not(target_os = "linux"))] + { + let _ = (config, rest, shutdown); + Ok(None) + } +} + +/// Only loss of process containment requires stopping the whole harness. +pub(crate) fn containment_failure(error: &anyhow::Error) -> bool { + #[cfg(target_os = "linux")] + { + error.is::() + } + #[cfg(not(target_os = "linux"))] + { + let _ = error; + false + } +} + +/// Shutdown cannot abandon a worker while the harness checkpoints its workspace. +pub(crate) async fn stop( + task: Option>>, +) -> anyhow::Result<()> { + let Some(mut task) = task else { return Ok(()) }; + // The maximum admission (15s), confirmed worker cleanup (15s), final audit + // attempt (5s), and socket close (1s) fit inside this absolute bound. + match tokio::time::timeout(std::time::Duration::from_secs(40), &mut task).await { + Ok(Ok(Ok(()))) => Ok(()), + Ok(Ok(Err(error))) if !containment_failure(&error) => { + tracing::warn!("terminal service stopped with its audit journal retained"); + Ok(()) + } + Ok(Ok(Err(error))) => Err(error), + Ok(Err(_)) => anyhow::bail!("terminal service exited without confirming containment"), + Err(_) => { + task.abort(); + anyhow::bail!("terminal service shutdown could not confirm containment") + } + } +} + /// 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. diff --git a/crates/buzz-acp/src/sandbox_terminal/authority.rs b/crates/buzz-acp/src/sandbox_terminal/authority.rs new file mode 100644 index 00000000000..eaa0ba76394 --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal/authority.rs @@ -0,0 +1,190 @@ +//! Current owner approval and independent channel/delegation authority. + +use anyhow::{ensure, Context, Result}; +use buzz_sdk::sandbox_attach::{Authority, Grant, Pending, Scope}; +use nostr::{Event, EventId, PublicKey, Timestamp}; +use serde_json::{json, Value}; + +use crate::relay::RestClient; + +pub(super) const AUTH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5); + +pub(super) struct Snapshot { + owner: PublicKey, + relay: PublicKey, + roster: Event, + delegations: Vec, + profiles: Vec, +} + +impl Snapshot { + fn authority(&self) -> Authority<'_> { + Authority { + owner: self.owner, + relay: self.relay, + roster: &self.roster, + delegations: &self.delegations, + profiles: &self.profiles, + } + } + + pub(super) fn authors(&self, scope: Scope) -> Vec { + let mut authors = vec![self.owner]; + authors.extend( + buzz_sdk::gateway::resolve( + self.owner, + &self.delegations, + &self.profiles, + &[self.owner, scope.agent], + Timestamp::now().as_secs(), + ) + .into_iter() + .filter(|live| { + buzz_sdk::gateway::channel_member( + &self.roster, + self.relay, + scope.channel, + live.gateway(), + ) + }) + .map(|live| live.gateway()), + ); + authors + } +} + +async fn query(rest: &RestClient, mut filter: Value, limit: usize) -> Result> { + filter["consistency"] = json!("strong"); + let events: Vec = serde_json::from_value(rest.query_raw(&[filter]).await?)?; + ensure!( + events.len() <= limit, + "terminal authority response exceeds bound" + ); + Ok(events) +} + +async fn read(rest: &RestClient, owner: PublicKey, scope: Scope) -> Result { + let relay = rest + .relay_self() + .await? + .and_then(|value| PublicKey::from_hex(&value).ok()) + .context("terminal relay identity unavailable")?; + let delegations = query( + rest, + json!({"kinds":[30181], "authors":[owner], + "limit":buzz_sdk::gateway::LOOKUP_LIMIT}), + buzz_sdk::gateway::LOOKUP_LIMIT, + ) + .await?; + let candidates: Vec<_> = + buzz_sdk::gateway::live_heads(owner, &delegations, Timestamp::now().as_secs()) + .into_iter() + .map(|live| live.gateway()) + .collect(); + let profiles = if candidates.is_empty() { + Vec::new() + } else { + query( + rest, + json!({"kinds":[0], "authors":candidates, "limit":candidates.len()}), + candidates.len(), + ) + .await? + }; + let mut rosters = query( + rest, + json!({"kinds":[39002], "authors":[relay], "#d":[scope.channel], "limit":1}), + 1, + ) + .await?; + let roster = rosters + .pop() + .context("terminal channel membership unavailable")?; + ensure!( + [owner, scope.agent] + .into_iter() + .all(|key| buzz_sdk::gateway::channel_member(&roster, relay, scope.channel, key)), + "terminal owner or agent no longer has posting membership" + ); + Ok(Snapshot { + owner, + relay, + roster, + delegations, + profiles, + }) +} + +pub(super) async fn snapshot( + rest: &RestClient, + owner: PublicKey, + scope: Scope, +) -> Result { + tokio::time::timeout(AUTH_TIMEOUT, read(rest, owner, scope)).await? +} + +async fn current(rest: &RestClient, snapshot: &Snapshot, grant: &Grant) -> Result<()> { + let coordinate = |author, phase| { + json!({"kinds":[30624], "authors":[author], + "#d":[format!("terminal:{}:{phase}", grant.session())], "limit":1}) + }; + let (approvals, pending) = tokio::try_join!( + query(rest, coordinate(snapshot.owner, "grant"), 1), + query(rest, coordinate(grant.requester(), "request"), 1), + )?; + ensure!( + approvals + .as_slice() + .first() + .is_some_and(|event| event.id == grant.id() && event.verify().is_ok()), + "terminal approval was revoked or replaced" + ); + let pending = Pending::parse( + pending.first().context("terminal request was revoked")?, + Timestamp::now().as_secs(), + )?; + ensure!( + grant.authorized(&pending, snapshot.authority(), Timestamp::now().as_secs()), + "terminal approval is not currently authorized" + ); + Ok(()) +} + +pub(super) async fn approve( + rest: &RestClient, + owner: PublicKey, + scope: Scope, + id: EventId, + requester: PublicKey, +) -> Result { + tokio::time::timeout(AUTH_TIMEOUT, async { + let snapshot = read(rest, owner, scope).await?; + let events = query(rest, json!({"kinds":[30624], "ids":[id], "limit":1}), 1).await?; + let event = events.first().context("terminal approval unavailable")?; + ensure!( + event.id == id && event.pubkey == owner, + "terminal approval signer mismatch" + ); + let grant = Grant::parse(event, scope, Timestamp::now().as_secs())?; + ensure!( + grant.requester() == requester, + "terminal requester mismatch" + ); + current(rest, &snapshot, &grant).await?; + Ok(grant) + }) + .await? +} + +pub(super) async fn refresh( + rest: &RestClient, + owner: PublicKey, + grant: &Grant, +) -> Result { + tokio::time::timeout(AUTH_TIMEOUT, async { + let snapshot = read(rest, owner, grant.scope()).await?; + current(rest, &snapshot, grant).await?; + Ok(snapshot) + }) + .await? +} diff --git a/crates/buzz-acp/src/sandbox_terminal/client.rs b/crates/buzz-acp/src/sandbox_terminal/client.rs new file mode 100644 index 00000000000..5173e6e3fc4 --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal/client.rs @@ -0,0 +1,472 @@ +//! Bounded IPC with the contained worker; terminal bytes never enter diagnostics. + +#![cfg(target_os = "linux")] + +use std::os::unix::process::CommandExt; +use std::path::Path; +use std::process::Stdio; +use std::time::Duration; + +use anyhow::{anyhow, bail, ensure, Result}; +use buzz_sdk::sandbox_attach::Packet; +use futures_util::StreamExt; +use serde::Deserialize; +use tokio::io::AsyncWriteExt; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; +use tokio::time::{timeout, timeout_at, Instant}; +use tokio_util::codec::{FramedRead, LinesCodec}; + +const MAX_LINE_BYTES: usize = 32 * 1024; +const IO_TIMEOUT: Duration = Duration::from_secs(5); +/// Full startup budget; admission must retain authority for this entire window. +pub(super) const READY_TIMEOUT: Duration = Duration::from_secs(5); +const CLOSE_TIMEOUT: Duration = Duration::from_secs(15); + +/// The caller must stop the whole harness when worker containment is unconfirmed. +#[derive(Debug, thiserror::Error)] +#[error("terminal worker containment could not be confirmed")] +pub(super) struct ContainmentError; + +/// Validated output from one local PTY worker. +pub(super) enum Event { + /// At most 16 KiB of terminal bytes, without logging or persistence. + Data(Vec), + /// The worker supplied its end receipt and exited after confirmed cleanup. + Ended { failed: bool }, +} + +#[derive(Deserialize)] +#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)] +enum Wire { + Ready, + Packet { packet: Packet }, + Ended { failed: bool }, +} + +/// One worker and its bounded pipes. Every normal exit path must call `close`. +pub(super) struct Pty { + child: Child, + input: Option, + output: Option>, + receipt: Option, + exit_deadline: Option, + close_deadline: Option, + confirmed: bool, +} + +impl Pty { + /// Spawn the internal worker and require readiness within one five-second deadline. + pub(super) async fn open( + executable: &Path, + workspace: &Path, + expires_at: u64, + rows: u16, + cols: u16, + ) -> Result { + Packet::Open { rows, cols }.bytes()?; + let environment = ["PATH", "HOME", "LANG", "LC_ALL", "TMPDIR"] + .into_iter() + .filter_map(|name| std::env::var_os(name).map(|value| (name, value))); + let mut command = Command::new(executable); + // Sprig selects its multicall entry point from argv[0]. + command.as_std_mut().arg0("buzz-acp"); + let mut child = command + .arg("sandbox-terminal-worker") + .arg(workspace) + .arg(expires_at.to_string()) + .env_clear() + .envs(environment) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + // SIGKILL-on-drop could interrupt the worker's descendant cleanup. + .kill_on_drop(false) + .spawn() + .map_err(|_| anyhow!("could not start terminal worker"))?; + let input = child.stdin.take(); + let output = child + .stdout + .take() + .map(|pipe| FramedRead::new(pipe, LinesCodec::new_with_max_length(MAX_LINE_BYTES))); + let mut pty = Self { + child, + input, + output, + receipt: None, + exit_deadline: None, + close_deadline: None, + confirmed: false, + }; + let started = timeout(READY_TIMEOUT, async { + pty.write(Packet::Open { rows, cols }).await?; + match pty.read().await? { + Some(Wire::Ready) => Ok(()), + Some(Wire::Ended { failed }) => { + pty.receipt = Some(failed); + bail!("terminal worker ended before readiness") + } + _ => bail!("terminal worker did not report readiness"), + } + }) + .await + .map_err(|_| anyhow!("terminal worker readiness timed out")) + .and_then(|result| result); + if let Err(error) = started { + // A setup refusal can itself mean containment is unavailable. Never + // downgrade an unconfirmed cleanup to an ordinary startup error. + pty.close().await?; + return Err(error); + } + Ok(pty) + } + + async fn write(&mut self, packet: Packet) -> Result<()> { + packet.bytes()?; + let mut bytes = serde_json::to_vec(&packet) + .map_err(|_| anyhow!("could not encode terminal worker command"))?; + ensure!( + bytes.len() < MAX_LINE_BYTES, + "terminal worker command exceeds bound" + ); + bytes.push(b'\n'); + let input = self + .input + .as_mut() + .ok_or_else(|| anyhow!("terminal worker input is closed"))?; + timeout(IO_TIMEOUT, async { + input.write_all(&bytes).await?; + input.flush().await + }) + .await + .map_err(|_| anyhow!("terminal worker command timed out"))? + .map_err(|_| anyhow!("terminal worker input failed")) + } + + async fn read(&mut self) -> Result> { + let output = self + .output + .as_mut() + .ok_or_else(|| anyhow!("terminal worker output is unavailable"))?; + match output.next().await { + None => Ok(None), + Some(Ok(line)) => serde_json::from_str(&line) + .map(Some) + .map_err(|_| anyhow!("invalid terminal worker output")), + Some(Err(_)) => bail!("terminal worker output exceeds its framing bound"), + } + } + + /// Deliver data, resize, or close; a second open and wire acknowledgements are refused. + pub(super) async fn send(&mut self, packet: Packet) -> Result<()> { + ensure!( + !self.confirmed && self.receipt.is_none(), + "terminal worker has ended" + ); + ensure!( + matches!( + packet, + Packet::Data { .. } | Packet::Resize { .. } | Packet::Close + ), + "invalid terminal worker command" + ); + self.write(packet).await + } + + /// Read one bounded data packet or an end whose process cleanup has been confirmed. + pub(super) async fn next(&mut self) -> Result { + ensure!(!self.confirmed, "terminal worker has ended"); + if self.receipt.is_some() { + // The caller may cancel next() while it awaits the final process + // exit. Keep the receipt so the next poll can finish confirmation. + let failed = self.confirm_bounded().await?; + return Ok(Event::Ended { failed }); + } + match self.read().await? { + Some(Wire::Packet { + packet: Packet::Data { data }, + }) => { + let bytes = Packet::Data { data } + .bytes() + .map_err(|_| anyhow!("invalid terminal worker data"))?; + Ok(Event::Data(bytes)) + } + Some(Wire::Ended { failed }) => { + self.receipt = Some(failed); + let failed = self.confirm_bounded().await?; + Ok(Event::Ended { failed }) + } + None => { + // No receipt is never a successful EOF, including signal exits. + self.confirm_bounded().await?; + Err(ContainmentError.into()) + } + _ => bail!("unexpected terminal worker output"), + } + } + + async fn confirm_bounded(&mut self) -> Result { + let deadline = *self + .exit_deadline + .get_or_insert(Instant::now() + IO_TIMEOUT); + timeout_at(deadline, self.confirm()) + .await + .map_err(|_| ContainmentError)? + } + + async fn confirm(&mut self) -> Result { + let status = self.child.wait().await.map_err(|_| ContainmentError)?; + let Some(failed) = self.receipt else { + return Err(ContainmentError.into()); + }; + if !matches!(status.code(), Some(0 | 1)) { + return Err(ContainmentError.into()); + } + self.confirmed = true; + Ok(failed || status.code() == Some(1)) + } + + fn stop(&mut self) { + // EOF also wakes the worker before it has installed signal handlers. + self.input.take(); + if let Some(pid) = self.child.id().and_then(|pid| i32::try_from(pid).ok()) { + // A live Child retains its PID until wait; after wait id() is None. + // The explicit close path verifies the outcome even if signalling fails. + let _ = nix::sys::signal::kill( + nix::unistd::Pid::from_raw(pid), + nix::sys::signal::Signal::SIGTERM, + ); + } + } + + /// Request cleanup, drain/discard bounded packets, and verify the end receipt and exit. + pub(super) async fn close(&mut self) -> Result<()> { + if self.confirmed { + return Ok(()); + } + self.stop(); + let deadline = *self + .close_deadline + .get_or_insert(Instant::now() + CLOSE_TIMEOUT); + timeout_at(deadline, async { + loop { + match self.read().await { + Ok(Some(Wire::Ended { failed })) => self.receipt = Some(failed), + Ok(Some(Wire::Ready)) => {} + Ok(Some(Wire::Packet { + packet: Packet::Data { data }, + })) => { + Packet::Data { data } + .bytes() + .map_err(|_| ContainmentError)?; + } + Ok(None) => break, + _ => return Err(ContainmentError.into()), + } + } + self.confirm().await.map(|_| ()) + }) + .await + .map_err(|_| ContainmentError)? + } +} + +impl Drop for Pty { + fn drop(&mut self) { + if !self.confirmed { + self.stop(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::os::unix::fs::PermissionsExt; + + const PEER: &str = r#"#!/usr/bin/python3 +import base64,json,pathlib,signal,sys,time +root=pathlib.Path(sys.argv[2]) +def end(signum=None,frame=None): + (root/'cleaned').write_text('yes') + print(json.dumps({'type':'ended','failed':False}),flush=True) + sys.exit(0) +signal.signal(signal.SIGTERM,end) +packet=json.loads(sys.stdin.readline()) +(root/'pid').write_text(str(__import__('os').getpid())) +if packet['rows']==47:signal.signal(signal.SIGTERM,signal.SIG_IGN) +if packet['rows']==41: + print(json.dumps({'type':'packet','packet':{'type':'data','data':'eA=='}}),flush=True) +else: + print(json.dumps({'type':'ready'}),flush=True) +if packet['rows']==42: + print(json.dumps({'type':'ended','failed':False}),flush=True) + sys.exit(1) +if packet['rows']==43: + print(json.dumps({'type':'ended','failed':False}),flush=True) + sys.exit(2) +if packet['rows']==44: + print(json.dumps({'type':'ended','failed':False}),flush=True) + while not (root/'exit-now').exists():time.sleep(.01) + sys.exit(0) +if packet['rows']==45: + print(json.dumps({'type':'packet','packet':{'type':'data','data':base64.b64encode(b'x'*16385).decode()}}),flush=True) +if packet['rows']==47: + while True:time.sleep(.01) +for line in sys.stdin: + incoming=json.loads(line) + if incoming['type']=='data': + print(json.dumps({'type':'packet','packet':incoming}),flush=True) +end() +"#; + + fn fixture() -> tempfile::TempDir { + let root = tempfile::tempdir().unwrap(); + let peer = root.path().join("peer"); + std::fs::write(&peer, PEER).unwrap(); + std::fs::set_permissions(&peer, std::fs::Permissions::from_mode(0o700)).unwrap(); + root + } + + async fn open(root: &tempfile::TempDir, rows: u16) -> Result { + Pty::open(&root.path().join("peer"), root.path(), 1, rows, 80).await + } + + #[tokio::test] + async fn production_client_bounds_data_and_confirms_startup_cleanup() { + let root = fixture(); + let mut pty = open(&root, 24).await.unwrap(); + pty.send(Packet::data(b"private bytes").unwrap()) + .await + .unwrap(); + assert!( + matches!(pty.next().await.unwrap(), Event::Data(bytes) if bytes == b"private bytes") + ); + assert!(pty.send(Packet::Open { rows: 24, cols: 80 }).await.is_err()); + pty.close().await.unwrap(); + assert!(root.path().join("cleaned").exists()); + assert!(pty.confirmed); + + let refused = fixture(); + let error = open(&refused, 41).await.err().unwrap(); + assert!( + error.downcast_ref::().is_none(), + "{error}" + ); + assert!(refused.path().join("cleaned").exists()); + + let oversized = fixture(); + let mut pty = open(&oversized, 45).await.unwrap(); + let error = pty.next().await.err().unwrap(); + assert_eq!(error.to_string(), "invalid terminal worker data"); + pty.close().await.unwrap(); + } + + #[tokio::test] + async fn process_status_controls_failure_and_containment() { + for (rows, fatal) in [(42, false), (43, true)] { + let root = fixture(); + let mut pty = open(&root, rows).await.unwrap(); + match pty.next().await { + Ok(Event::Ended { failed }) => { + assert!(!fatal); + assert!(failed, "exit 1 must not become a successful end"); + } + Err(error) => { + assert!(fatal); + assert!(error.downcast_ref::().is_some()); + } + _ => panic!("expected end or containment failure"), + } + } + } + + #[tokio::test] + async fn cancelled_exit_confirmation_retains_receipt_and_deadline() { + let root = fixture(); + let mut pty = open(&root, 44).await.unwrap(); + timeout(Duration::from_secs(2), async { + while pty.receipt.is_none() { + assert!(timeout(Duration::from_millis(25), pty.next()) + .await + .is_err()); + } + }) + .await + .unwrap(); + assert_eq!(pty.receipt, Some(false)); + let deadline = pty.exit_deadline.unwrap(); + assert!(timeout(Duration::from_millis(25), pty.next()) + .await + .is_err()); + assert_eq!(pty.exit_deadline, Some(deadline)); + std::fs::write(root.path().join("exit-now"), "yes").unwrap(); + assert!(matches!( + pty.next().await.unwrap(), + Event::Ended { failed: false } + )); + } + + #[tokio::test] + async fn shutdown_timeout_is_fatal_and_does_not_sigkill() { + let root = fixture(); + let mut pty = open(&root, 47).await.unwrap(); + let pidfd = rustix::process::pidfd_open( + rustix::process::Pid::from_raw(pty.child.id().unwrap() as i32).unwrap(), + rustix::process::PidfdFlags::empty(), + ) + .unwrap(); + // The same stored deadline is reused after a cancelled close. A short + // pre-existing deadline exercises its enforcement without a 15s test. + pty.close_deadline = Some(Instant::now() + Duration::from_millis(100)); + let outcome = timeout(Duration::from_secs(2), pty.close()).await; + let still_alive = pty.child.try_wait().unwrap().is_none(); + // This fixture intentionally ignores cleanup. Reap only this test peer. + let _ = rustix::process::pidfd_send_signal(&pidfd, rustix::process::Signal::KILL); + let _ = pty.child.wait().await; + assert!( + still_alive, + "close must not SIGKILL instead of confirming cleanup" + ); + let error = outcome.unwrap().unwrap_err(); + assert!(error.downcast_ref::().is_some()); + } + + #[tokio::test] + async fn drop_requests_cleanup_without_sigkill() { + let root = fixture(); + let pty = open(&root, 24).await.unwrap(); + let pid = nix::unistd::Pid::from_raw(pty.child.id().unwrap() as i32); + let pidfd = rustix::process::pidfd_open( + rustix::process::Pid::from_raw(pid.as_raw()).unwrap(), + rustix::process::PidfdFlags::empty(), + ) + .unwrap(); + nix::sys::signal::kill(pid, nix::sys::signal::Signal::SIGSTOP).unwrap(); + // Observe that the peer is actually stopped before testing Drop. + let mut stopped = false; + for _ in 0..100 { + let stat = std::fs::read_to_string(format!("/proc/{}/stat", pid.as_raw())).unwrap(); + if stat + .rsplit_once(')') + .unwrap() + .1 + .trim_start() + .starts_with('T') + { + stopped = true; + break; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + assert!(stopped, "peer did not stop at the scheduling barrier"); + drop(pty); + let _ = rustix::process::pidfd_send_signal(&pidfd, rustix::process::Signal::CONT); + timeout(Duration::from_secs(2), async { + while !root.path().join("cleaned").exists() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + } +} diff --git a/crates/buzz-acp/src/sandbox_terminal/journal.rs b/crates/buzz-acp/src/sandbox_terminal/journal.rs new file mode 100644 index 00000000000..4475a9c7c22 --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal/journal.rs @@ -0,0 +1,202 @@ +//! Durable single-use reservation and metadata-only audit outbox. + +use std::io::{Read, Write}; +use std::os::unix::fs::{OpenOptionsExt, PermissionsExt}; +use std::path::PathBuf; + +use anyhow::{ensure, Context, Result}; +use buzz_sdk::sandbox_attach::{audit, Grant, Scope}; +use nostr::{Event, EventId, Keys, PublicKey, Timestamp}; +use serde::{Deserialize, Serialize}; + +const MAX_RECORDS: usize = 256; +const MAX_RECORD_BYTES: u64 = 16 * 1024; + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Record { + pub(super) grant: Event, + pub(super) accepted: Event, + pub(super) ended: Option, + /// Only the exact first-claim receipt can set this before worker startup. + pub(super) claimed: bool, + pub(super) ended_sent: bool, + pub(super) bytes_in: u64, + pub(super) bytes_out: u64, +} + +pub(super) struct Journal { + directory: PathBuf, + scope: Scope, + owner: PublicKey, + _lease: std::fs::File, +} + +impl Journal { + pub(super) fn open(directory: PathBuf, scope: Scope, owner: PublicKey) -> Result { + std::fs::create_dir_all(&directory)?; + std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o700))?; + let lease = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .mode(0o600) + .custom_flags(nix::libc::O_NOFOLLOW) + .open(directory.join(".lock"))?; + rustix::fs::flock(&lease, rustix::fs::FlockOperation::NonBlockingLockExclusive) + .context("another terminal server owns this sandbox journal")?; + for (count, entry) in std::fs::read_dir(&directory)?.enumerate() { + ensure!( + count < MAX_RECORDS * 4 + 1, + "terminal journal directory exceeds bound" + ); + let entry = entry?; + if entry.file_name().to_string_lossy().starts_with(".tmp") { + ensure!( + entry.file_type()?.is_file(), + "invalid terminal journal staging file" + ); + std::fs::remove_file(entry.path())?; + } + } + Ok(Self { + directory, + scope, + owner, + _lease: lease, + }) + } + + fn path(&self, id: EventId) -> PathBuf { + self.directory.join(format!("{}.json", id.to_hex())) + } + + pub(super) fn save(&self, record: &Record) -> Result<()> { + let bytes = serde_json::to_vec(record)?; + ensure!( + bytes.len() <= MAX_RECORD_BYTES as usize, + "terminal journal record exceeds bound" + ); + let mut staged = tempfile::NamedTempFile::new_in(&self.directory)?; + staged.write_all(&bytes)?; + staged.as_file().sync_all()?; + staged.persist(self.path(record.grant.id))?; + std::fs::File::open(&self.directory)?.sync_all()?; + Ok(()) + } + + pub(super) fn records(&self) -> Result> { + let mut records = Vec::new(); + for entry in std::fs::read_dir(&self.directory)? { + let entry = entry?; + if entry.file_name() == ".lock" { + continue; + } + ensure!( + records.len() < MAX_RECORDS, + "terminal audit journal is full" + ); + ensure!( + entry.file_type()?.is_file(), + "invalid terminal journal entry" + ); + let mut bytes = Vec::new(); + std::fs::OpenOptions::new() + .read(true) + .custom_flags(nix::libc::O_NOFOLLOW) + .open(entry.path())? + .take(MAX_RECORD_BYTES + 1) + .read_to_end(&mut bytes)?; + ensure!( + bytes.len() <= MAX_RECORD_BYTES as usize, + "terminal journal record exceeds bound" + ); + let record: Record = serde_json::from_slice(&bytes)?; + ensure!( + entry.path() == self.path(record.grant.id), + "terminal journal coordinate mismatch" + ); + let grant = Grant::parse(&record.grant, self.scope, record.grant.created_at.as_secs())?; + ensure!( + record.grant.verify().is_ok() + && record.accepted.verify().is_ok() + && record + .ended + .as_ref() + .is_none_or(|event| event.verify().is_ok()), + "terminal journal signature invalid" + ); + ensure!( + record.grant.pubkey == self.owner + && record.accepted.pubkey == self.scope.agent + && buzz_core::sandbox_attach::acceptance_coordinate(&record.accepted) + == Some(("terminal", grant.session())) + && record + .ended + .as_ref() + .is_none_or(|event| event.pubkey == self.scope.agent), + "terminal journal authority mismatch" + ); + records.push(record); + } + Ok(records) + } + + /// An existing record consumes this approval, including after a crash or a + /// lost relay receipt. A new shell always requires a new owner approval. + pub(super) fn reserve(&self, keys: &Keys, grant: &Grant) -> Result { + ensure!( + !self.path(grant.id()).try_exists()?, + "terminal approval was already consumed" + ); + ensure!( + self.records()?.len() < MAX_RECORDS, + "terminal audit journal is full" + ); + let record = Record { + grant: grant.event().clone(), + accepted: audit(keys, grant, "accepted", 0, 0)?, + ended: None, + claimed: false, + ended_sent: false, + bytes_in: 0, + bytes_out: 0, + }; + self.save(&record)?; + Ok(record) + } + + pub(super) fn finish(&self, keys: &Keys, record: &mut Record) -> Result<()> { + // An unconfirmed or losing claim must not publish an end record that + // appears to end another replica's successful attempt. + if record.claimed && record.ended.is_none() { + // Recovery may finish a grant after expiry; parse at its creation + // time solely to rebuild audit metadata, never to reopen a shell. + let grant = Grant::parse(&record.grant, self.scope, record.grant.created_at.as_secs())?; + record.ended = Some(audit( + keys, + &grant, + "ended", + record.bytes_in, + record.bytes_out, + )?); + self.save(record)?; + } + Ok(()) + } + + pub(super) fn recover(&self, keys: &Keys) -> Result<()> { + for mut record in self.records()? { + self.finish(keys, &mut record)?; + let grant = Grant::parse(&record.grant, self.scope, record.grant.created_at.as_secs())?; + if (!record.claimed || record.ended_sent) + && grant.expires_at() <= Timestamp::now().as_secs() + { + std::fs::remove_file(self.path(record.grant.id))?; + std::fs::File::open(&self.directory)?.sync_all()?; + } + } + Ok(()) + } +} diff --git a/crates/buzz-acp/src/sandbox_terminal/server.rs b/crates/buzz-acp/src/sandbox_terminal/server.rs new file mode 100644 index 00000000000..34c87980a82 --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal/server.rs @@ -0,0 +1,556 @@ +//! One owner-approved PTY, with independent live authority and reliable relay I/O. + +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::{bail, ensure, Context, Result}; +use buzz_sdk::sandbox_attach::{self as attach, Flow, Frame, Grant, Packet, Rate, Scope}; +use buzz_ws_client::{NostrWsConnection, OkResponse, RelayMessage}; +use nostr::{Event, EventId, Keys, PublicKey, Tag, Timestamp}; +use serde_json::json; +use tokio::sync::watch; +use tokio::time::{timeout, Instant}; + +use super::{ + authority, client, + journal::{Journal, Record}, +}; +use crate::config::{BoundSession, RespondTo}; +use crate::relay::RestClient; + +const REFRESH: Duration = Duration::from_secs(5); +const IO_TIMEOUT: Duration = Duration::from_secs(1); +const IDLE: Duration = Duration::from_secs(300); +const DRAIN: Duration = Duration::from_secs(6); +const SUBSCRIPTION: &str = "sandbox-terminal"; + +#[derive(Clone)] +pub(super) struct Config { + pub(super) relay: String, + pub(super) keys: Keys, + pub(super) auth: Option, + pub(super) owner: PublicKey, + pub(super) scope: Scope, + pub(super) workspace: PathBuf, + pub(super) executable: PathBuf, + pub(super) journal: PathBuf, +} + +impl Config { + pub(super) fn from_harness(config: &crate::config::Config) -> Result> { + let Some(BoundSession::Thread { + channel_id, + thread_root, + }) = &config.bound_session + else { + return Ok(None); + }; + if config.respond_to != RespondTo::Owner + || url::Url::parse(&config.relay_url)?.scheme() != "wss" + { + return Ok(None); + } + let Some(owner) = crate::resolve_agent_owner(config) else { + return Ok(None); + }; + let owner = PublicKey::from_hex(&owner)?; + ensure!( + owner != config.keys.public_key(), + "terminal owner must be distinct from agent" + ); + let scope = Scope { + channel: *channel_id, + thread: EventId::from_hex(thread_root)?, + agent: config.keys.public_key(), + }; + let home = std::env::var_os("HOME").context("terminal journal requires HOME")?; + let journal = PathBuf::from(home) + .join(".local/state/buzz/terminal") + .join(scope.agent.to_hex()) + .join(scope.thread.to_hex()); + let auth = std::env::var("BUZZ_AUTH_TAG") + .ok() + .filter(|value| !value.is_empty()) + .map(|value| buzz_sdk::nip_oa::parse_auth_tag(&value)) + .transpose()?; + Ok(Some(Self { + relay: config.relay_url.clone(), + keys: config.keys.clone(), + auth, + owner, + scope, + workspace: std::env::current_dir()?, + executable: std::env::current_exe()?, + journal, + })) + } +} + +struct Session { + grant: Grant, + record: Record, + pty: client::Pty, + flow: Flow, + last_sent: u64, + acknowledged: u64, + idle: Instant, + heartbeat: Instant, + exited: Option, + close_sent: bool, + output: Option>, + authority_deadline: Instant, +} + +struct RefreshResult { + generation: Option, + checked_at: Instant, + verdict: Result, +} + +fn elapsed(start: Instant) -> u64 { + u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX) +} + +fn frame_admitted(event: &Event, budget: &mut Rate, ms: u64) -> bool { + // NIP44 padding/base64 can exceed twice the decoded packet budget. Use + // the shared codec's bounded envelope before any signature/decryption; + // Packet::bytes separately enforces the 16KiB decoded data limit. + buzz_core::observer::content_looks_like_nip44(&event.content) && budget.allow(ms) +} + +fn authority_deadline(grant: &Grant, checked_at: Instant) -> Instant { + let expiry = std::time::UNIX_EPOCH + Duration::from_secs(grant.expires_at()); + let remaining = expiry + .duration_since(std::time::SystemTime::now()) + .unwrap_or(Duration::ZERO); + (checked_at + REFRESH + authority::AUTH_TIMEOUT).min(Instant::now() + remaining) +} + +async fn raw(connection: &mut NostrWsConnection, value: serde_json::Value) -> Result<()> { + timeout(IO_TIMEOUT, connection.send_raw(&value)).await??; + Ok(()) +} + +/// Read the real per-write receipt without the shared client's unbounded +/// wait-for-OK buffer. No worker is running while metadata is published. +async fn publish(connection: &mut NostrWsConnection, event: &Event) -> Result { + timeout(REFRESH, async { + raw(connection, json!(["EVENT", event])).await?; + for _ in 0..256 { + if let RelayMessage::Ok(receipt) = connection.next_event(REFRESH).await? { + if receipt.event_id == event.id.to_hex() { + return Ok(receipt); + } + } + } + bail!("terminal receipt message budget exhausted") + }) + .await? +} + +fn first_claim(receipt: &OkResponse, event: &Event) -> Result<()> { + ensure!( + receipt.accepted + && receipt.event_id == event.id.to_hex() + && receipt.message == attach::CLAIM_OK_MESSAGE, + "terminal first claim was not confirmed" + ); + Ok(()) +} + +async fn subscribe( + connection: &mut NostrWsConnection, + config: &Config, + snapshot: &authority::Snapshot, +) -> Result<()> { + raw( + connection, + json!(["REQ", SUBSCRIPTION, {"kinds":[24203], + "authors":snapshot.authors(config.scope), "#p":[config.scope.agent], + "#h":[config.scope.channel], "#e":[config.scope.thread], "limit":0}]), + ) + .await +} + +async fn send( + connection: &mut NostrWsConnection, + config: &Config, + session: &Session, + frame: &Frame, +) -> Result<()> { + let event = attach::encrypt(&config.keys, session.grant.requester(), config.scope, frame)?; + tokio::time::timeout_at( + session.authority_deadline, + raw(connection, json!(["EVENT", event])), + ) + .await? +} + +async fn queue( + connection: &mut NostrWsConnection, + config: &Config, + session: &mut Session, + packet: Packet, + ms: u64, +) -> Result<()> { + let frame = session.flow.send(packet, ms)?; + session.last_sent = frame.seq; + send(connection, config, session, &frame).await +} + +async fn outbox(connection: &mut NostrWsConnection, journal: &Journal) -> Result<()> { + // One event per idle tick bounds both work and receipt-wait time. Retain + // every failure in the journal until the exact stored event is accepted. + for mut record in journal.records()? { + if record.ended_sent { + continue; + } + if let Some(event) = &record.ended { + let receipt = publish(connection, event).await?; + ensure!( + receipt.accepted && receipt.event_id == event.id.to_hex(), + "terminal end audit refused" + ); + record.ended_sent = true; + journal.save(&record)?; + break; + } + } + Ok(()) +} + +async fn open( + connection: &mut NostrWsConnection, + config: &Config, + rest: &RestClient, + journal: &Journal, + event: &Event, + outgoing: &mut Rate, + start: Instant, +) -> Result { + // Bound ciphertext before decryption. The second typed decryption binds + // the discovered approval ID to the complete signed envelope. + ensure!( + buzz_core::observer::content_looks_like_nip44(&event.content) && event.verify().is_ok(), + "invalid terminal open" + ); + let discovered: Frame = buzz_core::observer::decrypt_observer_payload(&config.keys, event)?; + let frame = attach::decrypt( + &config.keys, + event.pubkey, + config.scope, + discovered.grant, + event, + Timestamp::now().as_secs(), + )?; + let Packet::Open { rows, cols } = frame.packet else { + bail!("terminal requires Open") + }; + ensure!( + frame.seq == 1 && frame.ack == 0, + "terminal Open must start its transport" + ); + let checked_at = Instant::now(); + let grant = + authority::approve(rest, config.owner, config.scope, frame.grant, event.pubkey).await?; + let mut record = journal.reserve(&config.keys, &grant)?; + let receipt = publish(connection, &record.accepted).await?; + first_claim(&receipt, &record.accepted)?; + record.claimed = true; + journal.save(&record)?; + let deadline = authority_deadline(&grant, checked_at); + if Instant::now() + client::READY_TIMEOUT >= deadline { + journal.finish(&config.keys, &mut record)?; + bail!("terminal admission has insufficient live authority for readiness"); + } + let pty = match client::Pty::open( + &config.executable, + &config.workspace, + grant.expires_at(), + rows, + cols, + ) + .await + { + Ok(pty) => pty, + // The durable owned acceptance already permits recovery of its audit. + // A second journal failure must never downgrade lost containment. + Err(error) if error.is::() => return Err(error), + Err(error) => { + journal.finish(&config.keys, &mut record)?; + return Err(error); + } + }; + let mut flow = Flow::new(grant.id()); + flow.receive(Frame { + grant: frame.grant, + seq: 1, + ack: 0, + packet: Packet::Open { rows, cols }, + })?; + let mut session = Session { + grant, + record, + pty, + flow, + last_sent: 0, + acknowledged: 0, + idle: Instant::now(), + heartbeat: Instant::now(), + exited: None, + close_sent: false, + output: None, + authority_deadline: deadline, + }; + // An ACK cannot precede the worker's Ready receipt. Errors after spawn + // still confirm containment and durably finish the owned acceptance. + let ready = async { + ensure!( + outgoing.allow(elapsed(start)), + "terminal send budget exhausted" + ); + send( + connection, + config, + &session, + &session.flow.acknowledgement(), + ) + .await + } + .await; + if let Err(error) = ready { + session.pty.close().await?; + journal.finish(&config.keys, &mut session.record)?; + return Err(error); + } + Ok(session) +} + +async fn finish(config: &Config, journal: &Journal, active: &mut Option) -> Result<()> { + if let Some(mut session) = active.take() { + session.pty.close().await?; + journal.finish(&config.keys, &mut session.record)?; + } + Ok(()) +} + +/// The test-only constructor can supply loopback transport and a real worker +/// executable. Production reaches this engine only through the WSS factory. +pub(super) async fn serve( + config: Config, + rest: RestClient, + mut shutdown: watch::Receiver<()>, +) -> Result<()> { + for attempt in 0..5 { + match run(&config, &rest, shutdown.clone()).await { + Ok(()) => return Ok(()), + Err(error) if error.is::() => return Err(error), + Err(_) => { + tracing::warn!("terminal transport stopped; retrying with durable audit state") + } + } + tokio::select! { + _ = shutdown.changed() => return Ok(()), + _ = tokio::time::sleep(Duration::from_secs(1 << attempt)) => {}, + } + } + bail!("terminal transport retries exhausted") +} + +async fn run(config: &Config, rest: &RestClient, mut shutdown: watch::Receiver<()>) -> Result<()> { + let journal = Journal::open(config.journal.clone(), config.scope, config.owner)?; + journal.recover(&config.keys)?; + let mut connection = tokio::select! { + _ = shutdown.changed() => return Ok(()), + result = timeout(Duration::from_secs(45), NostrWsConnection::connect_authenticated( + &config.relay, &config.keys, config.auth.as_ref())) => result??, + }; + let snapshot = tokio::select! { + _ = shutdown.changed() => return Ok(()), + result = authority::snapshot(rest, config.owner, config.scope) => result?, + }; + subscribe(&mut connection, config, &snapshot).await?; + let start = Instant::now(); + let mut incoming = Rate::default(); + let mut outgoing = Rate::default(); + let mut active: Option = None; + let mut refresh: Option> = None; + let mut refreshed = Instant::now(); + let mut next_open = Instant::now(); + let mut tick = tokio::time::interval(Duration::from_millis(50)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let result: Result<()> = async { + loop { + if active.as_ref().is_some_and(|session| Instant::now() >= session.authority_deadline + || Timestamp::now().as_secs() >= session.grant.expires_at() || session.idle.elapsed() >= IDLE) { + finish(config, &journal, &mut active).await?; + } + if refresh.is_none() && refreshed.elapsed() >= REFRESH { + refreshed = Instant::now(); + let rest = rest.clone(); + let owner = config.owner; + let scope = config.scope; + let grant = active.as_ref().map(|session| session.grant.clone()); + refresh = Some(tokio::spawn(async move { + let checked_at = Instant::now(); + let generation = grant.as_ref().map(Grant::id); + let verdict = match grant { + Some(grant) => authority::refresh(&rest, owner, &grant).await, + None => authority::snapshot(&rest, owner, scope).await, + }; + RefreshResult { generation, checked_at, verdict } + })); + } + tokio::select! { + _ = shutdown.changed() => break, + ready = async { match refresh.as_mut() { + Some(task) => task.await, + None => std::future::pending().await, + } }, if refresh.is_some() => { + refresh = None; + let RefreshResult { generation, checked_at, verdict } = ready?; + if generation != active.as_ref().map(|session| session.grant.id()) { continue; } + match verdict { + Ok(snapshot) => { + if let Some(session) = &mut active { + // A late result cannot revive an expired lease. + if Instant::now() >= session.authority_deadline { finish(config, &journal, &mut active).await?; continue; } + session.authority_deadline = authority_deadline(&session.grant, checked_at); + } + if let Some(session) = &active { + tokio::time::timeout_at(session.authority_deadline, subscribe(&mut connection, config, &snapshot)).await??; + } else { subscribe(&mut connection, config, &snapshot).await?; } + } + Err(_) => { finish(config, &journal, &mut active).await?; } + } + } + message = connection.next_event(Duration::from_secs(30)) => { + match message? { + RelayMessage::Closed { subscription_id, .. } if subscription_id == SUBSCRIPTION => bail!("terminal subscription closed"), + RelayMessage::Event { subscription_id, event } if subscription_id == SUBSCRIPTION => { + let ms = elapsed(start); + if !frame_admitted(&event, &mut incoming, ms) { continue; } + if let Some(session) = &mut active { + if event.pubkey != session.grant.requester() { continue; } + let Ok(frame) = attach::decrypt(&config.keys, event.pubkey, config.scope, session.grant.id(), &event, Timestamp::now().as_secs()) else { continue; }; + let sequence = frame.seq; + let acknowledged = frame.ack; + let packet = session.flow.receive(frame)?; + session.acknowledged = session.acknowledged.max(acknowledged); + if let Some(packet) = packet { + match packet { + Packet::Close => { finish(config, &journal, &mut active).await?; continue; } + Packet::Open { .. } => bail!("terminal received a second Open"), + Packet::Ack => {}, + packet if session.exited.is_none() => { + let bytes = packet.bytes()?.len() as u64; + if bytes > 0 || matches!(packet, Packet::Resize { .. }) { session.idle = Instant::now(); } + session.record.bytes_in = session.record.bytes_in.saturating_add(bytes); + tokio::time::timeout_at(session.authority_deadline, session.pty.send(packet)).await??; + journal.save(&session.record)?; + } + _ => {}, + } + } + if sequence != 0 && outgoing.allow(ms) { + send(&mut connection, config, session, &session.flow.acknowledgement()).await?; + } + } else if Instant::now() >= next_open { + next_open = Instant::now() + Duration::from_secs(1); + match open(&mut connection, config, rest, &journal, &event, &mut outgoing, start).await { + Ok(session) => { + // Admission latency must not push the first refresh past + // the authority lease that began before its HTTP lookup. + refreshed = session.authority_deadline.checked_sub(REFRESH + authority::AUTH_TIMEOUT).unwrap_or_else(Instant::now); + active = Some(session); + } + Err(error) if error.is::() => return Err(error), + Err(_) => tracing::info!("terminal request refused"), + } + } + } + _ => {}, + } + } + worker = async { match active.as_mut() { + Some(session) => session.pty.next().await, + None => std::future::pending().await, + } }, if active.as_ref().is_some_and(|session| session.exited.is_none() && session.output.is_none() && session.flow.writable()) => { + let session = active.as_mut().context("terminal session disappeared")?; + match worker? { + client::Event::Data(bytes) => { + session.idle = Instant::now(); + session.output = Some(bytes); + } + client::Event::Ended { failed } => { + session.exited = Some(Instant::now()); + if failed { tracing::info!("terminal worker ended with an operation failure"); } + } + } + } + _ = tick.tick() => { + let ms = elapsed(start); + if let Some(session) = &mut active { + if Timestamp::now().as_secs() >= session.grant.expires_at() || session.idle.elapsed() >= IDLE { + finish(config, &journal, &mut active).await?; + continue; + } + for frame in session.flow.retries(ms)? { + if outgoing.allow(ms) { send(&mut connection, config, session, &frame).await?; } + } + if session.output.is_some() && session.flow.writable() && outgoing.allow(ms) { + let bytes = session.output.take().context("terminal output disappeared")?; + session.record.bytes_out = session.record.bytes_out.saturating_add(bytes.len() as u64); + journal.save(&session.record)?; + queue(&mut connection, config, session, Packet::data(&bytes)?, ms).await?; + } + if session.exited.is_some_and(|at| at.elapsed() >= DRAIN) + || (session.close_sent && session.acknowledged >= session.last_sent) { + finish(config, &journal, &mut active).await?; + continue; + } + if session.exited.is_some() && !session.close_sent && session.acknowledged >= session.last_sent && outgoing.allow(ms) { + queue(&mut connection, config, session, Packet::Close, ms).await?; + session.close_sent = true; + } else if session.exited.is_none() && session.heartbeat.elapsed() >= REFRESH && session.flow.writable() && outgoing.allow(ms) { + queue(&mut connection, config, session, Packet::data(&[])?, ms).await?; + session.heartbeat = Instant::now(); + } + } else { + // Transport failures leave the exact metadata event + // durable. The next idle tick retries after backoff. + if refreshed.elapsed() >= REFRESH { + if outbox(&mut connection, &journal).await.is_err() { + tracing::info!("terminal audit remains queued for retry"); + } + journal.recover(&config.keys)?; + } + } + } + } + } + Ok(()) + }.await; + if let Some(task) = refresh { + task.abort(); + } + // Containment failures take precedence over network or authorization errors. + let cleanup = finish(config, &journal, &mut active).await; + if result + .as_ref() + .is_err_and(|error| error.is::()) + { + return result; + } + cleanup?; + // Failure remains durable and does not turn a completed close into success + // without an audit retry record. + if outbox(&mut connection, &journal).await.is_err() { + tracing::info!("terminal audit retained for next startup"); + } + let _ = timeout(IO_TIMEOUT, connection.disconnect()).await; + result +} + +#[cfg(test)] +mod tests; diff --git a/crates/buzz-acp/src/sandbox_terminal/server/tests.rs b/crates/buzz-acp/src/sandbox_terminal/server/tests.rs new file mode 100644 index 00000000000..ed5376fe807 --- /dev/null +++ b/crates/buzz-acp/src/sandbox_terminal/server/tests.rs @@ -0,0 +1,723 @@ +//! Real HTTP/WS and spawned IPC bind the production admission path. + +use std::os::unix::fs::PermissionsExt; +use std::sync::{Arc, Mutex}; + +use axum::{ + extract::State, + routing::{get, post}, + Json, Router, +}; +use futures_util::{SinkExt, StreamExt}; +use nostr::{EventBuilder, Kind}; +use serde_json::Value; +use tokio_tungstenite::tungstenite::Message; +use uuid::Uuid; + +use super::*; + +struct Fixture { + config: Config, + rest: RestClient, + events: Arc>>, + request: Event, + approval: Event, + owner: Keys, + relay: Keys, + requester: Keys, + directory: tempfile::TempDir, + http: tokio::task::JoinHandle<()>, +} + +async fn query( + State(events): State>>>, + Json(filters): Json>, +) -> Json { + let filter = &filters[0]; + assert_eq!( + filter["consistency"], "strong", + "all authority/head reads must use writer consistency" + ); + let mut result: Vec<_> = events + .lock() + .unwrap() + .iter() + .filter(|event| { + let includes = |key, value| { + filter[key] + .as_array() + .is_none_or(|items| items.contains(&value)) + }; + includes("kinds", json!(event.kind.as_u16())) + && includes("authors", json!(event.pubkey)) + && includes("ids", json!(event.id)) + && filter["#d"].as_array().is_none_or(|items| { + event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.first().is_some_and(|key| key == "d") + && values + .get(1) + .is_some_and(|value| items.contains(&json!(value))) + }) + }) + }) + .cloned() + .collect(); + result.sort_by(|a, b| b.created_at.cmp(&a.created_at).then(a.id.cmp(&b.id))); + result.truncate(filter["limit"].as_u64().unwrap_or(256) as usize); + Json(json!(result)) +} + +impl Fixture { + async fn new(gateway: bool) -> Self { + let (owner, agent, relay, requester) = ( + Keys::generate(), + Keys::generate(), + Keys::generate(), + Keys::generate(), + ); + let requester = if gateway { requester } else { owner.clone() }; + let scope = Scope { + channel: Uuid::new_v4(), + thread: EventId::all_zeros(), + agent: agent.public_key(), + }; + let now = Timestamp::now().as_secs(); + let request = attach::request(&requester, scope, Uuid::new_v4(), now + 600).unwrap(); + let pending = attach::Pending::parse(&request, now).unwrap(); + let approval = attach::approve(&owner, &pending, now + 600).unwrap(); + let tags = std::iter::once(Tag::identifier(scope.channel.to_string())).chain( + [ + owner.public_key(), + agent.public_key(), + requester.public_key(), + ] + .into_iter() + .collect::>() + .into_iter() + .map(|key| Tag::parse(["p", &key.to_hex(), "", "member"]).unwrap()), + ); + let roster = EventBuilder::new(Kind::Custom(39002), "") + .tags(tags) + .sign_with_keys(&relay) + .unwrap(); + let mut events = vec![request.clone(), approval.clone(), roster]; + if gateway { + events.push( + buzz_sdk::gateway::Delegation::grant(requester.public_key(), now + 600, now) + .unwrap() + .sign_with_keys(&owner) + .unwrap(), + ); + events.push( + EventBuilder::new(Kind::Metadata, r#"{"name":"gateway"}"#) + .sign_with_keys(&requester) + .unwrap(), + ); + } + let events = Arc::new(Mutex::new(events)); + let relay_key = relay.public_key(); + let app = Router::new() + .route( + "/", + get(move || async move { Json(json!({"self":relay_key})) }), + ) + .route("/query", post(query)) + .with_state(events.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}", listener.local_addr().unwrap()); + let http = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let directory = tempfile::tempdir().unwrap(); + let executable = directory.path().join("worker"); + std::fs::write( + &executable, + r#"#!/usr/bin/env python3 +import os, sys, signal, json +os.chdir(sys.argv[2]) +with open('marker', 'w') as marker: marker.write('spawned') +def end(*_): + print('{"type":"ended","failed":false}', flush=True) + sys.exit(0) +signal.signal(signal.SIGTERM, end) +sys.stdin.readline() +print('{"type":"ready"}', flush=True) +for line in sys.stdin: + packet = json.loads(line) + if packet.get('type') == 'data': + print(json.dumps({'type':'packet', 'packet':packet}), flush=True) +end() +"#, + ) + .unwrap(); + std::fs::set_permissions(&executable, std::fs::Permissions::from_mode(0o700)).unwrap(); + Self { + config: Config { + relay: String::new(), + keys: agent.clone(), + auth: None, + owner: owner.public_key(), + scope, + workspace: directory.path().to_owned(), + executable, + journal: directory.path().join("journal"), + }, + rest: RestClient { + http: reqwest::Client::new(), + base_url, + keys: agent, + auth_tag_json: None, + }, + events, + request, + approval, + owner, + relay, + requester, + directory, + http, + } + } + fn open_event(&self) -> Event { + attach::encrypt( + &self.requester, + self.config.scope.agent, + self.config.scope, + &Frame { + grant: self.approval.id, + seq: 1, + ack: 0, + packet: Packet::Open { rows: 37, cols: 91 }, + }, + ) + .unwrap() + } + fn journal(&self) -> Journal { + Journal::open( + self.config.journal.clone(), + self.config.scope, + self.config.owner, + ) + .unwrap() + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.http.abort(); + } +} + +async fn relay_receipt(message: &str) -> (NostrWsConnection, tokio::task::JoinHandle<()>) { + relay_receipt_after(message, Duration::ZERO).await +} + +async fn relay_receipt_after( + message: &str, + delay: Duration, +) -> (NostrWsConnection, tokio::task::JoinHandle<()>) { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let message = message.to_owned(); + let task = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + while let Some(Ok(Message::Text(text))) = ws.next().await { + let value: Value = serde_json::from_str(&text).unwrap(); + if value[0] == "EVENT" { + tokio::time::sleep(delay).await; + ws.send(Message::Text( + json!(["OK", value[1]["id"], true, message]) + .to_string() + .into(), + )) + .await + .unwrap(); + } + } + }); + (NostrWsConnection::connect(&url).await.unwrap(), task) +} + +#[tokio::test] +async fn delayed_claim_receipt_cannot_start_a_worker_after_admission_expiry() { + let mut f = Fixture::new(false).await; + let pending = attach::Pending::parse(&f.request, Timestamp::now().as_secs()).unwrap(); + let approval = attach::approve(&f.owner, &pending, Timestamp::now().as_secs() + 3).unwrap(); + f.events + .lock() + .unwrap() + .retain(|event| event.id != f.approval.id); + f.events.lock().unwrap().push(approval.clone()); + f.approval = approval; + let journal = f.journal(); + let (mut connection, task) = + relay_receipt_after(attach::CLAIM_OK_MESSAGE, Duration::from_secs(4)).await; + let result = open( + &mut connection, + &f.config, + &f.rest, + &journal, + &f.open_event(), + &mut Rate::default(), + Instant::now(), + ) + .await; + let mut active = result.ok(); + finish(&f.config, &journal, &mut active).await.unwrap(); + assert!( + !f.directory.path().join("marker").exists(), + "expired admission spawned a worker before its lease check" + ); + let record = journal.records().unwrap().remove(0); + assert!( + record.claimed && record.ended.is_some(), + "expired owned claim was not durably ended" + ); + task.abort(); +} + +#[tokio::test] +async fn failed_audit_write_cannot_downgrade_unconfirmed_startup_containment() { + let f = Fixture::new(false).await; + std::fs::write( + &f.config.executable, + r#"#!/usr/bin/env python3 +import os, sys +os.chdir(sys.argv[2]) +os.rename('journal', 'retained-journal') +sys.exit(2) +"#, + ) + .unwrap(); + let journal = f.journal(); + let (mut connection, task) = relay_receipt(attach::CLAIM_OK_MESSAGE).await; + let result = open( + &mut connection, + &f.config, + &f.rest, + &journal, + &f.open_event(), + &mut Rate::default(), + Instant::now(), + ) + .await; + let error = match result { + Err(error) => error, + Ok(mut session) => { + session.pty.close().await.unwrap(); + panic!("uncontained worker unexpectedly started") + } + }; + assert!( + error.is::(), + "journal failure hid fatal startup containment: {error}" + ); + assert!( + f.directory + .path() + .join("retained-journal") + .join(format!("{}.json", f.approval.id)) + .exists(), + "durable owned acceptance was lost" + ); + task.abort(); +} + +#[tokio::test] +async fn open_requires_the_atomic_first_claim_receipt_before_worker_start() { + for message in [ + "saved", + "duplicate: already have this event", + attach::CLAIM_OK_MESSAGE, + ] { + let f = Fixture::new(false).await; + let journal = f.journal(); + let (mut connection, task) = relay_receipt(message).await; + let result = open( + &mut connection, + &f.config, + &f.rest, + &journal, + &f.open_event(), + &mut Rate::default(), + Instant::now(), + ) + .await; + let accepted = message == attach::CLAIM_OK_MESSAGE; + let did_open = result.is_ok(); + let mut active = result.ok(); + finish(&f.config, &journal, &mut active).await.unwrap(); + assert_eq!(did_open, accepted); + assert_eq!( + f.directory.path().join("marker").exists(), + accepted, + "old/generic/duplicate receipt started a worker" + ); + let records = journal.records().unwrap(); + assert_eq!(records.len(), 1); + assert_eq!(records[0].claimed, accepted); + assert_eq!( + records[0].ended.is_some(), + accepted, + "a losing claim falsely ended another replica's operation" + ); + // A restart/replayed Open never reuses this owner approval locally. + let grant = Grant::parse(&f.approval, f.config.scope, Timestamp::now().as_secs()).unwrap(); + assert!(journal.reserve(&f.config.keys, &grant).is_err()); + task.abort(); + } +} + +#[tokio::test] +async fn current_gateway_approval_requires_all_strong_heads_and_posting_membership() { + let f = Fixture::new(true).await; + let check = || { + authority::approve( + &f.rest, + f.config.owner, + f.config.scope, + f.approval.id, + f.requester.public_key(), + ) + }; + let grant = check().await.unwrap(); + let baseline = f.events.lock().unwrap().clone(); + f.events + .lock() + .unwrap() + .iter_mut() + .find(|event| event.id == f.approval.id) + .unwrap() + .content + .push(' '); + assert!( + authority::refresh(&f.rest, f.config.owner, &grant) + .await + .is_err(), + "malformed current approval bypassed refresh validation" + ); + *f.events.lock().unwrap() = baseline.clone(); + for remove in [f.request.id, f.approval.id] { + f.events.lock().unwrap().retain(|event| event.id != remove); + assert!( + check().await.is_err(), + "deleted current request/grant retained shell authority" + ); + *f.events.lock().unwrap() = baseline.clone(); + } + f.events + .lock() + .unwrap() + .retain(|event| event.kind.as_u16() != 30181); + assert!( + check().await.is_err(), + "revoked gateway retained shell authority" + ); + *f.events.lock().unwrap() = baseline.clone(); + let roster = baseline + .iter() + .find(|event| event.kind.as_u16() == 39002) + .unwrap(); + for missing in [ + f.config.owner, + f.config.scope.agent, + f.requester.public_key(), + ] { + let key = missing.to_hex(); + let tags = roster + .tags + .iter() + .filter(|tag| tag.as_slice().get(1) != Some(&key)) + .cloned(); + let removed = EventBuilder::new(roster.kind, "") + .tags(tags) + .sign_with_keys(&f.relay) + .unwrap(); + f.events + .lock() + .unwrap() + .retain(|event| event.kind.as_u16() != 39002); + f.events.lock().unwrap().push(removed); + assert!( + check().await.is_err(), + "loss of posting membership retained terminal authority" + ); + *f.events.lock().unwrap() = baseline.clone(); + } + let tags = roster + .tags + .iter() + .filter(|tag| tag.as_slice().get(1) != Some(&f.requester.public_key().to_hex())) + .cloned(); + // A spoofed roster with the right entries cannot replace relay authority. + let spoofed = EventBuilder::new(roster.kind, "") + .tags(tags) + .sign_with_keys(&f.owner) + .unwrap(); + f.events + .lock() + .unwrap() + .retain(|event| event.kind.as_u16() != 39002); + f.events.lock().unwrap().push(spoofed); + assert!(check().await.is_err()); +} + +#[tokio::test] +async fn journal_recovers_owned_audit_without_reopening_or_ending_a_losing_claim() { + let f = Fixture::new(false).await; + let grant = Grant::parse(&f.approval, f.config.scope, Timestamp::now().as_secs()).unwrap(); + let journal = f.journal(); + assert!( + Journal::open(f.config.journal.clone(), f.config.scope, f.config.owner).is_err(), + "two local servers acquired the same journal" + ); + let record = journal.reserve(&f.config.keys, &grant).unwrap(); + let accepted = record.accepted.id; + drop(journal); + let journal = f.journal(); + journal.recover(&f.config.keys).unwrap(); + assert!(journal.reserve(&f.config.keys, &grant).is_err()); + let mut record = journal.records().unwrap().remove(0); + assert_eq!(record.accepted.id, accepted); + assert!( + record.ended.is_none(), + "unconfirmed receipt falsely ended a different replica's terminal" + ); + record.claimed = true; + record.bytes_in = 12; + record.bytes_out = 34; + journal.save(&record).unwrap(); + journal.recover(&f.config.keys).unwrap(); + let record = journal.records().unwrap().remove(0); + let ended = record.ended.unwrap(); + let body: Value = serde_json::from_str(&ended.content).unwrap(); + assert_eq!(body["bytes_in"], 12); + assert_eq!(body["bytes_out"], 34); + assert_eq!(body["grant"]["id"], json!(grant.id())); + journal.recover(&f.config.keys).unwrap(); + assert_eq!( + journal.records().unwrap()[0].ended.as_ref().unwrap().id, + ended.id, + "audit retries regenerated their signed event" + ); + assert!(journal.reserve(&f.config.keys, &grant).is_err()); +} + +#[test] +fn production_factory_refuses_unbound_dm_non_owner_and_plaintext_transport() { + let mut config = crate::build_mcp_servers_tests::test_config(); + let owner = Keys::generate(); + config.agent_owner = Some(owner.public_key().to_hex()); + config.relay_url = "wss://relay.example.test".into(); + config.respond_to = RespondTo::Owner; + config.bound_session = Some(BoundSession::Thread { + channel_id: Uuid::new_v4(), + thread_root: EventId::all_zeros().to_hex(), + }); + assert!(Config::from_harness(&config).unwrap().is_some()); + config.relay_url = "ws://relay.example.test".into(); + assert!(Config::from_harness(&config).unwrap().is_none()); + config.relay_url = "wss://relay.example.test".into(); + config.respond_to = RespondTo::Anyone; + assert!(Config::from_harness(&config).unwrap().is_none()); + config.respond_to = RespondTo::Owner; + config.bound_session = Some(BoundSession::DirectMessage { + dm_channel_id: Uuid::new_v4(), + }); + assert!(Config::from_harness(&config).unwrap().is_none()); + config.bound_session = None; + assert!(Config::from_harness(&config).unwrap().is_none()); +} + +#[tokio::test] +async fn harness_shutdown_propagates_unconfirmed_containment_and_task_panics() { + let ordinary = tokio::spawn(async { Err(anyhow::anyhow!("transport unavailable")) }); + assert!(crate::sandbox_terminal::stop(Some(ordinary)).await.is_ok()); + let fatal = tokio::spawn(async { Err(client::ContainmentError.into()) }); + let error = crate::sandbox_terminal::stop(Some(fatal)) + .await + .unwrap_err(); + assert!(crate::sandbox_terminal::containment_failure(&error)); + let panic: tokio::task::JoinHandle> = + tokio::spawn(async { panic!("deliberate terminal task panic") }); + assert!(crate::sandbox_terminal::stop(Some(panic)).await.is_err()); +} + +struct EngineRelay { + url: String, + inject: tokio::sync::mpsc::Sender, + events: tokio::sync::mpsc::Receiver, + ready: tokio::sync::oneshot::Receiver<()>, + task: tokio::task::JoinHandle<()>, +} + +async fn engine_relay(agent: PublicKey) -> EngineRelay { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let (inject, mut injected) = tokio::sync::mpsc::channel::(16); + let (events_tx, events) = tokio::sync::mpsc::channel(16); + let (ready_tx, ready) = tokio::sync::oneshot::channel(); + let task = tokio::spawn(async move { + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap(); + ws.send(Message::Text( + json!(["AUTH", "terminal-engine-fixture"]) + .to_string() + .into(), + )) + .await + .unwrap(); + let mut ready = Some(ready_tx); + loop { + tokio::select! { + message=ws.next()=>{ + let Some(Ok(Message::Text(text)))=message else {break}; + let value:Value=serde_json::from_str(&text).unwrap(); + match value[0].as_str() { + Some("AUTH")=>{ + let event:Event=serde_json::from_value(value[1].clone()).unwrap(); + assert_eq!(event.pubkey,agent); + event.verify().unwrap(); + ws.send(Message::Text(json!(["OK",event.id,true,""]).to_string().into())).await.unwrap(); + } + Some("REQ")=>{if let Some(ready)=ready.take(){let _=ready.send(());}}, + Some("EVENT")=>{ + let event:Event=serde_json::from_value(value[1].clone()).unwrap(); + event.verify().unwrap(); + if event.kind.as_u16()==30624 { + let body:Value=serde_json::from_str(&event.content).unwrap(); + let receipt=if body["status"]=="accepted" {attach::CLAIM_OK_MESSAGE}else{"saved"}; + ws.send(Message::Text(json!(["OK",event.id,true,receipt]).to_string().into())).await.unwrap(); + } else {events_tx.send(event).await.unwrap();} + } + _=>{}, + } + } + Some(event)=injected.recv()=>{ + ws.send(Message::Text(json!(["EVENT",SUBSCRIPTION,event]).to_string().into())).await.unwrap(); + } + } + } + }); + EngineRelay { + url, + inject, + events, + ready, + task, + } +} + +#[tokio::test] +async fn production_engine_delivers_a_full_sdk_data_packet_through_worker_ipc() { + let f = Fixture::new(false).await; + let mut relay = engine_relay(f.config.scope.agent).await; + let mut config = f.config.clone(); + config.relay = relay.url.clone(); + let rest = f.rest.clone(); + let (shutdown_tx, shutdown) = watch::channel(()); + let service = tokio::spawn(serve(config, rest, shutdown)); + let bytes = vec![0xff; attach::MAX_DATA_BYTES]; + let journey = timeout(Duration::from_secs(5), async { + (&mut relay.ready).await.unwrap(); + relay.inject.send(f.open_event()).await.unwrap(); + loop { + let event = relay.events.recv().await.unwrap(); + let frame = attach::decrypt( + &f.requester, + f.config.scope.agent, + f.config.scope, + f.approval.id, + &event, + Timestamp::now().as_secs(), + ) + .unwrap(); + if frame.ack >= 1 { + break; + } + } + let data = attach::encrypt( + &f.requester, + f.config.scope.agent, + f.config.scope, + &Frame { + grant: f.approval.id, + seq: 2, + ack: 0, + packet: Packet::data(&bytes).unwrap(), + }, + ) + .unwrap(); + assert!( + data.content.len() > 32 * 1024, + "fixture did not cross the old encrypted-content guard" + ); + relay.inject.send(data).await.unwrap(); + loop { + let event = relay.events.recv().await.unwrap(); + let frame = attach::decrypt( + &f.requester, + f.config.scope.agent, + f.config.scope, + f.approval.id, + &event, + Timestamp::now().as_secs(), + ) + .unwrap(); + if matches!(frame.packet, Packet::Data { .. }) { + assert_eq!(frame.packet.bytes().unwrap(), bytes); + break; + } + } + }) + .await; + // Always confirm cleanup before asserting, including with the old guard + // restored by a mutation that deliberately prevents the packet's delivery. + shutdown_tx.send(()).unwrap(); + let stopped = timeout(Duration::from_secs(5), service).await; + relay.task.abort(); + assert!(stopped.unwrap().unwrap().is_ok()); + assert!( + journey.is_ok(), + "maximum SDK packet never reached/returned from the worker" + ); + let record = f.journal().records().unwrap().remove(0); + assert_eq!(record.bytes_in, attach::MAX_DATA_BYTES as u64); + assert_eq!(record.bytes_out, attach::MAX_DATA_BYTES as u64); +} + +#[test] +fn production_admission_rejects_oversize_before_crypto_without_spending_rate_budget() { + let (owner, agent) = (Keys::generate(), Keys::generate()); + let scope = Scope { + channel: Uuid::new_v4(), + thread: EventId::all_zeros(), + agent: agent.public_key(), + }; + let event = attach::encrypt( + &owner, + scope.agent, + scope, + &Frame { + grant: EventId::all_zeros(), + seq: u64::MAX, + ack: u64::MAX, + packet: Packet::data(&vec![0xff; attach::MAX_DATA_BYTES]).unwrap(), + }, + ) + .unwrap(); + assert!(event.content.len() > 32 * 1024); + assert!(event.content.len() <= buzz_core::observer::NIP44_MAX_CONTENT_LEN); + let mut oversized = event.clone(); + oversized.content = "z".repeat(buzz_core::observer::NIP44_MAX_CONTENT_LEN + 1); + let mut budget = Rate::default(); + // This deliberately has an invalid signature: the production gate only + // inspects size/admission and cannot invoke signature/decryption work. + assert!(!frame_admitted(&oversized, &mut budget, 0)); + for _ in 0..attach::MAX_FRAMES_PER_SECOND { + assert!(frame_admitted(&event, &mut budget, 0)); + } + assert!(!frame_admitted(&event, &mut budget, 0)); +} diff --git a/crates/buzz-cli/src/commands/sandbox.rs b/crates/buzz-cli/src/commands/sandbox.rs index 89ca3b36467..486db8bc2e2 100644 --- a/crates/buzz-cli/src/commands/sandbox.rs +++ b/crates/buzz-cli/src/commands/sandbox.rs @@ -21,6 +21,12 @@ mod tests; /// Owner approvals and interactive sandbox terminals. #[derive(Subcommand)] pub enum SandboxCmd { + /// Validate a local captured bundle before any Git import (internal worker seam) + #[command(hide = true)] + InspectBundle { + /// Captured bundle file + file: std::path::PathBuf, + }, /// Attach a local TTY; omitted targets are selected from BUZZ_PASEO_AGENTS Attach { /// Channel UUID containing the sandbox thread @@ -56,6 +62,17 @@ pub enum SandboxCmd { }, } +/// Print bounded local preflight metadata, without reading credentials or connecting. +pub(crate) fn inspect_bundle_cmd(file: &std::path::Path) -> Result<(), CliError> { + let summary = crate::sandbox_archive::inspect_bundle_file(file).map_err(failure)?; + let encoded = serde_json::to_string(&summary).map_err(failure)?; + if encoded.len() > 16 * 1024 * 1024 { + return Err(failure("bundle metadata exceeded budget")); + } + println!("{encoded}"); + Ok(()) +} + fn failure(message: impl ToString) -> CliError { CliError::Other(message.to_string()) } @@ -168,6 +185,7 @@ pub async fn dispatch(command: SandboxCmd, client: &BuzzClient) -> Result<(), Cl )); } match command { + SandboxCmd::InspectBundle { file } => inspect_bundle_cmd(&file), SandboxCmd::Authorize { request, minutes } => { let event = one(client, json!({"ids":[request],"kinds":[30624],"limit":1})).await?; if event.id != request { diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index 0aad53dccd8..9c92f57c8e9 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -4,6 +4,8 @@ mod commands; mod error; mod help_tree; mod links; +/// Safe local snapshot archive inspection and restoration. +pub mod sandbox_archive; mod validate; use clap::{CommandFactory, FromArgMatches, Parser, Subcommand, ValueEnum}; @@ -2303,6 +2305,10 @@ async fn run(cli: Cli) -> Result<(), CliError> { }; } + if let Cmd::Sandbox(commands::sandbox::SandboxCmd::InspectBundle { ref file }) = cli.command { + return commands::sandbox::inspect_bundle_cmd(file); + } + // Auth: private key is required for all relay operations. // The keypair IS the identity — no tokens, no other auth. let private_key_str = cli.private_key.ok_or_else(|| { diff --git a/crates/buzz-cli/src/sandbox_archive.rs b/crates/buzz-cli/src/sandbox_archive.rs new file mode 100644 index 00000000000..d21e45b77eb --- /dev/null +++ b/crates/buzz-cli/src/sandbox_archive.rs @@ -0,0 +1,99 @@ +//! Local-only bounded archive preflight. No identity, relay or Git process is used. +use std::fs::OpenOptions; +use std::io::Read; +#[cfg(unix)] +use std::os::unix::fs::OpenOptionsExt; +use std::path::Path; + +use buzz_sdk::sandbox_attach::snapshot::{ + archive::inspect_bundle, MAX_BLOB_BYTES, MAX_SOURCE_BYTES, +}; +use serde::{Deserialize, Serialize}; + +/// Local archive refusal; file contents and scanner diagnostics are never included. +#[derive(Debug, thiserror::Error)] +pub enum ArchiveError { + /// Filesystem access failed. + #[error("archive file access failed: {0}")] + Io(#[from] std::io::Error), + /// A hard file/type or metadata bound was exceeded. + #[error("{0}")] + Invalid(&'static str), + /// The SDK rejected malformed or oversized bundle contents. + #[error("{0}")] + Bundle(#[from] buzz_sdk::SdkError), +} + +/// Bounded admission metadata for the exact captured bundle. +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub struct BundleSummary { + /// Local helper format version. + pub version: u8, + /// Number of packed objects, checked before import. + pub objects: u32, + /// Sum of reconstructed object bytes, including delta targets. + pub expanded_bytes: u64, + /// Object-ID length in bytes (20 or 32). + pub hash_bytes: usize, + /// Object ID/ref name pairs; callers account for ref/index/block costs. + pub refs: Vec<(String, String)>, +} + +/// Inspect a regular file without allocating from unchecked file metadata. +/// The bounded descriptor read also catches growth after the initial metadata check. +pub fn inspect_bundle_file(path: &Path) -> Result { + if !std::fs::symlink_metadata(path)?.is_file() { + return Err(ArchiveError::Invalid("bundle must be a regular file")); + } + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(unix)] + options.custom_flags(nix::libc::O_NONBLOCK | nix::libc::O_NOFOLLOW); + let file = options.open(path)?; + let metadata = file.metadata()?; + if !metadata.is_file() || metadata.len() > MAX_BLOB_BYTES as u64 { + return Err(ArchiveError::Invalid( + "bundle exceeds regular-file byte budget", + )); + } + let mut bytes = Vec::new(); + file.take(MAX_BLOB_BYTES as u64 + 1) + .read_to_end(&mut bytes)?; + if bytes.len() > MAX_BLOB_BYTES { + return Err(ArchiveError::Invalid("bundle grew beyond byte budget")); + } + let inventory = inspect_bundle(&bytes, MAX_SOURCE_BYTES)?; + Ok(BundleSummary { + version: 1, + objects: inventory.objects, + expanded_bytes: inventory.expanded_bytes, + hash_bytes: inventory.hash_bytes, + refs: inventory.refs, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn inspector_refuses_oversized_files_before_reading_and_nonregular_inputs() { + let temp = tempfile::tempdir().unwrap(); + let path = temp.path().join("sparse.bundle"); + std::fs::File::create(&path) + .unwrap() + .set_len(MAX_BLOB_BYTES as u64 + 1) + .unwrap(); + assert!(matches!( + inspect_bundle_file(&path), + Err(ArchiveError::Invalid(_)) + )); + assert!(inspect_bundle_file(temp.path()).is_err()); + #[cfg(unix)] + { + let link = temp.path().join("link"); + std::os::unix::fs::symlink(&path, &link).unwrap(); + assert!(inspect_bundle_file(&link).is_err()); + } + } +} diff --git a/docs/sandbox-attach.md b/docs/sandbox-attach.md new file mode 100644 index 00000000000..96179844693 --- /dev/null +++ b/docs/sandbox-attach.md @@ -0,0 +1,74 @@ +# Owner-approved sandbox terminals + +The ACP harness accepts an interactive terminal through the existing community relay. +The service requires Linux, `--respond-to owner`, an immutable channel-thread binding, a known owner, and WSS. +Unbound agents, direct-message workspaces, other author modes, and plaintext transport never subscribe to terminal requests. +Rebuild the agent image and CLI after the dependent layers land. +See the [CLI runbook](sandbox-attach-cli.md) for request, approval, attachment, and revocation commands. + +Every attachment requires a separate owner-signed approval for its requester, agent, channel, thread, and session. +A global gateway grant permits requesting approval; it cannot authorize a shell or approve another gateway. +Terminal and snapshot approvals use distinct types and modes. +Terminal approval grants the sandbox user's shell permissions, including workspace changes and access to credentials available inside that sandbox. + +```mermaid +sequenceDiagram + participant Requester + participant Owner + participant Relay + participant ACP + participant Worker + Requester->>Relay: Signed pending request + Owner->>Relay: Signed session approval + Requester->>ACP: Encrypted Open through relay + ACP->>Relay: Strong current authority reads + ACP->>Relay: Unique durable acceptance + Relay-->>ACP: Atomic first-claim receipt + ACP->>Worker: Start isolated PTY worker + Worker-->>ACP: Ready + ACP-->>Requester: Encrypted Open acknowledgement +``` + +The server checks current pending request and approval heads, their signatures, immutable scope, posting membership, and live gateway delegation. +Missing, replaced, revoked, expired, malformed, or unavailable authority denies access. +The owner, agent, and requester must remain posting members in the relay-signed roster. +Per-session approval does not confer relay or channel membership. + +Authority refresh runs in a separate task every five seconds, with a five-second lookup deadline. +Results apply only to the same active approval generation. +The authority lease starts before its lookup; admission latency cannot extend it. +An absolute ten-second lease also bounds blocked relay and worker writes, capped by approval expiry. +Admission must retain the full five-second readiness budget before spawning a worker. +The server stops terminal I/O when that lease ends and immediately requests contained worker teardown. +Confirmed cleanup has a separate fifteen-second bound. +Slow authority queries do not block terminal output, input, or cancellation. + +The [atomic claim contract](sandbox-attachment-claims.md) makes each approval single-use across overlapping harnesses. +Generic success, duplicate receipts, mismatched IDs, and ambiguous acknowledgements never start a worker. +The private local journal reserves an approval before publishing its exact acceptance event. +After restart, a consumed approval cannot reopen; unfinished owned audits are recovered for retry. +An unconfirmed or losing claim never publishes an end record for another replica's operation. +The journal caps records at 256 and each record at 16 KiB. + +One terminal runs per bound sandbox. +Frames use recipient-encrypted NIP-44 payloads, four-frame windows, 16 KiB data packets, and a 64-frame-per-second budget. +Encrypted content uses the shared NIP-44 envelope bound of 87,472 bytes before cryptographic work. +The decoded data limit remains 16 KiB; padding and base64 make a full SDK packet exceed 32 KiB on the wire. +Reliable empty heartbeats run every five seconds without resetting idle expiry. +Five failed one-second retries close the transport; idle peer disappearance can therefore take about eleven seconds to detect. +Actual input, resize, or output resets the five-minute idle deadline. +The server drains acknowledged final output before reliably closing the transport. +It bounds that final transport drain to six seconds. + +The [contained worker](sandbox-terminal-worker.md) owns a real PTY and an isolated Linux subreaper process tree. +The harness propagates unconfirmed cleanup, worker exit 2, or a worker-task panic as a fatal containment error. +Fatal containment errors terminate the harness before checkpoint scans. +Ordinary transport failures close the PTY, retain metadata retries, and retry with bounded backoff. +After five failures, terminal service stops; restart the sandbox to restore it. +Ordinary model work can continue when worker containment was confirmed. + +Stored acceptance/end events and local retry records contain approval metadata and byte counters only. +Keystrokes and file bytes pass through encrypted ephemeral frames and local worker pipes. +The service does not persist terminal output or include it in diagnostics. +Failed audit publication remains durable for an idle retry or the next startup. +This feature adds no tunnel, broker, public exec endpoint, or substrate exec permission. diff --git a/docs/sandbox-snapshot-producer.md b/docs/sandbox-snapshot-producer.md index 6bc79a90fe9..0581ba948a8 100644 --- a/docs/sandbox-snapshot-producer.md +++ b/docs/sandbox-snapshot-producer.md @@ -36,6 +36,9 @@ Shallow clones retain their explicit local boundaries, without unshallow or netw Unborn HEADs are refused. Before packing, the helper inventories source objects and byte/count budgets. +After capture, the required local `buzz sandbox inspect-bundle` helper validates the actual bundle with the SDK before any Git import, independently of credentials or relay configuration. +The producer reserves the captured expansion, object/ref/index/block costs and aggregate counts, then verifies the neutral inventory matches before writing raw review bodies. +A source commit between preflight and capture therefore cannot inherit stale scratch admission; missing or malformed inspector output fails closed. It imports the bundle into a neutral Git repository, verifies its object graph, and scans that exact history with immutable gitleaks rules. It also scans the copied working files, manifest, ref names, and every archived raw object, including merge-only blobs and annotated-tag messages. Global/system configuration, replacement refs, grafts, lazy fetch, hooks, fsmonitor, repository ignore files, and `GITLEAKS_*` overrides cannot weaken the scan. diff --git a/voltage/images/buzz-agent/snapshot.py b/voltage/images/buzz-agent/snapshot.py index 9c50c81ecd7..855ba3b2ba9 100644 --- a/voltage/images/buzz-agent/snapshot.py +++ b/voltage/images/buzz-agent/snapshot.py @@ -49,7 +49,7 @@ def run(*args, cwd=None, timeout=60, env=None, input=None, output=None, if args and args[0] == 'git': args = (args[0], '-c', 'protocol.allow=never', '-c', 'core.hooksPath=/dev/null', '-c', 'core.fsmonitor=false', '-c', 'pack.threads=1', - '-c', 'pack.windowMemory=16m', *args[1:]) + '-c', 'pack.windowMemory=16m', '-c', 'core.logAllRefUpdates=false', *args[1:]) expires = time.monotonic() + remaining(timeout) with ExitStack() as stack: sink = stack.enter_context(open(output, 'xb')) if output is not None else None @@ -141,6 +141,30 @@ def inventory(repo, cp): return lines +def inspect_captured(bundle, cp): + # No fallback: source preflight cannot authorize costs of a different bundle + # captured after the source commits. The native SDK validates the exact pack. + result = json.loads(run('/usr/local/bin/buzz', 'sandbox', 'inspect-bundle', + str(bundle), limit=16 * 1024 * 1024)) + if (not isinstance(result, dict) + or set(result) != {'version', 'objects', 'expanded_bytes', 'hash_bytes', 'refs'} + or type(result['version']) is not int or result['version'] != 1 + or type(result['objects']) is not int or not 0 < result['objects'] <= MAX_OBJECTS + or type(result['expanded_bytes']) is not int or not 0 <= result['expanded_bytes'] <= cp.MAX_BYTES + or type(result['hash_bytes']) is not int or result['hash_bytes'] not in (20, 32) + or not isinstance(result['refs'], list) or not 0 < len(result['refs']) <= cp.MAX_FILES): + raise ValueError('Snapshot captured-bundle metadata is invalid') + parents = set() + for ref in result['refs']: + if (not isinstance(ref, list) or len(ref) != 2 or not all(isinstance(v, str) for v in ref) + or len(ref[0]) != result['hash_bytes'] * 2 + or any(c not in '0123456789abcdef' for c in ref[0]) + or not (ref[1] == 'HEAD' or ref[1].startswith('refs/'))): + raise ValueError('Snapshot captured reference is invalid') + parents.update(path(ref[1]).parents) + return result, len(parents) + + class ObjectStream: """Split one bounded cat-file batch without materializing object bodies.""" def __init__(self, lines, directory): @@ -362,7 +386,8 @@ def pack(workspace, output, work_bytes=512 * 1024 * 1024): # Inventory before packing is conservative: even unreachable source # objects count, so no large source object is decoded before a bound. source_objects = inventory(repo, cp) - expanded += sum(int(line.rsplit(' ', 1)[1]) for line in source_objects) + source_expanded = sum(int(line.rsplit(' ', 1)[1]) for line in source_objects) + expanded += source_expanded objects_count += len(source_objects) if expanded > cp.MAX_BYTES or objects_count > MAX_OBJECTS: raise Budget('Snapshot aggregate Git expansion exceeds budget') @@ -373,8 +398,8 @@ def pack(workspace, output, work_bytes=512 * 1024 * 1024): target = stage / 'repositories' / str(index) (target / 'files').mkdir(parents=True) bundle = target / 'history.bundle' - work.reserve(sum(int(line.rsplit(' ', 1)[1]) for line in source_objects) - + len(source_objects) * (work.block + 64)) + source_cost = source_expanded + len(source_objects) * (work.block + 64) + work.reserve(source_cost) allowance = min(MAX_BLOB, cp.MAX_BYTES - total, (work.limit - work.used) // 2) if allowance <= 0: raise Budget('Snapshot bundle capacity exceeds budget') @@ -384,9 +409,21 @@ def pack(workspace, output, work_bytes=512 * 1024 * 1024): work.reserve(2 * bundle_size) if total > cp.MAX_BYTES or bundle_size > MAX_BLOB: raise Budget('Snapshot source exceeds budget') + packed, ref_directories = inspect_captured(bundle, cp) + expanded += max(0, packed['expanded_bytes'] - source_expanded) + objects_count += max(0, packed['objects'] - len(source_objects)) + refs_count += len(packed['refs']) + if expanded > cp.MAX_BYTES or objects_count > MAX_OBJECTS or refs_count > cp.MAX_FILES: + raise Budget('Snapshot captured aggregate inventory exceeds budget') + actual_cost = (packed['expanded_bytes'] + packed['objects'] * (work.block + 64) + + (len(packed['refs']) + ref_directories) * work.block) + work.reserve(max(0, actual_cost - source_cost)) neutral = Path(temp) / f'history-{index}' cp.restore_history(bundle, record, neutral) objects = inventory(neutral, cp) + actual_expanded = sum(int(line.rsplit(' ', 1)[1]) for line in objects) + if actual_expanded != packed['expanded_bytes'] or len(objects) != packed['objects']: + raise ValueError('Snapshot neutral inventory differs from the captured bundle') object_review = review / str(index) object_review.mkdir() writer = ObjectStream(objects, object_review) @@ -405,9 +442,6 @@ def pack(workspace, output, work_bytes=512 * 1024 * 1024): stop = value.find(b'\n\n') if stop < 0: raise Budget('Snapshot bundle header exceeds budget') - refs_count += sum(1 for line in value[:stop].splitlines() if not line.startswith((b'#', b'@', b'-'))) - if refs_count > cp.MAX_FILES: - raise Budget('Snapshot reference count exceeds budget') (object_review / 'references').write_bytes(value[:stop]) names = cp.git(repo, 'ls-files', '-z', '--cached', '--others', '--exclude-standard') for raw in sorted(set(names.split(b'\0')) - {b''}): diff --git a/voltage/images/buzz-agent/test-snapshot.py b/voltage/images/buzz-agent/test-snapshot.py index d5ecfdf14f0..6417333d832 100644 --- a/voltage/images/buzz-agent/test-snapshot.py +++ b/voltage/images/buzz-agent/test-snapshot.py @@ -124,6 +124,54 @@ def expect_refusal(workspace,output,category=None,**options): assert package.extractfile('snapshot/repositories/1/files/inner.txt').read()==b'inner\n' print('PASS: real outer plus untracked nested Git repository have separate complete bundles') + # Real commits between source preflight and capture must be charged BEFORE import. + for succeeds,budget in ((False,20*1024*1024),(True,128*1024*1024)): + case=root/('racing-control' if succeeds else 'racing-refusal');repo=repository(case) + (repo/'safe').write_text('safe\n');git(repo,'add','.');git(repo,'commit','-m','initial') + cp=snapshot.checkpoint();original_git=cp.git;original_restore=cp.restore_history + seen={'changed':False,'imported':False} + def racing(directory,*args,**options): + if not seen['changed'] and args==('rev-parse','--verify','HEAD'): + seen['changed']=True + with (repo/'late-history').open('wb') as stream: stream.truncate(8*1024*1024) + git(repo,'add','late-history');git(repo,'commit','-m','concurrent source') + (repo/'late-history').unlink() + return original_git(directory,*args,**options) + def restoring(*args,**options): + seen['imported']=True + return original_restore(*args,**options) + cp.git=racing;cp.restore_history=restoring + factory=snapshot.checkpoint;snapshot.checkpoint=lambda:cp + output=case/'snapshot.tar.gz' + try: + if succeeds: + snapshot.pack(repo.parent,output,work_bytes=budget) + assert output.exists() and seen['imported'] + else: + try: + snapshot.pack(repo.parent,output,work_bytes=budget) + raise AssertionError('captured history escaped work accounting') + except snapshot.Budget: + assert not seen['imported'] and not output.exists() + assert seen['changed'] + finally: snapshot.checkpoint=factory + print('PASS: concurrent source growth refused before import at20MiB and accepted at128MiB control') + + # The packaged native inspector is required; invalid metadata cannot fall back. + case=root/'inspector-required';repo=repository(case) + (repo/'safe').write_text('safe\n');git(repo,'add','.');git(repo,'commit','-m','initial') + original_run=snapshot.run + for missing in (False,True): + def invalid_inspector(*args,**options): + if args[0]=='/usr/local/bin/buzz': + if missing: raise FileNotFoundError('required inspector is unavailable') + return b'{}' + return original_run(*args,**options) + snapshot.run=invalid_inspector + try: expect_refusal(repo.parent,case/'refused.tar.gz') + finally: snapshot.run=original_run + print('PASS: missing or invalid native inspector has no source-inventory fallback') + # No current or historical credential may be hidden by repo/global ignores. secret='sk-ant-oat01-'+'Z'*90 for historical in (False,True):