blaze-host: opt-in raw frame capture + auditable sanitizer

Evidence infrastructure, not protocol functionality. Built before gates 7-10
because those sessions cannot be reproduced -- a later run is a different
session, and the migration-validation runs happen once. Gates 5-6 already went
past without their bytes being recorded.

TWO LAYERS

  live FIFA traffic
    ├── raw capture      exact RX/TX bytes, mode 0600, gitignored
    └── blaze-sanitize   → repository-safe, replayable fixtures

CAPTURE. Off unless OPENFUT_BLAZE_CAPTURE names a file. Deterministic
big-endian container: 20-byte file header, then per-frame records carrying
connection id, a global monotonic sequence, timestamp, direction and the EXACT
frame bytes. RX is recorded as received; TX only AFTER a successful write, so a
record means the bytes were sent rather than intended.

Component/command/msgNum/msgType/payload length are deliberately NOT stored
beside 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.
Record::header() derives them, so every field the requirements name is
available without duplicating it.

SANITIZER. Redacts only the named tags in SENSITIVE_TAGS (KEY, AUTH, SESS,
MAIL, PML) and reports every substitution with path, kind and length.
Replacement is LENGTH-PRESERVING, so the TDF varint, payload length and Fire2
header are unchanged and the sanitized frame is exactly the size of the
captured one -- asserted per frame, failing rather than emitting a subtly
different conversation. Frames with nothing sensitive keep their exact wire
bytes. Payloads that will not decode are passed through and REPORTED, so a
reader knows they were never inspected rather than assuming they were checked.

TESTS. 39 in this crate. All nine required cases: capture disabled produces no
artefact; RX and TX captured exactly; ordering preserved; fragmented input
(one byte at a time) reconstructs the same frames as a single write; coalesced
input is captured as separate frames, not per-read; capture does not alter wire
output; sanitization removes a real session key from a real captured login;
malformed/truncated/wrong-version captures fail clearly; every listed sensitive
tag is provably reachable.

MUTATION TESTED. Dropping TX capture, truncating captured frames to their
header, and removing KEY from the sensitive list were each verified to turn the
suite red. One mutation was NOT caught: moving the TX capture above the write.
It is indistinguishable while writes succeed and only diverges when one fails.
That invariant is held by code placement and a comment saying so, not by a
test, and the code says as much rather than implying coverage it does not have.

Python oracle unchanged.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

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<String>) -> 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<Vec<u8>> {
+    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<Vec<u8>> = recs
+        .iter()
+        .filter(|r| r.dir == capture::Direction::Rx)
+        .map(|r| r.frame.clone())
+        .collect();
+    let tx_caps: Vec<Vec<u8>> = 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<u64> = 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::<Vec<_>>(),
+        without.iter().map(|f| hex(f)).collect::<Vec<_>>(),
+        "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<String> = whole.iter().map(|r| hex(&r.frame)).collect();
+    let f_frag: Vec<String> = 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<Vec<u8>> = 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);
+}