tdkpin/original/tools/audit_reconstruction.py

#!/usr/bin/env python3
"""Audit the exact binary-to-readable-C reconstruction boundary for TDKPIN."""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import re
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
TARGET_SHA256 = "a9022f1894e3e6e21fc42e8f6c932f7c549ca77f63aaa0c488bb9d55d9d0174c"
PRESERVED_RAW_SHA256 = (
    "987a47580ba5c92fa071c989bdaeddf138e285d2b07106c5909a4acbc3e27088"
)
LEDGER_COLUMNS = (
    "address",
    "raw_name",
    "segment",
    "body_bytes",
    "status",
    "reconstructed_name",
    "module",
    "evidence",
    "notes",
)
STATUS_ORDER = ("raw", "partial", "restored", "verified", "unknown", "blocked")
ALLOWED_STATUSES = set(STATUS_ORDER)
RAW_MARKER = re.compile(r"^/\* ===== ([0-9a-f]{4}:[0-9a-f]{4}) (\S+) ===== \*/$")
IMPORT_COLUMNS = (
    "address",
    "library",
    "import_name",
    "stack_purge_bytes",
    "call_sites",
    "status",
    "stage",
    "prototype",
    "evidence",
    "notes",
)
SHARED_ENTRY_COLUMNS = (
    "address",
    "containing_function",
    "status",
    "reconstructed_name",
    "evidence",
    "notes",
)
CODE_DATA_COLUMNS = (
    "start",
    "end",
    "bytes",
    "status",
    "object_id",
    "object_name",
    "c_type",
    "initialization",
    "readers",
    "evidence",
    "notes",
)

# These routines have readable companion fragments, but none yet has a complete,
# standalone, binary-faithful implementation. Keep them partial until the whole
# raw function has been reviewed and all callees/types are resolved.
PARTIAL_FUNCTIONS = {}

CALLBACK_APIS = {
    "REGISTERCLASS",
    "SYSTEMTIMERMAKE",
    "DIALOGBOXPARAM",
    "CREATEDIALOGPARAM",
    "MAKEPROCINSTANCE",
}


def read_tsv(path: Path) -> list[dict[str, str]]:
    with path.open(newline="", encoding="utf-8") as handle:
        return list(csv.DictReader(handle, delimiter="\t"))


def internal_functions() -> list[dict[str, str]]:
    rows = read_tsv(ROOT / "BINARY_FUNCTIONS.tsv")
    return [row for row in rows if row["entry_kind"] == "internal"]


def initialize_ledger(path: Path) -> None:
    if path.exists():
        raise SystemExit(f"refusing to overwrite existing ledger: {path}")
    rows = []
    for function in internal_functions():
        partial = PARTIAL_FUNCTIONS.get(function["address"])
        rows.append(
            {
                "address": function["address"],
                "raw_name": function["name"],
                "segment": function["block"],
                "body_bytes": function["body_bytes"],
                "status": "partial" if partial else "raw",
                "reconstructed_name": partial[0] if partial else "",
                "module": partial[1] if partial else "",
                "evidence": partial[2] if partial else "decompiler-only",
                "notes": (
                    "Readable fragment exists; full function and dependencies remain pending."
                    if partial
                    else "No reviewed readable-C implementation yet."
                ),
            }
        )
    with path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle, fieldnames=LEDGER_COLUMNS, delimiter="\t", lineterminator="\n"
        )
        writer.writeheader()
        writer.writerows(rows)


def sync_ledger(path: Path) -> None:
    """Add newly recovered functions while preserving manual semantic work."""
    if not path.exists():
        raise SystemExit(f"cannot sync missing ledger: {path}")
    existing = {row["address"]: row for row in read_tsv(path)}
    functions = internal_functions()
    current_addresses = {function["address"] for function in functions}
    removed = set(existing) - current_addresses
    if removed:
        raise SystemExit(
            "refusing to discard ledger rows: " + ", ".join(sorted(removed))
        )

    rows = []
    for function in functions:
        row = existing.get(function["address"])
        if row is None:
            row = {
                "address": function["address"],
                "status": "raw",
                "reconstructed_name": "",
                "module": "",
                "evidence": "decompiler-only",
                "notes": "Newly recovered function; no reviewed readable-C implementation yet.",
            }
        row["raw_name"] = function["name"]
        row["segment"] = function["block"]
        row["body_bytes"] = function["body_bytes"]
        rows.append(row)

    temporary = path.with_suffix(path.suffix + ".tmp")
    with temporary.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle, fieldnames=LEDGER_COLUMNS, delimiter="\t", lineterminator="\n"
        )
        writer.writeheader()
        writer.writerows(rows)
    temporary.replace(path)


def write_tsv(
    path: Path, columns: tuple[str, ...], rows: list[dict[str, object]]
) -> None:
    if path.exists():
        raise SystemExit(f"refusing to overwrite existing ledger: {path}")
    with path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle, fieldnames=columns, delimiter="\t", lineterminator="\n"
        )
        writer.writeheader()
        writer.writerows(rows)


def initialize_support_ledgers() -> None:
    functions = read_tsv(ROOT / "BINARY_FUNCTIONS.tsv")
    references = read_tsv(ROOT / "REFERENCES.tsv")
    calls_by_target = Counter(
        row["to_address"] for row in references if "CALL" in row["reference_type"]
    )
    imports = [row for row in functions if row["entry_kind"] == "external"]
    write_tsv(
        ROOT / "IMPORT_RECONSTRUCTION.tsv",
        IMPORT_COLUMNS,
        [
            {
                "address": row["address"],
                "library": row["namespace"],
                "import_name": row["name"],
                "stack_purge_bytes": row["stack_purge_bytes"],
                "call_sites": calls_by_target[row["address"]],
                "status": "unknown",
                "stage": "untyped-import",
                "prototype": "",
                "evidence": "NE import+Ghidra reference export",
                "notes": "Win16 prototype and every call site remain to be reviewed.",
            }
            for row in imports
        ],
    )

    data_references: dict[str, list[dict[str, str]]] = {}
    for row in references:
        if row["to_block"] == "Data6":
            data_references.setdefault(row["to_address"], []).append(row)
    write_tsv(
        ROOT / "DATA_RECONSTRUCTION.tsv",
        (
            "address",
            "status",
            "object_name",
            "type",
            "extent_start",
            "extent_end",
            "reads",
            "writes",
            "referring_functions",
            "initialization",
            "evidence",
            "notes",
        ),
        [
            {
                "address": address,
                "status": "unknown",
                "stage": "candidate-address",
                "object_name": next(
                    (row["to_symbol"] for row in rows if row["to_symbol"]), ""
                ),
                "type": "",
                "extent_start": address,
                "extent_end": address,
                "reads": sum("READ" in row["reference_type"] for row in rows),
                "writes": sum("WRITE" in row["reference_type"] for row in rows),
                "referring_functions": ",".join(
                    sorted(
                        {row["from_function"] for row in rows if row["from_function"]}
                    )
                ),
                "initialization": "",
                "evidence": "Ghidra DGROUP references",
                "notes": "Candidate address; semantic object boundary and type remain unresolved.",
            }
            for address, rows in sorted(data_references.items())
        ],
    )

    manifest = json.loads((ROOT / "assets/manifest.json").read_text(encoding="utf-8"))
    write_tsv(
        ROOT / "RESOURCE_RECONSTRUCTION.tsv",
        (
            "type",
            "id",
            "file_offset",
            "allocated_size",
            "raw_sha256",
            "raw_path",
            "decoded_paths",
            "status",
            "users",
            "behavior",
            "notes",
        ),
        [
            {
                "type": resource["type"],
                "id": resource["id"],
                "file_offset": resource["file_offset"],
                "allocated_size": resource["allocated_size"],
                "raw_sha256": resource["raw_sha256"],
                "raw_path": resource["raw_path"],
                "decoded_paths": ",".join(resource.get("decoded_paths", [])),
                "status": "partial",
                "stage": "extracted-unmapped",
                "users": "",
                "behavior": "",
                "notes": "Lossless extraction is verified; code users and behavior remain to be mapped.",
            }
            for resource in manifest["resources"]
        ],
    )

    callback_rows = []
    for row in references:
        if row["to_symbol"] in CALLBACK_APIS and "CALL" in row["reference_type"]:
            callback_rows.append(
                {
                    "registration_site": row["from_address"],
                    "registering_function": row["from_function"],
                    "api": row["to_symbol"],
                    "status": "unknown",
                    "stage": "registration-site",
                    "callback_address": "",
                    "callback_name": "",
                    "lifetime": "",
                    "evidence": "Ghidra call reference",
                    "notes": "Recover callback argument through Win16 ABI and stored state.",
                }
            )
    write_tsv(
        ROOT / "CALLBACK_RECONSTRUCTION.tsv",
        (
            "registration_site",
            "registering_function",
            "api",
            "status",
            "stage",
            "callback_address",
            "callback_name",
            "lifetime",
            "evidence",
            "notes",
        ),
        callback_rows,
    )


def sync_support_ledgers() -> None:
    path = ROOT / "IMPORT_RECONSTRUCTION.tsv"
    if not path.exists():
        raise SystemExit(f"cannot sync missing ledger: {path}")
    existing = {row["address"]: row for row in read_tsv(path)}
    import_statuses = {
        "unreviewed": ("unknown", "untyped-import"),
        "prototype-sourced": ("partial", "prototype-sourced"),
        "binary-reviewed": ("verified", "binary-reviewed"),
        "runtime-data": ("verified", "runtime-data"),
    }
    for row in existing.values():
        if not row.get("stage"):
            row["status"], row["stage"] = import_statuses.get(
                row["status"], (row["status"], row["status"])
            )
    functions = read_tsv(ROOT / "BINARY_FUNCTIONS.tsv")
    imports = [row for row in functions if row["entry_kind"] == "external"]
    references = read_tsv(ROOT / "REFERENCES.tsv")
    calls_by_target = Counter(
        row["to_address"] for row in references if "CALL" in row["reference_type"]
    )
    current_addresses = {row["address"] for row in imports}
    removed = set(existing) - current_addresses
    if removed:
        raise SystemExit(
            "refusing to discard import-ledger rows: " + ", ".join(sorted(removed))
        )
    rows = []
    for function in imports:
        row = existing.get(function["address"])
        if row is None:
            row = {
                "address": function["address"],
                "status": "unknown",
                "stage": "untyped-import",
                "prototype": "",
                "evidence": "NE import+Ghidra reference export",
                "notes": "Win16 prototype and every call site remain to be reviewed.",
            }
        row["library"] = function["namespace"]
        row["import_name"] = function["name"]
        row["stack_purge_bytes"] = function["stack_purge_bytes"]
        row["call_sites"] = calls_by_target[function["address"]]
        rows.append(row)
    temporary = path.with_suffix(path.suffix + ".tmp")
    with temporary.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle, fieldnames=IMPORT_COLUMNS, delimiter="\t", lineterminator="\n"
        )
        writer.writeheader()
        writer.writerows(rows)
    temporary.replace(path)

    normalize_support_statuses(
        ROOT / "DATA_RECONSTRUCTION.tsv",
        {"unresolved": ("unknown", "candidate-address")},
    )
    normalize_support_statuses(
        ROOT / "RESOURCE_RECONSTRUCTION.tsv",
        {"extracted-unmapped": ("partial", "extracted-unmapped")},
    )
    normalize_support_statuses(
        ROOT / "CALLBACK_RECONSTRUCTION.tsv",
        {"unresolved": ("unknown", "registration-site")},
    )


def normalize_support_statuses(path: Path, mapping: dict[str, tuple[str, str]]) -> None:
    with path.open(newline="", encoding="utf-8") as handle:
        reader = csv.DictReader(handle, delimiter="\t")
        columns = list(reader.fieldnames or ())
        rows = list(reader)
    if "stage" not in columns:
        columns.insert(columns.index("status") + 1, "stage")
    for row in rows:
        if not row.get("stage"):
            row["status"], row["stage"] = mapping.get(
                row["status"], (row["status"], row["status"])
            )
    temporary = path.with_suffix(path.suffix + ".tmp")
    with temporary.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(
            handle, fieldnames=columns, delimiter="\t", lineterminator="\n"
        )
        writer.writeheader()
        writer.writerows(rows)
    temporary.replace(path)


def require(condition: bool, message: str, failures: list[str]) -> None:
    if not condition:
        failures.append(message)


def status_summary(rows: list[dict[str, str]]) -> str:
    counts = Counter(row["status"] for row in rows)
    return ",".join(f"{status}:{counts[status]}" for status in STATUS_ORDER)


def dgroup_offset(address: str) -> int:
    selector, value = address.split(":", 1)
    if selector != "1028":
        raise ValueError(f"not a DGROUP address: {address}")
    return int(value, 16)


def audit(require_complete: bool = False) -> int:
    failures: list[str] = []
    digest = hashlib.sha256((ROOT / "TDKPIN.EXE").read_bytes()).hexdigest()
    require(digest == TARGET_SHA256, f"target SHA-256 changed: {digest}", failures)
    preserved_raw_digest = hashlib.sha256(
        (ROOT / "TDKPIN_GHIDRA_RAW.c").read_bytes()
    ).hexdigest()
    require(
        preserved_raw_digest == PRESERVED_RAW_SHA256,
        f"preserved raw decompiler evidence changed: {preserved_raw_digest}",
        failures,
    )

    functions = internal_functions()
    function_by_address = {row["address"]: row for row in functions}
    require(
        len(function_by_address) == len(functions),
        "duplicate binary function addresses",
        failures,
    )

    raw_markers: dict[str, str] = {}
    for line in (
        (ROOT / "TDKPIN_GHIDRA_COMPLETE_RAW.c").read_text(encoding="utf-8").splitlines()
    ):
        match = RAW_MARKER.match(line)
        if match:
            raw_markers[match.group(1)] = match.group(2)
    require(
        set(raw_markers) == set(function_by_address),
        "raw decompilation/function inventory mismatch",
        failures,
    )

    ledger_path = ROOT / "FUNCTION_RECONSTRUCTION.tsv"
    require(ledger_path.exists(), "FUNCTION_RECONSTRUCTION.tsv is missing", failures)
    ledger = read_tsv(ledger_path) if ledger_path.exists() else []
    ledger_by_address = {row["address"]: row for row in ledger}
    require(
        len(ledger_by_address) == len(ledger),
        "duplicate reconstruction-ledger addresses",
        failures,
    )
    require(
        set(ledger_by_address) == set(function_by_address),
        "reconstruction ledger/function inventory mismatch",
        failures,
    )

    for address, row in ledger_by_address.items():
        function = function_by_address[address]
        require(
            row["status"] in ALLOWED_STATUSES,
            f"{address}: invalid status {row['status']!r}",
            failures,
        )
        require(
            row["raw_name"] == function["name"], f"{address}: raw name drift", failures
        )
        require(
            row["segment"] == function["block"], f"{address}: segment drift", failures
        )
        require(
            row["body_bytes"] == function["body_bytes"],
            f"{address}: body-size drift",
            failures,
        )
        if row["status"] in {"restored", "verified"}:
            require(
                bool(row["reconstructed_name"]),
                f"{address}: restored row lacks a name",
                failures,
            )
            require(
                bool(row["module"]), f"{address}: restored row lacks a module", failures
            )
            require(
                (ROOT / row["module"]).is_file(),
                f"{address}: module does not exist",
                failures,
            )

    coverage = read_tsv(ROOT / "COMPLETE_COVERAGE.tsv")
    summaries = {
        row["name_or_start"]: row["bytes"]
        for row in coverage
        if row["record"] == "summary"
    }
    executable_exceptions = read_tsv(ROOT / "UNEXPLAINED.tsv")
    executable_exception_bytes = sum(
        int(row["bytes"]) for row in executable_exceptions
    )
    unclassified = [
        row
        for row in executable_exceptions
        if row["classification"] == "unclassified"
    ]
    require(
        executable_exception_bytes ==
            int(summaries["undefined-executable-bytes"]),
        "exception-byte evidence does not match coverage summary",
        failures,
    )

    references = read_tsv(ROOT / "REFERENCES.tsv")
    data_targets = {
        row["to_address"] for row in references if row["to_block"] == "Data6"
    }
    objects = read_tsv(ROOT / "OBJECTS.tsv")
    require(
        len(objects) == 175,
        f"expected 175 collision records, found {len(objects)}",
        failures,
    )

    support_ledgers = (
        "IMPORT_RECONSTRUCTION.tsv",
        "DATA_RECONSTRUCTION.tsv",
        "DATA_COVERAGE.tsv",
        "DATA_OBJECTS.tsv",
        "CODE_DATA_OBJECTS.tsv",
        "RESOURCE_RECONSTRUCTION.tsv",
        "CALLBACK_RECONSTRUCTION.tsv",
        "NE_SEGMENTS.tsv",
        "ENTRY_POINTS.tsv",
        "MODULE_REFERENCES.tsv",
        "RELOCATIONS.tsv",
        "SHARED_ENTRY_POINTS.tsv",
    )
    for name in support_ledgers:
        require((ROOT / name).exists(), f"{name} is missing", failures)
    shared_entries = read_tsv(ROOT / "SHARED_ENTRY_POINTS.tsv")
    for row in shared_entries:
        require(
            tuple(row) == SHARED_ENTRY_COLUMNS,
            f"shared entry {row.get('address', '?')}: schema drift",
            failures,
        )
        require(
            row["status"] in ALLOWED_STATUSES,
            f"shared entry {row['address']}: invalid status {row['status']!r}",
            failures,
        )
        require(
            row["containing_function"] in function_by_address,
            f"shared entry {row['address']}: unknown containing function",
            failures,
        )
        require(
            bool(row["reconstructed_name"]),
            f"shared entry {row['address']}: missing readable name",
            failures,
        )
    import_ledger: list[dict[str, str]] = []
    data_ledger: list[dict[str, str]] = []
    data_coverage: list[dict[str, str]] = []
    data_objects: list[dict[str, str]] = []
    code_data_objects: list[dict[str, str]] = []
    resource_ledger: list[dict[str, str]] = []
    if ROOT.joinpath("IMPORT_RECONSTRUCTION.tsv").exists():
        import_ledger = read_tsv(ROOT / "IMPORT_RECONSTRUCTION.tsv")
        external_addresses = {
            row["address"]
            for row in read_tsv(ROOT / "BINARY_FUNCTIONS.tsv")
            if row["entry_kind"] == "external"
        }
        require(
            {row["address"] for row in import_ledger} == external_addresses,
            "import ledger/inventory mismatch",
            failures,
        )
        for row in import_ledger:
            require(
                row["status"] in ALLOWED_STATUSES,
                f"import {row['address']}: invalid status {row['status']!r}",
                failures,
            )
    if ROOT.joinpath("DATA_RECONSTRUCTION.tsv").exists():
        data_ledger = read_tsv(ROOT / "DATA_RECONSTRUCTION.tsv")
        require(
            {row["address"] for row in data_ledger} == data_targets,
            "data ledger/reference-target mismatch",
            failures,
        )
        for row in data_ledger:
            require(
                row["status"] in ALLOWED_STATUSES,
                f"data {row['address']}: invalid status {row['status']!r}",
                failures,
            )
    if ROOT.joinpath("DATA_COVERAGE.tsv").exists():
        data_coverage = read_tsv(ROOT / "DATA_COVERAGE.tsv")
        next_offset = 0
        for row in data_coverage:
            start = dgroup_offset(row["start"])
            end = dgroup_offset(row["end"])
            require(
                start == next_offset,
                f"DGROUP coverage gap/overlap at {row['start']}",
                failures,
            )
            require(
                end >= start,
                f"invalid DGROUP coverage range at {row['start']}",
                failures,
            )
            require(
                int(row["bytes"]) == end - start + 1,
                f"DGROUP range length mismatch at {row['start']}",
                failures,
            )
            require(
                row["status"] in ALLOWED_STATUSES,
                f"DGROUP range {row['start']}: invalid status {row['status']!r}",
                failures,
            )
            next_offset = end + 1
        require(
            next_offset == 0x4D3A,
            f"DGROUP coverage ends at {next_offset:04x}",
            failures,
        )
    if ROOT.joinpath("DATA_OBJECTS.tsv").exists():
        data_objects = read_tsv(ROOT / "DATA_OBJECTS.tsv")
        previous_end = -1
        for row in sorted(data_objects, key=lambda item: dgroup_offset(item["start"])):
            start = dgroup_offset(row["start"])
            end = dgroup_offset(row["end"])
            require(
                start > previous_end,
                f"overlapping data object at {row['start']}",
                failures,
            )
            require(
                row["status"] in ALLOWED_STATUSES,
                f"data object {row['object_id']}: invalid status {row['status']!r}",
                failures,
            )
            previous_end = end
    if ROOT.joinpath("CODE_DATA_OBJECTS.tsv").exists():
        code_data_objects = read_tsv(ROOT / "CODE_DATA_OBJECTS.tsv")
        code_data_inventory = {
            row["address"]: int(row["length"])
            for row in read_tsv(ROOT / "DEFINED_DATA.tsv")
            if row["block"].startswith("Code")
        }
        require(
            {row["start"]: int(row["bytes"]) for row in code_data_objects}
            == code_data_inventory,
            "code-data ledger/defined-data inventory mismatch",
            failures,
        )
        for row in code_data_objects:
            require(
                tuple(row) == CODE_DATA_COLUMNS,
                f"code-data object {row.get('start', '?')}: schema drift",
                failures,
            )
            start_selector, start_text = row["start"].split(":", 1)
            end_selector, end_text = row["end"].split(":", 1)
            require(
                start_selector == end_selector,
                f"code-data object {row['start']}: crosses segments",
                failures,
            )
            require(
                int(row["bytes"]) == int(end_text, 16) - int(start_text, 16) + 1,
                f"code-data object {row['start']}: extent mismatch",
                failures,
            )
            require(
                row["status"] in ALLOWED_STATUSES,
                f"code-data object {row['object_id']}: invalid status {row['status']!r}",
                failures,
            )
    if ROOT.joinpath("RESOURCE_RECONSTRUCTION.tsv").exists():
        resource_ledger = read_tsv(ROOT / "RESOURCE_RECONSTRUCTION.tsv")
        manifest = json.loads(
            (ROOT / "assets/manifest.json").read_text(encoding="utf-8")
        )
        resource_keys = {
            (str(row["type"]), str(row["id"])) for row in manifest["resources"]
        }
        require(
            {(row["type"], row["id"]) for row in resource_ledger} == resource_keys,
            "resource ledger/manifest mismatch",
            failures,
        )
        for row in resource_ledger:
            require(
                row["status"] in ALLOWED_STATUSES,
                f"resource {row['type']}:{row['id']}: invalid status {row['status']!r}",
                failures,
            )

    callback_ledger = (
        read_tsv(ROOT / "CALLBACK_RECONSTRUCTION.tsv")
        if ROOT.joinpath("CALLBACK_RECONSTRUCTION.tsv").exists()
        else []
    )
    entry_points = (
        read_tsv(ROOT / "ENTRY_POINTS.tsv")
        if ROOT.joinpath("ENTRY_POINTS.tsv").exists()
        else []
    )
    relocations = (
        read_tsv(ROOT / "RELOCATIONS.tsv")
        if ROOT.joinpath("RELOCATIONS.tsv").exists()
        else []
    )
    require(
        len(entry_points) == 301,
        f"expected 301 NE entries, found {len(entry_points)}",
        failures,
    )
    for row in callback_ledger:
        require(
            row["status"] in ALLOWED_STATUSES,
            f"callback {row['registration_site']}: invalid status {row['status']!r}",
            failures,
        )

    if require_complete:
        require(
            not unclassified, "completion requires zero unclassified ranges", failures
        )
        for label, rows in (
            ("function", ledger),
            ("import", import_ledger),
            ("data", data_ledger),
            ("DGROUP coverage", data_coverage),
            ("data object", data_objects),
            ("code-data object", code_data_objects),
            ("resource", resource_ledger),
            ("callback", callback_ledger),
            ("shared entry", shared_entries),
        ):
            incomplete = [row for row in rows if row["status"] != "verified"]
            require(
                not incomplete,
                f"completion requires every {label} row verified; {len(incomplete)} remain",
                failures,
            )

    print(f"target_sha256={digest}")
    print(f"preserved_raw_sha256={preserved_raw_digest}")
    print(f"internal_functions={len(functions)}")
    print(f"function_statuses={status_summary(ledger)}")
    print(f"import_statuses={status_summary(import_ledger)}")
    print(f"data_statuses={status_summary(data_ledger)}")
    print(f"data_range_statuses={status_summary(data_coverage)}")
    print(f"data_object_statuses={status_summary(data_objects)}")
    print(f"code_data_object_statuses={status_summary(code_data_objects)}")
    print(f"resource_statuses={status_summary(resource_ledger)}")
    print(f"callback_statuses={status_summary(callback_ledger)}")
    print(f"shared_entry_statuses={status_summary(shared_entries)}")
    print(f"executable_bytes={summaries['executable-bytes']}")
    print(f"classified_executable_exception_ranges={len(executable_exceptions)}")
    print(f"classified_executable_exception_bytes={executable_exception_bytes}")
    print(f"unclassified_executable_ranges={len(unclassified)}")
    print(f"defined_data_units={len(read_tsv(ROOT / 'DEFINED_DATA.tsv'))}")
    print(f"referenced_data_addresses={len(data_targets)}")
    print(f"collision_records={len(objects)}")
    print(f"ne_entry_points={len(entry_points)}")
    print(f"relocation_sites={len(relocations)}")
    if failures:
        for failure in failures:
            print(f"FAIL: {failure}")
        return 1
    print("ledger_consistency=pass")
    return 0


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--initialize", action="store_true", help="create the initial function ledger"
    )
    parser.add_argument(
        "--sync",
        action="store_true",
        help="add newly recovered functions to the ledger",
    )
    parser.add_argument(
        "--initialize-support-ledgers",
        action="store_true",
        help="create import, data, resource, and callback ledgers",
    )
    parser.add_argument(
        "--sync-support-ledgers",
        action="store_true",
        help="refresh mechanically derived support-ledger fields",
    )
    parser.add_argument(
        "--require-complete",
        action="store_true",
        help="fail unless every required ledger row is verified",
    )
    args = parser.parse_args()
    if args.initialize:
        initialize_ledger(ROOT / "FUNCTION_RECONSTRUCTION.tsv")
    if args.sync:
        sync_ledger(ROOT / "FUNCTION_RECONSTRUCTION.tsv")
    if args.initialize_support_ledgers:
        initialize_support_ledgers()
    if args.sync_support_ledgers:
        sync_support_ledgers()
    return audit(args.require_complete)


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