funman300/OpenFUT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

utas: first real corpus, and two sanitiser gaps the audit caught

Browse Source 
24 transactions across 11 connections from a retail session: login,
hub, one pack open, two squad saves, a quick-sell, with before/after
state manifests. Raw .ofcap stays gitignored at 0600; the sanitized
corpus is committed as adapter fixtures.

TWO GAPS FOUND BY AUDITING THE OUTPUT, NOT BY TRUSTING THE SANITISER.

1. `POST /ut/auth` carries `macAddress` and `deviceId`. Session tokens
   were being redacted correctly and these were not. A committed fixture
   is a published fixture.

2. Then, with those fixed, the audit fired AGAIN on the file about to be
   committed: `GET .../phishing/trusteddevice?deviceId=...` puts the id in
   the QUERY STRING. Three input surfaces carry identifiers -- headers,
   JSON bodies, and query strings -- and the sanitiser knew about two.

Both fixed in the tool rather than by editing the file, with a
regression test and a mutation for the query path.

AND A THIRD ARTEFACT MIX-UP, in the mutation harness itself. It reported
the query-redaction mutation as SURVIVED while a hand-run of the same
mutation killed it. Cause: the harness pointed at a stale scratchpad copy
of the test that pre-dated the query assertion, so it was faithfully
testing the mutated tool against a test that could not detect the
mutation. That is the same class as the build guard checking the wrong
binary and cargo reusing a binary compiled from mutated source -- the
third instance today of measuring the wrong artifact. The harness now
resolves ROOT from its own location and runs the COMMITTED test; the
stale copy is deleted.

Harness committed as scripts/mutate-utas-observe.py so this is repeatable
rather than a thing that happened once in a scratch directory. 6/6 killed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
funman300
2026-08-11 18:33:59 +00:00
parent cdea85e214
commit 0b66662525
6 changed files with 405 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/openfut-utas-observe.py
+126 -48
View File
@@ -350,18 +350,48 @@ def header_list(head):
SECRET_HEADERS = {"authorization", "cookie", "set-cookie", "x-ut-sid", "easw-session-data-nucleus-id"}
SECRET_BODY_KEYS = ("sid", "token", "password", "answer", "secret")
# Substring-matched against lowercased JSON keys. Extended after auditing the
# first real corpus: session tokens were being redacted correctly, but the
# client also sends hardware and device identifiers in `POST /ut/auth`, and a
# committed fixture is a published fixture. Audit the output before publishing,
# every time -- "the sanitiser handles it" is a belief until it is checked.
SECRET_BODY_KEYS = (
"sid", "token", "password", "answer", "secret",
"macaddress", "deviceid", "authcode",
)
def sanitize(rec):
"""Repository-safe form. Bodies are preserved EXACTLY; only credentials go.
THREE input surfaces carry identifiers, not two. The first version handled
headers and JSON bodies and was still about to publish a device id, because
`GET /ut/game/fifa17/phishing/trusteddevice?deviceId=...` puts it in the
QUERY STRING. Caught by auditing the output rather than by trusting the
sanitiser -- which is the only reason it is a comment and not a leak.
Exact bodies first, sanitise second: the corpus is worthless if response
payloads are reshaped, so nothing here touches structure -- it replaces
known-secret header values and known-secret JSON keys, and records that it
did so.
"""
redacted = []
# Query string: same key rules as bodies.
q = rec["request"].get("query") or ""
if q:
parts, hit = [], False
for kv in q.split("&"):
k, eq, v = kv.partition("=")
if eq and any(x in k.lower() for x in SECRET_BODY_KEYS):
parts.append(k + "=<REDACTED>")
redacted.append("request.query." + k)
hit = True
else:
parts.append(kv)
if hit:
rec["request"]["query"] = "&".join(parts)
for side in ("request", "response"):
hs = rec[side]["headers"]
for pair in hs:
@@ -407,73 +437,121 @@ def redact_json(body):
# ────────────────────────────────────────────────────────────── snapshot ────
# Read-only GETs only. `/openfut/account/sync` is deliberately NOT here: it is a
# POST that selects and writes an account, so using it as a snapshot would
# mutate the state the snapshot exists to observe.
#
# All six were confirmed to answer 200 without a session token.
SNAPSHOT_ROUTES = [
("account", "/openfut/account/sync"),
("credits", "/ut/game/fifa17/user/credits"),
("userMassInfo", "/ut/game/fifa17/userMassInfo"),
("unassigned", "/ut/game/fifa17/purchased/items"),
("activeSquad", "/ut/game/fifa17/squad/active"),
("tradePile", "/ut/game/fifa17/tradePile"),
("accountInfo", "/ut/game/fifa17/user/accountinfo"),
]
def cmd_snapshot(args):
"""A light state manifest, for before/after pairing around a session.
def http_get(host, port, path):
s = socket.create_connection((host, int(port)), timeout=10)
s.sendall(("GET %s HTTP/1.1\r\nHost: %s:%s\r\nConnection: close\r\n\r\n"
% (path, host, port)).encode())
buf = b""
while True:
d = s.recv(65536)
if not d:
break
buf += d
s.close()
head, _, body = buf.partition(b"\r\n\r\n")
line = head.split(b"\r\n")[0].decode("latin1")
status = int(line.split()[1]) if len(line.split()) > 1 else None
return status, body
Deliberately a summary, not a database dump: the point is to say what
changed, and a committed copy of the save would be both huge and full of
things that are not evidence.
def cmd_snapshot(args):
"""A state manifest for pairing before/after around a capture session.
A summary plus a hash, not a database dump. The named fields are what a
human reads; the sha256 of each full body is the safety net, because a
summary can only report changes in fields somebody thought to list. If the
hash moves and no field does, the summary is incomplete -- and that is
itself a finding rather than a silent miss.
Raw bodies are written alongside at mode 0600 so a real diff is possible
later without re-running the session.
"""
import hashlib
uh, up = args.upstream.rsplit(":", 1)
out_path = os.path.abspath(args.out)
raw_dir = os.path.join(os.path.dirname(out_path), "bodies")
os.makedirs(raw_dir, exist_ok=True)
man = {"unix": time.time(), "upstream": args.upstream, "routes": {}}
for name, path in SNAPSHOT_ROUTES:
try:
s = socket.create_connection((uh, int(up)), timeout=5)
body = b'{"personaId":0}'
req = ("POST %s HTTP/1.1\r\nHost: %s:%s\r\nContent-Type: application/json\r\n"
"Content-Length: %d\r\nConnection: close\r\n\r\n" % (path, uh, up, len(body)))
s.sendall(req.encode() + body)
buf = b""
while True:
d = s.recv(4096)
if not d:
break
buf += d
s.close()
_, _, payload = buf.partition(b"\r\n\r\n")
man["routes"][name] = summarize(payload)
status, body = http_get(uh, up, path)
except Exception as e:
# Recorded as an error rather than omitted: a manifest missing a
# section must not look like a section that was empty.
# Recorded as an error, never omitted: a missing section must not
# be mistakable for an empty one.
man["routes"][name] = {"error": str(e)}
os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True)
with open(args.out, "w") as f:
continue
bp = os.path.join(raw_dir, name + ".json")
fd = os.open(bp, os.O_CREAT | os.O_WRONLY | os.O_TRUNC, 0o600)
with os.fdopen(fd, "wb") as f:
f.write(body)
man["routes"][name] = {
"status": status,
"body_len": len(body),
"sha256": hashlib.sha256(body).hexdigest(),
"fields": extract(name, body),
}
os.makedirs(os.path.dirname(out_path), exist_ok=True)
with open(out_path, "w") as f:
json.dump(man, f, indent=2, sort_keys=True)
f.write("\n")
print("wrote %s" % args.out)
print("wrote %s" % out_path)
for k, v in man["routes"].items():
print(" %s: %s" % (k, v))
print(" %-13s %s" % (k, v.get("fields", v)))
return 0
INTERESTING = ("coins", "unopenedPacks", "personaId", "personaName", "clubName",
"level", "experience", "accountFunds", "itemCount", "squadId")
def summarize(payload):
def extract(name, body):
"""Named fields per route. Unknown shapes report what they can rather than
raising -- the hash still covers everything."""
try:
doc = json.loads(payload)
d = json.loads(body)
except Exception:
return {"raw_len": len(payload)}
out = {}
def walk(o, prefix=""):
if isinstance(o, dict):
for k, v in o.items():
if k in INTERESTING and not isinstance(v, (dict, list)):
out[k] = v
else:
walk(v, prefix + k + ".")
elif isinstance(o, list):
out.setdefault(prefix.rstrip(".") + ".count", len(o))
walk(doc)
return out
return {"unparsed_len": len(body)}
f = {}
if name == "credits":
f["credits"] = d.get("credits")
elif name == "userMassInfo":
ui = d.get("userInfo", {})
for k in ("personaId", "clubName", "clubAbbr", "trophies"):
if k in ui:
f[k] = ui[k]
for k, v in d.items():
if isinstance(v, list):
f[k + ".count"] = len(v)
elif name == "unassigned":
f["itemData.count"] = len(d.get("itemData", []))
elif name == "activeSquad":
f["id"] = d.get("id")
f["formation"] = d.get("formation")
players = d.get("players", [])
f["players.count"] = len(players)
# Slot -> item id, so a two-player swap is visible in the diff.
f["slots"] = {
str(p.get("index", i)): (p.get("itemData") or {}).get("id")
for i, p in enumerate(players)
}
elif name == "tradePile":
f["auctionInfo.count"] = len(d.get("auctionInfo", []))
elif name == "accountInfo":
f["keys"] = sorted(d.keys()) if isinstance(d, dict) else None
return f
def main():