diff --git a/crates/bashkit/docs/threat-model.md b/crates/bashkit/docs/threat-model.md index 7cbc4c977..996e5ff5a 100644 --- a/crates/bashkit/docs/threat-model.md +++ b/crates/bashkit/docs/threat-model.md @@ -130,7 +130,7 @@ stdout capture limit remains downstream of pipelines and redirects. | 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 | +| 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; producers atomically claim the shared job cap before construction, borrow the AST and lease owned fork storage/future layouts before allocation | MITIGATED | | make amplification (TM-DOS-126) | Exponential recursive variables, self-including makefiles, deep `$(MAKE)` recursion, huge graphs | Expansion depth 200 and 4 MiB per expansion; substitution checks its exact size before construction, with shared capacity leases retained by expanded values and work/deadline checkpoints while matching and copying; include depth 16; target, query and glob caps per run; `$(MAKE)` stops at level 4; recipes run under the session budget and timeout | MITIGATED | | Random generator amplification (TM-DOS-123) | `openssl rand` asked for gigabytes | Requests capped at 1 MiB; fixed-size `uuidgen`/`$SRANDOM` | MITIGATED | | Child-shell recursion (TM-DOS-125) | Script that re-runs itself via `sh` | Child shells count against function depth, max 8 nested | MITIGATED | diff --git a/crates/bashkit/src/fs/proc_sub_fds.rs b/crates/bashkit/src/fs/proc_sub_fds.rs index 6647c834d..1aa6a0ca7 100644 --- a/crates/bashkit/src/fs/proc_sub_fds.rs +++ b/crates/bashkit/src/fs/proc_sub_fds.rs @@ -90,6 +90,17 @@ impl ProcSubFs { }) } + /// Fork metadata only: substitution byte payloads remain Arc-shared. + pub(crate) fn fork_storage_bytes(&self) -> usize { + std::mem::size_of::() + .saturating_add(2 * std::mem::size_of::()) + .saturating_add( + self.table() + .len() + .saturating_mul(std::mem::size_of::<(i32, ProcSubData)>()), + ) + } + /// Same filesystem, a copy of the open fds (a forked job). pub(crate) fn fork(&self) -> Arc { Arc::new(Self { diff --git a/crates/bashkit/src/interpreter/completion.rs b/crates/bashkit/src/interpreter/completion.rs index fb2dd8c01..310ce1432 100644 --- a/crates/bashkit/src/interpreter/completion.rs +++ b/crates/bashkit/src/interpreter/completion.rs @@ -250,6 +250,20 @@ pub(crate) struct CompletionState { pub(crate) bindings: super::readline_bind::Bindings, } +impl CompletionState { + pub(super) fn fork_storage( + &self, + storage: &mut super::pipeline_admission::ForkStorage<'_>, + ) -> Result<()> { + storage.slice::<(String, CompSpec)>(self.specs.len())?; + for (name, spec) in &self.specs { + storage.add(name.len())?; + storage.add(spec.text_bytes())?; + } + self.bindings.fork_storage(storage) + } +} + /// Parsed `complete`/`compgen` options. #[derive(Default)] struct CompArgs { diff --git a/crates/bashkit/src/interpreter/jobs.rs b/crates/bashkit/src/interpreter/jobs.rs index eaab67451..efdac057b 100644 --- a/crates/bashkit/src/interpreter/jobs.rs +++ b/crates/bashkit/src/interpreter/jobs.rs @@ -341,13 +341,13 @@ pub struct JobControl { live: Arc, } -/// One live job's claim on the session-wide job cap; released on drop -/// (finish or abort). -pub struct JobSlot(Arc); +/// A claim on the session-wide job cap; released on drop (finish or abort). +/// Pipelines retain a group claim until every producer owner is dropped. +pub struct JobSlot(Arc, usize); impl Drop for JobSlot { fn drop(&mut self) { - self.0.fetch_sub(1, AtomicOrdering::AcqRel); + self.0.fetch_sub(self.1, AtomicOrdering::AcqRel); } } @@ -371,12 +371,19 @@ impl JobControl { /// Claim a slot under `max` live jobs session-wide (TM-DOS-122). pub fn try_claim(&self, max: usize) -> Option { + self.try_claim_many(max, 1) + } + + /// Admit every producer before constructing a concurrent pipeline. + /// One drop-owned claim keeps its retained forks counted until completion + /// or cancellation; nested pipelines and background jobs share this cap. + pub(crate) fn try_claim_many(&self, max: usize, count: usize) -> Option { self.live .fetch_update(AtomicOrdering::AcqRel, AtomicOrdering::Acquire, |n| { - (n < max).then_some(n + 1) + n.checked_add(count).filter(|next| *next <= max) }) .ok() - .map(|_| JobSlot(Arc::clone(&self.live))) + .map(|_| JobSlot(Arc::clone(&self.live), count)) } /// Lock the table. Never held across an `.await`. @@ -462,6 +469,22 @@ pub fn new_shared_job_table() -> SharedJobTable { #[cfg(test)] mod tests { + #[test] + fn pipeline_admission_claims_atomically_across_forks_and_refunds_on_drop() { + let jobs = super::new_shared_job_table(); + let child = jobs.fork(); + let occupied = jobs.try_claim(2).unwrap(); + assert!(child.try_claim_many(2, 2).is_none()); + let remaining = child.try_claim(2).unwrap(); + assert!(jobs.try_claim(2).is_none()); + drop(occupied); + drop(remaining); + let group = child.try_claim_many(2, 2).unwrap(); + assert!(jobs.try_claim(2).is_none()); + drop(group); + assert!(jobs.try_claim_many(2, 2).is_some()); + } + use super::*; use std::time::Duration; diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index 10fce58d5..3c0ef5ea5 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -47,6 +47,7 @@ mod glob; mod history; mod jobs; pub(crate) mod pipe; +mod pipeline_admission; mod prompt; mod readline_bind; mod readline_defaults; @@ -1794,7 +1795,7 @@ pub struct Interpreter { http_client: Option>, /// Git client for git builtins #[cfg(feature = "git")] - git_client: Option, + git_client: Option>, /// SSH client for ssh/scp/sftp builtins #[cfg(feature = "ssh")] ssh_client: Option>, @@ -3148,6 +3149,7 @@ impl Interpreter { } self.getopts_char_idx = 0; self.pipeline_stdin = None; + self.clear_pipeline_streams(); self.regex_cache = RuntimeRegexCache::default(); self.bash_source_stack.clear(); self.arrays_mut().remove("BASH_SOURCE"); @@ -3156,6 +3158,15 @@ impl Interpreter { // Called after a host-backed deadline drops the in-flight execution future. pub(crate) fn clear_cancelled_execution_state(&mut self) { self.reconcile_cancelled_execution_state(0, 0, 0, None); + self.clear_pipeline_streams(); + } + + fn clear_pipeline_streams(&mut self) { + self.pipe_in = None; + self.pipe_out = None; + self.builtin_stdin_pipe = None; + self.builtin_stdout_pipe = None; + self.stream_stdout_command = None; } fn clear_pending_fd_redirect_state(&mut self) { @@ -3915,7 +3926,7 @@ impl Interpreter { /// This is only available when the `git` feature is enabled. #[cfg(feature = "git")] pub fn set_git_client(&mut self, client: crate::builtins::git::GitClient) { - self.git_client = Some(client); + self.git_client = Some(Arc::new(client)); } /// Set the SSH client for ssh/scp/sftp builtins. @@ -7870,14 +7881,34 @@ impl Interpreter { stdin: Option, lastpipe: bool, ) -> Result> { - type StageFuture = std::pin::Pin< - Box> + Send + 'static>, - >; + type StageFuture<'a> = + std::pin::Pin> + Send + 'a>>; let (last, producers) = commands.split_last().expect("pipeline tail"); + // THREAT[TM-DOS-124]: width, fork storage and construction work are + // admitted before allocation. Every retained producer counts against + // the same cap as background jobs, including descendants. + self.execution_budget.check()?; + let _slots = self + .jobs + .try_claim_many(self.limits.max_background_jobs, producers.len()) + .ok_or(LimitExceeded::MaxConcurrentJobs( + self.limits.max_background_jobs, + ))?; let mut input: Option> = None; - let mut stages: Vec = Vec::with_capacity(producers.len()); + let mut stages = + crate::limits::BudgetedVec::<( + StageFuture<'_>, + crate::limits::ExecutionBudgetLease, + crate::limits::ExecutionBudgetLease, + )>::try_with_capacity(Some(&self.execution_budget), producers.len())?; for (k, command) in producers.iter().enumerate() { - let pipe = pipe::Pipe::new(); + let inherited_stdin = if k == 0 { + stdin.as_ref().or(self.pipeline_stdin.as_ref()) + } else { + None + }; + let fork_storage = self.pipeline_fork_storage(inherited_stdin)?; + let pipe = pipe::Pipe::new_admitted(&self.execution_budget)?; let mut child = self.fork_for_job(); if k == 0 { // First stage reads what this shell would read. @@ -7898,7 +7929,6 @@ impl Interpreter { Arc::make_mut(&mut child.nofork).subshell_env = SUBSHELL_PIPE; let write_end = pipe::WriteEnd(Arc::clone(&pipe)); let read_end = input.take().map(pipe::ReadEnd); - let command = command.clone(); if matches!( command, Command::Simple(_) | Command::Compound(CompoundCommand::Subshell(_), _) @@ -7907,15 +7937,15 @@ impl Interpreter { // `( )` stage counts itself once. child.bash_subshell = self.bash_subshell; } - stages.push(Box::pin(async move { + let future = async move { let _read_end = read_end; - if let Command::Simple(simple) = &command + if let Command::Simple(simple) = command && Self::streams_stdout(simple) { child.stream_stdout_command = Some(simple as *const SimpleCommand as usize); } let jobs = Arc::clone(&child.jobs); - let result = jobs::with_jobs(&jobs, child.execute_command(&command)).await; + 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?; @@ -7934,16 +7964,26 @@ impl Interpreter { r.control_flow = ControlFlow::None; r.stdout = crate::StreamData::new(); Ok(r) - })); + }; + let future_storage = self + .execution_budget + .lease_bytes(std::mem::size_of_val(&future))?; + stages.try_push((Box::pin(future), fork_storage, future_storage))?; input = Some(pipe); } + let mut done = crate::limits::BudgetedVec::try_with_capacity( + Some(&self.execution_budget), + stages.len(), + )?; + for _ in 0..stages.len() { + done.try_push(None)?; + } // 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 { @@ -7968,7 +8008,9 @@ impl Interpreter { last_done = Some(r); cx.waker().wake_by_ref(); } - for (stage, slot) in stages.iter_mut().zip(done.iter_mut()).rev() { + for ((stage, _fork_storage, _future_storage), 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) { @@ -7986,11 +8028,12 @@ impl Interpreter { .await; } drop(read_end); - drop(stages); self.pipe_in = saved_pipe_in; self.pipeline_stdin = saved_stdin; + drop(stages); let mut results = Vec::with_capacity(done.len() + 1); + let (done, _done_storage) = done.into_parts(); for d in done { results.push(d.unwrap_or_else(|| Ok(ExecResult::ok(String::new())))?); } @@ -9876,7 +9919,7 @@ impl Interpreter { #[cfg(feature = "http_client")] http_client: self.http_client.as_ref(), #[cfg(feature = "git")] - git_client: self.git_client.as_ref(), + git_client: self.git_client.as_deref(), #[cfg(feature = "ssh")] ssh_client: self.ssh_client.as_deref(), shell: Some(shell_ref), @@ -9951,7 +9994,7 @@ impl Interpreter { #[cfg(feature = "http_client")] http_client: self.http_client.as_ref(), #[cfg(feature = "git")] - git_client: self.git_client.as_ref(), + git_client: self.git_client.as_deref(), #[cfg(feature = "ssh")] ssh_client: self.ssh_client.as_deref(), shell: Some(shell_ref), @@ -14963,6 +15006,131 @@ mod tests { use crate::fs::InMemoryFs; use crate::parser::Parser; + fn ordinary_pipeline_commands() -> Vec { + let parser = Parser::new("{ :; } | :"); + let script = parser.parse().unwrap(); + let Command::Pipeline(pipeline) = script.commands.into_iter().next().unwrap() else { + panic!("expected pipeline"); + }; + pipeline.commands + } + + #[tokio::test] + async fn pipeline_admission_respects_occupied_shared_job_slot() { + let mut bash = Bash::new(); + bash.interpreter.begin_execution_budget(); + bash.interpreter.limits.max_background_jobs = 1; + let slot = bash.interpreter.jobs.try_claim(1).unwrap(); + let result = bash + .interpreter + .execute_streaming_stages(&ordinary_pipeline_commands(), None, false) + .await; + assert!(matches!( + result, + Err(crate::Error::ResourceLimit( + LimitExceeded::MaxConcurrentJobs(1) + )) + )); + drop(slot); + assert!(bash.interpreter.jobs.try_claim(1).is_some()); + } + + #[tokio::test] + async fn pipeline_admission_reserves_storage_for_no_output_stage() { + let mut bash = Bash::new(); + bash.interpreter.begin_execution_budget(); + bash.interpreter.limits.max_live_intermediate_bytes = 0; + bash.interpreter.begin_execution_budget(); + let result = bash + .interpreter + .execute_streaming_stages(&ordinary_pipeline_commands(), None, false) + .await; + assert!(matches!( + result, + Err(crate::Error::ResourceLimit(LimitExceeded::ExecutionBudget( + crate::limits::ExecutionBudgetExceeded::LiveBytes { .. } + ))) + )); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + assert!(bash.interpreter.jobs.try_claim(1).is_some()); + } + + #[tokio::test] + async fn pipeline_admission_releases_storage_and_slots_after_completion() { + let mut bash = Bash::new(); + bash.interpreter.begin_execution_budget(); + bash.interpreter.limits.max_background_jobs = 1; + let result = bash + .interpreter + .execute_streaming_stages(&ordinary_pipeline_commands(), None, false) + .await + .unwrap(); + assert_eq!(result.len(), 2); + assert!(result.iter().all(|r| r.exit_code == 0)); + assert!(bash.interpreter.jobs.try_claim(1).is_some()); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + + #[tokio::test] + async fn pipeline_admission_checks_cancellation_before_construction() { + let mut bash = Bash::new(); + bash.interpreter.begin_execution_budget(); + bash.interpreter.cancelled.store(true, Ordering::Relaxed); + let result = bash + .interpreter + .execute_streaming_stages(&ordinary_pipeline_commands(), None, false) + .await; + assert!(matches!(result, Err(crate::Error::Cancelled))); + assert!(bash.interpreter.jobs.try_claim(1).is_some()); + assert_eq!(bash.interpreter.execution_budget.live_bytes_for_test(), 0); + } + + #[tokio::test] + async fn pipeline_admission_drop_pending_and_reconcile_releases_owners() { + use std::future::Future; + let mut bash = Bash::new(); + bash.interpreter.limits.max_background_jobs = 1; + bash.interpreter.begin_execution_budget(); + let script = Parser::new("{ sleep 1; } | cat").parse().unwrap(); + let Command::Pipeline(commands) = &script.commands[0] else { + panic!("expected pipeline"); + }; + let budget = bash.interpreter.execution_budget.clone(); + let jobs = Arc::clone(&bash.interpreter.jobs); + let mut future = Box::pin(bash.interpreter.execute_streaming_stages( + &commands.commands, + None, + false, + )); + let waker = futures_util::task::noop_waker(); + let mut context = std::task::Context::from_waker(&waker); + assert!(future.as_mut().poll(&mut context).is_pending()); + assert!(jobs.try_claim(1).is_none()); + assert!(budget.live_bytes_for_test() > 0); + drop(future); + assert!(jobs.try_claim(1).is_some()); + // The caller still owns the final pipe until the ordinary cancelled + // request reconciliation releases its transient streaming state. + assert!(budget.live_bytes_for_test() > 0); + bash.interpreter.clear_cancelled_execution_state(); + assert!(bash.interpreter.pipe_in.is_none()); + assert_eq!(budget.live_bytes_for_test(), 0); + } + + #[test] + fn pipeline_admission_environment_is_shared_and_isolated() { + let mut bash = Bash::new(); + bash.interpreter.begin_execution_budget(); + bash.interpreter + .env_mut() + .insert("DEMO".into(), "parent".into()); + let mut child = bash.interpreter.fork_for_job(); + assert!(Arc::ptr_eq(&bash.interpreter.env, &child.env)); + child.env_mut().insert("DEMO".into(), "child".into()); + assert_eq!(bash.interpreter.env["DEMO"], "parent"); + assert_eq!(child.env["DEMO"], "child"); + } + #[tokio::test] async fn function_metadata_shares_storage_and_drops_with_functions() { let mut bash = Bash::new(); diff --git a/crates/bashkit/src/interpreter/pipe.rs b/crates/bashkit/src/interpreter/pipe.rs index 14f404c05..58a68e4a2 100644 --- a/crates/bashkit/src/interpreter/pipe.rs +++ b/crates/bashkit/src/interpreter/pipe.rs @@ -72,6 +72,7 @@ impl State { #[derive(Default)] pub(crate) struct Pipe { state: Mutex, + _metadata_storage: Option, } impl Pipe { @@ -79,6 +80,19 @@ impl Pipe { Arc::new(Self::default()) } + /// Construction storage follows the pipe itself, including a final-stage + /// reference that survives until cancelled-request state is reconciled. + pub(crate) fn new_admitted( + budget: &crate::limits::ExecutionBudget, + ) -> crate::error::Result> { + let storage = + budget.lease_bytes(std::mem::size_of::() + 2 * std::mem::size_of::())?; + Ok(Arc::new(Self { + state: Mutex::new(State::default()), + _metadata_storage: Some(storage), + })) + } + fn lock(&self) -> MutexGuard<'_, State> { self.state.lock().unwrap_or_else(|e| e.into_inner()) } diff --git a/crates/bashkit/src/interpreter/pipeline_admission.rs b/crates/bashkit/src/interpreter/pipeline_admission.rs new file mode 100644 index 000000000..7c280fa22 --- /dev/null +++ b/crates/bashkit/src/interpreter/pipeline_admission.rs @@ -0,0 +1,151 @@ +//! Concurrent pipeline producers use the existing session-wide job cap and +//! request storage leases before forking. Shared CoW state is not copied; +//! owned fork fields are measured conservatively, and stage ASTs are borrowed. +//! Keep this inventory aligned with `fork_for_job` when its ownership changes. + +use super::*; +use crate::limits::{ExecutionBudget, ExecutionBudgetLease}; +use std::mem::size_of; + +pub(super) struct ForkStorage<'a> { + budget: &'a ExecutionBudget, + bytes: usize, +} + +impl<'a> ForkStorage<'a> { + fn new(budget: &'a ExecutionBudget) -> Self { + Self { budget, bytes: 0 } + } + + pub(super) fn add(&mut self, bytes: usize) -> Result<()> { + self.budget.consume_work(1 + bytes.div_ceil(1024) as u64)?; + self.bytes = self + .bytes + .checked_add(bytes) + .ok_or_else(|| LimitExceeded::Memory("pipeline fork storage size overflow".into()))?; + Ok(()) + } + + pub(super) fn slice(&mut self, len: usize) -> Result<()> { + self.add( + len.checked_mul(size_of::()).ok_or_else(|| { + LimitExceeded::Memory("pipeline fork storage size overflow".into()) + })?, + ) + } + + fn map(&mut self, capacity: usize) -> Result<()> { + // HashMap cloning retains the table capacity. Two buckets per usable + // entry cover spare/control buckets, including small tables. + self.slice::<(K, V)>( + capacity.checked_mul(2).ok_or_else(|| { + LimitExceeded::Memory("pipeline fork storage size overflow".into()) + })?, + )?; + self.add(capacity.saturating_mul(2).saturating_add(16)) + } + + fn strings(&mut self, strings: &[String]) -> Result<()> { + self.slice::(strings.len())?; + for value in strings { + self.add(value.len())?; + } + Ok(()) + } + + fn path(&mut self, path: &PathBuf) -> Result<()> { + // Also conservative for platforms whose OsString uses wide units. + self.add(path.capacity().saturating_mul(2)) + } +} + +impl Interpreter { + pub(super) fn pipeline_fork_storage( + &self, + stdin: Option<&crate::StreamData>, + ) -> Result { + let mut storage = ForkStorage::new(&self.execution_budget); + // Fixed heap owners created by the fork. Pipes hold their own lease. The future's + // actual layout is leased separately before Box::pin. + storage.add( + size_of::() + + size_of::() + + size_of::<[usize; 8]>() + + size_of::>(), + )?; + storage.add(self.proc_subs.fork_storage_bytes())?; + if let Some(stdin) = stdin { + storage.add(stdin.len())?; + storage.add(stdin.text_lossy().len())?; + } + storage.path(&self.cwd)?; + storage.add(self.last_command_name.len())?; + storage.add(self.prompt_user.len())?; + storage.add(self.prompt_host.len())?; + storage.add(self.last_bg_pid.as_ref().map_or(0, String::len))?; + storage.map::>(self.builtins.capacity())?; + for name in self.builtins.keys() { + storage.add(name.len())?; + } + storage.slice::(self.call_stack.len())?; + for frame in &self.call_stack { + storage.add(frame.name.len())?; + storage.strings(&frame.positional)?; + storage.map::(frame.saved_vars.capacity())?; + for (name, value) in &frame.saved_vars { + storage.add(name.len())?; + for value in [&value.value, &value.nameref, &value.env] + .into_iter() + .flatten() + { + storage.add(value.len())?; + } + } + storage.map::>>(frame.local_arrays.capacity())?; + for (name, values) in &frame.local_arrays { + storage.add(name.len())?; + if let Some(values) = values { + storage.map::(values.capacity())?; + for value in values.values() { + storage.add(value.len())?; + } + } + } + storage.map::>>( + frame.local_assoc_arrays.capacity(), + )?; + for (name, values) in &frame.local_assoc_arrays { + storage.add(name.len())?; + if let Some(values) = values { + storage.map::(values.capacity())?; + for (name, value) in values { + storage.add(name.len())?; + storage.add(value.len())?; + } + } + } + } + storage.slice::(self.bash_source_stack.len())?; + for frame in &self.bash_source_stack { + storage.add(frame.funcname.len())?; + } + storage.slice::(self.counters.loop_iterations.len())?; + storage.map::(self.exec_fd_table.capacity())?; + for target in self.exec_fd_table.values() { + storage.add(0)?; + if let FdTarget::WriteFile(path, text) | FdTarget::AppendFile(path, text) = target { + storage.path(path)?; + storage.add(text.len())?; + } + } + storage.map::(self.exec_input_fds.capacity())?; + storage.slice::(self.fd_redirect_scope.len())?; + if let Some(completion) = &self.completion { + storage.add(size_of::())?; + completion.fork_storage(&mut storage)?; + } + self.execution_budget + .lease_bytes(storage.bytes) + .map_err(Into::into) + } +} diff --git a/crates/bashkit/src/interpreter/readline_bind.rs b/crates/bashkit/src/interpreter/readline_bind.rs index 7ce88288a..81dde5f8b 100644 --- a/crates/bashkit/src/interpreter/readline_bind.rs +++ b/crates/bashkit/src/interpreter/readline_bind.rs @@ -163,6 +163,27 @@ fn unquote(s: &str) -> String { } impl Bindings { + pub(super) fn fork_storage( + &self, + storage: &mut super::pipeline_admission::ForkStorage<'_>, + ) -> Result<()> { + storage.slice::(self.changes.len())?; + for binding in &self.changes { + storage.add(binding.seq.len())?; + if let Some(Target::Function(text) | Target::Macro(text) | Target::Command(text)) = + &binding.target + { + storage.add(text.len())?; + } + } + storage.slice::<(String, String)>(self.variables.len())?; + for (name, value) in &self.variables { + storage.add(name.len())?; + storage.add(value.len())?; + } + Ok(()) + } + /// Every live binding of `keymap`: defaults minus unbound keys, plus /// changes (a later change of the same keys wins). fn effective(&self, keymap: &str) -> Vec<(Vec, String, Target)> { diff --git a/crates/bashkit/src/limits.rs b/crates/bashkit/src/limits.rs index f0d44d75d..1a29f0586 100644 --- a/crates/bashkit/src/limits.rs +++ b/crates/bashkit/src/limits.rs @@ -164,7 +164,8 @@ pub struct ExecutionLimits { // THREAT[TM-DOS-122]: each `&` job holds a forked shell state and a // pending future until it finishes or is killed. - /// Maximum background jobs running at once; past it `&` fails. + /// Maximum concurrent background jobs and retained pipeline producers. + /// Past it `&` fails or a streaming pipeline returns a resource-limit error. /// Default: 64 pub max_background_jobs: usize, @@ -249,7 +250,7 @@ impl ExecutionLimits { } } - /// Set the maximum number of background jobs running at once. + /// Set the maximum concurrent background jobs and retained pipeline producers. pub fn max_background_jobs(mut self, count: usize) -> Self { self.max_background_jobs = count; self @@ -743,6 +744,9 @@ pub enum LimitExceeded { #[error("maximum trap nesting depth exceeded ({0})")] MaxTrapDepth(usize), + #[error("maximum concurrent jobs exceeded ({0})")] + MaxConcurrentJobs(usize), + #[error("maximum file descriptors exceeded ({0})")] MaxFileDescriptors(usize), @@ -987,7 +991,7 @@ impl ExecutionBudget { } #[cfg(test)] - fn live_bytes_for_test(&self) -> u64 { + pub(crate) fn live_bytes_for_test(&self) -> u64 { self.inner.live_bytes.load(Ordering::Relaxed) } } diff --git a/crates/bashkit/tests/integration/main.rs b/crates/bashkit/tests/integration/main.rs index 19cb7ed61..a6a3f3f6a 100644 --- a/crates/bashkit/tests/integration/main.rs +++ b/crates/bashkit/tests/integration/main.rs @@ -125,6 +125,7 @@ pub mod od_resource_tests; pub mod output_truncation_tests; pub mod parallel_sessions_tests; pub mod partial_parse_tests; +pub mod pipeline_admission_tests; pub mod pipeline_tests; pub mod pr_resource_tests; pub mod process_substitution_fd_tests; diff --git a/crates/bashkit/tests/integration/pipeline_admission_tests.rs b/crates/bashkit/tests/integration/pipeline_admission_tests.rs new file mode 100644 index 000000000..31456fa5e --- /dev/null +++ b/crates/bashkit/tests/integration/pipeline_admission_tests.rs @@ -0,0 +1,59 @@ +//! Ordinary two-stage compatibility under producer admission; no stress inputs. +use bashkit::{Bash, ExecOptions, ExecutionLimits}; + +#[tokio::test] +async fn two_stage_pipeline_preserves_exports_status_and_isolation() { + let script = "export DEMO=parent; { export DEMO=child; printf '%s\\n' \"$DEMO\"; } | cat; printf '%s %s\\n' \"$DEMO\" \"${PIPESTATUS[*]}\""; + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_background_jobs(1)) + .build(); + let result = bash.exec(script).await.unwrap(); + assert_eq!(result.stdout, "child\nparent 0 0\n"); + assert!(result.stderr.is_empty()); + assert_eq!(result.exit_code, 0); + let oracle = std::process::Command::new("bash") + .args(["--noprofile", "--norc", "-c", script]) + .env_clear() + .output() + .unwrap(); + assert!(oracle.status.success()); + assert_eq!(oracle.stdout, result.stdout.as_bytes()); + assert_eq!(oracle.stderr, result.stderr.as_bytes()); +} + +#[tokio::test] +async fn sequential_two_stage_pipeline_accepts_zero_job_slots() { + let mut bash = Bash::builder() + .concurrent_jobs(false) + .limits(ExecutionLimits::new().max_background_jobs(0)) + .build(); + let result = bash.exec("{ printf 'ok\\n'; } | cat").await.unwrap(); + assert_eq!(result.stdout, "ok\n"); + assert_eq!(result.exit_code, 0); +} + +#[tokio::test] +async fn two_stage_lastpipe_retains_final_stage_assignment() { + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_background_jobs(1)) + .build(); + let result = bash.exec("shopt -s lastpipe; { printf 'ok\\n'; } | read -r value; printf '%s %s\\n' \"$value\" \"${PIPESTATUS[*]}\"").await.unwrap(); + assert_eq!(result.stdout, "ok 0 0\n"); + assert!(result.stderr.is_empty()); +} + +#[tokio::test] +async fn two_stage_pipeline_preserves_supplied_stdin_and_session_reuse() { + let mut bash = Bash::builder() + .limits(ExecutionLimits::new().max_background_jobs(1)) + .build(); + let result = bash + .exec_with_options("{ cat; } | cat", ExecOptions::new().stdin("input\n")) + .await + .unwrap(); + assert_eq!(result.stdout, "input\n"); + assert_eq!(result.exit_code, 0); + assert!(result.stderr.is_empty()); + let next = bash.exec("printf after").await.unwrap(); + assert_eq!(next.stdout, "after"); +} diff --git a/docs/security.md b/docs/security.md index df14808df..af0a47913 100644 --- a/docs/security.md +++ b/docs/security.md @@ -13,11 +13,21 @@ individual threat IDs and mitigation status, see the | Boundary | What it does | |----------|-------------| | **Virtual filesystem** | Scripts run against an in-memory VFS. No real filesystem access by default. Path traversal (`../../../etc/passwd`) is normalised away. Symlinks are followed inside the VFS only: a target is always a VFS path, so a link can never reach the host. | -| **No process execution** | `exec` is excluded entirely. `bash -c` re-invokes the virtual interpreter instead of spawning a real process. Background jobs (`&`) parse but run synchronously. | +| **No process execution** | `bash -c` and virtual `exec` execute inside the interpreter. Background jobs (`&`) run concurrently under the shared job cap. | | **Network allowlist** | HTTP/HTTPS only, pre-validated against an explicit host allowlist. No DNS resolution, no auto-redirect, no auto-decompression. | | **Resource limits** | Configurable caps on commands, loop iterations, recursion depth, AST depth, timeouts, and parser operations prevent denial-of-service from malicious scripts. | | **Filesystem limits** | Total bytes, per-file size, file count, path depth, and filename length are all capped to prevent storage exhaustion (zip bombs, tar bombs, recursive copies). | +Concurrent pipeline producers share `ExecutionLimits::max_background_jobs` +with background jobs, including nested work. All producer slots are admitted +before constructing the pipeline, and remain reserved until its producers +are released. Owned fork storage and producer futures count against +`max_live_intermediate_bytes` before allocation; construction also checks the +shared work budget, cancellation and deadline. The final stage needs no extra +job slot. Exceeding a limit returns a resource-limit error. Pipelines within +these budgets keep their streaming, isolation and `PIPESTATUS` behavior; +`concurrent_jobs(false)` uses sequential stages without producer slots. + `dd conv=sync` checks its padded block length against the 64 MiB copy cap before allocating or zero-filling. Copied payload capacity also counts against `max_live_intermediate_bytes` until output handling completes, including when diff --git a/knowledge/foundations/parallel-execution.md b/knowledge/foundations/parallel-execution.md index cf2243a01..e66175ab8 100644 --- a/knowledge/foundations/parallel-execution.md +++ b/knowledge/foundations/parallel-execution.md @@ -46,6 +46,31 @@ Implemented becomes 128+signal. - Limit: `max_background_jobs` (TM-DOS-122). +## Concurrent pipeline admission + +Non-final streaming producers atomically claim slots from the same +`max_background_jobs` counter as background jobs and descendants before any +fork or producer-vector allocation. A pipeline retains its entire group claim +until all producer owners drop; the final stage uses the caller's shell and +needs no extra slot. Refused admission returns `MaxConcurrentJobs` rather than +falling back to sequential execution. Sequential mode needs no producer slots. + +`pipeline_admission.rs` inventories the owned fields copied by `fork_for_job`: +local frames/arrays/positional values, builtin names/table capacity, source-frame +names, paths, descriptors, completion/bindings and substitution-fd metadata. +The first producer also admits both representations of copied stdin. +Its shared-budget checkpoints charge construction work and check cancellation +and deadlines; a storage lease is acquired before copying. Environment/scoped +state and immutable Git configuration remain Arc-shared. Stage commands borrow +the existing AST. Producer vectors and status slots use `BudgetedVec`; each +future's actual layout is leased before boxing. Pipe metadata owns its own +lease; cancelled-request reconciliation and the next host reset release the +caller's transient pipe references. Leases survive until producer +owners drop on completion, partial admission failure, error or cancellation. +Keep the inventory aligned with fork ownership changes. Two-stage fixtures in +`pipeline_admission_tests` cover Bash parity, export isolation, sequential mode +and lastpipe; unit fixtures cover shared admission and lease/slot cleanup. + ## Coprocesses `coproc [NAME] cmd` is a background job whose stdin and stdout are two diff --git a/knowledge/log.md b/knowledge/log.md index 38bfe70ac..2976c9a91 100644 --- a/knowledge/log.md +++ b/knowledge/log.md @@ -2,6 +2,8 @@ ## 2026-10-13 +* **Parallel execution / Security**: Streaming producers now atomically share background-job admission, lease owned fork state and future layouts before allocation, borrow stage ASTs, and check construction work/cancellation/deadlines (TM-DOS-124). Group slots and storage release with pipeline owners. Tiny admission/ownership units and ordinary two-stage Bash-parity fixtures cover the contract. + * **Operations**: Two CI holes with one cause — something was free to move to a version the pinned rustc cannot build, so an upstream release broke a tree nobody had touched. Nightly's `cargo-geiger` was installed with no version and built from source against the pinned toolchain; 0.14.0 shipped with its MSRV raised to 1.97 and run 259 went red the next morning, skipping the unsafe-code audit entirely. It is now pinned to 0.13.0 (MSRV 1.85), bumped deliberately with the toolchain, like `cargo-audit` before it. Separately, `wasmtime` is ignored in the `/` dependabot entry because the pin is exactly `=48.0.5`, but `ignore` is per-entry: the fuzz entry had none, its lockfile resolves the root workspace, and dependabot rewrote the root pin to 49.x (MSRV 1.96) in #2708, failing every rust job at resolution. `examples/hyperlight/host` had the same exposure from the example-workspaces entry. Both now carry the ignore, and `scripts/tests/test_dependabot_coverage.py` fails if any cargo entry can propose a wasmtime bump, since which entry reaches which manifest is not worth re-deriving each time one is added. The general rule: an unpinned tool against a pinned compiler defeats the point of pinning the compiler. Also corrected the [threat model](security/threat-model.md) tooling table, which listed `cargo-geiger` as running on every PR; it has only ever run nightly. ## 2026-10-12 diff --git a/knowledge/security/security-testing.md b/knowledge/security/security-testing.md index 5ffcf5bb3..6577cb5be 100644 --- a/knowledge/security/security-testing.md +++ b/knowledge/security/security-testing.md @@ -121,6 +121,15 @@ cancellation checkpoint without growing a buffer after rejection. Resource probes run in externally supervised subprocesses; full memory-exhaustion payloads are unnecessary. +## Pipeline admission fixtures + +`pipeline_admission_*` unit fixtures use ordinary two-stage no-output commands, +one occupied job slot, zero storage admission, and an inert environment value. +They assert typed errors, cancellation before construction, atomic shared job +admission, and lease/slot refunds. `pipeline_admission_tests` compares a tiny +export-isolation pipeline with real Bash and covers sequential zero-slot and +lastpipe behavior. These fixtures need no large exported values or fan-out. + ## Make Substitution Admission `builtins::make::expand::tests::subst_*` covers checked result-size arithmetic, diff --git a/knowledge/security/threat-model.md b/knowledge/security/threat-model.md index 514626b77..4e2c47c7d 100644 --- a/knowledge/security/threat-model.md +++ b/knowledge/security/threat-model.md @@ -295,7 +295,7 @@ on default 2 MiB test threads (TM-DOS-044). | 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-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. Producers atomically claim the session-wide job cap before construction, borrow the stage AST and lease owned fork metadata/future layouts before allocation. Construction checks shared work, cancellation and deadline budgets; group claims and leases drop with pipeline owners. Environment and immutable configuration stay Arc-shared. Stages never outlive the pipeline. Regressions: tiny `pipeline_admission_tests`, `pipeline_admission_*` ownership/cleanup units, and `pipeline_tests::threat_endless_producer_into_non_reader_ends`, `threat_recursive_pipeline_is_bounded`, `pipes-redirects.test.sh` SIGPIPE cases | **MITIGATED** | | TM-DOS-125 | Child-shell recursion stack overflow | A script that runs itself via `sh`/`bash` (`echo 'sh s.sh' > s.sh; sh s.sh`) nested interpreters until the process stack overflowed, aborting every tenant in the process | Each child shell counts against `max_function_depth` and a separate cap of `MAX_CHILD_SHELL_DEPTH` (8) nested shells, sized so the deepest mix fits a 2 MiB debug stack; over the cap the shell fails `maximum nesting depth exceeded` (exit 2). Forked jobs inherit the depth because they are polled on the caller's stack. Regression: `recursive_child_shell_is_bounded` | **MITIGATED** | | TM-DOS-126 | make amplification | Recursive variables (`X = $(X)$(X)`), self-including makefiles, `$(MAKE)` recursion, huge dependency graphs or `$(wildcard **)` fan-out turning one `make` call into unbounded CPU, memory or stack | Expansion depth 200 and 4 MiB per expansion; `$(subst)` computes its exact non-overlapping replacement size with checked arithmetic and admits it against the remaining allowance before construction. Arguments, replacement output and parent expansion buffers reserve capacity through the shared execution budget; immutable expanded variable values share storage and leases through scratch views and release them on replacement/drop. Matching/copying polls work, cancellation and deadlines. Tiny numeric admission and ordinary Unicode/assignment fixtures cover this ordering without oversized allocations (`subst_*`, `substitution_assignment_values_keep_and_release_leases`, `make_substitution_tests`). Recursive self-reference is fatal as in GNU; include depth 16; at most 100,000 targets, 10,000 `$(shell)`/`$(wildcard)`/`include` answers and 100,000 paths per `$(wildcard)` per run; `$(MAKE)` stops at MAKELEVEL 4, sized so the deepest recursion fits a 2 MiB debug thread stack (each level is roughly 200 KiB of make, `sh -c` and interpreter frames). Recipes run as ordinary `sh -c` commands under the session's budget, timeout and cancel token. Regressions: `builtins::make::expand::tests`, `l_make_003_sequential_sandbox_shell` | **MITIGATED** | | TM-DOS-127 | Pattern substitution amplification | `${x//pattern/rep}` with a glob pattern over a long value, or a huge pattern, turning one expansion into quadratic matching | Glob patterns compile to a size-capped (1 MiB) linear-time regex; extglob patterns fall back to at most 10,000 `glob_match` attempts and leave the value unchanged past the budget; results stay under `MAX_EXPANSION_RESULT_BYTES`. Regressions: `param-operand-quotes.test.sh`, `replace_pattern_extglob_budget` | **MITIGATED** |