#!/usr/bin/env python3 """Tests for the FIFA 17 verified-patched-client capability negotiation. Pins the additive empty-My-Packs switch built on top of the P2 65534 sentinel: the sentinel is suppressed for a session ONLY when the launcher has registered a verified resolver capability (v1) for the CURRENT FIFA process, bound to the peer IP, and the decision is frozen at the first /store/purchasegroup. Every failure / unknown / late / cross-process case is fail-closed to the active sentinel. Covers matrix A-J from docs/plans/FIFA17_PATCHED_CLIENT_CAPABILITY.md (Task 12/15): A no-capability, zero packs -> sentinel B verified v1, zero packs -> clean (no 65534) C real unopened pack + no capability -> genuine pack, no sentinel D real unopened pack + capability -> genuine pack, no sentinel E unsupported version -> endpoint 400 AND mode sentinel F late capability after sentinel freeze -> stays sentinel G capability disappears after clean freeze-> stays clean (immutable) H two concurrent IPs (A verified, B none) -> A clean, B sentinel (no global leak) I new session via reset clears capability -> fresh unpatched process -> sentinel J autopatch mismatch => never registers -> sentinel Standalone unit test in the project style: `python3 test_capability_negotiation.py`. """ import importlib import json import os import sys import tempfile TOOLS = os.path.dirname(os.path.abspath(__file__)) if TOOLS not in sys.path: sys.path.insert(0, TOOLS) SENTINEL_ID = 65534 REAL_PACK_ID = 1 class _H: """Minimal request-handler stand-in: peer IP + optional JSON body.""" def __init__(self, ip, body=None): self.client_address = (ip, 54321) self._body = json.dumps(body).encode("utf-8") if body is not None else b"" def _ids(catalog): return [p["id"] for p in catalog["purchase"]] def main(): with tempfile.TemporaryDirectory() as state: os.environ["FUT_ACCOUNT_PATH"] = os.path.join(state, "active_account.json") os.environ["FUT_PROFILE_ROOT"] = os.path.join(state, "accounts") os.environ.pop("FUT_PROFILE", None) import fut_account import fut_store import fut_accounts import utas_server importlib.reload(fut_account) importlib.reload(fut_store) importlib.reload(fut_accounts) importlib.reload(utas_server) fut_accounts.activate({"personaId": 111001, "personaName": "TEST_A"}) # ---- deterministic pack topology helpers ------------------------------- _orig_visible = utas_server.visible_unopened_packs def set_zero_packs(): utas_server.visible_unopened_packs = lambda: [] def set_real_pack(): utas_server.visible_unopened_packs = lambda: [REAL_PACK_ID] def reset_state(): """Fresh capability store between cases (no cross-case leakage).""" utas_server._FIFA17_STORE.clear() def register(ip, version, persona=42, pid=4242): return utas_server.fifa17_capability_route(_H(ip, { "capability": "empty_mypacks_resolver", "version": version, "personaId": persona, "fifaPid": pid, })) def store(ip): status, cat = utas_server.store_catalog(_H(ip)) assert status == 200, status return _ids(cat) try: # ---- A. no capability, zero packs -> sentinel ---------------------- reset_state() set_zero_packs() ip = "10.0.0.1" utas_server.fifa17_reset_session(ip) ids = store(ip) assert SENTINEL_ID in ids, ids assert utas_server._FIFA17_STORE[ip]["mode"] == utas_server.FIFA17_MODE_SENTINEL print("A no-capability zero-packs -> sentinel: OK") # ---- B. verified v1, zero packs -> clean --------------------------- reset_state() set_zero_packs() ip = "10.0.0.2" utas_server.fifa17_reset_session(ip) st, body = register(ip, 1) assert st == 200 and body == {"status": "OK"}, (st, body) ids = store(ip) assert SENTINEL_ID not in ids, ids assert utas_server._FIFA17_STORE[ip]["mode"] == utas_server.FIFA17_MODE_CLEAN print("B verified-v1 zero-packs -> clean: OK") # ---- C. real pack + no capability -> genuine, no sentinel ---------- reset_state() set_real_pack() ip = "10.0.0.3" utas_server.fifa17_reset_session(ip) ids = store(ip) assert SENTINEL_ID not in ids, ids assert REAL_PACK_ID in ids, ids print("C real-pack no-capability -> genuine, no sentinel: OK") # ---- D. real pack + capability -> genuine, no sentinel ------------- reset_state() set_real_pack() ip = "10.0.0.4" utas_server.fifa17_reset_session(ip) register(ip, 1) ids = store(ip) assert SENTINEL_ID not in ids, ids assert REAL_PACK_ID in ids, ids print("D real-pack capability -> genuine, no sentinel: OK") # ---- E. unsupported version -> 400 AND mode sentinel --------------- reset_state() set_zero_packs() ip = "10.0.0.5" utas_server.fifa17_reset_session(ip) for bad in (2, 99): st, body = register(ip, bad) assert st == 400 and "error" in body, (bad, st, body) # nothing recorded -> resolver stays None assert utas_server._FIFA17_STORE[ip]["resolver"] is None ids = store(ip) assert SENTINEL_ID in ids, ids assert utas_server._FIFA17_STORE[ip]["mode"] == utas_server.FIFA17_MODE_SENTINEL # a bad capability string with the right version is also rejected st, body = utas_server.fifa17_capability_route(_H("10.0.0.55", { "capability": "something_else", "version": 1})) assert st == 400, (st, body) print("E unsupported version -> 400 + sentinel: OK") # ---- F. late capability after sentinel freeze -> stays sentinel ---- reset_state() set_zero_packs() ip = "10.0.0.6" utas_server.fifa17_reset_session(ip) ids = store(ip) # freeze: sentinel assert SENTINEL_ID in ids, ids register(ip, 1) # arrives late; ignored for session ids = store(ip) assert SENTINEL_ID in ids, "late capability must not flip a frozen sentinel" assert utas_server._FIFA17_STORE[ip]["mode"] == utas_server.FIFA17_MODE_SENTINEL print("F late capability after sentinel freeze -> sentinel: OK") # ---- G. capability disappears after clean freeze -> stays clean ---- reset_state() set_zero_packs() ip = "10.0.0.7" utas_server.fifa17_reset_session(ip) register(ip, 1) ids = store(ip) # freeze: clean assert SENTINEL_ID not in ids, ids utas_server._FIFA17_STORE[ip]["resolver"] = None # capability vanishes ids = store(ip) assert SENTINEL_ID not in ids, "frozen clean mode must be immutable" assert utas_server._FIFA17_STORE[ip]["mode"] == utas_server.FIFA17_MODE_CLEAN print("G capability disappears after clean freeze -> clean: OK") # ---- H. two concurrent IPs -> no global leak ----------------------- reset_state() set_zero_packs() ip_a, ip_b = "10.0.1.1", "10.0.1.2" utas_server.fifa17_reset_session(ip_a) utas_server.fifa17_reset_session(ip_b) register(ip_a, 1) # A verified, B never registers ids_a = store(ip_a) ids_b = store(ip_b) assert SENTINEL_ID not in ids_a, ids_a assert SENTINEL_ID in ids_b, ids_b assert utas_server._FIFA17_STORE[ip_a]["mode"] == utas_server.FIFA17_MODE_CLEAN assert utas_server._FIFA17_STORE[ip_b]["mode"] == utas_server.FIFA17_MODE_SENTINEL print("H concurrent IPs (A clean, B sentinel) -> no global leak: OK") # ---- I. reset clears capability across sessions -------------------- reset_state() set_zero_packs() ip = "10.0.2.1" utas_server.fifa17_reset_session(ip) register(ip, 1) ids = store(ip) # session 1: clean assert SENTINEL_ID not in ids, ids utas_server.fifa17_reset_session(ip) # relaunch: fresh unpatched process assert utas_server._FIFA17_STORE[ip] == {"resolver": None, "mode": None} ids = store(ip) # session 2: no re-register -> sentinel assert SENTINEL_ID in ids, "capability must not leak across sessions" print("I new session clears capability -> sentinel: OK") # ---- J. autopatch mismatch => never registers -> sentinel ---------- reset_state() set_zero_packs() ip = "10.0.3.1" utas_server.fifa17_reset_session(ip) # autopatch verify FAILED: no register ids = store(ip) assert SENTINEL_ID in ids, ids assert utas_server._FIFA17_STORE[ip]["mode"] == utas_server.FIFA17_MODE_SENTINEL print("J autopatch mismatch (never registers) -> sentinel: OK") finally: utas_server.visible_unopened_packs = _orig_visible print("capability negotiation matrix A-J: OK") return 0 if __name__ == "__main__": raise SystemExit(main())