From 929b6bc267783c1036be8c611a5386d880d3058d Mon Sep 17 00:00:00 2001 From: Cyd Date: Sat, 4 Jul 2026 23:17:15 -0500 Subject: [PATCH] =?UTF-8?q?Networking:=20add=20egg=5Fcapture.py=20?= =?UTF-8?q?=E2=80=94=20console->pod=20protocol=20capture=20tool?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- emulator/net-tools/egg_capture.py | 99 +++++++++++++++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 emulator/net-tools/egg_capture.py diff --git a/emulator/net-tools/egg_capture.py b/emulator/net-tools/egg_capture.py new file mode 100644 index 0000000..ef9f793 --- /dev/null +++ b/emulator/net-tools/egg_capture.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""Capture + decode the Rel4.10 console -> pod egg-delivery TCP protocol. + +Stands in as a pod on TCP (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()