diff --git a/crates/bashkit/Cargo.toml b/crates/bashkit/Cargo.toml index 2a0037f09..a3a4b133a 100644 --- a/crates/bashkit/Cargo.toml +++ b/crates/bashkit/Cargo.toml @@ -450,3 +450,8 @@ wasm-bindgen-futures = { version = "0.4", optional = true } # not flag the vendored code. See knowledge/runtimes/coreutils-args-port.md. [lints.rust] unexpected_cfgs = { level = "warn", check-cfg = ['cfg(feature, values("i18n-decimal"))'] } + +# The process feature is native test infrastructure only; keep WASM builds +# on the workspace Tokio feature set without changing versions or the lock. +[target.'cfg(not(target_arch = "wasm32"))'.dev-dependencies] +tokio = { workspace = true, features = ["process"] } diff --git a/crates/bashkit/src/fs/proc_sub_fds.rs b/crates/bashkit/src/fs/proc_sub_fds.rs index 1aa6a0ca7..191759a69 100644 --- a/crates/bashkit/src/fs/proc_sub_fds.rs +++ b/crates/bashkit/src/fs/proc_sub_fds.rs @@ -121,6 +121,15 @@ impl ProcSubFs { self.table().iter().any(|(n, _)| *n == fd) } + /// Buffered read size for admission; pipe stat keeps its POSIX zero size. + pub(crate) fn input_len(&self, path: &Path) -> Option { + match self.slot(path) { + Slot::Open(_, ProcSubData::Input(data)) => Some(data.len()), + Slot::Open(_, ProcSubData::Output(_)) => Some(0), + Slot::Vfs | Slot::Closed => None, + } + } + pub(crate) fn open(&self, fd: i32, data: ProcSubData) { self.table().push((fd, data)); } diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index efe621a6f..0c7276845 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -9426,7 +9426,7 @@ impl Interpreter { } // Handle input redirections first - let stdin = match self + let stdin_setup = match self .process_input_redirections( stdin, &command.redirects, @@ -9441,6 +9441,10 @@ impl Interpreter { } Err(e) => return Err(e), }; + // Keep the materialized stdin's byte/text storage admitted + // for as long as the command retains it. + let _stdin_storage = stdin_setup.stdin_lease; + let stdin = stdin_setup.stdin; // `read -u FD` on a readable fd (`exec 3 return Err(e), }; + // Keep stdin's byte/text storage admitted through the body. + let _stdin_storage = stdin_setup.stdin_lease; + let stdin = stdin_setup.stdin; let prev_pipeline_stdin = if stdin.is_some() { let prev = self.pipeline_stdin.take(); self.pipeline_stdin = stdin; @@ -13020,10 +13027,12 @@ impl Interpreter { .await }; match read { - Ok(content) => { - if let Some(content) = content { - // Account for the VFS-owned input while it and the - // decoded substitution text are simultaneously live. + Ok(setup) => { + // Retain admitted input storage through substitution. + let _stdin_storage = setup.stdin_lease; + if let Some(content) = setup.stdin { + // Cover the temporary byte copy made by the + // command-substitution text boundary. let _content_lease = self.execution_budget.lease_bytes(content.len())?; stdout.try_push_str(&content.command_substitution_text())?; diff --git a/crates/bashkit/src/interpreter/redirection.rs b/crates/bashkit/src/interpreter/redirection.rs index 248276e4f..a28d2e2d3 100644 --- a/crates/bashkit/src/interpreter/redirection.rs +++ b/crates/bashkit/src/interpreter/redirection.rs @@ -56,6 +56,42 @@ fn redirect_target_may_split(redirect: &Redirect) -> bool { }) } +// Read admission, any excess backend capacity, and cached text travel together. +type InputStorage = [Option; 3]; +type StdinSlot = Option<(crate::StreamData, InputStorage)>; + +/// Assign the stdin slot, dropping any previous buffer and its clone lease +/// together, including fresh materializations as well as copies. +fn set_stdin_slot(slot: &mut StdinSlot, value: Option, lease: InputStorage) { + *slot = value.map(|data| (data, lease)); +} + +/// Result of input-redirection processing: the stdin buffer plus the +/// live-bytes leases for both representations. +/// Callers hold `stdin_lease` for as long as they retain `stdin`. +pub(crate) struct StdinSetup { + pub stdin: Option, + pub stdin_lease: InputStorage, +} + +/// Numbered-fd table with replace semantics: re-redirecting an fd drops the +/// superseded buffer — and its live-bytes lease — instead of piling clones. +/// Aggregate input consumption stays cumulative. +type OwnFdEntry = (i32, crate::StreamData, InputStorage); + +fn replace_own_fd( + own_fds: &mut Vec, + fd: i32, + data: crate::StreamData, + lease: InputStorage, +) { + if let Some(slot) = own_fds.iter_mut().find(|(existing, _, _)| *existing == fd) { + *slot = (fd, data, lease); + } else { + own_fds.push((fd, data, lease)); + } +} + /// `N<= 3: a readable fd for the command's body. fn high_fd_here(redirect: &Redirect) -> bool { matches!( @@ -229,6 +265,94 @@ impl Interpreter { self.diag(format!("{path}: {reason}\n")) } + /// Admit a backend read before loading and retain its capacity through decoding. + // THREAT[TM-DOS-103]: metadata and device ceilings precede VFS allocation. + async fn read_input_stream( + &self, + path: &Path, + create: bool, + ) -> Result<(crate::StreamData, InputStorage)> { + let budget = &self.execution_budget; + // Open substitution links describe a pipe, not a VFS symlink target. + // Its owner supplies the buffered size before read_file copies it. + let buffered = self.proc_subs.input_len(path); + let source = if buffered.is_some() { + path.to_path_buf() + } else { + match budget.run(crate::fs::canonicalize(&*self.fs, path)).await? { + Ok(source) => source, + Err(crate::Error::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => { + // Mounted backends can serve a path without synthetic + // parent entries. Its stat/read below decide existence. + path.to_path_buf() + } + Err(error) => return Err(error), + } + }; + // Preserve the configured VFS's following policy for actual I/O; + // canonical resolution above only identifies bounded device reads. + let size = match budget.run(self.fs.stat(path)).await? { + Ok(metadata) => metadata.size, + Err(crate::Error::Io(error)) + if create && error.kind() == std::io::ErrorKind::NotFound => + { + 0 + } + Err(error) => return Err(error), + }; + let reported = usize::try_from(size) + .map_err(|_| LimitExceeded::Memory("input redirection size overflow".into()))?; + let device = match source.to_str() { + // InMemoryFs devices report zero size but materialize these bounded reads. + Some("/dev/zero") => 1 << 20, + Some("/dev/random" | "/dev/urandom") => 8192, + _ => 0, + }; + let size = reported.max(device).max(buffered.unwrap_or(0)); + budget.consume_input(size)?; + budget.consume_work(1 + size.div_ceil(1024) as u64)?; + let raw = budget.lease_bytes(size)?; + if create { + // An empty append may load an existing lazy file, so admission + // must precede opening as well as the subsequent read. + budget.run(self.fs.append_file(path, b"")).await??; + } + let bytes = budget.run(self.fs.read_file(path)).await??; + budget.consume_input(bytes.len().saturating_sub(size))?; + let extra = budget.lease_bytes(bytes.capacity().saturating_sub(size))?; + let (stream, text) = crate::StreamData::from_admitted_bytes(bytes, budget)?; + Ok((stream, [Some(raw), Some(extra), text])) + } + + fn input_stream_from_text( + &self, + content: String, + newline: bool, + ) -> Result<(crate::StreamData, InputStorage)> { + let budget = &self.execution_budget; + let mut text = crate::limits::BudgetedString::try_from_string(content, Some(budget))?; + if newline { + text.try_push('\n')?; + } + budget.consume_input( + text.len() + .checked_mul(2) + .ok_or_else(|| LimitExceeded::Memory("input redirection size overflow".into()))?, + )?; + let (stream, [bytes, text]) = crate::StreamData::from_budgeted_string(text, Some(budget))?; + Ok((stream, [bytes, text, None])) + } + + fn clone_input_stream( + &self, + content: &crate::StreamData, + ) -> Result<(crate::StreamData, InputStorage)> { + self.execution_budget + .consume_input(content.retained_len())?; + let (stream, [bytes, text]) = content.try_clone_budgeted(&self.execution_budget)?; + Ok((stream, [bytes, text, None])) + } + /// Process input redirections (< file, <<< string). `demand` is how /// much the command reads, for stdin taken from a coproc's pipe. pub(super) async fn process_input_redirections( @@ -237,10 +361,13 @@ impl Interpreter { redirects: &[Redirect], demand: StdinDemand, high_fd_stdin: HighFdStdin, - ) -> Result> { - let mut stdin = existing_stdin; + ) -> crate::error::Result { + // Stdin slot: a budgeted clone's live lease travels with the buffer + // so replacing stdin drops both together; the surviving lease is + // handed to the caller with stdin. + let mut stdin: StdinSlot = existing_stdin.map(|data| (data, [None, None, None])); // Readable fds this command's own redirects open (`4< = Vec::new(); + let mut own_fds: Vec = Vec::new(); for redirect in redirects { // Fd N >= 1 for an input-side redirect (`3f`). @@ -250,19 +377,20 @@ impl Interpreter { // `<>`: open (creating) the file read-write. let target_path = self.expand_word(&redirect.target).await?; let path = self.resolve_path(&target_path); - let content = if is_dev_null(&path) { - crate::StreamData::new() + let (content, storage) = if is_dev_null(&path) { + (crate::StreamData::new(), [None, None, None]) } else if !self.shell_features.has_file_redirects() { return Err(crate::error::Error::Execution( self.diag(format!("{}: filesystem redirection disabled", target_path)), )); } else { - let opened = match self.fs.append_file(&path, b"").await { - Ok(()) => self.fs.read_file(&path).await, - Err(e) => Err(e), - }; + // Keep backend-admission state off recursive command frames. + let opened = Box::pin(self.read_input_stream(&path, true)).await; match opened { - Ok(content) => content.into(), + Ok(content) => content, + Err(e @ (crate::Error::ResourceLimit(_) | crate::Error::Cancelled)) => { + return Err(e); + } Err(e) => { return Err(crate::error::Error::CommandFailure( self.diag(format!("{target_path}: {}\n", io_error_reason(&e))), @@ -271,8 +399,10 @@ impl Interpreter { } }; match high_fd { - Some(fd) => own_fds.push((fd, content)), - None => stdin = Some(content), + Some(fd) => { + replace_own_fd(&mut own_fds, fd, content, storage); + } + None => set_stdin_slot(&mut stdin, Some(content), storage), } } RedirectKind::Input => { @@ -290,18 +420,18 @@ impl Interpreter { if self.fd_alias(&path) == Some(0) { // `< /dev/stdin` re-reads the current stdin: no-op. } else if is_dev_null(&path) { - stdin = Some(crate::StreamData::new()); // EOF + set_stdin_slot( + &mut stdin, + Some(crate::StreamData::new()), + [None, None, None], + ); // EOF } else if !self.shell_features.has_file_redirects() { return Err(crate::error::Error::Execution( self.diag(format!("{}: filesystem redirection disabled", target_path)), )); } else { - match self.fs.read_file(&path).await { - Ok(content) => { - let content: crate::StreamData = content.into(); - if let Some(fd) = high_fd { - own_fds.push((fd, content.clone())); - } + match Box::pin(self.read_input_stream(&path, false)).await { + Ok((content, storage)) => { // WTF: on a compound `3 fd == u, (Some(_), HighFdStdin::Never) => false, }; - if feeds { - stdin = Some(content); + if let Some(fd) = high_fd { + if feeds { + let (copy, copy_storage) = + self.clone_input_stream(&content)?; + replace_own_fd(&mut own_fds, fd, copy, copy_storage); + set_stdin_slot(&mut stdin, Some(content), storage); + } else { + replace_own_fd(&mut own_fds, fd, content, storage); + } + } else { + set_stdin_slot(&mut stdin, Some(content), storage); } } + Err(e @ (crate::Error::ResourceLimit(_) | crate::Error::Cancelled)) => { + return Err(e); + } Err(e) => { return Err(crate::error::Error::CommandFailure( self.redirect_error(&target_path, &io_error_reason(&e)), @@ -328,20 +470,24 @@ impl Interpreter { RedirectKind::HereString => { // <<< string - use the target as stdin content let content = self.expand_word(&redirect.target).await?; - let content: crate::StreamData = format!("{}\n", content).into(); + let (content, storage) = self.input_stream_from_text(content, true)?; match high_fd { - Some(fd) => own_fds.push((fd, content)), - None => stdin = Some(content), + Some(fd) => { + replace_own_fd(&mut own_fds, fd, content, storage); + } + None => set_stdin_slot(&mut stdin, Some(content), storage), } } RedirectKind::HereDoc | RedirectKind::HereDocStrip => { // << EOF / <<- EOF - use the heredoc content as stdin; // `3< own_fds.push((fd, content)), - None => stdin = Some(content), + Some(fd) => { + replace_own_fd(&mut own_fds, fd, content, storage); + } + None => set_stdin_slot(&mut stdin, Some(content), storage), } } RedirectKind::DupInput => { @@ -360,12 +506,15 @@ impl Interpreter { let Ok(src) = target.parse::() else { continue; }; - // `<&4` after this command's own `4< own_fds.push((fd, content)), - None => stdin = Some(content), + Some(fd) => replace_own_fd(&mut own_fds, fd, content, lease), + None => set_stdin_slot(&mut stdin, Some(content), lease), } continue; } @@ -391,11 +540,16 @@ impl Interpreter { text.push('\n'); } } - stdin = Some(text.into()); + let (content, storage) = self.input_stream_from_text(text, false)?; + set_stdin_slot(&mut stdin, Some(content), storage); } Some(coproc::InputFd::Pipe(end)) => { let pipe = Arc::clone(&end.0); - stdin = Some(Box::pin(coproc::read_coproc_input(pipe, demand)).await); + set_stdin_slot( + &mut stdin, + Some(Box::pin(coproc::read_coproc_input(pipe, demand)).await), + [None, None, None], + ); } // `<&5` with fd 5 never opened (or closed): bash // reports it and does not run the command. @@ -416,7 +570,12 @@ impl Interpreter { } } - Ok(stdin) + let mut stdin_lease = [None, None, None]; + let stdin = stdin.map(|(data, lease)| { + stdin_lease = lease; + data + }); + Ok(StdinSetup { stdin, stdin_lease }) } /// Apply output redirections to command output @@ -1560,3 +1719,248 @@ fn redirect_reads(kind: RedirectKind) -> bool { | RedirectKind::OutputBoth ) } + +#[cfg(test)] +mod input_admission_tests { + use super::*; + + fn ordinary_input(kind: RedirectKind) -> Redirect { + Redirect { + kind, + fd: None, + fd_var: None, + target: Word::quoted_literal("/a"), + heredoc_delim: None, + } + } + + #[tokio::test] + async fn input_admission_mount_refuses_lazy_before_loading() { + let loads = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let observed = Arc::clone(&loads); + let mounted = Arc::new(crate::InMemoryFs::new()); + mounted.add_lazy_file( + "/a", + 2, + 0o644, + Arc::new(move || { + observed.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + b"ab".to_vec() + }), + ); + let fs = crate::MountableFs::new(Arc::new(crate::InMemoryFs::new())); + fs.mount("/mnt/input", mounted).unwrap(); + let mut bash = crate::Bash::builder() + .fs(Arc::new(fs)) + .limits(crate::ExecutionLimits::new().max_live_intermediate_bytes(1)) + .build(); + bash.interpreter.begin_execution_budget(); + let result = bash + .interpreter + .read_input_stream(Path::new("/mnt/input/a"), false) + .await; + assert!(matches!(result, Err(crate::Error::ResourceLimit(_)))); + assert_eq!(loads.load(std::sync::atomic::Ordering::Relaxed), 0); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + + #[tokio::test] + async fn input_admission_tiny_buffered_pipe() { + for (cap, succeeds) in [(1, false), (4, true)] { + let mut bash = crate::Bash::builder() + .limits(crate::ExecutionLimits::new().max_live_intermediate_bytes(cap)) + .build(); + bash.interpreter.begin_execution_budget(); + bash.interpreter + .proc_subs + .open(63, crate::fs::ProcSubData::Input(Arc::from(&b"ab"[..]))); + let path = Path::new("/dev/fd/63"); + assert_eq!(bash.interpreter.fs.stat(path).await.unwrap().size, 0); + let result = bash.interpreter.read_input_stream(path, false).await; + if succeeds { + let (stream, storage) = result.unwrap(); + assert_eq!(stream.as_bytes(), b"ab"); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 4); + drop((stream, storage)); + } else { + assert!(matches!(result, Err(crate::Error::ResourceLimit(_)))); + } + assert!(bash.interpreter.proc_subs.is_open(63)); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + } + + #[tokio::test] + async fn input_admission_readwrite_refuses_lazy_load_before_open() { + let loads = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let observed = Arc::clone(&loads); + let fs = Arc::new(crate::InMemoryFs::new()); + fs.add_lazy_file( + "/a", + 2, + 0o644, + Arc::new(move || { + observed.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + b"ab".to_vec() + }), + ); + let mut bash = crate::Bash::builder() + .fs(fs) + .limits(crate::ExecutionLimits::new().max_aggregate_input_bytes(1)) + .build(); + bash.interpreter.begin_execution_budget(); + let redirect = ordinary_input(RedirectKind::ReadWrite); + let result = bash + .interpreter + .process_input_redirections(None, &[redirect], StdinDemand::All, HighFdStdin::Never) + .await; + assert!(matches!(result, Err(crate::Error::ResourceLimit(_)))); + assert_eq!(loads.load(std::sync::atomic::Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn input_admission_preserves_cancellation_during_tiny_read() { + for kind in [RedirectKind::Input, RedirectKind::ReadWrite] { + let fs = Arc::new(crate::InMemoryFs::new()); + let mut bash = crate::Bash::builder().fs(fs.clone()).build(); + bash.interpreter.begin_execution_budget(); + let token = bash.interpreter.cancellation_token(); + fs.add_lazy_file( + "/a", + 2, + 0o644, + Arc::new(move || { + token.store(true, std::sync::atomic::Ordering::Relaxed); + b"ab".to_vec() + }), + ); + let redirect = ordinary_input(kind); + let result = bash + .interpreter + .process_input_redirections(None, &[redirect], StdinDemand::All, HighFdStdin::Never) + .await; + assert!( + matches!(result, Err(crate::Error::Cancelled)), + "cancellation must not become a command failure" + ); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + } + + #[tokio::test] + async fn input_admission_refuses_lazy_read_before_loading() { + for limits in [ + crate::ExecutionLimits::new().max_aggregate_input_bytes(1), + crate::ExecutionLimits::new().max_live_intermediate_bytes(1), + ] { + let loads = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let observed = Arc::clone(&loads); + let fs = Arc::new(crate::InMemoryFs::new()); + fs.add_lazy_file( + "/a", + 2, + 0o644, + Arc::new(move || { + observed.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + b"ab".to_vec() + }), + ); + let mut bash = crate::Bash::builder().fs(fs).limits(limits).build(); + bash.interpreter.begin_execution_budget(); + let result = bash + .interpreter + .read_input_stream(Path::new("/a"), false) + .await; + assert!(matches!(result, Err(crate::Error::ResourceLimit(_)))); + assert_eq!(loads.load(std::sync::atomic::Ordering::Relaxed), 0); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + } + + #[tokio::test] + async fn input_admission_tiny_read_exact_boundary_and_failure_refund() { + for (cap, succeeds) in [(3, false), (4, true)] { + let mut bash = crate::Bash::builder() + .limits(crate::ExecutionLimits::new().max_live_intermediate_bytes(cap)) + .build(); + bash.fs().write_file(Path::new("/a"), b"ab").await.unwrap(); + bash.interpreter.begin_execution_budget(); + let result = bash + .interpreter + .read_input_stream(Path::new("/a"), false) + .await; + if succeeds { + let (stream, storage) = result.unwrap(); + assert_eq!(stream.as_bytes(), b"ab"); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 4); + drop((stream, storage)); + } else { + assert!(matches!(result, Err(crate::Error::ResourceLimit(_)))); + } + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + } + + #[tokio::test] + async fn input_admission_preserves_missing_file_error() { + let mut bash = crate::Bash::new(); + bash.interpreter.begin_execution_budget(); + assert!( + bash.interpreter + .read_input_stream(Path::new("/absent"), false) + .await + .is_err() + ); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + + #[test] + fn input_admission_text_and_clone_hold_both_representations() { + for newline in [false, true] { + let mut bash = crate::Bash::new(); + bash.interpreter.begin_execution_budget(); + let (stream, storage) = bash + .interpreter + .input_stream_from_text("a".into(), newline) + .unwrap(); + let expected = if newline { + b"a\n".as_slice() + } else { + b"a".as_slice() + }; + assert_eq!(stream.as_bytes(), expected); + let original_live = bash.interpreter.execution_budget.live_bytes_for_test(); + assert!(original_live >= (2 * expected.len()) as u64); + let (copy, copy_storage) = bash.interpreter.clone_input_stream(&stream).unwrap(); + assert_eq!(copy, stream); + assert_eq!( + bash.interpreter.execution_budget.live_bytes_for_test(), + original_live + (2 * expected.len()) as u64 + ); + drop((copy, copy_storage)); + assert_eq!( + bash.interpreter.execution_budget.live_bytes_for_test(), + original_live + ); + drop((stream, storage)); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + } + + #[tokio::test] + async fn input_admission_retains_fresh_stdin_storage() { + let mut bash = crate::Bash::new(); + bash.fs().write_file(Path::new("/a"), b"ab").await.unwrap(); + bash.interpreter.begin_execution_budget(); + let redirect = ordinary_input(RedirectKind::Input); + let setup = bash + .interpreter + .process_input_redirections(None, &[redirect], StdinDemand::All, HighFdStdin::Never) + .await + .unwrap(); + assert_eq!(setup.stdin.as_ref().unwrap().as_bytes(), b"ab"); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 4); + drop(setup); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } +} diff --git a/crates/bashkit/src/stream.rs b/crates/bashkit/src/stream.rs index 928a3929d..1af845df8 100644 --- a/crates/bashkit/src/stream.rs +++ b/crates/bashkit/src/stream.rs @@ -17,6 +17,41 @@ pub struct StreamData { } impl StreamData { + /// Decode a VFS buffer whose byte capacity is already leased by its caller. + pub(crate) fn from_admitted_bytes( + bytes: Vec, + budget: &crate::limits::ExecutionBudget, + ) -> Result<(Self, Option), crate::limits::LimitExceeded> + { + let mut text = crate::limits::BudgetedString::new(Some(budget))?; + for chunk in bytes.utf8_chunks() { + budget.consume_work(1 + chunk.valid().len().div_ceil(1024) as u64)?; + budget.consume_input(chunk.valid().len())?; + text.try_push_str(chunk.valid())?; + if !chunk.invalid().is_empty() { + budget.consume_input('\u{fffd}'.len_utf8())?; + text.try_push('\u{fffd}')?; + } + } + let (text, storage) = text.into_parts(); + Ok((Self { bytes, text }, storage)) + } + + /// Copy both representations through the existing capacity-owning builders. + pub(crate) fn try_clone_budgeted( + &self, + budget: &crate::limits::ExecutionBudget, + ) -> Result<(Self, StreamStorage), crate::limits::LimitExceeded> { + budget.consume_work(1 + self.retained_len().div_ceil(1024) as u64)?; + let mut bytes = crate::limits::BudgetedBytes::new(Some(budget))?; + let mut text = crate::limits::BudgetedString::new(Some(budget))?; + bytes.try_extend_from_slice(&self.bytes)?; + text.try_push_str(&self.text)?; + let (bytes, byte_storage) = bytes.into_parts(); + let (text, text_storage) = text.into_parts(); + Ok((Self { bytes, text }, [byte_storage, text_storage])) + } + /// Convert admitted text to the byte/text stream without an uncharged copy. pub(crate) fn from_budgeted_string( text: crate::limits::BudgetedString, @@ -73,6 +108,12 @@ impl StreamData { pub fn len(&self) -> usize { self.bytes.len() } + + /// Live bytes retained by this buffer: the raw bytes plus the decoded + /// text kept alongside them. + pub(crate) fn retained_len(&self) -> usize { + self.bytes.len().saturating_add(self.text.len()) + } pub fn is_empty(&self) -> bool { self.bytes.is_empty() } @@ -220,6 +261,43 @@ impl serde::Serialize for StreamData { mod tests { use super::StreamData; + #[test] + fn admitted_bytes_preserve_lossy_text_and_lease() { + let budget = crate::limits::ExecutionBudget::new( + &crate::ExecutionLimits::new(), + std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + ); + let bytes = b"a\xff\0".to_vec(); + let raw = budget.lease_bytes(bytes.capacity()).unwrap(); + let (stream, text) = StreamData::from_admitted_bytes(bytes, &budget).unwrap(); + assert_eq!(stream.as_bytes(), b"a\xff\0"); + assert_eq!(stream.text_lossy(), "a\u{fffd}\0"); + assert!(budget.live_bytes_for_test() >= 8); + drop((stream, raw, text)); + assert_eq!(budget.live_bytes_for_test(), 0); + } + + #[test] + fn budgeted_clone_exact_boundary_and_refund() { + for (limit, succeeds) in [(3, false), (4, true)] { + let budget = crate::limits::ExecutionBudget::new( + &crate::ExecutionLimits::new().max_live_intermediate_bytes(limit), + std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), + ); + let source = StreamData::from("ab"); + let result = source.try_clone_budgeted(&budget); + if succeeds { + let (copy, storage) = result.unwrap(); + assert_eq!(copy, source); + assert_eq!(budget.live_bytes_for_test(), 4); + drop((copy, storage)); + } else { + assert!(result.is_err()); + } + assert_eq!(budget.live_bytes_for_test(), 0); + } + } + #[test] fn budgeted_string_stream_admits_both_representations() { for limit in [7, 8] { diff --git a/crates/bashkit/tests/integration/input_redirect_budget_tests.rs b/crates/bashkit/tests/integration/input_redirect_budget_tests.rs new file mode 100644 index 000000000..200843527 --- /dev/null +++ b/crates/bashkit/tests/integration/input_redirect_budget_tests.rs @@ -0,0 +1,1825 @@ +//! Repeated numbered input redirects must charge the input budget: +//! +//! Each `N/new").await.unwrap(); + assert_eq!(result.exit_code, 0); + assert!( + bash.fs() + .read_file(Path::new("/new")) + .await + .unwrap() + .is_empty() + ); + bash.fs().write_file(Path::new("/a"), b"ab").await.unwrap(); + let result = bash.exec("cat <>/a").await.unwrap(); + assert_eq!(result.exit_code, 0); + assert_eq!(result.stdout.as_bytes(), b"ab"); + assert_eq!(bash.fs().read_file(Path::new("/a")).await.unwrap(), b"ab"); +} + +#[tokio::test] +async fn input_redirection_missing_file_remains_shell_failure() { + let mut bash = Bash::new(); + let result = bash.exec("cat /mnt/input/a", "ab"), + (": <>/mnt/input/new", ""), + ("cat = + std::sync::Arc::new(bashkit::OverlayFs::new(lower.clone())); + let backend = if readonly { + std::sync::Arc::new(bashkit::ReadOnlyFs::new(overlay)) + as std::sync::Arc + } else { + overlay + }; + let fs = bashkit::MountableFs::new(std::sync::Arc::new(bashkit::InMemoryFs::new())); + fs.mount("/nested/data", backend).unwrap(); + let mut bash = Bash::builder().fs(std::sync::Arc::new(fs)).build(); + let read = bash.exec("cat /nested/data/a").await.unwrap(); + assert_eq!(opened.exit_code, if readonly { 1 } else { 0 }); + if readonly { + assert!(opened.stderr.contains("filesystem is read-only")); + assert!(opened.stdout.is_empty()); + } else { + assert_eq!(opened.stdout, "ab"); + } + assert_eq!( + bashkit::FileSystem::read_file(&*lower, Path::new("/a")) + .await + .unwrap(), + b"ab" + ); + let missing = bash + .exec("cat /nested/data/a").await.unwrap(); + assert_eq!(opened.exit_code, 1); + assert!(!opened.stderr.is_empty()); + assert_eq!( + bash.fs() + .read_file(Path::new("/nested/data/a")) + .await + .unwrap(), + b"ab" + ); +} + +#[tokio::test] +async fn input_redirection_respects_symlink_configuration() { + for follows in [false, true] { + let mut bash = Bash::builder().follow_symlinks(follows).build(); + bash.fs().write_file(Path::new("/a"), b"ab").await.unwrap(); + bash.fs() + .symlink(Path::new("/a"), Path::new("/link")) + .await + .unwrap(); + for script in ["cat /link"] { + let result = bash.exec(script).await.unwrap(); + assert_eq!(result.exit_code, if follows { 0 } else { 1 }); + assert_eq!( + result.stdout.as_bytes(), + if follows { &b"ab"[..] } else { &[] } + ); + } + } +} + +#[tokio::test] +async fn input_redirection_tiny_process_substitution() { + let mut bash = Bash::new(); + for script in [ + "cat < <(printf 'x\\ny\\n')", + "mapfile -t a < <(printf 'x\\ny\\n'); printf '%s\\n' \"${a[@]}\"", + ] { + let result = bash.exec(script).await.unwrap(); + assert_eq!(result.exit_code, 0); + assert_eq!(result.stdout.as_bytes(), b"x\ny\n"); + assert!(result.stderr.is_empty(), "{}", result.stderr); + } +} + +#[tokio::test] +async fn input_redirection_tiny_here_inputs() { + let mut bash = Bash::new(); + for script in ["cat <</big").await; + assert!( + matches!( + result, + Err(Error::ResourceLimit(LimitExceeded::ExecutionBudget( + ExecutionBudgetExceeded::InputBytes { .. } + ))) + ), + "ReadWrite retention should trip InputBytes, got {result:?}" + ); +} + +#[tokio::test] +/// The session stays usable after a budget error: unrelated commands run. +async fn session_recovers_after_budget_error() { + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_aggregate_input_bytes(100)) + .build(); + bash.fs() + .write_file(Path::new("/big"), &[b'x'; 32]) + .await + .unwrap(); + let tripped = bash.exec(": 3 String { + format!( + "ulimit -s 4096 || exit 99; \ + ulimit -t 10 || exit 99; \ + ulimit -f {fileblk} || exit 99; \ + mem_kb=unsupported; mem_bytes=unsupported; file_bytes=unsupported; \ + if [ \"$(uname)\" = Linux ] && [ -r /proc/self/limits ]; then \ + file_bytes=$(awk '/Max file size/ {{print $4}}' /proc/self/limits); \ + fi; \ + echo \"ir-limits: stack=$(ulimit -s) cpu=$(ulimit -t) \ + mem_kb=$mem_kb mem_bytes=$mem_bytes \ + fileblk=$(ulimit -Sf) file_bytes=$file_bytes\" >&2; \ + exec \"$@\"", + fileblk = IR_FILE_BLOCKS, + ) +} + +/// Parsed `ir-limits:` receipt. Raw fields are the shell's own units; +/// `*_bytes` are kernel byte counts (Linux) or `None` (explicitly +/// unsupported elsewhere). +#[derive(Debug)] +struct IrLimitsNote { + stack_raw: String, + cpu_raw: String, + mem_kb_raw: String, + mem_bytes: Option, + fileblk_raw: String, + file_bytes: Option, +} + +fn parse_ir_limits_note(line: &str) -> Result { + let body = line + .strip_prefix("ir-limits:") + .ok_or_else(|| format!("missing ir-limits prefix: {line:?}"))?; + let get = |key: &str| -> Result { + body.split_whitespace() + .find_map(|field| field.strip_prefix(&format!("{key}="))) + .map(str::to_string) + .ok_or_else(|| format!("missing field {key} in {line:?}")) + }; + let parse_bytes = |key: &str, raw: &str| -> Result, String> { + if raw == "unsupported" { + return Ok(None); + } + raw.parse::() + .map(Some) + .map_err(|_| format!("bad byte count {key}={raw:?} in {line:?}")) + }; + let mem_bytes_raw = get("mem_bytes")?; + let file_bytes_raw = get("file_bytes")?; + Ok(IrLimitsNote { + stack_raw: get("stack")?, + cpu_raw: get("cpu")?, + mem_kb_raw: get("mem_kb")?, + mem_bytes: parse_bytes("mem_bytes", &mem_bytes_raw)?, + fileblk_raw: get("fileblk")?, + file_bytes: parse_bytes("file_bytes", &file_bytes_raw)?, + }) +} + +/// Classified artifact-sink outcome: absence is normal off-coverage, a +/// present artifact must fit the explicit bound, anything else fails loudly +/// (never confused with absence). +#[derive(Debug)] +enum ArtifactCheck { + Absent, + Within(u64), + Over(u64), +} + +fn check_artifact_bytes(profile: &std::path::Path, bound: u64) -> std::io::Result { + match std::fs::metadata(profile) { + Ok(meta) => Ok(if meta.len() < bound { + ArtifactCheck::Within(meta.len()) + } else { + ArtifactCheck::Over(meta.len()) + }), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(ArtifactCheck::Absent), + Err(e) => Err(e), + } +} + +/// Owned temp working dir. Ownership is established FIRST (the guard object +/// exists before any fallible use); every operation runs through the guard. +/// `cleanup` disarms only after removal succeeds, so a failed cleanup keeps +/// the guard armed and fails loudly instead of leaking silently. `Drop` is a +/// backstop for unwind paths only and documents its silent best-effort. +#[derive(Debug)] +struct TempWorkdir { + path: Option, +} + +impl TempWorkdir { + fn create(tag: &str) -> std::io::Result { + let path = std::env::temp_dir().join(format!("bashkit-{tag}-{}", std::process::id())); + // Exclusive creation: existence is NOT proof this invocation created + // the directory. A collision rejects WITHOUT deleting so preexisting + // contents (possibly another owner's) are always preserved. + match std::fs::create_dir(&path) { + Ok(()) => Ok(Self { path: Some(path) }), + Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => Err(std::io::Error::new( + std::io::ErrorKind::AlreadyExists, + format!("work dir collision, refusing to adopt: {}", path.display()), + )), + Err(e) => Err(e), + } + } + fn path(&self) -> &std::path::Path { + self.path.as_ref().expect("workdir already cleaned") + } + + fn cleanup(mut self) -> std::io::Result<()> { + let path = self.path.as_ref().expect("workdir already cleaned"); + std::fs::remove_dir_all(path)?; + self.path = None; + Ok(()) + } +} + +impl Drop for TempWorkdir { + // Backstop only: the explicit `cleanup` path above is the truthful + // receipt. A Drop-time failure has nowhere to report, hence best-effort; + // normal paths never reach here armed. + fn drop(&mut self) { + if let Some(path) = self.path.take() { + let _ = std::fs::remove_dir_all(path); + } + } +} + +/// Parse `ps -o rss=` output (KiB on Linux and macOS) to a byte-exact... +/// KiB count. Anything else fails loudly: a policing loop must never parse +/// garbage as a small number. +fn parse_ps_rss(out: &str) -> Result { + out.trim() + .parse::() + .map_err(|_| format!("unparseable ps rss output: {out:?}")) +} + +/// Attested file-size unit factor: the byteproof must equal blocks times one +/// of the historically attested shell units (512B POSIX, 1024B measured +/// bash). Anything else fails closed instead of converting silently. +fn file_size_unit_factor(blocks: u64, bytes: u64) -> Result { + for unit in [512u64, 1024u64] { + if blocks.saturating_mul(unit) == bytes { + return Ok(unit); + } + } + Err(format!( + "file byteproof {bytes} matches neither 512B nor 1024B units for {blocks} blocks" + )) +} + +/// Outcome of one opt-in bounded input-shape run: the reviewed verdict plus +/// job-specific receipts. Setup faults panic with context (fail-closed); +/// runtime observations are flags for the caller to assert on. +#[cfg(not(target_arch = "wasm32"))] +struct InputShapeRun { + exit_code: Option, + wall_limit_seconds: u64, + stdout: String, + stderr: String, + timed_out: bool, + oversized: bool, + incomplete_readers: bool, + sampler_complete: bool, + lifecycle_error: Option, + read_error: bool, + reaped: bool, + limits: IrLimitsNote, + artifact: ArtifactCheck, + leftovers: Vec, + file_unit: Option, + rss_pid: u32, + rss_samples: u32, + peak_rss_kb: u64, + rss_exceeded: bool, + exited_fast: bool, +} + +/// Sample resident-set size (KiB) of `pid` via `ps` (`rss` is KiB on +/// Linux and macOS). Any failure is loud: blind policing is worse than none, +/// and the caller kills first on doubt. +/// Bounded async drain: keeps at most `cap` bytes, counts the rest without +/// storing. Cancellation-safe per `AsyncReadExt::read` (locked tokio +/// 1.53.2): dropping the future abandons only unread bytes, so a deadline +/// timeout followed by kill+reap stays finite. Never collects unbounded +/// `read_to_end`. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum StreamNotice { + Excess, + ReadError, +} + +async fn ir_drain_capped_async( + pipe: &mut R, + cap: usize, + mut notice: Option>, +) -> (Vec, usize, bool) +where + R: tokio::io::AsyncRead + Unpin, +{ + use tokio::io::AsyncReadExt; + let mut kept = Vec::new(); + let mut total = 0usize; + let mut had_error = false; + let mut chunk = [0u8; 8192]; + loop { + match pipe.read(&mut chunk).await { + Ok(0) => break, + Ok(n) => { + let room = cap.saturating_sub(kept.len()).min(n); + kept.extend_from_slice(&chunk[..room]); + total = total.saturating_add(n); + // One notice per stream; the shared channel has two slots. + // Notify before EOF while continuing bounded capture. + if total > cap + && let Some(sender) = notice.take() + && sender.try_send(StreamNotice::Excess).is_err() + { + had_error = true; + break; + } + } + Err(_) => { + had_error = true; + if let Some(sender) = notice.take() { + // A closed receiver cannot turn a failed read into success. + let _ = sender.try_send(StreamNotice::ReadError); + } + break; + } + } + } + let kept_len = kept.len(); + (kept, total.saturating_sub(kept_len), had_error) +} + +#[derive(Debug)] +enum ReapOutcome { + Reaped(T), + Failed(std::io::Error), + TimedOut, +} + +async fn reap_before( + future: impl Future>, + deadline: tokio::time::Instant, +) -> ReapOutcome { + match tokio::time::timeout_at(deadline, future).await { + Ok(Ok(status)) => ReapOutcome::Reaped(status), + Ok(Err(error)) => ReapOutcome::Failed(error), + Err(_) => ReapOutcome::TimedOut, + } +} + +#[derive(Debug)] +struct SampleError { + message: String, + cleanup_complete: bool, +} + +/// One lifecycle owns wait, captures and the optional RSS sampler. Errors +/// request kill once, then retain all futures through finite completion. +/// Drop remains a best-effort backstop, never a completion receipt. +#[cfg(not(target_arch = "wasm32"))] +struct ChildCompletion { + status: Option, + out: (Vec, usize, bool), + err: (Vec, usize, bool), + reaped: bool, + readers_complete: bool, + sampler_complete: bool, + timed_out: bool, + failure: Option, + rss_samples: u32, + peak_rss_kb: u64, + rss_exceeded: bool, + exited_fast: bool, +} + +#[cfg(not(target_arch = "wasm32"))] +async fn complete_owned_child( + child: &mut OwnedChild, + cap: usize, + deadline: tokio::time::Instant, + sample_pid: Option, + grace: Duration, +) -> ChildCompletion { + let (mut out_pipe, mut err_pipe) = child.take_pipes(); + let missing_pipe = out_pipe.is_none() || err_pipe.is_none(); + let (notice_tx, mut notices) = tokio::sync::mpsc::channel(2); + let err_notice = notice_tx.clone(); + let mut out_fut = std::pin::pin!(async { + match out_pipe.as_mut() { + Some(pipe) => ir_drain_capped_async(pipe, cap, Some(notice_tx)).await, + None => (Vec::new(), 0, true), + } + }); + let mut err_fut = std::pin::pin!(async { + match err_pipe.as_mut() { + Some(pipe) => ir_drain_capped_async(pipe, cap, Some(err_notice)).await, + None => (Vec::new(), 0, true), + } + }); + let mut completion = ChildCompletion { + status: None, + out: (Vec::new(), 0, false), + err: (Vec::new(), 0, false), + reaped: false, + readers_complete: false, + sampler_complete: false, + timed_out: false, + failure: missing_pipe.then(|| "child capture pipe missing".into()), + rss_samples: 0, + peak_rss_kb: 0, + rss_exceeded: false, + exited_fast: false, + }; + let mut stop_requested = missing_pipe; + let mut kill_sent = false; + let mut wait_failed = false; + let mut out_done = false; + let mut err_done = false; + let mut notices_open = true; + let mut sample_fut: Option = None; + let mut ticker = tokio::time::interval_at( + tokio::time::Instant::now() + Duration::from_millis(50), + Duration::from_millis(500), + ); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let finish_deadline = deadline + grace; + let mut sampler_cleanup_complete = true; + loop { + if stop_requested && !kill_sent && completion.status.is_none() { + kill_sent = true; + if let Err(error) = child.start_kill() { + completion + .failure + .get_or_insert_with(|| format!("kill child: {error}")); + } + } + if (completion.status.is_some() || wait_failed) + && out_done + && err_done + && sample_fut.is_none() + { + break; + } + tokio::select! { + biased; + _ = tokio::time::sleep_until(finish_deadline) => { + completion.failure.get_or_insert_with(|| "child completion timed out".into()); + // Send before leaving the owner loop, including a failed wait. + if !kill_sent && completion.status.is_none() + && let Err(error) = child.start_kill() + { + completion.failure.get_or_insert_with(|| format!("kill child: {error}")); + } + break; + } + _ = tokio::time::sleep_until(deadline), + if !stop_requested && completion.status.is_none() => + { + completion.timed_out = true; + stop_requested = true; + } + result = child.wait(), if completion.status.is_none() && !wait_failed => { + match result { + Ok(status) => { + completion.status = Some(status); + completion.reaped = true; + completion.exited_fast = completion.rss_samples == 0; + } + Err(error) => { + completion.failure = Some(format!("wait child: {error}")); + wait_failed = true; + stop_requested = true; + } + } + } + notice = notices.recv(), if notices_open => { + match notice { + Some(notice) => { + completion.failure.get_or_insert_with(|| format!("child stream: {notice:?}")); + stop_requested = true; + } + None => notices_open = false, + } + } + out = &mut out_fut, if !out_done => { + if out.1 > 0 || out.2 { + completion.failure.get_or_insert_with(|| "stdout capture failed".into()); + stop_requested = true; + } + completion.out = out; + out_done = true; + } + err = &mut err_fut, if !err_done => { + if err.1 > 0 || err.2 { + completion.failure.get_or_insert_with(|| "stderr capture failed".into()); + stop_requested = true; + } + completion.err = err; + err_done = true; + } + res = poll_owned_sample(&mut sample_fut), if sample_fut.is_some() => { + sample_fut = None; + match res { + Ok(rss) => { + completion.rss_samples += 1; + completion.peak_rss_kb = completion.peak_rss_kb.max(rss); + if rss > IR_RSS_CAP_KIB { + completion.rss_exceeded = true; + stop_requested = true; + } + } + Err(error) => { + if !error.cleanup_complete { + sampler_cleanup_complete = false; + completion.failure = Some(error.message); + stop_requested = true; + } else if completion.status.is_none() { + match child.try_wait() { + Ok(Some(status)) => { + completion.status = Some(status); + completion.reaped = true; + completion.exited_fast = completion.rss_samples == 0; + } + Ok(None) => { + completion.failure = Some(error.message); + stop_requested = true; + } + Err(poll_error) => { + completion.failure = Some(format!("poll child: {poll_error}; {}", error.message)); + stop_requested = true; + wait_failed = true; + } + } + } + } + } + } + _ = ticker.tick(), + if sample_pid.is_some() && !stop_requested + && completion.status.is_none() && sample_fut.is_none() => + { + sample_fut = Some(Box::pin(sample_child_rss(sample_pid.expect("sampling enabled")))); + } + } + } + completion.readers_complete = out_done && err_done; + match finish_owned_sample(&mut sample_fut, finish_deadline).await { + Ok(Some(rss)) => { + completion.rss_samples += 1; + completion.peak_rss_kb = completion.peak_rss_kb.max(rss); + completion.rss_exceeded |= rss > IR_RSS_CAP_KIB; + } + Ok(None) => {} + Err(error) => { + sampler_cleanup_complete &= error.cleanup_complete; + if !error.cleanup_complete { + completion.failure.get_or_insert(error.message); + } + } + } + completion.sampler_complete = sample_fut.is_none() && sampler_cleanup_complete; + if completion.status.is_none() { + match reap_before( + child.wait(), + tokio::time::Instant::now() + grace.min(Duration::from_secs(5)), + ) + .await + { + ReapOutcome::Reaped(status) => { + completion.status = Some(status); + completion.reaped = true; + } + ReapOutcome::Failed(error) => { + completion + .failure + .get_or_insert_with(|| format!("reap child: {error}")); + } + ReapOutcome::TimedOut => { + completion + .failure + .get_or_insert_with(|| "reap child timed out".into()); + } + } + } + completion +} + +/// RSS sampling uses the same finite owned lifecycle as the main child. +/// Sampling failure and incomplete reap/capture are distinct verdicts. +#[cfg(not(target_arch = "wasm32"))] +async fn sample_child_rss(pid: u32) -> Result { + let ps = Command::new("ps") + .arg("-o") + .arg("rss=") + .arg("-p") + .arg(pid.to_string()) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true) + .spawn() + .map_err(|error| SampleError { + message: format!("spawn ps for rss of {pid}: {error}"), + cleanup_complete: true, // Spawn failed; no child exists. + })?; + let mut ps = OwnedChild::new(ps); + let capture = complete_owned_child( + &mut ps, + 4096, + tokio::time::Instant::now() + Duration::from_secs(5), + None, + Duration::from_secs(2), + ) + .await; + let cleanup_complete = capture.reaped && capture.readers_complete && capture.sampler_complete; + if !cleanup_complete + || capture.timed_out + || capture.failure.is_some() + || !capture.status.is_some_and(|status| status.success()) + { + return Err(SampleError { + message: format!( + "ps rss of {pid} failed: {} (reaped={}, captures={})", + capture.failure.as_deref().unwrap_or("unsuccessful sample"), + capture.reaped, + capture.readers_complete + ), + cleanup_complete, + }); + } + parse_ps_rss(&String::from_utf8_lossy(&capture.out.0)).map_err(|message| SampleError { + message, + cleanup_complete, + }) +} +type SampleFuture = std::pin::Pin>>>; + +async fn poll_owned_sample(slot: &mut Option) -> Result { + // Losing a select branch must not destroy the pending sampler it borrows. + slot.as_mut().expect("armed with sampler").as_mut().await +} + +async fn finish_owned_sample( + slot: &mut Option, + deadline: tokio::time::Instant, +) -> Result, SampleError> { + if slot.is_none() { + return Ok(None); + } + match tokio::time::timeout_at(deadline, poll_owned_sample(slot)).await { + Ok(result) => { + *slot = None; + result.map(Some) + } + Err(_) => Err(SampleError { + message: "owned sampler completion timed out".into(), + cleanup_complete: false, + }), + } +} + +#[tokio::test] +async fn ir_helper_reap_outcomes() { + let deadline = tokio::time::Instant::now() + Duration::from_secs(1); + assert!(matches!( + reap_before(std::future::ready(Ok(7_u8)), deadline).await, + ReapOutcome::Reaped(7) + )); + match reap_before( + std::future::ready(Err::(std::io::Error::other("injected"))), + deadline, + ) + .await + { + ReapOutcome::Failed(error) => assert_eq!(error.kind(), std::io::ErrorKind::Other), + outcome => panic!("failed wait must remain an error: {outcome:?}"), + } + assert!(matches!( + reap_before( + std::future::pending::>(), + tokio::time::Instant::now() + ) + .await, + ReapOutcome::TimedOut + )); +} + +#[tokio::test] +async fn ir_helper_live_excess_notice() { + let (mut writer, mut reader) = tokio::io::duplex(64); + tokio::io::AsyncWriteExt::write_all(&mut writer, &[1_u8; 32]) + .await + .unwrap(); + let (tx, mut rx) = tokio::sync::mpsc::channel(2); + let mut drain = std::pin::pin!(ir_drain_capped_async(&mut reader, 16, Some(tx))); + let notice = tokio::time::timeout(Duration::from_secs(1), async { + tokio::select! { + notice = rx.recv() => notice, + _ = &mut drain => panic!("writer is open; capture must remain pending"), + } + }) + .await + .unwrap(); + assert_eq!(notice, Some(StreamNotice::Excess)); + drop(writer); + let (kept, truncated, error) = drain.await; + assert_eq!(kept, vec![1_u8; 16]); + assert_eq!(truncated, 16); + assert!(!error); +} + +#[tokio::test] +async fn ir_helper_live_error_notice() { + let (tx, mut rx) = tokio::sync::mpsc::channel(2); + let mut reader = FailOnce { failed: false }; + let (notice, (kept, truncated, error)) = + tokio::join!(rx.recv(), ir_drain_capped_async(&mut reader, 16, Some(tx)),); + assert_eq!(notice, Some(StreamNotice::ReadError)); + assert!(kept.is_empty() && truncated == 0 && error); +} + +#[tokio::test] +async fn ir_helper_terminal_sampler_completion() { + let (tx, rx) = tokio::sync::oneshot::channel::(); + let mut slot: Option = Some(Box::pin(async move { + rx.await.map_err(|error| SampleError { + message: error.to_string(), + cleanup_complete: true, + }) + })); + let deadline = tokio::time::Instant::now() + Duration::from_secs(1); + let (result, ()) = tokio::join!(finish_owned_sample(&mut slot, deadline), async { + tx.send(7).unwrap(); + },); + assert_eq!(result.unwrap(), Some(7)); + assert!(slot.is_none()); + slot = Some(Box::pin(std::future::pending())); + let error = finish_owned_sample(&mut slot, tokio::time::Instant::now()) + .await + .unwrap_err(); + assert!(!error.cleanup_complete); + assert_eq!(error.message, "owned sampler completion timed out"); + assert!( + slot.is_some(), + "timeout cannot claim completion or discard the owned future" + ); +} + +#[cfg(not(target_arch = "wasm32"))] +async fn run_input_shape_bounded(script: &str, wall_secs: u64, tag: &str) -> InputShapeRun { + let sink = TempWorkdir::create(&format!("{tag}-sink")).expect("artifact sink setup"); + let work = TempWorkdir::create(&format!("{tag}-work")).expect("producer workdir setup"); + let profile = sink.path().join("child.profraw"); + let cli = bashkit_cli_path(); + let mut cmd = Command::new("bash"); + cmd.arg("-c") + .arg(input_shape_launcher()) + .arg("ir-child") + .arg(&cli) + .arg("-c") + .arg(script) + .env("LLVM_PROFILE_FILE", &profile) + .current_dir(work.path()) + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + // Backstop only: SIGKILL on drop if the explicit kill+reap below + // never runs. Not proof of anything; receipts are. + .kill_on_drop(true); + let mut child = OwnedChild::new(cmd.spawn().expect("spawn bounded child")); + let rss_pid = child.id(); + let completion = complete_owned_child( + &mut child, + PIPE_CAP, + tokio::time::Instant::now() + Duration::from_secs(wall_secs), + Some(rss_pid), + Duration::from_secs(15), + ) + .await; + let (out_kept, out_trunc, out_err) = completion.out; + let (err_kept, err_trunc, err_err) = completion.err; + // Receipt first: the launcher note proves the caps ran; without it the + // run is untrusted regardless of exit code. + let stderr_full = String::from_utf8_lossy(&err_kept).into_owned(); + let (note, rest) = stderr_full + .split_once('\n') + .expect("launcher receipt missing: empty stderr"); + let limits = parse_ir_limits_note(note).unwrap_or_else(|e| panic!("bad launcher receipt: {e}")); + let stderr = rest.to_string(); + #[cfg(target_os = "linux")] + let file_unit = { + let blocks: u64 = limits + .fileblk_raw + .parse() + .expect("fileblk receipt must be numeric"); + let bytes = limits.file_bytes.expect("file byteproof required on Linux"); + Some(file_size_unit_factor(blocks, bytes).unwrap_or_else(|e| panic!("file unit cap: {e}"))) + }; + #[cfg(not(target_os = "linux"))] + let file_unit = None; + let artifact = check_artifact_bytes(&profile, IR_ARTIFACT_BOUND) + .unwrap_or_else(|e| panic!("cannot stat artifact sink: {e:?}")); + // Inventory before cleanup: any producer file fails closed by naming it; + // entry errors are loud, never filtered away. + let mut leftovers = Vec::new(); + for entry in std::fs::read_dir(work.path()) + .unwrap_or_else(|e| panic!("cannot inventory work dir: {e:?}")) + { + match entry { + Ok(e) => leftovers.push(e.file_name().to_string_lossy().into_owned()), + Err(e) => panic!("cannot inventory work dir entry: {e:?}"), + } + } + sink.cleanup().expect("artifact sink cleanup"); + work.cleanup().expect("work dir cleanup"); + InputShapeRun { + exit_code: completion.status.and_then(|s| s.code()), + wall_limit_seconds: wall_secs, + stdout: String::from_utf8_lossy(&out_kept).into_owned(), + stderr, + timed_out: completion.timed_out, + oversized: out_trunc > 0 || err_trunc > 0, + incomplete_readers: !completion.readers_complete, + sampler_complete: completion.sampler_complete, + lifecycle_error: completion.failure, + read_error: out_err || err_err, + reaped: completion.reaped, + limits, + artifact, + leftovers, + file_unit, + rss_pid, + rss_samples: completion.rss_samples, + peak_rss_kb: completion.peak_rss_kb, + rss_exceeded: completion.rss_exceeded, + exited_fast: completion.exited_fast, + } +} + +/// Effective-limits assertions shared by bounded input-shape runs: exact raw +/// round-trips everywhere; kernel byte proofs on Linux, explicit unsupported +/// markers elsewhere (fail-closed on any unit surprise). +/// Effective-bounds assertions shared by bounded input-shape runs: exact raw +/// round-trips everywhere; kernel byte proofs on Linux (including the +/// shell-unit factor for `-f`); explicit unsupported markers elsewhere +/// (fail-closed on any unit surprise); RSS policing receipts (a real pid, +/// no observed exceedance, sampled maximum under the cutoff, samples taken unless the child +/// exited before the first poll). +#[cfg(not(target_arch = "wasm32"))] +fn assert_effective_bounds(r: &InputShapeRun) { + use std::io::Write as _; + + let (artifact_class, artifact_bytes) = match &r.artifact { + ArtifactCheck::Absent => ("absent", None), + ArtifactCheck::Within(bytes) => ("within", Some(*bytes)), + ArtifactCheck::Over(bytes) => ("over", Some(*bytes)), + }; + let receipt = serde_json::json!({ + "producer_exit_code": r.exit_code, + "wall_limit_seconds": r.wall_limit_seconds, + "per_stream_capture_limit_bytes": PIPE_CAP, + "stack_kib": r.limits.stack_raw, + "cpu_seconds": r.limits.cpu_raw, + "address_space_kib": r.limits.mem_kb_raw, + "address_space_bytes": r.limits.mem_bytes, + "address_space_enforced_by_helper": false, + "memory_guard": "sampled-rss", + "rss_limit_kib": IR_RSS_CAP_KIB, + "file_blocks": r.limits.fileblk_raw, + "file_bytes": r.limits.file_bytes, + "file_unit_bytes": r.file_unit, + "artifact_class": artifact_class, + "artifact_bytes": artifact_bytes, + "artifact_limit_bytes": IR_ARTIFACT_BOUND, + "reaped": r.reaped, + "readers_complete": !r.incomplete_readers, + "sampler_complete": r.sampler_complete, + "lifecycle_error": r.lifecycle_error, + "read_error": r.read_error, + "timed_out": r.timed_out, + "oversized": r.oversized, + "leftover_count": r.leftovers.len(), + "rss_samples": r.rss_samples, + "sampled_peak_rss_kib": r.peak_rss_kb, + "rss_exceeded": r.rss_exceeded, + }); + // Direct stderr preserves the measured receipt in successful CI logs too. + writeln!(std::io::stderr().lock(), "input-shape-receipt: {receipt}") + .expect("write bounded input receipt"); + assert!(r.sampler_complete, "owned sampler completion unknown"); + assert!( + r.lifecycle_error.is_none(), + "child lifecycle failed: {:?}", + r.lifecycle_error + ); + let limits = &r.limits; + assert_eq!(limits.stack_raw, "4096", "stack cap round-trip"); + assert_eq!(limits.cpu_raw, "10", "cpu cap round-trip"); + assert_eq!( + limits.mem_kb_raw, "unsupported", + "no AS enforcement by helper" + ); + assert_eq!(limits.mem_bytes, None, "no AS byteproof claimed"); + assert_eq!( + limits.fileblk_raw, + IR_FILE_BLOCKS.to_string(), + "file cap round-trip" + ); + #[cfg(target_os = "linux")] + { + assert!( + r.file_unit.is_some(), + "file unit factor must be attested on Linux" + ); + } + #[cfg(not(target_os = "linux"))] + { + assert_eq!(limits.file_bytes, None, "no file byteproof here"); + assert_eq!(r.file_unit, None, "no file unit attested here"); + } + assert_ne!(r.rss_pid, 0, "RSS policing must attach to a real pid"); + assert!( + !r.rss_exceeded, + "RSS cutoff tripped at {} KiB", + r.peak_rss_kb + ); + assert!( + r.peak_rss_kb < IR_RSS_CAP_KIB, + "sampled peak RSS {} KiB over cutoff", + r.peak_rss_kb + ); + assert!( + r.rss_samples >= 1 || r.exited_fast, + "RSS policing never sampled a live child" + ); +} + +/// Original shape under default limits in the opt-in bounded runner: `:` +/// plus 1024 same-fd `/dev/zero` redirects is denied quickly by the +/// aggregate input budget (each 1MiB read retains ~2MiB against the 100MB +/// default cap, tripping after ~50 redirects) instead of piling ~2GiB of +/// retained clones. Asserts the exact denial: clean exit code 1 naming +/// `aggregate input bytes` (production retention admission, far above the +/// ~7KiB script text), distinguishing budget denial from OOM/signal death +/// (no exit code) and from pre-fix success (exit 0). +/// NOT EXECUTED in this environment (hostile-fixture ban); kept as the +/// blocked-verification target with full harness assertions intact. +#[cfg(all(unix, not(target_arch = "wasm32")))] +#[tokio::test] +async fn original_shape_bounded_child_default_limits() { + let r = + run_input_shape_bounded(&format!(":{}", " 3 {} + Err(e) => panic!("fixture root busy, clean {path:?} manually: {e:?}"), + } + std::fs::write(path.join("precious"), b"precious").expect("plant collision file"); + let err = TempWorkdir::create(tag).expect_err("collision must fail"); + assert_eq!(err.kind(), std::io::ErrorKind::AlreadyExists); + assert_eq!( + std::fs::read(path.join("precious")).expect("preexisting contents preserved"), + b"precious" + ); + std::fs::remove_dir_all(&path).expect("remove owned planted dir"); + assert!(!path.exists()); +} + +#[test] +fn ir_helper_ps_parse_cases() { + assert_eq!(parse_ps_rss(" 1234\n"), Ok(1234)); + assert_eq!(parse_ps_rss("0"), Ok(0)); + assert!(parse_ps_rss("").is_err()); + assert!(parse_ps_rss("abc").is_err()); + assert!(parse_ps_rss("12 34").is_err()); +} + +/// Owned child handle: established immediately after spawn, before any +/// fallible pipe/thread/poll operation. Every later expect runs with this +/// owner alive, so unwind paths always have a defined backstop. Normal +/// paths reap explicitly (see `reap_with_grace` callers) and then drop the +/// guard; Drop kills best-effort (documented backstop, never the primary +/// path) but never waits, so Drop cannot block. +/// Owned child handle: established immediately after spawn, before any +/// fallible pipe/poll operation. Every later expect runs with this owner +/// alive, so unwind paths always have a defined backstop. Normal paths reap +/// explicitly with a finite deadline and then drop the guard; Drop sends +/// SIGKILL best-effort via sync `start_kill` (documented backstop, never the +/// primary path) but never waits, so Drop cannot block. +#[cfg(not(target_arch = "wasm32"))] +struct OwnedChild { + inner: Option, +} + +#[cfg(not(target_arch = "wasm32"))] +impl OwnedChild { + fn new(child: tokio::process::Child) -> Self { + Self { inner: Some(child) } + } + + fn id(&self) -> u32 { + self.inner + .as_ref() + .expect("child already released") + .id() + .expect("child has pid") + } + + fn take_pipes( + &mut self, + ) -> ( + Option, + Option, + ) { + let inner = self.inner.as_mut().expect("child already released"); + (inner.stdout.take(), inner.stderr.take()) + } + + fn try_wait(&mut self) -> std::io::Result> { + self.inner + .as_mut() + .expect("child already released") + .try_wait() + } + + fn start_kill(&mut self) -> std::io::Result<()> { + self.inner + .as_mut() + .expect("child already released") + .start_kill() + } + + async fn wait(&mut self) -> std::io::Result { + self.inner + .as_mut() + .expect("child already released") + .wait() + .await + } +} + +#[cfg(not(target_arch = "wasm32"))] +impl Drop for OwnedChild { + fn drop(&mut self) { + if let Some(child) = self.inner.as_mut() { + let _ = child.start_kill(); + } + } +} + +/// In-memory async drain contract: 100B through a duplex pair with a 16B +/// cap keeps 16, counts 84 truncated, no error. Harmless synthetic fixture: +/// proves the bounded-drain primitive admits before growth without any +/// child, pipe, or resource pressure. +#[tokio::test] +async fn ir_helper_async_drain_bounded() { + let (mut writer, mut reader) = tokio::io::duplex(256); + tokio::io::AsyncWriteExt::write_all(&mut writer, &[9u8; 100]) + .await + .expect("fill duplex"); + drop(writer); + let (kept, trunc, err) = ir_drain_capped_async(&mut reader, 16, None).await; + assert_eq!(kept.len(), 16); + assert_eq!(trunc, 84); + assert!(!err); +} + +/// Injected I/O error contract: a reader failing on first poll reports +/// had_error instead of hanging or misattributing bytes. +struct FailOnce { + failed: bool, +} + +impl tokio::io::AsyncRead for FailOnce { + fn poll_read( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + _buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + if self.failed { + std::task::Poll::Ready(Ok(())) + } else { + self.get_mut().failed = true; + std::task::Poll::Ready(Err(std::io::Error::other("injected"))) + } + } +} + +#[tokio::test] +async fn ir_helper_async_drain_error() { + let mut reader = FailOnce { failed: false }; + let (kept, trunc, err) = ir_drain_capped_async(&mut reader, 16, None).await; + assert!(kept.is_empty()); + assert_eq!(trunc, 0); + assert!(err); +} + +/// Deterministic timer contract: a ready timer completes under deadline; a +/// pending future does not. Proves timeout_at behaves before trusting it +/// for kill/reap/drain bounds. No wall-clock flakiness: margins are 500x. +#[tokio::test] +async fn ir_helper_timeout_contract() { + use std::time::Duration; + let ok = tokio::time::timeout_at( + tokio::time::Instant::now() + Duration::from_secs(5), + tokio::time::sleep(Duration::from_millis(10)), + ) + .await; + assert!(ok.is_ok()); + let timed_out = tokio::time::timeout_at( + tokio::time::Instant::now() + Duration::from_millis(10), + std::future::pending::<()>(), + ) + .await; + assert!(timed_out.is_err()); +} + +/// Scheduling contract: a release-gated sampler, two 32B duplex drains, and +/// a timer must ALL complete concurrently in one select! scope. The release +/// fires only after both drains finish. Poll the wrapper's actual owned-slot +/// helper so losing a selection cannot destroy that release path. 64B total fixture +/// data, 2s outer bound, fully deterministic (no timing margins asserted, +/// only completion). +#[tokio::test] +async fn ir_helper_concurrent_schedule() { + use std::time::Duration; + let (tx, rx) = tokio::sync::oneshot::channel::(); + let (mut w1, mut r1) = tokio::io::duplex(64); + let (mut w2, mut r2) = tokio::io::duplex(64); + tokio::io::AsyncWriteExt::write_all(&mut w1, &[1u8; 32]) + .await + .expect("fill one"); + tokio::io::AsyncWriteExt::write_all(&mut w2, &[2u8; 32]) + .await + .expect("fill two"); + drop((w1, w2)); + let mut sampler: Option = Some(Box::pin(async move { + rx.await.map(u64::from).map_err(|error| SampleError { + message: error.to_string(), + cleanup_complete: true, + }) + })); + let mut d1 = std::pin::pin!(ir_drain_capped_async(&mut r1, 64, None)); + let mut d2 = std::pin::pin!(ir_drain_capped_async(&mut r2, 64, None)); + let mut tx_opt = Some(tx); + let mut got = 0u64; + let mut k1 = Vec::new(); + let mut k2 = Vec::new(); + let mut sampler_done = false; + let mut d1_done = false; + let mut d2_done = false; + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let finished = tokio::time::timeout_at(deadline, async { + loop { + if sampler_done && d1_done && d2_done { + break; + } + tokio::select! { + biased; + v = poll_owned_sample(&mut sampler), if sampler.is_some() => { + got = v.expect("sampler completed"); + sampler = None; + sampler_done = true; + } + _ = tokio::time::sleep(Duration::from_millis(5)) => { + if d1_done && d2_done && let Some(tx) = tx_opt.take() { + let _ = tx.send(7); + } + } + r = &mut d1, if !d1_done => { + let (kept, trunc, err) = r; + k1 = kept; + assert_eq!(trunc, 0); + assert!(!err); + d1_done = true; + } + r = &mut d2, if !d2_done => { + let (kept, trunc, err) = r; + k2 = kept; + assert_eq!(trunc, 0); + assert!(!err); + d2_done = true; + } + } + } + }) + .await; + assert!(finished.is_ok(), "concurrent schedule must finish fast"); + assert_eq!(got, 7); + assert_eq!(k1, vec![1u8; 32]); + assert_eq!(k2, vec![2u8; 32]); +} diff --git a/crates/bashkit/tests/integration/main.rs b/crates/bashkit/tests/integration/main.rs index ddb66bed3..45d70d476 100644 --- a/crates/bashkit/tests/integration/main.rs +++ b/crates/bashkit/tests/integration/main.rs @@ -53,6 +53,7 @@ pub mod curl_data_compat_tests; pub mod custom_builtins_tests; pub mod custom_fs_tests; pub mod function_metadata_security_tests; +pub mod input_redirect_budget_tests; pub mod interactive_builtins_tests; pub mod make_substitution_tests; pub mod memory_growth_security_tests; diff --git a/crates/bashkit/tests/integration/threat_model_tests.rs b/crates/bashkit/tests/integration/threat_model_tests.rs index ba045b87f..a59aa807f 100644 --- a/crates/bashkit/tests/integration/threat_model_tests.rs +++ b/crates/bashkit/tests/integration/threat_model_tests.rs @@ -19,7 +19,7 @@ use std::time::{Duration, Instant}; // 1. RESOURCE EXHAUSTION TESTS // ============================================================================= -mod resource_exhaustion { +pub(crate) mod resource_exhaustion { use super::*; /// Ancestor levels searched for the CLI binary. Five are needed for the @@ -80,7 +80,7 @@ mod resource_exhaustion { /// unexplained exit 127 with the bound assertions failing — both read /// like product failures. Fail on the precondition instead, naming every /// path ruled out. - fn bashkit_cli_path() -> std::path::PathBuf { + pub(crate) fn bashkit_cli_path() -> std::path::PathBuf { let exe = std::env::current_exe().expect("current test executable"); match cli_in_ancestors(&exe) { Ok(cli) => cli, @@ -99,30 +99,30 @@ mod resource_exhaustion { } } - struct BoundedRun { - timed_out: bool, - oversized: bool, + pub(crate) struct BoundedRun { + pub(crate) timed_out: bool, + pub(crate) oversized: bool, /// True when the deadline passed with readers unfinished: a foreign /// descendant retains the pipes. Detection, not reaping: without /// platform process-group APIs (no new dependencies) a foreign /// descendant cannot be reaped portably, so the run fails visibly. - incomplete_readers: bool, + pub(crate) incomplete_readers: bool, /// A pipe reader hit a real I/O error (anything but `Interrupted`, /// which is retried). Capture is then partial by definition, so the /// run fails closed instead of relabeling the error as a clean EOF. - read_error: bool, + pub(crate) read_error: bool, /// OS termination accounted for: the loop observed an exit or /// the bounded reap recovered one. Orthogonal to `exit_code` /// (signaled victims have no code) and to injected observation /// faults (which discard the observation, never the cleanup). - reaped: bool, + pub(crate) reaped: bool, /// First stderr line from the launcher (`harness-limits: ...`), /// proving the CPU/stack caps were applied to this run. Empty when /// the launcher itself refused (fail-closed exit 99). - limits_note: String, - exit_code: Option, - stdout: String, - stderr: String, + pub(crate) limits_note: String, + pub(crate) exit_code: Option, + pub(crate) stdout: String, + pub(crate) stderr: String, } /// Fail-closed launcher limits: 10 CPU seconds, 4 MiB main-thread @@ -186,7 +186,7 @@ mod resource_exhaustion { /// the capture with the bytes kept so far and `read_error` set, never /// relabeled as a clean EOF. Reaching exactly the cap followed by EOF /// is not oversize; only real excess is flagged. - fn drain_capped(reader: &mut R) -> (Vec, bool, bool) { + pub(crate) fn drain_capped(reader: &mut R) -> (Vec, bool, bool) { let mut buf = Vec::new(); let mut chunk = [0u8; 4096]; loop { @@ -935,7 +935,7 @@ mod resource_exhaustion { /// Returns the observed status, or `None` when the victim cannot be /// reaped inside the grace (or polling itself errors): the safe seam /// for the unavailable-reap verdict. - fn reap_with_grace( + pub(crate) fn reap_with_grace( poll: &mut dyn FnMut() -> std::io::Result>, grace: Duration, ) -> Option { @@ -981,7 +981,7 @@ mod resource_exhaustion { // drives the real lifecycle path with exactly one hostile condition. // All flags stay off outside these tests. #[derive(Clone, Default)] - struct LifecycleFaults { + pub(crate) struct LifecycleFaults { /// Child poll fails: the parent must fail closed, never panic. fail_poll: bool, /// Stdout pipe reads fail: the parent verdict must carry @@ -1011,7 +1011,7 @@ mod resource_exhaustion { /// Standard runner: core lifecycle plus the bounded join gate. Reader /// threads that never reported stay detached by contract. - fn run_command_bounded_with( + pub(crate) fn run_command_bounded_with( cmd: Command, limit: Duration, faults: LifecycleFaults, diff --git a/docs/security.md b/docs/security.md index f88c28318..a8e5e07a6 100644 --- a/docs/security.md +++ b/docs/security.md @@ -38,6 +38,18 @@ and deadlines. Refused admission returns a resource-limit error even when output would be discarded by a redirect. Ordinary aliases and append/copy semantics remain unchanged within the configured budgets. +Command-local file-input redirections reserve reported file sizes before +opening or reading, and admit cached text and descriptor copies before growth. +Both representations count toward `max_live_intermediate_bytes` and aggregate +input consumption. Storage charges survive with stdin; replacing a numbered +descriptor releases its old storage without resetting aggregate consumption. +Admission uses the original backend path for stat/open/read, preserving mounted +namespaces, symlink policy, and read-only diagnostics; auxiliary device-path +canonicalization does not require synthetic mount-parent entries. +Budget and cancellation errors abort the request, while ordinary missing-file +errors remain shell command failures. Custom filesystems must report reliable +read sizes; backend-internal allocations are still the backend's responsibility. + `dd conv=sync` checks its padded block length against the 64 MiB copy cap before allocating or zero-filling. Copied payload capacity also counts against `max_live_intermediate_bytes` until output handling completes, including when diff --git a/knowledge/foundations/architecture.md b/knowledge/foundations/architecture.md index 915be0bba..516263045 100644 --- a/knowledge/foundations/architecture.md +++ b/knowledge/foundations/architecture.md @@ -152,6 +152,19 @@ compression paths use these builders because their expansion ratios and nested buffers make post-hoc leasing unsafe. Atomic compare/exchange admission keeps concurrent descendants from wrapping or temporarily exceeding the shared cap. +Command-local input redirections admit canonical VFS metadata/device read +ceilings before loading bytes, including before the empty open for `<>`. +Backend capacity beyond its reported size is charged before decoding. Cached +lossy text and fd copies use the existing owning builders; byte/text leases +travel with both the numbered-fd slot and surviving stdin. Reopening a numbered +fd replaces its old storage, without refunding aggregate input consumption. +An input not needed for stdin moves into the fd table instead of being copied. +Budget and cancellation errors abort the request; ordinary backend I/O errors +remain shell command failures. A custom VFS must report reliable read sizes; +the legacy `read_file` trait cannot prevent allocations inside a backend that +underreports or concurrently changes those sizes. Persistent/coprocess fd +storage remains a separate ownership boundary. + Shell diagnostic names are truncated to 1,024 UTF-8 bytes before copying, without changing positional `$0`. Builtin diagnostic prefix rewriting uses `BudgetedBytes` and fallible `StreamData` text construction under the same diff --git a/knowledge/operations/testing.md b/knowledge/operations/testing.md index 44ff6e173..5442f712e 100644 --- a/knowledge/operations/testing.md +++ b/knowledge/operations/testing.md @@ -172,6 +172,30 @@ plus the package-named `build/bashkit` *directory* not being mistaken for the binary and a decoy above the target directory being refused. `bashkit_cli_path()` fails closed naming every path it ruled out. +## Command-local input admission + +`interpreter::redirection::input_admission_tests` uses two-byte ordinary and +lazy files to check admission before loading, including the empty open for +`<>`, exact byte/text storage boundaries, failed-admission refunds, surviving +stdin storage, and injected cancellation during a read. `stream::tests` covers +lossy decoding without changing bytes and capacity-owned copy boundaries. +The ordinary `input_redirection_*` integration cases cover binary stdin/fd +copies, empty read-write creation, existing-file preservation, here inputs, +and tiny process-substitution input (including mapfile). A two-byte buffered +pipe fixture checks admission and lease release while pipe stat stays zero +and the existing substitution owner remains open. The owner exposes buffered +length without copying; admission must precede its `read_file` copy. +The cases also cover nonfatal missing-file diagnostics and enabled/disabled +symlink-following for ordinary read/read-write input. These fixtures require no +host mount: a two-byte in-memory mounted file without synthetic parent entries +covers direct read, read-write/create and refusal before lazy loading. A missing +canonical parent is not proof the backend cannot stat/read the original path. +Two-byte overlay/read-only and rebased-namespace fixtures also check successful +read, rejected read-write open with backend diagnostics, and unchanged content. +These fixtures require no +subprocess, device read, large allocation, or original resource-exhaustion shape. Selecting +them is scoped proof, not a claim that the unchanged full/coverage suites ran. + ## Coverage Uploaded to Codecov from three sources: Rust unit/integration coverage via @@ -249,6 +273,35 @@ surfaces as a 404 in an already-green job. ## Resource-cap tests must not race the deadline +The opt-in input-redirect helper polls its one in-flight RSS sampling future by +mutable reference across `select!` rounds. A losing branch must leave that future +in its owned slot; only observing its result clears the slot. The ordinary +`ir_helper_concurrent_schedule` fixture uses the wrapper's same +`poll_owned_sample` helper with 64 bytes of in-memory pipe data and a release-gated +future. This verifies scheduling ownership, not subprocess cleanup, effective +resource caps, platform coverage or merged security proof. + +The native input-redirect runner shares one owned lifecycle for the main child +and RSS sampler: wait, both capped captures and live first-failure notices are +polled together. Ordinary errors request kill before finite reap/capture completion; +completion waits for an owned pending sample and records incomplete cleanup as a +negative verdict. Reap timeout, I/O failure and success are separate outcomes. +The process feature is a native-only Tokio dev-dependency; WASM builds keep the +workspace feature set. Drop remains a best-effort backstop, not evidence of +reap or capture completion on panic or external cancellation. Tiny in-memory +fixtures verify notice delivery before EOF, nested reap failures and pending-sample +ownership. They do not establish subprocess, Linux or full security shipping proof. + +The input helper follows the shared sanitizer-compatible launcher +policy: no virtual-AS cap; its receipt explicitly marks AS unsupported and +labels resident-memory observations as sampled RSS with an unchanged 1 GiB +cutoff, not hard OS enforcement or a between-sample peak guarantee. CPU 10s, +stack 4096 KiB, file backstop 128 MiB on measured Bash units, separate artifact +acceptance below 64 MiB, wall deadlines and capped drains remain unchanged. +Backend-admission futures are boxed at the recursive redirect boundary to +avoid embedding their state in each command frame; layout-only observations +do not establish a depth-execution result. + Controlled retained-pipe fixtures keep a caller-owned reader-handoff vector outside their cleanup guard. A bounded `Drop` returns any unfinished ORIGINAL handle to that owner rather than recording a count and detaching it. The owner diff --git a/knowledge/security/threat-model.md b/knowledge/security/threat-model.md index 8207b979e..c3e72e28b 100644 --- a/knowledge/security/threat-model.md +++ b/knowledge/security/threat-model.md @@ -1429,6 +1429,24 @@ This section maps former vulnerability IDs to the new threat ID scheme and track | TM-DOS-106 | Zip creation allocation-before-check | `zip` recursively collected complete file contents and encoded its output before acquiring live-byte leases, allowing transient allocations above the configured intermediate-memory ceiling | Zip traversal collects paths only; each VFS file size is leased before `read_file`, leases remain live with the entries, and archive encoding uses `BudgetedBytes` to charge capacity before growth — **MITIGATED** | | TM-DOS-108 | CLI host-stdin read can stall before execution | A producer keeps the one-shot CLI's stdin pipe open while idle or trickles fewer than 10 MiB, so the pre-execution read never reaches EOF or the byte cap | Read stdin on a dedicated thread and bound the wait with the selected execution timeout; timeout exits with a diagnostic suggesting `--no-stdin`, while the existing byte cap remains enforced — **MITIGATED** | | TM-DOS-112 | `sed r` output amplification | A script can read the complete stdin after every input line, synchronously growing output quadratically inside the sed engine before any interpreter-level truncation applies | Every sed sink (stdout, `w`/`W` files, in-place edits) is a `BudgetedString` that leases capacity from the shared live-intermediate budget (`max_live_intermediate_bytes`) *before* growing, and a refused lease becomes `sed: ` with exit 1. The stdout *capture* cap is deliberately not a sink limit: sed output is routinely piped onward or redirected to a file, neither of which is captured stdout, so bounding sinks by it truncated ordinary transformations (#2455). Regression tests: `sed_resource_tests` — **MITIGATED** | + +Command-local input redirection storage also follows TM-DOS-103: canonical +file metadata and bounded device sizes are admitted before opening/reading; +cached text and fd copies use capacity-owning builders. Leases remain attached +to fd slots and surviving stdin, including failure and replacement paths. +Open process-substitution input uses its owner's buffered length before the +read copy, not POSIX pipe stat's zero size; its descriptive `pipe:[N]` link is +not canonicalized as a VFS target. The original buffered owner is unchanged. +Canonical resolution only identifies device ceilings; actual stat/open/read +use the original path through the configured VFS, preserving its symlink policy. +If canonical lookup lacks synthetic mount parents, the original backend path's +stat still decides read size/existence before allocation; non-NotFound lookup +errors remain visible. +Tiny `input_admission_tests` and `input_redirection_*` fixtures cover the +read/open ordering, byte fidelity, error typing and lifetime contracts. +This does not certify backend-internal allocation or persistent/coprocess fd +storage; a custom backend's reported read size remains a trusted contract. + ### Accepted (Low Priority) | Threat ID | Vulnerability | Impact | Rationale |