Merge nucleic/sleek-thistle-egret-fyej into dev

nash-observe/src/lib.rs

@@ -25,6 +25,8 @@ const FLUSH_MAX_AGE: Duration = Duration::from_millis(500);
 const CHANNEL_BOUND: usize = 4096;
 const IO_TIMEOUT: Duration = Duration::from_millis(1500);
 const EXIT_FLUSH_TIMEOUT: Duration = Duration::from_millis(2000);
+/// Per-redirection / per-cmdsub preview cap in bytes (docs/NASH.md §5.4).
+const PREVIEW_CAP: usize = 64 * 1024;
 
 // ---------------------------------------------------------------------------
 // Event model (docs/NASH.md §6.1)
@@ -58,6 +60,28 @@ enum Event {
         reason: String,
         input: String,
     },
+    #[serde(rename = "redirect", rename_all = "camelCase")]
+    Redirect {
+        seq: u64,
+        ts: u64,
+        cmd_seq: u64,
+        op: String,
+        fd: u32,
+        target: Option<String>,
+        bytes: u64,
+        truncated: bool,
+        hash: String,
+        preview_b64: String,
+    },
+    #[serde(rename = "cmdsub", rename_all = "camelCase")]
+    Cmdsub {
+        seq: u64,
+        ts: u64,
+        bytes: u64,
+        truncated: bool,
+        hash: String,
+        preview_b64: String,
+    },
     #[serde(rename = "dropped", rename_all = "camelCase")]
     Dropped { seq: u64, ts: u64, count: u64 },
 }
@@ -141,6 +165,7 @@ struct PendingExec {
     argv: Vec<String>,
     cwd: PathBuf,
     started: Instant,
+    redirects: Vec<brush_core::gate::RedirectRecord>,
 }
 
 struct Observer {
@@ -185,6 +210,121 @@ fn now_millis() -> u64 {
         .unwrap_or(0)
 }
 
+// --- data-flow capture helpers (docs/NASH.md §5.2–5.4) ---------------------
+
+const B64: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
+
+/// Standard base64 (no external crate — the transport is a hand-rolled HTTP client).
+fn b64(input: &[u8]) -> String {
+    let mut out = String::with_capacity(input.len().div_ceil(3) * 4);
+    for chunk in input.chunks(3) {
+        let b = [
+            chunk[0],
+            *chunk.get(1).unwrap_or(&0),
+            *chunk.get(2).unwrap_or(&0),
+        ];
+        let n = (u32::from(b[0]) << 16) | (u32::from(b[1]) << 8) | u32::from(b[2]);
+        out.push(B64[((n >> 18) & 63) as usize] as char);
+        out.push(B64[((n >> 12) & 63) as usize] as char);
+        out.push(if chunk.len() > 1 { B64[((n >> 6) & 63) as usize] as char } else { '=' });
+        out.push(if chunk.len() > 2 { B64[(n & 63) as usize] as char } else { '=' });
+    }
+    out
+}
+
+/// 64-bit FNV-1a, hex — a cheap content fingerprint for the full (uncapped) data.
+fn fnv1a_hex(input: &[u8]) -> String {
+    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
+    for &byte in input {
+        h ^= u64::from(byte);
+        h = h.wrapping_mul(0x0000_0100_0000_01b3);
+    }
+    format!("{h:016x}")
+}
+
+/// Redact obvious credential shapes from a preview before it leaves the guest
+/// (docs/NASH.md §5.4). Best-effort pattern masking on whitespace-delimited tokens.
+fn redact(text: &str) -> String {
+    text.split_inclusive(|c: char| c.is_whitespace())
+        .map(|tok| {
+            let (word, trailer) = match tok.char_indices().rev().find(|(_, c)| !c.is_whitespace()) {
+                Some((i, c)) => tok.split_at(i + c.len_utf8()),
+                None => return tok.to_string(),
+            };
+            if looks_secret(word) {
+                format!("«redacted»{trailer}")
+            } else {
+                tok.to_string()
+            }
+        })
+        .collect()
+}
+
+fn looks_secret(word: &str) -> bool {
+    let w = word.trim_matches(|c: char| c == '"' || c == '\'' || c == ',');
+    // Common token prefixes (GitHub PATs, Slack, Stripe, AWS, OpenAI/Anthropic, …).
+    const PREFIXES: [&str; 10] =
+        ["ghp_", "gho_", "ghs_", "github_pat_", "xoxb-", "xoxp-", "sk-", "AKIA", "ASIA", "AIza"];
+    if PREFIXES.iter().any(|p| w.starts_with(p)) && w.len() >= 12 {
+        return true;
+    }
+    // Long high-entropy-ish blobs (base64/hex secrets).
+    if w.len() >= 32
+        && w.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '+' | '/' | '=' | '_' | '-'))
+        && w.chars().any(|c| c.is_ascii_digit())
+        && w.chars().any(|c| c.is_ascii_alphabetic())
+    {
+        return true;
+    }
+    false
+}
+
+/// The captured (bytes-total, capped-preview, truncated) for one redirect record.
+fn read_back(record: &brush_core::gate::RedirectRecord) -> Option<(u64, Vec<u8>, bool)> {
+    use brush_core::gate::RedirectReadback;
+    if let Some(inline) = &record.inline {
+        let full = inline.as_bytes();
+        let capped = full.len().min(PREVIEW_CAP);
+        return Some((full.len() as u64, full[..capped].to_vec(), full.len() > capped));
+    }
+    let path = record.path.as_ref()?;
+    // Only read back regular files; skip /dev/null, ttys, fifos, sockets, devices.
+    let meta = std::fs::metadata(path).ok()?;
+    if !meta.is_file() {
+        return None;
+    }
+    let size = meta.len();
+    let (offset, total) = match record.readback {
+        RedirectReadback::Append => (record.size_before, size.saturating_sub(record.size_before)),
+        RedirectReadback::Truncate | RedirectReadback::Input => (0, size),
+        RedirectReadback::Inline => return None,
+    };
+    let mut file = std::fs::File::open(path).ok()?;
+    if offset > 0 {
+        use std::io::Seek;
+        file.seek(std::io::SeekFrom::Start(offset)).ok()?;
+    }
+    let want = usize::try_from(total.min(PREVIEW_CAP as u64)).unwrap_or(0);
+    let mut buf = vec![0u8; want];
+    let n = std::io::Read::read(&mut file, &mut buf).unwrap_or(0);
+    buf.truncate(n);
+    Some((total, buf, total > n as u64))
+}
+
+/// Build a preview string (redacted, base64) from captured bytes. Binary data is
+/// previewed as a hex head rather than mangled UTF-8.
+fn preview_b64(data: &[u8]) -> String {
+    let looks_text = std::str::from_utf8(data).is_ok();
+    if looks_text {
+        // SAFETY-adjacent: validated above.
+        let redacted = redact(&String::from_utf8_lossy(data));
+        b64(redacted.as_bytes())
+    } else {
+        let hex: String = data.iter().take(1024).map(|b| format!("{b:02x}")).collect();
+        b64(format!("<binary> {hex}").as_bytes())
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Gate implementation (allow-all; docs/NASH.md §4.2)
 // ---------------------------------------------------------------------------
@@ -203,6 +343,7 @@ impl brush_core::gate::Gate for RecordingGate {
                         argv: ev.argv,
                         cwd: ev.cwd,
                         started: Instant::now(),
+                        redirects: ev.redirects,
                     },
                 );
             }
@@ -260,14 +401,63 @@ impl brush_core::gate::Gate for RecordingGate {
                 _ => {}
             }
 
+            let exec_seq = obs.seq();
             obs.send(Event::Exec {
-                seq: obs.seq(),
+                seq: exec_seq,
                 ts,
                 argv: pending.argv,
                 cwd: cwd_str,
                 exit_code: ev.exit_code,
                 duration_ms: u64::try_from(pending.started.elapsed().as_millis()).unwrap_or(0),
             });
+
+            // Data-flow read-back for this command's redirects (docs/NASH.md §5.2).
+            for record in &pending.redirects {
+                let Some((total, data, truncated)) = read_back(record) else {
+                    // Special file / unreadable — emit metadata only.
+                    obs.send(Event::Redirect {
+                        seq: obs.seq(),
+                        ts,
+                        cmd_seq: exec_seq,
+                        op: record.op.clone(),
+                        fd: record.fd,
+                        target: record.path.as_ref().map(|p| p.to_string_lossy().into_owned()),
+                        bytes: 0,
+                        truncated: false,
+                        hash: String::new(),
+                        preview_b64: String::new(),
+                    });
+                    continue;
+                };
+                obs.send(Event::Redirect {
+                    seq: obs.seq(),
+                    ts,
+                    cmd_seq: exec_seq,
+                    op: record.op.clone(),
+                    fd: record.fd,
+                    target: record.path.as_ref().map(|p| p.to_string_lossy().into_owned()),
+                    bytes: total,
+                    truncated,
+                    hash: fnv1a_hex(&data),
+                    preview_b64: preview_b64(&data),
+                });
+            }
+        }));
+    }
+
+    fn on_cmdsub(&self, output: &str) {
+        let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            let Some(obs) = OBSERVER.get() else { return };
+            let full = output.as_bytes();
+            let capped = full.len().min(PREVIEW_CAP);
+            obs.send(Event::Cmdsub {
+                seq: obs.seq(),
+                ts: now_millis(),
+                bytes: full.len() as u64,
+                truncated: full.len() > capped,
+                hash: fnv1a_hex(full),
+                preview_b64: preview_b64(&full[..capped]),
+            });
         }));
     }
 }