OpenFUT/scripts/openfut-interception-preflight.py

#!/usr/bin/env python3
"""Report host-level packet interception affecting OpenFUT endpoints.

WHY THIS EXISTS
---------------
During the 2026-08-22 Rust production cutover, four stale `openfut-switch` nft
rules were still redirecting production-facing traffic to staging:

    42127 -> :42227    8081 -> :8281    8094 -> :18094    8099 -> :18106

They matched `ip daddr 10.10.0.120`, so every server-side probe via 127.0.0.1 or
the container IP passed while the CLIENT was refused or silently sent to
staging. That cost an entire false-negative acceptance round: a retail
quick-sell landed on staging while production sat untouched, and the launcher
reported "OpenFUT server is not answering".

The lesson is mechanical, so the check is too: a connectivity gate that only
probes loopback proves nothing about what the client reaches.

READ-ONLY. This tool never deletes a rule. Removing interception is a
deliberate operator act (`openfut-switch.sh off --name <id>`).

Exit status: 0 CLEAN, 1 INTERCEPTION_PRESENT, 2 could not determine.

    python3 scripts/openfut-interception-preflight.py
    python3 scripts/openfut-interception-preflight.py --advertise 10.10.0.120
"""
from __future__ import annotations

import argparse
import re
import shutil
import socket
import subprocess
import sys

# The endpoints a FIFA 17 client actually dials, plus the staging twins that
# stale rules historically pointed at.
PRODUCTION_PORTS = {
    8099: "UTAS (Rust utas-host)",
    8081: "roster",
    8094: "POW api",
    8085: "POW content",
    4216: "LSX (client-side)",
    42127: "Blaze redirector",
    42130: "Blaze main",
    42131: "Nucleus",
}

# LSX runs on the GAME machine, not here: the compose file publishes 4216 but
# `OPENFUT_SERVERS` excludes lsx by default, so "published but not served" is
# its normal state and must not be reported as interception. Every other port
# above is expected to be served on this host.
NOT_SERVED_HERE = {4216}
STAGING_PORTS = {8299: "staging UTAS", 42327: "staging redirector",
                 42330: "staging blaze main", 8281: "staging roster",
                 18094: "staging POW", 18106: "retired season-shim"}
ALL_PORTS = dict(PRODUCTION_PORTS)
ALL_PORTS.update(STAGING_PORTS)


def _run(cmd: list[str]) -> str:
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
        return r.stdout
    except Exception:
        return ""


def switch_status() -> tuple[str, list[str]]:
    """`openfut-switch.sh status`, which owns the redirect lifecycle."""
    for path in ("/home/alex/OpenFUT/openfut-blaze-host/openfut-switch.sh",
                 "openfut-blaze-host/openfut-switch.sh"):
        if shutil.which("bash") and subprocess.run(
                ["test", "-x", path], capture_output=True).returncode == 0:
            out = _run([path, "status"])
            active = [l.strip() for l in out.splitlines()
                      if "->" in l and "openfut-switch" in l]
            return ("INACTIVE" if "INACTIVE" in out else
                    ("ACTIVE" if active else "UNKNOWN")), active
    return "UNKNOWN", []


def nft_redirects(advertise: str) -> tuple[list[str], list[str]]:
    """Split nft rules touching our ports into REDIRECTs (interception) and
    Docker's own DNAT (legitimate publishing)."""
    out = _run(["sudo", "-n", "nft", "list", "ruleset"])
    if not out:
        out = _run(["nft", "list", "ruleset"])
    redirects, dnats = [], []
    port_re = re.compile(r"dport (\d+)")
    for line in out.splitlines():
        s = line.strip()
        m = port_re.search(s)
        if not m or int(m.group(1)) not in ALL_PORTS:
            continue
        if "redirect to" in s:
            redirects.append(s)
        elif "dnat to" in s:
            dnats.append(s)
    return redirects, dnats


def reachable(host: str, port: int, timeout: float = 2.0) -> bool:
    s = socket.socket()
    s.settimeout(timeout)
    try:
        s.connect((host, port))
        return True
    except OSError:
        return False
    finally:
        s.close()


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--advertise", default="10.10.0.120",
                    help="the address the CLIENT dials (not loopback)")
    args = ap.parse_args()
    adv = args.advertise

    print("OpenFUT interception preflight — advertise=%s" % adv)
    print()

    state, active = switch_status()
    print("openfut-switch : %s" % state)
    for a in active:
        print("   %s" % a)

    redirects, dnats = nft_redirects(adv)
    print("\nnft REDIRECTs on OpenFUT ports : %d%s"
          % (len(redirects), "  <-- INTERCEPTION" if redirects else ""))
    for r in redirects:
        print("   %s" % r[:150])
    print("nft DNAT (docker publishing)   : %d (expected, not interception)"
          % len(dnats))

    # The point of the whole tool: compare loopback with the address the client
    # actually dials. A redirect keyed on the LAN IP is invisible to loopback.
    print("\neffective endpoint, loopback vs advertised:")
    disagree = []
    for port, label in sorted(PRODUCTION_PORTS.items()):
        lo = reachable("127.0.0.1", port)
        wan = reachable(adv, port)
        if lo == wan:
            flag = ""
        elif port in NOT_SERVED_HERE:
            flag = "   (not served here -- expected)"
        else:
            flag = "   <-- DISAGREE"
        if lo != wan and port not in NOT_SERVED_HERE:
            disagree.append((port, label, lo, wan))
        print("   :%-6d %-24s loopback=%-5s advertised=%-5s%s"
              % (port, label, lo, wan, flag))

    intercepted = bool(redirects) or state == "ACTIVE" or bool(disagree)
    print()
    if intercepted:
        print("RESULT: INTERCEPTION_PRESENT")
        if disagree:
            print("  loopback and the advertised address disagree on: %s"
                  % ", ".join(":%d" % p for p, _, _, _ in disagree))
        print("  Nothing was changed. To clear a switch rule, run explicitly:")
        print("    openfut-blaze-host/openfut-switch.sh off --name <id>")
        return 1
    print("RESULT: CLEAN")
    return 0


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