CydandClaude Fable 5
929b6bc267
Listens on TCP 1501 as a stand-in pod, logs the console's egg-delivery bytes, and auto-decodes each NetworkPacket in both byte orders to pin the framing + PPC/x86 endianness (the two open caveats). Works whether the console reaches it via a slirp gateway redirect or a bridged segment. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
100 lines
3.9 KiB
Python
100 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Capture + decode the Rel4.10 console -> pod egg-delivery TCP protocol.
|
|
|
|
Stands in as a pod on TCP <port> (default 1501, the port from Console.ini).
|
|
Logs every byte the console sends, saves the raw stream to a file, and
|
|
parses each accumulated frame as a NetworkPacket in BOTH byte orders to
|
|
auto-detect the console's endianness (Power Mac console = big-endian; the
|
|
real pods are little-endian x86, so one side byte-swaps -- this tool tells
|
|
us which). Resolves the two open caveats in NET-NOTES.md (stream framing +
|
|
endianness) the moment the console connects.
|
|
|
|
Wire layout (decoded from CODE/RP/MUNGA headers):
|
|
NetworkPacketHeader (16B): clientID, gameID, fromHost, timeStamp (4x i32)
|
|
Receiver::Message (12B): messageLength, messageID, messageFlags
|
|
ReceiveEggFileMessage : +seq(i32) +notationFileLength +thisMessageLength
|
|
+notationData[1000] (messageLength should=1024)
|
|
|
|
Usage: python egg_capture.py [port]
|
|
Point the console's target cockpit (e.g. "Puck") addressIP at this host,
|
|
or bridge to it, then have the console send a mission.
|
|
"""
|
|
import socket, struct, sys, time
|
|
|
|
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 1501
|
|
RAW = f"egg_capture_{int(time.time())}.bin"
|
|
|
|
def parse(buf, bo):
|
|
if len(buf) < 28:
|
|
return None
|
|
clientID, gameID, fromHost, ts = struct.unpack(bo + "iiii", buf[0:16])
|
|
mlen, mid, mflags = struct.unpack(bo + "Iii", buf[16:28])
|
|
d = dict(clientID=clientID, gameID=gameID, fromHost=fromHost, ts=ts,
|
|
msgLen=mlen, msgID=mid, msgFlags=mflags)
|
|
if len(buf) >= 40:
|
|
seq, flen, tlen = struct.unpack(bo + "iii", buf[28:40])
|
|
d.update(seq=seq, fileLen=flen, thisLen=tlen)
|
|
return d
|
|
|
|
def score(d):
|
|
"""Heuristic: how 'sane' this decode looks -> picks the real endianness."""
|
|
s = 0
|
|
if d.get("msgLen") == 1024: s += 5 # egg message is 1024 bytes
|
|
if 0 <= d.get("clientID", -1) <= 6: s += 2 # valid ClientID enum
|
|
if 0 <= d.get("msgID", -1) < 100: s += 1
|
|
if 0 <= d.get("seq", -1) < 100000: s += 2
|
|
if 0 < d.get("fileLen", 0) < 10_000_000: s += 3
|
|
if 0 <= d.get("thisLen", -1) <= 1000: s += 2
|
|
return s
|
|
|
|
def hexdump(b, n=80):
|
|
out = []
|
|
for i in range(0, min(len(b), n), 16):
|
|
c = b[i:i+16]
|
|
hx = " ".join(f"{x:02X}" for x in c)
|
|
asc = "".join(chr(x) if 32 <= x < 127 else "." for x in c)
|
|
out.append(f" {i:04X} {hx:<48} {asc}")
|
|
if len(b) > n:
|
|
out.append(f" ... (+{len(b)-n} more bytes)")
|
|
return "\n".join(out)
|
|
|
|
def main():
|
|
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
srv.bind(("0.0.0.0", PORT))
|
|
srv.listen(1)
|
|
print(f"egg_capture: listening on 0.0.0.0:{PORT} (raw -> {RAW})")
|
|
raw = open(RAW, "wb")
|
|
while True:
|
|
conn, addr = srv.accept()
|
|
print(f"\n=== connection from {addr} ===")
|
|
total = b""
|
|
conn.settimeout(15)
|
|
try:
|
|
while True:
|
|
chunk = conn.recv(4096)
|
|
if not chunk:
|
|
break
|
|
raw.write(chunk); raw.flush()
|
|
total += chunk
|
|
print(f"\nrecv {len(chunk)}B (total {len(total)})")
|
|
print(hexdump(total))
|
|
best = None
|
|
for bo, name in ((">", "BIG/PPC"), ("<", "little")):
|
|
d = parse(total, bo)
|
|
if d:
|
|
sc = score(d)
|
|
print(f" [{name:8}] score={sc} {d}")
|
|
if best is None or sc > best[0]:
|
|
best = (sc, name, d)
|
|
if best and best[0] >= 8:
|
|
print(f" >>> endianness looks {best[1]} "
|
|
f"(score {best[0]})")
|
|
except socket.timeout:
|
|
print("(idle 15s -- console waiting for an ACK reply?)")
|
|
print(f"=== closed; {len(total)} bytes captured to {RAW} ===")
|
|
conn.close()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|