OpenFUT/scripts/openfut-reachability.py

#!/usr/bin/env python3
"""OpenFUT UTAS route-reachability reporter.

Parses the openfut-utas-host `owner=` dispatch log (the line the host prints for
EVERY request) into per-owner and per-Python-domain hit counts, so a staging
preflight can assert the post-P1 invariants WITHOUT mutating anything:

  * economy Python hits  == 0        (any nonzero => P1 economy regression)
  * migrated non-economy Python hits == 0   (accountinfo/settings/leaderboards/
      match-reset/phishing are Rust-owned since post-P1)
  * remaining Python domains are exactly the documented residual set.

This is dev/staging tooling (Python is fine here — production authority is Rust).
It only READS a log (file path arg, or stdin); it never touches production.

Log grammar (openfut-utas-host, src/lib.rs eprintln! lines), examples:
  utas-host owner=RUST route=economy method=GET path=/ut/game/fifa17/user/credits status=200
  utas-host owner=PYTHON_FALLBACK method=GET path=/ut/game/fifa17/hub status=200
  utas-host owner=RUST_OVERLAY route=userMassInfo status=200 ...
  utas-host owner=RUST_OBSERVE route=auth status=200 ...
  utas-host owner=RUST route=accountinfo status=200

Usage:
  openfut-reachability.py [LOGFILE]        # report + gate (exit 1 on regression)
  hub logs ... | openfut-reachability.py   # read from stdin
"""
import re
import sys
from collections import Counter

OWNER_RE = re.compile(r"utas-host owner=(\S+)")
PATH_RE = re.compile(r"path=(\S+)")
ROUTE_RE = re.compile(r"route=(\S+)")

# Non-economy routes migrated to Rust ownership post-P1. A PYTHON_FALLBACK hit on
# any of these is a MIGRATION REGRESSION (the classifier lost the route).
MIGRATED_NON_ECONOMY = {
    "user/accountinfo",
    "settings",
    "leaderboards/options",
    "match/reset",
    "phishing",  # phishing/{trusteddevice,question,validate}
    "hub",
}

# Documented residual Python-owned non-economy domains (expected > 0 until
# migrated). Keep in sync with docs/PRODUCTION_AUTHORITY_MATRIX.md.
RESIDUAL_PYTHON = {
    "openfut/account/sync",
    "club/stats",  # year/consumables still Python; club/stats/staff is Rust
    "clientdata/userHubData",
}


def python_domain(path: str) -> str:
    """Normalize a proxied path to its domain key."""
    p = path.split("?", 1)[0]
    if p.startswith("/openfut/account/sync"):
        return "openfut/account/sync"
    # strip /ut/game/<sku>/ or /ut/v2/game/<sku>/ prefix
    m = re.match(r"/ut/(?:v2/)?game/[^/]+/(.*)", p)
    tail = m.group(1) if m else p.lstrip("/")
    if tail.startswith("phishing/"):
        return "phishing"
    if tail.startswith("club/stats"):
        return "club/stats"
    if tail.startswith("clientdata/"):
        return "clientdata/userHubData"
    return tail


def is_economy(line: str, route: str | None) -> bool:
    return route == "economy"


def main() -> int:
    src = sys.stdin
    if len(sys.argv) > 1:
        src = open(sys.argv[1], encoding="utf-8", errors="replace")
    owners = Counter()
    economy_python = []          # economy routes that escaped to Python (BAD)
    migrated_regressions = []    # migrated non-economy routes that hit Python (BAD)
    python_domains = Counter()
    for line in src:
        mo = OWNER_RE.search(line)
        if not mo:
            continue
        owner = mo.group(1)
        owners[owner] += 1
        route = (ROUTE_RE.search(line) or [None, None])[1] if ROUTE_RE.search(line) else None
        if owner == "PYTHON_FALLBACK":
            pm = PATH_RE.search(line)
            path = pm.group(1) if pm else "?"
            dom = python_domain(path)
            python_domains[dom] += 1
            if dom in MIGRATED_NON_ECONOMY:
                migrated_regressions.append(path)
        # A properly classified economy route is owner=RUST route=economy. If an
        # economy tail ever shows up as PYTHON_FALLBACK that is the regression.
        if owner == "PYTHON_FALLBACK" and route is None:
            pm = PATH_RE.search(line)
            path = pm.group(1) if pm else ""
            if re.search(r"/(user/credits|store/(transaction|purchasegroup)|purchased|tradepile|"
                         r"transfermarket|auctionhouse|trade/|item)", path, re.I):
                economy_python.append(path)

    print("=== OWNER COUNTS ===")
    for owner, n in owners.most_common():
        print(f"  {owner:18} {n}")
    print("=== PYTHON_FALLBACK DOMAINS ===")
    for dom, n in python_domains.most_common():
        tag = ""
        if dom in MIGRATED_NON_ECONOMY:
            tag = "  <-- REGRESSION (should be Rust)"
        elif dom not in RESIDUAL_PYTHON:
            tag = "  <-- UNEXPECTED (not in residual set)"
        print(f"  {dom:28} {n}{tag}")

    ok = True
    if economy_python:
        ok = False
        print("\nFAIL: economy routes reached Python (P1 REGRESSION):")
        for p in economy_python:
            print(f"  {p}")
    if migrated_regressions:
        ok = False
        print("\nFAIL: migrated non-economy routes reached Python:")
        for p in migrated_regressions:
            print(f"  {p}")
    unexpected = [d for d in python_domains
                  if d not in RESIDUAL_PYTHON and d not in MIGRATED_NON_ECONOMY]
    if unexpected:
        print("\nWARN: undocumented Python domains (classify + add to matrix):")
        for d in unexpected:
            print(f"  {d} ({python_domains[d]})")

    print("\nGATE:", "PASS" if ok else "FAIL")
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())