diff --git a/crates/nono-cli/src/audit_integrity.rs b/crates/nono-cli/src/audit_integrity.rs index 7bdd53afc..118a4b06f 100644 --- a/crates/nono-cli/src/audit_integrity.rs +++ b/crates/nono-cli/src/audit_integrity.rs @@ -6,6 +6,10 @@ use sha2::{Digest, Sha256}; use std::fs::{File, OpenOptions}; use std::io::{BufRead, BufReader, Write}; use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::{sync_channel, Receiver, SyncSender, TrySendError}; +use std::sync::{Arc, Mutex}; +use std::thread::JoinHandle; pub(crate) const AUDIT_EVENTS_FILENAME: &str = "audit-events.ndjson"; const EVENT_DOMAIN: &[u8] = b"nono.audit.event.alpha\n"; @@ -189,6 +193,199 @@ impl AuditRecorder { } } +/// Bounded queue capacity for the audit writer channel. +/// +/// A typical event is a few hundred bytes; this bound caps in-flight memory +/// well under 10 MB even for the largest events. Bursts from a runaway +/// agent fit here; sustained overflow drops events with rate-limited +/// `warn!`s and bumps `dropped_events`. +const AUDIT_WRITER_QUEUE: usize = 16_384; + +enum WriterCmd { + /// A network event to append to the recorder. Dropped silently by the + /// writer once `done` has been set (see `RecorderStreamingSink` docs). + Event(NetworkAuditEvent), + /// Drain marker. The writer flips `done` and acks via the included + /// `SyncSender<()>`; `flush()` blocks on the ack to provide a + /// synchronous "everything before this is on disk" barrier. + Flush(SyncSender<()>), + /// Terminate the writer loop. Sent from `Drop`. + Shutdown, +} + +/// Streaming sink that forwards each network audit event into an +/// `AuditRecorder` via a bounded channel drained by a dedicated writer +/// thread, instead of buffering until session exit. +/// +/// `record()` is non-blocking: it `try_send`s into the queue and returns +/// immediately. The writer thread is the sole owner of the recorder mutex +/// on the audit path, so concurrent proxied connections never contend on +/// disk I/O. Order is preserved because a single consumer drains FIFO, +/// which is required by the recorder's hash chain and Merkle accumulator. +/// +/// `flush()` seals the sink: once it returns, the writer has marked itself +/// `done` and silently drops any event that was enqueued during the +/// `AuditLog::close` race window. This guarantees the caller can append +/// further records (e.g. `session_ended`) directly to the same recorder +/// without the writer racing them through the recorder mutex and +/// extending the file past the Merkle root captured in `SessionMetadata`. +/// +/// Errors are logged (not propagated) because audit recording must never +/// abort an in-flight network request — a poisoned mutex or transient I/O +/// failure should drop a single event rather than break the proxy. +pub(crate) struct RecorderStreamingSink { + tx: SyncSender, + /// Count of events rejected because the queue was full or the writer + /// was gone. Surfaced via `dropped_events()` and a final `warn!` on + /// `Drop` so the operator never silently loses count. + dropped: AtomicU64, + /// Set by the writer when it processes a `Flush` command. Subsequent + /// `Event`s are silently ignored — see struct-level docs. + done: Arc, + writer: Mutex>>, +} + +impl RecorderStreamingSink { + pub(crate) fn new(recorder: Arc>) -> Result { + Self::with_capacity(recorder, AUDIT_WRITER_QUEUE) + } + + fn with_capacity(recorder: Arc>, capacity: usize) -> Result { + let (tx, rx) = sync_channel::(capacity); + let done = Arc::new(AtomicBool::new(false)); + let done_writer = Arc::clone(&done); + let writer = std::thread::Builder::new() + .name("nono-audit-writer".to_string()) + .spawn(move || writer_loop(recorder, rx, done_writer)) + .map_err(|e| { + NonoError::Snapshot(format!("Failed to spawn audit writer thread: {e}")) + })?; + Ok(Self { + tx, + dropped: AtomicU64::new(0), + done, + writer: Mutex::new(Some(writer)), + }) + } + + #[cfg(test)] + pub(crate) fn new_with_capacity( + recorder: Arc>, + capacity: usize, + ) -> Result { + Self::with_capacity(recorder, capacity) + } + + /// Number of events dropped because the bounded queue was full or the + /// writer thread had already terminated. Exposed mainly for tests and + /// for surfacing in diagnostics; the final value is logged on `Drop`. + pub(crate) fn dropped_events(&self) -> u64 { + self.dropped.load(Ordering::Relaxed) + } +} + +fn writer_loop( + recorder: Arc>, + rx: Receiver, + done: Arc, +) { + while let Ok(cmd) = rx.recv() { + match cmd { + WriterCmd::Event(event) => { + // Once Flush has sealed the sink, any Event that lost the + // is_closed/record race in `push_event` and reached the + // queue after the seal must NOT be appended — doing so + // would extend the file past the Merkle root that + // `finalize_supervised_exit` is about to commit, breaking + // `verify_audit_log`. + if done.load(Ordering::Acquire) { + continue; + } + match recorder.lock() { + Ok(mut r) => { + if let Err(e) = r.record_network_event(event) { + tracing::warn!("Audit writer: record_network_event failed: {}", e); + } + } + Err(e) => { + tracing::warn!("Audit writer: recorder mutex poisoned: {}", e); + } + } + } + WriterCmd::Flush(ack) => { + // Set `done` *before* acking so that any concurrent + // producer that enqueues an Event after this Flush + // observes the seal by the time the writer dequeues it. + done.store(true, Ordering::Release); + let _ = ack.send(()); + } + WriterCmd::Shutdown => break, + } + } +} + +impl nono_proxy::audit::NetworkAuditSink for RecorderStreamingSink { + fn record(&self, event: &NetworkAuditEvent) { + match self.tx.try_send(WriterCmd::Event(event.clone())) { + Ok(()) => {} + Err(TrySendError::Full(_)) => { + let prev = self.dropped.fetch_add(1, Ordering::Relaxed); + if prev == 0 || prev.is_power_of_two() { + tracing::warn!( + "Audit writer queue full (cap {}); dropped {} events so far", + AUDIT_WRITER_QUEUE, + prev.saturating_add(1) + ); + } + } + Err(TrySendError::Disconnected(_)) => { + let prev = self.dropped.fetch_add(1, Ordering::Relaxed); + if prev == 0 || prev.is_power_of_two() { + tracing::warn!( + "Audit writer thread terminated; dropped {} events so far", + prev.saturating_add(1) + ); + } + } + } + } + + fn flush(&self) { + let (ack_tx, ack_rx) = sync_channel::<()>(1); + if self.tx.send(WriterCmd::Flush(ack_tx)).is_err() { + tracing::warn!("Audit writer thread is gone; flush is a no-op"); + // Seal locally so callers that observe a successful flush + // (even degraded) cannot have late events appended — there + // is no writer to append anything, but the contract still + // holds. + self.done.store(true, Ordering::Release); + return; + } + if ack_rx.recv().is_err() { + tracing::warn!("Audit writer disconnected before flush ack"); + self.done.store(true, Ordering::Release); + } + } +} + +impl Drop for RecorderStreamingSink { + fn drop(&mut self) { + let dropped = self.dropped_events(); + if dropped > 0 { + tracing::warn!( + "Audit writer sink dropped {} events total over the session", + dropped + ); + } + let _ = self.tx.send(WriterCmd::Shutdown); + if let Ok(mut writer) = self.writer.lock() { + if let Some(h) = writer.take() { + let _ = h.join(); + } + } + } +} + fn hash_event(event_bytes: &[u8]) -> ContentHash { let mut hasher = Sha256::new(); hasher.update(EVENT_DOMAIN); @@ -557,6 +754,369 @@ mod tests { assert!(verified.records_verified); } + #[test] + fn recorder_streaming_sink_writes_events_to_ndjson() { + use nono_proxy::audit::NetworkAuditSink; + + let dir = tempfile::tempdir().unwrap(); + let recorder = Arc::new(Mutex::new( + AuditRecorder::new(dir.path().to_path_buf()).unwrap(), + )); + recorder + .lock() + .unwrap() + .record_session_started("2026-04-21T00:00:00Z".to_string(), vec!["pwd".to_string()]) + .unwrap(); + + let sink = RecorderStreamingSink::new(Arc::clone(&recorder)).unwrap(); + sink.record(&NetworkAuditEvent { + timestamp_unix_ms: 1000, + mode: NetworkAuditMode::Connect, + decision: NetworkAuditDecision::Allow, + route_id: None, + auth_mechanism: None, + auth_outcome: None, + managed_credential_active: None, + injection_mode: None, + denial_category: None, + target: "api.example.com".to_string(), + port: Some(443), + method: Some("CONNECT".to_string()), + path: None, + status: None, + reason: None, + }); + sink.record(&NetworkAuditEvent { + timestamp_unix_ms: 2000, + mode: NetworkAuditMode::Connect, + decision: NetworkAuditDecision::Deny, + route_id: None, + auth_mechanism: None, + auth_outcome: None, + managed_credential_active: None, + injection_mode: None, + denial_category: None, + target: "evil.example".to_string(), + port: Some(443), + method: None, + path: None, + status: None, + reason: Some("blocked".to_string()), + }); + // Drain the writer queue before appending session_ended directly. + // In production this happens via AuditLog::close -> sink.flush(). + sink.flush(); + + recorder + .lock() + .unwrap() + .record_session_ended("2026-04-21T00:00:02Z".to_string(), 0) + .unwrap(); + + let summary = recorder.lock().unwrap().finalize().unwrap(); + assert_eq!( + summary.event_count, 4, + "session_started + 2 network + session_ended" + ); + let verified = verify_audit_log(dir.path(), Some(&summary)).unwrap(); + assert_eq!(verified.event_count, 4); + assert!(verified.records_verified); + } + + #[test] + fn recorder_streaming_sink_preserves_order_under_concurrent_writers() { + use nono_proxy::audit::NetworkAuditSink; + + let dir = tempfile::tempdir().unwrap(); + let recorder = Arc::new(Mutex::new( + AuditRecorder::new(dir.path().to_path_buf()).unwrap(), + )); + recorder + .lock() + .unwrap() + .record_session_started("2026-04-21T00:00:00Z".to_string(), vec!["pwd".to_string()]) + .unwrap(); + + let sink = Arc::new(RecorderStreamingSink::new(Arc::clone(&recorder)).unwrap()); + + // Simulate the proxy hot path: many concurrent senders, none of + // which should block waiting on the recorder mutex. With sync I/O + // inside record() this would serialize through one lock + one + // file.flush() per event. + let threads: usize = 8; + let per_thread: usize = 100; + let handles: Vec<_> = (0..threads) + .map(|t| { + let sink = Arc::clone(&sink); + std::thread::spawn(move || { + for i in 0..per_thread { + sink.record(&NetworkAuditEvent { + timestamp_unix_ms: (t * per_thread + i) as u64, + mode: NetworkAuditMode::Connect, + decision: NetworkAuditDecision::Allow, + route_id: None, + auth_mechanism: None, + auth_outcome: None, + managed_credential_active: None, + injection_mode: None, + denial_category: None, + target: format!("t{t}.example"), + port: Some(443), + method: Some("CONNECT".to_string()), + path: None, + status: None, + reason: None, + }); + } + }) + }) + .collect(); + for h in handles { + h.join().unwrap(); + } + + sink.flush(); + + recorder + .lock() + .unwrap() + .record_session_ended("2026-04-21T00:00:02Z".to_string(), 0) + .unwrap(); + + let summary = recorder.lock().unwrap().finalize().unwrap(); + let expected = 1 + threads * per_thread + 1; + assert_eq!(summary.event_count as usize, expected); + // The recorder builds its hash chain by appending in the order it + // receives events. verify_audit_log re-derives that chain from the + // file and refuses to validate if any link is broken — so a passing + // verification proves the writer thread preserved FIFO order. + let verified = verify_audit_log(dir.path(), Some(&summary)).unwrap(); + assert_eq!(verified.event_count as usize, expected); + assert!(verified.records_verified); + } + + fn make_audit_event(target: &str, ts: u64) -> NetworkAuditEvent { + NetworkAuditEvent { + timestamp_unix_ms: ts, + mode: NetworkAuditMode::Connect, + decision: NetworkAuditDecision::Allow, + route_id: None, + auth_mechanism: None, + auth_outcome: None, + managed_credential_active: None, + injection_mode: None, + denial_category: None, + target: target.to_string(), + port: Some(443), + method: Some("CONNECT".to_string()), + path: None, + status: None, + reason: None, + } + } + + #[test] + fn recorder_streaming_sink_drops_events_when_queue_full() { + use nono_proxy::audit::NetworkAuditSink; + + let dir = tempfile::tempdir().unwrap(); + let recorder = Arc::new(Mutex::new( + AuditRecorder::new(dir.path().to_path_buf()).unwrap(), + )); + recorder + .lock() + .unwrap() + .record_session_started("2026-04-21T00:00:00Z".to_string(), vec!["pwd".to_string()]) + .unwrap(); + + // Block the writer thread by holding the recorder mutex. The + // writer will dequeue at most one event before parking on the + // recorder lock; everything else queues up to `capacity`, then + // try_send starts returning Full and the sink increments + // `dropped`. + let blocker = recorder.lock().unwrap(); + let capacity: usize = 4; + let sink = + RecorderStreamingSink::new_with_capacity(Arc::clone(&recorder), capacity).unwrap(); + + let burst: usize = 200; + for i in 0..burst { + sink.record(&make_audit_event("api.example.com", i as u64)); + } + + // The sink may not have observed every Full *yet* — the writer + // is still parked. Just assert that the writer's drop accounting + // saw at least some drops. + let dropped_during_burst = sink.dropped_events(); + assert!( + dropped_during_burst > 0, + "expected the bounded queue to drop events when writer is parked, got 0" + ); + + // Releasing the blocker lets the writer drain whatever made it + // into the queue. + drop(blocker); + sink.flush(); + + recorder + .lock() + .unwrap() + .record_session_ended("2026-04-21T00:00:02Z".to_string(), 0) + .unwrap(); + + let summary = recorder.lock().unwrap().finalize().unwrap(); + // session_started + at most (capacity + 1 in-flight on the + // writer) network events + session_ended. + let max_recorded = 1 + capacity + 1 + 1; + assert!( + (summary.event_count as usize) <= max_recorded, + "wrote {} events; expected at most {} given queue cap {}", + summary.event_count, + max_recorded, + capacity + ); + let verified = verify_audit_log(dir.path(), Some(&summary)).unwrap(); + assert!(verified.records_verified); + // dropped_events + recorded_network_events should match the burst. + let recorded_network = (summary.event_count as usize).saturating_sub(2); + assert_eq!( + sink.dropped_events() as usize + recorded_network, + burst, + "every event must be either recorded or counted as dropped" + ); + } + + #[test] + fn recorder_streaming_sink_drops_post_flush_events() { + use nono_proxy::audit::NetworkAuditSink; + + let dir = tempfile::tempdir().unwrap(); + let recorder = Arc::new(Mutex::new( + AuditRecorder::new(dir.path().to_path_buf()).unwrap(), + )); + recorder + .lock() + .unwrap() + .record_session_started("2026-04-21T00:00:00Z".to_string(), vec!["pwd".to_string()]) + .unwrap(); + + let sink = RecorderStreamingSink::new(Arc::clone(&recorder)).unwrap(); + sink.record(&make_audit_event("before.example", 1)); + sink.flush(); + let count_after_flush = recorder.lock().unwrap().event_count(); + + // After flush() returns the sink is sealed: events submitted now + // must be silently dropped, otherwise they would land in the file + // after the Merkle root that finalize_supervised_exit is about to + // commit, breaking verify_audit_log. + for i in 0..20 { + sink.record(&make_audit_event("after.example", 100 + i)); + } + + // Give the writer enough wall-clock time to *try* appending the + // post-flush events if the seal were missing. + std::thread::sleep(std::time::Duration::from_millis(50)); + + let count_now = recorder.lock().unwrap().event_count(); + assert_eq!( + count_now, count_after_flush, + "post-flush events must not extend the audit log" + ); + + // Append session_ended directly — this is the production pattern + // that the seal protects. + recorder + .lock() + .unwrap() + .record_session_ended("2026-04-21T00:00:02Z".to_string(), 0) + .unwrap(); + let summary = recorder.lock().unwrap().finalize().unwrap(); + // session_started + 1 network + session_ended = 3. + assert_eq!(summary.event_count, 3); + let verified = verify_audit_log(dir.path(), Some(&summary)).unwrap(); + assert!(verified.records_verified); + } + + #[test] + fn recorder_streaming_sink_flush_is_idempotent() { + use nono_proxy::audit::NetworkAuditSink; + + let dir = tempfile::tempdir().unwrap(); + let recorder = Arc::new(Mutex::new( + AuditRecorder::new(dir.path().to_path_buf()).unwrap(), + )); + recorder + .lock() + .unwrap() + .record_session_started("2026-04-21T00:00:00Z".to_string(), vec!["pwd".to_string()]) + .unwrap(); + + let sink = RecorderStreamingSink::new(Arc::clone(&recorder)).unwrap(); + sink.record(&make_audit_event("api.example.com", 1)); + sink.flush(); + let after_first = recorder.lock().unwrap().event_count(); + + // Second flush on an already-sealed sink must not panic, deadlock, + // or otherwise affect the recorder. The writer keeps acking + // Flush commands; the seal is sticky. + sink.flush(); + sink.flush(); + let after_extra = recorder.lock().unwrap().event_count(); + assert_eq!( + after_first, after_extra, + "extra flushes must not change the recorder state" + ); + + // Late events stay dropped, as before. + sink.record(&make_audit_event("late.example", 999)); + std::thread::sleep(std::time::Duration::from_millis(20)); + assert_eq!(after_extra, recorder.lock().unwrap().event_count()); + } + + #[test] + fn recorder_streaming_sink_drop_joins_writer_cleanly() { + use nono_proxy::audit::NetworkAuditSink; + + let dir = tempfile::tempdir().unwrap(); + let recorder = Arc::new(Mutex::new( + AuditRecorder::new(dir.path().to_path_buf()).unwrap(), + )); + recorder + .lock() + .unwrap() + .record_session_started("2026-04-21T00:00:00Z".to_string(), vec!["pwd".to_string()]) + .unwrap(); + + { + let sink = RecorderStreamingSink::new(Arc::clone(&recorder)).unwrap(); + for i in 0..5 { + sink.record(&make_audit_event("api.example.com", i)); + } + sink.flush(); + // sink dropped here: writer thread receives Shutdown and + // joins synchronously inside Drop. After this block the + // recorder is sole-owned by `recorder`. + } + + // If the writer were still alive, this Arc count would be 2. + assert_eq!( + Arc::strong_count(&recorder), + 1, + "writer thread must release its Arc clone on shutdown" + ); + + // Recorder is fully usable post-shutdown — no leftover state. + recorder + .lock() + .unwrap() + .record_session_ended("2026-04-21T00:00:02Z".to_string(), 0) + .unwrap(); + let summary = recorder.lock().unwrap().finalize().unwrap(); + assert_eq!(summary.event_count, 1 + 5 + 1); + let verified = verify_audit_log(dir.path(), Some(&summary)).unwrap(); + assert!(verified.records_verified); + } + #[test] fn verifier_rejects_alpha_records_missing_event_json() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/nono-cli/src/rollback_runtime.rs b/crates/nono-cli/src/rollback_runtime.rs index 2dcda897c..484c8edb6 100644 --- a/crates/nono-cli/src/rollback_runtime.rs +++ b/crates/nono-cli/src/rollback_runtime.rs @@ -473,16 +473,32 @@ pub(crate) fn finalize_supervised_exit(ctx: RollbackExitContext<'_>) -> Result<( rollback_prompt_disabled, } = ctx; + // Stop accepting further proxy audit events before draining and recording + // session_ended. Any in-flight response still being audited at this point + // would otherwise extend the audit file past the Merkle root we are about + // to commit to session metadata. + if let Some(handle) = proxy_handle { + handle.close_audit_log(); + } let mut network_events = proxy_handle.map_or_else( Vec::new, nono_proxy::server::ProxyHandle::drain_audit_events, ); + // When the proxy was streaming events into the recorder live, the drained + // buffer is already in the file — re-recording would produce duplicates + // and break the chain hash. The buffer is still kept for SessionMetadata + // below as a snapshot of network activity. + let streaming_already_recorded = proxy_handle + .map(nono_proxy::server::ProxyHandle::audit_streaming_active) + .unwrap_or(false); let (audit_event_count, audit_integrity) = if let Some(recorder_mutex) = audit_recorder { let mut recorder = recorder_mutex .lock() .map_err(|_| nono::NonoError::Snapshot("Audit recorder lock poisoned".to_string()))?; - for event in &network_events { - recorder.record_network_event(event.clone())?; + if !streaming_already_recorded { + for event in &network_events { + recorder.record_network_event(event.clone())?; + } } recorder.record_session_ended(ended.to_string(), exit_code)?; let event_count = recorder.event_count(); diff --git a/crates/nono-cli/src/supervised_runtime.rs b/crates/nono-cli/src/supervised_runtime.rs index b791449d6..5387f0072 100644 --- a/crates/nono-cli/src/supervised_runtime.rs +++ b/crates/nono-cli/src/supervised_runtime.rs @@ -15,7 +15,7 @@ use colored::Colorize; use nono::undo::ExecutableIdentity; use nono::{CapabilitySet, Result}; use std::io::IsTerminal; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; struct SessionRuntimeState { started: String, @@ -202,12 +202,14 @@ pub(crate) fn execute_supervised_runtime(ctx: SupervisedRuntimeContext<'_>) -> R } else { None }; - let audit_recorder = if audit_state.is_some() && !rollback.no_audit_integrity { + let audit_recorder: Option>> = if audit_state.is_some() + && !rollback.no_audit_integrity + { audit_state .as_ref() .map(|state| { AuditRecorder::new_with_policy(state.session_dir.clone(), redaction_policy.clone()) - .map(Mutex::new) + .map(|r| Arc::new(Mutex::new(r))) }) .transpose()? } else { @@ -220,6 +222,35 @@ pub(crate) fn execute_supervised_runtime(ctx: SupervisedRuntimeContext<'_>) -> R recorder.record_session_started(started.clone(), command.to_vec())?; } + // Stream network audit events into the recorder as they occur. Without + // this, events sit in the proxy's in-memory buffer until session exit and + // are lost on crash or once `MAX_AUDIT_EVENTS` is reached. + // + // Streaming is best-effort: if the writer thread fails to spawn we log + // and continue with the legacy "drain at session_ended" path rather than + // aborting the run. The session still gets a complete audit log on + // clean shutdown; only the crash-durability and unbounded-buffer wins + // are forfeited. + if let (Some(recorder_mutex), Some(handle)) = (audit_recorder.as_ref(), proxy_handle) { + match crate::audit_integrity::RecorderStreamingSink::new(Arc::clone(recorder_mutex)) { + Ok(sink) => { + let sink: Arc = Arc::new(sink); + if handle.set_audit_streaming_sink(sink).is_err() { + tracing::warn!( + "Network audit streaming sink already attached; refusing to overwrite" + ); + } + } + Err(e) => { + tracing::warn!( + "Failed to start network audit streaming writer; continuing without live \ + streaming (events will still be drained at session end): {}", + e + ); + } + } + } + let protected_roots = protected_paths::ProtectedRoots::from_defaults()?; let approval_backend = terminal_approval::TerminalApproval; let supervisor_session_id = build_supervisor_session_id(audit_state.as_ref()); @@ -231,7 +262,7 @@ pub(crate) fn execute_supervised_runtime(ctx: SupervisedRuntimeContext<'_>) -> R detach_sequence: session.detach_sequence.as_deref(), open_url_origins: &proxy.open_url_origins, open_url_allow_localhost: proxy.open_url_allow_localhost, - audit_recorder: audit_recorder.as_ref(), + audit_recorder: audit_recorder.as_deref(), redaction_policy, allow_launch_services_active: proxy.allow_launch_services_active, #[cfg(target_os = "linux")] @@ -270,7 +301,7 @@ pub(crate) fn execute_supervised_runtime(ctx: SupervisedRuntimeContext<'_>) -> R rollback_state, audit_snapshot_state, audit_tracked_paths, - audit_recorder: audit_recorder.as_ref(), + audit_recorder: audit_recorder.as_deref(), audit_integrity_enabled: !rollback.no_audit_integrity, proxy_handle, executable_identity, diff --git a/crates/nono-proxy/src/audit.rs b/crates/nono-proxy/src/audit.rs index ec4b8d552..209f018ad 100644 --- a/crates/nono-proxy/src/audit.rs +++ b/crates/nono-proxy/src/audit.rs @@ -8,15 +8,114 @@ use nono::undo::{ NetworkAuditAuthMechanism, NetworkAuditAuthOutcome, NetworkAuditDecision, NetworkAuditDenialCategory, NetworkAuditEvent, NetworkAuditInjectionMode, NetworkAuditMode, }; -use std::sync::{Arc, Mutex}; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; use std::time::{SystemTime, UNIX_EPOCH}; use tracing::{info, warn}; /// Maximum number of in-memory network audit events kept per proxy session. const MAX_AUDIT_EVENTS: usize = 4096; +/// Sink for streaming network audit events to a durable destination as they +/// occur. Implementations must be cheap to call from async request handlers +/// and must not panic; failures should be logged and swallowed so audit +/// recording cannot break network operations. +/// +/// `record` is the hot path and must not block. Implementations that buffer +/// or offload writes asynchronously must drain their buffer in `flush`, +/// which the audit log calls during `close` so that the caller can safely +/// append further events (e.g. a `session_ended` record) directly to the +/// underlying store without racing the sink's background writer. +pub trait NetworkAuditSink: Send + Sync { + fn record(&self, event: &NetworkAuditEvent); + + /// Drain buffered events and seal the sink against late writes. + /// + /// Returns only after every event passed to `record` *before this call + /// started* is durable in the sink's destination. + /// + /// After `flush()` returns, implementations are permitted to silently + /// drop any event submitted via subsequent `record()` calls. This lets + /// the caller append further records to the same store without racing + /// the sink's background writer. The `close()` race is unavoidable — + /// a producer thread may pass the `is_closed()` check, lose the CPU, + /// and only enqueue its event after this method has been called — so + /// honouring the seal is how implementations preserve any external + /// integrity boundary (hash chains, content roots) anchored at the + /// flush point. + /// + /// Default no-op for sinks that write synchronously inside `record`. + fn flush(&self) {} +} + +/// Shared sink for network audit events. +/// +/// Holds an in-memory buffer (used to populate `SessionMetadata.network_events` +/// at session end) and an optional streaming sink that writes each event to a +/// durable log as it occurs. Without the streaming sink, events are only +/// persisted at session exit, which loses them on crash and silently drops +/// past `MAX_AUDIT_EVENTS`. +pub struct AuditLog { + events: Mutex>, + streaming_sink: OnceLock>, + closed: AtomicBool, +} + +impl AuditLog { + fn new() -> Self { + Self { + events: Mutex::new(Vec::new()), + streaming_sink: OnceLock::new(), + closed: AtomicBool::new(false), + } + } + + /// Attach a streaming sink. Returns `Err` if a sink was already attached. + pub fn set_streaming_sink( + &self, + sink: Arc, + ) -> std::result::Result<(), Arc> { + self.streaming_sink.set(sink) + } + + /// True if a streaming sink has been attached. + #[must_use] + pub fn streaming_active(&self) -> bool { + self.streaming_sink.get().is_some() + } + + /// Stop accepting new events, then drain the streaming sink. + /// + /// Called by the supervisor immediately before recording `session_ended`, + /// so that late events (in-flight responses still being audited after the + /// child has exited) cannot append past the Merkle root that gets written + /// into the session metadata. Without this, post-finalize events would + /// extend the file past the stored root and cause `verify_audit_log` to + /// fail with a Merkle mismatch. + /// + /// The `closed` flag is set with release ordering *before* `flush()` so + /// that any producer that has not yet read `is_closed()` sees the seal + /// on the next access. Producers that have already passed the + /// `is_closed()` check (in `push_event`) but not yet called + /// `sink.record()` are unavoidable: their event can still reach the + /// sink after `close()` returns. The trait contract on + /// `NetworkAuditSink::flush` makes this benign — sinks may silently + /// drop post-flush events, which is exactly how + /// `RecorderStreamingSink` preserves the Merkle root invariant. + pub fn close(&self) { + self.closed.store(true, Ordering::Release); + if let Some(sink) = self.streaming_sink.get() { + sink.flush(); + } + } + + fn is_closed(&self) -> bool { + self.closed.load(Ordering::Acquire) + } +} + /// Shared in-memory sink for network audit events. -pub type SharedAuditLog = Arc>>; +pub type SharedAuditLog = Arc; /// Proxy mode for audit logging. #[derive(Debug, Clone, Copy)] @@ -57,13 +156,13 @@ impl std::fmt::Display for ProxyMode { /// Create a shared in-memory audit log. #[must_use] pub fn new_audit_log() -> SharedAuditLog { - Arc::new(Mutex::new(Vec::new())) + Arc::new(AuditLog::new()) } /// Drain all network audit events collected so far. #[must_use] pub fn drain_audit_events(audit_log: &SharedAuditLog) -> Vec { - match audit_log.lock() { + match audit_log.events.lock() { Ok(mut events) => events.drain(..).collect(), Err(e) => { warn!( @@ -110,7 +209,19 @@ fn push_event(audit_log: Option<&SharedAuditLog>, event: NetworkAuditEvent) { return; }; - match audit_log.lock() { + if audit_log.is_closed() { + // Session is over — dropping post-finalize events keeps the file + // consistent with the Merkle root recorded in session metadata. + return; + } + + // Stream to the durable sink first so events survive a process crash + // and are not lost when the in-memory buffer hits MAX_AUDIT_EVENTS. + if let Some(sink) = audit_log.streaming_sink.get() { + sink.record(&event); + } + + match audit_log.events.lock() { Ok(mut events) => { if events.len() < MAX_AUDIT_EVENTS { events.push(event); @@ -339,4 +450,135 @@ mod tests { Some("blocked by metadata deny list") ); } + + #[derive(Default)] + struct CountingSink { + events: Mutex>, + } + + impl NetworkAuditSink for CountingSink { + fn record(&self, event: &NetworkAuditEvent) { + self.events.lock().unwrap().push(event.clone()); + } + } + + #[test] + fn streaming_sink_receives_events_immediately() { + let log = new_audit_log(); + let counter = Arc::new(CountingSink::default()); + let sink: Arc = counter.clone(); + assert!(log.set_streaming_sink(sink).is_ok()); + + log_allowed( + Some(&log), + ProxyMode::Connect, + &EventContext::default(), + "api.openai.com", + 443, + "CONNECT", + ); + log_denied( + Some(&log), + ProxyMode::Connect, + &EventContext::default(), + "evil.example", + 443, + "blocked", + ); + + let streamed = counter.events.lock().unwrap(); + assert_eq!( + streamed.len(), + 2, + "sink must receive each event as it occurs" + ); + assert_eq!(streamed[0].target, "api.openai.com"); + assert_eq!(streamed[1].target, "evil.example"); + } + + #[test] + fn streaming_does_not_skip_in_memory_buffer() { + let log = new_audit_log(); + let counter = Arc::new(CountingSink::default()); + let sink: Arc = counter.clone(); + assert!(log.set_streaming_sink(sink).is_ok()); + + log_allowed( + Some(&log), + ProxyMode::Connect, + &EventContext::default(), + "api.openai.com", + 443, + "CONNECT", + ); + + let drained = drain_audit_events(&log); + assert_eq!( + drained.len(), + 1, + "buffer must still hold the event for session metadata" + ); + assert_eq!(counter.events.lock().unwrap().len(), 1); + } + + #[test] + fn streaming_active_reflects_sink_attachment() { + let log = new_audit_log(); + assert!(!log.streaming_active()); + let sink: Arc = Arc::new(CountingSink::default()); + assert!(log.set_streaming_sink(sink).is_ok()); + assert!(log.streaming_active()); + } + + #[test] + fn close_drops_subsequent_events() { + let log = new_audit_log(); + let counter = Arc::new(CountingSink::default()); + let sink: Arc = counter.clone(); + assert!(log.set_streaming_sink(sink).is_ok()); + + log_allowed( + Some(&log), + ProxyMode::Connect, + &EventContext::default(), + "before.example", + 443, + "CONNECT", + ); + log.close(); + log_allowed( + Some(&log), + ProxyMode::Connect, + &EventContext::default(), + "after.example", + 443, + "CONNECT", + ); + + let drained = drain_audit_events(&log); + assert_eq!( + drained.len(), + 1, + "post-close events must not enter the buffer" + ); + assert_eq!(drained[0].target, "before.example"); + let streamed = counter.events.lock().unwrap(); + assert_eq!( + streamed.len(), + 1, + "post-close events must not reach the sink" + ); + } + + #[test] + fn second_set_streaming_sink_returns_err() { + let log = new_audit_log(); + let first: Arc = Arc::new(CountingSink::default()); + assert!(log.set_streaming_sink(first).is_ok()); + let second: Arc = Arc::new(CountingSink::default()); + assert!( + log.set_streaming_sink(second).is_err(), + "OnceLock must reject second sink attachment" + ); + } } diff --git a/crates/nono-proxy/src/server.rs b/crates/nono-proxy/src/server.rs index 59da845c8..2353613f9 100644 --- a/crates/nono-proxy/src/server.rs +++ b/crates/nono-proxy/src/server.rs @@ -141,6 +141,32 @@ impl ProxyHandle { rows } + /// Attach a streaming sink for live forwarding of network audit events. + /// + /// Once attached, every recorded event is also forwarded to `sink` as it + /// occurs (in addition to the in-memory buffer drained at session end). + /// Returns `Err` if a sink was already attached. + pub fn set_audit_streaming_sink( + &self, + sink: std::sync::Arc, + ) -> std::result::Result<(), std::sync::Arc> { + self.audit_log.set_streaming_sink(sink) + } + + /// True when a streaming sink has been attached. Used by the supervisor to + /// avoid double-recording drained events that were already streamed. + #[must_use] + pub fn audit_streaming_active(&self) -> bool { + self.audit_log.streaming_active() + } + + /// Stop accepting new audit events. Must be called before the session's + /// final integrity summary is computed; further events would extend the + /// file past the recorded Merkle root and break verification. + pub fn close_audit_log(&self) { + self.audit_log.close(); + } + /// Environment variables to inject into the child process. /// /// The proxy URL includes `nono:@` userinfo so that standard HTTP