tdkpin/original/tools/build_data_coverage.py

#!/usr/bin/env python3
"""Build a byte-complete, non-overlapping DGROUP reconstruction ledger."""

from __future__ import annotations

import csv
import hashlib
from itertools import pairwise
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
OUTPUT = ROOT / "DATA_COVERAGE.tsv"
COLUMNS = (
    "start",
    "end",
    "bytes",
    "storage",
    "status",
    "stage",
    "object_id",
    "object_name",
    "c_type",
    "initial_value",
    "readers",
    "writers",
    "relocations",
    "evidence",
    "notes",
)
RELOCATION_WIDTHS = {
    "low-byte": 1,
    "selector-16": 2,
    "far-pointer-32": 4,
    "offset-16": 2,
    "offset-32": 4,
    "pointer-48": 6,
}


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 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 address(value: int) -> str:
    return f"1028:{value:04x}"


def initial_value(payload: bytes, start: int, end: int, file_length: int) -> str:
    if start >= file_length:
        return "zero-fill"
    value = payload[start : min(end + 1, file_length)]
    if end + 1 > file_length:
        value += bytes(end + 1 - file_length)
    if len(value) <= 32:
        return value.hex()
    return f"sha256:{hashlib.sha256(value).hexdigest()}"


def project_semantic_objects(
    objects: list[dict[str, str]],
) -> None:
    path = ROOT / "DATA_RECONSTRUCTION.tsv"
    rows = read_tsv(path)
    with path.open(newline="", encoding="utf-8") as handle:
        columns = tuple(csv.DictReader(handle, delimiter="\t").fieldnames or ())
    for row in rows:
        anchor = offset(row["address"])
        matches = [
            item
            for item in objects
            if offset(item["start"]) <= anchor <= offset(item["end"])
        ]
        if len(matches) > 1:
            raise ValueError(f"overlapping semantic objects at {row['address']}")
        if not matches:
            continue
        item = matches[0]
        row["status"] = item["status"]
        row["stage"] = "semantic-object-field"
        row["object_name"] = item["object_name"]
        row["type"] = item["c_type"]
        row["extent_start"] = item["start"]
        row["extent_end"] = item["end"]
        row["initialization"] = item["initialization"]
        row["evidence"] = item["evidence"]
        row["notes"] = item["notes"]
    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 main() -> int:
    segments = read_tsv(ROOT / "NE_SEGMENTS.tsv")
    data_segment = next(row for row in segments if row["segment"] == "6")
    file_offset = int(data_segment["file_offset"])
    file_length = int(data_segment["file_length"])
    allocation_size = int(data_segment["allocation_size"])
    binary = (ROOT / "TDKPIN.EXE").read_bytes()
    payload = binary[file_offset : file_offset + file_length]

    references = [
        row for row in read_tsv(ROOT / "REFERENCES.tsv") if row["to_block"] == "Data6"
    ]
    defined_data = [
        row for row in read_tsv(ROOT / "DEFINED_DATA.tsv") if row["block"] == "Data6"
    ]
    relocations = [
        row for row in read_tsv(ROOT / "RELOCATIONS.tsv") if row["segment"] == "6"
    ]
    objects = (
        read_tsv(ROOT / "DATA_OBJECTS.tsv")
        if (ROOT / "DATA_OBJECTS.tsv").exists()
        else []
    )
    project_semantic_objects(objects)

    boundaries = {0, file_length, allocation_size}
    for row in references:
        boundaries.add(offset(row["to_address"]))
    for row in defined_data:
        start = offset(row["address"])
        boundaries.add(start)
        boundaries.add(min(start + int(row["length"]), allocation_size))
    for row in relocations:
        start = offset(row["source_address"])
        width = RELOCATION_WIDTHS.get(row["source_type"])
        if width is None:
            raise ValueError(f"unknown relocation width: {row['source_type']}")
        boundaries.add(start)
        boundaries.add(min(start + width, allocation_size))
    for row in objects:
        start = offset(row["start"])
        end = offset(row["end"])
        boundaries.add(start)
        boundaries.add(min(end + 1, allocation_size))
    ordered = sorted(
        boundary for boundary in boundaries if 0 <= boundary <= allocation_size
    )

    old = {}
    if OUTPUT.exists():
        old = {(row["start"], row["end"]): row for row in read_tsv(OUTPUT)}

    rows = []
    current_keys = set()
    for start, next_start in pairwise(ordered):
        if start == next_start:
            continue
        end = next_start - 1
        key = (address(start), address(end))
        current_keys.add(key)
        row = old.get(key, {})
        range_refs = [
            reference
            for reference in references
            if start <= offset(reference["to_address"]) <= end
        ]
        range_data = [
            data for data in defined_data if start <= offset(data["address"]) <= end
        ]
        range_relocations = [
            relocation
            for relocation in relocations
            if start <= offset(relocation["source_address"]) <= end
        ]
        range_objects = [
            item
            for item in objects
            if offset(item["start"]) <= start and end <= offset(item["end"])
        ]
        if len(range_objects) > 1:
            raise ValueError(f"overlapping semantic data objects at {key[0]}-{key[1]}")
        semantic_object = range_objects[0] if range_objects else None
        row.update(
            {
                "start": key[0],
                "end": key[1],
                "bytes": end - start + 1,
                "storage": "initialized" if end < file_length else "zero-fill",
                "status": (
                    semantic_object["status"]
                    if semantic_object
                    else row.get("status", "unknown")
                ),
                "stage": (
                    "semantic-object"
                    if semantic_object
                    else row.get("stage", "candidate-range")
                ),
                "object_id": (
                    semantic_object["object_id"]
                    if semantic_object
                    else row.get("object_id", "")
                ),
                "object_name": (
                    semantic_object["object_name"]
                    if semantic_object
                    else row.get("object_name", "")
                ),
                "c_type": (
                    semantic_object["c_type"]
                    if semantic_object
                    else row.get("c_type", "")
                ),
                "initial_value": initial_value(payload, start, end, file_length),
                "readers": ",".join(
                    sorted(
                        {
                            reference["from_function"]
                            for reference in range_refs
                            if "READ" in reference["reference_type"]
                            and reference["from_function"]
                        }
                    )
                ),
                "writers": ",".join(
                    sorted(
                        {
                            reference["from_function"]
                            for reference in range_refs
                            if "WRITE" in reference["reference_type"]
                            and reference["from_function"]
                        }
                    )
                ),
                "relocations": ",".join(
                    f"{item['source_address']}->{item['target']}"
                    for item in range_relocations
                ),
                "evidence": ";".join(
                    [
                        *([semantic_object["evidence"]] if semantic_object else []),
                        *[
                            f"{item['address']}:{item['data_type']}[{item['length']}]"
                            for item in range_data
                        ],
                    ]
                ),
                "notes": (
                    semantic_object["notes"]
                    if semantic_object
                    else row.get(
                        "notes",
                        "Candidate byte range; semantic object grouping and type remain unresolved.",
                    )
                ),
            }
        )
        rows.append(row)

    lost_review = [
        row
        for key, row in old.items()
        if key not in current_keys and row["status"] != "unknown"
    ]
    unsafe_lost_review = []
    for reviewed in lost_review:
        reviewed_start = offset(reviewed["start"])
        reviewed_end = offset(reviewed["end"])
        replacements = [
            row
            for row in rows
            if offset(row["end"]) >= reviewed_start
            and offset(row["start"]) <= reviewed_end
        ]
        cursor = reviewed_start
        safe = True
        for replacement in replacements:
            replacement_start = max(offset(replacement["start"]), reviewed_start)
            replacement_end = min(offset(replacement["end"]), reviewed_end)
            if replacement_start != cursor or replacement["status"] != "verified" or (
                replacement["stage"] != "semantic-object"
            ):
                safe = False
                break
            cursor = replacement_end + 1
        if not safe or cursor != reviewed_end + 1:
            unsafe_lost_review.append(reviewed)
    if unsafe_lost_review:
        raise RuntimeError(
            "refusing to discard "
            f"{len(unsafe_lost_review)} reviewed data ranges after boundary drift"
        )

    temporary = OUTPUT.with_suffix(OUTPUT.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(OUTPUT)
    print(f"data_ranges={len(rows)}")
    print(f"covered_bytes={sum(int(row['bytes']) for row in rows)}")
    print(f"initialized_bytes={file_length}")
    print(f"zero_fill_bytes={allocation_size - file_length}")
    return 0


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