funman300
cdea85e214
UTAS needs a real request/response corpus before any Rust is written: it is where protocol shape and FUT state start being coupled, so guessing is worse here than it was for Blaze. The oracle truncates logged bodies at ~200 chars, and raising that cap would mean editing the behavioural specification to make it easier to copy -- backwards. A proxy gets the same evidence and leaves the oracle untouched. THE DESIGN RULE: TEE, DO NOT REBUILD. UTAS is plaintext HTTP/1.1 on ThreadingHTTPServer, so keep-alive, pipelining and chunked transfer are all live. A proxy that parses a request and re-emits it can corrupt the traffic it exists to observe -- and that corruption would present as a UTAS bug, pointing the investigation in exactly the wrong direction. So bytes are copied verbatim in both directions and a second copy goes to disk; transactions are reconstructed later, offline, from that copy. A parser bug therefore spoils the record and never the session. Standalone, NOT in the container, so the same tool can later sit in front of a Rust UTAS host and replay an identical captured request against both. Two layers, as with the Blaze captures: raw/*.ofcap is exact bytes at mode 0600 and gitignored; sanitized/transactions.jsonl is the committed artefact. Bodies are preserved EXACTLY and sanitised second -- only known-secret headers and JSON keys are replaced, structure is never reshaped, and every redaction is recorded in the transaction so a reader knows what was touched. Captured per transaction: connection id, sequence, relative and wall time, elapsed ms, method, path, query, HTTP version, headers IN RECEIVED ORDER as pairs (a dict would drop duplicates and ordering), raw body and length for both directions, status, and observed keep-alive. Verified as two independent properties, because they fail differently: transparency (bytes through the proxy identical to bytes direct, Date masked, with the mask asserted to have fired) and fidelity (parsed transactions match what was sent, including a dechunked response and a 300-byte POST body). 5/5 mutations killed, including "record but do not forward", "drop the last byte of every chunk" and "stop redacting". scripts/test-utas-observe.py is committed alongside it: a capture tool nobody can re-verify is not evidence. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
505 lines
18 KiB
Python
Executable File
505 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Record real UTAS traffic between FIFA 17 and the Python oracle.
|
|
|
|
openfut-utas-observe.py record --listen 0.0.0.0:8199 --upstream 127.0.0.1:8099 \
|
|
--session captures/utas/session-001
|
|
openfut-utas-observe.py parse --session captures/utas/session-001
|
|
openfut-utas-observe.py snapshot --upstream 127.0.0.1:8099 --out before/state-manifest.json
|
|
|
|
WHY THIS EXISTS
|
|
---------------
|
|
Porting UTAS needs the exact bytes of real requests AND real responses. The
|
|
oracle's own log truncates bodies at ~200 chars, and raising that cap would mean
|
|
editing the behavioural specification to make it easier to copy -- which is
|
|
exactly backwards. A proxy gets the same evidence and leaves the oracle alone.
|
|
|
|
Standalone on purpose: NOT built into the Python container. The same tool can
|
|
later sit in front of a Rust UTAS host, so the identical captured request can be
|
|
replayed against both and diffed.
|
|
|
|
THE DESIGN RULE: TEE, DO NOT REBUILD
|
|
------------------------------------
|
|
UTAS is plaintext HTTP/1.1 served by `ThreadingHTTPServer`, so keep-alive,
|
|
pipelining and chunked transfer are all possible. A proxy that parses a request
|
|
and re-emits it can corrupt the traffic it exists to observe -- and a subtle
|
|
corruption here would look like a UTAS bug, sending the investigation in exactly
|
|
the wrong direction.
|
|
|
|
So this copies bytes VERBATIM in both directions and writes a second copy to
|
|
disk. Transactions are reconstructed later, offline, from that copy. A bug in
|
|
the parser therefore spoils the record and never the session. This is the same
|
|
reasoning that keeps the oracle unmodified: do not let the measuring apparatus
|
|
change what is being measured.
|
|
|
|
TWO LAYERS, LIKE THE BLAZE CAPTURES
|
|
-----------------------------------
|
|
raw/utas.ofcap exact bytes, mode 0600, gitignored
|
|
|
|
|
sanitizer
|
|
|
|
|
sanitized/transactions.jsonl repository-safe fixtures
|
|
|
|
Raw captures are never normalised. Normalisation happens only when producing the
|
|
committed fixture, and preserves exact bodies first, sanitises second.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
MAGIC = "OFCAP1"
|
|
C2S = "c2s"
|
|
S2C = "s2c"
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────── capture file ────
|
|
class Capture:
|
|
"""Append-only JSONL of raw byte chunks. One line per read().
|
|
|
|
JSONL with base64 rather than a packed binary format: this is a research
|
|
artefact that people will grep and eyeball, and the size cost is irrelevant
|
|
next to being able to inspect it without a decoder. Base64 is an encoding,
|
|
not a normalisation -- the bytes round-trip exactly.
|
|
"""
|
|
|
|
def __init__(self, path):
|
|
self.path = path
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
# 0600 before anything is written: captures carry session tokens.
|
|
fd = os.open(path, os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
|
|
self.f = os.fdopen(fd, "w")
|
|
self.lock = threading.Lock()
|
|
self._write({"magic": MAGIC, "started_unix": time.time()})
|
|
|
|
def _write(self, obj):
|
|
with self.lock:
|
|
self.f.write(json.dumps(obj, sort_keys=True) + "\n")
|
|
self.f.flush()
|
|
|
|
def chunk(self, conn_id, direction, data, t0):
|
|
self._write({
|
|
"conn": conn_id,
|
|
"dir": direction,
|
|
"t": round(time.monotonic() - t0, 6),
|
|
"unix": time.time(),
|
|
"len": len(data),
|
|
"b64": base64.b64encode(data).decode(),
|
|
})
|
|
|
|
def event(self, conn_id, kind, detail, t0):
|
|
self._write({
|
|
"conn": conn_id,
|
|
"event": kind,
|
|
"detail": detail,
|
|
"t": round(time.monotonic() - t0, 6),
|
|
"unix": time.time(),
|
|
})
|
|
|
|
def close(self):
|
|
with self.lock:
|
|
self.f.close()
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────── record ────
|
|
def pump(src, dst, cap, conn_id, direction, t0, done):
|
|
"""Copy src->dst verbatim, teeing every chunk into the capture.
|
|
|
|
Forward FIRST, record second: the client must never wait on our disk.
|
|
"""
|
|
try:
|
|
while True:
|
|
data = src.recv(65536)
|
|
if not data:
|
|
break
|
|
try:
|
|
dst.sendall(data)
|
|
except OSError as e:
|
|
cap.event(conn_id, "forward_failed", "%s %s" % (direction, e), t0)
|
|
break
|
|
cap.chunk(conn_id, direction, data, t0)
|
|
except OSError as e:
|
|
cap.event(conn_id, "read_failed", "%s %s" % (direction, e), t0)
|
|
finally:
|
|
# Half-close so the peer sees EOF in the same direction the original
|
|
# did. Connection semantics are part of what is being captured.
|
|
try:
|
|
dst.shutdown(socket.SHUT_WR)
|
|
except OSError:
|
|
pass
|
|
cap.event(conn_id, "eof", direction, t0)
|
|
done.set()
|
|
|
|
|
|
def handle(client, addr, upstream_addr, cap, conn_id, t0):
|
|
cap.event(conn_id, "open", "%s:%d" % addr, t0)
|
|
try:
|
|
up = socket.create_connection(upstream_addr, timeout=30)
|
|
except OSError as e:
|
|
cap.event(conn_id, "upstream_failed", str(e), t0)
|
|
client.close()
|
|
return
|
|
client.settimeout(None)
|
|
up.settimeout(None)
|
|
|
|
a, b = threading.Event(), threading.Event()
|
|
t1 = threading.Thread(target=pump, args=(client, up, cap, conn_id, C2S, t0, a), daemon=True)
|
|
t2 = threading.Thread(target=pump, args=(up, client, cap, conn_id, S2C, t0, b), daemon=True)
|
|
t1.start(); t2.start()
|
|
t1.join(); t2.join()
|
|
for s in (client, up):
|
|
try:
|
|
s.close()
|
|
except OSError:
|
|
pass
|
|
cap.event(conn_id, "close", "", t0)
|
|
|
|
|
|
def cmd_record(args):
|
|
lh, lp = args.listen.rsplit(":", 1)
|
|
uh, up = args.upstream.rsplit(":", 1)
|
|
session = os.path.abspath(args.session)
|
|
cap = Capture(os.path.join(session, "raw", "utas.ofcap"))
|
|
|
|
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
srv.bind((lh, int(lp)))
|
|
srv.listen(64)
|
|
t0 = time.monotonic()
|
|
print("recording %s -> %s:%s into %s" % (args.listen, uh, up, cap.path), flush=True)
|
|
print(" raw capture is mode 0600 and must stay out of git", flush=True)
|
|
|
|
n = 0
|
|
try:
|
|
while True:
|
|
client, addr = srv.accept()
|
|
n += 1
|
|
threading.Thread(
|
|
target=handle,
|
|
args=(client, addr, (uh, int(up)), cap, n, t0),
|
|
daemon=True,
|
|
).start()
|
|
except KeyboardInterrupt:
|
|
print("\nstopped after %d connection(s)" % n, flush=True)
|
|
finally:
|
|
cap.close()
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────── parsing ────
|
|
def read_capture(path):
|
|
conns = {}
|
|
with open(path) as f:
|
|
for line in f:
|
|
r = json.loads(line)
|
|
if "magic" in r:
|
|
continue
|
|
c = conns.setdefault(r["conn"], {"c2s": bytearray(), "s2c": bytearray(),
|
|
"events": [], "marks": {"c2s": [], "s2c": []}})
|
|
if "b64" in r:
|
|
data = base64.b64decode(r["b64"])
|
|
c["marks"][r["dir"]].append((len(c[r["dir"]]), r["t"], r["unix"]))
|
|
c[r["dir"]].extend(data)
|
|
else:
|
|
c["events"].append(r)
|
|
return conns
|
|
|
|
|
|
def split_messages(buf, is_response):
|
|
"""Split an HTTP/1.1 byte stream into messages. Returns list of (head, body, end).
|
|
|
|
Framing follows RFC 7230: Transfer-Encoding: chunked wins over
|
|
Content-Length. A stream that cannot be framed stops the walk rather than
|
|
guessing -- a wrong split would silently mis-attribute bodies.
|
|
"""
|
|
out, i = [], 0
|
|
while True:
|
|
sep = buf.find(b"\r\n\r\n", i)
|
|
if sep < 0:
|
|
break
|
|
head = bytes(buf[i:sep])
|
|
hs = head.decode("latin1")
|
|
start = sep + 4
|
|
te = re.search(r"(?im)^transfer-encoding:\s*(.+)$", hs)
|
|
cl = re.search(r"(?im)^content-length:\s*(\d+)\s*$", hs)
|
|
if te and "chunked" in te.group(1).lower():
|
|
j, body = start, bytearray()
|
|
while True:
|
|
nl = buf.find(b"\r\n", j)
|
|
if nl < 0:
|
|
return out
|
|
try:
|
|
size = int(bytes(buf[j:nl]).split(b";")[0], 16)
|
|
except ValueError:
|
|
return out
|
|
j = nl + 2
|
|
if size == 0:
|
|
j = buf.find(b"\r\n", j)
|
|
j = (j + 2) if j >= 0 else len(buf)
|
|
break
|
|
body.extend(buf[j:j + size])
|
|
j += size + 2
|
|
out.append((head, bytes(body), j)); i = j
|
|
elif cl:
|
|
n = int(cl.group(1))
|
|
end = start + n
|
|
if end > len(buf):
|
|
break
|
|
out.append((head, bytes(buf[start:end]), end)); i = end
|
|
else:
|
|
status_no_body = is_response and re.match(r"HTTP/\d\.\d (1\d\d|204|304)", hs)
|
|
if is_response and not status_no_body:
|
|
# No framing header: the body runs to end-of-stream.
|
|
out.append((head, bytes(buf[start:]), len(buf))); i = len(buf)
|
|
break
|
|
out.append((head, b"", start)); i = start
|
|
return out
|
|
|
|
|
|
def time_at(marks, offset):
|
|
"""Wall/relative time of the chunk containing byte `offset`."""
|
|
last = (0.0, 0.0)
|
|
for pos, t, unix in marks:
|
|
if pos > offset:
|
|
break
|
|
last = (t, unix)
|
|
return last
|
|
|
|
|
|
def cmd_parse(args):
|
|
session = os.path.abspath(args.session)
|
|
raw = os.path.join(session, "raw", "utas.ofcap")
|
|
conns = read_capture(raw)
|
|
out_dir = os.path.join(session, "sanitized")
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
out_path = os.path.join(out_dir, "transactions.jsonl")
|
|
|
|
total, skipped = 0, 0
|
|
with open(out_path, "w") as out:
|
|
for conn_id in sorted(conns):
|
|
c = conns[conn_id]
|
|
reqs = split_messages(c["c2s"], is_response=False)
|
|
resps = split_messages(c["s2c"], is_response=True)
|
|
if len(reqs) != len(resps):
|
|
# Reported, never silently trimmed: an unmatched count means the
|
|
# framing is not understood, and a quiet truncation would look
|
|
# like the client simply asked for less.
|
|
print(" conn %d: %d requests but %d responses -- pairing the common prefix"
|
|
% (conn_id, len(reqs), len(resps)), file=sys.stderr)
|
|
skipped += abs(len(reqs) - len(resps))
|
|
keepalive = None
|
|
for seq, ((rh, rb, rend), (sh, sb, send)) in enumerate(zip(reqs, resps), 1):
|
|
rl = rh.split(b"\r\n")[0].decode("latin1")
|
|
sl = sh.split(b"\r\n")[0].decode("latin1")
|
|
m = re.match(r"(\S+)\s+(\S+)\s+(HTTP/\d\.\d)", rl)
|
|
method, target, ver = (m.group(1), m.group(2), m.group(3)) if m else ("?", rl, "?")
|
|
path, _, query = target.partition("?")
|
|
st = re.match(r"HTTP/\d\.\d\s+(\d+)", sl)
|
|
req_t, req_unix = time_at(c["marks"]["c2s"], rend - 1)
|
|
res_t, res_unix = time_at(c["marks"]["s2c"], max(send - 1, 0))
|
|
keepalive = len(reqs) > 1
|
|
rec = {
|
|
"conn": conn_id,
|
|
"seq": seq,
|
|
"t_request": req_t,
|
|
"t_response": res_t,
|
|
"elapsed_ms": round((res_t - req_t) * 1000, 3),
|
|
"unix_request": req_unix,
|
|
"request": {
|
|
"method": method, "path": path, "query": query, "version": ver,
|
|
"headers": header_list(rh),
|
|
"body_b64": base64.b64encode(rb).decode(),
|
|
"body_len": len(rb),
|
|
},
|
|
"response": {
|
|
"status": int(st.group(1)) if st else None,
|
|
"status_line": sl,
|
|
"headers": header_list(sh),
|
|
"body_b64": base64.b64encode(sb).decode(),
|
|
"body_len": len(sb),
|
|
},
|
|
"connection": {
|
|
"requests_on_this_connection": len(reqs),
|
|
"keepalive_observed": keepalive,
|
|
},
|
|
}
|
|
out.write(json.dumps(sanitize(rec), sort_keys=True) + "\n")
|
|
total += 1
|
|
|
|
os.chmod(out_path, 0o644)
|
|
print("wrote %s (%d transactions across %d connections%s)"
|
|
% (out_path, total, len(conns),
|
|
"; %d unpaired messages reported above" % skipped if skipped else ""))
|
|
return 0
|
|
|
|
|
|
def header_list(head):
|
|
"""Headers in RECEIVED ORDER, as a list of pairs. Order can be protocol-
|
|
relevant and a dict would silently discard duplicates."""
|
|
lines = head.decode("latin1").split("\r\n")[1:]
|
|
out = []
|
|
for l in lines:
|
|
if not l:
|
|
continue
|
|
k, _, v = l.partition(":")
|
|
out.append([k, v.strip()])
|
|
return out
|
|
|
|
|
|
SECRET_HEADERS = {"authorization", "cookie", "set-cookie", "x-ut-sid", "easw-session-data-nucleus-id"}
|
|
SECRET_BODY_KEYS = ("sid", "token", "password", "answer", "secret")
|
|
|
|
|
|
def sanitize(rec):
|
|
"""Repository-safe form. Bodies are preserved EXACTLY; only credentials go.
|
|
|
|
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 = []
|
|
for side in ("request", "response"):
|
|
hs = rec[side]["headers"]
|
|
for pair in hs:
|
|
if pair[0].lower() in SECRET_HEADERS:
|
|
redacted.append("%s.%s" % (side, pair[0]))
|
|
pair[1] = "<REDACTED>"
|
|
body = base64.b64decode(rec[side]["body_b64"])
|
|
new, hit = redact_json(body)
|
|
if hit:
|
|
redacted.extend("%s.body.%s" % (side, h) for h in hit)
|
|
rec[side]["body_b64"] = base64.b64encode(new).decode()
|
|
rec[side]["body_len_original"] = len(body)
|
|
if redacted:
|
|
rec["redacted"] = sorted(set(redacted))
|
|
return rec
|
|
|
|
|
|
def redact_json(body):
|
|
if not body[:1] in (b"{", b"["):
|
|
return body, []
|
|
try:
|
|
doc = json.loads(body)
|
|
except Exception:
|
|
return body, []
|
|
hits = []
|
|
|
|
def walk(o):
|
|
if isinstance(o, dict):
|
|
for k in list(o):
|
|
if any(s in k.lower() for s in SECRET_BODY_KEYS) and isinstance(o[k], str):
|
|
o[k] = "<REDACTED>"
|
|
hits.append(k)
|
|
else:
|
|
walk(o[k])
|
|
elif isinstance(o, list):
|
|
for v in o:
|
|
walk(v)
|
|
|
|
walk(doc)
|
|
if not hits:
|
|
return body, []
|
|
return json.dumps(doc, sort_keys=True).encode(), hits
|
|
|
|
|
|
# ────────────────────────────────────────────────────────────── snapshot ────
|
|
SNAPSHOT_ROUTES = [
|
|
("account", "/openfut/account/sync"),
|
|
]
|
|
|
|
|
|
def cmd_snapshot(args):
|
|
"""A light state manifest, for before/after pairing around a session.
|
|
|
|
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.
|
|
"""
|
|
uh, up = args.upstream.rsplit(":", 1)
|
|
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)
|
|
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.
|
|
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:
|
|
json.dump(man, f, indent=2, sort_keys=True)
|
|
f.write("\n")
|
|
print("wrote %s" % args.out)
|
|
for k, v in man["routes"].items():
|
|
print(" %s: %s" % (k, v))
|
|
return 0
|
|
|
|
|
|
INTERESTING = ("coins", "unopenedPacks", "personaId", "personaName", "clubName",
|
|
"level", "experience", "accountFunds", "itemCount", "squadId")
|
|
|
|
|
|
def summarize(payload):
|
|
try:
|
|
doc = json.loads(payload)
|
|
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
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
r = sub.add_parser("record")
|
|
r.add_argument("--listen", required=True)
|
|
r.add_argument("--upstream", required=True)
|
|
r.add_argument("--session", required=True)
|
|
r.set_defaults(fn=cmd_record)
|
|
|
|
p = sub.add_parser("parse")
|
|
p.add_argument("--session", required=True)
|
|
p.set_defaults(fn=cmd_parse)
|
|
|
|
s = sub.add_parser("snapshot")
|
|
s.add_argument("--upstream", required=True)
|
|
s.add_argument("--out", required=True)
|
|
s.set_defaults(fn=cmd_snapshot)
|
|
|
|
args = ap.parse_args()
|
|
return args.fn(args) or 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|