From 8e1fb640b6652ba0700d443f77965ced36314581 Mon Sep 17 00:00:00 2001 From: funman300 Date: Wed, 19 Aug 2026 19:20:28 +0000 Subject: [PATCH] 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. --- scripts/core-snapshot.py | 146 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) create mode 100755 scripts/core-snapshot.py diff --git a/scripts/core-snapshot.py b/scripts/core-snapshot.py new file mode 100755 index 0000000..d791e8a --- /dev/null +++ b/scripts/core-snapshot.py @@ -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())