Files
OpenFUT/fifa17-recon/tools/test_capability_negotiation.py
T
funman300 b25761ea31 feat(fifa17): negotiate clean empty My Packs mode
Backend side of the handshake: suppress the synthetic 65534 My-Packs sentinel
ONLY for a session whose client has registered a verified resolver-guard
capability. Additive; the P2 active-sentinel path is retained as the else-branch
and the universal default. Fail-closed everywhere.

- Per-client state keyed by source IP (client_address[0]; the only per-connection
  discriminator in this single-account, stateless backend): _FIFA17_STORE[ip] =
  {resolver, mode}; mode in {None, "sentinel", "clean-v1"}, guarded by a lock.
- New POST /openfut/fifa17/capability endpoint: accepts only
  {"capability":"empty_mypacks_resolver","version":1,...}; unknown capability or
  version => 400 and records nothing (=> sentinel).
- account_sync (the launcher's required per-launch call) resets the per-ip record
  => a new FIFA process starts unfrozen with no inherited capability.
- Store topology is frozen at the FIRST /store/purchasegroup per session:
  clean-v1 iff a v1 capability is registered, else sentinel; immutable thereafter
  (late capability logged + ignored this session; a disappeared capability does
  not un-freeze a clean session). This enforces the SESSION-STABLE invariant.
- store_catalog zero-owned-packs branch: clean-v1 emits NO mypacks group (the
  client guard routes category -1 to Browse); every other case emits the existing
  active 65534 sentinel verbatim. PACK_CATALOG / pack 70 / normal packs / profile
  untouched. FIFA-17 only; not lifted into game-independent Core.
- Tests: full matrix A-J incl. concurrency isolation (two IPs, no global leak) and
  no cross-process capability leak.

Design: docs/plans/FIFA17_PATCHED_CLIENT_CAPABILITY.md.
2026-08-13 04:03:37 +00:00

230 lines
9.7 KiB
Python
Executable File

#!/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())