mirror of
https://git.k-space.ee/arti/vt1200.git
synced 2026-11-02 12:23:17 +02:00
arti
d211ff0770
Two faults the 60 fps work exposed, both in the delivery tooling: * `demo.sh --no-load` waited for the terminal to answer ping. A loaded payload owns the LANCE and stops the OS's stack with it, so that gate can never pass once the payload is running -- the recovery path documented for "already jumped" deadlocked for its full 180 s. The gate now runs only when this invocation is going to load the payload. * `rdebug_load.py` used a 5 s socket timeout and no retry, so a readback peek that timed out on a freshly booted OS failed a load whose pokes had already been acknowledged. Timeout is 15 s (--timeout) and the readback retries once before it is believed. Tests: the retry path is pinned with a fake remote that times out on the first peek (one timeout, then the successful pass over every chunk) and one that never answers (it must raise rather than loop). 9 tests in harness/badapple pass.
218 lines
8.6 KiB
Python
Executable File
218 lines
8.6 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Deliver a payload image into a running VT1200 over rdebug, then jump to it.
|
|
|
|
This is the real delivery path, not a debugger shortcut: the image is written
|
|
with the firmware's own remote-debug service (``peek``/``poke``/``jump``, TCP
|
|
771 -- ``dwt/notes/CROSS.md``), exactly as ``dwt/exploit.py`` does for its
|
|
one-shot commands. ``harness/mame/vidstream.lua`` remains the *test* path: it
|
|
pokes through MAME's debugger API, which is fine for checking a payload and is
|
|
not how anything reaches the hardware.
|
|
|
|
Two things the protocol forces, both measured in ``harness/badapple/README.md``:
|
|
|
|
* a **512-byte chunk**. The firmware's documented clamp is 1024, but a
|
|
1024-byte poke is mis-parsed -- the chip reads the body's first word as the
|
|
next opcode and answers "Invalid command 0xA5A5A5A5" for a 0xA5-filled
|
|
chunk. 512 is accepted and read back byte for byte.
|
|
* **no probing after the jump**. A non-returning payload takes the remote-debug
|
|
process with it, so the load is verified *before* the jump and observed
|
|
afterwards from the machine itself (MAME's RAM dumps on the emulator, the
|
|
screen on the unit).
|
|
|
|
Usage
|
|
-----
|
|
rdebug_load.py # image + entry from the build
|
|
rdebug_load.py --dry-run # what would be written where
|
|
rdebug_load.py --no-jump # load and verify only
|
|
rdebug_load.py --host 192.168.99.2 --image build/vidstream.bin
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
ROOT = os.path.dirname(os.path.dirname(HERE))
|
|
DWT = os.path.join(ROOT, "dwt")
|
|
|
|
if DWT not in sys.path:
|
|
sys.path.insert(0, DWT)
|
|
|
|
#: Largest poke the firmware actually accepts. See the module docstring.
|
|
CHUNK = 512
|
|
|
|
#: The image's link origin: the .text address `build-stream.sh` links at. Used
|
|
#: only when the environment file is absent, so a hand-built image still loads.
|
|
DEFAULT_LOAD = 0x018C0000
|
|
|
|
#: Socket timeout. Generous on purpose: a freshly booted OS answers rdebug
|
|
#: slowly (measured: a readback peek timed out at 5 s while the machine was still
|
|
#: settling), and a slow peek must not fail a load that already succeeded.
|
|
DEFAULT_TIMEOUT = 15.0
|
|
|
|
|
|
def read_env(path) -> dict:
|
|
"""Parse ``vidstream.env`` (``KEY=VALUE`` lines) into ints where possible."""
|
|
out = {}
|
|
with open(path) as handle:
|
|
for line in handle:
|
|
line = line.strip()
|
|
if not line or "=" not in line:
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
value = value.strip()
|
|
try:
|
|
out[key.strip()] = int(value, 0)
|
|
except ValueError:
|
|
out[key.strip()] = value
|
|
return out
|
|
|
|
|
|
def chunks(data: bytes, size: int = CHUNK) -> list[bytes]:
|
|
"""Split ``data`` into transfers the firmware will accept."""
|
|
if size < 1 or size > CHUNK:
|
|
raise ValueError(
|
|
"a poke chunk must be 1..%d bytes (the firmware mis-parses %d); got %r"
|
|
% (CHUNK, CHUNK * 2, size))
|
|
return [data[off:off + size] for off in range(0, len(data), size)]
|
|
|
|
|
|
def chunk_addr(base: int, index: int, size: int = CHUNK) -> int:
|
|
"""Bit address of chunk ``index``: the TMS34010 addresses bits, so a
|
|
``size``-byte transfer advances the pointer by ``size * 8``."""
|
|
return base + index * size * 8
|
|
|
|
|
|
def plan(image: bytes, load: int, size: int = CHUNK):
|
|
"""The transfers this load will perform, as ``(bit address, bytes)``."""
|
|
return [(chunk_addr(load, n, size), part)
|
|
for n, part in enumerate(chunks(image, size))]
|
|
|
|
|
|
def readback(rd, transfers, retries=1):
|
|
"""Peek every chunk back, retrying a stalled read once.
|
|
|
|
A timeout here is the machine being busy, not a bad image -- the poke has
|
|
already been acknowledged -- so it is retried before it is believed.
|
|
"""
|
|
back = b""
|
|
for attempt in range(retries + 1):
|
|
try:
|
|
return b"".join(rd.peek(addr, len(part)) for addr, part in transfers)
|
|
except (OSError, TimeoutError):
|
|
if attempt == retries:
|
|
raise
|
|
return back
|
|
|
|
|
|
def load_and_jump(image_path, entry, load, do_jump=True, chunk=CHUNK,
|
|
host="192.168.99.2", port=771, verbose=False,
|
|
timeout=DEFAULT_TIMEOUT):
|
|
"""Poke ``image_path`` at ``load``, verify it, then jump to ``entry``."""
|
|
import exploit # dwt/exploit.py: the protocol the firmware speaks
|
|
|
|
data = open(image_path, "rb").read()
|
|
transfers = plan(data, load, chunk)
|
|
rd = exploit.RDebug(host, port, timeout, verbose)
|
|
rd.connect()
|
|
try:
|
|
for addr, part in transfers:
|
|
rd.poke(addr, part)
|
|
print("poked %d bytes in %d chunk(s) at %#x"
|
|
% (len(data), len(transfers), load))
|
|
|
|
back = readback(rd, transfers)
|
|
if back != data:
|
|
bad = next(i for i in range(min(len(back), len(data)))
|
|
if back[i] != data[i])
|
|
print("readback MISMATCH at byte %d: wrote %02x, read %02x"
|
|
% (bad, data[bad], back[bad]), file=sys.stderr)
|
|
return 1
|
|
print("readback OK: the machine holds the image we sent")
|
|
|
|
if do_jump:
|
|
rd.jump(entry, [], wait=False)
|
|
print("jump sent to %#x (the payload never returns)" % entry)
|
|
finally:
|
|
rd.close()
|
|
return 0
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
ap = argparse.ArgumentParser(
|
|
description=__doc__,
|
|
formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("--host", default="192.168.99.2",
|
|
help="terminal address (default %(default)s)")
|
|
ap.add_argument("--port", type=int, default=771,
|
|
help="rdebug TCP port (default %(default)s)")
|
|
ap.add_argument("--image",
|
|
default=os.path.join(ROOT, "harness", "udpstream", "build",
|
|
"vidstream.bin"),
|
|
help="payload image (default: the streaming payload)")
|
|
ap.add_argument("--env",
|
|
default=os.path.join(ROOT, "harness", "udpstream", "build",
|
|
"vidstream.env"),
|
|
help="build environment carrying ENTRY_PTR")
|
|
ap.add_argument("--load", type=lambda s: int(s, 0), default=None,
|
|
help="load bit address (default: %#x)" % DEFAULT_LOAD)
|
|
ap.add_argument("--entry", type=lambda s: int(s, 0), default=None,
|
|
help="entry bit address (default: ENTRY_PTR from --env)")
|
|
ap.add_argument("--chunk", type=int, default=CHUNK,
|
|
help="poke size in bytes (max %d; default %%(default)s)"
|
|
% CHUNK)
|
|
ap.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT,
|
|
help="socket timeout in seconds (default %(default)s)")
|
|
ap.add_argument("--no-jump", action="store_true",
|
|
help="load and verify, do not start it")
|
|
ap.add_argument("--dry-run", action="store_true",
|
|
help="print the transfer plan and exit")
|
|
args = ap.parse_args(argv)
|
|
|
|
try:
|
|
env = read_env(args.env)
|
|
except OSError:
|
|
print("error: no build environment at %s -- build the payload first "
|
|
"(harness/udpstream/build-stream.sh)" % args.env,
|
|
file=sys.stderr)
|
|
return 2
|
|
|
|
load = args.load if args.load is not None else env.get("LOAD_PTR",
|
|
DEFAULT_LOAD)
|
|
entry = args.entry if args.entry is not None else env.get("ENTRY_PTR")
|
|
if entry is None:
|
|
print("error: %s has no ENTRY_PTR (pass --entry)" % args.env,
|
|
file=sys.stderr)
|
|
return 2
|
|
|
|
try:
|
|
image = open(args.image, "rb").read()
|
|
transfers = plan(image, load, args.chunk)
|
|
except (OSError, ValueError) as exc:
|
|
print("error: %s" % exc, file=sys.stderr)
|
|
return 2
|
|
|
|
if args.dry_run:
|
|
print("%s: %d bytes, %d chunk(s) of %d, load %#x, entry %#x"
|
|
% (args.image, len(image), len(transfers), args.chunk,
|
|
load, entry))
|
|
for addr, part in transfers[:3]:
|
|
print(" %#010x %d B" % (addr, len(part)))
|
|
if len(transfers) > 3:
|
|
print(" ... %d more" % (len(transfers) - 3))
|
|
return 0
|
|
|
|
try:
|
|
return load_and_jump(args.image, entry, load,
|
|
do_jump=not args.no_jump, chunk=args.chunk,
|
|
host=args.host, port=args.port,
|
|
timeout=args.timeout)
|
|
except Exception as exc: # noqa: BLE001 - one line, not a traceback
|
|
print("error: %s: %s" % (type(exc).__name__, exc), file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|