OpenFUT/scripts/openfut-utas-observe.py

#!/usr/bin/env python3
"""Record real UTAS traffic between FIFA 17 and the Python oracle.

    openfut-utas-observe.py record  --listen 0.0.0.0:8199 --upstream 127.0.0.1:8099 \
                                    --session captures/utas/session-001
    openfut-utas-observe.py parse   --session captures/utas/session-001
    openfut-utas-observe.py snapshot --upstream 127.0.0.1:8099 --out before/state-manifest.json

WHY THIS EXISTS
---------------
Porting UTAS needs the exact bytes of real requests AND real responses. The
oracle's own log truncates bodies at ~200 chars, and raising that cap would mean
editing the behavioural specification to make it easier to copy -- which is
exactly backwards. A proxy gets the same evidence and leaves the oracle alone.

Standalone on purpose: NOT built into the Python container. The same tool can
later sit in front of a Rust UTAS host, so the identical captured request can be
replayed against both and diffed.

THE DESIGN RULE: TEE, DO NOT REBUILD
------------------------------------
UTAS is plaintext HTTP/1.1 served by `ThreadingHTTPServer`, so keep-alive,
pipelining and chunked transfer are all possible. A proxy that parses a request
and re-emits it can corrupt the traffic it exists to observe -- and a subtle
corruption here would look like a UTAS bug, sending the investigation in exactly
the wrong direction.

So this copies bytes VERBATIM in both directions and writes a second copy to
disk. Transactions are reconstructed later, offline, from that copy. A bug in
the parser therefore spoils the record and never the session. This is the same
reasoning that keeps the oracle unmodified: do not let the measuring apparatus
change what is being measured.

TWO LAYERS, LIKE THE BLAZE CAPTURES
-----------------------------------
    raw/utas.ofcap        exact bytes, mode 0600, gitignored
            |
        sanitizer
            |
    sanitized/transactions.jsonl    repository-safe fixtures

Raw captures are never normalised. Normalisation happens only when producing the
committed fixture, and preserves exact bodies first, sanitises second.
"""
import argparse
import base64
import hashlib
import json
import os
import re
import socket
import sys
import threading
import time

MAGIC = "OFCAP1"
C2S = "c2s"
S2C = "s2c"


# ─────────────────────────────────────────────────────────── capture file ────
class Capture:
    """Append-only JSONL of raw byte chunks. One line per read().

    JSONL with base64 rather than a packed binary format: this is a research
    artefact that people will grep and eyeball, and the size cost is irrelevant
    next to being able to inspect it without a decoder. Base64 is an encoding,
    not a normalisation -- the bytes round-trip exactly.
    """

    def __init__(self, path):
        self.path = path
        os.makedirs(os.path.dirname(path), exist_ok=True)
        # 0600 before anything is written: captures carry session tokens.
        fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
        self.f = os.fdopen(fd, "w")
        self.lock = threading.Lock()
        self._write({"magic": MAGIC, "started_unix": time.time()})

    def _write(self, obj):
        with self.lock:
            self.f.write(json.dumps(obj, sort_keys=True) + "\n")
            self.f.flush()

    def chunk(self, conn_id, direction, data, t0):
        self._write({
            "conn": conn_id,
            "dir": direction,
            "t": round(time.monotonic() - t0, 6),
            "unix": time.time(),
            "len": len(data),
            "b64": base64.b64encode(data).decode(),
        })

    def event(self, conn_id, kind, detail, t0):
        self._write({
            "conn": conn_id,
            "event": kind,
            "detail": detail,
            "t": round(time.monotonic() - t0, 6),
            "unix": time.time(),
        })

    def close(self):
        with self.lock:
            self.f.close()


# ──────────────────────────────────────────────────────────────── record ────
def pump(src, dst, cap, conn_id, direction, t0, done):
    """Copy src->dst verbatim, teeing every chunk into the capture.

    Forward FIRST, record second: the client must never wait on our disk.
    """
    try:
        while True:
            data = src.recv(65536)
            if not data:
                break
            try:
                dst.sendall(data)
            except OSError as e:
                cap.event(conn_id, "forward_failed", "%s %s" % (direction, e), t0)
                break
            cap.chunk(conn_id, direction, data, t0)
    except OSError as e:
        cap.event(conn_id, "read_failed", "%s %s" % (direction, e), t0)
    finally:
        # Half-close so the peer sees EOF in the same direction the original
        # did. Connection semantics are part of what is being captured.
        try:
            dst.shutdown(socket.SHUT_WR)
        except OSError:
            pass
        cap.event(conn_id, "eof", direction, t0)
        done.set()


def handle(client, addr, upstream_addr, cap, conn_id, t0):
    cap.event(conn_id, "open", "%s:%d" % addr, t0)
    try:
        up = socket.create_connection(upstream_addr, timeout=30)
    except OSError as e:
        cap.event(conn_id, "upstream_failed", str(e), t0)
        client.close()
        return
    client.settimeout(None)
    up.settimeout(None)

    a, b = threading.Event(), threading.Event()
    t1 = threading.Thread(target=pump, args=(client, up, cap, conn_id, C2S, t0, a), daemon=True)
    t2 = threading.Thread(target=pump, args=(up, client, cap, conn_id, S2C, t0, b), daemon=True)
    t1.start(); t2.start()
    t1.join(); t2.join()
    for s in (client, up):
        try:
            s.close()
        except OSError:
            pass
    cap.event(conn_id, "close", "", t0)


def cmd_record(args):
    lh, lp = args.listen.rsplit(":", 1)
    uh, up = args.upstream.rsplit(":", 1)
    session = os.path.abspath(args.session)
    cap = Capture(os.path.join(session, "raw", "utas.ofcap"))

    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((lh, int(lp)))
    srv.listen(64)
    t0 = time.monotonic()
    print("recording %s -> %s:%s into %s" % (args.listen, uh, up, cap.path), flush=True)
    print("  raw capture is mode 0600 and must stay out of git", flush=True)

    n = 0
    try:
        while True:
            client, addr = srv.accept()
            n += 1
            threading.Thread(
                target=handle,
                args=(client, addr, (uh, int(up)), cap, n, t0),
                daemon=True,
            ).start()
    except KeyboardInterrupt:
        print("\nstopped after %d connection(s)" % n, flush=True)
    finally:
        cap.close()


# ─────────────────────────────────────────────────────────────── parsing ────
def read_capture(path):
    conns = {}
    with open(path) as f:
        for line in f:
            r = json.loads(line)
            if "magic" in r:
                continue
            c = conns.setdefault(r["conn"], {"c2s": bytearray(), "s2c": bytearray(),
                                             "events": [], "marks": {"c2s": [], "s2c": []}})
            if "b64" in r:
                data = base64.b64decode(r["b64"])
                c["marks"][r["dir"]].append((len(c[r["dir"]]), r["t"], r["unix"]))
                c[r["dir"]].extend(data)
            else:
                c["events"].append(r)
    return conns


def split_messages(buf, is_response):
    """Split an HTTP/1.1 byte stream into messages. Returns list of (head, body, end).

    Framing follows RFC 7230: Transfer-Encoding: chunked wins over
    Content-Length. A stream that cannot be framed stops the walk rather than
    guessing -- a wrong split would silently mis-attribute bodies.
    """
    out, i = [], 0
    while True:
        sep = buf.find(b"\r\n\r\n", i)
        if sep < 0:
            break
        head = bytes(buf[i:sep])
        hs = head.decode("latin1")
        start = sep + 4
        te = re.search(r"(?im)^transfer-encoding:\s*(.+)$", hs)
        cl = re.search(r"(?im)^content-length:\s*(\d+)\s*$", hs)
        if te and "chunked" in te.group(1).lower():
            j, body = start, bytearray()
            while True:
                nl = buf.find(b"\r\n", j)
                if nl < 0:
                    return out
                try:
                    size = int(bytes(buf[j:nl]).split(b";")[0], 16)
                except ValueError:
                    return out
                j = nl + 2
                if size == 0:
                    j = buf.find(b"\r\n", j)
                    j = (j + 2) if j >= 0 else len(buf)
                    break
                body.extend(buf[j:j + size])
                j += size + 2
            out.append((head, bytes(body), j)); i = j
        elif cl:
            n = int(cl.group(1))
            end = start + n
            if end > len(buf):
                break
            out.append((head, bytes(buf[start:end]), end)); i = end
        else:
            status_no_body = is_response and re.match(r"HTTP/\d\.\d (1\d\d|204|304)", hs)
            if is_response and not status_no_body:
                # No framing header: the body runs to end-of-stream.
                out.append((head, bytes(buf[start:]), len(buf))); i = len(buf)
                break
            out.append((head, b"", start)); i = start
    return out


def time_at(marks, offset):
    """Wall/relative time of the chunk containing byte `offset`."""
    last = (0.0, 0.0)
    for pos, t, unix in marks:
        if pos > offset:
            break
        last = (t, unix)
    return last


def cmd_parse(args):
    session = os.path.abspath(args.session)
    raw = os.path.join(session, "raw", "utas.ofcap")
    conns = read_capture(raw)
    out_dir = os.path.join(session, "sanitized")
    os.makedirs(out_dir, exist_ok=True)
    out_path = os.path.join(out_dir, "transactions.jsonl")

    total, skipped, filtered = 0, 0, 0
    with open(out_path, "w") as out:
        for conn_id in sorted(conns):
            c = conns[conn_id]
            reqs = split_messages(c["c2s"], is_response=False)
            resps = split_messages(c["s2c"], is_response=True)
            if len(reqs) != len(resps):
                # Reported, never silently trimmed: an unmatched count means the
                # framing is not understood, and a quiet truncation would look
                # like the client simply asked for less.
                print("  conn %d: %d requests but %d responses -- pairing the common prefix"
                      % (conn_id, len(reqs), len(resps)), file=sys.stderr)
                skipped += abs(len(reqs) - len(resps))
            keepalive = None
            for seq, ((rh, rb, rend), (sh, sb, send)) in enumerate(zip(reqs, resps), 1):
                rl = rh.split(b"\r\n")[0].decode("latin1")
                sl = sh.split(b"\r\n")[0].decode("latin1")
                m = re.match(r"(\S+)\s+(\S+)\s+(HTTP/\d\.\d)", rl)
                method, target, ver = (m.group(1), m.group(2), m.group(3)) if m else ("?", rl, "?")
                path, _, query = target.partition("?")
                if args.path_prefix and not path.startswith(args.path_prefix):
                    filtered += 1
                    continue
                st = re.match(r"HTTP/\d\.\d\s+(\d+)", sl)
                req_t, req_unix = time_at(c["marks"]["c2s"], rend - 1)
                res_t, res_unix = time_at(c["marks"]["s2c"], max(send - 1, 0))
                keepalive = len(reqs) > 1
                rec = {
                    "conn": conn_id,
                    "seq": seq,
                    "t_request": req_t,
                    "t_response": res_t,
                    "elapsed_ms": round((res_t - req_t) * 1000, 3),
                    "unix_request": req_unix,
                    "request": {
                        "method": method, "path": path, "query": query, "version": ver,
                        "headers": header_list(rh),
                        "body_b64": base64.b64encode(rb).decode(),
                        "body_len": len(rb),
                    },
                    "response": {
                        "status": int(st.group(1)) if st else None,
                        "status_line": sl,
                        "headers": header_list(sh),
                        "body_b64": base64.b64encode(sb).decode(),
                        "body_len": len(sb),
                    },
                    "connection": {
                        "requests_on_this_connection": len(reqs),
                        "keepalive_observed": keepalive,
                    },
                }
                rec = sanitize(rec)
                if getattr(args, "max_body", 0):
                    for side in ("request", "response"):
                        raw = base64.b64decode(rec[side]["body_b64"])
                        if len(raw) > args.max_body:
                            rec[side]["body_b64"] = base64.b64encode(
                                raw[: args.max_body]
                            ).decode()
                            rec[side]["body_truncated"] = True
                            rec[side]["body_full_len"] = len(raw)
                            rec[side]["body_full_sha256"] = hashlib.sha256(raw).hexdigest()
                out.write(json.dumps(rec, sort_keys=True) + "\n")
                total += 1

    os.chmod(out_path, 0o644)
    detail = []
    if skipped:
        detail.append("%d unpaired messages reported above" % skipped)
    if filtered:
        detail.append("%d transactions excluded by path prefix" % filtered)
    print("wrote %s (%d transactions across %d connections%s)"
          % (out_path, total, len(conns),
             "; " + "; ".join(detail) if detail else ""))
    return 0


def header_list(head):
    """Headers in RECEIVED ORDER, as a list of pairs. Order can be protocol-
    relevant and a dict would silently discard duplicates."""
    lines = head.decode("latin1").split("\r\n")[1:]
    out = []
    for l in lines:
        if not l:
            continue
        k, _, v = l.partition(":")
        out.append([k, v.strip()])
    return out


SECRET_HEADERS = {"authorization", "cookie", "set-cookie", "x-ut-sid", "easw-session-data-nucleus-id"}
# Substring-matched against lowercased JSON keys. Extended after auditing the
# first real corpus: session tokens were being redacted correctly, but the
# client also sends hardware and device identifiers in `POST /ut/auth`, and a
# committed fixture is a published fixture. Audit the output before publishing,
# every time -- "the sanitiser handles it" is a belief until it is checked.
SECRET_BODY_KEYS = (
    "sid", "token", "password", "answer", "secret",
    "macaddress", "deviceid", "authcode",
)


def sanitize(rec):
    """Repository-safe form. Bodies are preserved EXACTLY; only credentials go.

    THREE input surfaces carry identifiers, not two. The first version handled
    headers and JSON bodies and was still about to publish a device id, because
    `GET /ut/game/fifa17/phishing/trusteddevice?deviceId=...` puts it in the
    QUERY STRING. Caught by auditing the output rather than by trusting the
    sanitiser -- which is the only reason it is a comment and not a leak.

    Exact bodies first, sanitise second: the corpus is worthless if response
    payloads are reshaped, so nothing here touches structure -- it replaces
    known-secret header values and known-secret JSON keys, and records that it
    did so.
    """
    redacted = []

    # Query string: same key rules as bodies.
    q = rec["request"].get("query") or ""
    if q:
        parts, hit = [], False
        for kv in q.split("&"):
            k, eq, v = kv.partition("=")
            if eq and any(x in k.lower() for x in SECRET_BODY_KEYS):
                parts.append(k + "=<REDACTED>")
                redacted.append("request.query." + k)
                hit = True
            else:
                parts.append(kv)
        if hit:
            rec["request"]["query"] = "&".join(parts)

    for side in ("request", "response"):
        hs = rec[side]["headers"]
        for pair in hs:
            if pair[0].lower() in SECRET_HEADERS:
                redacted.append("%s.%s" % (side, pair[0]))
                pair[1] = "<REDACTED>"
        body = base64.b64decode(rec[side]["body_b64"])
        new, hit = redact_json(body)
        if hit:
            redacted.extend("%s.body.%s" % (side, h) for h in hit)
            rec[side]["body_b64"] = base64.b64encode(new).decode()
            rec[side]["body_len_original"] = len(body)
    if redacted:
        rec["redacted"] = sorted(set(redacted))
    return rec


def redact_json(body):
    if not body[:1] in (b"{", b"["):
        return body, []
    try:
        doc = json.loads(body)
    except Exception:
        return body, []
    hits = []

    def walk(o):
        if isinstance(o, dict):
            for k in list(o):
                if any(s in k.lower() for s in SECRET_BODY_KEYS) and isinstance(o[k], str):
                    o[k] = "<REDACTED>"
                    hits.append(k)
                else:
                    walk(o[k])
        elif isinstance(o, list):
            for v in o:
                walk(v)

    walk(doc)
    if not hits:
        return body, []
    return json.dumps(doc, sort_keys=True).encode(), hits


# ────────────────────────────────────────────────────────────── snapshot ────
# Read-only GETs only. `/openfut/account/sync` is deliberately NOT here: it is a
# POST that selects and writes an account, so using it as a snapshot would
# mutate the state the snapshot exists to observe.
#
# All six were confirmed to answer 200 without a session token.
SNAPSHOT_ROUTES = [
    ("credits", "/ut/game/fifa17/user/credits"),
    ("userMassInfo", "/ut/game/fifa17/userMassInfo"),
    ("unassigned", "/ut/game/fifa17/purchased/items"),
    ("activeSquad", "/ut/game/fifa17/squad/active"),
    ("tradePile", "/ut/game/fifa17/tradePile"),
    ("accountInfo", "/ut/game/fifa17/user/accountinfo"),
]


def http_get(host, port, path):
    s = socket.create_connection((host, int(port)), timeout=10)
    s.sendall(("GET %s HTTP/1.1\r\nHost: %s:%s\r\nConnection: close\r\n\r\n"
               % (path, host, port)).encode())
    buf = b""
    while True:
        d = s.recv(65536)
        if not d:
            break
        buf += d
    s.close()
    head, _, body = buf.partition(b"\r\n\r\n")
    line = head.split(b"\r\n")[0].decode("latin1")
    status = int(line.split()[1]) if len(line.split()) > 1 else None
    return status, body


def cmd_snapshot(args):
    """A state manifest for pairing before/after around a capture session.

    A summary plus a hash, not a database dump. The named fields are what a
    human reads; the sha256 of each full body is the safety net, because a
    summary can only report changes in fields somebody thought to list. If the
    hash moves and no field does, the summary is incomplete -- and that is
    itself a finding rather than a silent miss.

    Raw bodies are written alongside at mode 0600 so a real diff is possible
    later without re-running the session.
    """
    import hashlib
    uh, up = args.upstream.rsplit(":", 1)
    out_path = os.path.abspath(args.out)
    raw_dir = os.path.join(os.path.dirname(out_path), "bodies")
    os.makedirs(raw_dir, exist_ok=True)

    man = {"unix": time.time(), "upstream": args.upstream, "routes": {}}
    for name, path in SNAPSHOT_ROUTES:
        try:
            status, body = http_get(uh, up, path)
        except Exception as e:
            # Recorded as an error, never omitted: a missing section must not
            # be mistakable for an empty one.
            man["routes"][name] = {"error": str(e)}
            continue
        bp = os.path.join(raw_dir, name + ".json")
        fd = os.open(bp, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "wb") as f:
            f.write(body)
        man["routes"][name] = {
            "status": status,
            "body_len": len(body),
            "sha256": hashlib.sha256(body).hexdigest(),
            "fields": extract(name, body),
        }

    os.makedirs(os.path.dirname(out_path), exist_ok=True)
    with open(out_path, "w") as f:
        json.dump(man, f, indent=2, sort_keys=True)
        f.write("\n")
    print("wrote %s" % out_path)
    for k, v in man["routes"].items():
        print("  %-13s %s" % (k, v.get("fields", v)))
    return 0


def extract(name, body):
    """Named fields per route. Unknown shapes report what they can rather than
    raising -- the hash still covers everything."""
    try:
        d = json.loads(body)
    except Exception:
        return {"unparsed_len": len(body)}
    f = {}
    if name == "credits":
        f["credits"] = d.get("credits")
    elif name == "userMassInfo":
        ui = d.get("userInfo", {})
        for k in ("personaId", "clubName", "clubAbbr", "trophies"):
            if k in ui:
                f[k] = ui[k]
        for k, v in d.items():
            if isinstance(v, list):
                f[k + ".count"] = len(v)
    elif name == "unassigned":
        f["itemData.count"] = len(d.get("itemData", []))
    elif name == "activeSquad":
        f["id"] = d.get("id")
        f["formation"] = d.get("formation")
        players = d.get("players", [])
        f["players.count"] = len(players)
        # Slot -> item id, so a two-player swap is visible in the diff.
        f["slots"] = {
            str(p.get("index", i)): (p.get("itemData") or {}).get("id")
            for i, p in enumerate(players)
        }
    elif name == "tradePile":
        f["auctionInfo.count"] = len(d.get("auctionInfo", []))
    elif name == "accountInfo":
        f["keys"] = sorted(d.keys()) if isinstance(d, dict) else None
    return f


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)

    r = sub.add_parser("record")
    r.add_argument("--listen", required=True)
    r.add_argument("--upstream", required=True)
    r.add_argument("--session", required=True)
    r.set_defaults(fn=cmd_record)

    p = sub.add_parser("parse")
    p.add_argument("--session", required=True)
    # Repository-safe size. The raw .ofcap keeps every byte privately; a
    # committed fixture that repeats an identical 1.1MB response 25 times is
    # 13MB of git history proving something its own sha256 already proves.
    # Bodies over the limit are replaced by a bounded head sample, and the
    # omission is RECORDED per transaction so nobody mistakes a truncated
    # fixture for the whole response.
    p.add_argument("--max-body", type=int, default=0,
                   help="bytes; 0 = keep every body in full")
    p.add_argument("--path-prefix", default="",
                   help="emit only transactions whose request path starts with this prefix")
    p.set_defaults(fn=cmd_parse)

    s = sub.add_parser("snapshot")
    s.add_argument("--upstream", required=True)
    s.add_argument("--out", required=True)
    s.set_defaults(fn=cmd_snapshot)

    args = ap.parse_args()
    return args.fn(args) or 0


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