OpenFUT/openfut-blaze-host/src/capture.rs

//! 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<Direction> {
        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<Mutex<File>>,
    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<Capture> {
        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<u8>,
}

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> {
        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<Vec<Record>, 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<Vec<Record>, 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<u64> = recs.iter().map(|r| r.seq).collect();
        assert_eq!(seqs, (0..6).collect::<Vec<_>>());
        // And per-connection order is recoverable from the same total order.
        let conn0: Vec<u8> = 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);
    }
}