diff --git a/NUCLEIC_FORK.md b/NUCLEIC_FORK.md index a511ec3..c9f6eab 100644 --- a/NUCLEIC_FORK.md +++ b/NUCLEIC_FORK.md @@ -22,6 +22,8 @@ Every fork change is marked with a `// nash:` comment. Current patches: | `brush-core/src/interp.rs` | `ExecutionParameters.nash_redirects` accumulator; `setup_redirect` records file (`>`,`>>`,`<`,`2>`,`&>`,`<>`), heredoc, and here-string redirects | | `brush-core/src/commands.rs` | gate wrapper drains `nash_redirects` into `ExecStart` for post-run read-back | | `brush-core/src/expansion.rs` | command-substitution arm calls `gate().on_cmdsub` with the trimmed output | +| `brush-core/src/gate.rs` | pipe tap (docs/NASH.md §5.3): `PipeEvent` + `Gate::on_pipe` default hook + `is_active()`/`next_pipeline_id()` gating | +| `brush-core/src/interp.rs` | `spawn_pipeline_processes` interposes a tee on each `a \| b` link when observation is active (`nash_spawn_pipe_tee`): a copier thread mirrors a bounded prefix, preserving backpressure and SIGPIPE, reports before closing the write end. No tee on the unobserved fast path. | Verification (this container): brush compat suite `2067 ran: 1684 succeeded, 6 failed, 377 known to fail, 38 skipped` — the 6 failures are identical to a @@ -30,11 +32,9 @@ pristine `brush-shell-v0.4.0` baseline run (environmental: ANSI-C quote and zero fork-caused regressions; +3 successes vs baseline are the repaired IFS tests. nash corpus: 94/94. -Planned (M2 remainder, per docs/NASH.md §5.3): inline tees on the pipe wiring -in `interp.rs::spawn_pipeline_processes` (the `std::io::pipe()` links between -`a | b` stages) — the one data-flow channel not yet tapped. Deferred as the most -invasive change (async copier + backpressure + SIGPIPE); regular-file -redirections, heredoc/here-string, and command substitution are all captured. +Planned (M3, per docs/NASH.md / NAROS.md): image-side delivery — nash baked and +forced on every Linux surface via narOS. No further brush-core patches expected +for M3; enforcement (`Hold`/`Deny` verdicts) remains post-M6. Updating: diff against the upstream tag, re-vendor, re-apply `// nash:` patches (grep for the marker), then re-run `shell/corpus/replay.py` and the brush compat diff --git a/brush-core/src/gate.rs b/brush-core/src/gate.rs index 6794b0a..bf53e24 100644 --- a/brush-core/src/gate.rs +++ b/brush-core/src/gate.rs @@ -8,6 +8,7 @@ //! any further changes to this crate. use std::path::PathBuf; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::OnceLock; /// Details about a simple command about to be executed (builtin, function, or @@ -62,6 +63,24 @@ pub struct ExecEnd { pub exit_code: i32, } +/// Data observed flowing across one pipe link `a | b` (docs/NASH.md §5.3). +/// Reported once the link drains (producer closed or consumer went away). +#[derive(Clone, Debug)] +pub struct PipeEvent { + /// Correlates the links of one pipeline invocation. + pub pipeline_id: u64, + /// Zero-based index of the producing stage. + pub from_index: usize, + /// Zero-based index of the consuming stage. + pub to_index: usize, + /// Total bytes that crossed the link. + pub total_bytes: u64, + /// A bounded prefix of the bytes (observer applies its own caps/redaction). + pub captured: Vec, + /// Whether `total_bytes` exceeded `captured`. + pub truncated: bool, +} + /// The gate's decision for a command about to execute. pub enum Verdict { /// Run the command; the token is echoed back via `on_exit`. @@ -87,6 +106,9 @@ pub trait Gate: Send + Sync { /// nash: called with the (trimmed) output of a command substitution /// `$(…)` / backticks (docs/NASH.md §5.3). Default no-op. fn on_cmdsub(&self, _output: &str) {} + /// nash: called once a pipe link `a | b` drains (docs/NASH.md §5.3). + /// Invoked from the tee copier thread. Default no-op. + fn on_pipe(&self, _ev: PipeEvent) {} } struct AllowAll; @@ -111,3 +133,16 @@ pub fn set_gate(gate: Box) { pub(crate) fn gate() -> &'static dyn Gate { GATE.get().map_or(&ALLOW_ALL as &dyn Gate, AsRef::as_ref) } + +/// Whether a real (non-default) gate is installed. Pipe teeing — which adds a +/// thread + extra pipe per link — is only done when this is true (docs/NASH.md §5.3). +pub(crate) fn is_active() -> bool { + GATE.get().is_some() +} + +static PIPELINE_COUNTER: AtomicU64 = AtomicU64::new(1); + +/// A fresh id correlating the links of one pipeline invocation. +pub(crate) fn next_pipeline_id() -> u64 { + PIPELINE_COUNTER.fetch_add(1, Ordering::Relaxed) +} diff --git a/brush-core/src/interp.rs b/brush-core/src/interp.rs index ce15ff9..80727dd 100644 --- a/brush-core/src/interp.rs +++ b/brush-core/src/interp.rs @@ -460,7 +460,28 @@ async fn spawn_pipeline_processes( // Create pipes to use between commands, but only bother doing so if there's more than one // command. if pipeline_len > 1 { - for _ in 0..(pipeline_len - 1) { + // nash: when observation is active, interpose a tee on each link so the + // bytes flowing `a | b` are captured (docs/NASH.md §5.3). Off otherwise — + // no extra pipe/thread on the unobserved fast path. + let nash_pipeline_id = if crate::gate::is_active() { + Some(crate::gate::next_pipeline_id()) + } else { + None + }; + for link in 0..(pipeline_len - 1) { + if let Some(pipeline_id) = nash_pipeline_id { + // Producer writes `a_write`; a copier moves bytes to `b_write`; + // consumer reads `b_read`. Slots stay identical to the untapped + // case, so the pop-ordering below is unchanged. + let (a_read, a_write) = std::io::pipe()?; + let (b_read, b_write) = std::io::pipe()?; + // Link `link` connects producer stage (len-2-link) → consumer (len-1-link). + let from_index = pipeline_len - 2 - link; + nash_spawn_pipe_tee(a_read, b_write, pipeline_id, from_index, from_index + 1); + pipe_readers.push(Some(openfiles::OpenFile::PipeReader(b_read))); + pipe_writers.push(Some(openfiles::OpenFile::PipeWriter(a_write))); + continue; + } let (reader, writer) = std::io::pipe()?; pipe_readers.push(Some(reader.into())); pipe_writers.push(Some(writer.into())); @@ -1888,6 +1909,69 @@ pub(crate) async fn setup_redirect( Ok(()) } +// nash: bounded prefix captured per pipe link before reporting (docs/NASH.md §5.3). +const NASH_PIPE_CAPTURE_CAP: usize = 64 * 1024; + +// nash: spawn a copier that moves bytes from `src` (producer's output) to `dst` +// (consumer's input), mirroring a bounded prefix, then reports the link to the +// gate (docs/NASH.md §5.3). The synchronous copy preserves pipe backpressure; +// EOF on `src` or EPIPE on `dst` (consumer gone, e.g. `yes | head`) ends it and +// closes `dst` so the consumer sees EOF. +fn nash_spawn_pipe_tee( + mut src: std::io::PipeReader, + mut dst: std::io::PipeWriter, + pipeline_id: u64, + from_index: usize, + to_index: usize, +) { + std::thread::Builder::new() + .name("nash-pipe-tee".into()) + .spawn(move || { + use std::io::{Read, Write}; + let mut buf = [0u8; 32 * 1024]; + let mut captured: Vec = Vec::new(); + let mut total: u64 = 0; + let mut truncated = false; + loop { + let n = match src.read(&mut buf) { + Ok(0) => break, // producer closed → done + Ok(n) => n, + Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => continue, + Err(_) => break, + }; + total += n as u64; + if captured.len() < NASH_PIPE_CAPTURE_CAP { + let room = NASH_PIPE_CAPTURE_CAP - captured.len(); + let take = room.min(n); + captured.extend_from_slice(&buf[..take]); + if take < n { + truncated = true; + } + } else { + truncated = true; + } + // Pass through; if the consumer went away, stop (its EOF is enough). + if dst.write_all(&buf[..n]).is_err() { + break; + } + } + // Report BEFORE closing `dst`: dropping it unblocks the consumer, which + // lets the shell reach its exit-flush — so the event must already be + // enqueued to avoid a lost-event race on short-lived shells. + crate::gate::gate().on_pipe(crate::gate::PipeEvent { + pipeline_id, + from_index, + to_index, + total_bytes: total, + captured, + truncated, + }); + // Now close the write end so the consumer sees EOF. + drop(dst); + }) + .ok(); +} + // nash: record a regular-file redirect for post-run read-back (docs/NASH.md §5.2a). // The child still gets the real fd; this only notes what to read back afterward. fn nash_record_file_redirect(