OpenFUT/openfut-protocol-blaze/fixtures/generate.py

#!/usr/bin/env python3
"""Generate protocol fixtures from the PROVEN Python FIFA 17 backend.

WHY THIS EXISTS
---------------
The Python backend in `fifa17-recon/tools/` is the behavioural oracle: it is the
implementation that actually walked a retail FIFA 17 client from Origin login to
an opened FUT pack. Nothing in Rust may claim parity with it by assertion; parity
has to be *replayable*.

The `captures/` directory that earlier findings cite is gitignored (`captures/`,
`*.bin`), so the original wire captures are not recoverable from Git. The next
best oracle is the Python code that produced the bytes the client accepted:
`heat2.encode_tdf` (Heat2/TDF codec) and `blaze_responder_v3b.fire2` (Fire2
framing). This script freezes their output into committed, machine-testable
vectors so the Rust port can be diffed byte-for-byte, forever, without a running
FIFA client.

WHAT IS AND IS NOT AN ORACLE HERE
---------------------------------
  * "live" vectors are built from the responder's real payload builders — the
    exact field trees FIFA 17 accepted in the working run. Highest confidence.
  * "synthetic" vectors exercise codec edge cases (varint boundaries, tag
    padding, empty string, nesting, every type byte). They prove the Rust codec
    matches the Python codec; they do NOT prove either matches EA, because the
    complex types (list/map/union/varlist/objtype/objid/float) are flagged
    UNVERIFIED in heat2.py's own docstring. Vectors carry that flag through so a
    future reader cannot mistake self-consistency for wire truth.

NO SECRETS. The FIFA 17 identity in these fixtures (persona 33068179 / "CAGE")
is the project's fixed synthetic offline identity, not a credential. Session keys
are generated from a seeded PRNG below so output is deterministic and carries no
entropy from any real session.

Usage:  python3 fixtures/generate.py        (writes tdf.jsonl + fire2.jsonl)
        python3 fixtures/generate.py --check (verify committed files are current)
"""
from __future__ import annotations

import json
import os
import random
import sys
from collections import OrderedDict

HERE = os.path.dirname(os.path.abspath(__file__))
TOOLS = os.path.normpath(os.path.join(HERE, "..", "..", "fifa17-recon", "tools"))
if not os.path.isdir(TOOLS):
    sys.exit("cannot find the Python oracle at %s" % TOOLS)
sys.path.insert(0, TOOLS)

# Read our own flags BEFORE clearing argv. The responders inspect sys.argv at
# import time (blaze_responder_v3b honours --selftest), so it has to be emptied
# for them — but doing that first would silently swallow --check and turn the
# staleness guard into a no-op that always rewrites and always passes.
CHECK_ONLY = "--check" in sys.argv[1:]
sys.argv = [sys.argv[0]]
import heat2  # noqa: E402
import blaze_responder_v3b as B  # noqa: E402

TYPE_NAME = {
    heat2.INT: "int", heat2.STRING: "string", heat2.BLOB: "blob",
    heat2.STRUCT: "struct", heat2.LIST: "list", heat2.MAP: "map",
    heat2.UNION: "union", heat2.VARLIST: "varlist", heat2.OBJTYPE: "objtype",
    heat2.OBJID: "objid", heat2.FLOAT: "float",
}


# ---------------------------------------------------------------- value model
#
# A TDF value is serialised to JSON as {"t": <type-name>, ...}. The shape per
# type mirrors heat2.py's Python representation exactly so the Rust side can
# rebuild an identical tree and re-encode it.

def val_to_json(typ: int, value):
    t = TYPE_NAME[typ]
    if typ == heat2.INT:
        return {"t": t, "v": int(value)}
    if typ == heat2.STRING:
        s = value.decode("utf-8") if isinstance(value, (bytes, bytearray)) else value
        return {"t": t, "v": s}
    if typ == heat2.BLOB:
        return {"t": t, "v": bytes(value).hex()}
    if typ == heat2.STRUCT:
        return {"t": t, "v": fields_to_json(value)}
    if typ == heat2.LIST:
        etype, items = value
        return {"t": t, "elem": TYPE_NAME[etype],
                "v": [val_to_json(etype, it) for it in items]}
    if typ == heat2.MAP:
        ktype, vtype, items = value
        return {"t": t, "key": TYPE_NAME[ktype], "val": TYPE_NAME[vtype],
                "v": [[val_to_json(ktype, k), val_to_json(vtype, v)]
                      for k, v in items]}
    if typ == heat2.UNION:
        key, member = value
        if member is None:
            return {"t": t, "key": key, "member": None}
        mtag, mtype, mval = member
        return {"t": t, "key": key,
                "member": {"tag": mtag, "value": val_to_json(mtype, mval)}}
    if typ == heat2.VARLIST:
        return {"t": t, "v": [int(n) for n in value]}
    if typ == heat2.OBJTYPE:
        return {"t": t, "v": [int(value[0]), int(value[1])]}
    if typ == heat2.OBJID:
        return {"t": t, "v": [int(value[0]), int(value[1]), int(value[2])]}
    if typ == heat2.FLOAT:
        return {"t": t, "v": float(value)}
    raise ValueError("unhandled TDF type 0x%02x" % typ)


def fields_to_json(fields):
    """Ordered [[tag, value], ...]. Order is informational: both encoders sort
    by packed tag, and the fixture's encoded_hex is what actually binds."""
    items = list(fields.items()) if isinstance(fields, dict) else list(fields)
    return [[tag, val_to_json(typ, val)] for tag, (typ, val) in items]


# ---------------------------------------------------------------- collection

TDF_VECTORS = []
FIRE2_VECTORS = []


def tdf_case(name, fields, *, origin, verified, note=""):
    """Encode with the oracle, prove it round-trips, and record the vector."""
    encoded = heat2.encode_tdf(fields)
    decoded = heat2.decode_tdf(encoded)
    reencoded = heat2.encode_tdf(decoded)
    if reencoded != encoded:
        raise AssertionError("oracle does not round-trip: %s" % name)
    TDF_VECTORS.append(OrderedDict((
        ("name", name),
        ("origin", origin),          # "live" | "synthetic"
        ("verified", verified),      # False => heat2.py marks the layout UNVERIFIED
        ("note", note),
        ("fields", fields_to_json(fields)),
        ("encoded_hex", encoded.hex()),
    )))


def fire2_case(name, *, component, command, msg_num, msg_type, payload,
               metadata=b"", user_index=0, options=0, origin, note=""):
    frame = B.fire2(component, command, msg_num, msg_type, payload,
                    metadata=metadata, user_index=user_index, options=options)
    hdr = B.parse_fire2_header(frame)
    assert hdr["component"] == component and hdr["command"] == command
    assert hdr["msg_num"] == msg_num and hdr["msg_type"] == msg_type
    assert hdr["user_index"] == user_index and hdr["options"] == options
    assert hdr["payload_len"] == len(payload)
    assert hdr["metadata_len"] == len(metadata)
    FIRE2_VECTORS.append(OrderedDict((
        ("name", name),
        ("origin", origin),
        ("note", note),
        ("component", component),
        ("command", command),
        ("msg_num", msg_num),
        ("msg_type", msg_type),
        ("user_index", user_index),
        ("options", options),
        ("metadata_hex", metadata.hex()),
        ("payload_hex", payload.hex()),
        ("frame_hex", frame.hex()),
    )))


# ---------------------------------------------------------------- live vectors
#
# These field trees come from the responder's own builders — the payloads the
# retail client accepted during the working end-to-end run.

# A fixed epoch (2025-08-11T00:00:00Z) so regeneration is reproducible and
# `--check` is meaningful.
FIXED_NOW = 1754870400


def collect_live():
    random.seed(0xF17A)          # deterministic forged session key
    sess = B.Session()
    now = FIXED_NOW

    builders = [
        ("preauth_response", lambda: B.preauth_response_fields()),
        ("ping_response", lambda: B.ping_response_fields()),
        ("qos_config", lambda: B.qos_config()),
        ("census_subscribe_response",
         lambda: B.subscribe_census_data_updates_response_fields()),
        ("persona_details", lambda: B.persona_details_fields(now)),
        ("user_login_info", lambda: B.user_login_info_fields(sess, now)),
        ("login_response", lambda: B.login_response_fields(sess)),
        ("account_info", lambda: B.account_info_fields(sess, now)),
        ("persona_info", lambda: B.persona_info_fields(now)),
        ("get_persona_response", lambda: B.get_persona_response_fields(now)),
        ("list_personas_response", lambda: B.list_personas_response_fields(now)),
        ("user_session_login_info",
         lambda: B.user_session_login_info_fields(sess, now)),
        ("network_qos_data", lambda: B.network_qos_data_fields()),
        ("user_session_extended_data",
         lambda: B.user_session_extended_data_fields()),
        ("user_session_extended_data_update",
         lambda: B.user_session_extended_data_update_fields()),
        ("user_identification", lambda: B.user_identification_fields(sess)),
        ("user_data", lambda: B.user_data_fields(sess)),
        ("post_auth_response", lambda: B.post_auth_response_fields(sess)),
        ("entitlements_response", lambda: B.entitlements_response_fields()),
        ("get_auth_token_response",
         lambda: B.get_auth_token_response_fields(sess)),
        ("user_settings_response", lambda: B.user_settings_response_fields()),
        ("get_lists_response", lambda: B.get_lists_response_fields()),
    ]
    for name, fn in builders:
        fields = fn()
        if not isinstance(fields, (dict, OrderedDict)):
            raise AssertionError("%s did not return a TDF struct" % name)
        tdf_case("live/" + name, fields, origin="live", verified=True,
                 note="payload builder from blaze_responder_v3b.py")

    for cfid in ("BlazeSDK", "UTAS", "FUT"):
        try:
            fields = B.fetch_config_response_fields(cfid)
        except Exception:
            continue
        tdf_case("live/fetch_config_%s" % cfid, fields, origin="live",
                 verified=True, note="fetchClientConfig CFID=%s" % cfid)

    # --- framing: the real login burst, exactly as the client received it.
    login_payload = heat2.encode_tdf(B.login_response_fields(sess))
    fire2_case("live/login_reply", component=B.COMP_AUTH, command=B.CMD_LOGIN,
               msg_num=7, msg_type=B.REPLY, payload=login_payload,
               origin="live", note="Authentication::login REPLY")

    for tag, frame in B.build_login_notifications(sess, now):
        hdr = B.parse_fire2_header(frame)
        body = frame[16 + hdr["metadata_len"]:]
        fire2_case("live/notify_%s" % str(tag).lower().replace(" ", "_"),
                   component=hdr["component"], command=hdr["command"],
                   msg_num=hdr["msg_num"], msg_type=hdr["msg_type"],
                   payload=body, user_index=hdr["user_index"],
                   options=hdr["options"], origin="live",
                   note="post-login server push")

    fire2_case("live/preauth_reply", component=B.COMP_UTIL,
               command=B.CMD_PREAUTH, msg_num=1, msg_type=B.REPLY,
               payload=heat2.encode_tdf(B.preauth_response_fields()),
               origin="live", note="Util::preAuth REPLY, the first RPC")
    fire2_case("live/ping_reply", component=B.COMP_UTIL, command=B.CMD_PING,
               msg_num=2, msg_type=B.REPLY,
               payload=heat2.encode_tdf(B.ping_response_fields()),
               origin="live", note="Util::ping REPLY")


# ------------------------------------------------------------ synthetic vectors

def collect_synthetic():
    I, S, BL, ST = heat2.INT, heat2.STRING, heat2.BLOB, heat2.STRUCT
    L, M, U = heat2.LIST, heat2.MAP, heat2.UNION
    VL, OT, OI, F = heat2.VARLIST, heat2.OBJTYPE, heat2.OBJID, heat2.FLOAT

    def case(name, fields, verified=True, note=""):
        tdf_case("synthetic/" + name, fields, origin="synthetic",
                 verified=verified, note=note)

    # Varint boundaries. The first byte carries only 6 data bits, so 0x3F/0x40
    # is the one-to-two byte edge; the rest are the 7-bit group edges.
    case("varint_boundaries", OrderedDict([
        ("V000", (I, 0)), ("V001", (I, 1)), ("V03F", (I, 0x3F)),
        ("V040", (I, 0x40)), ("V07F", (I, 0x7F)), ("V080", (I, 0x80)),
        ("V1FF", (I, 0x1FFF)), ("V200", (I, 0x2000)),
        ("VBIG", (I, 0xFFFFFFFF)), ("VMAX", (I, 0x7FFFFFFFFFFFFFFF)),
    ]), note="6-bit first group then 7-bit groups")

    case("varint_negative", OrderedDict([("NEG1", (I, -1)), ("NEGB", (I, -300))]),
         verified=False,
         note="sign bit 0x40 is UNVERIFIED in heat2.py; never seen on the wire")

    # Tag packing: 6-bit chars, space padded to 4. Short tags and the full
    # 4-char case must both survive a round trip.
    case("tag_padding", OrderedDict([
        ("A", (I, 1)), ("AB", (I, 2)), ("ABC", (I, 3)), ("ABCD", (I, 4)),
        ("ENV", (S, "prod")), ("Z", (I, 26)),
    ]), note="trailing-space padding must strip on decode")

    case("tag_ordering", OrderedDict([
        ("ZZZZ", (I, 3)), ("AAAA", (I, 1)), ("MMMM", (I, 2)),
    ]), note="members must serialise in ascending packed-tag order")

    case("strings", OrderedDict([
        ("EMPT", (S, "")), ("ONEC", (S, "x")),
        ("LONG", (S, "x" * 200)),
        ("UTF8", (S, "Grün-Weiß Ünïcode")),
    ]), note="length varint INCLUDES the NUL terminator")

    case("blobs", OrderedDict([
        ("BEMP", (BL, b"")), ("BONE", (BL, b"\x00")),
        ("BFUL", (BL, bytes(range(256)))),
    ]), note="blob length varint EXCLUDES any terminator")

    case("nested_structs", OrderedDict([
        ("OUTR", (ST, OrderedDict([
            ("INNR", (ST, OrderedDict([
                ("DEEP", (ST, OrderedDict([("LEAF", (I, 42))]))),
                ("SIBL", (S, "sibling")),
            ]))),
            ("AFTR", (I, 7)),
        ]))),
        ("TAIL", (I, 9)),
    ]), note="nested structs are 0x00 terminated; the top level is not")

    case("empty_struct", OrderedDict([("MTST", (ST, OrderedDict()))]),
         note="an empty nested struct is a bare 0x00")

    case("lists", OrderedDict([
        ("LINT", (L, (I, [1, 2, 3, 0x40, 0x2000]))),
        ("LSTR", (L, (S, ["a", "bb", ""]))),
        ("LEMP", (L, (I, []))),
        ("LST2", (L, (ST, [OrderedDict([("A", (I, 1))]),
                           OrderedDict([("B", (I, 2))])]))),
    ]), verified=False, note="list layout UNVERIFIED (absent from the capture)")

    # A tag is exactly 4 packed 6-bit chars. Longer labels are TRUNCATED, not
    # rejected, so "LSTR2" and "LSTR" are the same wire tag and one silently
    # overwrites the other in a struct. Byte-level round-trip still holds.
    case("tag_truncation", OrderedDict([("LONGTAG", (I, 1))]),
         note="labels over 4 chars truncate silently — a collision hazard")

    case("maps", OrderedDict([
        ("MSI", (M, (S, I, [("one", 1), ("two", 2)]))),
        ("MII", (M, (I, I, [(1, 10), (2, 20)]))),
        ("MEMP", (M, (S, I, []))),
    ]), verified=False, note="map layout UNVERIFIED")

    case("unions", OrderedDict([
        ("USET", (U, (0, ("VALU", I, 1234)))),
        ("UUNS", (U, (heat2.UNION_UNSET, None))),
    ]), verified=False, note="union layout UNVERIFIED")

    case("scalar_misc", OrderedDict([
        ("VLST", (VL, [1, 0x40, 0x2000])),
        ("OTYP", (OT, (4, 1))),
        ("OIDD", (OI, (4, 1, 33068179))),
        ("FLTP", (F, 1.5)),
        ("FLTN", (F, -0.25)),
        ("FLTZ", (F, 0.0)),
    ]), verified=False,
        note="varintlist/objtype/objid/float layouts UNVERIFIED; float is >f")

    # Framing edge cases.
    fire2_case("synthetic/empty_payload", component=0x0009, command=0x0002,
               msg_num=0, msg_type=B.MESSAGE, payload=b"", origin="synthetic",
               note="a zero-length body is legal (ping)")
    fire2_case("synthetic/msgnum_24bit", component=0x0001, command=0x000A,
               msg_num=0xFFFFFF, msg_type=B.REPLY, payload=b"\x01\x02",
               origin="synthetic", note="msgNum is a 24-bit field at [10:13]")
    fire2_case("synthetic/user_index_max", component=0x7802, command=0x0008,
               msg_num=1, msg_type=B.NOTIFICATION, payload=b"\xaa",
               user_index=0x1F, origin="synthetic",
               note="byte[13] = (msgType<<5) | (userIndex & 0x1F)")
    fire2_case("synthetic/with_metadata", component=0x0009, command=0x0007,
               msg_num=5, msg_type=B.MESSAGE, payload=b"payload",
               metadata=b"\xde\xad\xbe\xef", origin="synthetic",
               note="wire = header(16) || metadata || payload")
    fire2_case("synthetic/options_byte", component=0x0009, command=0x0007,
               msg_num=5, msg_type=B.MESSAGE, payload=b"x", options=0x42,
               origin="synthetic", note="byte[14] is the options byte")
    for mt, label in ((B.MESSAGE, "message"), (B.REPLY, "reply"),
                      (B.NOTIFICATION, "notification"),
                      (B.ERROR_REPLY, "error_reply"), (B.PING, "ping"),
                      (B.PING_REPLY, "ping_reply")):
        fire2_case("synthetic/msgtype_%s" % label, component=0x0009,
                   command=0x0001, msg_num=3, msg_type=mt, payload=b"\x00",
                   origin="synthetic", note="msgType %d in the top 3 bits of byte[13]" % mt)
    fire2_case("synthetic/large_payload", component=0x0001, command=0x000A,
               msg_num=1, msg_type=B.REPLY, payload=bytes(70000 * [0x5A]),
               origin="synthetic",
               note="payloadLen is a full u32 — Fire2 has no jumbo-flag escape")


# ---------------------------------------------------------------- output

def write(path, vectors):
    body = "".join(json.dumps(v, separators=(",", ":")) + "\n" for v in vectors)
    if CHECK_ONLY:
        if not os.path.exists(path):
            sys.exit("MISSING: %s has never been generated" % path)
        with open(path, "r", encoding="utf-8") as fh:
            if fh.read() != body:
                sys.exit("STALE: %s does not match the oracle; re-run without "
                         "--check" % path)
        print("current: %s (%d vectors)" % (os.path.basename(path), len(vectors)))
        return
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(body)
    print("wrote %s (%d vectors)" % (os.path.basename(path), len(vectors)))


def frozen_clock():
    """Pin the oracle's clock for the duration of sampling.

    Some builders take a `now` argument; others (`ping_response_fields`,
    `login_response_fields`) call `int(time.time())` internally and ignore it.
    Left alone, every regeneration produces different bytes for those vectors,
    which makes `--check` permanently red and the committed fixtures
    unreproducible.

    This patches the clock the oracle reads, not the oracle's logic — the
    encoding path under test is untouched. Restored in a finally block so an
    imported responder is never left with a frozen clock.
    """
    import time as _time

    original = _time.time
    _time.time = lambda: float(FIXED_NOW)
    return original, _time


def main():
    original_time, time_mod = frozen_clock()
    try:
        collect_live()
    finally:
        time_mod.time = original_time
    collect_synthetic()
    write(os.path.join(HERE, "tdf.jsonl"), TDF_VECTORS)
    write(os.path.join(HERE, "fire2.jsonl"), FIRE2_VECTORS)
    live = sum(1 for v in TDF_VECTORS if v["origin"] == "live")
    print("TDF: %d vectors (%d live, %d synthetic); Fire2: %d vectors"
          % (len(TDF_VECTORS), live, len(TDF_VECTORS) - live, len(FIRE2_VECTORS)))


if __name__ == "__main__":
    main()