Select TP6 procedures from bounded TPU9 entry metadata

docs/tpu9.md

@@ -0,0 +1,41 @@
+# TPU9 procedure boundaries
+
+`pydasm.tpu9` reads the code geometry of Turbo Pascal 6 units. It uses the
+TPU9 entry and code-block tables, not a sweep over the whole code section.
+It has no compiler, emulator, Ghidra, or third-party package dependency.
+
+```python
+from pathlib import Path
+from pydasm.tpu9 import primary_routine, procedure_spans
+from pydasm import Decoder, DecoderOptions
+
+product = Path("PROBE.TPU").read_bytes()
+file_offset, code = primary_routine(product)
+decoder = Decoder(code, DecoderOptions())
+```
+
+A string literal can precede an entry in the same code block. For example,
+`01 4C` is a one-character Pascal string, not the start of an ADD instruction.
+A linear decoder started there can consume the next procedure's PUSH BP as
+an operand and miss its real entry. Entry metadata supplies the correct seed.
+
+The parser checks the symbol/code extents, table bounds and record alignment,
+code-block size sum, block selectors, and block-relative entry offsets.
+Duplicate aliases are accepted; the all-FFFF unassigned entry is ignored.
+Invalid tables never enable a raw byte-search fallback.
+
+`procedure_spans` returns absolute file-coordinate intervals bounded by the
+owning block and the next distinct entry in that block. `primary_span` selects
+the earliest local entry and requires PUSH BP / MOV BP,SP at that exact entry.
+It will not substitute a later carrier if the primary entry has a bad prologue.
+`primary_routine` decodes within that interval through its first complete
+RET/RETF. It rejects a truncated instruction or missing return; a return in
+another entry, block, symbol table, or trailer cannot complete the procedure.
+
+This API is for controlled, single-epilogue BP-framed probes. It is not a full
+TPU symbol/relocation reader, named-routine selector, arbitrary code finder,
+or control-flow reconstruction for multi-return assembly. Those callers must
+use their own authoritative entry/extent information and the normal decoder.
+
+`tests/test_tpu9.py` contains synthetic positive and adversarial controls, so
+these guarantees are testable without redistributing a vendor or game binary.

src/pydasm/tpu9.py

@@ -0,0 +1,113 @@
+"""Bounded TP6 procedure selection from TPU9 entry/code-block metadata.
+
+This is a container reader, not code discovery. Literal pools are legal before
+an entry inside its code block. Never linear-sweep those pools to find a BP
+prologue, or search past the owning block for a convenient return instruction.
+"""
+from __future__ import annotations
+
+import struct
+
+from .decoder import Decoder, DecoderOptions
+from .errors import Eof, UnexpectedEof
+
+PROLOGUE = b'\x55\x89\xe5'
+RECORD_SIZE = 8
+
+
+def code_span(data: bytes) -> tuple[int, int]:
+    """The absolute file bounds of the declared TPU9 code section."""
+    if len(data) < 0x40 or data[:4] != b'TPU9':
+        raise ValueError('not a complete TPU9 header')
+    symbols, size = struct.unpack_from('<HH', data, 0x1c)
+    start = (symbols + 15) & ~15
+    if symbols < 0x40 or not size or start + size > len(data):
+        raise ValueError('incomplete TPU9 code section')
+    return start, start + size
+
+
+def procedure_spans(data: bytes) -> tuple[tuple[int, int], ...]:
+    """Local entry spans in file coordinates, bounded by their owning blocks.
+
+    Entry and code-block records are eight bytes. An entry's block selector is
+    an offset into the block table, NOT a code-section offset. Block sizes are
+    cumulative; entry offsets are block-relative. Duplicate entry aliases are
+    allowed. The next distinct entry in the same block also limits a span.
+
+    Missing/invalid tables are errors, not permission to scan arbitrary bytes.
+    This reads only TPU9 code geometry, not symbols, imports or relocations.
+    """
+    start, end = code_span(data)
+    symbols = struct.unpack_from('<H', data, 0x1c)[0]
+    entries_at, blocks_at, blocks_end = struct.unpack_from('<HHH', data, 0x0c)
+    if not 0x40 <= entries_at <= blocks_at < blocks_end <= symbols:
+        raise ValueError('invalid TPU9 entry/code-block table bounds')
+    if (blocks_at - entries_at) % RECORD_SIZE or (blocks_end - blocks_at) % RECORD_SIZE:
+        raise ValueError('partial TPU9 entry/code-block record')
+    blocks = []
+    running = 0
+    for at in range(blocks_at, blocks_end, RECORD_SIZE):
+        size = struct.unpack_from('<H', data, at + 2)[0]
+        blocks.append((running, size))
+        running += size
+    if running != end - start:
+        raise ValueError('TPU9 code-block sizes disagree with code section')
+    offsets: dict[int, set[int]] = {}
+    for at in range(entries_at, blocks_at, RECORD_SIZE):
+        selector, offset = struct.unpack_from('<HH', data, at + 4)
+        if (selector, offset) == (0xffff, 0xffff):
+            continue  # Unassigned/external entry, including the usual entry 0.
+        if selector % RECORD_SIZE or selector // RECORD_SIZE >= len(blocks):
+            raise ValueError('TPU9 entry names an invalid code block')
+        block = selector // RECORD_SIZE
+        if offset >= blocks[block][1]:
+            raise ValueError('TPU9 entry is outside its owning code block')
+        offsets.setdefault(block, set()).add(offset)
+    spans = []
+    for block, values in offsets.items():
+        base, size = blocks[block]
+        ordered = sorted(values)
+        for index, offset in enumerate(ordered):
+            limit = ordered[index + 1] if index + 1 < len(ordered) else size
+            spans.append((start + base + offset, start + base + limit))
+    return tuple(sorted(spans))
+
+
+def primary_span(data: bytes) -> tuple[int, int]:
+    """Earliest declared local entry and its exclusive, metadata-owned end.
+
+    The spelling PUSH BP / MOV BP,SP is required at the entry itself. This does
+    not infer names, select by a matching byte pattern, or choose a later entry
+    because the first one failed a reconstruction comparison.
+    """
+    spans = procedure_spans(data)
+    if spans:
+        start, end = spans[0]
+        if end - start >= len(PROLOGUE) and data[start:start + 3] == PROLOGUE:
+            return start, end
+    raise ValueError('earliest declared TPU9 entry is not a BP-framed procedure')
+
+
+def primary_routine(data: bytes) -> tuple[int, bytes]:
+    """Decode the primary BP entry through its first complete RET/RETF.
+
+    For controlled single-epilogue TP6 probes. This is NOT control-flow recovery
+    for arbitrary assembly or multi-return functions. A missing return before
+    the next entry/block is an error; adjacent carrier code cannot complete it.
+    """
+    start, limit = primary_span(data)
+    code = data[start:limit]
+    decoder = Decoder(code, DecoderOptions())
+    position = 0
+    try:
+        while position < len(code):
+            instruction = decoder.decode_next()
+            if (not instruction.is_valid or not instruction.is_complete
+                    or not 0 < instruction.length <= len(code) - position):
+                raise ValueError('invalid instruction in TPU9 procedure')
+            position += instruction.length
+            if instruction.mnemonic.name in ('RET', 'RETF'):
+                return start, bytes(code[:position])
+    except (Eof, UnexpectedEof) as error:
+        raise ValueError('incomplete TPU9 procedure instruction') from error
+    raise ValueError('no complete return inside the TPU9 procedure span')

tests/test_tpu9.py

@@ -0,0 +1,101 @@
+"""Synthetic container controls; no licensed compiler or game binary needed."""
+import struct
+import pytest
+from pydasm.tpu9 import code_span, procedure_spans, primary_span, primary_routine
+
+NEAR = bytes.fromhex('5589e55dc20600')
+FAR = bytes.fromhex('5589e55dca0800')
+
+
+def tpu(blocks, entries):
+    """Minimal structurally valid header, local tables, and code section."""
+    entries_at = 64
+    records = [(0xffff, 0xffff), *entries]
+    blocks_at = entries_at + 8 * len(records)
+    blocks_end = blocks_at + 8 * len(blocks)
+    data = bytearray((blocks_end + 15) & ~15)
+    data[:4] = b'TPU9'
+    struct.pack_into('<HHH', data, 0x0c, entries_at, blocks_at, blocks_end)
+    struct.pack_into('<HH', data, 0x1c, blocks_end, sum(map(len, blocks)))
+    for i, (selector, offset) in enumerate(records):
+        struct.pack_into('<HH', data, entries_at + 8 * i + 4, selector, offset)
+    for i, block in enumerate(blocks):
+        struct.pack_into('<H', data, blocks_at + 8 * i + 2, len(block))
+    return bytes(data) + b''.join(blocks)
+
+
+@pytest.mark.parametrize('pool', [b'\x01L', b'\xff', bytes.fromhex('b85589e5'), b'\0' * 32])
+@pytest.mark.parametrize('routine', [NEAR, FAR])
+def test_metadata_skips_constants_even_when_a_sweep_loses_the_prologue(pool, routine):
+    data = tpu([pool + routine], [(0, len(pool))])
+    start, _ = code_span(data)
+    assert primary_routine(data) == (start + len(pool), routine)
+
+
+def test_block_selector_and_entry_offset_are_not_section_offsets():
+    data = tpu([b'\x90\xc3', b'\x01L' + FAR], [(8, 2)])
+    start, end = code_span(data)
+    assert procedure_spans(data) == ((start + 4, end),)
+    assert primary_routine(data) == (start + 4, FAR)
+
+
+def test_unsorted_entries_and_aliases_still_select_the_earliest_entry():
+    data = tpu([NEAR, FAR], [(8, 0), (0, 0), (0, 0)])
+    start, _ = code_span(data)
+    assert primary_routine(data) == (start, NEAR)
+    assert len(procedure_spans(data)) == 2
+
+
+@pytest.mark.parametrize('same_block', [True, False])
+def test_adjacent_carrier_return_cannot_complete_an_unterminated_primary(same_block):
+    broken = bytes.fromhex('5589e590')
+    data = (tpu([broken + FAR], [(0, 0), (0, len(broken))]) if same_block
+            else tpu([broken, FAR], [(0, 0), (8, 0)]))
+    start, _ = code_span(data)
+    assert primary_span(data) == (start, start + len(broken))
+    with pytest.raises(ValueError, match='return'):
+        primary_routine(data)
+
+
+def test_undeclared_prologue_inside_literal_or_operand_is_not_an_entry():
+    data = tpu([b'\x05' + NEAR + b'\x90\xc3'], [(0, 1 + len(NEAR))])
+    with pytest.raises(ValueError, match='declared'):
+        primary_routine(data)
+
+
+@pytest.mark.parametrize('selector,offset', [(1, 0), (8, 0), (0, len(NEAR)),
+                                             (0xffff, 0), (0, 0xffff)])
+def test_invalid_entry_never_falls_back_to_scanning(selector, offset):
+    with pytest.raises(ValueError):
+        primary_routine(tpu([NEAR], [(selector, offset)]))
+
+
+@pytest.mark.parametrize('offset,value', [(0x0c, 0), (0x0c, 65), (0x0e, 63),
+                                         (0x10, 0xffff), (0x1c, 63), (0x1e, 0)])
+def test_invalid_header_or_table_geometry_is_rejected(offset, value):
+    data = bytearray(tpu([NEAR], [(0, 0)]))
+    struct.pack_into('<H', data, offset, value)
+    with pytest.raises(ValueError):
+        primary_routine(bytes(data))
+
+
+def test_block_sizes_must_cover_exactly_the_declared_code():
+    data = bytearray(tpu([NEAR], [(0, 0)]))
+    blocks_at = struct.unpack_from('<H', data, 0x0e)[0]
+    struct.pack_into('<H', data, blocks_at + 2, len(NEAR) - 1)
+    with pytest.raises(ValueError, match='sizes'):
+        primary_routine(bytes(data))
+
+
+@pytest.mark.parametrize('data', [b'', b'TPU9', tpu([FAR], [(0, 0)])[:-1],
+                                  tpu([FAR[:-1]], [(0, 0)]),
+                                  tpu([NEAR[:-1]], [(0, 0)]) + b'\0'])
+def test_truncation_and_bytes_after_code_cannot_supply_a_return_operand(data):
+    with pytest.raises(ValueError):
+        primary_routine(data)
+
+
+def test_bad_primary_prologue_is_not_replaced_by_a_later_carrier():
+    data = tpu([b'\x90\xc3', FAR], [(0, 0), (8, 0)])
+    with pytest.raises(ValueError, match='earliest declared'):
+        primary_routine(data)