diff --git a/crates/bashkit/docs/threat-model.md b/crates/bashkit/docs/threat-model.md index dca9446fe..be1fb24af 100644 --- a/crates/bashkit/docs/threat-model.md +++ b/crates/bashkit/docs/threat-model.md @@ -114,6 +114,7 @@ through configurable limits. | sleep bypasses timeout (TM-DOS-057) | `sleep N` ignores `ExecutionLimits::timeout` | Host-backed timeout; non-JS wasm blocking sleep is clamped to the execution deadline | **MITIGATED** | | Interactive terminal flooding (TM-DOS-119) | Unbounded type-ahead, undrained output, or a `vi` buffer grown by pastes | Caps on unread input (1 MiB), retained output (4 MiB), scrollback (1000 rows), transcript (1 MiB), line length, `vi` buffer (8 MiB) and undo history; `:w` goes through VFS limits | MITIGATED | | Input wait vs. timeout (TM-DOS-120) | Abusing the keyboard-wait exclusion to run work past the deadline | Only time blocked reading the terminal is excluded; CPU work and `sleep` still count | MITIGATED | +| Concurrent pipeline amplification (TM-DOS-124) | An endless loop piped into `head`, a producer whose reader quit, or recursive pipelines `f() { f \| f; }` | 4 KiB pipes with backpressure; a writer whose reader is gone ends with 141 (SIGPIPE); streaming nesting counts toward `max_subshell_depth` and falls back to sequential stages past 4 levels; stages share the session budget, timeout and cancellation | MITIGATED | | Random generator amplification (TM-DOS-123) | `openssl rand` asked for gigabytes | Requests capped at 1 MiB; fixed-size `uuidgen`/`$SRANDOM` | MITIGATED | | Unbounded builtin output (TM-DOS-058) | `seq 1 1000000` produces 1M lines | Add `max_stdout_bytes` limit | **MITIGATED** | | Silent truncation at builtin caps (TM-DOS-109) | `seq 200000`, an awk loop past its cap, or an oversized `sprintf` expression returns incomplete output with exit 0 | Caps report `: limit () exceeded` on stderr and exit non-zero; awk caps and formatting errors are fatal | **MITIGATED** | diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index 323df5174..74a9ec813 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -15,6 +15,7 @@ mod brace_expansion; mod expansion; mod glob; mod jobs; +mod pipe; mod redirection; mod state; mod time_command; @@ -559,6 +560,111 @@ impl Drop for ExecutionExtensionsGuard { } } +/// Subshell depth above which pipelines run their stages one after another. +/// +/// THREAT[TM-DOS-124]: every concurrent level polls the next one from inside +/// its own poll, so nesting costs extra native stack (~25 KB per level in +/// debug builds). Past this depth (`f() { f | f; }`) stages fall back to running in +/// sequence, which recurses no deeper than a plain function call. +const MAX_STREAMING_NESTING: usize = 4; + +/// Append a stage's stderr to the pipeline's, within `max` bytes. +fn append_stage_stderr( + acc: &mut crate::StreamData, + truncated: &mut bool, + result: &ExecResult, + max: usize, +) { + if *truncated { + return; + } + let remaining = max.saturating_sub(acc.len()); + if result.stderr.len() <= remaining { + acc.append(&result.stderr); + *truncated = result.stderr_truncated; + } else { + acc.append(&result.stderr.prefix(remaining)); + *truncated = true; + } +} + +/// How much of a streaming stdin a command needs before it starts. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum StdinDemand { + /// Never reads stdin. + Nothing, + /// One line (`read`). + Line, + /// First N lines, then stops reading (`head -n N`). + Lines(usize), + /// First N bytes, then stops reading (`head -c N`). + Bytes(usize), + /// Everything up to end of input. + All, +} + +impl StdinDemand { + fn satisfied(self, buf: &[u8]) -> bool { + match self { + Self::Nothing => true, + Self::Line => buf.contains(&b'\n'), + Self::Lines(n) => buf.iter().filter(|&&b| b == b'\n').count() >= n, + Self::Bytes(n) => buf.len() >= n, + Self::All => false, + } + } +} + +fn stdin_demand(name: &str, args: &[String]) -> StdinDemand { + match name { + "echo" | "printf" | "true" | "false" | ":" | "exit" | "return" | "break" | "continue" + | "sleep" | "test" | "[" | "cd" | "export" | "local" | "declare" | "set" | "unset" + | "shift" | "" => StdinDemand::Nothing, + "read" + if !args + .iter() + .any(|a| a.starts_with('-') && (a.contains('d') || a.contains('N'))) => + { + StdinDemand::Line + } + "head" => head_demand(args), + _ => StdinDemand::All, + } +} + +/// `head [-n N | -c N | -N]` with no file operands; anything else reads all. +fn head_demand(args: &[String]) -> StdinDemand { + let mut demand = StdinDemand::Lines(10); + let mut it = args.iter(); + while let Some(arg) = it.next() { + let (lines, value) = match arg.as_str() { + "-n" | "--lines" => (true, it.next().map(String::as_str)), + "-c" | "--bytes" => (false, it.next().map(String::as_str)), + "-q" | "-v" | "--quiet" | "--silent" | "--verbose" => continue, + a if a.starts_with("--lines=") => (true, Some(&a[8..])), + a if a.starts_with("--bytes=") => (false, Some(&a[8..])), + a if a.starts_with("-n") => (true, Some(&a[2..])), + a if a.starts_with("-c") => (false, Some(&a[2..])), + a if a.len() > 1 + && a.starts_with('-') + && a[1..].bytes().all(|b| b.is_ascii_digit()) => + { + (true, Some(&a[1..])) + } + _ => return StdinDemand::All, + }; + let Some(n) = value.and_then(|v| v.parse::().ok()) else { + return StdinDemand::All; + }; + demand = if lines { + StdinDemand::Lines(n) + } else { + StdinDemand::Bytes(n) + }; + } + demand +} + #[cfg(test)] mod execution_extensions_guard_tests { use super::*; @@ -1182,6 +1288,24 @@ struct ScopedState { /// (`Arc::make_mut`). For substitutions that don't mutate state at all (the /// common case — `$(echo $x)`, command queries) this saves an entire deep /// HashMap clone per substitution. +/// Results gathered while a pipeline runs (see `execute_pipeline`). +#[derive(Default)] +struct PipelineAcc { + statuses: Vec, + stderr: crate::StreamData, + stderr_truncated: bool, + last: ExecResult, +} + +/// Subshell state a pipeline stage rolls back. +type StageSnapshot = (SubshellSnapshot, Vec, HashMap>); + +/// State held across one pipeline stage (see `enter_pipeline_stage`). +struct PipelineStageScope { + saved: Option, + prev_pipeline_stdin: Option>, +} + struct SubshellSnapshot { scoped: ScopedState, flags: BashFlags, @@ -1265,6 +1389,12 @@ pub struct Interpreter { /// Stdin inherited from pipeline for compound commands (while read, etc.) /// Each read operation consumes one line, advancing through the data. pipeline_stdin: Option, + /// Streaming stdin from a concurrent pipeline stage; read into + /// `pipeline_stdin` on demand (see `fill_stdin_from_pipe`). + pipe_in: Option>, + /// Stdout pipe of this forked pipeline stage: command boundaries wait + /// here for room and stop with 141 once the reader is gone. + pipe_out: Option>, /// Position within the current argument while `getopts` walks a clustered /// short-option group (e.g. `-abc`). Interpreter-internal working state for /// `execute_getopts`; `0` means "at the start of the next option group". @@ -1788,6 +1918,8 @@ impl Interpreter { #[cfg(feature = "ssh")] ssh_client: None, pipeline_stdin: None, + pipe_in: None, + pipe_out: None, getopts_char_idx: 0, last_bg_pid: None, output_callback: None, @@ -1938,6 +2070,8 @@ impl Interpreter { #[cfg(feature = "ssh")] ssh_client: self.ssh_client.clone(), pipeline_stdin: None, + pipe_in: None, + pipe_out: None, getopts_char_idx: self.getopts_char_idx, last_bg_pid: self.last_bg_pid.clone(), output_callback: None, @@ -2823,7 +2957,12 @@ impl Interpreter { .limits .max_stderr_bytes .saturating_sub(self.output_stream_stderr_bytes); - let stdout_chunk = stdout.prefix(stdout_remaining); + // A stage's pipe is not caller-visible output: no stdout cap. + let stdout_chunk = if self.pipe_out.is_some() { + stdout.clone() + } else { + stdout.prefix(stdout_remaining) + }; let stderr_chunk = stderr.prefix(stderr_remaining); if stdout_chunk.is_empty() && stderr_chunk.is_empty() { return false; @@ -3198,6 +3337,12 @@ impl Interpreter { self.charge_command_execution()?; + if self.pipe_out.is_some() + && let Some(killed) = Box::pin(self.wait_for_pipe_room()).await + { + return Ok(killed); + } + match command { Command::Simple(simple) => { // One Result local (not `?` + a second ExecResult copy): @@ -5390,25 +5535,34 @@ impl Interpreter { } /// Execute a pipeline (cmd1 | cmd2 | cmd3) + /// + /// Leading stages that are single commands run first, one after another, + /// handing over complete output. From the first stage that runs shell + /// code (a loop, group, function), the rest run concurrently over + /// bounded pipes (`execute_streaming_stages`), unless concurrency is off. async fn execute_pipeline(&mut self, pipeline: &Pipeline) -> Result { + // Kept small: recursive pipelines (`f() { f | f; }`) nest this poll + // frame once per level, so bookkeeping lives in sync helpers. + let (stream_from, lastpipe) = self.plan_pipeline(pipeline); + let mut acc = PipelineAcc::default(); let mut stdin_data: Option = None; - let mut last_result = ExecResult::ok(String::new()); - let mut pipeline_stderr = crate::StreamData::new(); - let mut pipeline_stderr_truncated = false; - let mut pipe_statuses = Vec::new(); - - // Every stage of a multi-command pipeline runs in a subshell, except - // the last one under `shopt -s lastpipe` (bash, job control off). - let multi = pipeline.commands.len() > 1; - let lastpipe = self - .scoped - .variables - .get("SHOPT_lastpipe") - .is_some_and(|v| v == "1"); + let count = pipeline.commands.len(); for (i, command) in pipeline.commands.iter().enumerate() { - let is_last = i == pipeline.commands.len() - 1; - let subshell = multi && !(is_last && lastpipe); + if stream_from == Some(i) { + let stages = self + .execute_streaming_group(&pipeline.commands[i..], stdin_data.take(), lastpipe) + .await?; + for result in stages { + self.absorb_stage(&mut acc, result, true); + } + break; + } + let is_last = i == count - 1; + // Every stage of a multi-command pipeline runs in a subshell, + // except the last one under `shopt -s lastpipe` (bash, job + // control off). + let subshell = count > 1 && !(is_last && lastpipe); // A non-last stage's stdout feeds the next stage, never the // streaming observer. let saved_callback = if is_last { @@ -5416,37 +5570,100 @@ impl Interpreter { } else { self.output_callback.take() }; - let result = self - .execute_pipeline_stage(command, stdin_data.take(), subshell) - .await; + let result = match self.enter_pipeline_stage(command, stdin_data.take(), subshell) { + Ok((scope, stdin)) => { + let result = match command { + Command::Simple(simple) => self.execute_simple_command(simple, stdin).await, + _ => self.execute_command(command).await, + }; + self.exit_pipeline_stage(scope, result) + } + Err(e) => Err(e), + }; if let Some(cb) = saved_callback { self.output_callback = Some(cb); } - let result = result?; + stdin_data = self.absorb_stage(&mut acc, result?, is_last); + } + Ok(self.finish_pipeline(pipeline, acc)) + } - pipe_statuses.push(result.exit_code); - if !pipeline_stderr_truncated { - let remaining = self - .limits - .max_stderr_bytes - .saturating_sub(pipeline_stderr.len()); - if result.stderr.len() <= remaining { - pipeline_stderr.append(&result.stderr); - pipeline_stderr_truncated = result.stderr_truncated; - } else { - pipeline_stderr.append(&result.stderr.prefix(remaining)); - pipeline_stderr_truncated = true; - } - } + /// Where streaming starts (first stage that runs shell code), and lastpipe. + #[inline(never)] + fn plan_pipeline(&self, pipeline: &Pipeline) -> (Option, bool) { + let count = pipeline.commands.len(); + let lastpipe = self + .scoped + .variables + .get("SHOPT_lastpipe") + .is_some_and(|v| v == "1"); + let stream_from = if count > 1 + && self.concurrent_jobs + && self.counters.subshell_depth < MAX_STREAMING_NESTING + { + pipeline.commands[..count - 1] + .iter() + .position(|c| !self.is_single_command_stage(c)) + } else { + None + }; + (stream_from, lastpipe) + } - if is_last { - last_result = result; - } else { - stdin_data = Some(result.stdout); - } + /// Run a concurrent tail of a pipeline as one subshell nesting level. + fn execute_streaming_group<'a>( + &'a mut self, + commands: &'a [Command], + stdin: Option, + lastpipe: bool, + ) -> std::pin::Pin>> + Send + 'a>> + { + Box::pin(async move { + // THREAT[TM-DOS-124]: concurrent stages nest like subshells + // (`f() { f | f; }`), so the whole group counts as one level. + self.counters.push_subshell(&self.limits)?; + let stages = self + .execute_streaming_stages(commands, stdin, lastpipe) + .await; + self.counters.pop_subshell(); + stages + }) + } + + /// Record a finished stage; returns its stdout as the next stage's stdin + /// (the last stage's stdout stays in the accumulator). + #[inline(never)] + fn absorb_stage( + &self, + acc: &mut PipelineAcc, + mut result: ExecResult, + is_last: bool, + ) -> Option { + acc.statuses.push(result.exit_code); + append_stage_stderr( + &mut acc.stderr, + &mut acc.stderr_truncated, + &result, + self.limits.max_stderr_bytes, + ); + if is_last { + acc.last = result; + return None; } - last_result.stderr = pipeline_stderr; - last_result.stderr_truncated |= pipeline_stderr_truncated; + Some(std::mem::take(&mut result.stdout)) + } + + /// PIPESTATUS, pipefail and `!` for a finished pipeline. + #[inline(never)] + fn finish_pipeline(&mut self, pipeline: &Pipeline, acc: PipelineAcc) -> ExecResult { + let PipelineAcc { + statuses: pipe_statuses, + stderr, + stderr_truncated, + last: mut last_result, + } = acc; + last_result.stderr = stderr; + last_result.stderr_truncated |= stderr_truncated; // Store PIPESTATUS array self.pipestatus = pipe_statuses.clone(); @@ -5467,20 +5684,216 @@ impl Interpreter { if pipeline.negated { last_result.exit_code = if last_result.exit_code == 0 { 1 } else { 0 }; } + last_result + } - Ok(last_result) + /// A stage that is one builtin call produces its whole output at once, + /// so running it before the next stage loses nothing. Anything that can + /// run shell code (loops, groups, functions, `eval`, nested shells) + /// streams instead. + fn is_single_command_stage(&self, command: &Command) -> bool { + let Command::Simple(simple) = command else { + return false; + }; + let name = match simple.name.parts.as_slice() { + [] => return true, + [WordPart::Literal(name)] => name.as_str(), + _ => return false, + }; + !matches!(name, "eval" | "source" | "." | "bash" | "sh") + && !self.scoped.functions.contains_key(name) + && !self.scoped.aliases.contains_key(name) } - /// Run one pipeline stage with `stdin` from the previous stage. - /// - /// With `subshell`, state changes (variables, cwd, options, fds) are - /// rolled back afterwards and `exit`/`return` end only the stage. - async fn execute_pipeline_stage( + /// Run `commands` (the tail of a pipeline) concurrently: every stage but + /// the last on a forked shell writing into a bounded [`pipe::Pipe`], the + /// last one here (a subshell unless `lastpipe`) so its output streams. + /// A stage whose reader went away ends with 141 (SIGPIPE). Returns one + /// result per stage; stdout is set only on the last. + async fn execute_streaming_stages( + &mut self, + commands: &[Command], + stdin: Option, + lastpipe: bool, + ) -> Result> { + type StageFuture = std::pin::Pin< + Box> + Send + 'static>, + >; + let (last, producers) = commands.split_last().expect("pipeline tail"); + let mut input: Option> = None; + let mut stages: Vec = Vec::with_capacity(producers.len()); + for (k, command) in producers.iter().enumerate() { + let pipe = pipe::Pipe::new(); + let mut child = self.fork_for_job(); + if k == 0 { + // First stage reads what this shell would read. + child.pipeline_stdin = stdin.clone().or_else(|| self.pipeline_stdin.clone()); + if stdin.is_none() { + child.pipe_in = self.pipe_in.clone(); + } + } else { + child.pipe_in = input.clone(); + } + let sink = Arc::clone(&pipe); + child.output_callback = Some(Box::new(move |out, _err| { + sink.write(out.as_bytes()); + })); + child.pipe_out = Some(Arc::clone(&pipe)); + let write_end = pipe::WriteEnd(Arc::clone(&pipe)); + let read_end = input.take().map(pipe::ReadEnd); + let command = command.clone(); + stages.push(Box::pin(async move { + let _read_end = read_end; + let jobs = Arc::clone(&child.jobs); + let result = jobs::with_jobs(&jobs, child.execute_command(&command)).await; + jobs.finish_all().await; + let (out, err) = jobs.lock().take_finished_output(); + let mut r = result?; + r.stdout.append(&out); + r.stderr.append(&err); + // Output not yet sent at a command boundary goes now. + let sent = child.output_stream_stdout_bytes; + if r.stdout.len() > sent { + write_end.0.write(&r.stdout.as_bytes()[sent..]); + } + if write_end.0.is_broken() { + r.exit_code = 141; + } else if let ControlFlow::Exit(code) | ControlFlow::Return(code) = r.control_flow { + r.exit_code = code; + } + r.control_flow = ControlFlow::None; + r.stdout = crate::StreamData::new(); + Ok(r) + })); + input = Some(pipe); + } + + // The last stage reads the last pipe through `pipe_in`. + let last_pipe = input.expect("at least one producer"); + let saved_pipe_in = self.pipe_in.replace(Arc::clone(&last_pipe)); + let saved_stdin = self.pipeline_stdin.take(); + let mut read_end = Some(pipe::ReadEnd(last_pipe)); + let mut done: Vec>> = (0..stages.len()).map(|_| None).collect(); + let mut last_done: Option> = None; + { + let mut last_fut = Box::pin(async { + let (scope, stdin) = self.enter_pipeline_stage(last, None, !lastpipe)?; + let result = match last { + Command::Simple(simple) => self.execute_simple_command(simple, stdin).await, + _ => self.execute_command(last).await, + }; + self.exit_pipeline_stage(scope, result) + }); + // Readers first (last stage, then upstream), so each blocks on + // its pipe before the writer feeding it runs. + std::future::poll_fn(|cx| { + if last_done.is_none() + && let std::task::Poll::Ready(r) = last_fut.as_mut().poll(cx) + { + // Nobody reads any more: writers get SIGPIPE. + read_end = None; + last_done = Some(r); + cx.waker().wake_by_ref(); + } + for (stage, slot) in stages.iter_mut().zip(done.iter_mut()).rev() { + if slot.is_none() + && let std::task::Poll::Ready(r) = stage.as_mut().poll(cx) + { + *slot = Some(r); + } + } + let failed = done.iter().any(|d| matches!(d, Some(Err(_)))) + || matches!(last_done, Some(Err(_))); + if failed || (last_done.is_some() && done.iter().all(Option::is_some)) { + std::task::Poll::Ready(()) + } else { + std::task::Poll::Pending + } + }) + .await; + } + drop(read_end); + drop(stages); + self.pipe_in = saved_pipe_in; + self.pipeline_stdin = saved_stdin; + + let mut results = Vec::with_capacity(done.len() + 1); + for d in done { + results.push(d.unwrap_or_else(|| Ok(ExecResult::ok(String::new())))?); + } + results.push(last_done.unwrap_or_else(|| Ok(ExecResult::ok(String::new())))?); + Ok(results) + } + + /// At a command boundary in a forked pipeline stage: wait while the + /// pipe is full; once a write found the reader gone, end the stage + /// with 141 like SIGPIPE. + async fn wait_for_pipe_room(&mut self) -> Option { + let pipe = Arc::clone(self.pipe_out.as_ref()?); + if pipe.is_broken() { + return Some(ExecResult { + exit_code: 141, + control_flow: ControlFlow::Exit(141), + ..Default::default() + }); + } + pipe.writable().await; + None + } + + /// Before command `name` takes its stdin from a streaming pipe, pull what + /// it needs into `pipeline_stdin`: one line for `read`, the requested + /// lines/bytes for `head` (then close the pipe so the writer gets + /// SIGPIPE), nothing for commands that never read stdin, and everything + /// up to end of input otherwise. + async fn fill_stdin_from_pipe(&mut self, name: &str, args: &[String]) { + let Some(pipe) = self.pipe_in.clone() else { + return; + }; + let want = stdin_demand(name, args); + if want == StdinDemand::Nothing { + return; + } + let mut buf = self.pipeline_stdin.take().unwrap_or_default().into_bytes(); + let mut eof = false; + while !want.satisfied(&buf) { + let chunk = if matches!(want, StdinDemand::All | StdinDemand::Bytes(_)) { + pipe.read_some().await + } else { + pipe.read_line_chunk().await + }; + if chunk.is_empty() { + eof = true; + break; + } + buf.extend_from_slice(&chunk); + } + if eof || matches!(want, StdinDemand::All) { + self.pipe_in = None; + } else if matches!(want, StdinDemand::Lines(_) | StdinDemand::Bytes(_)) { + // `head` exits after its input: the writer sees a closed pipe. + pipe.close_read(); + self.pipe_in = None; + } + self.pipeline_stdin = Some(buf.into()); + } + + /// Start a pipeline stage: with `subshell`, snapshot state so changes + /// (variables, cwd, options, fds) roll back in `exit_pipeline_stage` and + /// `exit`/`return` end only the stage. The caller awaits the stage's + /// command itself (`execute_simple_command` / `execute_command`, both + /// boxed): recursive pipelines (`f() { f | f; }`) then add no extra poll + /// frame per level. + fn enter_pipeline_stage( &mut self, command: &Command, stdin: Option, subshell: bool, - ) -> Result { + ) -> Result<(PipelineStageScope, Option)> { + if matches!(command, Command::Simple(_)) { + self.execution_budget.consume_work(1)?; + self.counters.tick_command(&self.limits)?; + } let saved = subshell.then(|| { ( self.snapshot_subshell_state(), @@ -5488,27 +5901,28 @@ impl Interpreter { self.coproc_buffers.clone(), ) }); - let result = match command { - Command::Simple(simple) => { - let ticked = self - .execution_budget - .consume_work(1) - .and_then(|()| self.counters.tick_command(&self.limits)); - match ticked { - Ok(()) => self.execute_simple_command(simple, stdin).await, - Err(e) => Err(e.into()), - } - } - _ => { - // Compound commands, lists, etc. in pipeline: - // set pipeline_stdin so inner commands (read, cat, etc.) can consume it - let prev_pipeline_stdin = std::mem::replace(&mut self.pipeline_stdin, stdin); - let result = self.execute_command(command).await; - self.pipeline_stdin = prev_pipeline_stdin; - result - } + let mut scope = PipelineStageScope { + saved, + prev_pipeline_stdin: None, }; - let Some((snap, call_stack, coproc)) = saved else { + if matches!(command, Command::Simple(_)) { + return Ok((scope, stdin)); + } + // Compound commands, lists, etc. in pipeline: + // set pipeline_stdin so inner commands (read, cat, etc.) can consume it + scope.prev_pipeline_stdin = Some(std::mem::replace(&mut self.pipeline_stdin, stdin)); + Ok((scope, None)) + } + + fn exit_pipeline_stage( + &mut self, + scope: PipelineStageScope, + result: Result, + ) -> Result { + if let Some(prev) = scope.prev_pipeline_stdin { + self.pipeline_stdin = prev; + } + let Some((snap, call_stack, coproc)) = scope.saved else { return result; }; self.restore_subshell_state(snap); @@ -6453,6 +6867,10 @@ impl Interpreter { stdin }; + if stdin.is_none() && self.pipe_in.is_some() { + Box::pin(self.fill_stdin_from_pipe(name, &args)).await; + } + // If no explicit stdin, inherit from pipeline_stdin (for compound cmds in pipes). // For `read`, consume one line; for other commands, provide all remaining data. let stdin = if stdin.is_some() { diff --git a/crates/bashkit/src/interpreter/pipe.rs b/crates/bashkit/src/interpreter/pipe.rs new file mode 100644 index 000000000..e03dd47d4 --- /dev/null +++ b/crates/bashkit/src/interpreter/pipe.rs @@ -0,0 +1,249 @@ +//! In-process pipe between concurrent pipeline stages. +//! +//! Important decision: a pipe is a shared byte buffer, not a channel of +//! chunks. The writing stage appends whole command outputs at command +//! boundaries (overshooting the capacity is allowed, so a single builtin's +//! output is never split) and then waits at its next command until the buffer +//! drops below [`PIPE_CAPACITY`]. The reading stage pulls bytes on demand. +//! Closing the read end makes the next append fail, which the writer turns +//! into SIGPIPE (exit 141), like a write to a pipe with no readers. +//! +//! Both ends are polled on the same task (see `execute_pipeline`), so there +//! are no threads: each side keeps one waker slot. + +use std::future::poll_fn; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::task::{Context, Poll, Waker}; + +/// Bytes buffered before a writer waits. +/// +/// Linux buffers 64 KiB, but every byte a producer runs ahead costs +/// commands from the execution budget: an endless `while :; do echo x; done` +/// would spend ~64K commands before `head` could stop it. 4 KiB (`PIPE_BUF`) +/// keeps that waste small; the only visible effect is that a finite +/// producer writing between 4 and 64 KiB into a reader that quits early +/// exits 141 where bash would usually exit 0. +pub(crate) const PIPE_CAPACITY: usize = 4 * 1024; + +#[derive(Default)] +struct State { + buf: Vec, + /// Bytes at the start of `buf` already consumed. + pos: usize, + write_closed: bool, + read_closed: bool, + /// A write found the read end closed. + broken: bool, + read_waker: Option, + write_waker: Option, +} + +impl State { + fn available(&self) -> usize { + self.buf.len() - self.pos + } + + fn wake_reader(&mut self) { + if let Some(w) = self.read_waker.take() { + w.wake(); + } + } + + fn wake_writer(&mut self) { + if let Some(w) = self.write_waker.take() { + w.wake(); + } + } + + fn take(&mut self, n: usize) -> Vec { + let end = self.pos + n.min(self.available()); + let out = self.buf[self.pos..end].to_vec(); + self.pos = end; + if self.pos == self.buf.len() { + self.buf.clear(); + self.pos = 0; + } + self.wake_writer(); + out + } +} + +/// Shared pipe state; ends are tracked with [`WriteEnd`] / [`ReadEnd`]. +#[derive(Default)] +pub(crate) struct Pipe { + state: Mutex, +} + +impl Pipe { + pub(crate) fn new() -> Arc { + Arc::new(Self::default()) + } + + fn lock(&self) -> MutexGuard<'_, State> { + self.state.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Append `data`. Returns false (and marks the pipe broken) when nobody + /// reads any more. + pub(crate) fn write(&self, data: &[u8]) -> bool { + if data.is_empty() { + return true; + } + let mut s = self.lock(); + if s.read_closed { + s.broken = true; + return false; + } + s.buf.extend_from_slice(data); + s.wake_reader(); + true + } + + /// A write hit a closed read end. + pub(crate) fn is_broken(&self) -> bool { + self.lock().broken + } + + /// Wait until the buffer has room or the reader is gone. + pub(crate) async fn writable(&self) { + poll_fn(|cx: &mut Context<'_>| { + let mut s = self.lock(); + if s.read_closed || s.available() < PIPE_CAPACITY { + Poll::Ready(()) + } else { + s.write_waker = Some(cx.waker().clone()); + Poll::Pending + } + }) + .await + } + + pub(crate) fn close_write(&self) { + let mut s = self.lock(); + s.write_closed = true; + s.wake_reader(); + } + + pub(crate) fn close_read(&self) { + let mut s = self.lock(); + s.read_closed = true; + s.buf.clear(); + s.pos = 0; + s.wake_writer(); + } + + /// Wait for data or end of input; returns the available bytes, empty at + /// end of input. + pub(crate) async fn read_some(&self) -> Vec { + self.read_chunk(false).await + } + + /// Like [`read_some`](Self::read_some), but stops after the first + /// newline, leaving the rest in the pipe (and the writer paused). + pub(crate) async fn read_line_chunk(&self) -> Vec { + self.read_chunk(true).await + } + + async fn read_chunk(&self, line: bool) -> Vec { + poll_fn(|cx: &mut Context<'_>| { + let mut s = self.lock(); + let mut n = s.available(); + if line && let Some(i) = s.buf[s.pos..].iter().position(|&b| b == b'\n') { + n = i + 1; + } + if n > 0 { + Poll::Ready(s.take(n)) + } else if s.write_closed { + Poll::Ready(Vec::new()) + } else { + s.read_waker = Some(cx.waker().clone()); + Poll::Pending + } + }) + .await + } + + /// Read everything until the writer closes. + #[cfg(test)] + pub(crate) async fn read_to_end(&self) -> Vec { + let mut out = Vec::new(); + loop { + let chunk = self.read_some().await; + if chunk.is_empty() { + return out; + } + out.extend_from_slice(&chunk); + } + } +} + +/// Write end owned by a producing stage; closes on drop so the reader sees +/// end of input even if the stage is cancelled. +pub(crate) struct WriteEnd(pub(crate) Arc); + +impl Drop for WriteEnd { + fn drop(&mut self) { + self.0.close_write(); + } +} + +/// Read end owned by a consuming stage; closes on drop so the writer gets +/// SIGPIPE on its next write. +pub(crate) struct ReadEnd(pub(crate) Arc); + +impl Drop for ReadEnd { + fn drop(&mut self) { + self.0.close_read(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::future::Future; + + #[tokio::test] + async fn read_sees_writes_then_eof() { + let p = Pipe::new(); + assert!(p.write(b"ab")); + assert!(p.write(b"c")); + p.close_write(); + assert_eq!(p.read_to_end().await, b"abc"); + assert!(p.read_some().await.is_empty()); + } + + #[tokio::test] + async fn line_reads_leave_the_rest() { + let p = Pipe::new(); + p.write(b"a\nb\nc"); + p.close_write(); + assert_eq!(p.read_line_chunk().await, b"a\n"); + assert_eq!(p.read_line_chunk().await, b"b\n"); + assert_eq!(p.read_line_chunk().await, b"c"); + assert!(p.read_line_chunk().await.is_empty()); + } + + #[tokio::test] + async fn write_after_reader_closed_breaks() { + let p = Pipe::new(); + assert!(p.write(b"x")); + drop(ReadEnd(Arc::clone(&p))); + assert!(!p.is_broken()); + assert!(!p.write(b"y")); + assert!(p.is_broken()); + // An empty write is not a write. + assert!(p.write(b"")); + } + + #[tokio::test] + async fn writer_waits_until_drained() { + let p = Pipe::new(); + p.write(&vec![b'a'; PIPE_CAPACITY]); + let mut wait = Box::pin(p.writable()); + let waker = futures_util::task::noop_waker(); + let mut cx = Context::from_waker(&waker); + assert!(wait.as_mut().poll(&mut cx).is_pending()); + assert_eq!(p.read_some().await.len(), PIPE_CAPACITY); + assert!(wait.as_mut().poll(&mut cx).is_ready()); + } +} diff --git a/crates/bashkit/src/lib.rs b/crates/bashkit/src/lib.rs index b63bcb88c..127cdf194 100644 --- a/crates/bashkit/src/lib.rs +++ b/crates/bashkit/src/lib.rs @@ -3398,6 +3398,11 @@ impl BashBuilder { /// deterministic (eval replay, snapshot tests). See /// `knowledge/foundations/parallel-execution.md` ("Background jobs"). /// + /// The same switch makes pipeline stages that run shell code stream into + /// each other concurrently, so `while :; do echo x; done | head -1` ends + /// the loop with SIGPIPE (141); with `false` stages run one after another + /// ("Pipelines"). + /// /// ```rust /// # use bashkit::Bash; /// # #[tokio::main] diff --git a/crates/bashkit/tests/integration/limitations_evidence_tests.rs b/crates/bashkit/tests/integration/limitations_evidence_tests.rs index fa07b514e..6ddd53973 100644 --- a/crates/bashkit/tests/integration/limitations_evidence_tests.rs +++ b/crates/bashkit/tests/integration/limitations_evidence_tests.rs @@ -165,8 +165,8 @@ async fn l_term_002_cat_does_not_wait_for_terminal_input() { ); } -/// L-PIPE-001: stages run one after another, so the producer runs to its own -/// cap before `head` reads, and never sees SIGPIPE. +/// L-PIPE-001: a leading single-builtin stage runs to its own cap before +/// `head` reads, and never sees SIGPIPE. #[tokio::test] async fn l_pipe_001_stages_run_sequentially() { let mut bash = Bash::new(); diff --git a/crates/bashkit/tests/integration/pipeline_tests.rs b/crates/bashkit/tests/integration/pipeline_tests.rs index fc8838a64..e42c8b7b4 100644 --- a/crates/bashkit/tests/integration/pipeline_tests.rs +++ b/crates/bashkit/tests/integration/pipeline_tests.rs @@ -46,3 +46,78 @@ async fn last_stage_output_streams() { assert_eq!(result.stdout, "go-1\ngo-2\n"); assert_eq!(chunks.lock().unwrap().concat(), "go-1\ngo-2\n"); } + +/// A compound producer streams into the next stage through a bounded pipe; +/// large output arrives intact and in order. +#[tokio::test] +async fn large_producer_output_is_intact() { + let mut bash = Bash::new(); + let result = bash + .exec("{ seq 1 100000; seq 1 100000; } | wc -l; { seq 1 3; echo err >&2; } | tail -1") + .await + .unwrap(); + assert_eq!(result.stdout, "200000\n3\n"); + assert_eq!(result.stderr, "err\n"); +} + +/// Stages of a function producer run as a forked stage: small output finishes +/// before `head` reads, so the producer exits 0 like bash usually does. +#[tokio::test] +async fn finite_producer_is_not_killed() { + let mut bash = Bash::new(); + let result = bash + .exec("f() { for i in 1 2 3; do echo $i; done; }; f | head -1; echo \"${PIPESTATUS[*]}\"") + .await + .unwrap(); + assert_eq!(result.stdout, "1\n0 0\n"); +} + +/// With concurrent jobs off, stages keep running one after another. +#[tokio::test] +async fn sequential_mode_keeps_buffered_stages() { + let mut bash = Bash::builder().concurrent_jobs(false).build(); + let result = bash + .exec("for i in 1 2 3; do echo $i; done | tac") + .await + .unwrap(); + assert_eq!(result.stdout, "3\n2\n1\n"); +} + +/// TM-DOS-124: a stage that never reads ends an endless producer; nothing +/// buffers without bound. +#[tokio::test] +async fn threat_endless_producer_into_non_reader_ends() { + let mut bash = Bash::new(); + let result = bash + .exec("while :; do echo x; done | true; echo \"${PIPESTATUS[*]}\"") + .await + .unwrap(); + assert_eq!(result.stdout, "141 0\n"); +} + +/// TM-DOS-124: recursive pipelines stop at a depth limit instead of +/// forking without bound. Concurrent nesting is capped (deeper levels run +/// stages in sequence), so the stack cost stays near plain function +/// recursion; the test thread gets headroom for unoptimized test builds. +#[test] +fn threat_recursive_pipeline_is_bounded() { + let handle = std::thread::Builder::new() + .stack_size(8 * 1024 * 1024) + .spawn(|| { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let mut bash = Bash::new(); + bash.exec("f() { f | f; }; f; echo after").await + }) + }) + .unwrap(); + let result = handle.join().unwrap(); + let failed = match &result { + Err(e) => e.to_string().contains("limit") || e.to_string().contains("depth"), + Ok(r) => r.exit_code != 0 || r.stderr.contains("limit"), + }; + assert!(failed, "{result:?}"); +} diff --git a/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh b/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh index 48e2ef383..2c23f2bd8 100644 --- a/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh +++ b/crates/bashkit/tests/spec_cases/bash/pipes-redirects.test.sh @@ -332,3 +332,42 @@ echo "still here" after 3 still here ### end + +### pipeline_endless_producer_into_head +# head exits after two lines; the endless loop gets SIGPIPE (141) +while :; do echo x; done | head -n 2 +echo "${PIPESTATUS[*]}" +### expect +x +x +141 0 +### end + +### pipeline_endless_producer_into_reads +# A group that reads two lines and leaves ends the producer too +i=0 +while :; do echo $((i++)); done | { read -r a; read -r b; echo "$a $b"; } +echo "${PIPESTATUS[*]}" +### expect +0 1 +141 0 +### end + +### pipeline_endless_through_middle_stage +# Every stage upstream of head is ended by SIGPIPE +while :; do echo a; done | while read -r l; do echo "<$l>"; done | head -n 2 +echo "${PIPESTATUS[*]}" +### expect + + +141 141 0 +### end + +### pipeline_reader_breaks_out_early +# A while-read loop that breaks ends an endless producer +while :; do echo y; done | while read -r l; do echo "got $l"; break; done +echo "${PIPESTATUS[*]}" +### expect +got y +141 0 +### end diff --git a/knowledge/foundations/parallel-execution.md b/knowledge/foundations/parallel-execution.md index 855e042bd..ed0bd0898 100644 --- a/knowledge/foundations/parallel-execution.md +++ b/knowledge/foundations/parallel-execution.md @@ -53,11 +53,30 @@ Implemented `exit`/`return` end only that stage. `shopt -s lastpipe` keeps the last stage in the current shell (bash does this when job control is off, which is always the case here). So `echo x | read v` leaves `v` unset, as in bash. -- Stages reuse the cheap subshell snapshot (`execute_pipeline_stage`), not a - fork: they still run one after another (L-PIPE-001). -- A non-last stage's stdout feeds the next stage only; the streaming output - callback is suspended while it runs, so `for ... | tac` streams the - reversed lines, not the loop's. +- Leading stages that are one builtin call (`echo`, `grep`, `seq` ...) run + first, one after another, on the cheap subshell snapshot + (`enter_pipeline_stage`/`exit_pipeline_stage`): a builtin produces its + output as one value anyway, and this keeps `echo x | grep x` fork-free. +- From the first stage that runs shell code (loop, group, function, `eval`, + nested shell), the rest run concurrently (`execute_streaming_stages`, on + with `concurrent_jobs`). Each non-last stage is a forked interpreter like a + background job; the last stays in this shell so its output streams to the + caller. All stage futures are polled on the caller's task, readers first. +- Stages talk through `interpreter/pipe.rs`: a 4 KiB shared buffer. A writer + appends at command boundaries (through the streaming output hook, so + command substitutions and redirects stay out of the pipe) and waits at its + next command while the buffer is full. Before a command in a reading stage + takes stdin, `fill_stdin_from_pipe` pulls what it needs: one line for + `read`, N lines/bytes for `head` (then closes the read end), nothing for + commands that never read stdin, all of it otherwise. +- When the reader is gone (stage finished, or `head` done), the writer's next + append fails and its next command ends the stage with 141, like SIGPIPE: + `while :; do echo x; done | head -n 2` prints two lines and `PIPESTATUS` + is `141 0`. +- Concurrent nesting costs native stack (each level polls the next from its + own poll), so a pipeline nested more than 4 subshells deep runs its stages + in sequence (TM-DOS-124). `concurrent_jobs(false)` runs every stage in + sequence. Gaps: L-PIPE-001. ## Benchmark diff --git a/knowledge/log.md b/knowledge/log.md index 5fcff54ef..f6b85471a 100644 --- a/knowledge/log.md +++ b/knowledge/log.md @@ -2,6 +2,7 @@ ## 2026-10-07 +* **Feature**: Pipelines stream. From the first stage that runs shell code, stages run concurrently on forked shells over 4 KiB pipes, with backpressure and SIGPIPE: `while :; do echo x; done | head -n 2` stops the loop (`PIPESTATUS` `141 0`) instead of running it into the command limit. Single-builtin leading stages keep running in sequence (no fork). The pipe is 4 KiB rather than Linux's 64 KiB because every byte an endless producer runs ahead costs budgeted commands. Nested streaming is capped at 4 levels after a debug-build stack measurement (TM-DOS-124). L-PIPE-001 narrowed. See [Parallel Execution](foundations/parallel-execution.md#pipelines). * **Fix**: Pipeline stages are subshells, as in bash: `echo x | read v` leaves `v` unset and `cd /tmp | true` keeps the cwd; `shopt -s lastpipe` keeps the last stage in the shell. Streaming callers no longer see a non-last stage's output (`for ...; done | tac` streamed the loop's lines in input order). Stages still run sequentially (new L-PIPE-001). See [Parallel Execution](foundations/parallel-execution.md#pipelines). * **Performance**: CPython `random` seeds lazily. The snapshot's `random._inst` gets a subclass whose entry points (`random`, `getrandbits`, `getstate`, `seed`, `setstate`) seed from `os.urandom` on first use, then switch it back to `Random` and rebind those five module functions. Calls that never touch `random` skip the reseed (`pass` 3.7 to 3.5 ms). See [CPython WebAssembly Runtime](runtimes/cpython-wasm.md). * **Decision**: Native code for CPython is an opt-in feature, `cpython-native` (companion crate `native`): `build.rs` compiles the snapshot for Cargo's `TARGET` (baseline ISA) instead of Pulley. Warm `print(1)` 0.95 ms vs 3.7 ms, CPU-bound ~10x faster; needs executable memory and a ~47 ms first load (module copy). Pulley stays the default. CI runs the CPython suites on both. See [CPython WebAssembly Runtime](runtimes/cpython-wasm.md), TM-PY-CPY-011. diff --git a/knowledge/operations/limitations.md b/knowledge/operations/limitations.md index c3db537d2..7746f316c 100644 --- a/knowledge/operations/limitations.md +++ b/knowledge/operations/limitations.md @@ -45,7 +45,7 @@ execution model. Evidence is a threat-model ID, a test, or `stance` | L-PROC-001 | `exec` does not replace the process; `exec cmd` runs cmd then stops execution (fd redirects work) | True process replace would break sandbox containment | TM-ESC-005 | | 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 | Core sandbox model: no fork/exec escape surface | `l_proc_003_no_process_spawning` | -| L-PIPE-001 | Pipeline stages run one after another, not concurrently: a stage finishes before the next one reads its output. An endless producer piped into an early-exiting reader (`yes \| head -1`, `while :; do echo; done \| head`) runs until its own cap or a limit, and no stage gets SIGPIPE (exit 141) | Stages hand over complete output buffers; streaming stages over a bounded pipe are the next step of the fidelity roadmap | `l_pipe_001_stages_run_sequentially` | +| L-PIPE-001 | Pipeline stages stream only from the first stage that runs shell code (loop, group, function, `eval`). Leading single-builtin stages run to completion first and hand over their whole output, so they never get SIGPIPE: `yes \| head -1` gives `PIPESTATUS` `1 0` plus yes's output-cap notice, where bash gives `141 0`. Downstream of a streaming stage, a single builtin reads all its input before it starts, except `head` and `read`: `while :; do echo; done \| cat \| 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-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` 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` | | L-ROOTFS-001 | Default rootfs is static and read-only: `/proc` has no `self`, pid dirs, `uptime` or live counters; `/etc/passwd` has no root entry; `/dev/zero` yields 1 MiB per read; `/bin`, `/usr/bin` are stubs that dispatch builtins | Host state must not leak (TM-INF-003, TM-ISO-018); fixed values keep runs deterministic | `rootfs_layout`, `threat_etc_passwd_blocked` | diff --git a/knowledge/security/threat-model.md b/knowledge/security/threat-model.md index 1e32fba8e..ed05fad91 100644 --- a/knowledge/security/threat-model.md +++ b/knowledge/security/threat-model.md @@ -291,6 +291,7 @@ runaway scripts without permanently breaking the session. | TM-DOS-121 | `find` traversal amplification | `find -L` over a symlink cycle (`ln -s .. d/up`) re-enters the same directories forever; `-exec ... \;` as a predicate or `{} +` batches could also queue unbounded command lines; the old `-name` glob matcher recursed with string clones per `*` (exponential on `*a*a*...b`) | `-L`/`-H` resolve each directory to its canonical path (realpath with a 40-hop ELOOP cap) and refuse to enter one already on the current path (`File system loop detected`, exit 1); every directory listing charges `consume_work`; find's own output stays capped by `FIND_MAX_OUTPUT_BYTES`; `{} +` batches flush every 4096 paths; `fnmatch` uses a single star restore point (O(value x pattern)). Regression tests: `find_follow_detects_loops` spec case, `ls::glob::tests::many_stars_are_linear` | **MITIGATED** | | TM-DOS-122 | Background job flooding | `while :; do sleep 99 & done`, or a fork bomb `f(){ f & f & }; f`, keeps unbounded concurrent forked interpreters alive | `ExecutionLimits::max_background_jobs` (default 64, `hardened()` 16) caps live jobs; the next `&` prints bash's `fork: retry: Resource temporarily unavailable` and returns 1 without forking. Jobs are polled on the caller's task (no `tokio::spawn`), share the session's command/loop budget, timeout and cancel token, and `exec()` waits for or drops every job before returning, so nothing outlives the call. Regressions: `jobs::tests::*`, `jobs.test.sh` spec cases, `background_job_limit` | **MITIGATED** | | TM-DOS-123 | Random generator amplification | `openssl rand 99999999999` asks one builtin call for gigabytes of CSPRNG output | `openssl rand` caps a request at `OPENSSL_RAND_MAX_BYTES` (1 MiB) and fails above it; `uuidgen` and `$SRANDOM` emit fixed sizes. Bytes come from `getrandom` (OS CSPRNG), never the `$RANDOM` LCG. Regression: `builtins::random::tests::openssl_rand_limits` | **MITIGATED** | +| TM-DOS-124 | Concurrent pipeline amplification | An endless producer (`while :; do echo x; done \| head`) buffering without bound, a producer whose reader quit running forever, or recursive pipelines (`f() { f \| f; }`) nesting forked stages and native stack | Pipes hold 4 KiB and writers wait at command boundaries; a writer whose reader is gone ends with 141 at its next command; a reader that never reads stdin ends the producer when it finishes. A streaming group counts as one subshell level (`max_subshell_depth`) and past 4 levels stages run in sequence, keeping stack cost near plain function recursion. Stages are forked like jobs: they share the execution budget, timeout and cancel token and never outlive the pipeline. Regressions: `pipeline_tests::threat_endless_producer_into_non_reader_ends`, `threat_recursive_pipeline_is_bounded`, `pipes-redirects.test.sh` SIGPIPE cases | **MITIGATED** | | TM-DOS-101 | yq structured-data amplification | YAML aliases can expand exponentially during deserialization; deep YAML/JSON, document floods, jaq generators, and YAML re-serialization can consume stack, CPU, or memory beyond the source size | Reject YAML alias tokens with a bounded lexical pass before deserialization; VFS/stdin and aggregate input budgets; serde_yaml_ng recursion cap 128 plus Bashkit depth 100; 4096-document cap; shared jaq work/deadline/output limits; post-serialization stdout cap; YAML+JSON depth/document/input/output regressions plus `yq_fuzz` and arbitrary-input proptest | **MITIGATED** | | TM-DOS-107 | Static-analysis command validation amplification | A large comment before thousands of commands makes a whole-script ordered-subsequence scan per command | Build one source-character position index, then validate each analyzed command-name character with a logarithmic position lookup instead of rescanning source | **MITIGATED** | **TM-DOS-051** is historical. The custom parser was deleted with the `yaml`