From fa513e4ace91b4d452a2321cb0f495d3ccb97e08 Mon Sep 17 00:00:00 2001 From: Mykhailo Chalyi Date: Sun, 18 Oct 2026 23:24:00 +0000 Subject: [PATCH 1/5] fix(streaming): enforce budgets and preserve routed byte output --- crates/bashkit/src/builtins/cat.rs | 14 +- crates/bashkit/src/builtins/cuttr.rs | 11 +- crates/bashkit/src/builtins/grep.rs | 31 +- crates/bashkit/src/builtins/mod.rs | 166 ++++++++--- crates/bashkit/src/builtins/seq.rs | 4 +- crates/bashkit/src/builtins/yes.rs | 2 +- crates/bashkit/src/command.rs | 131 ++++++--- crates/bashkit/src/fs/proc_sub_fds.rs | 13 +- crates/bashkit/src/interpreter/mod.rs | 98 +++++-- crates/bashkit/src/interpreter/pipe.rs | 1 + crates/bashkit/src/interpreter/redirection.rs | 2 +- crates/bashkit/src/limits.rs | 4 + crates/bashkit/tests/integration/main.rs | 1 + .../integration/streaming_audit_tests.rs | 270 ++++++++++++++++++ 14 files changed, 601 insertions(+), 147 deletions(-) create mode 100644 crates/bashkit/tests/integration/streaming_audit_tests.rs diff --git a/crates/bashkit/src/builtins/cat.rs b/crates/bashkit/src/builtins/cat.rs index 8389ab1ed..deeb2752c 100644 --- a/crates/bashkit/src/builtins/cat.rs +++ b/crates/bashkit/src/builtins/cat.rs @@ -73,7 +73,8 @@ impl Builtin for Cat { return stream_plain(&ctx, &files, &stream).await; } - let mut raw = Vec::new(); + let budget = ctx.shell.as_ref().map(|s| s.execution_budget); + let mut raw = crate::limits::BudgetedBytes::new(budget)?; let mut stderr = String::new(); // Buffered stdin is read once: `cat - -` sees it at the first `-`. let mut stdin_used = false; @@ -82,10 +83,10 @@ impl Builtin for Cat { if let Some(stdin) = ctx.stdin && !std::mem::replace(&mut stdin_used, true) { - raw.extend_from_slice(stdin.as_bytes()); + raw.try_extend_from_slice(stdin.as_bytes())?; } if let Some(input) = ctx.stdin_stream() { - raw.extend_from_slice(&input.read_to_end().await); + raw.try_extend_from_slice(&input.read_to_end().await?)?; } } else { let path = if Path::new(file).is_absolute() { @@ -94,7 +95,7 @@ impl Builtin for Cat { vfs_join(ctx.cwd, file).to_string_lossy().into_owned() }; match ctx.fs.read_file(Path::new(&path)).await { - Ok(bytes) => raw.extend_from_slice(&bytes), + Ok(bytes) => raw.try_extend_from_slice(&bytes)?, // GNU cat reports the operand and goes on with the rest. Err(e) => { let reason = crate::error::io_error_reason(&e); @@ -104,6 +105,7 @@ impl Builtin for Cat { } } + let (raw, _raw_storage) = raw.into_parts(); let mut result = if plain { ExecResult::ok_bytes(raw) } else { @@ -136,7 +138,7 @@ async fn stream_plain( async fn send(ctx: &Context<'_>, stream: &super::StdoutStream, data: &[u8]) -> Result { for chunk in data.chunks(STREAM_CHUNK_BYTES) { ctx.consume_budget_work(1)?; - if !stream.write(chunk).await { + if !stream.write(chunk).await? { return Ok(false); } } @@ -155,7 +157,7 @@ async fn stream_plain( }; if let Some(input) = ctx.stdin_stream() { while ok { - let chunk = input.read().await; + let chunk = input.read().await?; if chunk.is_empty() { break; } diff --git a/crates/bashkit/src/builtins/cuttr.rs b/crates/bashkit/src/builtins/cuttr.rs index f5eaa8d62..8362aa949 100644 --- a/crates/bashkit/src/builtins/cuttr.rs +++ b/crates/bashkit/src/builtins/cuttr.rs @@ -357,7 +357,7 @@ impl Builtin for Tr { // `yes | tr y n | head -1` ends with SIGPIPE. Squeezing looks across // piece boundaries, so `-s` reads everything first. if !squeeze && let Some(out) = ctx.stdout_stream() { - let mut input = super::InputChunks::stdin(&ctx); + let mut input = super::InputChunks::stdin(&ctx)?; let cut = |data: &[u8]| { if spec.c_locale { data.len() @@ -365,20 +365,21 @@ impl Builtin for Tr { super::cut_utf8(data) } }; - while let Some(piece) = input.next(cut).await { + while let Some(piece) = input.next(cut).await? { ctx.consume_budget_work(1)?; - let output = match spec.apply(&piece.into()) { + let piece = ctx.collected_input(piece)?; + let output = match spec.apply(&piece) { Ok(output) => output, Err(msg) => return Ok(Self::err(&msg, 1)), }; - if !output.is_empty() && !out.write(&output).await { + if !output.is_empty() && !out.write(&output).await? { return Ok(ExecResult::with_code(String::new(), 141)); } } return Ok(ExecResult::ok(String::new())); } - let stdin = ctx.stdin_to_end().await.unwrap_or_default(); + let stdin = ctx.stdin_to_end().await?; match spec.apply(&stdin) { Ok(output) => Ok(ExecResult::ok_bytes(output)), Err(msg) => Ok(Self::err(&msg, 1)), diff --git a/crates/bashkit/src/builtins/grep.rs b/crates/bashkit/src/builtins/grep.rs index 6abe9fcbb..ff3688c92 100644 --- a/crates/bashkit/src/builtins/grep.rs +++ b/crates/bashkit/src/builtins/grep.rs @@ -900,10 +900,10 @@ async fn scan_input( let opts = sh.opts; let term = if opts.null_data { b'\0' } else { b'\n' }; let (name, mut input, whole_binary) = match source { - Source::Stdin(name) => (name, super::InputChunks::stdin(ctx), false), + Source::Stdin(name) => (name, super::InputChunks::stdin(ctx)?, false), Source::File(name, bytes) => { let nul = bytes.contains(&0); - (name, super::InputChunks::bytes(bytes), nul) + (name, super::InputChunks::bytes(ctx, bytes)?, nul) } }; let mut scan = FileScan::new(&name, opts); @@ -911,7 +911,7 @@ async fn scan_input( scan.set_binary(opts); } while !scan.done { - let Some(piece) = input.next(super::cut_records(term)).await else { + let Some(piece) = input.next(super::cut_records(term)).await? else { break; }; ctx.consume_budget_work(1)?; @@ -928,7 +928,7 @@ async fn scan_input( if let Some(s) = stream && !out.is_empty() { - if !s.write(out).await { + if !s.write(out).await? { return Ok((scan.nlines, true)); } out.clear(); @@ -938,7 +938,7 @@ async fn scan_input( if let Some(s) = stream && !out.is_empty() { - if !s.write(out).await { + if !s.write(out).await? { return Ok((n, true)); } out.clear(); @@ -969,7 +969,11 @@ impl Builtin for Grep { for pattern_file in &opts.pattern_files { let content = match read_operand_file(&ctx, pattern_file).await { Ok(c) => c, + Err(error @ (crate::Error::ResourceLimit(_) | crate::Error::Cancelled)) => { + return Err(error); + } Err(reason) => { + let reason = crate::error::io_error_reason(&reason); return Ok(ExecResult::err( format!("grep: {pattern_file}: {reason}\n"), 2, @@ -991,7 +995,11 @@ impl Builtin for Grep { opts.exclude_patterns .extend(text.lines().filter(|l| !l.is_empty()).map(str::to_string)); } + Err(error @ (crate::Error::ResourceLimit(_) | crate::Error::Cancelled)) => { + return Err(error); + } Err(reason) => { + let reason = crate::error::io_error_reason(&reason); return Ok(ExecResult::err( format!("grep: {exclude_file}: {reason}\n"), 2, @@ -1182,18 +1190,11 @@ fn resolve(cwd: &std::path::Path, file: &str) -> std::path::PathBuf { /// Read a file named on the command line (`-f`, `--exclude-from`); `-` /// is standard input. `Err` is the strerror-style reason. -async fn read_operand_file(ctx: &Context<'_>, name: &str) -> std::result::Result, String> { +async fn read_operand_file(ctx: &Context<'_>, name: &str) -> Result> { if name == "-" { - return Ok(ctx - .stdin_to_end() - .await - .map(|s| s.as_bytes().to_vec()) - .unwrap_or_default()); + return Ok(ctx.stdin_to_end().await?.as_bytes().to_vec()); } - ctx.fs - .read_file(&resolve(ctx.cwd, name)) - .await - .map_err(|e| crate::error::io_error_reason(&e)) + ctx.fs.read_file(&resolve(ctx.cwd, name)).await } /// A resolved directory ancestor. Sibling aliases remain independent searches. diff --git a/crates/bashkit/src/builtins/mod.rs b/crates/bashkit/src/builtins/mod.rs index 93b468816..7cbdc726d 100644 --- a/crates/bashkit/src/builtins/mod.rs +++ b/crates/bashkit/src/builtins/mod.rs @@ -1016,17 +1016,46 @@ pub struct Context<'a> { } /// Stdin pipe of a streaming filter stage, see [`Context::stdin_stream`]. -pub(crate) struct StdinStream(std::sync::Arc); +pub(crate) struct StdinStream { + pipe: std::sync::Arc, + budget: crate::ExecutionBudget, +} impl StdinStream { /// Next chunk of input; empty at end of input. - pub(crate) async fn read(&self) -> Vec { - self.0.read_some().await + pub(crate) async fn read(&self) -> Result> { + let bytes = self.budget.run(self.pipe.read_some()).await?; + // THREAT[TM-DOS-096]: streaming consumers share the same input ledger + // as buffered utilities and public CommandInput readers. + self.budget.consume_input(bytes.len())?; + self.budget + .consume_work(1 + bytes.len().div_ceil(1024) as u64)?; + Ok(bytes) } /// Everything left, for filters that need their whole input. - pub(crate) async fn read_to_end(&self) -> Vec { - self.0.read_to_end().await + pub(crate) async fn read_to_end(&self) -> Result { + let mut bytes = crate::limits::BudgetedBytes::new(Some(&self.budget))?; + loop { + let chunk = self.read().await?; + if chunk.is_empty() { + return Ok(bytes); + } + bytes.try_extend_from_slice(&chunk)?; + } + } +} + +/// Collected input keeps both byte and text storage admitted while in use. +pub(crate) struct CollectedInput { + data: crate::StreamData, + _storage: crate::stream::StreamStorage, +} + +impl std::ops::Deref for CollectedInput { + type Target = crate::StreamData; + fn deref(&self) -> &Self::Target { + &self.data } } @@ -1040,39 +1069,44 @@ pub(crate) const STREAM_CHUNK_BYTES: usize = 4 * 1024; /// [`STREAM_CHUNK_BYTES`] unless one record is longer; the tail comes at /// end of input. pub(crate) struct InputChunks<'c> { - buf: Vec, + buf: crate::limits::BudgetedBytes, pos: usize, - buffered: Option<&'c [u8]>, + buffered: Option>, + buffered_pos: usize, + budget: Option, input: Option, eof: bool, } impl<'c> InputChunks<'c> { /// Stdin of this builtin: `ctx.stdin`, then the streaming pipe if any. - pub(crate) fn stdin(ctx: &Context<'c>) -> Self { - Self { - buf: Vec::new(), + pub(crate) fn stdin(ctx: &Context<'c>) -> Result { + let budget = ctx.shell.as_ref().map(|s| s.execution_budget.clone()); + Ok(Self { + buf: crate::limits::BudgetedBytes::new(budget.as_ref())?, pos: 0, - buffered: ctx.stdin.map(|s| s.as_bytes()), + buffered: ctx.stdin.map(|s| std::borrow::Cow::Borrowed(s.as_bytes())), + buffered_pos: 0, + budget, input: ctx.stdin_stream(), eof: false, - } + }) } /// Input that is already all in memory (a file's contents). - pub(crate) fn bytes(bytes: Vec) -> Self { - Self { - buf: bytes, - pos: 0, - buffered: None, - input: None, - eof: true, - } + pub(crate) fn bytes(ctx: &Context<'c>, bytes: Vec) -> Result { + let mut input = Self::stdin(ctx)?; + input.buffered = Some(std::borrow::Cow::Owned(bytes)); + input.input = None; + Ok(input) } /// Next piece, or `None` at end of input. `cut(data)` returns how many /// leading bytes of `data` can be handled now (0 for "need more"). - pub(crate) async fn next(&mut self, cut: impl Fn(&[u8]) -> usize) -> Option> { + pub(crate) async fn next( + &mut self, + cut: impl Fn(&[u8]) -> usize, + ) -> Result> { loop { let rest = &self.buf[self.pos..]; let window = rest.len().min(STREAM_CHUNK_BYTES); @@ -1084,27 +1118,38 @@ impl<'c> InputChunks<'c> { n = rest.len(); } if n > 0 { - let piece = rest[..n].to_vec(); + // THREAT[TM-DOS-103]: include long records and their returned + // copies in live storage; a bounded pipe alone is insufficient. + let mut piece = crate::limits::BudgetedBytes::new(self.budget.as_ref())?; + piece.try_extend_from_slice(&rest[..n])?; self.pos += n; - return Some(piece); + return Ok(Some(piece)); } if self.eof { - return None; + return Ok(None); } - self.buf.drain(..self.pos); + self.buf.copy_within(self.pos.., 0); + let remaining = self.buf.len() - self.pos; + self.buf.truncate(remaining); self.pos = 0; - if let Some(buffered) = self.buffered.take() { - self.buf.extend_from_slice(buffered); + if let Some(buffered) = &self.buffered { + let end = (self.buffered_pos + STREAM_CHUNK_BYTES).min(buffered.len()); + self.buf + .try_extend_from_slice(&buffered[self.buffered_pos..end])?; + self.buffered_pos = end; + if end == buffered.len() { + self.buffered = None; + } continue; } let more = match &self.input { - Some(input) => input.read().await, + Some(input) => input.read().await?, None => Vec::new(), }; if more.is_empty() { self.eof = true; } else { - self.buf.extend_from_slice(&more); + self.buf.try_extend_from_slice(&more)?; } } } @@ -1125,12 +1170,18 @@ pub(crate) fn cut_utf8(data: &[u8]) -> usize { } /// Stdout pipe of a streaming pipeline stage, see [`Context::stdout_stream`]. -pub(crate) struct StdoutStream(std::sync::Arc); +pub(crate) struct StdoutStream(crate::command::CommandOutput); impl StdoutStream { /// Wait for room in the pipe, then write. False once nobody reads. - pub(crate) async fn write(&self, data: &[u8]) -> bool { - self.0.write_bounded(data).await.is_ok() + pub(crate) async fn write(&self, data: &[u8]) -> Result { + match self.0.write_all(data).await { + Ok(()) => Ok(true), + Err(crate::Error::Io(error)) if error.kind() == std::io::ErrorKind::BrokenPipe => { + Ok(false) + } + Err(error) => Err(error), + } } } @@ -1223,7 +1274,7 @@ impl<'a> Context<'a> { pub(crate) fn stdout_stream(&self) -> Option { self.shell .as_ref() - .and_then(|shell| shell.stdout_pipe.clone()) + .and_then(|shell| shell.stdout_stream.clone()) .map(StdoutStream) } @@ -1231,24 +1282,47 @@ impl<'a> Context<'a> { /// pipeline (`loop | cat | head -1`). `ctx.stdin` then holds only data /// already buffered before the stage started; read the rest here. pub(crate) fn stdin_stream(&self) -> Option { - self.shell - .as_ref() - .and_then(|shell| shell.stdin_pipe.clone()) - .map(StdinStream) + self.shell.as_ref().and_then(|shell| { + shell.stdin_pipe.clone().map(|pipe| StdinStream { + pipe, + budget: shell.execution_budget.clone(), + }) + }) } /// All of stdin, including what is still in a streaming stdin pipe, for /// a filter stage that needs its whole input before it can write. - pub(crate) async fn stdin_to_end(&self) -> Option { - let Some(input) = self.stdin_stream() else { - return self.stdin.cloned(); + pub(crate) async fn stdin_to_end(&self) -> Result { + let budget = self.shell.as_ref().map(|s| s.execution_budget.clone()); + let mut all = crate::limits::BudgetedBytes::new(budget.as_ref())?; + if let Some(stdin) = self.stdin { + all.try_extend_from_slice(stdin.as_bytes())?; + } + if let Some(input) = self.stdin_stream() { + loop { + let chunk = input.read().await?; + if chunk.is_empty() { + break; + } + all.try_extend_from_slice(&chunk)?; + } + } + self.collected_input(all) + } + + pub(crate) fn collected_input( + &self, + bytes: crate::limits::BudgetedBytes, + ) -> Result { + let (data, storage) = if let Some(shell) = &self.shell { + crate::StreamData::from_budgeted_bytes(bytes, shell.execution_budget)? + } else { + (bytes.into_parts().0.into(), [None, None]) }; - let mut all = self - .stdin - .map(|s| s.as_bytes().to_vec()) - .unwrap_or_default(); - all.extend_from_slice(&input.read_to_end().await); - Some(all.into()) + Ok(CollectedInput { + data, + _storage: storage, + }) } /// Remaining wall-clock budget of the current `exec*` call, if limited. diff --git a/crates/bashkit/src/builtins/seq.rs b/crates/bashkit/src/builtins/seq.rs index eed9e67e1..924549e41 100644 --- a/crates/bashkit/src/builtins/seq.rs +++ b/crates/bashkit/src/builtins/seq.rs @@ -292,7 +292,7 @@ impl Builtin for Seq { && output.len() >= STREAM_CHUNK_BYTES { ctx.consume_budget_work(1)?; - if !stream.write(output.as_bytes()).await { + if !stream.write(output.as_bytes()).await? { return Ok(ExecResult::with_code("", 141)); } flushed += output.len(); @@ -319,7 +319,7 @@ impl Builtin for Seq { output.push('\n'); } if let Some(stream) = &stream { - if !stream.write(output.as_bytes()).await { + if !stream.write(output.as_bytes()).await? { return Ok(ExecResult::with_code("", 141)); } output.clear(); diff --git a/crates/bashkit/src/builtins/yes.rs b/crates/bashkit/src/builtins/yes.rs index 483e3ebdb..cd742f40d 100644 --- a/crates/bashkit/src/builtins/yes.rs +++ b/crates/bashkit/src/builtins/yes.rs @@ -75,7 +75,7 @@ impl Builtin for Yes { let lines = output.lines().count(); for chunk in output.as_bytes().chunks(STREAM_CHUNK_BYTES) { ctx.consume_budget_work(1)?; - if !stream.write(chunk).await { + if !stream.write(chunk).await? { return Ok(ExecResult::with_code("", 141)); } } diff --git a/crates/bashkit/src/command.rs b/crates/bashkit/src/command.rs index d4a28137f..904feb0be 100644 --- a/crates/bashkit/src/command.rs +++ b/crates/bashkit/src/command.rs @@ -226,13 +226,16 @@ pub(crate) struct CommandAdapter { pub command: Arc, } -#[async_trait] -impl Builtin for CommandAdapter { - fn streams_io(&self) -> bool { - true - } +/// Shared routed byte destinations for host commands and streaming utilities. +pub(crate) struct CommandIo { + pub stdout: CommandOutput, + pub stderr: CommandOutput, + out_capture: Arc>, + err_capture: Arc>, +} - async fn execute(&self, ctx: BuiltinContext<'_>) -> Result { +impl CommandIo { + pub(crate) fn new(ctx: &BuiltinContext<'_>) -> Result { let budget = ctx .execution_budget() .ok_or_else(|| Error::Internal("missing command budget".into()))? @@ -277,6 +280,79 @@ impl Builtin for CommandAdapter { Some([out, err]) => (resolve(out), resolve(err)), None => (stdout, stderr), }; + Ok(Self { + stdout: CommandOutput { + destination: stdout, + budget: budget.clone(), + }, + stderr: CommandOutput { + destination: stderr, + budget, + }, + out_capture, + err_capture, + }) + } + + pub(crate) fn collect(&self, mut result: ExecResult) -> Result { + let mut out = self.out_capture.lock().unwrap_or_else(|e| e.into_inner()); + let mut err = self.err_capture.lock().unwrap_or_else(|e| e.into_inner()); + result.stdout = std::mem::replace(&mut out.bytes, crate::limits::BudgetedVec::new(None)?) + .into_parts() + .0 + .into(); + result.stderr = std::mem::replace(&mut err.bytes, crate::limits::BudgetedVec::new(None)?) + .into_parts() + .0 + .into(); + result.stdout_truncated |= out.truncated; + result.stderr_truncated |= err.truncated; + Ok(result) + } + + pub(crate) async fn write_error(&self, name: &str, error: std::io::Error) -> Result { + if error.kind() == std::io::ErrorKind::BrokenPipe { + return Ok(141); + } + let message = format!( + "{name}: write error: {}\n", + crate::error::io_error_reason(&Error::Io(error)) + ); + write_diagnostic(&self.stderr, message.as_bytes()).await?; + Ok(1) + } + + /// Buffered options and diagnostics use the same already-open destinations; + /// applying redirects again would truncate files and lose streamed bytes. + pub(crate) async fn complete(&self, name: &str, mut result: ExecResult) -> Result { + for (output, bytes) in [ + (&self.stdout, std::mem::take(&mut result.stdout)), + (&self.stderr, std::mem::take(&mut result.stderr)), + ] { + if bytes.is_empty() { + continue; + } + match output.write_all(bytes.as_bytes()).await { + Ok(()) => {} + Err(Error::Io(error)) => { + result.exit_code = self.write_error(name, error).await?; + } + Err(error) => return Err(error), + } + } + self.collect(result) + } +} + +#[async_trait] +impl Builtin for CommandAdapter { + fn streams_io(&self) -> bool { + true + } + + async fn execute(&self, ctx: BuiltinContext<'_>) -> Result { + let io = CommandIo::new(&ctx)?; + let budget = io.stdout.budget.clone(); let input_pipe = ctx.shell.as_ref().and_then(|s| s.stdin_pipe.clone()); let input_storage = budget.lease_bytes(ctx.stdin.map_or(0, StreamData::len))?; let buffered = ctx.stdin.cloned().unwrap_or_default(); @@ -289,10 +365,6 @@ impl Builtin for CommandAdapter { _storage: input_storage, })), }; - let error_output = CommandOutput { - destination: stderr.clone(), - budget: budget.clone(), - }; let result = self.command.execute(CommandContext { name: &self.name, args: ctx.args, @@ -300,14 +372,8 @@ impl Builtin for CommandAdapter { cwd: ctx.cwd, fs: ctx.fs.clone(), stdin: input, - stdout: CommandOutput { - destination: stdout, - budget: budget.clone(), - }, - stderr: CommandOutput { - destination: stderr, - budget: budget.clone(), - }, + stdout: io.stdout.clone(), + stderr: io.stderr.clone(), extensions: ctx .shell .as_ref() @@ -320,37 +386,14 @@ impl Builtin for CommandAdapter { .await? { Ok(Ok(status)) => status, - Ok(Err(Error::Io(e))) if e.kind() == std::io::ErrorKind::BrokenPipe => { - CommandStatus::with_code(141) - } - Ok(Err(Error::Io(e))) => { - let message = format!( - "{}: write error: {}\n", - self.name, - crate::error::io_error_reason(&Error::Io(e)) - ); - write_diagnostic(&error_output, message.as_bytes()).await?; - CommandStatus::with_code(1) - } + Ok(Err(Error::Io(e))) => CommandStatus::with_code(io.write_error(&self.name, e).await?), Ok(Err(e)) => return Err(e), Err(_) => { - write_diagnostic(&error_output, b"command failed unexpectedly\n").await?; + write_diagnostic(&io.stderr, b"command failed unexpectedly\n").await?; CommandStatus::with_code(1) } }; - let mut out = out_capture.lock().unwrap_or_else(|e| e.into_inner()); - let mut err = err_capture.lock().unwrap_or_else(|e| e.into_inner()); - Ok(ExecResult { - stdout: std::mem::replace(&mut out.bytes, crate::limits::BudgetedVec::new(None)?) - .into_parts() - .0 - .into(), - stderr: std::mem::replace(&mut err.bytes, crate::limits::BudgetedVec::new(None)?) - .into_parts() - .0 - .into(), - stdout_truncated: out.truncated, - stderr_truncated: err.truncated, + io.collect(ExecResult { exit_code: status.exit_code, ..Default::default() }) diff --git a/crates/bashkit/src/fs/proc_sub_fds.rs b/crates/bashkit/src/fs/proc_sub_fds.rs index 191759a69..4fc3969a2 100644 --- a/crates/bashkit/src/fs/proc_sub_fds.rs +++ b/crates/bashkit/src/fs/proc_sub_fds.rs @@ -36,11 +36,20 @@ pub(crate) type ProcSubBuffer = Arc>>; #[derive(Clone)] pub(crate) enum ProcSubData { /// `<(cmd)`: the list's output, read by the command. - Input(Arc<[u8]>), + Input(Arc), /// `>(cmd)`: what the command writes, fed to the list afterwards. Output(ProcSubBuffer), } +#[cfg(test)] +impl ProcSubData { + pub(crate) fn test_input(data: &[u8]) -> Self { + let mut bytes = crate::limits::BudgetedBytes::new(None).unwrap(); + bytes.try_extend_from_slice(data).unwrap(); + Self::Input(Arc::new(bytes)) + } +} + pub(crate) struct ProcSubFs { inner: Arc, /// Open substitutions in allocation order. @@ -345,7 +354,7 @@ mod tests { let a = ProcSubFs::new(Arc::clone(&shared)); let b = ProcSubFs::new(Arc::clone(&shared)); let p = Path::new("/dev/fd/63"); - a.open(63, ProcSubData::Input(Arc::from(&b"a-data"[..]))); + a.open(63, ProcSubData::test_input(b"a-data")); assert_eq!(a.read_file(p).await.unwrap(), b"a-data"); assert!(a.exists(p).await.unwrap()); assert_eq!(a.stat(p).await.unwrap().file_type, FileType::Fifo); diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index 71777d11d..407de802f 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -421,6 +421,7 @@ pub(crate) struct ShellRef<'a> { pub(crate) stdin_pipe: Option>, pub(crate) stderr_pipe: Option>, pub(crate) command_routes: Option<[crate::command::CommandRoute; 2]>, + pub(crate) stdout_stream: Option, /// Enclosing loops in the current function (`break`/`continue`). pub(crate) loop_depth: usize, /// Active function calls and `source`s (`return` is valid when > 0). @@ -4233,7 +4234,7 @@ impl Interpreter { let emit_before = self.output_emit_count; self.maybe_emit_output(&result.stdout, &result.stderr, emit_before) .await?; - self.flush_exec_passthrough(&mut result); + Box::pin(self.flush_exec_passthrough(&mut result)).await?; Ok(result) }) } @@ -4357,7 +4358,7 @@ impl Interpreter { self.maybe_emit_output(&result.stdout, &result.stderr, emit_before) .await?; if top_level { - self.flush_exec_passthrough(&mut result); + Box::pin(self.flush_exec_passthrough(&mut result)).await?; } // Accumulate stdout with truncation @@ -7984,18 +7985,18 @@ impl Interpreter { /// large, so `yes | head -1` and `seq 1000000 | head -1` stop at the /// first line with SIGPIPE instead of running into the output caps. fn streams_stdout(simple: &SimpleCommand) -> bool { - simple.redirects.is_empty() - && matches!( - simple.name.parts.as_slice(), - [WordPart::Literal(name)] - if matches!(name.as_str(), "yes" | "seq" | "cat" | "grep" | "tr") - ) + matches!( + simple.name.parts.as_slice(), + [WordPart::Literal(name)] + if matches!(name.as_str(), "yes" | "seq" | "cat" | "grep" | "tr") + ) } /// Streaming stages that also read their stdin pipe incrementally /// (`Context::stdin_stream`), so `loop | grep y | head -1` stops early. - /// They read the pipe to the end themselves when an option needs the - /// whole input (`grep -c`, `tr -s`). + /// They consume incrementally; options such as `tr -s` explicitly collect + /// budget-owned whole input. Grep summaries can delay output, while -q + /// still exits after its first matching record. fn streams_stdin(name: &str) -> bool { matches!(name, "cat" | "grep" | "tr") } @@ -8489,6 +8490,9 @@ impl Interpreter { } self.maybe_emit_output(&result.stdout, &result.stderr, emit_before) .await?; + if route { + Box::pin(self.flush_exec_passthrough(&mut result)).await?; + } if self.retain_streamed_output() { stdout.append(&result.stdout); stderr.append(&result.stderr); @@ -8595,6 +8599,9 @@ impl Interpreter { } self.maybe_emit_output(&result.stdout, &result.stderr, emit_before) .await?; + if route { + Box::pin(self.flush_exec_passthrough(&mut result)).await?; + } if self.retain_streamed_output() { stdout.append(&result.stdout); stderr.append(&result.stderr); @@ -9534,8 +9541,7 @@ impl Interpreter { let output = self .execution_output() .filter(|_| self.output_callback.is_some()); - let can_stream = !Self::has_output_redirect(&command.redirects) - && !self.scoped.functions.contains_key(name); + let can_stream = !self.scoped.functions.contains_key(name); self.builtin_stdout_pipe = if can_stream && (self.stream_stdout_command == Some(command as *const SimpleCommand as usize) || (self.output_callback.is_some() @@ -9843,10 +9849,22 @@ impl Interpreter { /// Hand output written to a saved original stdout/stderr (`>&3` after /// `exec 3>&1 >log`) to the caller, unrouted. - fn flush_exec_passthrough(&mut self, result: &mut ExecResult) { + async fn flush_exec_passthrough(&mut self, result: &mut ExecResult) -> Result<()> { let (out, err) = std::mem::take(&mut self.exec_passthrough); if out.is_empty() && err.is_empty() { - return; + return Ok(()); + } + // Saved original descriptors bypass persistent redirects, but must + // still use the request's awaited, capped byte-output pumps. + if self.output_callback.is_some() + && let Some(output) = self.execution_output().filter(|_| self.pipe_out.is_none()) + { + self.execution_budget + .run(output.stdout.write_bounded(out.as_bytes())) + .await??; + self.execution_budget + .run(output.stderr.write_bounded(err.as_bytes())) + .await??; } if let Some(cb) = self.output_callback.as_mut() { cb(&out, &err); @@ -9856,6 +9874,7 @@ impl Interpreter { } result.stdout.append(&out); result.stderr.append(&err); + Ok(()) } fn exec_fd_alias_target(&self, target_fd: i32) -> FdTarget { @@ -10094,6 +10113,7 @@ impl Interpreter { stdin_pipe: None, stderr_pipe: None, command_routes: None, + stdout_stream: None, loop_depth: self.loop_depth, return_depth: self.return_depth, last_exit_code: self.last_exit_code, @@ -10151,7 +10171,11 @@ impl Interpreter { (fs, None) }; let streaming_command = builtin.streams_io(); - let command_routes = if streaming_command { + let streaming_utility = bundled + && matches!(name, "yes" | "seq" | "cat" | "grep" | "tr") + && self.builtin_stdout_pipe.is_some(); + let routed_output = streaming_command || streaming_utility; + let command_routes = if routed_output { self.builtin_stdout_pipe = self .pipe_out .clone() @@ -10172,7 +10196,7 @@ impl Interpreter { } else { None }; - let stderr_pipe = if builtin.streams_io() && self.output_callback.is_some() { + let stderr_pipe = if routed_output && self.output_callback.is_some() { self.execution_output().map(|o| o.stderr.clone()) } else { None @@ -10200,11 +10224,12 @@ impl Interpreter { stdin_pipe: self.builtin_stdin_pipe.take(), stderr_pipe, command_routes, + stdout_stream: None, loop_depth: self.loop_depth, return_depth: self.return_depth, last_exit_code: self.last_exit_code, }; - let ctx = builtins::Context { + let mut ctx = builtins::Context { args, env: &self.env, variables: Arc::make_mut(&mut self.scoped.variables), @@ -10220,6 +10245,14 @@ impl Interpreter { shell: Some(shell_ref), }; + let utility_io = if streaming_utility { + let io = crate::command::CommandIo::new(&ctx)?; + ctx.shell.as_mut().expect("builtin shell").stdout_stream = Some(io.stdout.clone()); + Some(io) + } else { + None + }; + // THREAT[TM-INT-001]: Execute builtin with panic catching for security let result = AssertUnwindSafe(builtin.execute(ctx)).catch_unwind().await; @@ -10235,6 +10268,14 @@ impl Interpreter { Ok(Err(crate::error::Error::Regex(e))) if bundled => { builtin_usage_error(&format!("{name}: {e}")) } + Ok(Err(crate::Error::Io(error))) if utility_io.is_some() => ExecResult { + exit_code: utility_io + .as_ref() + .expect("routed utility") + .write_error(name, error) + .await?, + ..Default::default() + }, Ok(Err(e)) => return Err(e), Err(_panic) => ExecResult::err( self.diag(format!("{name}: builtin failed unexpectedly\n")), @@ -10282,7 +10323,9 @@ impl Interpreter { self.apply_builtin_side_effects(&mut result).await?; - let result = if streaming_command { + let result = if let Some(io) = utility_io { + io.complete(name, result).await? + } else if streaming_command { result } else { self.apply_redirections(result, redirects).await? @@ -12837,7 +12880,7 @@ impl Interpreter { } if is_input { - let mut stdout = String::new(); + let mut stdout = crate::limits::BudgetedBytes::new(Some(&self.execution_budget))?; // The substituted list runs in a subshell: nothing it changes // (variables, cwd, options, `$?`) reaches the parent. let snapshot = Box::new(self.snapshot_subshell_state()); @@ -12853,7 +12896,14 @@ impl Interpreter { for cmd in commands { match self.execute_command(cmd).await { Ok(cmd_result) => { - stdout.push_str(&cmd_result.stdout.command_substitution_text()); + // Process substitution is byte transport, unlike $(...). + // Admit capture growth without applying shell-word decoding. + if let Err(error) = + stdout.try_extend_from_slice(cmd_result.stdout.as_bytes()) + { + failed = Some(error.into()); + break; + } if matches!( cmd_result.control_flow, ControlFlow::Exit(_) | ControlFlow::Abort @@ -12877,10 +12927,8 @@ impl Interpreter { // Allocated after the list ran: in bash the child closes its own // end, so a substitution nested inside sees the number free. let fd = self.allocate_proc_sub_fd()?; - self.proc_subs.open( - fd, - crate::fs::ProcSubData::Input(Arc::from(stdout.into_bytes())), - ); + self.proc_subs + .open(fd, crate::fs::ProcSubData::Input(Arc::new(stdout))); Ok(format!("/dev/fd/{fd}")) } else { let fd = self.allocate_proc_sub_fd()?; @@ -18369,7 +18417,7 @@ cat /tmp/test_fd_exec_public.txt"#, // While open in `owner`, the fd exists only there. owner .proc_subs - .open(63, crate::fs::ProcSubData::Input(Arc::from(&b"secret"[..]))); + .open(63, crate::fs::ProcSubData::test_input(b"secret")); assert_eq!( owner.fs.read_file(Path::new("/dev/fd/63")).await.unwrap(), b"secret" diff --git a/crates/bashkit/src/interpreter/pipe.rs b/crates/bashkit/src/interpreter/pipe.rs index 89a36ba60..43e5a13ec 100644 --- a/crates/bashkit/src/interpreter/pipe.rs +++ b/crates/bashkit/src/interpreter/pipe.rs @@ -212,6 +212,7 @@ impl Pipe { } /// Read everything until the writer closes. + #[cfg(test)] pub(crate) async fn read_to_end(&self) -> Vec { let mut out = Vec::new(); loop { diff --git a/crates/bashkit/src/interpreter/redirection.rs b/crates/bashkit/src/interpreter/redirection.rs index 4334e980c..d74c675f3 100644 --- a/crates/bashkit/src/interpreter/redirection.rs +++ b/crates/bashkit/src/interpreter/redirection.rs @@ -1882,7 +1882,7 @@ mod input_admission_tests { bash.interpreter.begin_execution_budget(); bash.interpreter .proc_subs - .open(63, crate::fs::ProcSubData::Input(Arc::from(&b"ab"[..]))); + .open(63, crate::fs::ProcSubData::test_input(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; diff --git a/crates/bashkit/src/limits.rs b/crates/bashkit/src/limits.rs index b24991942..9bcc6dcbb 100644 --- a/crates/bashkit/src/limits.rs +++ b/crates/bashkit/src/limits.rs @@ -1184,6 +1184,10 @@ impl BudgetedVec { self.inner.pop() } + pub(crate) fn truncate(&mut self, len: usize) { + self.inner.truncate(len); + } + pub(crate) fn into_parts(self) -> (Vec, Option) { (self.inner, self.lease) } diff --git a/crates/bashkit/tests/integration/main.rs b/crates/bashkit/tests/integration/main.rs index 8131f2242..1044b5706 100644 --- a/crates/bashkit/tests/integration/main.rs +++ b/crates/bashkit/tests/integration/main.rs @@ -165,6 +165,7 @@ pub mod sqlite_integration_tests; pub mod sqlite_security_tests; pub mod stack_overflow_regression_tests; pub mod stderr_fidelity_tests; +pub mod streaming_audit_tests; pub mod subst_depth_limit_tests; pub mod symlink_overlay_security_tests; #[cfg(feature = "terminal")] diff --git a/crates/bashkit/tests/integration/streaming_audit_tests.rs b/crates/bashkit/tests/integration/streaming_audit_tests.rs new file mode 100644 index 000000000..f831eb0e5 --- /dev/null +++ b/crates/bashkit/tests/integration/streaming_audit_tests.rs @@ -0,0 +1,270 @@ +//! Streaming invariants across builtin, descriptor and substitution boundaries. + +use bashkit::{ + Bash, Command, CommandContext, CommandStatus, ExecutionBudgetExceeded, ExecutionLimits, + LimitExceeded, async_trait, +}; +use std::sync::Arc; +use tokio::sync::Notify; + +struct Chunks; + +#[async_trait] +impl Command for Chunks { + async fn execute(&self, ctx: CommandContext<'_>) -> bashkit::Result { + let count: usize = ctx.args[0].parse().unwrap(); + for _ in 0..count { + ctx.stdout.write_all(&[b'x'; 1024]).await?; + } + ctx.stdout.write_all(b"\n").await?; + Ok(CommandStatus::success()) + } +} + +#[tokio::test] +async fn streaming_builtin_input_consumes_shared_budget() { + for filter in [ + "cat", + "grep -q x", + "grep absent", + "tr x y", + "cat -n", + "tr -s x", + ] { + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_aggregate_input_bytes(16_384)) + .command("chunks", Chunks) + .build(); + let script = format!("chunks 32 | {filter}"); + let result = bash.execute(&script).await; + assert!( + matches!( + result, + Err(bashkit::Error::ResourceLimit( + LimitExceeded::ExecutionBudget(ExecutionBudgetExceeded::InputBytes { .. }) + )) + ), + "{script}: {result:?}" + ); + assert_eq!( + bash.execute("echo recovered") + .capture(32) + .await + .unwrap() + .stdout, + "recovered\n" + ); + } +} + +#[tokio::test] +async fn streaming_records_and_collectors_admit_memory_before_growth() { + for filter in ["grep -q x", "grep absent", "cat -n", "tr -s x"] { + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_live_intermediate_bytes(65_536)) + .command("chunks", Chunks) + .build(); + let script = format!("chunks 128 | {filter}"); + let result = bash.execute(&script).await; + assert!( + matches!( + result, + Err(bashkit::Error::ResourceLimit( + LimitExceeded::ExecutionBudget(ExecutionBudgetExceeded::LiveBytes { .. }) + )) + ), + "{script}: {result:?}" + ); + assert_eq!( + bash.execute("echo recovered") + .capture(32) + .await + .unwrap() + .stdout, + "recovered\n" + ); + } + // Total volume may exceed the live-storage cap when no record accumulates. + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_live_intermediate_bytes(65_536)) + .command("chunks", Chunks) + .build(); + assert_eq!(bash.execute("chunks 128 | cat").await.unwrap().exit_code, 0); +} + +#[tokio::test] +async fn saved_output_descriptors_reach_sinks_and_capture() { + for (script, stdout, stderr) in [ + ("exec 3>&1 >/out; echo hi >&3", "hi\n", ""), + ("exec 3>&2 2>/out; echo hi >&3", "", "hi\n"), + ( + "exec 3>&1 >/out; printf hi >&3; printf file; cat /out >&3", + "hifile", + "", + ), + ] { + let mut bash = Bash::new(); + let (mut out, mut err) = (Vec::new(), Vec::new()); + let result = bash + .execute(script) + .stdout(&mut out) + .stderr(&mut err) + .capture(64) + .await + .unwrap(); + assert_eq!(result.exit_code, 0, "{script}"); + assert_eq!(out, stdout.as_bytes(), "{script}"); + assert_eq!(err, stderr.as_bytes(), "{script}"); + assert_eq!(result.stdout.as_bytes(), out, "{script}"); + assert_eq!(result.stderr.as_bytes(), err, "{script}"); + } +} + +#[tokio::test] +async fn process_substitution_preserves_bytes_in_buffered_and_live_execution() { + for script in [ + "cat <(printf '\\377\\000A')", + "cat <(printf '\\377'; printf '\\000A')", + "printf '\\377\\000A' > >(cat)", + ] { + let oracle = std::process::Command::new("bash") + .args(["--noprofile", "--norc", "-c", script]) + .output() + .unwrap(); + assert!(oracle.status.success()); + for live in [false, true] { + let mut bash = Bash::new(); + let result = if live { + bash.execute(script).capture(64).await + } else { + bash.exec(script).await + } + .unwrap(); + assert_eq!( + result.stdout.as_bytes(), + oracle.stdout, + "{script}, live={live}" + ); + assert_eq!(result.stderr.as_bytes(), oracle.stderr); + assert_eq!(result.exit_code, 0); + } + } +} + +struct Gated(Arc); + +#[async_trait] +impl Command for Gated { + async fn execute(&self, ctx: CommandContext<'_>) -> bashkit::Result { + ctx.stdout.write_all(b"first\n").await?; + self.0.notified().await; + ctx.stdout.write_all(b"last\n").await?; + Ok(CommandStatus::success()) + } +} + +#[tokio::test] +async fn descriptor_redirects_keep_final_filters_live() { + for filter in [ + "cat 2>&1", + "grep first 2>&1", + "tr a-z A-Z 2>&1", + "cat 2>/errors", + "cat 3>&1 >&3", + ] { + let release = Arc::new(Notify::new()); + let sink_release = release.clone(); + let mut bash = Bash::builder().command("gated", Gated(release)).build(); + let script = format!("gated | {filter}"); + let result = tokio::time::timeout( + std::time::Duration::from_secs(1), + bash.execute(&script).capture(64).stdout_fn(move |bytes| { + assert!(!bytes.is_empty()); + sink_release.notify_one(); + async { Ok(()) } + }), + ) + .await + .expect("redirect buffered output until gated producer completion") + .unwrap(); + assert_eq!(result.exit_code, 0, "{script}"); + assert!(result.stderr.is_empty(), "{script}"); + } +} + +#[tokio::test] +async fn redirected_streaming_utilities_match_bash() { + for script in [ + "printf 'a\\nb\\n' | cat 2>&1", + "printf 'a\\nb\\n' | cat 1>&2", + "printf 'a\\nb\\n' | cat >out 2>&1; cat out", + "printf 'a\\nb\\n' | cat -n >out; cat out", + "printf 'a\\nb\\n' | grep -c a >out; cat out", + "printf 'a\\nb\\n' | tr a-z A-Z 3>&1 >&3", + "seq 3 >out; cat out", + "printf 'a\\nb\\n' | cat 3>&1 1>&2 2>&3", + ] { + let scratch = tempfile::tempdir().unwrap(); + let oracle = std::process::Command::new("bash") + .args(["--noprofile", "--norc", "-c", script]) + .current_dir(scratch.path()) + .output() + .unwrap(); + let mut bash = Bash::new(); + let result = bash.execute(script).capture(1024).await.unwrap(); + assert_eq!(result.stdout.as_bytes(), oracle.stdout, "{script}"); + assert_eq!(result.stderr.as_bytes(), oracle.stderr, "{script}"); + assert_eq!(result.exit_code, oracle.status.code().unwrap(), "{script}"); + } +} + +#[tokio::test] +async fn saved_descriptor_delivery_respects_output_caps_and_sink_failure() { + let script = "exec 3>&1 >/out; printf abcdef >&3"; + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_stdout_bytes(3)) + .build(); + let mut delivered = Vec::new(); + let result = bash + .execute(script) + .stdout(&mut delivered) + .capture(2) + .await + .unwrap(); + assert_eq!(delivered, b"abc"); + assert_eq!(result.stdout.as_bytes(), b"ab"); + assert!(result.stdout_truncated); + let mut bash = Bash::new(); + let result = bash + .execute(script) + .stdout_fn(|_| async { Err(std::io::Error::other("sink refused output").into()) }) + .await; + assert!(matches!(result, Err(bashkit::Error::Io(_)))); + assert_eq!( + bash.execute("exec 1>&3; echo recovered") + .capture(32) + .await + .unwrap() + .stdout, + "recovered\n" + ); +} + +#[tokio::test] +async fn streaming_write_errors_remain_command_statuses() { + let mut bash = Bash::new(); + let result = bash + .execute("printf x | cat 1>&-; echo status=$?") + .capture(1024) + .await + .unwrap(); + assert_eq!(result.stdout, "status=1\n"); + assert!(result.stderr.contains("Bad file descriptor")); + let result = bash + .execute("printf '' | cat 1>&-") + .capture(64) + .await + .unwrap(); + assert_eq!(result.exit_code, 0); + assert!(result.stderr.is_empty()); +} From 0af8b7ff27d5486891738917bb8203477c718608 Mon Sep 17 00:00:00 2001 From: Mykhailo Chalyi Date: Sun, 18 Oct 2026 23:24:00 +0000 Subject: [PATCH 2/5] chore(streaming): document byte routing and budget guarantees --- crates/bashkit/docs/streaming.md | 7 +++++++ docs/streaming.md | 17 +++++++++++++++++ knowledge/foundations/streaming-roadmap.md | 17 +++++++++++++++++ knowledge/operations/limitations.md | 4 ++-- knowledge/security/security-testing.md | 11 +++++++++++ knowledge/security/threat-model.md | 8 ++++++++ 6 files changed, 62 insertions(+), 2 deletions(-) diff --git a/crates/bashkit/docs/streaming.md b/crates/bashkit/docs/streaming.md index 35be12dd0..949ae938b 100644 --- a/crates/bashkit/docs/streaming.md +++ b/crates/bashkit/docs/streaming.md @@ -70,6 +70,13 @@ sandbox delivery truncation. before the next execution. Deadlines cover pending sinks as well as engine work. Command I/O and filesystem handles expire when the execution ends. +Streaming builtin readers share the aggregate input budget. Record buffers +and whole-input collectors reserve intermediate memory before growth; even a +sink-only execution can reject an oversized record. Streaming `cat`, `grep` +and `tr` output follows descriptor routing without disabling live delivery. +Process substitutions preserve raw bytes but remain buffered; `$(...)` is +still a text boundary. + Use buffered execution when `after_tool` or `after_exec` must inspect or replace whole output. Live execution rejects those hooks and `ExecOptions` output observers; use async byte sinks for live delivery. diff --git a/docs/streaming.md b/docs/streaming.md index 627ba8856..a62cb27be 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -203,6 +203,23 @@ host array or queue can still retain the whole output even when capture is disabled. Rust async writers flush at completion; flush inside a byte callback when a buffered destination must make each chunk visible immediately. +Streaming builtin readers charge input to the request's aggregate byte budget. +Record buffers and whole-input collectors also reserve intermediate memory +before growing. A very long line can therefore fail the memory budget even +when pipe queues remain within 4 KiB; disabling capture does not disable these +resource limits. + +Plain `cat`, line-printing `grep`, and nonsqueezing `tr` preserve live delivery +through descriptor duplication and file redirection, including `2>&1`. +Redirected files are opened once before execution. Options that collect or +summarize output can still delay delivery until their result is available. +Other utilities, including `sed`, `awk`, and `tee`, retain their buffered +execution paths. + +Process substitutions `<(...)` and `>(...)` preserve raw bytes, including NUL +and invalid UTF-8, but still buffer rather than running both sides as live +tasks. Command substitution `$(...)` remains a text boundary and removes NUL. + ## Routing, errors and cancellation A command context provides its name, expanded arguments (`args` in Rust, diff --git a/knowledge/foundations/streaming-roadmap.md b/knowledge/foundations/streaming-roadmap.md index e8cb8fa61..b84b40dd3 100644 --- a/knowledge/foundations/streaming-roadmap.md +++ b/knowledge/foundations/streaming-roadmap.md @@ -19,6 +19,8 @@ utilities and legacy tool/terminal queues remain separate work. Input process substitutions suppress outer output sinks while collecting their producer bytes, including incremental builtins and live host commands; this preserves the existing buffered `/dev/fd` behavior without leaking intermediate output. +Their input captures retain budget-owned bytes with the descriptor instead of +decoding through command-substitution text; NUL and invalid UTF-8 survive. Native process spawning, a browser process host, and jobs surviving `exec()` require separate designs; the default remains an in-process sandbox. @@ -73,6 +75,21 @@ state on dropped futures. Async output emission stays boxed to respect the existing native debug-build stack budget. Live execution yields every 64 commands for fairness; buffered synchronous APIs retain their single-poll model. +Streaming utilities and host commands share `CommandIo` output destinations. +Their redirects open once; incremental writes, buffered option results and +diagnostics use those same destinations. Saved original stdout/stderr copies +(`exec 3>&1 >log; echo hi >&3`) reach the awaited output pumps, including caps +and sink failure. Top-level list elements flush saved-descriptor output before +the next element can emit, preserving per-destination order. + +Internal streaming readers charge each pipe read to aggregate input/work. +`InputChunks` uses capacity-owning byte buffers for records and returned pieces; +`cat` and `tr` whole-input paths keep capture leases while consuming data. A +long delimiter-free record can fail admission even with sink-only execution. +`streaming_audit_tests` pins input/live-byte rejection, session recovery, byte +substitution, saved-fd delivery and gated redirected filters. This does not +establish memory bounds for every buffered utility or tool/terminal queue. + Native Python and NAPI JS expose live commands and async byte sinks on Bash and BashTool; JS/WASM uses the same core on its cooperative executor. C ABI exposes output-only synchronous callbacks. diff --git a/knowledge/operations/limitations.md b/knowledge/operations/limitations.md index d36913dfd..2c0cb2381 100644 --- a/knowledge/operations/limitations.md +++ b/knowledge/operations/limitations.md @@ -43,13 +43,13 @@ execution model. Evidence is a threat-model ID, a test, or `stance` | ID | Limitation | Why | Evidence | |----|------------|-----|----------| | L-DIAG-001 | Shell diagnostic names show at most 1,024 UTF-8 bytes of `$0` or the current source filename; the actual `$0` is unchanged. Prefix amplification can abort a request at the shared live-memory/work cap, including redirected or streamed diagnostics | Prevent attacker-controlled diagnostic names from multiplying into unchecked host allocations (TM-DOS-103) | `diagnostic_prefix_tests` | -| L-PROC-001 | `exec` does not replace the process; `exec cmd` runs cmd then stops execution. fd redirects work; `exec >log 2>&1` routes the shell's output per top-level command (each element of a top-level `;`/`&&`/`||` list counts as one), so output written earlier in the same top-level command (not streamed yet) also lands in the log, and output under the redirect reaches a streaming caller only when that command ends. `exec N>&-` closes fd 1 or 2 for the rest of the script: later writes to it are dropped, the command reports `write error: Bad file descriptor` naming itself, and its status is 1. An `exec` redirect inside a subshell applies to the subshell's own output, routed at the subshell boundary; output the subshell wrote before the `exec` still reaches the caller. Inside a `bash -c` child it still does not take effect. `exec &>file` and `exec >&word` are ignored | True process replace would break sandbox containment; there is no real fd 1 to swap, the shell's output is a value returned to the caller | TM-ESC-005, `exec-redirect-shell-output.test.sh` | +| L-PROC-001 | `exec` does not replace the process; `exec cmd` runs cmd then stops execution. fd redirects work; `exec >log 2>&1` routes the shell's output per top-level command (each element of a top-level `;`/`&&`/`||` list counts as one), so output written earlier in the same top-level command (not streamed yet) also lands in the log, and buffered commands route redirected output when that command ends; live commands and streaming utilities route each write. `exec N>&-` closes fd 1 or 2 for the rest of the script: later writes to it are dropped, the command reports `write error: Bad file descriptor` naming itself, and its status is 1. An `exec` redirect inside a subshell applies to the subshell's own output, routed at the subshell boundary; output the subshell wrote before the `exec` still reaches the caller. Inside a `bash -c` child it still does not take effect. `exec &>file` and `exec >&word` are ignored | True process replace would break sandbox containment; there is no real fd 1 to swap, the shell's output is a value returned to the caller | TM-ESC-005, `exec-redirect-shell-output.test.sh` | | L-PROC-002 | Background jobs run concurrently (`jobs`, `wait -n`, `kill`, `ps`, `pgrep`), but cannot be stopped or resumed: `kill -STOP/-CONT` are ignored, there is no `suspend`/Ctrl-Z, `bg` is a no-op and `fg` just waits; jobs end when `exec()` returns | No terminal process groups in a virtual shell; a job must not outlive the call that owns it (TM-DOS-122) | `l_proc_002_no_job_control` | | L-PROC-003 | No process spawning; external commands run as builtins. bash's fork suppression is modeled only for `$SHLVL` (see `NoforkScope` in the interpreter): a `bash`/`sh` child bash would exec in place keeps the level, but another command run there still sees the unlowered value (bash's `bash -c 'printenv SHLVL'` prints one less than the child's level), nor do `env`/`timeout`/`xargs` starting a shell, and a `SHLVL=n` prefix on the in-place command counts | Core sandbox model: no fork/exec escape surface | `l_proc_003_no_process_spawning` | | L-PROC-005 | Coprocesses run concurrently with bash's fd numbers and cleanup, with these gaps: a coproc fd as a compound command's stdin (`while read l; do ...; done <&${C[0]}`) is read to end of input up front; `read -n`/`-N` on a coproc fd take a whole line (the rest of it is lost); writing to a coproc that exited is silently dropped (bash takes SIGPIPE); a finished coproc stays (NAME set, output readable) until `wait` reaps it, where bash reaps on SIGCHLD; coprocs end with the `exec()` call; an empty `<&"${C[0]}"` reports `ambiguous redirect` without the quotes where bash says `Bad file descriptor` | Deterministic, sandboxed job model (no SIGCHLD, no process outliving the call) | `coproc_tests.rs` | | L-PROC-004 | `bash`/`sh` child shells nest at most 8 deep (and count against `max_function_depth`); deeper nesting fails `maximum nesting depth exceeded` | Child shells run in-process on one stack; unbounded nesting crashed the process (TM-DOS-125) | `l_proc_004_child_shell_depth` | | L-HIST-001 | `fc` with no `-l`/`-s` (and `fc -e EDITOR`) fails `fc: editing history is not supported in bashkit` instead of opening an editor on the selected commands; `fc -s` and `fc -e -` rerun them | No terminal and no external editor process in the sandbox | `history-fc.test.sh` | -| L-PIPE-001 | Pipeline stages stream from the first stage that runs shell code (loop, group, function, `eval`) or a registered live `Command`. Leading single-builtin stages run to completion first and hand over their whole output, so they never get SIGPIPE (`sort big \| head -1` gives `PIPESTATUS` `0 0`, bash `141 0`); `yes`, `seq`, plain `cat`, `grep` and `tr` are the exception and stream. Downstream of a streaming stage, a single builtin reads all its input before it starts, except registered live commands, `head`, `read`, plain `cat`, `grep` (not `-c/-l/-L/-q`, context, `-b`) and `tr` (not `-s`): `while :; do echo; done \| sed p \| head -1` runs until a limit. Pipes hold 4 KiB, not 64 KiB, so a finite loop writing more than that into an early-exiting reader exits 141 where bash would exit 0. Pipelines nested more than 4 subshells deep run their stages in sequence | Builtins return their output as one value, not a stream; streaming them needs a reader/writer `Context`. The smaller pipe bounds how far a producer runs ahead (each byte costs budgeted commands); the nesting cap bounds native stack (TM-DOS-124) | `l_pipe_001_stages_run_sequentially` | +| L-PIPE-001 | Pipeline stages stream from the first stage that runs shell code (loop, group, function, `eval`) or a registered live `Command`. Leading single-builtin stages run to completion first and hand over their whole output, so they never get SIGPIPE (`sort big \| head -1` gives `PIPESTATUS` `0 0`, bash `141 0`); `yes`, `seq`, plain `cat`, `grep` and `tr` are the exception and stream. Downstream of a streaming stage, a single builtin reads all its input before it starts, except registered live commands, `head`, `read`, plain `cat`, `grep` and `tr` (not `-s`); grep summary/context options can delay output, while `-q` still exits on the first matching record: `while :; do echo; done \| sed p \| head -1` runs until a limit. Pipes hold 4 KiB, not 64 KiB, so a finite loop writing more than that into an early-exiting reader exits 141 where bash would exit 0. Pipelines nested more than 4 subshells deep run their stages in sequence | Builtins return their output as one value, not a stream; streaming them needs a reader/writer `Context`. The smaller pipe bounds how far a producer runs ahead (each byte costs budgeted commands); the nesting cap bounds native stack (TM-DOS-124) | `l_pipe_001_stages_run_sequentially` | | L-SUDO-001 | `sudo` runs its command as the one sandbox user: `-u`, `-g`, `-i` change nothing, `whoami` still prints `sandbox`, and there is never a password prompt. `busybox APPLET` runs bashkit's builtin of that name, not BusyBox's own applet (flags and messages are bashkit's) | The sandbox has no privilege boundary to cross; the VFS has no permission enforcement (L-FS-002) | `sudo-busybox.test.sh` | | L-RAND-001 | `uuidgen` makes random (v4) UUIDs only (no `-t`, `-m`, `-s`); `openssl` implements only `rand` | Time/MAC-based UUIDs would leak host identity; a full TLS/crypto toolkit is out of scope | `uuidgen_time_based_unsupported` | | L-FS-001 | Symlinks are followed, but `..` after a linked directory resolves lexically (`/link/..` is the link's parent, the `cd -L` view), and `ln` without `-s` (and `link`) makes a symlink, not a hard link | The interpreter normalizes paths before the VFS sees them; the VFS has no inodes to share | `symlink.test.sh`, `ln_default_symbolic` | diff --git a/knowledge/security/security-testing.md b/knowledge/security/security-testing.md index 984cc1f9b..22232fee6 100644 --- a/knowledge/security/security-testing.md +++ b/knowledge/security/security-testing.md @@ -161,6 +161,17 @@ zero/three-byte admission configurations, session recovery, discarded redirects, and a two-byte alias/append comparison with real Bash. No amplified-output or host-exhaustion workload is required. +## Streaming boundary regressions + +`streaming_audit_tests` uses a producer emitting 1 KiB stack chunks, a 16 KiB +aggregate input budget, and a 64 KiB live-memory budget. It checks builtin +stream readers, delimiter-free records and whole-input collectors without host +exhaustion workloads, plus recovery after refusal. Gated producers prove +redirected filters deliver before producer completion. Saved-fd tests cover +stdout/stderr, ordering, output caps and failed sinks; byte substitution and +redirect permutations compare against real Bash. These complement the existing +`streaming_command_tests` cancellation/revocation and pipe backpressure tests. + ## Adding New Fail Points Add `fail_point!("module::function", |action| ...)` under diff --git a/knowledge/security/threat-model.md b/knowledge/security/threat-model.md index bfbc80dc0..557d570bb 100644 --- a/knowledge/security/threat-model.md +++ b/knowledge/security/threat-model.md @@ -1454,6 +1454,14 @@ This section maps former vulnerability IDs to the new threat ID scheme and track | 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** | +Streaming builtin pipe reads also follow TM-DOS-096: each read consumes the +shared input/work budget and polls request cancellation/deadlines. Delimited +record assembly, returned record copies, `cat` accumulation and `tr` whole-input +collection use capacity-owning buffers before growth (TM-DOS-103). Input +process-substitution captures keep their byte-storage leases attached to their +shared descriptors. `streaming_audit_tests` covers small-budget rejection and +recovery; bounded pipe queues do not imply that record storage is bounded. + 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 From a62e3609db97428e411a201d159385c79728a087 Mon Sep 17 00:00:00 2001 From: Mykhailo Chalyi Date: Sun, 18 Oct 2026 23:28:12 +0000 Subject: [PATCH 3/5] fix(streaming): avoid writes for empty sequence output --- crates/bashkit/src/builtins/seq.rs | 2 +- crates/bashkit/tests/integration/streaming_audit_tests.rs | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/bashkit/src/builtins/seq.rs b/crates/bashkit/src/builtins/seq.rs index 924549e41..1a823b7dd 100644 --- a/crates/bashkit/src/builtins/seq.rs +++ b/crates/bashkit/src/builtins/seq.rs @@ -319,7 +319,7 @@ impl Builtin for Seq { output.push('\n'); } if let Some(stream) = &stream { - if !stream.write(output.as_bytes()).await? { + if !output.is_empty() && !stream.write(output.as_bytes()).await? { return Ok(ExecResult::with_code("", 141)); } output.clear(); diff --git a/crates/bashkit/tests/integration/streaming_audit_tests.rs b/crates/bashkit/tests/integration/streaming_audit_tests.rs index f831eb0e5..7d1e32f03 100644 --- a/crates/bashkit/tests/integration/streaming_audit_tests.rs +++ b/crates/bashkit/tests/integration/streaming_audit_tests.rs @@ -267,4 +267,7 @@ async fn streaming_write_errors_remain_command_statuses() { .unwrap(); assert_eq!(result.exit_code, 0); assert!(result.stderr.is_empty()); + let result = bash.execute("seq 0 1>&-").capture(64).await.unwrap(); + assert_eq!(result.exit_code, 0); + assert!(result.stderr.is_empty()); } From b7fd4b4ca0f24180464434a1255aceb893035ae2 Mon Sep 17 00:00:00 2001 From: Mykhailo Chalyi Date: Sun, 18 Oct 2026 23:41:57 +0000 Subject: [PATCH 4/5] fix(streaming): drain completed background output before closing execution --- crates/bashkit/src/interpreter/mod.rs | 19 ++++++- .../integration/streaming_audit_tests.rs | 55 +++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index 407de802f..970764365 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -4035,7 +4035,22 @@ impl Interpreter { self.close_coprocs(); jobs.finish_all().await; let (out, err) = jobs.lock().take_finished_output(); - if let Ok(r) = &mut result { + // Jobs completing after the last command boundary still owe their + // output to live sinks. Drain before closing the execution writers; + // keep cleanup unconditional when delivery fails. + let emission = if result.is_ok() + && self.output_callback.is_some() + && (!out.is_empty() || !err.is_empty()) + { + self.maybe_emit_output(&out, &err, self.output_emit_count) + .await + .map(|_| ()) + } else { + Ok(()) + }; + if self.retain_streamed_output() + && let Ok(r) = &mut result + { r.stdout.append(&out); r.stderr.append(&err); } @@ -4043,7 +4058,7 @@ impl Interpreter { if let Some(error) = &self.memory_limit_error { Err(crate::error::Error::ResourceLimit(error.clone())) } else { - result + emission.and(result) } }; diff --git a/crates/bashkit/tests/integration/streaming_audit_tests.rs b/crates/bashkit/tests/integration/streaming_audit_tests.rs index 7d1e32f03..ac9b46ed9 100644 --- a/crates/bashkit/tests/integration/streaming_audit_tests.rs +++ b/crates/bashkit/tests/integration/streaming_audit_tests.rs @@ -153,6 +153,61 @@ async fn process_substitution_preserves_bytes_in_buffered_and_live_execution() { struct Gated(Arc); +struct DelayedOutput { + started: Arc, + release: Arc, +} + +#[async_trait] +impl Command for DelayedOutput { + async fn execute(&self, ctx: CommandContext<'_>) -> bashkit::Result { + ctx.stdout.write_all(b"first").await?; + self.started.notify_one(); + self.release.notified().await; + ctx.stdout.write_all(b"last\n").await?; + ctx.stderr.write_all(b"diagnostic\n").await?; + Ok(CommandStatus::success()) + } +} + +#[tokio::test] +async fn background_completion_drains_to_live_sinks_and_capture_once() { + let started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let mut bash = Bash::builder() + .command( + "delayed", + DelayedOutput { + started: started.clone(), + release: release.clone(), + }, + ) + .build(); + let (mut stdout, mut stderr) = (Vec::new(), Vec::new()); + let execution = async { + bash.execute("delayed &") + .stdout(&mut stdout) + .stderr(&mut stderr) + .capture(64) + .await + }; + let unblock = async { + started.notified().await; + release.notify_one(); + }; + let (result, ()) = tokio::time::timeout(std::time::Duration::from_secs(2), async { + tokio::join!(execution, unblock) + }) + .await + .unwrap(); + let result = result.unwrap(); + assert_eq!(result.exit_code, 0); + assert_eq!(stdout, b"firstlast\n"); + assert_eq!(stderr, b"diagnostic\n"); + assert_eq!(result.stdout.as_bytes(), stdout); + assert_eq!(result.stderr.as_bytes(), stderr); +} + #[async_trait] impl Command for Gated { async fn execute(&self, ctx: CommandContext<'_>) -> bashkit::Result { From c090e7576669b506b1ddba152abe2c379db0f8f4 Mon Sep 17 00:00:00 2001 From: Mykhailo Chalyi Date: Sun, 18 Oct 2026 23:41:57 +0000 Subject: [PATCH 5/5] chore(streaming): document final background output delivery --- docs/streaming.md | 4 ++++ knowledge/foundations/streaming-roadmap.md | 6 ++++++ 2 files changed, 10 insertions(+) diff --git a/docs/streaming.md b/docs/streaming.md index a62cb27be..2bf883e84 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -220,6 +220,10 @@ Process substitutions `<(...)` and `>(...)` preserve raw bytes, including NUL and invalid UTF-8, but still buffer rather than running both sides as live tasks. Command substitution `$(...)` remains a text boundary and removes NUL. +Background jobs remain scoped to one execution and buffer output until +completion. Final job output reaches live sinks before execution completes, +including when the script does not explicitly call `wait`. + ## Routing, errors and cancellation A command context provides its name, expanded arguments (`args` in Rust, diff --git a/knowledge/foundations/streaming-roadmap.md b/knowledge/foundations/streaming-roadmap.md index b84b40dd3..d735f199f 100644 --- a/knowledge/foundations/streaming-roadmap.md +++ b/knowledge/foundations/streaming-roadmap.md @@ -75,6 +75,12 @@ state on dropped futures. Async output emission stays boxed to respect the existing native debug-build stack budget. Live execution yields every 64 commands for fairness; buffered synchronous APIs retain their single-poll model. +Background output remains buffered until completion. Jobs finishing after the +last top-level command drain into live stdout/stderr before execution writers +close, including scripts with no explicit `wait`. Job cleanup runs even if that +delivery fails; explicit live execution does not copy delivered job bytes into +its discarded final engine result. + Streaming utilities and host commands share `CommandIo` output destinations. Their redirects open once; incremental writes, buffered option results and diagnostics use those same destinations. Saved original stdout/stderr copies