d9e80a774a
Case R makes the F3 session-topology invariant explicit alongside the A-Q matrix: for a single X-UT-SID the frozen empty-My-Packs mode never flips in either direction (Sentinel stays Sentinel even if a capability later appears; Clean stays Clean even if the capability is wiped), while a fresh SID from the same IP decides independently. Complements F/G/K.
302 lines
14 KiB
Python
Executable File
302 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Tests for the FIFA 17 verified-patched-client capability negotiation.
|
|
|
|
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.
|
|
|
|
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
|
|
R topology freeze immutable per SID -> no flip either way; new SID fresh
|
|
|
|
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
|
|
PERSONA = 111001
|
|
|
|
|
|
class _H:
|
|
"""Minimal request-handler stand-in: peer IP, optional X-UT-SID, optional body."""
|
|
|
|
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""
|
|
|
|
|
|
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)
|
|
|
|
us = utas_server
|
|
CLEAN, SENT = us.FIFA17_MODE_CLEAN, us.FIFA17_MODE_SENTINEL
|
|
|
|
_orig_visible = us.visible_unopened_packs
|
|
|
|
def set_zero_packs():
|
|
us.visible_unopened_packs = lambda: []
|
|
|
|
def set_real_pack():
|
|
us.visible_unopened_packs = lambda: [REAL_PACK_ID]
|
|
|
|
def reset_state():
|
|
us._FIFA17_SESSIONS.clear()
|
|
us._FIFA17_PENDING.clear()
|
|
|
|
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(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()
|
|
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()
|
|
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 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()
|
|
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()
|
|
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 / 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 -> 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 -> 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 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. 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()
|
|
auth("sidJ", "10.0.1.4")
|
|
assert SENTINEL_ID in store("sidJ", "10.0.1.4")
|
|
print("J autopatch mismatch (never registers) -> sentinel: 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")
|
|
|
|
# ---- R. topology freeze immutable per SID; new SID decides fresh ----
|
|
# F3 invariant: once a SID's store topology is decided it NEVER flips,
|
|
# in either direction, and a different SID may decide differently.
|
|
reset_state(); set_zero_packs()
|
|
IP = "10.0.2.8"
|
|
# frozen Sentinel never becomes Clean, even if a capability appears later
|
|
auth("sidR_s", IP)
|
|
assert SENTINEL_ID in store("sidR_s", IP) # freeze Sentinel
|
|
register(IP, 1)
|
|
us._FIFA17_SESSIONS["sidR_s"]["resolver"] = 1 # force-present capability
|
|
assert SENTINEL_ID in store("sidR_s", IP) # STILL Sentinel
|
|
assert mode_of("sidR_s") == SENT
|
|
# frozen Clean never becomes Sentinel, even if the capability is wiped
|
|
auth("sidR_c", IP); register(IP, 1)
|
|
assert SENTINEL_ID not in store("sidR_c", IP) # freeze Clean
|
|
us._FIFA17_SESSIONS["sidR_c"]["resolver"] = None # capability vanishes
|
|
assert SENTINEL_ID not in store("sidR_c", IP) # STILL Clean
|
|
assert mode_of("sidR_c") == CLEAN
|
|
# a fresh SID (same IP) decides independently
|
|
auth("sidR_new", IP)
|
|
assert SENTINEL_ID in store("sidR_new", IP)
|
|
print("R topology freeze immutable per SID; new SID fresh: OK")
|
|
|
|
finally:
|
|
us.visible_unopened_packs = _orig_visible
|
|
|
|
print("capability negotiation matrix A-R: OK")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|