arti/vt1200
1
0 Fork 1
mirror of https://git.k-space.ee/arti/vt1200.git synced 2026-10-30 12:03:54 +02:00
Code Wiki Activity
Files
8588340527261e73e85a0363b6b5a02571d4d4a0
vt1200/dwt/loader/hostload.py
T
SDD 8588340527 dwt/loader: end-to-end host loader over rdebug 
hostload.py drives one rdebug connection: jump(malloc) for the COFF buffer,
poke the COFF, poke the loader at 0x01880000, write the descriptor, call the
loader, and verify status 0 plus app evidence (marker cells or xhello stage 6
and a VRAM screenshot). --tftp uses the existing tftpc.bin fetch instead of a
host poke. Adds evidence for two/three fresh boots and appends the integrate
section to dwt/notes/LOADER.md, including the bit-offset result-block layout
and the create_process clobber of the descriptor.
2026-10-18 15:34:25 +03:00

575 lines
23 KiB
Python
Raw Blame History

#!/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("<H", d, off)[0]
def u32(d: bytes, off: int) -> int:
return struct.unpack_from("<I", d, off)[0]
def overlaps(a0: int, a1: int, b0: int, b1: int) -> 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("<I", peek_bytes(rd, bit_addr, 4))[0]
def compare(rd: exploit.RDebug, bit_addr: int, data: bytes) -> 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("<HHIII", 0x4C44, 0x5231, buf, length, 1)
poke_bytes(rd, DESC, desc)
got = read_desc(rd)
if got != desc:
raise RuntimeError(f"descriptor readback mismatch: wrote {desc.hex(' ')}, "
f"read {got.hex(' ')}")
return desc
def read_result(rd: exploit.RDebug) -> 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))
View Git Blame Copy Permalink