fix(fifa17): isolate patched-client capability per session

Harden the empty-My-Packs capability binding so a verified FIFA process can never
enable clean/no-sentinel Store topology for another unverified process that merely
shares its source IP. The prototype keyed the decision by source IP alone; two FIFA
processes (concurrent, or a relaunch) share an IP, so an unpatched process could
inherit a patched one's clean-v1 mode and crash. Source IP is now auxiliary only.

- Authoritative key = the per-login UTAS session id (X-UT-SID). /ut/auth now mints
  a fresh unique SID per login (was a shared constant) and opens a session record
  keyed by that SID; the client echoes it on every later call incl.
  /store/purchasegroup (live-confirmed). The legacy constant is still accepted by
  the retired security-question gate only, never to grant clean-v1.
- Session state: _FIFA17_SESSIONS[sid] = {ip, persona, resolver, mode, created,
  last_seen}. Store mode freezes at the first /store/purchasegroup of the session
  and is immutable thereafter. Fail-closed: unknown SID, or a SID presented from a
  different source IP than it was opened on, resolves to the sentinel.
- Launcher capability (out-of-band; cannot know the SID) is matched by (ip, persona)
  as a SINGLE-USE, short-TTL pending, bound to exactly one session at whichever comes
  first: its login (pending predates auth), the registration (session already live),
  or its first store request. Ambiguous same-(ip,persona) concurrent registration is
  ignored-late -> both sentinel (never a wrong clean).
- Session cleanup: activity-based TTL sweep (sessions 3600s idle, pendings 120s);
  reaping only removes expired entries and never affects another live session.
- account_sync now clears only stale pending for the machine (pre-launch hygiene);
  it no longer resets a per-IP mode (there is no per-IP mode any more).

Backend-only: the launcher registration payload (already carries personaId) is
unchanged. Additive; P2 sentinel remains the else-branch and the default.

Tests: matrix A-Q incl. same-IP concurrent (K), same-IP+persona relaunch (L),
same-IP failed-patch (M), late-registration-vs-frozen-sessions (N), TTL expiry (O),
duplicate/idempotent registration (P), and register-before-login pending (Q).
This commit is contained in:
funman300
2026-08-13 04:39:53 +00:00
parent d4c3811665
commit 805d754dc8
2 changed files with 347 additions and 189 deletions
+189 -141
View File
@@ -1,23 +1,33 @@
#!/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.
The additive empty-My-Packs switch on top of the P2 65534 sentinel: the sentinel is
suppressed for ONE FIFA session only when the launcher has registered a verified
resolver capability (v1) that binds to THAT process's UTAS session (keyed by the
per-login-unique X-UT-SID; source IP + persona are auxiliary). Every failure /
unknown / late / cross-process / cross-session case is fail-closed to the 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
The initial prototype keyed by source IP alone; this suite proves the hardened
per-session binding, including two sessions that SHARE a source IP.
Matrix (docs/plans/FIFA17_PATCHED_CLIENT_CAPABILITY.md):
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 / capability -> endpoint 400 AND mode sentinel
F late capability after sentinel freeze -> stays sentinel
G capability disappears after clean freeze -> stays clean (immutable)
H two IPs (A verified, B none) -> A clean, B sentinel (no global leak)
I new session after reset -> fresh unpatched -> sentinel
J autopatch mismatch => never registers -> sentinel
K SAME IP, two sessions (A patched, B not) -> A clean, B sentinel
L SAME IP+persona relaunch (old ok, new not) -> new session sentinel
M SAME IP, failed-patch second session -> first clean, second sentinel
N late registration when sessions are frozen -> does not modify active sessions
O session cleanup / TTL expiry -> capability gone, sentinel
P duplicate registration for a session -> idempotent; no post-freeze change
Q register-before-login (pending consumed) -> clean
Standalone unit test in the project style: `python3 test_capability_negotiation.py`.
"""
@@ -33,13 +43,15 @@ if TOOLS not in sys.path:
SENTINEL_ID = 65534
REAL_PACK_ID = 1
PERSONA = 111001
class _H:
"""Minimal request-handler stand-in: peer IP + optional JSON body."""
"""Minimal request-handler stand-in: peer IP, optional X-UT-SID, optional body."""
def __init__(self, ip, body=None):
def __init__(self, ip, body=None, sid=None):
self.client_address = (ip, 54321)
self.headers = {"X-UT-SID": sid} if sid is not None else {}
self._body = json.dumps(body).encode("utf-8") if body is not None else b""
@@ -62,166 +74,202 @@ def main():
importlib.reload(fut_accounts)
importlib.reload(utas_server)
fut_accounts.activate({"personaId": 111001, "personaName": "TEST_A"})
us = utas_server
CLEAN, SENT = us.FIFA17_MODE_CLEAN, us.FIFA17_MODE_SENTINEL
# ---- deterministic pack topology helpers -------------------------------
_orig_visible = utas_server.visible_unopened_packs
_orig_visible = us.visible_unopened_packs
def set_zero_packs():
utas_server.visible_unopened_packs = lambda: []
us.visible_unopened_packs = lambda: []
def set_real_pack():
utas_server.visible_unopened_packs = lambda: [REAL_PACK_ID]
us.visible_unopened_packs = lambda: [REAL_PACK_ID]
def reset_state():
"""Fresh capability store between cases (no cross-case leakage)."""
utas_server._FIFA17_STORE.clear()
us._FIFA17_SESSIONS.clear()
us._FIFA17_PENDING.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 auth(sid, ip, persona=PERSONA):
"""Simulate /ut/auth opening a per-login session with a chosen sid."""
us.fifa17_open_session(sid, ip, persona)
def register(ip, version, persona=PERSONA, pid=4242):
return us.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))
def store(sid, ip):
status, cat = us.store_catalog(_H(ip, sid=sid))
assert status == 200, status
return _ids(cat)
def mode_of(sid):
return us._FIFA17_SESSIONS[sid]["mode"]
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
reset_state(); set_zero_packs()
auth("sidA", "10.0.0.1")
assert SENTINEL_ID in store("sidA", "10.0.0.1")
assert mode_of("sidA") == SENT
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)
reset_state(); set_zero_packs()
auth("sidB", "10.0.0.2")
assert register("10.0.0.2", 1)[0] == 200
ids = store("sidB", "10.0.0.2")
assert SENTINEL_ID not in ids, ids
assert utas_server._FIFA17_STORE[ip]["mode"] == utas_server.FIFA17_MODE_CLEAN
assert mode_of("sidB") == 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
reset_state(); set_real_pack()
auth("sidC", "10.0.0.3")
ids = store("sidC", "10.0.0.3")
assert REAL_PACK_ID in ids and SENTINEL_ID not 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
reset_state(); set_real_pack()
auth("sidD", "10.0.0.4"); register("10.0.0.4", 1)
ids = store("sidD", "10.0.0.4")
assert REAL_PACK_ID in ids and SENTINEL_ID not 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")
# ---- E. unsupported version / capability -> 400 + sentinel ---------
reset_state(); set_zero_packs()
auth("sidE", "10.0.0.5")
assert register("10.0.0.5", 2)[0] == 400
assert register("10.0.0.5", 99)[0] == 400
assert us.fifa17_capability_route(
_H("10.0.0.5", {"capability": "bogus", "version": 1}))[0] == 400
assert SENTINEL_ID in store("sidE", "10.0.0.5")
assert mode_of("sidE") == SENT
print("E unsupported version/capability -> 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
# ---- F. late capability after sentinel freeze -> sentinel ----------
reset_state(); set_zero_packs()
auth("sidF", "10.0.0.6")
assert SENTINEL_ID in store("sidF", "10.0.0.6") # freezes sentinel
assert register("10.0.0.6", 1)[0] == 200 # session frozen -> ignored-late
assert SENTINEL_ID in store("sidF", "10.0.0.6")
assert mode_of("sidF") == SENT
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
# ---- G. capability disappears after clean freeze -> clean ----------
reset_state(); set_zero_packs()
auth("sidG", "10.0.0.7"); register("10.0.0.7", 1)
assert SENTINEL_ID not in store("sidG", "10.0.0.7") # freezes clean
us._FIFA17_SESSIONS["sidG"]["resolver"] = None # capability vanishes
assert SENTINEL_ID not in store("sidG", "10.0.0.7")
assert mode_of("sidG") == 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")
# ---- H. two IPs (A verified, B none) -> no global leak -------------
reset_state(); set_zero_packs()
auth("sidH1", "10.0.1.1"); register("10.0.1.1", 1)
auth("sidH2", "10.0.1.2")
assert SENTINEL_ID not in store("sidH1", "10.0.1.1")
assert SENTINEL_ID in store("sidH2", "10.0.1.2")
print("H two 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")
# ---- I. new session after reset -> fresh unpatched -> sentinel -----
reset_state(); set_zero_packs()
auth("sidI1", "10.0.1.3"); register("10.0.1.3", 1)
assert SENTINEL_ID not in store("sidI1", "10.0.1.3") # A clean
us.fifa17_clear_pending("10.0.1.3") # relaunch boundary
auth("sidI2", "10.0.1.3") # new SID, autopatch failed
assert SENTINEL_ID in store("sidI2", "10.0.1.3")
print("I new session after reset -> sentinel (no cross-process leak): 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
reset_state(); set_zero_packs()
auth("sidJ", "10.0.1.4")
assert SENTINEL_ID in store("sidJ", "10.0.1.4")
print("J autopatch mismatch (never registers) -> sentinel: OK")
finally:
utas_server.visible_unopened_packs = _orig_visible
print("capability negotiation matrix A-J: OK")
# ---- K. SAME IP, two sessions: patched A clean, unpatched B sent ---
reset_state(); set_zero_packs()
IP = "10.0.2.1"
auth("sidK_A", IP)
assert register(IP, 1)[0] == 200 # A sole candidate -> bound
auth("sidK_B", IP) # B joins, never registers
assert SENTINEL_ID not in store("sidK_A", IP)
assert SENTINEL_ID in store("sidK_B", IP)
print("K same-IP two sessions -> A clean, B sentinel: OK")
# ---- L. SAME IP+persona relaunch: old ok, new not -> new sentinel --
reset_state(); set_zero_packs()
IP = "10.0.2.2"
auth("sidL_old", IP, PERSONA); register(IP, 1, PERSONA)
assert SENTINEL_ID not in store("sidL_old", IP)
us.fifa17_clear_pending(IP)
auth("sidL_new", IP, PERSONA) # same persona, unverified
assert SENTINEL_ID in store("sidL_new", IP)
print("L same-IP+persona relaunch -> new session sentinel: OK")
# ---- M. SAME IP, failed-patch second session -----------------------
reset_state(); set_zero_packs()
IP = "10.0.2.3"
auth("sidM1", IP); register(IP, 1)
assert SENTINEL_ID not in store("sidM1", IP)
auth("sidM2", IP) # autopatch failed
assert SENTINEL_ID in store("sidM2", IP)
print("M same-IP failed-patch second session -> sentinel: OK")
# ---- N. late reg when sessions frozen -> no active session change --
reset_state(); set_zero_packs()
IP = "10.0.2.4"
auth("sidN1", IP); register(IP, 1)
assert SENTINEL_ID not in store("sidN1", IP) # N1 frozen clean
auth("sidN2", IP)
assert SENTINEL_ID in store("sidN2", IP) # N2 frozen sentinel
assert register(IP, 1)[0] == 200 # late: both frozen -> ignored
assert SENTINEL_ID not in store("sidN1", IP) # unchanged
assert SENTINEL_ID in store("sidN2", IP) # unchanged
print("N late registration does not modify active sessions: OK")
# ---- O. session cleanup / TTL expiry -> capability gone ------------
reset_state(); set_zero_packs()
IP = "10.0.2.5"
auth("sidO", IP); register(IP, 1)
assert SENTINEL_ID not in store("sidO", IP) # clean while live
us._FIFA17_SESSIONS["sidO"]["last_seen"] = (
us._fifa17_now() - us.FIFA17_SESSION_TTL - 10.0)
store("sidUNKNOWN", IP) # any op triggers reap
assert "sidO" not in us._FIFA17_SESSIONS, "expired session not reaped"
assert SENTINEL_ID in store("sidO", IP) # gone -> sentinel
print("O session cleanup / TTL expiry -> sentinel: OK")
# ---- P. duplicate registration -> idempotent, no post-freeze change
reset_state(); set_zero_packs()
IP = "10.0.2.6"
auth("sidP", IP)
assert register(IP, 1)[0] == 200 # bound
assert register(IP, 1)[0] == 200 # duplicate -> ignored-late
assert SENTINEL_ID not in store("sidP", IP) # still clean
assert register(IP, 1)[0] == 200 # after freeze
assert SENTINEL_ID not in store("sidP", IP) # unchanged
assert mode_of("sidP") == CLEAN
print("P duplicate registration -> idempotent: OK")
# ---- Q. register-before-login: pending consumed at auth -> clean ---
reset_state(); set_zero_packs()
IP = "10.0.2.7"
assert register(IP, 1)[0] == 200 # no session yet -> pending
assert (IP, PERSONA) in us._FIFA17_PENDING
auth("sidQ", IP, PERSONA) # consumes pending
assert (IP, PERSONA) not in us._FIFA17_PENDING # single-use
assert SENTINEL_ID not in store("sidQ", IP)
assert mode_of("sidQ") == CLEAN
print("Q register-before-login pending consumed -> clean: OK")
finally:
us.visible_unopened_packs = _orig_visible
print("capability negotiation matrix A-Q: OK")
return 0