vt1200/harness/badapple/rdebug_load.py

#!/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


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 load_and_jump(image_path, entry, load, do_jump=True, chunk=CHUNK,
                  host="192.168.99.2", port=771, verbose=False):
    """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, 5.0, 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 = b"".join(rd.peek(addr, len(part)) for addr, part in 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("--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)
    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())