diff --git a/Cargo.toml b/Cargo.toml index dea0f08..aa921df 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "openfut-core", "openfut-protocol-blaze", "openfut-adapter-fifa17", + "openfut-blaze-host", "openfut-bridge", "openfut-launcher", "openfut-launcher/openfut-hook", diff --git a/openfut-blaze-host/Cargo.toml b/openfut-blaze-host/Cargo.toml new file mode 100644 index 0000000..7a4cfdd --- /dev/null +++ b/openfut-blaze-host/Cargo.toml @@ -0,0 +1,30 @@ +[package] +name = "openfut-blaze-host" +version = "0.1.0" +edition = "2021" +license = "MIT" +description = "Thin TCP host for the FIFA 17 Blaze RPC surface; runs beside the Python backend" +publish = false + +[dependencies] +openfut-protocol-blaze = { path = "../openfut-protocol-blaze" } +openfut-adapter-fifa17 = { path = "../openfut-adapter-fifa17" } +# Session keys. The client never validates them, but they must be distinct per +# connection; seeding from the clock would not be. +rand = "0.8" +# The probe replays the adapter's recorded fixture conversation. +serde_json = "1" + +# No async runtime and no TLS, both deliberate: +# * A FIFA client opens a handful of connections, so a thread each mirrors the +# Python oracle and keeps the host readable. +# * The Blaze main port is plaintext — verified against the running backend. +# TLS belongs to the redirector phase. + +[[bin]] +name = "openfut-blaze-host" +path = "src/main.rs" + +[[bin]] +name = "blaze-probe" +path = "src/bin/blaze-probe.rs" diff --git a/openfut-blaze-host/README.md b/openfut-blaze-host/README.md new file mode 100644 index 0000000..3a2667e --- /dev/null +++ b/openfut-blaze-host/README.md @@ -0,0 +1,148 @@ +# openfut-blaze-host + +A deliberately thin TCP host for the FIFA 17 Blaze RPC surface. Runs **beside** +the working Python backend, never instead of it. + +``` + listener → Fire2 stream framing → per-connection Session + │ + openfut-adapter-fifa17::dispatch + │ + write returned frames, in order +``` + +## Scope + +Owns: a socket, a read buffer, one `Session` per connection, diagnostics. +That is the complete list. + +Must never acquire: coins, club state, packs, profiles, market state, UTAS +logic. Those belong to OpenFUT Core, reached later through the adapter. A +transport host that starts holding game state becomes a second backend — the +architecture this migration exists to avoid. + +## No TLS + +The Blaze main port is **plaintext**. Verified against the running Python +backend by sending a raw Fire2 `Util::ping` and getting a plaintext +`PingResponse` back; `blaze_responder_v3b.py::blaze_handle` uses the raw socket +and only `redir_handle` wraps `ssl`. TLS belongs to the redirector phase. + +## Running it + +Both critical settings are required — there is no default port anywhere in this +crate, so it cannot silently collide with the Python container. + +```bash +cargo build -p openfut-blaze-host + +OPENFUT_ADVERTISE= \ +OPENFUT_BIND=0.0.0.0 \ +POW_CONTENT_HOST=:8085 \ +OPENFUT_BLAZE_HOST_BIND=0.0.0.0 \ +OPENFUT_BLAZE_HOST_PORT= \ +OPENFUT_BLAZE_TRACE=/tmp/rust-blaze.trace \ + ./target/debug/openfut-blaze-host +``` + +`OPENFUT_BIND` is the **advertised-config** bind, not the listener's. The +adapter derives `nucleusConnect` from it (reproducing the oracle — see the +adapter README and the vault's known-issue entry), so it must mirror whatever +the Python container runs with, or the two will not compare. The listener has +its own `OPENFUT_BLAZE_HOST_BIND`. + +For a FIFA test from another machine, `OPENFUT_BLAZE_HOST_BIND` must be `0.0.0.0` +(or the LAN address); the default follows `OPENFUT_BIND`. + +## Live A/B against Python + +```bash +./check-live-parity.sh 127.0.0.1:42130 127.0.0.1: +``` + +Replays the recorded conversations against both endpoints over real sockets and +diffs the normalized traces. Session keys and server timestamps are masked, so +anything that differs is a real behavioural difference. + +**The diff is the test, not the probe's exit code.** The probe only detects +anomalies visible live — missing frames, an early close. A same-length content +change deep inside a notification body shows up *only* as a digest difference in +the trace. Both cases were verified by mutation. + +For the comparison to mean anything both servers must run the same config — +same `OPENFUT_ADVERTISE`, `OPENFUT_BIND`, `POW_CONTENT_HOST` and active persona +— or legitimate config differences read as parity failures. + +Current result against the live container: + +``` + main: IDENTICAL (82 frames) + fallbacks: IDENTICAL (12 frames) + locale: IDENTICAL (7 frames) + LIVE PARITY OK — 101 frames, identical normalized traces. +``` + +## Tests + +`cargo test -p openfut-blaze-host` — 18 tests. The integration suite runs the +real host on an ephemeral port and replays the recorded conversations over TCP, +covering what fixtures cannot: + +* byte-for-byte replay over a socket +* requests dribbled **one byte at a time** (fragmentation) +* several requests in **one write** (coalescing) +* the four-frame login burst arriving in order on the wire +* session state persisting across frames, and *not* leaking between connections +* an absurd payload length closing the connection instead of allocating +* an undecodable body still getting a reply + +Byte-exactness is possible only because `Hooks` injects the session key and +clock — they appear inside response bodies, so with the real ones no live run +could reproduce a recording. That is the crate's only test seam. + +## Promotion gates + +Automated, re-runnable now: + +1. ✅ All migration tests pass (116 across the three crates). +2. ✅ Python contract suite still 446/446. +3. ✅ Scripted connection to the Rust host succeeds. +4. ✅ Recorded conversations work through the real TCP host, and the live A/B + against Python is identical over 101 frames. + +Requiring a FIFA client on the game machine — **not yet done**: + +5. ⬜ FIFA reaches FUT with Rust Blaze. +6. ⬜ Close FIFA completely. +7. ⬜ FIFA reaches FUT a second time. +8. ⬜ Restore Python Blaze and confirm rollback works. +9. ⬜ Switch back to Rust and confirm again. +10. ⬜ Only then is Rust Blaze a viable runtime replacement. + +Gates 8 and 9 matter as much as 5: `Python → Rust → Python → Rust` proves the +rollback path rather than asserting one exists. + +### Running gate 5 + +Keep the Python container exactly as it is. On the game machine, redirect **only +the Blaze destination** to the sidecar's port; the redirector, Nucleus, roster, +UTAS and POW keep hitting Python. The single changed variable is then Python +Blaze vs Rust Blaze. + +The redirector is what tells the client where Blaze lives, so the cleanest +switch is to point the Python redirector's advertised Blaze port at the sidecar +rather than reconfiguring the client. + +## Diagnostics + +The log mirrors the Python responder's shape so the two can be read side by +side: connection id, peer, frame number, route, msgType, msgNum, userIndex, +options, payload and metadata sizes, every response emitted, and a close reason. + +`OPENFUT_BLAZE_TRACE=` additionally writes the normalized structural +trace — the same format the probe emits, so a live FIFA session against Rust can +be diffed against one against Python. + +No credential or token is logged. Volatile values are replaced before they reach +the line, not truncated after, and a test asserts a known secret never appears +in trace output. diff --git a/openfut-blaze-host/check-live-parity.sh b/openfut-blaze-host/check-live-parity.sh new file mode 100755 index 0000000..aed2f23 --- /dev/null +++ b/openfut-blaze-host/check-live-parity.sh @@ -0,0 +1,67 @@ +#!/usr/bin/env bash +# Live A/B: the Python Blaze responder vs the Rust sidecar, over real sockets. +# +# ./check-live-parity.sh [outdir] +# +# Replays the recorded conversations against BOTH endpoints and diffs the +# normalized traces. Session keys and server timestamps are masked, so anything +# that differs is a real behavioural difference. +# +# IMPORTANT: the diff is the test, not the probe's exit code. The probe only +# detects structural anomalies it can see live (missing frames, early close); a +# same-length content change deep inside a notification body shows up ONLY as a +# digest difference in the trace. Verified by mutation. +# +# Read-only against both backends: it opens client connections and sends +# recorded requests. Safe to run while the Python backend is serving. +# +# For the comparison to mean anything, BOTH servers must run the same config — +# same OPENFUT_ADVERTISE, OPENFUT_BIND, POW_CONTENT_HOST and active persona. +# Otherwise legitimate config differences read as parity failures. +set -uo pipefail + +PY="${1:-}" +RS="${2:-}" +OUT="${3:-$(mktemp -d)}" + +if [[ -z "$PY" || -z "$RS" ]]; then + echo "usage: $0 [outdir]" >&2 + echo "example: $0 127.0.0.1:42130 127.0.0.1:42230" >&2 + exit 2 +fi + +cd "$(dirname "$(readlink -f "$0")")/.." +PROBE="./target/debug/blaze-probe" +[[ -x "$PROBE" ]] || PROBE="./target/release/blaze-probe" +if [[ ! -x "$PROBE" ]]; then + echo "blaze-probe not built; run: cargo build -p openfut-blaze-host" >&2 + exit 2 +fi + +mkdir -p "$OUT" +fail=0 +frames=0 + +for sess in main fallbacks locale; do + "$PROBE" "$PY" --session "$sess" > "$OUT/python-$sess.trace" 2>"$OUT/python-$sess.err" || true + "$PROBE" "$RS" --session "$sess" > "$OUT/rust-$sess.trace" 2>"$OUT/rust-$sess.err" || true + + n=$(grep -c '^conn-' "$OUT/python-$sess.trace" || true) + if diff -u "$OUT/python-$sess.trace" "$OUT/rust-$sess.trace" > "$OUT/diff-$sess.txt"; then + echo " $sess: IDENTICAL ($n frames)" + frames=$((frames + n)) + else + echo " $sess: DIFFERS -> $OUT/diff-$sess.txt" + head -30 "$OUT/diff-$sess.txt" + fail=1 + fi +done + +echo +if [[ $fail -eq 0 ]]; then + echo "LIVE PARITY OK — $frames frames, identical normalized traces." + echo "traces: $OUT" +else + echo "LIVE PARITY FAILED — see $OUT" +fi +exit $fail diff --git a/openfut-blaze-host/src/bin/blaze-probe.rs b/openfut-blaze-host/src/bin/blaze-probe.rs new file mode 100644 index 0000000..9913240 --- /dev/null +++ b/openfut-blaze-host/src/bin/blaze-probe.rs @@ -0,0 +1,205 @@ +//! Replay a recorded Blaze conversation against a LIVE endpoint and emit a +//! normalized trace. +//! +//! This is the A/B instrument. Point it at the Python backend, then at the Rust +//! sidecar, and diff the two traces: +//! +//! ```text +//! blaze-probe 127.0.0.1:42130 > /tmp/python.trace +//! blaze-probe 127.0.0.1:42230 > /tmp/rust.trace +//! diff /tmp/python.trace /tmp/rust.trace +//! ``` +//! +//! Byte comparison is impossible across two live servers — session keys and +//! server timestamps differ by design — so the trace masks known-volatile +//! values while preserving their tag, type and length. Everything else must +//! match exactly, including frame counts and notification ordering. +//! +//! The requests come from `openfut-adapter-fifa17/fixtures/blaze_transactions.jsonl`, +//! so this exercises the same conversation the offline parity suite does, but +//! over a real socket against a real server process. +//! +//! Read-only: it opens a client connection and sends recorded requests. It +//! writes nothing and mutates no state beyond the server's own per-connection +//! session. + +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::time::Duration; + +use openfut_blaze_host::trace; +use openfut_protocol_blaze::fire2::{Frame, Header, HEADER_LEN}; + +fn usage() -> ! { + eprintln!("usage: blaze-probe [--verbose] [--session main|fallbacks|locale]"); + eprintln!(); + eprintln!("Replays the recorded Blaze conversation against a live endpoint and"); + eprintln!("prints a normalized, volatile-masked trace on stdout."); + std::process::exit(2); +} + +fn fixtures_path() -> String { + format!( + "{}/../openfut-adapter-fifa17/fixtures/blaze_transactions.jsonl", + env!("CARGO_MANIFEST_DIR") + ) +} + +fn main() { + let args: Vec = std::env::args().skip(1).collect(); + if args.is_empty() || args[0].starts_with("--") { + usage(); + } + let endpoint = args[0].clone(); + let verbose = args.iter().any(|a| a == "--verbose"); + let want_session = args + .windows(2) + .find(|w| w[0] == "--session") + .map(|w| w[1].clone()) + .unwrap_or_else(|| "main".to_string()); + + let path = fixtures_path(); + let text = std::fs::read_to_string(&path).unwrap_or_else(|e| { + eprintln!("cannot read {path}: {e}"); + std::process::exit(1) + }); + + let records: Vec = text + .lines() + .filter(|l| !l.trim().is_empty()) + .map(|l| serde_json::from_str(l).expect("fixture line is valid JSON")) + .collect(); + + let requests: Vec<(String, Vec)> = records + .iter() + .filter(|r| r["kind"] == "tx" && r["session"] == want_session.as_str()) + .map(|r| { + ( + r["name"].as_str().unwrap().to_string(), + unhex(r["request_hex"].as_str().unwrap()), + ) + }) + .collect(); + + if requests.is_empty() { + eprintln!("no transactions for session {want_session:?}"); + std::process::exit(1); + } + + eprintln!( + "blaze-probe: {} requests from session {want_session:?} -> {endpoint}", + requests.len() + ); + + let mut stream = TcpStream::connect(&endpoint).unwrap_or_else(|e| { + eprintln!("cannot connect to {endpoint}: {e}"); + std::process::exit(1); + }); + let _ = stream.set_nodelay(true); + // Generous but finite: a server that answers nothing must not hang the probe. + let _ = stream.set_read_timeout(Some(Duration::from_secs(10))); + + println!("# blaze-probe conversation: session={want_session}"); + println!("# endpoint intentionally omitted so traces from different hosts diff cleanly"); + + let mut buf: Vec = Vec::new(); + let mut mismatched = 0usize; + + for (n, (name, request)) in requests.iter().enumerate() { + let req_frame = Frame::parse(request).expect("fixture request parses").0; + println!("--- {:02} {name}", n + 1); + println!( + "{}", + trace::trace_line(0, "TX", &format!("{}", n + 1), &req_frame) + ); + + if let Err(e) = stream.write_all(request) { + println!("!! send failed: {e}"); + mismatched += 1; + break; + } + let _ = stream.flush(); + + // How many frames to expect is recorded; a server that sends fewer is + // the failure this probe exists to catch, so read with a timeout rather + // than blocking forever. + let expected = records + .iter() + .find(|r| r["kind"] == "tx" && r["name"] == name.as_str()) + .and_then(|r| r["responses"].as_array()) + .map(|a| a.len()) + .unwrap_or(0); + + let mut got = 0usize; + while got < expected { + match read_frame(&mut stream, &mut buf) { + Ok(Some(frame)) => { + println!( + "{}", + trace::trace_line(0, "RX", &format!("{}.{}", n + 1, got), &frame) + ); + if verbose && !frame.payload.is_empty() { + if let Ok(body) = openfut_protocol_blaze::heat2::decode(&frame.payload) { + for line in trace::masked_dump(&body).lines() { + println!(" {line}"); + } + } + } + got += 1; + } + Ok(None) => { + println!("!! connection closed after {got}/{expected} frame(s)"); + mismatched += 1; + break; + } + Err(e) => { + println!("!! read failed after {got}/{expected} frame(s): {e}"); + mismatched += 1; + break; + } + } + } + if got != expected { + println!("!! frame count {got}, recorded {expected}"); + mismatched += 1; + } + } + + println!( + "# done: {} request(s), {mismatched} anomaly/anomalies", + requests.len() + ); + if mismatched > 0 { + std::process::exit(1); + } +} + +fn read_frame(stream: &mut TcpStream, buf: &mut Vec) -> std::io::Result> { + let mut chunk = [0u8; 65536]; + while buf.len() < HEADER_LEN { + match stream.read(&mut chunk)? { + 0 => return Ok(None), + got => buf.extend_from_slice(&chunk[..got]), + } + } + let header = + Header::parse(&buf[..HEADER_LEN]).map_err(|e| std::io::Error::other(e.to_string()))?; + let total = header.frame_len(); + while buf.len() < total { + match stream.read(&mut chunk)? { + 0 => return Ok(None), + got => buf.extend_from_slice(&chunk[..got]), + } + } + let (frame, used) = + Frame::parse(&buf[..total]).map_err(|e| std::io::Error::other(e.to_string()))?; + buf.drain(..used); + Ok(Some(frame)) +} + +fn unhex(s: &str) -> Vec { + (0..s.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("valid hex")) + .collect() +} diff --git a/openfut-blaze-host/src/config.rs b/openfut-blaze-host/src/config.rs new file mode 100644 index 0000000..4ea67cf --- /dev/null +++ b/openfut-blaze-host/src/config.rs @@ -0,0 +1,175 @@ +//! Host configuration, entirely from the environment. +//! +//! Two rules carried over from the Python deployment: +//! +//! * **No silent loopback.** `OPENFUT_ADVERTISE` is required, exactly as the +//! Python entrypoint requires it. A backend that guesses its own reachable +//! address is the bug the client/server split removed. +//! * **No default port.** The sidecar runs beside the working Python container +//! and must never collide with it, so the listen port is explicit. There is +//! no "test port" constant anywhere in this crate. + +use std::env; +use std::fmt; + +use openfut_adapter_fifa17::blaze::{AdapterConfig, Endpoints, Identity}; + +#[derive(Debug)] +pub struct ConfigError(String); + +impl fmt::Display for ConfigError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.0) + } +} + +impl std::error::Error for ConfigError {} + +#[derive(Debug, Clone)] +pub struct HostConfig { + /// Address the listener binds. + pub listen_addr: String, + /// Port the listener binds. Required; no default. + pub listen_port: u16, + /// Seconds of inactivity before a connection is dropped. Matches the + /// oracle's 300s socket timeout. + pub idle_timeout_secs: u64, + /// Reject a frame claiming a larger payload than this, as the oracle does. + pub max_payload_bytes: u32, + /// Optional path for the normalized structural trace. + pub trace_path: Option, + /// What the adapter answers with. + pub adapter: AdapterConfig, +} + +fn required(key: &str, why: &str) -> Result { + match env::var(key) { + Ok(v) if !v.trim().is_empty() => Ok(v), + _ => Err(ConfigError(format!("{key} must be set — {why}"))), + } +} + +fn optional(key: &str, default: &str) -> String { + env::var(key) + .ok() + .filter(|v| !v.trim().is_empty()) + .unwrap_or_else(|| default.to_string()) +} + +impl HostConfig { + pub fn from_env() -> Result { + // The address handed to the CLIENT for every next hop. + let advertise = required( + "OPENFUT_ADVERTISE", + "it is the address the game machine uses to reach this host; \ + there is no loopback fallback in remote mode", + )?; + + // NOTE: this is the *advertised-config* bind, not the listener bind. + // The adapter derives nucleusConnect from it, reproducing the oracle + // (see the adapter's config docs and the vault's known-issue entry), so + // it must mirror whatever the Python container runs with if the two are + // to be compared. The listener has its own setting below. + let config_bind = optional("OPENFUT_BIND", "127.0.0.1"); + + let listen_port_raw = required( + "OPENFUT_BLAZE_HOST_PORT", + "the sidecar runs beside the working Python backend and must not \ + collide with it, so the port is explicit and has no default", + )?; + let listen_port: u16 = listen_port_raw.trim().parse().map_err(|_| { + ConfigError(format!( + "OPENFUT_BLAZE_HOST_PORT is not a valid port: {listen_port_raw:?}" + )) + })?; + + let listen_addr = optional("OPENFUT_BLAZE_HOST_BIND", &config_bind); + + let endpoints = Endpoints { + advertise, + bind: config_bind, + pow_content_host: optional("POW_CONTENT_HOST", "127.0.0.1:8080"), + pow_host: optional("POW_HOST", "127.0.0.1:8094"), + ..Endpoints::default() + }; + + Ok(HostConfig { + listen_addr, + listen_port, + idle_timeout_secs: optional("OPENFUT_BLAZE_IDLE_TIMEOUT", "300") + .parse() + .unwrap_or(300), + max_payload_bytes: 4 * 1024 * 1024, + trace_path: env::var("OPENFUT_BLAZE_TRACE") + .ok() + .filter(|v| !v.trim().is_empty()), + adapter: AdapterConfig { + identity: Identity::default(), + endpoints, + server_version: AdapterConfig::default().server_version, + }, + }) + } + + pub fn listen_on(&self) -> String { + format!("{}:{}", self.listen_addr, self.listen_port) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Env is process-global, so these run under one lock rather than as + /// separate tests that would race each other. + #[test] + fn env_contract() { + let keys = [ + "OPENFUT_ADVERTISE", + "OPENFUT_BIND", + "OPENFUT_BLAZE_HOST_PORT", + "OPENFUT_BLAZE_HOST_BIND", + ]; + let saved: Vec<_> = keys.iter().map(|k| (*k, env::var(k).ok())).collect(); + for k in keys { + env::remove_var(k); + } + + // Missing advertise is refused, not defaulted. + let err = HostConfig::from_env().unwrap_err().to_string(); + assert!(err.contains("OPENFUT_ADVERTISE"), "{err}"); + + // Missing port is refused too — no default that could collide. + env::set_var("OPENFUT_ADVERTISE", "10.0.0.5"); + let err = HostConfig::from_env().unwrap_err().to_string(); + assert!(err.contains("OPENFUT_BLAZE_HOST_PORT"), "{err}"); + + // A non-numeric port is a clear error, not a silent fallback. + env::set_var("OPENFUT_BLAZE_HOST_PORT", "not-a-port"); + let err = HostConfig::from_env().unwrap_err().to_string(); + assert!(err.contains("not a valid port"), "{err}"); + + // Happy path: listener bind defaults to the config bind. + env::set_var("OPENFUT_BLAZE_HOST_PORT", "42230"); + env::set_var("OPENFUT_BIND", "0.0.0.0"); + let cfg = HostConfig::from_env().expect("configured"); + assert_eq!(cfg.listen_on(), "0.0.0.0:42230"); + assert_eq!(cfg.adapter.endpoints.advertise, "10.0.0.5"); + // The adapter's nucleus URL follows the CONFIG bind, reproducing the + // oracle's behaviour rather than the listener's address. + assert_eq!(cfg.adapter.nucleus_base(), "http://0.0.0.0:42131"); + + // The listener bind can differ from the advertised-config bind. + env::set_var("OPENFUT_BLAZE_HOST_BIND", "127.0.0.1"); + let cfg = HostConfig::from_env().expect("configured"); + assert_eq!(cfg.listen_on(), "127.0.0.1:42230"); + assert_eq!(cfg.adapter.nucleus_base(), "http://0.0.0.0:42131"); + + for (k, v) in saved { + match v { + Some(v) => env::set_var(k, v), + None => env::remove_var(k), + } + } + } +} diff --git a/openfut-blaze-host/src/conn.rs b/openfut-blaze-host/src/conn.rs new file mode 100644 index 0000000..91b9d38 --- /dev/null +++ b/openfut-blaze-host/src/conn.rs @@ -0,0 +1,264 @@ +//! Per-connection Fire2 stream handling. +//! +//! This is the layer fixtures cannot test: a socket delivers bytes, not frames. +//! Frames arrive split across reads and coalesced into one, several arrive per +//! connection, and the connection outlives every individual RPC. +//! +//! The loop mirrors the Python oracle's exactly — read a 16-byte header, derive +//! the total length from it, read the rest, consume, dispatch, write every +//! returned frame in order — because that behaviour is part of what was proven +//! against the real client. + +use std::io::{self, Read, Write}; +use std::net::TcpStream; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +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::config::HostConfig; +use crate::trace::{self, Tracer}; +use crate::Hooks; + +/// Why a connection ended. Logged so a live run can be compared with Python's. +#[derive(Debug, PartialEq, Eq)] +pub enum CloseReason { + /// Client closed cleanly between frames. + ClientClosed, + /// Stream ended part-way through a frame. + EofMidFrame { want: usize, have: usize }, + /// A header claimed a payload larger than the configured ceiling. + AbsurdPayload { claimed: u32 }, + /// No bytes within the idle timeout. + IdleTimeout, + /// Socket error. + Io(String), +} + +impl std::fmt::Display for CloseReason { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + CloseReason::ClientClosed => write!(f, "client closed"), + CloseReason::EofMidFrame { want, have } => { + write!(f, "EOF mid-frame (want {want}, have {have})") + } + CloseReason::AbsurdPayload { claimed } => { + write!(f, "absurd payload_len {claimed}, dropping connection") + } + CloseReason::IdleTimeout => write!(f, "idle timeout"), + CloseReason::Io(e) => write!(f, "io error: {e}"), + } + } +} + +pub(crate) fn unix_now() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs() as i64) + .unwrap_or(0) +} + +/// Mint a Blaze-shaped session key. +/// +/// The client never validates the format, so this only has to be stable within +/// a connection and distinct between them. +pub(crate) fn mint_session_key() -> String { + use rand::Rng; + let mut rng = rand::thread_rng(); + const ALPHA: &[u8] = b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789$*"; + let tail: String = (0..44) + .map(|_| ALPHA[rng.gen_range(0..ALPHA.len())] as char) + .collect(); + openfut_adapter_fifa17::blaze::session::format_session_key(rng.gen::(), &tail) +} + +/// Serve one connection until it closes. +pub fn handle( + mut stream: TcpStream, + conn_id: u64, + cfg: &HostConfig, + adapter: &Adapter, + tracer: &Tracer, + hooks: &Hooks, +) -> CloseReason { + let peer = stream + .peer_addr() + .map(|a| a.to_string()) + .unwrap_or_else(|_| "".into()); + + let _ = stream.set_read_timeout(Some(Duration::from_secs(cfg.idle_timeout_secs))); + // Blaze is request/response with server pushes; Nagle would add latency to + // every burst for no benefit. + let _ = stream.set_nodelay(true); + + let mut session = Session::new((hooks.session_key)(), cfg.adapter.identity.account_locale); + trace::log(&format!( + "conn-{conn_id:04} CONNECT from {peer} (session key minted: [REDACTED])" + )); + tracer.write_line(&format!("conn-{conn_id:04} OPEN")); + + let mut buf: Vec = Vec::with_capacity(16 * 1024); + let mut frame_no: u64 = 0; + + let reason = loop { + // ---- header + match fill_to(&mut stream, &mut buf, HEADER_LEN) { + Ok(true) => {} + Ok(false) => { + break if buf.is_empty() { + CloseReason::ClientClosed + } else { + CloseReason::EofMidFrame { + want: HEADER_LEN, + have: buf.len(), + } + }; + } + Err(e) => break classify(e), + } + + let header = match Header::parse(&buf[..HEADER_LEN]) { + Ok(h) => h, + // Unreachable: fill_to guaranteed 16 bytes. Treated as a close + // rather than a panic because this is a network path. + Err(e) => break CloseReason::Io(e.to_string()), + }; + + if header.payload_len > cfg.max_payload_bytes { + trace::log(&format!( + "conn-{conn_id:04} REJECT payload_len {} exceeds {} — {}", + header.payload_len, + cfg.max_payload_bytes, + hex_prefix(&buf) + )); + break CloseReason::AbsurdPayload { + claimed: header.payload_len, + }; + } + + // ---- body + let total = header.frame_len(); + match fill_to(&mut stream, &mut buf, total) { + Ok(true) => {} + Ok(false) => { + break CloseReason::EofMidFrame { + want: total, + have: buf.len(), + } + } + Err(e) => break classify(e), + } + + let (frame, used) = match Frame::parse(&buf[..total]) { + Ok(v) => v, + Err(e) => break CloseReason::Io(e.to_string()), + }; + buf.drain(..used); + + frame_no += 1; + trace::log(&format!( + "conn-{conn_id:04} RX #{frame_no} {}", + trace::describe(&frame.header) + )); + tracer.frame(conn_id, "RX", &frame_no.to_string(), &frame); + + // A body that will not decode is NOT fatal: the oracle logs it and + // dispatches with no fields, letting the RPC fall back to defaults. + // An empty Struct is equivalent to the oracle's `None` on every path + // dispatch takes (verified: every read is guarded or defaulted). + let body = if frame.payload.is_empty() { + Struct::new() + } else { + match heat2::decode(&frame.payload) { + Ok(s) => s, + Err(e) => { + trace::log(&format!( + "conn-{conn_id:04} RX #{frame_no} TDF DECODE FAILED: {e}" + )); + Struct::new() + } + } + }; + + let out = adapter.dispatch(&frame.header, &body, &mut session, (hooks.now)()); + + for (k, resp) in out.iter().enumerate() { + let bytes = resp.encode(); + if let Err(e) = stream.write_all(&bytes) { + trace::log(&format!("conn-{conn_id:04} TX #{frame_no}.{k} FAILED: {e}")); + break; + } + trace::log(&format!( + "conn-{conn_id:04} TX #{frame_no}.{k} {} ({}B total)", + trace::describe(&resp.header), + bytes.len() + )); + tracer.frame(conn_id, "TX", &format!("{frame_no}.{k}"), resp); + } + if let Err(e) = stream.flush() { + break CloseReason::Io(e.to_string()); + } + }; + + trace::log(&format!( + "conn-{conn_id:04} CLOSE after {frame_no} frame(s): {reason}" + )); + tracer.write_line(&format!("conn-{conn_id:04} CLOSE frames={frame_no}")); + reason +} + +/// Read until `buf` holds at least `n` bytes. `Ok(false)` on clean EOF. +fn fill_to(stream: &mut TcpStream, buf: &mut Vec, n: usize) -> io::Result { + let mut chunk = [0u8; 65536]; + while buf.len() < n { + match stream.read(&mut chunk) { + Ok(0) => return Ok(false), + Ok(got) => buf.extend_from_slice(&chunk[..got]), + Err(ref e) if e.kind() == io::ErrorKind::Interrupted => continue, + Err(e) => return Err(e), + } + } + Ok(true) +} + +fn classify(e: io::Error) -> CloseReason { + match e.kind() { + io::ErrorKind::WouldBlock | io::ErrorKind::TimedOut => CloseReason::IdleTimeout, + _ => CloseReason::Io(e.to_string()), + } +} + +fn hex_prefix(buf: &[u8]) -> String { + buf.iter() + .take(16) + .map(|b| format!("{b:02x}")) + .collect::>() + .join(" ") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn minted_keys_have_the_blaze_shape_and_differ() { + let a = mint_session_key(); + let b = mint_session_key(); + assert_eq!(a.len(), 16 + 1 + 44); + assert_ne!(a, b, "a session key must be distinct per connection"); + assert!(a[..16].chars().all(|c| c.is_ascii_hexdigit())); + assert_eq!(&a[16..17], "_"); + } + + #[test] + fn close_reasons_render_readably() { + assert!(CloseReason::ClientClosed.to_string().contains("closed")); + assert!(CloseReason::EofMidFrame { want: 20, have: 5 } + .to_string() + .contains("want 20")); + assert!(CloseReason::AbsurdPayload { claimed: 99 } + .to_string() + .contains("99")); + } +} diff --git a/openfut-blaze-host/src/lib.rs b/openfut-blaze-host/src/lib.rs new file mode 100644 index 0000000..c18f95b --- /dev/null +++ b/openfut-blaze-host/src/lib.rs @@ -0,0 +1,147 @@ +//! # openfut-blaze-host +//! +//! A deliberately thin TCP host for the FIFA 17 Blaze RPC surface. +//! +//! ```text +//! listener → Fire2 stream framing → per-connection Session +//! │ +//! openfut-adapter-fifa17::dispatch +//! │ +//! write returned frames, in order +//! ``` +//! +//! ## What it owns +//! +//! A socket, a read buffer, one `Session` per connection, and diagnostics. +//! That is the complete list. +//! +//! ## What it must never acquire +//! +//! Coins, club state, packs, profiles, market state, UTAS logic. Those belong +//! to OpenFUT Core, reached later through the adapter. A transport host that +//! starts holding game state becomes a second backend, which is exactly the +//! architecture this migration exists to avoid. +//! +//! ## No TLS +//! +//! The Blaze main port is **plaintext**. Verified against the running Python +//! backend by sending a raw Fire2 `Util::ping` and receiving a plaintext +//! `PingResponse`; `blaze_responder_v3b.py::blaze_handle` uses the raw socket, +//! and only `redir_handle` wraps `ssl`. TLS belongs to the redirector phase. +//! +//! ## Running beside Python, never instead of it +//! +//! The listen port is required configuration with no default, so this cannot +//! silently collide with the working container. See the crate README for the +//! A/B procedure and the gate list. + +pub mod config; +pub mod conn; +pub mod trace; + +pub use config::HostConfig; +pub use conn::CloseReason; + +use std::io; +use std::net::{SocketAddr, TcpListener}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; + +use openfut_adapter_fifa17::blaze::Adapter; + +/// The two values a live host reads from the outside world. +/// +/// Injectable so a test can replay a recorded conversation byte-for-byte: the +/// session key and the server clock appear inside response bodies, so with the +/// real ones no live run could ever reproduce a fixture exactly. This is the +/// only test seam in the crate, and it exists because the alternative is to +/// verify transport *structurally only* and lose the byte-level guarantee. +pub struct Hooks { + pub session_key: Box String + Send + Sync>, + pub now: Box i64 + Send + Sync>, +} + +impl Default for Hooks { + fn default() -> Hooks { + Hooks { + session_key: Box::new(conn::mint_session_key), + now: Box::new(conn::unix_now), + } + } +} + +impl std::fmt::Debug for Hooks { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Hooks { .. }") + } +} + +/// A bound listener plus everything a connection needs. +pub struct Server { + pub local_addr: SocketAddr, + listener: TcpListener, + cfg: Arc, + adapter: Arc, + tracer: Arc, + hooks: Arc, +} + +/// Bind without accepting, so a caller can learn the real port first (useful +/// when binding port 0). +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 adapter = Arc::new(Adapter::new(cfg.adapter.clone())); + Ok(Server { + local_addr, + listener, + cfg: Arc::new(cfg), + adapter, + tracer, + hooks: Arc::new(hooks), + }) +} + +impl Server { + /// Accept forever, one thread per connection. + /// + /// A FIFA client opens a handful of connections, so a thread each mirrors + /// the Python oracle and keeps the host readable; a work-stealing runtime + /// would be weight without benefit. + pub fn run(self) -> io::Result<()> { + trace::log(&format!( + "blaze-host listening on {} (advertise={}, config-bind={})", + self.local_addr, self.cfg.adapter.endpoints.advertise, self.cfg.adapter.endpoints.bind + )); + if self.tracer.enabled() { + trace::log(&format!( + "structural trace -> {}", + self.cfg.trace_path.as_deref().unwrap_or("") + )); + } + + let counter = AtomicU64::new(0); + for incoming in self.listener.incoming() { + match incoming { + Ok(stream) => { + let id = counter.fetch_add(1, Ordering::Relaxed) + 1; + let cfg = self.cfg.clone(); + let adapter = self.adapter.clone(); + let tracer = self.tracer.clone(); + let hooks = self.hooks.clone(); + std::thread::spawn(move || { + conn::handle(stream, id, &cfg, &adapter, &tracer, &hooks); + }); + } + Err(e) => trace::log(&format!("accept failed: {e}")), + } + } + Ok(()) + } +} + +/// Bind and serve until the process is killed. +pub fn serve(cfg: HostConfig) -> io::Result<()> { + bind(cfg, Hooks::default())?.run() +} diff --git a/openfut-blaze-host/src/main.rs b/openfut-blaze-host/src/main.rs new file mode 100644 index 0000000..c2b56aa --- /dev/null +++ b/openfut-blaze-host/src/main.rs @@ -0,0 +1,38 @@ +//! Entry point for the Blaze sidecar. +//! +//! Configuration is entirely environmental and both critical values are +//! required, so a misconfigured launch fails immediately and loudly rather than +//! binding a default port next to the working Python container. + +use openfut_blaze_host::{serve, HostConfig}; + +fn main() { + let cfg = match HostConfig::from_env() { + Ok(cfg) => cfg, + Err(e) => { + eprintln!("openfut-blaze-host: {e}\n"); + eprintln!("Required:"); + eprintln!( + " OPENFUT_ADVERTISE address the game machine uses to reach this host" + ); + eprintln!( + " OPENFUT_BLAZE_HOST_PORT port to listen on (no default, to avoid colliding" + ); + eprintln!(" with the Python backend it runs beside)"); + eprintln!("Optional:"); + eprintln!(" OPENFUT_BIND advertised-config bind, mirrors the Python"); + eprintln!(" container's value so nucleusConnect matches"); + eprintln!(" OPENFUT_BLAZE_HOST_BIND listener bind (defaults to OPENFUT_BIND)"); + eprintln!(" POW_CONTENT_HOST host:port for POW content"); + eprintln!(" POW_HOST host:port for the POW/EASFC API"); + eprintln!(" OPENFUT_BLAZE_TRACE path for the normalized structural trace"); + eprintln!(" OPENFUT_BLAZE_IDLE_TIMEOUT seconds, default 300"); + std::process::exit(2); + } + }; + + if let Err(e) = serve(cfg) { + eprintln!("openfut-blaze-host: fatal: {e}"); + std::process::exit(1); + } +} diff --git a/openfut-blaze-host/src/trace.rs b/openfut-blaze-host/src/trace.rs new file mode 100644 index 0000000..24d2020 --- /dev/null +++ b/openfut-blaze-host/src/trace.rs @@ -0,0 +1,300 @@ +//! Diagnostics and the normalized structural trace. +//! +//! Two outputs with different jobs: +//! +//! * **The log** is for a human watching a live run. Free-form, timestamped. +//! * **The trace** is for `diff`. One line per frame, deterministic, with every +//! volatile value masked so a Python session and a Rust session of the same +//! conversation produce identical text. +//! +//! Masking is what makes the trace useful. A session key, a server timestamp +//! and an auth token differ on every run by design, so comparing raw bytes +//! across two live servers can only ever fail. Their *presence, tag, type and +//! length* are what must match, and that is what the trace records. +//! +//! Nothing here writes a credential: masked values are replaced before they +//! reach the line, not truncated after. + +use std::fmt::Write as _; +use std::io::Write as _; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +use openfut_adapter_fifa17::blaze::ids; +use openfut_protocol_blaze::fire2::{Frame, Header, MsgType}; +use openfut_protocol_blaze::heat2::{self, Struct, Value}; + +/// Tags whose values legitimately differ between two runs of the same +/// conversation. Masked in the trace so it stays diffable. +/// +/// Deliberately a denylist of *known-volatile* tags rather than an allowlist: +/// a new field appearing in a response should show up as a diff, not be hidden. +const VOLATILE_TAGS: &[&str] = &[ + "KEY", // session key + "AUTH", // auth token (also a credential-shaped value) + "STIM", // server time + "LLOG", "LAST", "LADT", "LATH", // login / auth timestamps + "GDAY", "DTCR", // grant/create dates (stable today, timestamp-shaped) + "SESS", // telemetry session echo (string form) +]; + +fn now_millis() -> u128 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis()) + .unwrap_or(0) +} + +pub fn log(msg: &str) { + let ms = now_millis(); + eprintln!("[{}.{:03}] {msg}", ms / 1000, ms % 1000); +} + +/// Describe a frame header the way the Python responder logs it, so the two +/// logs can be read side by side. +pub fn describe(header: &Header) -> String { + let is_notify = header.msg_type == MsgType::Notification; + format!( + "{} msgType={} msgNum={} userIdx={} opts=0x{:02x} meta={}B payload={}B", + ids::describe(header.component, header.command, is_notify), + header.msg_type.name(), + header.msg_num, + header.user_index, + header.options, + header.metadata_len, + header.payload_len + ) +} + +/// A deterministic, volatile-masked rendering of one frame. +pub struct Tracer { + sink: Option>, +} + +impl Tracer { + pub fn new(path: Option<&str>) -> std::io::Result { + Ok(Tracer { + sink: match path { + Some(p) => Some(Mutex::new(std::fs::File::create(p)?)), + None => None, + }, + }) + } + + pub fn disabled() -> Tracer { + Tracer { sink: None } + } + + pub fn enabled(&self) -> bool { + self.sink.is_some() + } + + pub fn write_line(&self, line: &str) { + if let Some(sink) = &self.sink { + if let Ok(mut f) = sink.lock() { + let _ = writeln!(f, "{line}"); + let _ = f.flush(); + } + } + } + + pub fn frame(&self, conn: u64, direction: &str, index: &str, frame: &Frame) { + if !self.enabled() { + return; + } + self.write_line(&trace_line(conn, direction, index, frame)); + } +} + +/// `conn-0001 TX 1.0 Authentication::login REPLY num=12 uidx=0 payload=201 tdf=` +pub fn trace_line(conn: u64, direction: &str, index: &str, frame: &Frame) -> String { + let h = &frame.header; + let is_notify = h.msg_type == MsgType::Notification; + let body = if frame.payload.is_empty() { + "(empty)".to_string() + } else { + match heat2::decode(&frame.payload) { + Ok(s) => format!("{:016x}", structure_digest(&s)), + Err(_) => "DECODE-FAILED".to_string(), + } + }; + format!( + "conn-{conn:04} {direction} {index} {} {} num={} uidx={} payload={} tdf={}", + ids::describe(h.component, h.command, is_notify), + h.msg_type.name(), + h.msg_num, + h.user_index, + frame.payload.len(), + body + ) +} + +/// A masked, human-readable dump of a decoded body. Used by the probe's +/// verbose mode when a digest mismatch needs explaining. +pub fn masked_dump(s: &Struct) -> String { + let mut out = String::new(); + write_masked(s, 0, &mut out); + out +} + +fn write_masked(s: &Struct, depth: usize, out: &mut String) { + // Sort so two implementations that differ only in member order still + // produce identical text. (They should not — the encoder sorts — but the + // trace must not be the thing that hides it if they do.) + let mut fields: Vec<_> = s.iter().collect(); + fields.sort_by_key(|(t, _)| *t); + + for (tag, value) in fields { + let pad = " ".repeat(depth); + let label = tag.to_label(); + let volatile = VOLATILE_TAGS.contains(&label.as_str()); + match value { + Value::Struct(inner) => { + let _ = writeln!(out, "{pad}{label} (struct)"); + write_masked(inner, depth + 1, out); + } + Value::List { elem, items } => { + let _ = writeln!(out, "{pad}{label} (list[{}] x{})", elem.name(), items.len()); + for it in items { + if let Value::Struct(inner) = it { + write_masked(inner, depth + 1, out); + } + } + } + Value::Map { key, val, entries } => { + let _ = writeln!( + out, + "{pad}{label} (map[{}->{}] x{})", + key.name(), + val.name(), + entries.len() + ); + } + other => { + let rendered = if volatile { + masked_scalar(other) + } else { + render_scalar(other) + }; + let _ = writeln!( + out, + "{pad}{label} ({}) = {rendered}", + other.type_id().name() + ); + } + } + } +} + +/// Replace the value but keep its shape, so a length or type change still diffs. +fn masked_scalar(v: &Value) -> String { + match v { + Value::String(s) => format!("", s.len()), + Value::Int(_) => "".to_string(), + Value::Blob(b) => format!("", b.len()), + other => format!("", other.type_id().name()), + } +} + +fn render_scalar(v: &Value) -> String { + match v { + Value::Int(n) => n.to_string(), + Value::String(s) => format!("{s:?}"), + Value::Blob(b) => format!("", b.len()), + Value::VarList(items) => format!("{items:?}"), + Value::Float(f) => format!("{f}"), + Value::ObjType { component, ty } => format!("({component},{ty})"), + Value::ObjId { component, ty, id } => format!("({component},{ty},{id})"), + other => other.type_id().name().to_string(), + } +} + +/// FNV-1a over the masked dump: a stable structural fingerprint. +/// +/// Hand-rolled because it is eight lines and pulling a hashing crate into a +/// diagnostics path would be the heavier choice. Not a security hash and never +/// used as one. +pub fn structure_digest(s: &Struct) -> u64 { + let mut hash: u64 = 0xcbf2_9ce4_8422_2325; + for byte in masked_dump(s).as_bytes() { + hash ^= *byte as u64; + hash = hash.wrapping_mul(0x1000_0000_01b3); + } + hash +} + +#[cfg(test)] +mod tests { + use super::*; + + fn frame_with(body: Struct) -> Frame { + Frame::new(0x0001, 0x000A, 12, MsgType::Reply, heat2::encode(&body)) + } + + #[test] + fn volatile_values_are_masked_but_their_shape_survives() { + let a = Struct::new() + .with("KEY", Value::String("aaaaaaaa".into())) + .with("UID", Value::Int(33068179)); + let b = Struct::new() + .with("KEY", Value::String("bbbbbbbb".into())) + .with("UID", Value::Int(33068179)); + // Different session keys of the same length: identical trace. + assert_eq!(structure_digest(&a), structure_digest(&b)); + + // A different length is a real change and must still diff. + let c = Struct::new() + .with("KEY", Value::String("short".into())) + .with("UID", Value::Int(33068179)); + assert_ne!(structure_digest(&a), structure_digest(&c)); + } + + #[test] + fn non_volatile_values_are_not_masked() { + let a = Struct::new().with("UID", Value::Int(1)); + let b = Struct::new().with("UID", Value::Int(2)); + assert_ne!(structure_digest(&a), structure_digest(&b)); + } + + #[test] + fn a_new_field_shows_up_as_a_difference() { + // The denylist must not hide additions. + let a = Struct::new().with("UID", Value::Int(1)); + let b = Struct::new() + .with("UID", Value::Int(1)) + .with("NEWF", Value::Int(0)); + assert_ne!(structure_digest(&a), structure_digest(&b)); + } + + #[test] + fn trace_line_is_stable_across_sessions() { + let a = frame_with(Struct::new().with("KEY", Value::String("k1-aaaaaa".into()))); + let b = frame_with(Struct::new().with("KEY", Value::String("k2-bbbbbb".into()))); + assert_eq!( + trace_line(1, "TX", "1.0", &a), + trace_line(1, "TX", "1.0", &b) + ); + assert!(trace_line(1, "TX", "1.0", &a).contains("Authentication::login")); + } + + #[test] + fn no_masked_value_leaks_into_the_line() { + let secret = "SUPER-SECRET-SESSION-KEY"; + let f = frame_with(Struct::new().with("KEY", Value::String(secret.into()))); + let line = trace_line(1, "TX", "1.0", &f); + assert!(!line.contains(secret)); + assert!(!masked_dump(&heat2::decode(&f.payload).unwrap()).contains(secret)); + } + + #[test] + fn an_undecodable_body_is_reported_not_hidden() { + let f = Frame::new(9, 7, 1, MsgType::Reply, vec![0xFF; 8]); + assert!(trace_line(1, "TX", "1.0", &f).contains("DECODE-FAILED")); + } + + #[test] + fn empty_payloads_are_distinguishable_from_failures() { + let f = Frame::new(9, 7, 1, MsgType::Reply, vec![]); + assert!(trace_line(1, "TX", "1.0", &f).contains("(empty)")); + } +} diff --git a/openfut-blaze-host/tests/live_transport.rs b/openfut-blaze-host/tests/live_transport.rs new file mode 100644 index 0000000..16ab623 --- /dev/null +++ b/openfut-blaze-host/tests/live_transport.rs @@ -0,0 +1,472 @@ +//! Transport parity: the recorded conversations over a real TCP socket. +//! +//! The adapter's own suite calls `dispatch()` directly. That proves what the +//! adapter decides, and nothing about what a socket does. This suite runs the +//! actual host process-internally, connects over real TCP, and replays the same +//! recorded conversations — which is what exercises the things fixtures cannot: +//! +//! * a stream that delivers bytes, not frames +//! * requests split across reads and coalesced into one +//! * many frames per connection, and session state surviving between them +//! * multiple frames written back in order for a single request +//! * connection lifecycle and close reasons +//! +//! Byte-exactness is only achievable because `Hooks` injects the session key +//! and clock; with the real ones, no live run could reproduce a recording, and +//! the guarantee would drop to structural. + +use std::io::{Read, Write}; +use std::net::TcpStream; +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_protocol_blaze::fire2::{Frame, Header, HEADER_LEN}; +use serde_json::Value as J; + +fn records() -> Vec { + let path = format!( + "{}/../openfut-adapter-fifa17/fixtures/blaze_transactions.jsonl", + env!("CARGO_MANIFEST_DIR") + ); + std::fs::read_to_string(&path) + .unwrap_or_else(|e| panic!("cannot read {path}: {e}")) + .lines() + .filter(|l| !l.trim().is_empty()) + .map(|l| serde_json::from_str(l).expect("valid JSON")) + .collect() +} + +fn unhex(s: &str) -> Vec { + (0..s.len()) + .step_by(2) + .map(|i| u8::from_str_radix(&s[i..i + 2], 16).expect("valid hex")) + .collect() +} + +fn hex(b: &[u8]) -> String { + b.iter().map(|x| format!("{x:02x}")).collect() +} + +fn st(j: &J, k: &str) -> String { + j[k].as_str().expect("string").to_string() +} + +fn num(j: &J, k: &str) -> i64 { + j[k].as_i64().expect("number") +} + +struct Harness { + addr: String, + records: Vec, +} + +/// Start the real host on an ephemeral port with the recording's config. +/// +/// Session keys are handed out in the order the recording declares them, so the +/// 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 { + let records = records(); + let cfg_rec = records + .iter() + .find(|r| r["kind"] == "config") + .expect("config record") + .clone(); + let id = &cfg_rec["identity"]; + + let adapter_cfg = AdapterConfig { + identity: Identity { + persona_id: num(id, "persona_id"), + persona_name: st(id, "persona_name"), + user_id: num(id, "user_id"), + ext_id: num(id, "ext_id"), + email: st(id, "email"), + namespace: st(id, "namespace"), + client_platform: num(id, "client_platform"), + persona_status: num(id, "persona_status"), + user_session_type: num(id, "user_session_type"), + account_locale: num(id, "account_locale_int"), + locale: st(id, "locale"), + content_id: st(id, "content_id"), + entitlement_tag: st(id, "entitlement_tag"), + entitlement_group: st(id, "entitlement_group"), + title_id: st(id, "title_id"), + client_id: st(id, "client_id"), + platform: st(id, "platform"), + }, + endpoints: Endpoints { + advertise: st(&cfg_rec, "advertise"), + bind: st(&cfg_rec, "bind"), + pow_content_host: st(&cfg_rec, "pow_content_host"), + pow_host: st(&cfg_rec, "pow_host"), + ..Endpoints::default() + }, + server_version: st(id, "server_version"), + }; + + let host_cfg = HostConfig { + // Port 0: the OS picks a free one, so this can never collide with the + // running Python backend or with a parallel test. + listen_addr: "127.0.0.1".into(), + listen_port: 0, + idle_timeout_secs: 10, + max_payload_bytes: 4 * 1024 * 1024, + trace_path: None, + adapter: adapter_cfg, + }; + + let keys: Vec = records + .iter() + .filter(|r| r["kind"] == "session") + .map(|r| st(r, "session_key")) + .collect(); + let queue = Mutex::new(keys.into_iter().collect::>()); + let now = num(&cfg_rec, "now"); + + let hooks = Hooks { + session_key: Box::new(move || { + queue + .lock() + .expect("key queue") + .pop_front() + .expect("a recorded session key for each connection") + }), + now: Box::new(move || now), + }; + + let server = bind(host_cfg, hooks).expect("bind ephemeral port"); + let addr = server.local_addr.to_string(); + std::thread::spawn(move || { + let _ = server.run(); + }); + + Harness { addr, records } +} + +fn connect(addr: &str) -> TcpStream { + let s = TcpStream::connect(addr).expect("connect to host"); + s.set_read_timeout(Some(Duration::from_secs(10))).unwrap(); + s.set_nodelay(true).unwrap(); + s +} + +fn read_frame(stream: &mut TcpStream, buf: &mut Vec) -> Option { + let mut chunk = [0u8; 65536]; + while buf.len() < HEADER_LEN { + match stream.read(&mut chunk) { + Ok(0) | Err(_) => return None, + Ok(got) => buf.extend_from_slice(&chunk[..got]), + } + } + let total = Header::parse(&buf[..HEADER_LEN]).ok()?.frame_len(); + while buf.len() < total { + match stream.read(&mut chunk) { + Ok(0) | Err(_) => return None, + Ok(got) => buf.extend_from_slice(&chunk[..got]), + } + } + let (frame, used) = Frame::parse(&buf[..total]).ok()?; + buf.drain(..used); + Some(frame) +} + +/// Transactions for one recorded session, in order. +fn transactions<'a>(records: &'a [J], session: &str) -> Vec<&'a J> { + records + .iter() + .filter(|r| r["kind"] == "tx" && r["session"] == session) + .collect() +} + +fn session_ids(records: &[J]) -> Vec { + records + .iter() + .filter(|r| r["kind"] == "session") + .map(|r| st(r, "id")) + .collect() +} + +/// The headline transport test: every recorded conversation, over TCP, +/// byte-for-byte, one connection per recorded session. +#[test] +fn recorded_conversations_replay_byte_for_byte_over_tcp() { + let h = start(); + let mut total_frames = 0usize; + + for sid in session_ids(&h.records) { + let mut stream = connect(&h.addr); + let mut buf = Vec::new(); + + for tx in transactions(&h.records, &sid) { + let name = st(tx, "name"); + let request = unhex(&st(tx, "request_hex")); + let expected: Vec = tx["responses"] + .as_array() + .unwrap() + .iter() + .map(|v| v.as_str().unwrap().to_string()) + .collect(); + + stream.write_all(&request).expect("send request"); + stream.flush().unwrap(); + + for (i, want) in expected.iter().enumerate() { + let frame = read_frame(&mut stream, &mut buf) + .unwrap_or_else(|| panic!("{sid}/{name}: no frame {i}, expected one")); + assert_eq!( + hex(&frame.encode()), + *want, + "\n{sid}/{name}: frame {i} differs over the wire" + ); + total_frames += 1; + } + } + + // A clean close between frames must be recognised as such. + drop(stream); + } + + assert!( + total_frames >= 50, + "expected the full recorded conversation, saw {total_frames} frames" + ); +} + +/// The same conversation with every request written ONE BYTE AT A TIME. +/// +/// This is the fragmentation case: a real client's frames arrive split across +/// reads, and a host that assumed one read per frame would pass every fixture +/// test and then fail against FIFA. +#[test] +fn requests_split_across_reads_are_reassembled() { + let h = start(); + 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(6) { + let name = st(tx, "name"); + let request = unhex(&st(tx, "request_hex")); + for byte in &request { + stream.write_all(&[*byte]).expect("dribble a byte"); + stream.flush().unwrap(); + } + + for (i, want) in tx["responses"].as_array().unwrap().iter().enumerate() { + let frame = read_frame(&mut stream, &mut buf) + .unwrap_or_else(|| panic!("{name}: no frame {i} after fragmented send")); + assert_eq!( + hex(&frame.encode()), + want.as_str().unwrap(), + "{name} frame {i}" + ); + } + } +} + +/// Several requests written as ONE write, so the host sees them coalesced in a +/// single read and must split them itself. +#[test] +fn coalesced_requests_in_one_write_are_split() { + let h = start(); + 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).expect("one big write"); + stream.flush().unwrap(); + + for tx in &txs { + let name = st(tx, "name"); + for (i, want) in tx["responses"].as_array().unwrap().iter().enumerate() { + let frame = read_frame(&mut stream, &mut buf) + .unwrap_or_else(|| panic!("{name}: no frame {i} after coalesced send")); + assert_eq!( + hex(&frame.encode()), + want.as_str().unwrap(), + "{name} frame {i}" + ); + } + } +} + +/// Login must deliver four frames in order on a real socket, not just from +/// `dispatch()`. This is the ordering guarantee the client depends on. +#[test] +fn login_burst_arrives_in_order_over_the_wire() { + let h = start(); + 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) { + let request = unhex(&st(tx, "request_hex")); + stream.write_all(&request).unwrap(); + stream.flush().unwrap(); + + let expected = tx["responses"].as_array().unwrap().len(); + let mut got = Vec::new(); + for _ in 0..expected { + got.push(read_frame(&mut stream, &mut buf).expect("frame")); + } + + if st(tx, "name") == "login" { + assert_eq!(got.len(), 4, "reply + three pushes"); + let routes: Vec<(u16, u16)> = got + .iter() + .map(|f| (f.header.component, f.header.command)) + .collect(); + assert_eq!( + routes, + vec![ + (0x0001, 0x000A), // Authentication::login REPLY + (0x7802, 0x0008), // UserAuthenticated + (0x7802, 0x0001), // UserSessionExtendedDataUpdate + (0x7802, 0x0002), // UserAdded + ] + ); + return; + } + } + panic!("login transaction never reached"); +} + +/// Session state must persist across frames on one connection: the auth code +/// login records has to still be there when getAuthToken asks for it later. +#[test] +fn session_state_persists_across_frames_on_one_connection() { + let h = start(); + let sid = session_ids(&h.records).into_iter().next().unwrap(); + let mut stream = connect(&h.addr); + let mut buf = Vec::new(); + + let txs = transactions(&h.records, &sid); + let mut saw_before = false; + let mut saw_after = false; + + for tx in txs { + let name = st(tx, "name"); + stream.write_all(&unhex(&st(tx, "request_hex"))).unwrap(); + stream.flush().unwrap(); + + for want in tx["responses"].as_array().unwrap() { + let frame = read_frame(&mut stream, &mut buf).expect("frame"); + let got = hex(&frame.encode()); + assert_eq!(got, want.as_str().unwrap(), "{name}"); + + if name == "get_auth_token_before_login" { + let body = openfut_protocol_blaze::heat2::decode(&frame.payload).unwrap(); + let tok = body.get("AUTH").and_then(|v| v.as_str()).unwrap(); + assert!( + tok.starts_with("OPENFUT-"), + "synthesised before login: {tok}" + ); + saw_before = true; + } + if name == "get_auth_token_after_login" { + let body = openfut_protocol_blaze::heat2::decode(&frame.payload).unwrap(); + let tok = body.get("AUTH").and_then(|v| v.as_str()).unwrap(); + assert_eq!(tok, "OPENFUT-TEST-AUTHCODE", "echoes the login's code"); + saw_after = true; + } + } + } + assert!( + saw_before && saw_after, + "both getAuthToken phases must be exercised" + ); +} + +/// A separate connection must get a separate session — no leakage. +#[test] +fn a_second_connection_gets_a_fresh_session() { + let h = start(); + let ids = session_ids(&h.records); + + // Connection 1: log in. + let mut c1 = connect(&h.addr); + let mut b1 = Vec::new(); + for tx in transactions(&h.records, &ids[0]) { + c1.write_all(&unhex(&st(tx, "request_hex"))).unwrap(); + c1.flush().unwrap(); + for _ in 0..tx["responses"].as_array().unwrap().len() { + read_frame(&mut c1, &mut b1).expect("frame"); + } + } + + // Connection 2 asks for an auth token without logging in. If session state + // leaked it would echo connection 1's code instead of synthesising one. + let mut c2 = connect(&h.addr); + let mut b2 = Vec::new(); + let get_token = transactions(&h.records, &ids[0]) + .into_iter() + .find(|t| st(t, "name") == "get_auth_token_before_login") + .expect("fixture present"); + c2.write_all(&unhex(&st(get_token, "request_hex"))).unwrap(); + c2.flush().unwrap(); + + let frame = read_frame(&mut c2, &mut b2).expect("frame"); + let body = openfut_protocol_blaze::heat2::decode(&frame.payload).unwrap(); + let tok = body.get("AUTH").and_then(|v| v.as_str()).unwrap(); + assert!( + tok.starts_with("OPENFUT-"), + "a fresh connection must not inherit a login: {tok}" + ); + assert_ne!(tok, "OPENFUT-TEST-AUTHCODE"); +} + +/// A header claiming an absurd payload must drop the connection rather than +/// try to allocate for it. +#[test] +fn an_absurd_payload_length_drops_the_connection() { + let h = start(); + let mut stream = connect(&h.addr); + + let mut header = [0u8; HEADER_LEN]; + header[0..4].copy_from_slice(&0x7FFF_FFFFu32.to_be_bytes()); // ~2 GiB + header[6..8].copy_from_slice(&0x0009u16.to_be_bytes()); + header[8..10].copy_from_slice(&0x0002u16.to_be_bytes()); + stream.write_all(&header).unwrap(); + stream.flush().unwrap(); + + let mut buf = Vec::new(); + assert!( + read_frame(&mut stream, &mut buf).is_none(), + "the host must close, not answer, an absurd frame" + ); +} + +/// A body that will not decode as TDF must not kill the connection: the oracle +/// logs it and dispatches with no fields. +#[test] +fn an_undecodable_body_still_gets_a_reply() { + let h = start(); + let mut stream = connect(&h.addr); + + // Util::ping with a garbage payload. ping ignores its body, so the reply + // must arrive exactly as if the body had been empty. + let mut frame = Frame::new( + 0x0009, + 0x0002, + 1, + openfut_protocol_blaze::fire2::MsgType::Message, + vec![0xFF; 8], + ); + frame.header.payload_len = 8; + stream.write_all(&frame.encode()).unwrap(); + stream.flush().unwrap(); + + let mut buf = Vec::new(); + let reply = read_frame(&mut stream, &mut buf).expect("a reply despite the bad body"); + assert_eq!(reply.header.component, 0x0009); + assert_eq!(reply.header.command, 0x0002); + assert!(!reply.payload.is_empty(), "ping still answers with STIM"); +}