sgi-indy/tools/rom-map.py

#!/usr/bin/env python3
"""Build a structural map of an SGI Indy IP24 PROM image.

Combines byte-level layout facts (padding, printable density, MIPS jump
vectors) with procedure and string inventories exported from a REA session.

Usage:
    tools/rom-map.py roms/ip24prom.070-9101-011.bin \
        --evidence evidence/rea-bundle.json \
        --output docs/rom-map.json
"""

from __future__ import annotations

import argparse
import hashlib
import json
import struct
from collections import Counter
from pathlib import Path

BASE = 0x1FC00000
PAGE = 0x1000
VECTOR_STRIDE = 0x10
# 60 entries of 16 bytes. The reset handler itself starts at 0x1FC003C0 and
# overwrites what would otherwise be the final four entries.
VECTOR_TABLE_SIZE = 0x3C0


def sha1(data: bytes) -> str:
    return hashlib.sha1(data).hexdigest()


def sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def printable_ratio(chunk: bytes) -> float:
    if not chunk:
        return 0.0
    printable = sum(
        1 for b in chunk if 0x20 <= b <= 0x7E or b in (0x09, 0x0A, 0x0D)
    )
    return round(printable / len(chunk), 4)


def zero_ratio(chunk: bytes) -> float:
    return round(chunk.count(0) / len(chunk), 4) if chunk else 0.0


def ff_ratio(chunk: bytes) -> float:
    return round(chunk.count(0xFF) / len(chunk), 4) if chunk else 0.0


def j_target(word: int, pc: int) -> int | None:
    """Decode a MIPS J/JAL absolute target, or None for other opcodes."""
    if (word >> 26) not in (0x02, 0x03):
        return None
    return ((pc + 4) & 0xF0000000) | ((word & 0x03FFFFFF) << 2)


def vector_table(rom: bytes) -> list[dict]:
    rows = []
    for off in range(0, min(VECTOR_TABLE_SIZE, len(rom)), VECTOR_STRIDE):
        word = struct.unpack(">I", rom[off : off + 4])[0]
        target = j_target(word, BASE + off)
        rows.append(
            {
                "offset": off,
                "address": f"0x{BASE + off:08x}",
                "word": f"0x{word:08x}",
                "target": f"0x{target:08x}" if target is not None else None,
            }
        )
    return rows


def load_inventory(path: Path | None) -> tuple[list[dict], list[dict]]:
    if path is None or not path.exists():
        return [], []
    bundle = json.loads(path.read_text())
    procedures: list[dict] = []
    strings: list[dict] = []
    for record in bundle.get("records", []):
        operation = record.get("operation")
        if operation == "list_procedures":
            procedures = record.get("normalized_result", [])
        elif operation == "list_strings":
            strings = record.get("normalized_result", [])
    return procedures, strings


def classify(page: dict, last_non_ff: int) -> str:
    if page["offset"] >= last_non_ff or page["ff_ratio"] > 0.9:
        return "padding"
    if page["address"] < BASE + VECTOR_TABLE_SIZE:
        return "exception-vectors"
    if page["procedures"]:
        return "code"
    if page["strings"]:
        return "strings"
    if page["printable_ratio"] > 0.6:
        return "text-data"
    if page["zero_ratio"] > 0.5:
        return "zero-data"
    return "binary-data"


def build(rom: bytes, procedures: list[dict], strings: list[dict]) -> dict:
    proc_addrs = sorted(int(p["address"], 16) for p in procedures)
    str_addrs = sorted(int(s["address"], 16) for s in strings)

    last_non_ff = max((i for i, b in enumerate(rom) if b != 0xFF), default=-1)

    pages = []
    for off in range(0, len(rom), PAGE):
        chunk = rom[off : off + PAGE]
        addr = BASE + off
        page = {
            "offset": off,
            "address": addr,
            "procedures": sum(1 for a in proc_addrs if addr <= a < addr + PAGE),
            "strings": sum(1 for a in str_addrs if addr <= a < addr + PAGE),
            "printable_ratio": printable_ratio(chunk),
            "zero_ratio": zero_ratio(chunk),
            "ff_ratio": ff_ratio(chunk),
        }
        page["kind"] = classify(page, last_non_ff)
        pages.append(page)

    # Collapse equal neighbouring classifications into ranges.
    regions = []
    for page in pages:
        if regions and regions[-1]["kind"] == page["kind"]:
            regions[-1]["end"] = page["address"] + PAGE
            regions[-1]["procedures"] += page["procedures"]
            regions[-1]["strings"] += page["strings"]
        else:
            regions.append(
                {
                    "kind": page["kind"],
                    "start": page["address"],
                    "end": page["address"] + PAGE,
                    "procedures": page["procedures"],
                    "strings": page["strings"],
                }
            )

    version = next(
        (s["value"] for s in strings if s["value"].startswith("SGI Version")), None
    )

    code_span = None
    if proc_addrs:
        code_span = {
            "start": f"0x{min(proc_addrs):08x}",
            "last_function": f"0x{max(proc_addrs):08x}",
            "count": len(proc_addrs),
        }

    string_span = None
    if str_addrs:
        string_span = {
            "start": f"0x{min(str_addrs):08x}",
            "end": f"0x{max(str_addrs):08x}",
            "count": len(str_addrs),
        }

    histogram = Counter(p["kind"] for p in pages)

    return {
        "source": {
            "bytes": len(rom),
            "sha1": sha1(rom),
            "sha256": sha256(rom),
            "version_string": version,
            "last_non_ff_offset": last_non_ff,
            "last_non_ff_address": (
                f"0x{BASE + last_non_ff:08x}" if last_non_ff >= 0 else None
            ),
        },
        "image": {
            "base": f"0x{BASE:08x}",
            "end": f"0x{BASE + len(rom):08x}",
            "page_size": PAGE,
        },
        "code": code_span,
        "strings": string_span,
        "page_kinds": dict(histogram),
        "regions": [
            {
                **r,
                "start": f"0x{r['start']:08x}",
                "end": f"0x{r['end']:08x}",
            }
            for r in regions
        ],
        "exception_vectors": vector_table(rom),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("rom")
    parser.add_argument("--evidence", type=Path, default=None)
    parser.add_argument("--output", type=Path, default=None)
    parser.add_argument(
        "--procedures-csv",
        type=Path,
        default=None,
        help="write the analysed function start addresses as CSV",
    )
    args = parser.parse_args()

    rom = Path(args.rom).read_bytes()
    procedures, strings = load_inventory(args.evidence)
    doc = build(rom, procedures, strings)

    if args.procedures_csv and procedures:
        args.procedures_csv.parent.mkdir(parents=True, exist_ok=True)
        rows = ["address,name,thunk_target"]
        for proc in sorted(procedures, key=lambda p: int(p["address"], 16)):
            rows.append(
                "{address},{value},{target}".format(
                    address=proc["address"],
                    value=proc["value"],
                    target=proc.get("procedure", {}).get("thunk_target") or "",
                )
            )
        args.procedures_csv.write_text("\n".join(rows) + "\n")
        print(args.procedures_csv)

    text = json.dumps(doc, indent=2)
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(text + "\n")
        print(args.output)
    else:
        print(text)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())