path: root/psp_emu/harness/mailbox.py

#!/usr/bin/env python3
"""Host-side mailbox client for the emulated PSP: submit commands, read replies.

This is the piece that turns a booted-but-idle PSP into one that *serves the
host*, which is the surface every host-reachable bug lives on.

Protocol (read off `bl_legacy_mailbox_loop` @VA 0x2025EC, the loop the part
actually runs -- hardware-confirmed in ATTACK-SURFACE.md, and confirmed again
here by observing which entries sOS passes to SVC 0x51):

    header  = MMIO 0x3010A00          ( *(0x2060E4) + 2560 )
    word0   : bit31      = 'done' -- firmware sets it when the reply is ready,
                                     and SKIPS the command if it is already set
              bits 16-19 = command id
              bits 0-15  = status on reply
    +0x14/+0x18/+0x1C    = ring parameters for the ring-init commands

    command ids: 1 ring-1 init, 2 ring-2 init, 3/0xB/0xC teardown, 4 query,
                 5/6 set flag, 7 SYS_DRV info, 8/0xA SYS_DRV RPC

The loop blocks in `SVC 0x5B` on resources registered by `bl_tee_init` via
`svc_59_obj_register(80/81/83)`, so the doorbell is simply IRQ **80**: the FIQ
handler calls `sos_release_irq_objects(80)`, which releases the resource and
makes the task runnable.
"""
import os, sys, time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

MBOX = 0x03010A00          # 32-byte legacy mailbox command header (MMIO, RAM-backed)
MBOX_LEN = 0x20
DOORBELL_IRQ = 80          # bl_tee_init registers 80/81/83; 80 is the command doorbell
RING_STATUS = 0x03010544   # bl_ring_wait_doorbell writes 0x80000000 here when ready

DONE = 0x80000000


class Mailbox:
    def __init__(self, comp, irqs, run_slice=300_000):
        self.c = comp
        self.irqs = irqs
        self.run_slice = run_slice
        self.log = []

    # ------------------------------------------------------------ raw access
    def rd(self, n=MBOX_LEN):
        return bytes(self.c.uc.mem_read(MBOX, n))

    def wr(self, data, off=0):
        self.c.uc.mem_write(MBOX + off, bytes(data))

    def word0(self):
        return int.from_bytes(self.c.uc.mem_read(MBOX, 4), "little")

    # ------------------------------------------------------------- submitting
    def submit(self, cmd_id, params=b"", timeout=3.0, runner=None):
        """Post a command and wait for the firmware to set the done bit.

        `params` is written at +4. Returns (status, raw_header) where status is
        the low 16 bits of word0, or (None, raw) on timeout.
        """
        hdr = bytearray(MBOX_LEN)
        hdr[0:4] = ((cmd_id & 0xF) << 16).to_bytes(4, "little")   # done bit CLEAR
        if params:
            hdr[4:4 + len(params)] = params
        self.wr(hdr)
        self.irqs.raise_irq(DOORBELL_IRQ)

        for _ in range(int(timeout)):
            runner(self.run_slice)
            w = self.word0()
            if w & DONE:
                st = w & 0xFFFF
                self.log.append((cmd_id, st))
                return st, self.rd()
        self.log.append((cmd_id, None))
        return None, self.rd()


def install_debug_gpio(c, lo=0x1000000, hi=0x3000000):
    """Model the debug-GPIO pulse handshake the TEE dispatcher depends on.

    `tee_cmd_dispatch` calls `bl_debug_gpio_pulse` BEFORE it even looks at the
    command. That routine writes bit12 to SMN register 0x3898 and then spins
    while 0x389C bit12 is clear -- a pulse-request / pulse-done register pair.
    Backed by plain RAM the request bit sticks and 'done' never arrives, so the
    very first ring command hangs the dispatcher and every later one is never
    drained. Completing the handshake is what takes the ring from 1 dispatched
    command to many.

    Implemented by watching for the request write rather than hard-coding a
    mapped address, because the register is reached through a VA window whose
    slot is assigned at runtime.
    """
    from unicorn import UC_HOOK_MEM_WRITE
    seen = {}

    # Match the REGISTER, not just the value. An earlier version triggered on
    # any 4-byte write of 0x1000 anywhere in the aperture and corrupted
    # unrelated state -- mailbox commands that had been working started
    # wedging. SMN 0x3898 is reached through a 1 MB-granular window, so the
    # offset within the page is preserved: low 12 bits == 0x898.
    def on_w(uc, access, address, size, value, user):
        if size == 4 and value == 0x1000 and (address & 0xFFF) == 0x898:
            seen.setdefault("req", address)
            uc.mem_write(address + 4, (0x1000).to_bytes(4, "little"))

    c.uc.hook_add(UC_HOOK_MEM_WRITE, on_w, begin=lo, end=hi)
    return seen


class Ring:
    """Host-side command ring: the path that reaches tee_cmd_dispatch.

    Frame is 64 bytes at ring_va + read_idx*4; +0/+4 hold the PA of a 0x400
    command buffer whose **offset +8 is the command id** (1/4 load TA,
    2 unload, 3 TA command, 5.. forwarded to SYS_DRV). `write_idx` is a DWORD
    index (multiple of 16) published at mailbox header +0x0C.
    """

    # The firmware does NOT read the ring at the physical address the host
    # supplied. tee_ring_init maps it and the dispatcher reads through the
    # resulting guest VA, which resolves to a DIFFERENT physical page: with a
    # host ring at PA 0x4800000 the firmware's ring VA 0x4000000 resolved to
    # PA 0x8800000 (the Syshub aperture adds 0x4000000). Writing frames at the
    # host PA therefore hands the firmware a page of zeros -- it dispatched, read
    # an all-zero frame, and tried to map physical address 0, which lands in the
    # reserved [sos_dram_window_base, +8MB) window and fails with 10.
    #
    # So: resolve the firmware's ring VA through the guest page tables and write
    # frames where it will actually look. Buffers referenced BY a frame are
    # mapped by the firmware the same way, so they are mirrored at both the host
    # PA and the aliased view.
    ALIAS = 0x04000000

    slice_blocks = 300_000

    def __init__(self, comp, irqs, ring_pa, ring_size, cmdbuf_pa, runner,
                 ring_va=0x4000000, pt=None):
        self.c, self.irqs = comp, irqs
        self.ring_pa, self.size = ring_pa, ring_size
        self.cmdbuf = cmdbuf_pa
        self.run = runner
        self.ndw = ring_size >> 2
        self.widx = 0
        if pt is None:
            from pspemu import armmmu
            pt = armmmu.PageTableEditor(comp.uc)
        self.pt = pt
        _k, self.ring_phys = pt.describe(ring_va)
        if self.ring_phys is None:
            raise RuntimeError("firmware ring VA %#x is not mapped" % ring_va)

    def _mirror(self, pa, data):
        """Write a buffer where the host put it AND where the firmware maps it."""
        self.c.uc.mem_write(pa, data)
        try:
            self.c.uc.mem_write(pa + self.ALIAS, data)
        except Exception:
            pass

    RESP_OFF = 0x360          # psp_gfx_cmd_resp.resp.status, verified on hardware

    def response(self):
        """Read the firmware's response status for the last submitted command.

        IMPORTANT: the firmware writes the response into the ALIAS copy of the
        command buffer (`cmdbuf + ALIAS`), not into `cmdbuf` itself. Reading
        `cmdbuf` gives a stale 0 for every command, which looks exactly like a
        flat oracle and silently invalidates any status comparison -- an earlier
        session concluded "full_system's ring response path is unfaithful" on
        that basis, and it was the readback that was wrong, not the path.

        Verified against hardware semantics: an unknown command id returns
        0x100 here, the same value an RX 7900 XTX returns
        (HW-VALIDATION-RX7900XTX.md).
        """
        import struct
        raw = bytes(self.c.uc.mem_read(self.cmdbuf + self.ALIAS + self.RESP_OFF, 8))
        status, session = struct.unpack("<2I", raw)
        return status, session

    # TA invoke ABI note, learned the hard way: for GFX_CMD_ID_LOAD_TA the
    # cmd_buf_len you pass becomes params[0].size at the TA's invoke entry, and
    # the GP TAs REJECT anything <= 0x2817 (HDCP returns 0x10 -> ring status
    # 0xffff0010). A 0x1000 command buffer therefore looks exactly like "the
    # emulator cannot invoke TAs" while the invoke path is perfectly fine.
    # Use >= 0x4000.
    TA_CMDBUF_MIN = 0x2818

    def submit(self, cmd_id, buf_extra=b"", timeout=5.0, probe=None):
        import struct
        buf = bytearray(0x400)
        buf[0:4] = struct.pack("<I", 0x400)       # buf_size
        buf[4:8] = struct.pack("<I", 1)           # buf_version
        buf[8:12] = struct.pack("<I", cmd_id)     # command id
        if buf_extra:
            buf[0x1C:0x1C + len(buf_extra)] = buf_extra
        self._mirror(self.cmdbuf, bytes(buf))

        frame = bytearray(64)
        frame[0:4] = struct.pack("<I", self.cmdbuf)
        self.c.uc.mem_write(self.ring_phys + self.widx * 4, bytes(frame))
        nxt = (self.widx + 16) % self.ndw

        hdr = bytearray(MBOX_LEN)
        hdr[0:4] = struct.pack("<I", DONE)        # done set -> loop goes straight to drain
        hdr[0x0C:0x10] = struct.pack("<I", nxt)   # publish write_idx
        self.c.uc.mem_write(MBOX, bytes(hdr))

        n0 = len(probe) if probe is not None else 0
        self.irqs.raise_irq(DOORBELL_IRQ)
        for _ in range(int(timeout)):
            self.run(self.slice_blocks)
            if probe is not None and len(probe) > n0:
                break
        self.widx = nxt
        return (len(probe) - n0) if probe is not None else None


def make_det_runner(c, irqs):
    """DETERMINISTIC runner: advances the guest by a BLOCK BUDGET, not seconds.

    sOS idles on WFI, so emu_start returns cleanly and must be resumed; and a
    task that spins never returns at all. Both are handled by letting the
    interrupt block-hook stop emulation once the budget is spent. Results are
    then reproducible instead of depending on how loaded the machine is.
    """
    from unicorn.arm_const import UC_ARM_REG_PC, UC_ARM_REG_CPSR
    state = {"pc": None}

    def run(blocks):
        target = irqs.n + int(blocks)
        while irqs.n < target:
            irqs.budget = target
            thumb = bool(c.uc.reg_read(UC_ARM_REG_CPSR) & 0x20)
            pc = state["pc"] if state["pc"] is not None else 0x98
            begin = (pc | 1) if thumb else (pc & ~1)
            try:
                c.uc.emu_start(begin, 0xFF0000)
            except Exception:
                state["pc"] = c.uc.reg_read(UC_ARM_REG_PC)
                return False
            state["pc"] = c.uc.reg_read(UC_ARM_REG_PC)
        irqs.budget = None
        return True

    return run, state


def make_runner(c):
    """Resume-across-WFI runner usable in slices (see boot_sos_and_tas)."""
    from unicorn.arm_const import UC_ARM_REG_PC, UC_ARM_REG_CPSR

    state = {"pc": None}

    def run(seconds):
        deadline = time.time() + seconds
        while time.time() < deadline:
            slice_us = max(1, int((deadline - time.time()) * 1_000_000))
            thumb = bool(c.uc.reg_read(UC_ARM_REG_CPSR) & 0x20)
            pc = state["pc"] if state["pc"] is not None else 0x98
            begin = (pc | 1) if thumb else (pc & ~1)
            try:
                c.uc.emu_start(begin, 0xFF0000, timeout=slice_us)
            except Exception:
                state["pc"] = c.uc.reg_read(UC_ARM_REG_PC)
                return False
            state["pc"] = c.uc.reg_read(UC_ARM_REG_PC)
        return True

    return run, state