#!/usr/bin/env python3 """hostload.py -- end-to-end host-side app loader over the stock rdebug service. One driver, one TCP connection to the terminal's `DEBUG TCP Port` (771): rdebug ---- malloc(size) --> bit pointer ---- poke COFF ---> that buffer ---- poke loader -> bit 0x01880000 ---- poke descriptor @ 0x01882C00 ---- jump 0x01880000 -> status in A8 + result block ---- peek markers / screenshot The COFF loader itself (`dwt/loader/loader.asm`, 1220 B) is position-dependent and lives at bit `0x01880000`. This script only supplies its input and reads its result; it does not choose load addresses (the COFF section headers do) and does not patch the ROM. Two transports for the COFF: * default -- poke the file into a `malloc`'d buffer from the host. * --tftp -- poke the existing `dwt/tcpip/tftpc.bin` (bit 0x01870000), jump it, and read the buffer/length it fetched. The TFTP server must be serving the COFF *content* for any requested name, because tftpc.asm's RRQ name is fixed (`dwt/tcpip/probe.bin`): sudo python3 -u dwt/tcpip/serve_probe.py 192.168.99.1 --bootp \ --file dwt/loader/build/marker.coff That server must already be answering BOOTP so the terminal has 192.168.99.2. Addresses are TMS34010 **bit** addresses throughout; `poke`/`peek` convert by 8 bits per byte, and the malloc return is used verbatim as a bit pointer. Exit status: 0 only when the loader returned status 0, the result block is consistent, and the app's own evidence verifies (markers, or X stage + frame). Examples: # marker app: markers at 0x018E0000.. (start/data/data2/bss/counter) python3 dwt/loader/hostload.py --host 192.168.99.2 \ --coff dwt/loader/build/marker.coff --marker \ --out dwt/loader/evidence/hostload --tag marker # xhello: X window, verified by stage==6 and a VRAM screenshot python3 dwt/loader/hostload.py --host 192.168.99.2 \ --coff dwt/loader/build/xhello.coff \ --out dwt/loader/evidence/hostload --tag xhello # self-contained fetch: no COFF poke, tftpc fetches it into malloc memory python3 dwt/loader/hostload.py --host 192.168.99.2 --tftp \ --coff dwt/loader/build/marker.coff --marker \ --tftp-log dwt/loader/evidence/hostload/serve_tftp_marker.log \ --tag tftp_marker """ from __future__ import annotations import argparse import hashlib import os import struct import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) # dwt/loader DWT = os.path.dirname(HERE) # dwt ROOT = os.path.dirname(DWT) # repo root sys.path.insert(0, DWT) import exploit # noqa: E402 # --- fixed loader contract (dwt/notes/LOADER_CROSS.md) --------------------- LOADER_BASE = 0x01880000 LOADER_PAGE_END = 0x01883200 DESC = 0x01882C00 RESULT = 0x01882D00 MALLOC = 0x3F701230 RAM_BITS = 0x2000000 # 4 MiB, in bits # Result-block cell offsets. These are BIT offsets (loader.asm uses # `RESULT_BASE+020h`, and the TMS34010 assembler's addresses are bit # addresses), so 0x020 bits is byte +4, not +0x20 bytes. The block is 13 # adjacent 32-bit cells, 52 bytes total. RESULT_CELLS = { 0x000: "status", 0x020: "entry", 0x040: "text_addr", 0x060: "text_size", 0x080: "data_addr", 0x0A0: "data_size", 0x0C0: "bss_addr", 0x0E0: "bss_size", 0x100: "relocs", 0x120: "pcb", 0x140: "nscns", 0x160: "coff_len", 0x180: "loader_base", } RESULT_BYTES = 0x34 # 0x180 bits + 4 bytes STATUS_NAMES = { 0: "OK", 1: "bad descriptor", 2: "truncated COFF / section headers", 3: "bad COFF magic", 4: "section size not whole words", 5: "target outside RAM", 6: "target overlaps loader page or COFF buffer", 7: "relocation records present (unsupported)", 8: "target not in a free-list block", 9: "no entry point", 10: "create_process returned 0", } # tftpc.asm markers (dwt/tcpip/tftpc.asm) TFTPC_BASE = 0x01870000 TFTP_MARK = 0x0187C000 TFTP_DONE = 0x05A5A5A5 # app marker cells MARKER_CELLS = [ (0x018E0000, "start"), (0x018E0020, "data"), (0x018E0040, "data2"), (0x018E0060, "bss"), (0x018E0080, "count"), ] XHELLO_STAGE = 0x01890140 XHELLO_STRLEN = 0x01890160 # --------------------------------------------------------------------------- # small helpers # --------------------------------------------------------------------------- def parse_int(text: str) -> int: text = text.strip().replace("_", "") if text.lower().startswith("0x"): return int(text, 16) if any(c in "abcdefABCDEF" for c in text): return int(text, 16) return int(text, 10) def u16(d: bytes, off: int) -> int: return struct.unpack_from(" int: return struct.unpack_from(" bool: """Half-open ranges [a0,a1) and [b0,b1) in the same unit.""" return a0 < b1 and b0 < a1 def parse_coff(data: bytes) -> dict: """Minimal TI COFF header/section reader (same fields as coff.py).""" if len(data) < 20: raise ValueError("shorter than a COFF file header") magic, nscns = u16(data, 0), u16(data, 2) opthdr = u16(data, 16) if magic != 0x0090: raise ValueError(f"COFF magic {magic:#06x}, expected 0x0090") if 20 + opthdr > len(data): raise ValueError("optional header past EOF") entry = textstart = None if opthdr: entry = u32(data, 20 + 16) textstart = u32(data, 20 + 20) sections = [] base = 20 + opthdr for i in range(nscns): off = base + i * 40 if off + 40 > len(data): raise ValueError(f"section header {i} past EOF") name = data[off:off + 8].rstrip(b"\0").decode("latin-1") paddr = u32(data, off + 8) size = u32(data, off + 16) scnptr = u32(data, off + 20) nreloc = u16(data, off + 32) flags = u32(data, off + 36) sections.append(dict(name=name, paddr=paddr, size=size, scnptr=scnptr, nreloc=nreloc, flags=flags)) return dict(magic=magic, nscns=nscns, opthdr=opthdr, entry=entry or 0, textstart=textstart or 0, sections=sections) def poke_bytes(rd: exploit.RDebug, bit_addr: int, data: bytes, chunk: int = exploit.MAX_XFER) -> None: for off in range(0, len(data), chunk): rd.poke(bit_addr + off * 8, data[off:off + chunk]) def peek_bytes(rd: exploit.RDebug, bit_addr: int, length: int) -> bytes: out = bytearray() while len(out) < length: n = min(exploit.MAX_XFER, length - len(out)) out += rd.peek(bit_addr + len(out) * 8, n) return bytes(out) def peek_u32(rd: exploit.RDebug, bit_addr: int) -> int: return struct.unpack(" tuple[int, int]: """Peek *data* back and return (words, differing bytes).""" got = peek_bytes(rd, bit_addr, len(data)) bad = sum(1 for a, b in zip(got, data) if a != b) return (len(data) + 1) // 2, bad def check_buffer(ptr: int, nbytes: int, coff: dict) -> list[str]: """Sanity-check a malloc'd bit pointer. Returns a list of problems.""" problems = [] end = ptr + nbytes * 8 if ptr == 0: problems.append("malloc returned 0") return problems if ptr % 8: problems.append(f"pointer {ptr:#010x} is not byte-aligned " f"(byte offset {ptr % 8})") if ptr < 0x100 or end > RAM_BITS: problems.append(f"range {ptr:#010x}..{end:#010x} is outside RAM") if overlaps(ptr, end, LOADER_BASE, LOADER_PAGE_END): problems.append("buffer overlaps the loader page") for sec in coff["sections"]: if sec["size"] and overlaps(ptr, end, sec["paddr"], sec["paddr"] + sec["size"]): problems.append(f"buffer overlaps section {sec['name']} " f"at {sec['paddr']:#010x}") for cell, name in MARKER_CELLS: if overlaps(ptr, end, cell, cell + 0x20): problems.append(f"buffer overlaps marker cell {name}") if overlaps(ptr, end, 0x01890100, 0x01890170): problems.append("buffer overlaps the xhello marker cells") return problems def describe_pointer(ptr: int, nbytes: int) -> str: return (f"bit {ptr:#010x} = byte {ptr >> 3:#08x}, {nbytes} bytes, " f"bit range {ptr:#010x}..{ptr + nbytes * 8:#010x}") # --------------------------------------------------------------------------- # screenshot / frame helpers # --------------------------------------------------------------------------- def screenshot_vram(rd: exploit.RDebug, out_png: str, out_raw: str | None = None) -> bytes: """Grab the 1024x1024 VRAM canvas using dwt/screenshot.py primitives.""" import screenshot # imported here so marker-only runs need no PNG path data = screenshot.grab(rd, 0, screenshot.HEIGHT, True) gray = screenshot.decode(data) screenshot.write_png(out_png, screenshot.WIDTH, screenshot.HEIGHT, gray) if out_raw: with open(out_raw, "wb") as fh: fh.write(data) return data def wait_for_frame(path: str, timeout: float) -> bool: end = time.monotonic() + timeout while time.monotonic() < end: if os.path.exists(path) and os.path.getsize(path) > 0: return True time.sleep(0.5) return os.path.exists(path) and os.path.getsize(path) > 0 # --------------------------------------------------------------------------- # the loader sequence # --------------------------------------------------------------------------- def read_desc(rd: exploit.RDebug) -> bytes: return peek_bytes(rd, DESC, 16) def write_desc(rd: exploit.RDebug, buf: int, length: int) -> bytes: desc = struct.pack(" dict: raw = peek_bytes(rd, RESULT, RESULT_BYTES) # RESULT_CELLS keys are bit offsets; the peeked buffer is bytes. return {name: u32(raw, off // 8) for off, name in RESULT_CELLS.items()} def run_loader(rd: exploit.RDebug, buf: int, coff_len: int, jump_wait: float) -> tuple[int, dict]: # Rewritten immediately before the call: create_process uses ARG_AREA # (0x01882B00) and its 0x100-bit call frame lands on the descriptor, so a # descriptor left from a previous run is not trustworthy. desc = write_desc(rd, buf, coff_len) print(f"[.] descriptor @ {DESC:#010x}: {desc.hex(' ')} " f"(magic 4C44/5231, buf={buf:#010x}, len={coff_len}, mode=1)") res = rd.jump(LOADER_BASE, [], wait=True, wait_timeout=jump_wait) if res is None: raise RuntimeError(f"loader at {LOADER_BASE:#010x} did not return within " f"{jump_wait}s") status = res[0] result = read_result(rd) print(f"[.] loader A8={status:#x} ({STATUS_NAMES.get(status, 'unknown')})") print("[.] result block @ {:#x} (cell offsets are bit offsets):".format( RESULT)) for off, name in RESULT_CELLS.items(): print(f" +{off:#05x} {name:<12} {result[name]:#010x}") problems = [] if status != 0: problems.append(f"loader status {status} ({STATUS_NAMES.get(status, '?')})") if result["status"] != status: problems.append(f"result status {result['status']} != A8 {status}") if result["loader_base"] != LOADER_BASE: problems.append(f"result loader_base {result['loader_base']:#x} != " f"{LOADER_BASE:#x}") if result["pcb"] == 0: problems.append("create_process returned a null PCB") if result["coff_len"] != coff_len: problems.append(f"result coff_len {result['coff_len']} != {coff_len}") if problems: raise RuntimeError("; ".join(problems)) return status, result def verify_markers(rd: exploit.RDebug, result: dict, settle: float) -> list[str]: problems = [] vals = {name: peek_u32(rd, cell) for cell, name in MARKER_CELLS} time.sleep(settle) vals2 = {name: peek_u32(rd, cell) for cell, name in MARKER_CELLS} for name, want in (("start", result["entry"]), ("data", 0xDEADBEEF), ("data2", 0x12345678), ("bss", 0)): got = vals[name] print(f"[.] marker {name:<6} {got:#010x} (want {want:#010x})") if got != want: problems.append(f"marker {name}={got:#x}, want {want:#x}") c0, c1 = vals["count"], vals2["count"] print(f"[.] marker count {c0} -> {c1} (child scheduled)") if c1 == c0: problems.append("marker counter did not advance") return problems def verify_xhello(rd: exploit.RDebug, out_dir: str, tag: str, settle: float, take_shot: bool) -> list[str]: problems = [] time.sleep(settle) stage = peek_u32(rd, XHELLO_STAGE) strlen = peek_u32(rd, XHELLO_STRLEN) print(f"[.] xhello stage={stage} strlen={strlen}") if stage != 6: problems.append(f"xhello stage={stage}, want 6 (XSync done)") if strlen != 4: problems.append(f"xhello strlen={strlen}, want 4 " f"(calling-convention check)") if take_shot: png = os.path.join(out_dir, f"hostload_{tag}_xhello.png") data = screenshot_vram(rd, png) print(f"[+] screenshot {png} ({os.path.getsize(png)} bytes, " f"{len(data)} bytes of VRAM)") return problems def tftp_fetch(rd: exploit.RDebug, tftpc_path: str, coff_len: int, jump_wait: float) -> tuple[int, int]: tftpc = open(tftpc_path, "rb").read() poke_bytes(rd, TFTPC_BASE, tftpc) words, bad = compare(rd, TFTPC_BASE, tftpc) print(f"[.] tftpc poked {len(tftpc)} B at {TFTPC_BASE:#010x} " f"({words} words, {bad} differing)") if bad: raise RuntimeError("tftpc readback mismatch") res = rd.jump(TFTPC_BASE, [], wait=True, wait_timeout=jump_wait) if res is None: raise RuntimeError(f"tftpc at {TFTPC_BASE:#010x} did not return within " f"{jump_wait}s (server running? file served?)") marks = {name: peek_u32(rd, TFTP_MARK + off) for off, name in ((0x00, "stage"), (0x20, "fd"), (0x40, "buf"), (0x60, "total"), (0x80, "last"), (0xA0, "block"), (0xC0, "done"))} print("[.] tftpc markers: " + " ".join( f"{k}={v:#x}" if k in ("buf", "done", "block") else f"{k}={v}" for k, v in marks.items())) if marks["stage"] != 4 or marks["done"] != TFTP_DONE: raise RuntimeError("tftpc did not complete (stage/done marker wrong)") if marks["total"] != coff_len: raise RuntimeError(f"tftpc fetched {marks['total']} bytes, " f"COFF is {coff_len}") return marks["buf"], marks["total"] def print_tftp_rrq(path: str | None) -> None: if not path or not os.path.exists(path): return lines = [ln.rstrip() for ln in open(path, errors="replace") if "RRQ" in ln] if lines: print(f"[.] server RRQ ({path}): {lines[-1]}") # --------------------------------------------------------------------------- # main # --------------------------------------------------------------------------- def main(argv: list[str]) -> int: ap = argparse.ArgumentParser( description="End-to-end rdebug app loader: malloc -> poke COFF -> " "COFF loader @0x01880000 -> create_process -> verify.", formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--host", default="192.168.99.2") ap.add_argument("--port", type=int, default=771) ap.add_argument("--timeout", type=float, default=10.0) ap.add_argument("--coff", required=True, help="linked GSPLNK COFF image") ap.add_argument("--loader", default=os.path.join(HERE, "build", "loader.bin"), help="raw loader image (default dwt/loader/build/loader.bin)") ap.add_argument("--no-poke-loader", action="store_true", help="do not write the loader; verify it is already resident") ap.add_argument("--marker", action="store_true", help="verify marker.coff evidence cells (otherwise inferred " "from the file name)") ap.add_argument("--tftp", action="store_true", help="fetch the COFF with dwt/tcpip/tftpc.bin instead of " "poking it") ap.add_argument("--tftpc", default=os.path.join(ROOT, "dwt", "tcpip", "build", "tftpc.bin"), help="raw TFTP client image (default dwt/tcpip/build/tftpc.bin)") ap.add_argument("--tftp-log", default=None, help="server log to quote the RRQ line from (optional)") ap.add_argument("--frame", type=parse_int, default=None, metavar="N", help="wait for MAME's VT1200_DUMP_FRAME=N PPM (boot-relative; " "the dump must be configured on the MAME side)") ap.add_argument("--dump-dir", default=os.getcwd(), help="directory searched for vt1200-frame-N.ppm") ap.add_argument("--frame-timeout", type=float, default=120.0) ap.add_argument("--out", default=os.path.join(HERE, "evidence", "hostload"), help="evidence directory (default dwt/loader/evidence/hostload)") ap.add_argument("--tag", default=None, help="evidence file tag (default: COFF basename)") ap.add_argument("--settle", type=float, default=1.5, help="seconds to wait for the child/window after the load") ap.add_argument("--no-screenshot", action="store_true", help="do not grab VRAM for xhello") ap.add_argument("--jump-wait", type=float, default=20.0, help="seconds to wait for the loader call to return") ap.add_argument("--verbose", action="store_true") args = ap.parse_args(argv[1:]) coff_path = os.path.abspath(args.coff) coff = open(coff_path, "rb").read() try: info = parse_coff(coff) except ValueError as exc: print(f"error: {coff_path}: {exc}", file=sys.stderr) return 2 tag = args.tag or os.path.splitext(os.path.basename(coff_path))[0] os.makedirs(args.out, exist_ok=True) marker_mode = args.marker or "marker" in os.path.basename(coff_path).lower() xhello_mode = "xhello" in os.path.basename(coff_path).lower() print(f"[.] COFF {coff_path}: {len(coff)} B, {info['nscns']} sections, " f"entry={info['entry'] or info['textstart']:#010x} " f"sha256={hashlib.sha256(coff).hexdigest()}") for sec in info["sections"]: print(f" {sec['name']:<8} paddr={sec['paddr']:#010x} " f"size={sec['size']} bits ({sec['size'] // 8} B) " f"nreloc={sec['nreloc']}") rd = exploit.RDebug(args.host, args.port, args.timeout, args.verbose) rc = 1 try: rd.connect() flag, payload = rd.liveness(7) print(f"[.] connected {args.host}:{args.port}; opcode 7 -> " f"flag={flag:#x} {payload!r}") # 1. allocate the COFF buffer (or fetch it with the TFTP client) if args.tftp: buf, fetched = tftp_fetch(rd, args.tftpc, len(coff), args.jump_wait) print_tftp_rrq(args.tftp_log) problems = check_buffer(buf, fetched, info) if problems: raise RuntimeError("unsafe TFTP buffer: " + "; ".join(problems)) got = peek_bytes(rd, buf, fetched) if got != coff: diff = next((i for i, (a, b) in enumerate(zip(got, coff)) if a != b), None) raise RuntimeError(f"TFTP buffer differs from {coff_path} " f"(first mismatch at {diff})") print(f"[+] TFTP fetch verified byte-for-byte: {fetched} B at " f"{buf:#010x} (sha256={hashlib.sha256(got).hexdigest()})") else: res = rd.jump(MALLOC, [len(coff)], wait=True, wait_timeout=args.jump_wait) if res is None: raise RuntimeError("jump(malloc) did not return") buf = res[0] print(f"[+] jump(malloc) -> {buf:#010x}; " f"{describe_pointer(buf, len(coff))}") problems = check_buffer(buf, len(coff), info) if problems: raise RuntimeError("unsafe malloc buffer: " + "; ".join(problems)) # 2. poke the COFF and prove the readback poke_bytes(rd, buf, coff) words, bad = compare(rd, buf, coff) print(f"[+] COFF poked {len(coff)} B at {buf:#010x} " f"({words} words, {bad} differing bytes)") if bad: raise RuntimeError("COFF readback mismatch") first = min(exploit.MAX_XFER, len(coff)) print(f" readback[0:{first}] {peek_bytes(rd, buf, first).hex(' ')}") # 3. loader resident? loader = open(args.loader, "rb").read() if args.no_poke_loader: words, bad = compare(rd, LOADER_BASE, loader) print(f"[.] --no-poke-loader: {args.loader} readback " f"({words} words, {bad} differing)") if bad: raise RuntimeError("loader is not resident at " f"{LOADER_BASE:#010x}") else: poke_bytes(rd, LOADER_BASE, loader) words, bad = compare(rd, LOADER_BASE, loader) print(f"[+] loader poked {len(loader)} B at {LOADER_BASE:#010x} " f"({words} words, {bad} differing)") if bad: raise RuntimeError("loader readback mismatch") # 4. descriptor + call + result block status, result = run_loader(rd, buf, len(coff), args.jump_wait) # 5. app evidence if marker_mode: problems = verify_markers(rd, result, args.settle) elif xhello_mode: problems = verify_xhello(rd, args.out, tag, args.settle, not args.no_screenshot) else: problems = [] print("[.] no app-specific verifier (name is neither marker nor " "xhello); status/result only") if problems: raise RuntimeError("app verification failed: " + "; ".join(problems)) # 6. optional MAME frame dump (boot-relative, externally configured) if args.frame is not None: ppm = os.path.join(args.dump_dir, f"vt1200-frame-{args.frame}.ppm") if wait_for_frame(ppm, args.frame_timeout): print(f"[+] frame dump {ppm} " f"({os.path.getsize(ppm)} bytes)") else: raise RuntimeError(f"no frame dump {ppm} within " f"{args.frame_timeout}s") print(f"[+] PASS: status=0 pcb={result['pcb']:#010x} " f"entry={result['entry']:#010x} ({tag})") rc = 0 except (OSError, exploit.ProtocolError, RuntimeError, ValueError) as exc: print(f"[-] FAIL: {exc}", file=sys.stderr) rc = 2 finally: rd.close() return rc if __name__ == "__main__": raise SystemExit(main(sys.argv))