diff --git a/.gitignore b/.gitignore index 54885b4..8870de0 100644 --- a/.gitignore +++ b/.gitignore @@ -31,3 +31,8 @@ Thumbs.db # Frozen baseline archives / inspects / manifests /docker-backups/ gate-evidence/ + +# Raw Fire2 frame captures — forensic evidence, may contain session material. +# Sanitize with `blaze-sanitize` before anything leaves this machine. +*.ofcap +captures/ diff --git a/openfut-blaze-host/Cargo.toml b/openfut-blaze-host/Cargo.toml index 7a4cfdd..ebfe4f4 100644 --- a/openfut-blaze-host/Cargo.toml +++ b/openfut-blaze-host/Cargo.toml @@ -28,3 +28,7 @@ path = "src/main.rs" [[bin]] name = "blaze-probe" path = "src/bin/blaze-probe.rs" + +[[bin]] +name = "blaze-sanitize" +path = "src/bin/blaze-sanitize.rs" diff --git a/openfut-blaze-host/src/bin/blaze-sanitize.rs b/openfut-blaze-host/src/bin/blaze-sanitize.rs new file mode 100644 index 0000000..691db8b --- /dev/null +++ b/openfut-blaze-host/src/bin/blaze-sanitize.rs @@ -0,0 +1,170 @@ +//! Read a raw capture and emit a repository-safe, replayable fixture. +//! +//! ```text +//! blaze-sanitize [-o out.jsonl] [--report report.txt] +//! ``` +//! +//! Prints an audit of exactly what was replaced. Nothing is redacted silently: +//! every substitution appears in the report with its path, kind and length, and +//! untouched frames keep their exact wire bytes. +//! +//! The output is a JSONL conversation that replays against either backend — +//! see `blaze-probe --capture`. + +use std::io::Write; + +use openfut_blaze_host::capture::{self, Direction}; +use openfut_blaze_host::sanitize::{self, SENSITIVE_TAGS}; +use openfut_protocol_blaze::fire2::Header; + +fn usage() -> ! { + eprintln!("usage: blaze-sanitize [-o out.jsonl] [--report report.txt]"); + eprintln!(); + eprintln!("Reads a raw frame capture and writes a sanitized, replayable JSONL"); + eprintln!("conversation plus an audit of every redaction made."); + std::process::exit(2); +} + +fn arg_after(args: &[String], flag: &str) -> Option { + args.windows(2).find(|w| w[0] == flag).map(|w| w[1].clone()) +} + +fn hex(b: &[u8]) -> String { + b.iter().map(|x| format!("{x:02x}")).collect() +} + +fn json_escape(s: &str) -> String { + s.chars() + .flat_map(|c| match c { + '"' => "\\\"".chars().collect::>(), + '\\' => "\\\\".chars().collect(), + '\n' => "\\n".chars().collect(), + c => vec![c], + }) + .collect() +} + +fn main() { + let args: Vec = std::env::args().skip(1).collect(); + if args.is_empty() || args[0].starts_with('-') { + usage(); + } + let input = args[0].clone(); + let out_path = arg_after(&args, "-o").unwrap_or_else(|| format!("{input}.sanitized.jsonl")); + let report_path = arg_after(&args, "--report"); + + let records = match capture::read_file(&input) { + Ok(r) => r, + Err(e) => { + eprintln!("blaze-sanitize: cannot read {input}: {e}"); + std::process::exit(1); + } + }; + eprintln!("read {} frame(s) from {input}", records.len()); + + let (out, report) = match sanitize::sanitize(&records) { + Ok(v) => v, + Err(e) => { + eprintln!("blaze-sanitize: {e}"); + std::process::exit(1); + } + }; + + // ------------------------------------------------------------- fixture + let mut f = match std::fs::File::create(&out_path) { + Ok(f) => f, + Err(e) => { + eprintln!("blaze-sanitize: cannot write {out_path}: {e}"); + std::process::exit(1); + } + }; + + let _ = writeln!( + f, + "{{\"kind\":\"capture\",\"source\":\"{}\",\"frames\":{},\"redacted_frames\":{},\ + \"sensitive_tags\":[{}],\"note\":\"sanitized; redacted values are \ + length-preserving fillers, all other bytes are exact\"}}", + json_escape(&input), + report.frames_out, + report.frames_redacted, + SENSITIVE_TAGS + .iter() + .map(|t| format!("\"{t}\"")) + .collect::>() + .join(",") + ); + + for (rec, bytes) in &out { + let h = Header::parse(bytes).ok(); + let (comp, cmd, mtype, mnum, uidx, plen) = match h { + Some(h) => ( + h.component, + h.command, + h.msg_type.as_bits(), + h.msg_num, + h.user_index, + h.payload_len, + ), + None => (0, 0, 0, 0, 0, 0), + }; + let _ = writeln!( + f, + "{{\"kind\":\"frame\",\"conn\":{},\"seq\":{},\"dir\":\"{}\",\ + \"component\":{},\"command\":{},\"msg_type\":{},\"msg_num\":{},\ + \"user_index\":{},\"payload_len\":{},\"frame_hex\":\"{}\"}}", + rec.conn_id, + rec.seq, + rec.dir.label(), + comp, + cmd, + mtype, + mnum, + uidx, + plen, + hex(bytes) + ); + } + eprintln!("wrote {out_path}"); + + // -------------------------------------------------------------- report + let mut lines = Vec::new(); + lines.push(format!("sanitization report for {input}")); + lines.push(format!(" frames in: {}", report.frames_in)); + lines.push(format!(" frames out: {}", report.frames_out)); + lines.push(format!(" frames redacted: {}", report.frames_redacted)); + lines.push(format!(" redactions: {}", report.redactions.len())); + lines.push(format!(" sensitive tags: {}", SENSITIVE_TAGS.join(", "))); + if !report.undecodable.is_empty() { + lines.push(format!( + " NOT decodable as TDF (passed through unchanged, NOT inspected): seq {:?}", + report.undecodable + )); + } + lines.push(String::new()); + if report.redactions.is_empty() { + lines.push(" no sensitive values found".into()); + } else { + lines.push(" seq conn path kind bytes".into()); + for r in &report.redactions { + lines.push(format!( + " {:<5} {:<5} {:<20} {:<7} {} -> {}", + r.seq, r.conn_id, r.path, r.kind, r.original_len, r.replacement_len + )); + } + } + + // Direction/route summary, so a reviewer can see the conversation shape. + lines.push(String::new()); + lines.push(" conversation shape:".into()); + let rx = out.iter().filter(|(r, _)| r.dir == Direction::Rx).count(); + let tx = out.len() - rx; + lines.push(format!(" {rx} RX, {tx} TX")); + + let text = lines.join("\n") + "\n"; + print!("{text}"); + if let Some(p) = report_path { + if std::fs::write(&p, &text).is_ok() { + eprintln!("wrote {p}"); + } + } +} diff --git a/openfut-blaze-host/src/capture.rs b/openfut-blaze-host/src/capture.rs new file mode 100644 index 0000000..5aee609 --- /dev/null +++ b/openfut-blaze-host/src/capture.rs @@ -0,0 +1,430 @@ +//! Opt-in raw Fire2 frame capture. +//! +//! **Evidence infrastructure, not protocol functionality.** Disabled unless +//! `OPENFUT_BLAZE_CAPTURE` names a file. Nothing in the dispatch path consults +//! it, and it never changes a byte, an ordering, or a session. +//! +//! # Why this exists +//! +//! The wire captures every RE finding cites are gitignored and gone from disk, +//! and a complete live FIFA Blaze session ran past during gates 5–6 without its +//! bytes being recorded. Those sessions cannot be reproduced: a later run is a +//! different session, and the migration-validation runs in particular happen +//! once. +//! +//! # Two layers +//! +//! ```text +//! live FIFA traffic +//! ├── raw capture exact RX/TX bytes, gitignored, mode 0600 +//! └── blaze-sanitize → repository-safe, auditable fixtures +//! ``` +//! +//! Raw captures are forensic evidence and may carry session material. They are +//! never committed and never printed to the ordinary log. +//! +//! # Format +//! +//! Deterministic and self-describing, big-endian throughout to match the +//! protocol. +//! +//! ```text +//! file header (20 bytes) +//! magic 8 "OFUTCAP1" +//! version u16 +//! flags u16 +//! created u64 unix seconds +//! +//! record (repeated) +//! len u32 bytes after this field +//! conn_id u64 +//! seq u64 global, monotonic — total order across connections +//! ts_ms u64 +//! dir u8 0 = RX (from client), 1 = TX (to client) +//! pad u8×3 +//! frame_len u32 +//! frame [frame_len] EXACT bytes, one whole Fire2 frame +//! ``` +//! +//! Component, command, message number, message type and payload length are +//! deliberately **not** stored alongside the frame: they are already in its +//! 16-byte header, and a redundant copy can disagree with the bytes, leaving a +//! reader unable to tell which is true. The frame is authoritative and readers +//! derive those fields from it — [`Record::header`] does exactly that, so every +//! field named in the capture requirements is available without duplicating it. + +use std::fs::File; +use std::io::{self, Read, Write}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +use openfut_protocol_blaze::fire2::Header; + +pub const MAGIC: &[u8; 8] = b"OFUTCAP1"; +pub const VERSION: u16 = 1; +pub const FILE_HEADER_LEN: usize = 20; +/// Bytes of a record after its own length field, excluding the frame. +const RECORD_FIXED: usize = 8 + 8 + 8 + 1 + 3 + 4; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Direction { + /// Received from the client. + Rx, + /// Sent to the client. + Tx, +} + +impl Direction { + pub fn as_byte(self) -> u8 { + match self { + Direction::Rx => 0, + Direction::Tx => 1, + } + } + pub fn from_byte(b: u8) -> Option { + match b { + 0 => Some(Direction::Rx), + 1 => Some(Direction::Tx), + _ => None, + } + } + pub fn label(self) -> &'static str { + match self { + Direction::Rx => "RX", + Direction::Tx => "TX", + } + } +} + +/// A capture sink. Cheap and inert when disabled. +pub struct Capture { + sink: Option>, + seq: AtomicU64, +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +impl Capture { + /// Disabled sink. + pub fn disabled() -> Capture { + Capture { + sink: None, + seq: AtomicU64::new(0), + } + } + + /// Create a capture file, replacing any existing one. + /// + /// Created mode 0600: these bytes are forensic evidence and may contain + /// session material. + pub fn create(path: &str) -> io::Result { + let mut opts = File::options(); + opts.write(true).create(true).truncate(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + opts.mode(0o600); + } + let mut file = opts.open(path)?; + + let mut hdr = Vec::with_capacity(FILE_HEADER_LEN); + hdr.extend_from_slice(MAGIC); + hdr.extend_from_slice(&VERSION.to_be_bytes()); + hdr.extend_from_slice(&0u16.to_be_bytes()); // flags + hdr.extend_from_slice( + &SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) + .to_be_bytes(), + ); + file.write_all(&hdr)?; + file.flush()?; + + Ok(Capture { + sink: Some(Mutex::new(file)), + seq: AtomicU64::new(0), + }) + } + + pub fn enabled(&self) -> bool { + self.sink.is_some() + } + + /// Record one whole frame. + /// + /// Callers pass the exact bytes that crossed the socket, and call this + /// AFTER the I/O has happened — so a TX record means "these bytes were + /// written", not "these bytes were intended". Errors are swallowed + /// deliberately: a capture problem must never take down a live session that + /// is otherwise healthy. + pub fn record(&self, conn_id: u64, dir: Direction, frame: &[u8]) { + let Some(sink) = &self.sink else { return }; + let seq = self.seq.fetch_add(1, Ordering::SeqCst); + + let mut buf = Vec::with_capacity(4 + RECORD_FIXED + frame.len()); + buf.extend_from_slice(&((RECORD_FIXED + frame.len()) as u32).to_be_bytes()); + buf.extend_from_slice(&conn_id.to_be_bytes()); + buf.extend_from_slice(&seq.to_be_bytes()); + buf.extend_from_slice(&now_ms().to_be_bytes()); + buf.push(dir.as_byte()); + buf.extend_from_slice(&[0u8; 3]); + buf.extend_from_slice(&(frame.len() as u32).to_be_bytes()); + buf.extend_from_slice(frame); + + if let Ok(mut f) = sink.lock() { + let _ = f.write_all(&buf); + // Flushed per record: a crash mid-session must not cost the + // evidence collected so far. Volume is a few dozen frames per FIFA + // session, so this is not a hot path. + let _ = f.flush(); + } + } +} + +/// One captured frame. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Record { + pub conn_id: u64, + pub seq: u64, + pub ts_ms: u64, + pub dir: Direction, + /// Exact bytes as they crossed the socket. + pub frame: Vec, +} + +impl Record { + /// Parse the Fire2 header from the captured bytes. + /// + /// Component, command, msgNum, msgType, userIndex and payload length all + /// come from here rather than from stored copies. + pub fn header(&self) -> Option
{ + Header::parse(&self.frame).ok() + } +} + +#[derive(Debug)] +pub enum ReadError { + NotACapture, + UnsupportedVersion(u16), + Truncated { at: usize, want: usize, have: usize }, + Io(String), +} + +impl std::fmt::Display for ReadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ReadError::NotACapture => write!(f, "not an OpenFUT capture (bad magic)"), + ReadError::UnsupportedVersion(v) => write!(f, "unsupported capture version {v}"), + ReadError::Truncated { at, want, have } => { + write!( + f, + "truncated capture at offset {at}: want {want} bytes, have {have}" + ) + } + ReadError::Io(e) => write!(f, "io error: {e}"), + } + } +} + +impl std::error::Error for ReadError {} + +/// Read a whole capture. Fails clearly rather than returning partial data. +pub fn read_file(path: &str) -> Result, ReadError> { + let mut buf = Vec::new(); + File::open(path) + .and_then(|mut f| f.read_to_end(&mut buf)) + .map_err(|e| ReadError::Io(e.to_string()))?; + parse(&buf) +} + +pub fn parse(buf: &[u8]) -> Result, ReadError> { + if buf.len() < FILE_HEADER_LEN || &buf[..8] != MAGIC { + return Err(ReadError::NotACapture); + } + let version = u16::from_be_bytes([buf[8], buf[9]]); + if version != VERSION { + return Err(ReadError::UnsupportedVersion(version)); + } + + let mut out = Vec::new(); + let mut i = FILE_HEADER_LEN; + while i < buf.len() { + if i + 4 > buf.len() { + return Err(ReadError::Truncated { + at: i, + want: 4, + have: buf.len() - i, + }); + } + let len = u32::from_be_bytes([buf[i], buf[i + 1], buf[i + 2], buf[i + 3]]) as usize; + i += 4; + if len < RECORD_FIXED || i + len > buf.len() { + return Err(ReadError::Truncated { + at: i, + want: len, + have: buf.len().saturating_sub(i), + }); + } + let rec = &buf[i..i + len]; + let conn_id = u64::from_be_bytes(rec[0..8].try_into().unwrap()); + let seq = u64::from_be_bytes(rec[8..16].try_into().unwrap()); + let ts_ms = u64::from_be_bytes(rec[16..24].try_into().unwrap()); + let dir = Direction::from_byte(rec[24]).ok_or(ReadError::Truncated { + at: i + 24, + want: 1, + have: 1, + })?; + let frame_len = u32::from_be_bytes(rec[28..32].try_into().unwrap()) as usize; + if 32 + frame_len != len { + return Err(ReadError::Truncated { + at: i + 32, + want: frame_len, + have: len.saturating_sub(32), + }); + } + out.push(Record { + conn_id, + seq, + ts_ms, + dir, + frame: rec[32..32 + frame_len].to_vec(), + }); + i += len; + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use openfut_protocol_blaze::fire2::{Frame, MsgType}; + + fn tmp(name: &str) -> String { + let dir = std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()); + format!("{dir}/ofcap-test-{}-{name}.ofcap", std::process::id()) + } + + #[test] + fn disabled_capture_writes_nothing() { + let c = Capture::disabled(); + assert!(!c.enabled()); + c.record(1, Direction::Rx, &[1, 2, 3]); // must not panic or create anything + } + + #[test] + fn round_trips_frames_exactly_and_in_order() { + let path = tmp("roundtrip"); + let c = Capture::create(&path).unwrap(); + + let a = Frame::new(0x0009, 0x0007, 1, MsgType::Message, vec![1, 2, 3]).encode(); + let b = Frame::new(0x0009, 0x0007, 1, MsgType::Reply, vec![4, 5]).encode(); + c.record(7, Direction::Rx, &a); + c.record(7, Direction::Tx, &b); + drop(c); + + let recs = read_file(&path).unwrap(); + assert_eq!(recs.len(), 2); + assert_eq!(recs[0].frame, a, "RX bytes must be exact"); + assert_eq!(recs[1].frame, b, "TX bytes must be exact"); + assert_eq!(recs[0].dir, Direction::Rx); + assert_eq!(recs[1].dir, Direction::Tx); + assert_eq!(recs[0].conn_id, 7); + assert!(recs[0].seq < recs[1].seq, "sequence preserves order"); + + // Metadata is derivable rather than stored. + let h = recs[0].header().unwrap(); + assert_eq!(h.component, 0x0009); + assert_eq!(h.command, 0x0007); + assert_eq!(h.msg_type, MsgType::Message); + assert_eq!(h.payload_len, 3); + + let _ = std::fs::remove_file(&path); + } + + #[test] + fn interleaved_connections_keep_a_total_order() { + let path = tmp("interleaved"); + let c = Capture::create(&path).unwrap(); + for i in 0..6u64 { + let f = Frame::new(9, 2, i as u32, MsgType::Message, vec![i as u8]).encode(); + c.record(i % 2, Direction::Rx, &f); + } + drop(c); + + let recs = read_file(&path).unwrap(); + assert_eq!(recs.len(), 6); + let seqs: Vec = recs.iter().map(|r| r.seq).collect(); + assert_eq!(seqs, (0..6).collect::>()); + // And per-connection order is recoverable from the same total order. + let conn0: Vec = recs + .iter() + .filter(|r| r.conn_id == 0) + .map(|r| r.frame[16]) + .collect(); + assert_eq!(conn0, vec![0, 2, 4]); + + let _ = std::fs::remove_file(&path); + } + + #[cfg(unix)] + #[test] + fn capture_files_are_owner_only() { + use std::os::unix::fs::PermissionsExt; + let path = tmp("perms"); + let _c = Capture::create(&path).unwrap(); + let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777; + assert_eq!(mode, 0o600, "captures may contain session material"); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn rejects_a_file_that_is_not_a_capture() { + assert!(matches!( + parse(b"hello world padding.."), + Err(ReadError::NotACapture) + )); + assert!(matches!(parse(b""), Err(ReadError::NotACapture))); + } + + #[test] + fn rejects_an_unsupported_version() { + let mut buf = MAGIC.to_vec(); + buf.extend_from_slice(&99u16.to_be_bytes()); + buf.extend_from_slice(&0u16.to_be_bytes()); + buf.extend_from_slice(&0u64.to_be_bytes()); + assert!(matches!( + parse(&buf), + Err(ReadError::UnsupportedVersion(99)) + )); + } + + #[test] + fn truncation_fails_clearly_rather_than_returning_partial_data() { + let path = tmp("truncated"); + let c = Capture::create(&path).unwrap(); + c.record( + 1, + Direction::Rx, + &Frame::new(9, 2, 1, MsgType::Message, vec![7; 40]).encode(), + ); + drop(c); + + let full = std::fs::read(&path).unwrap(); + for cut in [FILE_HEADER_LEN + 2, FILE_HEADER_LEN + 10, full.len() - 5] { + match parse(&full[..cut]) { + Err(ReadError::Truncated { .. }) => {} + other => panic!("cut at {cut} should be Truncated, got {other:?}"), + } + } + // The whole file still parses. + assert_eq!(parse(&full).unwrap().len(), 1); + let _ = std::fs::remove_file(&path); + } +} diff --git a/openfut-blaze-host/src/config.rs b/openfut-blaze-host/src/config.rs index 4ea67cf..fa8e052 100644 --- a/openfut-blaze-host/src/config.rs +++ b/openfut-blaze-host/src/config.rs @@ -38,6 +38,8 @@ pub struct HostConfig { pub max_payload_bytes: u32, /// Optional path for the normalized structural trace. pub trace_path: Option, + /// Optional path for the raw frame capture. Opt-in; forensic evidence. + pub capture_path: Option, /// What the adapter answers with. pub adapter: AdapterConfig, } @@ -103,6 +105,9 @@ impl HostConfig { trace_path: env::var("OPENFUT_BLAZE_TRACE") .ok() .filter(|v| !v.trim().is_empty()), + capture_path: env::var("OPENFUT_BLAZE_CAPTURE") + .ok() + .filter(|v| !v.trim().is_empty()), adapter: AdapterConfig { identity: Identity::default(), endpoints, diff --git a/openfut-blaze-host/src/conn.rs b/openfut-blaze-host/src/conn.rs index 91b9d38..fac321c 100644 --- a/openfut-blaze-host/src/conn.rs +++ b/openfut-blaze-host/src/conn.rs @@ -17,6 +17,7 @@ use openfut_adapter_fifa17::blaze::{Adapter, Session}; use openfut_protocol_blaze::fire2::{Frame, Header, HEADER_LEN}; use openfut_protocol_blaze::heat2::{self, Struct}; +use crate::capture::{Capture, Direction}; use crate::config::HostConfig; use crate::trace::{self, Tracer}; use crate::Hooks; @@ -81,6 +82,7 @@ pub fn handle( adapter: &Adapter, tracer: &Tracer, hooks: &Hooks, + capture: &Capture, ) -> CloseReason { let peer = stream .peer_addr() @@ -157,6 +159,11 @@ pub fn handle( buf.drain(..used); frame_no += 1; + // Capture the exact bytes as received, before anything interprets them. + // `&buf[..total]` was already consumed above, so re-encode from the + // parsed frame — which is byte-identical by construction and covered by + // the protocol crate's round-trip tests. + capture.record(conn_id, Direction::Rx, &frame.encode()); trace::log(&format!( "conn-{conn_id:04} RX #{frame_no} {}", trace::describe(&frame.header) @@ -189,6 +196,16 @@ pub fn handle( trace::log(&format!("conn-{conn_id:04} TX #{frame_no}.{k} FAILED: {e}")); break; } + // AFTER the write: a TX record means these bytes were sent, not + // merely intended. Capturing here also keeps the client's latency + // untouched — it already has the bytes. + // + // NOT COVERED BY A TEST: moving this above the write is + // indistinguishable while writes succeed, and only diverges when + // one fails (it would record a frame that never reached the + // client). Mutation-tested and confirmed undetected. The invariant + // is held by this placement and this comment; do not move it. + capture.record(conn_id, Direction::Tx, &bytes); trace::log(&format!( "conn-{conn_id:04} TX #{frame_no}.{k} {} ({}B total)", trace::describe(&resp.header), diff --git a/openfut-blaze-host/src/lib.rs b/openfut-blaze-host/src/lib.rs index 04a4b5c..f6e06dd 100644 --- a/openfut-blaze-host/src/lib.rs +++ b/openfut-blaze-host/src/lib.rs @@ -35,8 +35,10 @@ //! silently collide with the working container. See the crate README for the //! A/B procedure and the gate list. +pub mod capture; pub mod config; pub mod conn; +pub mod sanitize; pub mod trace; pub use config::HostConfig; @@ -118,6 +120,7 @@ pub struct Server { adapter: Arc, tracer: Arc, hooks: Arc, + capture: Arc, } /// Bind without accepting, so a caller can learn the real port first (useful @@ -126,6 +129,10 @@ pub fn bind(cfg: HostConfig, hooks: Hooks) -> io::Result { let listener = TcpListener::bind(cfg.listen_on())?; let local_addr = listener.local_addr()?; let tracer = Arc::new(trace::Tracer::new(cfg.trace_path.as_deref())?); + let capture = Arc::new(match cfg.capture_path.as_deref() { + Some(p) => capture::Capture::create(p)?, + None => capture::Capture::disabled(), + }); let adapter = Arc::new(Adapter::new(cfg.adapter.clone())); Ok(Server { local_addr, @@ -134,6 +141,7 @@ pub fn bind(cfg: HostConfig, hooks: Hooks) -> io::Result { adapter, tracer, hooks: Arc::new(hooks), + capture, }) } @@ -164,6 +172,13 @@ impl Server { self.cfg.trace_path.as_deref().unwrap_or("") )); } + if self.capture.enabled() { + trace::log(&format!( + "RAW FRAME CAPTURE -> {} (forensic evidence: never commit, \ + sanitize with blaze-sanitize before sharing)", + self.cfg.capture_path.as_deref().unwrap_or("") + )); + } let counter = AtomicU64::new(0); for incoming in self.listener.incoming() { @@ -174,8 +189,9 @@ impl Server { let adapter = self.adapter.clone(); let tracer = self.tracer.clone(); let hooks = self.hooks.clone(); + let capture = self.capture.clone(); std::thread::spawn(move || { - conn::handle(stream, id, &cfg, &adapter, &tracer, &hooks); + conn::handle(stream, id, &cfg, &adapter, &tracer, &hooks, &capture); }); } Err(e) => trace::log(&format!("accept failed: {e}")), diff --git a/openfut-blaze-host/src/sanitize.rs b/openfut-blaze-host/src/sanitize.rs new file mode 100644 index 0000000..28d2923 --- /dev/null +++ b/openfut-blaze-host/src/sanitize.rs @@ -0,0 +1,359 @@ +//! Turn a raw capture into a repository-safe fixture, auditably. +//! +//! Raw captures are forensic evidence and may carry session material. This +//! produces something committable from one, and the governing rule is that +//! sanitization is **explicit and reported**, never a silent byte substitution +//! followed by a claim of equivalence. +//! +//! # What is preserved +//! +//! Everything except identified sensitive values: Fire2 headers, component and +//! command IDs, message types and numbers, user index, frame ordering, +//! connection identity, payload structure, and every non-sensitive protocol +//! value. +//! +//! # What is replaced, and how +//! +//! Only the tags in [`SENSITIVE_TAGS`], and only their values. Replacement is +//! **length-preserving**: a redacted string occupies exactly as many bytes as +//! the original, so the TDF length varint, the payload length and the Fire2 +//! header are all unchanged and the sanitized frame is the same size as the +//! captured one. That is asserted per frame — a size change means the +//! sanitizer is wrong, and it fails rather than emitting a subtly different +//! conversation. +//! +//! # What this costs +//! +//! A sanitized frame is NOT byte-identical to the wire: that is the entire +//! point, and it is why the raw capture is kept separately. Every difference is +//! enumerated in the report. + +use openfut_protocol_blaze::fire2::Frame; +use openfut_protocol_blaze::heat2::{self, Struct, Value}; + +use crate::capture::Record; + +/// Tags whose values are session-specific or credential-shaped. +/// +/// A deliberately small, named list. Anything not here is preserved, so a +/// reviewer can see exactly what was touched — and a new sensitive field shows +/// up as un-redacted rather than being silently caught by a broad pattern. +pub const SENSITIVE_TAGS: &[&str] = &[ + "KEY", // Blaze session key (LoginResponse.SESS.KEY, UserAuthenticated.KEY) + "AUTH", // auth code / token (LoginRequest.AUTH, GetAuthTokenResponse.AUTH) + "SESS", // telemetry session echo, when carried as a string + "MAIL", // account email + "PML", // parental email +]; + +/// One replacement that was made. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Redaction { + pub seq: u64, + pub conn_id: u64, + /// Dotted path to the member, e.g. `SESS.KEY`. + pub path: String, + pub kind: &'static str, + pub original_len: usize, + pub replacement_len: usize, +} + +#[derive(Debug, Default)] +pub struct Report { + pub frames_in: usize, + pub frames_out: usize, + pub frames_redacted: usize, + pub redactions: Vec, + /// Frames whose payload did not decode as TDF and were passed through. + pub undecodable: Vec, +} + +/// A length-preserving, obviously-synthetic replacement. +fn filler(len: usize) -> String { + const PATTERN: &[u8] = b"REDACTED_"; + (0..len) + .map(|i| PATTERN[i % PATTERN.len()] as char) + .collect() +} + +fn redact_struct(s: &Struct, prefix: &str, rec: &Record, out: &mut Vec) -> Struct { + let mut fields = Vec::with_capacity(s.len()); + for (tag, value) in s.iter() { + let label = tag.to_label(); + let path = if prefix.is_empty() { + label.clone() + } else { + format!("{prefix}.{label}") + }; + + let new_value = match value { + Value::Struct(inner) => Value::Struct(redact_struct(inner, &path, rec, out)), + Value::List { elem, items } => Value::List { + elem: *elem, + items: items + .iter() + .map(|it| match it { + Value::Struct(inner) => { + Value::Struct(redact_struct(inner, &path, rec, out)) + } + other => other.clone(), + }) + .collect(), + }, + Value::String(v) if SENSITIVE_TAGS.contains(&label.as_str()) && !v.is_empty() => { + // Byte length, not char count: the wire length is what must be + // preserved, and the filler is ASCII so the two coincide. + let n = v.len(); + out.push(Redaction { + seq: rec.seq, + conn_id: rec.conn_id, + path: path.clone(), + kind: "string", + original_len: n, + replacement_len: n, + }); + Value::String(filler(n)) + } + other => other.clone(), + }; + fields.push((*tag, new_value)); + } + Struct { fields } +} + +/// Sanitize one captured frame. Returns the new frame bytes. +pub fn sanitize_record(rec: &Record, report: &mut Report) -> Result, String> { + let (frame, _) = Frame::parse(&rec.frame).map_err(|e| format!("seq {}: {e}", rec.seq))?; + + if frame.payload.is_empty() { + return Ok(rec.frame.clone()); + } + let body = match heat2::decode(&frame.payload) { + Ok(b) => b, + Err(_) => { + // Not decodable as TDF. Passed through unchanged and reported, so a + // reader knows it was never inspected rather than assuming it was + // checked and found clean. + report.undecodable.push(rec.seq); + return Ok(rec.frame.clone()); + } + }; + + let before = report.redactions.len(); + let redacted = redact_struct(&body, "", rec, &mut report.redactions); + if report.redactions.len() == before { + // Nothing sensitive: emit the original bytes, not a re-encoding, so + // untouched frames stay byte-identical to the wire. + return Ok(rec.frame.clone()); + } + report.frames_redacted += 1; + + let mut out = Frame::new( + frame.header.component, + frame.header.command, + frame.header.msg_num, + frame.header.msg_type, + heat2::encode(&redacted), + ); + out.header.user_index = frame.header.user_index; + out.header.options = frame.header.options; + let bytes = out.encode(); + + // Length preservation is the contract that keeps a sanitized capture a + // faithful conversation. If it does not hold, something is wrong with the + // replacement and emitting the result would be worse than failing. + if bytes.len() != rec.frame.len() { + return Err(format!( + "seq {}: sanitized frame is {} bytes, original {} — replacement was not \ + length-preserving", + rec.seq, + bytes.len(), + rec.frame.len() + )); + } + Ok(bytes) +} + +/// A captured record paired with its sanitized frame bytes. +pub type Sanitized = (Record, Vec); + +/// Sanitize a whole capture. +pub fn sanitize(records: &[Record]) -> Result<(Vec, Report), String> { + let mut report = Report { + frames_in: records.len(), + ..Default::default() + }; + let mut out = Vec::with_capacity(records.len()); + for rec in records { + let bytes = sanitize_record(rec, &mut report)?; + out.push((rec.clone(), bytes)); + } + report.frames_out = out.len(); + Ok((out, report)) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::capture::Direction; + use openfut_protocol_blaze::fire2::MsgType; + + fn rec(seq: u64, body: &Struct) -> Record { + Record { + conn_id: 1, + seq, + ts_ms: 0, + dir: Direction::Tx, + frame: Frame::new(0x0001, 0x000A, 7, MsgType::Reply, heat2::encode(body)).encode(), + } + } + + #[test] + fn redacts_a_session_key_without_changing_frame_size() { + let body = Struct::new() + .with( + "KEY", + Value::String("a-real-looking-session-key-0123456789".into()), + ) + .with("UID", Value::Int(33068179)); + let r = rec(1, &body); + + let mut report = Report::default(); + let out = sanitize_record(&r, &mut report).unwrap(); + + assert_eq!(out.len(), r.frame.len(), "framing must be preserved"); + assert_eq!(report.redactions.len(), 1); + assert_eq!(report.redactions[0].path, "KEY"); + assert_eq!( + report.redactions[0].original_len, + report.redactions[0].replacement_len + ); + + let (f, _) = Frame::parse(&out).unwrap(); + let decoded = heat2::decode(&f.payload).unwrap(); + let key = decoded.get("KEY").and_then(Value::as_str).unwrap(); + assert!(!key.contains("real-looking"), "secret must be gone"); + assert!( + key.starts_with("REDACTED"), + "and obviously synthetic: {key}" + ); + // Non-sensitive values survive untouched. + assert_eq!(decoded.get("UID").and_then(Value::as_int), Some(33068179)); + } + + #[test] + fn redacts_nested_and_reports_the_path() { + let inner = Struct::new() + .with("KEY", Value::String("secret-key-value".into())) + .with("MAIL", Value::String("someone@example.com".into())); + let body = Struct::new() + .with("SESS", Value::Struct(inner)) + .with("ANON", Value::Int(0)); + + let mut report = Report::default(); + let out = sanitize_record(&rec(2, &body), &mut report).unwrap(); + + let paths: Vec<&str> = report.redactions.iter().map(|r| r.path.as_str()).collect(); + assert!(paths.contains(&"SESS.KEY"), "{paths:?}"); + assert!(paths.contains(&"SESS.MAIL"), "{paths:?}"); + assert!(!String::from_utf8_lossy(&out).contains("someone@example.com")); + assert!(!String::from_utf8_lossy(&out).contains("secret-key-value")); + } + + #[test] + fn redacts_inside_lists() { + let body = Struct::new().with( + "NLST", + Value::List { + elem: openfut_protocol_blaze::heat2::TypeId::Struct, + items: vec![Value::Struct( + Struct::new().with("AUTH", Value::String("token-abcdef".into())), + )], + }, + ); + let mut report = Report::default(); + let out = sanitize_record(&rec(3, &body), &mut report).unwrap(); + assert_eq!(report.redactions.len(), 1, "list members must be reached"); + assert!(!String::from_utf8_lossy(&out).contains("token-abcdef")); + } + + #[test] + fn frames_with_nothing_sensitive_stay_byte_identical() { + let body = Struct::new() + .with("UID", Value::Int(1)) + .with("CO", Value::String("US".into())); + let r = rec(4, &body); + let mut report = Report::default(); + let out = sanitize_record(&r, &mut report).unwrap(); + assert_eq!(out, r.frame, "untouched frames keep the exact wire bytes"); + assert!(report.redactions.is_empty()); + assert_eq!(report.frames_redacted, 0); + } + + #[test] + fn empty_payloads_pass_through() { + let r = Record { + conn_id: 1, + seq: 5, + ts_ms: 0, + dir: Direction::Rx, + frame: Frame::new(9, 2, 1, MsgType::Message, vec![]).encode(), + }; + let mut report = Report::default(); + assert_eq!(sanitize_record(&r, &mut report).unwrap(), r.frame); + } + + #[test] + fn undecodable_payloads_are_passed_through_and_flagged() { + let r = Record { + conn_id: 1, + seq: 6, + ts_ms: 0, + dir: Direction::Rx, + frame: Frame::new(9, 2, 1, MsgType::Message, vec![0xFF; 8]).encode(), + }; + let mut report = Report::default(); + let out = sanitize_record(&r, &mut report).unwrap(); + assert_eq!(out, r.frame); + assert_eq!( + report.undecodable, + vec![6], + "must be reported, not silently kept" + ); + } + + #[test] + fn ordering_and_identity_survive_a_whole_capture() { + let recs: Vec = (0..5) + .map(|i| { + rec( + i, + &Struct::new().with("KEY", Value::String(format!("key-{i}-padding"))), + ) + }) + .collect(); + let (out, report) = sanitize(&recs).unwrap(); + assert_eq!(report.frames_in, 5); + assert_eq!(report.frames_out, 5); + assert_eq!(report.frames_redacted, 5); + for (i, (r, _)) in out.iter().enumerate() { + assert_eq!(r.seq, i as u64, "sequence order preserved"); + assert_eq!(r.conn_id, 1); + } + } + + #[test] + fn every_sensitive_tag_is_actually_reachable() { + // Guards against a tag being listed but never matched — e.g. a typo. + for tag in SENSITIVE_TAGS { + let body = Struct::new().with(tag, Value::String("sensitive-value".into())); + let mut report = Report::default(); + let out = sanitize_record(&rec(9, &body), &mut report).unwrap(); + assert_eq!(report.redactions.len(), 1, "tag {tag} was not redacted"); + assert!( + !String::from_utf8_lossy(&out).contains("sensitive-value"), + "tag {tag} leaked" + ); + } + } +} diff --git a/openfut-blaze-host/tests/live_transport.rs b/openfut-blaze-host/tests/live_transport.rs index 16ab623..08fa1b3 100644 --- a/openfut-blaze-host/tests/live_transport.rs +++ b/openfut-blaze-host/tests/live_transport.rs @@ -21,7 +21,7 @@ use std::sync::Mutex; use std::time::Duration; use openfut_adapter_fifa17::blaze::{AdapterConfig, Endpoints, Identity}; -use openfut_blaze_host::{bind, config::HostConfig, Hooks}; +use openfut_blaze_host::{bind, capture, config::HostConfig, sanitize, Hooks}; use openfut_protocol_blaze::fire2::{Frame, Header, HEADER_LEN}; use serde_json::Value as J; @@ -68,6 +68,10 @@ struct Harness { /// Nth connection this test opens gets the Nth recorded key. That is why the /// connection order below must match the fixture's session order. fn start() -> Harness { + start_with_capture(None) +} + +fn start_with_capture(capture_path: Option) -> Harness { let records = records(); let cfg_rec = records .iter() @@ -114,6 +118,7 @@ fn start() -> Harness { idle_timeout_secs: 10, max_payload_bytes: 4 * 1024 * 1024, trace_path: None, + capture_path, adapter: adapter_cfg, }; @@ -470,3 +475,276 @@ fn an_undecodable_body_still_gets_a_reply() { assert_eq!(reply.header.command, 0x0002); assert!(!reply.payload.is_empty(), "ping still answers with STIM"); } + +// ───────────────────────────── raw frame capture ───────────────────────────── +// +// Capture is evidence infrastructure. These tests exist because the value of a +// capture is entirely in its fidelity: a capture that quietly drops, reorders +// or alters frames is worse than none, since it would be trusted. + +fn cap_path(name: &str) -> String { + let dir = std::env::var("TMPDIR").unwrap_or_else(|_| "/tmp".into()); + format!("{dir}/ofcap-it-{}-{name}.ofcap", std::process::id()) +} + +/// Drive a conversation over TCP and return (captured records, wire responses). +fn converse(h: &Harness, take: usize) -> Vec> { + let sid = session_ids(&h.records).into_iter().next().unwrap(); + let mut stream = connect(&h.addr); + let mut buf = Vec::new(); + let mut got = Vec::new(); + for tx in transactions(&h.records, &sid).into_iter().take(take) { + stream.write_all(&unhex(&st(tx, "request_hex"))).unwrap(); + stream.flush().unwrap(); + for _ in 0..tx["responses"].as_array().unwrap().len() { + got.push(read_frame(&mut stream, &mut buf).expect("frame").encode()); + } + } + got +} + +#[test] +fn capture_disabled_produces_no_artefact() { + let path = cap_path("disabled"); + let _ = std::fs::remove_file(&path); + + let h = start(); // capture_path: None + converse(&h, 4); + + assert!( + !std::path::Path::new(&path).exists(), + "no capture file may appear when capture is off" + ); +} + +#[test] +fn capture_records_rx_and_tx_frames_exactly_and_in_order() { + let path = cap_path("exact"); + let _ = std::fs::remove_file(&path); + + let h = start_with_capture(Some(path.clone())); + let sid = session_ids(&h.records).into_iter().next().unwrap(); + let txs: Vec<&J> = transactions(&h.records, &sid).into_iter().take(6).collect(); + + let mut stream = connect(&h.addr); + let mut buf = Vec::new(); + let mut sent = Vec::new(); + let mut received = Vec::new(); + for tx in &txs { + let req = unhex(&st(tx, "request_hex")); + stream.write_all(&req).unwrap(); + stream.flush().unwrap(); + sent.push(req); + for _ in 0..tx["responses"].as_array().unwrap().len() { + received.push(read_frame(&mut stream, &mut buf).expect("frame").encode()); + } + } + drop(stream); + std::thread::sleep(std::time::Duration::from_millis(200)); + + let recs = capture::read_file(&path).expect("capture parses"); + + let rx: Vec> = recs + .iter() + .filter(|r| r.dir == capture::Direction::Rx) + .map(|r| r.frame.clone()) + .collect(); + let tx_caps: Vec> = recs + .iter() + .filter(|r| r.dir == capture::Direction::Tx) + .map(|r| r.frame.clone()) + .collect(); + + assert_eq!( + rx, sent, + "captured RX must equal the exact bytes sent by the client" + ); + assert_eq!( + tx_caps, received, + "captured TX must equal the exact bytes the client received" + ); + + // Total order is monotonic, and RX precedes its own TX. + let seqs: Vec = recs.iter().map(|r| r.seq).collect(); + let mut sorted = seqs.clone(); + sorted.sort(); + assert_eq!(seqs, sorted, "records are written in order"); + assert_eq!(recs[0].dir, capture::Direction::Rx); + + let _ = std::fs::remove_file(&path); +} + +#[test] +fn capture_does_not_alter_the_wire_output() { + // The same conversation with and without capture must produce identical + // response bytes. Capture must be observation, not participation. + let without = converse(&start(), 6); + + let path = cap_path("noalter"); + let _ = std::fs::remove_file(&path); + let with = converse(&start_with_capture(Some(path.clone())), 6); + + assert_eq!( + with.iter().map(|f| hex(f)).collect::>(), + without.iter().map(|f| hex(f)).collect::>(), + "enabling capture changed the wire output" + ); + let _ = std::fs::remove_file(&path); +} + +#[test] +fn fragmented_input_reconstructs_the_same_captured_frames() { + // A frame dribbled one byte at a time must be captured as ONE whole frame, + // identical to the same frame sent in a single write. + let sid_take = 3; + + let whole_path = cap_path("whole"); + let _ = std::fs::remove_file(&whole_path); + converse(&start_with_capture(Some(whole_path.clone())), sid_take); + std::thread::sleep(std::time::Duration::from_millis(150)); + let whole = capture::read_file(&whole_path).unwrap(); + + let frag_path = cap_path("frag"); + let _ = std::fs::remove_file(&frag_path); + let h = start_with_capture(Some(frag_path.clone())); + let sid = session_ids(&h.records).into_iter().next().unwrap(); + let mut stream = connect(&h.addr); + let mut buf = Vec::new(); + for tx in transactions(&h.records, &sid).into_iter().take(sid_take) { + for byte in unhex(&st(tx, "request_hex")) { + stream.write_all(&[byte]).unwrap(); + stream.flush().unwrap(); + } + for _ in 0..tx["responses"].as_array().unwrap().len() { + read_frame(&mut stream, &mut buf).expect("frame"); + } + } + drop(stream); + std::thread::sleep(std::time::Duration::from_millis(200)); + let frag = capture::read_file(&frag_path).unwrap(); + + let f_whole: Vec = whole.iter().map(|r| hex(&r.frame)).collect(); + let f_frag: Vec = frag.iter().map(|r| hex(&r.frame)).collect(); + assert_eq!( + f_frag, f_whole, + "fragmentation must not change captured frames" + ); + + let _ = std::fs::remove_file(&whole_path); + let _ = std::fs::remove_file(&frag_path); +} + +#[test] +fn coalesced_input_is_captured_as_separate_frames() { + let path = cap_path("coalesced"); + let _ = std::fs::remove_file(&path); + + let h = start_with_capture(Some(path.clone())); + let sid = session_ids(&h.records).into_iter().next().unwrap(); + let txs: Vec<&J> = transactions(&h.records, &sid).into_iter().take(4).collect(); + + let mut stream = connect(&h.addr); + let mut buf = Vec::new(); + let mut blob = Vec::new(); + for tx in &txs { + blob.extend_from_slice(&unhex(&st(tx, "request_hex"))); + } + stream.write_all(&blob).unwrap(); // ONE write + stream.flush().unwrap(); + for tx in &txs { + for _ in 0..tx["responses"].as_array().unwrap().len() { + read_frame(&mut stream, &mut buf).expect("frame"); + } + } + drop(stream); + std::thread::sleep(std::time::Duration::from_millis(200)); + + let recs = capture::read_file(&path).unwrap(); + let rx: Vec> = recs + .iter() + .filter(|r| r.dir == capture::Direction::Rx) + .map(|r| r.frame.clone()) + .collect(); + + assert_eq!(rx.len(), txs.len(), "one record per frame, not per read"); + for (got, tx) in rx.iter().zip(txs.iter()) { + assert_eq!( + hex(got), + st(tx, "request_hex"), + "frame boundaries preserved" + ); + } + let _ = std::fs::remove_file(&path); +} + +#[test] +fn sanitization_removes_the_session_key_from_a_real_captured_login() { + let path = cap_path("sanitize"); + let _ = std::fs::remove_file(&path); + + let h = start_with_capture(Some(path.clone())); + let sid = session_ids(&h.records).into_iter().next().unwrap(); + let mut stream = connect(&h.addr); + let mut buf = Vec::new(); + for tx in transactions(&h.records, &sid) { + stream.write_all(&unhex(&st(tx, "request_hex"))).unwrap(); + stream.flush().unwrap(); + for _ in 0..tx["responses"].as_array().unwrap().len() { + read_frame(&mut stream, &mut buf).expect("frame"); + } + if st(tx, "name") == "login" { + break; + } + } + drop(stream); + std::thread::sleep(std::time::Duration::from_millis(200)); + + let recs = capture::read_file(&path).unwrap(); + let session_key = h + .records + .iter() + .find(|r| r["kind"] == "session" && r["id"] == sid.as_str()) + .map(|r| st(r, "session_key")) + .expect("recorded session key"); + + // The raw capture DOES contain the key — that is what makes it forensic + // evidence, and why it is gitignored. + let raw_has_key = recs + .iter() + .any(|r| String::from_utf8_lossy(&r.frame).contains(&session_key)); + assert!( + raw_has_key, + "raw capture should contain the real session key" + ); + + let (out, report) = sanitize::sanitize(&recs).expect("sanitizes"); + assert!( + report.frames_redacted > 0, + "login frames carry a session key" + ); + + for (_, bytes) in &out { + assert!( + !String::from_utf8_lossy(bytes).contains(&session_key), + "sanitized output still contains the session key" + ); + } + // Structure is untouched: same frame count, same sizes, same order. + assert_eq!(out.len(), recs.len()); + for ((rec, bytes), orig) in out.iter().zip(recs.iter()) { + assert_eq!( + bytes.len(), + orig.frame.len(), + "frame size must be preserved" + ); + assert_eq!(rec.seq, orig.seq, "order must be preserved"); + } + // And every redaction is reported, not silent. + assert!( + report.redactions.iter().any(|r| r.path.ends_with("KEY")), + "{:?}", + report.redactions + ); + + let _ = std::fs::remove_file(&path); +}