tools: authoritative Core-state snapshot for FUT loop verification

Reads openfut-core's authoritative API directly (balance, entitlements, profile, club, collection, squad/ext) with the fifa17 game header and emits a machine-readable JSON snapshot plus integrity checks: coin cross-check, duplicate owned_card_id, squad-references-non-owned, owned-set + card-multiset hashes for drift/resurrection detection across operations and restarts. Read-only oracle tooling; used to verify Phases 1-9 of the Core-backed FUT loop audit.
2026-08-19 19:20:28 +00:00
parent 0019806a3b
commit 8e1fb640b6
1 changed files with 146 additions and 0 deletions
@@ -0,0 +1,146 @@
+#!/usr/bin/env python3
+"""Authoritative Core state snapshot for FIFA17 FUT loop verification.
+
+Queries openfut-core's authoritative read API directly (NOT the FIFA UI, NOT the
+host projection, NOT Python) and emits a machine-readable JSON snapshot with the
+integrity fields needed to detect, across any economy operation or a full stack
+restart:
+
+  * coin drift            (balance vs club.coins cross-check + absolute value)
+  * duplicate ownership   (repeated owned_card_id)
+  * missing ownership     (squad references an owned_card_id not in the collection)
+  * resurrection          (owned-set hash changes when it must not)
+  * entitlement dup/count (unopened pack entitlements)
+  * stale squad mappings  (squad player ids absent from ownership)
+
+This is verification/oracle tooling, not production behavior. It reads only.
+
+Usage:
+  core-snapshot.py [--core URL] [--game fifa17] [--squad-ns fifa17.squad]
+                   [--out FILE] [--label NAME]
+Env fallbacks: OPENFUT_CORE_URL, OPENFUT_GAME, OPENFUT_SQUAD_NS.
+Exit code 0 always for a successful read; integrity problems are reported IN the
+snapshot (`integrity.ok` / `integrity.problems`) so callers can diff/assert.
+"""
+import argparse
+import hashlib
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+from datetime import datetime, timezone
+
+
+def fetch(core, game, path):
+    req = urllib.request.Request(core + path, headers={"X-OpenFUT-Game": game})
+    try:
+        with urllib.request.urlopen(req, timeout=15) as f:
+            return f.status, f.read()
+    except urllib.error.HTTPError as e:
+        return e.code, e.read()
+    except Exception as e:  # noqa: BLE001 - surface transport errors in the snapshot
+        return None, str(e).encode()
+
+
+def fetch_json(core, game, path):
+    st, body = fetch(core, game, path)
+    if st != 200:
+        raise SystemExit(f"Core read {path} failed: status={st} body={body[:200]!r}")
+    return json.loads(body)
+
+
+def sha(items):
+    h = hashlib.sha256()
+    for it in items:
+        h.update(str(it).encode())
+        h.update(b"\x00")
+    return h.hexdigest()
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--core", default=os.environ.get("OPENFUT_CORE_URL", "http://127.0.0.1:18101"))
+    ap.add_argument("--game", default=os.environ.get("OPENFUT_GAME", "fifa17"))
+    ap.add_argument("--squad-ns", default=os.environ.get("OPENFUT_SQUAD_NS", "fifa17.squad"))
+    ap.add_argument("--out")
+    ap.add_argument("--label", default="")
+    args = ap.parse_args()
+
+    balance = fetch_json(args.core, args.game, "/economy/balance")
+    entitlements = fetch_json(args.core, args.game, "/economy/entitlements")
+    profile = fetch_json(args.core, args.game, "/profile")
+    club = fetch_json(args.core, args.game, "/club")
+    collection = fetch_json(args.core, args.game, "/collection")["collection"]
+    squad = fetch_json(args.core, args.game, f"/squad/ext?namespace={args.squad_ns}")
+
+    coins_balance = balance.get("balance")
+    coins_club = club.get("coins")
+
+    owned_ids = sorted(x["owned_card_id"] for x in collection)
+    card_ids = sorted(x["card"]["id"] for x in collection)
+    dup_owned = sorted({i for i in owned_ids if owned_ids.count(i) > 1}) if len(owned_ids) != len(set(owned_ids)) else []
+    loans = [x["owned_card_id"] for x in collection if x.get("is_loan")]
+
+    # Squad player ownership references (from the opaque extension payload).
+    squad_player_ids = []
+    ext = squad.get("extension") or {}
+    payload_raw = ext.get("payload")
+    kit_numbers = {}
+    if payload_raw:
+        try:
+            payload = json.loads(payload_raw)
+            kit_numbers = payload.get("kit_numbers", {}) or {}
+            squad_player_ids = sorted(kit_numbers.keys())
+        except (json.JSONDecodeError, TypeError):
+            pass
+
+    owned_set = set(owned_ids)
+    squad_missing = sorted(pid for pid in squad_player_ids if pid not in owned_set)
+
+    problems = []
+    if coins_balance != coins_club:
+        problems.append(f"coin cross-check mismatch: balance={coins_balance} club={coins_club}")
+    if dup_owned:
+        problems.append(f"duplicate owned_card_id: {dup_owned[:10]} (+{max(0,len(dup_owned)-10)} more)")
+    if squad_missing:
+        problems.append(f"squad references non-owned ids: {squad_missing}")
+
+    snap = {
+        "label": args.label,
+        "captured_at": datetime.now(timezone.utc).isoformat(),
+        "core": args.core,
+        "game": args.game,
+        "coins": {"balance": coins_balance, "club": coins_club},
+        "profile": {"username": profile.get("username"), "level": profile.get("level"), "xp": profile.get("xp")},
+        "entitlements": {"count": len(entitlements), "ids": entitlements},
+        "collection": {
+            "owned": len(collection),
+            "distinct_owned_card_id": len(set(owned_ids)),
+            "distinct_card_id": len(set(card_ids)),
+            "loans": len(loans),
+            "duplicate_owned_card_id": dup_owned,
+            "owned_set_sha256": sha(owned_ids),
+            "card_multiset_sha256": sha(card_ids),
+        },
+        "squad": {
+            "namespace": args.squad_ns,
+            "verdict": squad.get("verdict"),
+            "player_count": len(squad_player_ids),
+            "player_ids": squad_player_ids,
+            "missing_from_ownership": squad_missing,
+            "ext_schema_version": ext.get("schema_version"),
+        },
+        "integrity": {"ok": not problems, "problems": problems},
+    }
+
+    text = json.dumps(snap, indent=2, sort_keys=True)
+    if args.out:
+        with open(args.out, "w") as f:
+            f.write(text + "\n")
+    print(text)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())