OpenFUT/scripts/test-utas-observe.py

#!/usr/bin/env python3
"""Prove the UTAS recorder is byte-transparent and parses what it saw.

Two independent properties, because they fail differently:

  TRANSPARENCY  what the client receives through the proxy is byte-identical to
                what it receives going direct. If this breaks, the tool corrupts
                the session it exists to observe -- and the damage would look
                like a UTAS bug.

  FIDELITY      the parsed transactions match what was actually sent. If this
                breaks, only the record is wrong, which is why the design
                separates the two.

The upstream here is a purpose-built server, not the live oracle: it can be made
to exercise keep-alive, bodies in both directions and chunked encoding on
demand, and no live client is at risk.
"""
import base64
import http.server
import json
import os
import shutil
import socket
import subprocess
import sys
import threading
import time

TOOL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "openfut-utas-observe.py")
SESSION = os.path.join(os.environ.get("TMPDIR", "/tmp"), "openfut-utas-observe-selftest")

fails = []


def check(name, ok, detail=""):
    print(("  ok   " if ok else "  FAIL ") + name + (("  " + detail) if detail and not ok else ""))
    if not ok:
        fails.append(name)


class H(http.server.BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"          # keep-alive, like the real UTAS

    def log_message(self, *a):
        pass

    def _read_body(self):
        n = int(self.headers.get("Content-Length", 0) or 0)
        return self.rfile.read(n) if n else b""

    def do_GET(self):
        if self.path.split("?")[0] == "/chunked":
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Transfer-Encoding", "chunked")
            self.end_headers()
            for part in (b'{"a":1,', b'"b":[2,3],', b'"c":"end"}'):
                self.wfile.write(b"%x\r\n" % len(part) + part + b"\r\n")
            self.wfile.write(b"0\r\n\r\n")
            return
        body = json.dumps({"path": self.path}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        got = self._read_body()
        body = json.dumps({"echo_len": len(got),
                           "echo": got.decode("latin1"),
                           "sid": "SECRET-SESSION-VALUE"}).encode()
        self.send_response(201)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


def free_port():
    s = socket.socket()
    s.bind(("127.0.0.1", 0))
    p = s.getsockname()[1]
    s.close()
    return p


def raw_exchange(port, requests):
    """Send requests on ONE keep-alive connection; return all bytes received."""
    s = socket.create_connection(("127.0.0.1", port), timeout=5)
    for r in requests:
        s.sendall(r)
        time.sleep(0.15)
    s.settimeout(1.5)
    out = b""
    try:
        while True:
            d = s.recv(65536)
            if not d:
                break
            out += d
    except socket.timeout:
        pass
    s.close()
    return out


def main():
    shutil.rmtree(SESSION, ignore_errors=True)
    up_port, proxy_port = free_port(), free_port()

    srv = http.server.ThreadingHTTPServer(("127.0.0.1", up_port), H)
    threading.Thread(target=srv.serve_forever, daemon=True).start()

    proxy = subprocess.Popen(
        [sys.executable, TOOL, "record",
         "--listen", "127.0.0.1:%d" % proxy_port,
         "--upstream", "127.0.0.1:%d" % up_port,
         "--session", SESSION],
        stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    time.sleep(1.0)

    body = json.dumps({"squad": [1, 2, 3], "note": "x" * 300}).encode()
    reqs = [
        b"GET /ut/game/fifa17/sbs/sets HTTP/1.1\r\nHost: t\r\n\r\n",
        b"POST /ut/game/fifa17/sbs/challenge/101/squad HTTP/1.1\r\nHost: t\r\n"
        b"Content-Type: application/json\r\nContent-Length: %d\r\n\r\n" % len(body) + body,
        b"GET /chunked?deviceId=DEADBEEFCAFE&keep=yes HTTP/1.1\r\nHost: t\r\n\r\n",
        b"GET /ut/game/fifa17/hub HTTP/1.1\r\nHost: t\r\nConnection: close\r\n\r\n",
    ]

    print("== transparency ==")
    direct = raw_exchange(up_port, reqs)
    through = raw_exchange(proxy_port, reqs)
    # The two exchanges happen seconds apart, so `Date:` legitimately differs.
    # Mask it -- but ONLY it, and assert the mask actually fired, so this cannot
    # quietly hide a real difference.
    import re as _re
    def mask(b):
        return _re.sub(rb"Date: [^\r\n]+", b"Date: <M>", b)
    dm, tm = mask(direct), mask(through)
    check("masking Date actually applied", dm != direct and tm != through)
    check("bytes through the proxy are identical to bytes direct (Date masked)",
          dm == tm,
          "direct=%dB through=%dB firstdiff=%s" % (
              len(direct), len(through),
              next((i for i in range(min(len(dm), len(tm))) if dm[i] != tm[i]), "len")))
    check("nothing but Date differed in the unmasked bytes",
          len(direct) == len(through))
    check("all four responses arrived", through.count(b"HTTP/1.1 ") == 4,
          "saw %d" % through.count(b"HTTP/1.1 "))
    check("keep-alive was actually used (one connection, four responses)",
          through.count(b"HTTP/1.1 ") == 4)

    time.sleep(0.6)
    proxy.terminate()
    try:
        proxy.wait(timeout=5)
    except subprocess.TimeoutExpired:
        proxy.kill()

    print("== capture file ==")
    raw = os.path.join(SESSION, "raw", "utas.ofcap")
    check("raw capture exists", os.path.exists(raw))
    check("raw capture is mode 0600",
          oct(os.stat(raw).st_mode & 0o777) == "0o600",
          oct(os.stat(raw).st_mode & 0o777))

    print("== fidelity ==")
    r = subprocess.run([sys.executable, TOOL, "parse", "--session", SESSION],
                       capture_output=True, text=True)
    print("     " + r.stdout.strip().replace("\n", "\n     "))
    if r.stderr.strip():
        print("     stderr: " + r.stderr.strip().replace("\n", "\n     "))
    txs = [json.loads(l) for l in
           open(os.path.join(SESSION, "sanitized", "transactions.jsonl"))]
    # Two client connections were made (direct + through); only one went via the proxy.
    check("four transactions parsed", len(txs) == 4, "got %d" % len(txs))
    if len(txs) == 4:
        check("methods and paths in order",
              [(t["request"]["method"], t["request"]["path"]) for t in txs] ==
              [("GET", "/ut/game/fifa17/sbs/sets"),
               ("POST", "/ut/game/fifa17/sbs/challenge/101/squad"),
               ("GET", "/chunked"),
               ("GET", "/ut/game/fifa17/hub")])
        check("request body preserved byte-for-byte",
              base64.b64decode(txs[1]["request"]["body_b64"]) == body)
        check("statuses recorded", [t["response"]["status"] for t in txs] == [200, 201, 200, 200])
        chunked = base64.b64decode(txs[2]["response"]["body_b64"])
        check("chunked response body dechunked correctly",
              chunked == b'{"a":1,"b":[2,3],"c":"end"}', chunked[:60].decode("latin1"))
        check("headers kept in received order, as pairs",
              txs[0]["response"]["headers"][0][0] == "Server")
        check("keep-alive recorded", txs[0]["connection"]["requests_on_this_connection"] == 4)
        check("timing recorded", all(t["elapsed_ms"] is not None for t in txs))
        # Sanitiser: the secret goes, the rest of the body does not.
        echoed = json.loads(base64.b64decode(txs[1]["response"]["body_b64"]))
        check("secret JSON value redacted", echoed.get("sid") == "<REDACTED>")
        check("non-secret payload preserved exactly", echoed.get("echo_len") == len(body))
        check("redaction is recorded, not silent", "redacted" in txs[1])
        # Regression: the sanitiser handled headers and JSON bodies but not the
        # QUERY STRING, and was one audit away from publishing a device id from
        # `?deviceId=...`. Three surfaces carry identifiers, not two.
        qtx = next((t for t in txs if t["request"]["query"]), None)
        check("a secret query parameter is redacted",
              qtx is not None and "deviceId=<REDACTED>" in qtx["request"]["query"],
              qtx["request"]["query"] if qtx else "no query captured")
        check("a non-secret query parameter is preserved",
              qtx is not None and "keep=yes" in qtx["request"]["query"],
              qtx["request"]["query"] if qtx else "")

    print("== path-scoped fixture ==")
    r = subprocess.run(
        [sys.executable, TOOL, "parse", "--session", SESSION,
         "--path-prefix", "/ut/game/fifa17/sbs/"],
        capture_output=True, text=True)
    print("     " + r.stdout.strip().replace("\n", "\n     "))
    filtered = [json.loads(l) for l in
                open(os.path.join(SESSION, "sanitized", "transactions.jsonl"))]
    check("SBC prefix emits only the two SBC transactions", len(filtered) == 2,
          "got %d" % len(filtered))
    check("SBC save body remains byte-exact after scoped parse",
          len(filtered) == 2 and
          base64.b64decode(filtered[1]["request"]["body_b64"]) == body)

    srv.shutdown()
    print()
    print("all checks passed" if not fails else "%d FAILED: %s" % (len(fails), fails))
    return 1 if fails else 0


if __name__ == "__main__":
    sys.exit(main())