boxsim-fileformat-rev/boxsim/_oracle.py

"""Development-only reference decoder: runs Boxsim's own machine code.

This module emulates `ExpandA2F` (VA 0xa11880) out of an unmodified Boxsim.exe
under Unicorn, with faked Delphi TStream objects. It is exact by construction
and exists solely as the oracle that the pure Python decoder in `boxsim.a2f`
is differentially tested against.

It is NOT part of the shipped tool: it needs `unicorn`, `pefile` and an 8 MB
proprietary Windows binary. Point BOXSIM_EXE at a Boxsim.exe to use it.

    python -m boxsim.cli decode --oracle file.bpj
"""

from __future__ import annotations

import os
import struct

MAGIC = b"MyIni2CK"

ENTRY = 0xA11880  # ExpandA2F(self, instream, outstream)
ERR_FUNC = 0x97E584  # Boxsim's error reporter
RAISE_FUNC = 0x80ACC4  # Delphi RaiseExcept

OBJ_BASE = 0x20000000  # decoder object: ~11 MB of zero-init tables
OBJ_SIZE = 0x01000000
FAKE_BASE = 0x40000000  # fake stream objects, vtables, trap stubs
STACK_BASE = 0x60000000
STACK_SIZE = 0x100000
GDT_BASE = 0x50000000
FS_SEG = 0x50002000  # Delphi SEH chain scratch space
SENTINEL = 0x7F00DEAD

IN_OBJ, OUT_OBJ = FAKE_BASE, FAKE_BASE + 0x100
IN_VT, OUT_VT = FAKE_BASE + 0x1000, FAKE_BASE + 0x2000
STUB = FAKE_BASE + 0x3000
S_IN_SIZE, S_IN_READ, S_IN_SEEK, S_OUT_WRITE, S_OUT_SIZE, S_OUT_SEEK = range(6)
STUB_SPAN = 256


def default_exe() -> str:
    return os.environ.get(
        "BOXSIM_EXE",
        os.path.join(
            os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
            "boxsim_dl",
            "BoxSim 2.1 DE (Stand 12.2025)",
            "Boxsim.exe",
        ),
    )


def _gdt_entry(base: int, limit: int, access: int, gran: int) -> bytes:
    e = limit & 0xFFFF
    e |= (base & 0xFFFFFF) << 16
    e |= (access & 0xFF) << 40
    e |= ((limit >> 16) & 0xF) << 48
    e |= (gran & 0xF) << 52
    e |= ((base >> 24) & 0xFF) << 56
    return struct.pack("<Q", e)


def expand_a2f(data: bytes, exe: str | None = None) -> bytes:
    """Decompress a descrambled MyIni2CK stream using Boxsim's own decoder."""
    import pefile
    from unicorn import UC_ARCH_X86, UC_HOOK_CODE, UC_MODE_32, Uc
    from unicorn.x86_const import (
        UC_X86_REG_CS,
        UC_X86_REG_DS,
        UC_X86_REG_EAX,
        UC_X86_REG_ECX,
        UC_X86_REG_EDX,
        UC_X86_REG_ES,
        UC_X86_REG_ESP,
        UC_X86_REG_FS,
        UC_X86_REG_GDTR,
        UC_X86_REG_GS,
        UC_X86_REG_SS,
    )

    exe = exe or default_exe()
    if not os.path.exists(exe):
        raise FileNotFoundError(
            f"oracle needs Boxsim.exe; set BOXSIM_EXE (looked at {exe!r})"
        )

    pe = pefile.PE(exe)
    base = pe.OPTIONAL_HEADER.ImageBase
    img = pe.get_memory_mapped_image(ImageBase=base)

    uc = Uc(UC_ARCH_X86, UC_MODE_32)
    uc.mem_map(base, (len(img) + 0xFFF) & ~0xFFF)
    uc.mem_write(base, img)
    for addr, size in (
        (OBJ_BASE, OBJ_SIZE),
        (FAKE_BASE, 0x10000),
        (STACK_BASE, STACK_SIZE),
        (GDT_BASE, 0x4000),
    ):
        uc.mem_map(addr, size)

    uc.mem_write(IN_OBJ, struct.pack("<I", IN_VT))
    uc.mem_write(OUT_OBJ, struct.pack("<I", OUT_VT))
    layout = (
        (IN_VT, {0x00: S_IN_SIZE, 0x0C: S_IN_READ, 0x28: S_IN_SEEK}),
        (OUT_VT, {0x00: S_OUT_SIZE, 0x0C: S_IN_READ, 0x10: S_OUT_WRITE, 0x28: S_OUT_SEEK}),
    )
    for vt, stubs in layout:
        for off in range(0, 0x30, 4):
            sid = stubs.get(off, 15)
            uc.mem_write(vt + off, struct.pack("<I", STUB + sid * 16))
    for sid in range(16):
        # Delphi Seek is `stdcall`-ish with 8 bytes of args; the rest are cdecl.
        body = b"\xc2\x08\x00" if sid in (S_IN_SEEK, S_OUT_SEEK) else b"\xc3"
        uc.mem_write(STUB + sid * 16, body)

    # A full flat GDT. Loading FS alone resets the other segment registers'
    # hidden descriptors in unicorn, which faults the very first push.
    uc.mem_write(GDT_BASE + 8, _gdt_entry(0, 0xFFFFF, 0x9B, 0xC))
    uc.mem_write(GDT_BASE + 16, _gdt_entry(0, 0xFFFFF, 0x93, 0xC))
    uc.mem_write(GDT_BASE + 24, _gdt_entry(FS_SEG, 0xFFF, 0x93, 0x4))
    uc.reg_write(UC_X86_REG_GDTR, (0, GDT_BASE, 0x1F, 0))
    uc.reg_write(UC_X86_REG_CS, 8)
    for seg in (UC_X86_REG_DS, UC_X86_REG_ES, UC_X86_REG_SS, UC_X86_REG_GS):
        uc.reg_write(seg, 16)
    uc.reg_write(UC_X86_REG_FS, 24)

    state = {"pos": 0, "out": bytearray(), "err": None}

    def hook_code(u, addr, _size, _ud):
        if STUB <= addr < STUB + STUB_SPAN:
            sid = (addr - STUB) // 16
            eax = u.reg_read(UC_X86_REG_EAX)
            edx = u.reg_read(UC_X86_REG_EDX)
            ecx = u.reg_read(UC_X86_REG_ECX)
            if sid == S_IN_READ and eax == IN_OBJ:
                n = min(ecx, len(data) - state["pos"])
                if n > 0:
                    u.mem_write(edx, data[state["pos"] : state["pos"] + n])
                state["pos"] += n
                u.reg_write(UC_X86_REG_EAX, n)
            elif sid == S_IN_SIZE:
                u.reg_write(UC_X86_REG_EAX, len(data))
                u.reg_write(UC_X86_REG_EDX, 0)
            elif sid == S_IN_SEEK:
                esp = u.reg_read(UC_X86_REG_ESP)
                off = struct.unpack("<i", u.mem_read(esp + 4, 4))[0]
                origin = edx & 0xFF
                state["pos"] = (
                    off
                    if origin == 0
                    else state["pos"] + off
                    if origin == 1
                    else len(data) + off
                )
                u.reg_write(UC_X86_REG_EAX, state["pos"] & 0xFFFFFFFF)
                u.reg_write(UC_X86_REG_EDX, 0)
            elif sid == S_OUT_WRITE:
                state["out"] += bytes(u.mem_read(edx, ecx))
                u.reg_write(UC_X86_REG_EAX, ecx)
            elif sid in (S_OUT_SIZE, S_OUT_SEEK):
                u.reg_write(UC_X86_REG_EAX, len(state["out"]))
                u.reg_write(UC_X86_REG_EDX, 0)
            else:
                state["err"] = f"unexpected vtable stub {sid}"
                u.emu_stop()
        elif addr == ERR_FUNC:
            eax = u.reg_read(UC_X86_REG_EAX)
            msg = bytes(u.mem_read(eax, 64)).split(b"\x00")[0]
            state["err"] = f"boxsim error: {msg.decode('latin1')}"
            u.emu_stop()
        elif addr == RAISE_FUNC:
            state["err"] = "delphi exception"
            u.emu_stop()

    # Bind the hooks to the trap addresses only. A single unbounded UC_HOOK_CODE
    # fires the callback on every emulated instruction and costs ~5x wall clock.
    uc.hook_add(UC_HOOK_CODE, hook_code, None, STUB, STUB + STUB_SPAN - 1)
    uc.hook_add(UC_HOOK_CODE, hook_code, None, ERR_FUNC, ERR_FUNC)
    uc.hook_add(UC_HOOK_CODE, hook_code, None, RAISE_FUNC, RAISE_FUNC)

    esp = STACK_BASE + STACK_SIZE - 0x1000
    uc.mem_write(esp, struct.pack("<I", SENTINEL))
    uc.reg_write(UC_X86_REG_ESP, esp)
    # Codec config, as the compressor entry point (0xa1013c) sets it.
    uc.mem_write(OBJ_BASE + 0x04, b"\x01\x01")
    uc.mem_write(OBJ_BASE + 0x08, struct.pack("<I", 0x20))
    uc.mem_write(OBJ_BASE + 0x0C, struct.pack("<I", 0x47))
    uc.mem_write(OBJ_BASE + 0x10, b"\x01")
    uc.reg_write(UC_X86_REG_EAX, OBJ_BASE)
    uc.reg_write(UC_X86_REG_EDX, IN_OBJ)
    uc.reg_write(UC_X86_REG_ECX, OUT_OBJ)

    uc.emu_start(ENTRY, SENTINEL, timeout=600 * 1000000)

    if state["err"]:
        raise RuntimeError(state["err"])
    if state["pos"] != len(data):
        raise RuntimeError(f"decoder stopped early: {state['pos']}/{len(data)}")
    del uc, img, pe
    return bytes(state["out"])


def decode_bpj(raw: bytes, exe: str | None = None) -> bytes:
    """Full container decode using the oracle for the compressed layer."""
    from .krypt import unscramble

    if raw[:8] != MAGIC:
        raise ValueError("not a MyIni2CK file")
    return expand_a2f(MAGIC + unscramble(raw[8:]), exe=exe)