[AUTOMATED] Add bounded PDB MSF stream reader

decbench/utils/pdb_msf.py

@@ -0,0 +1,109 @@
+"""Bounded MSF 7 stream and PDB information transport.
+
+This module does not find PDBs, match them to PE files or feed any metric.
+Unsupported or invalid MSF encodings raise PdbFormatError; operating-system
+read failures are left to the caller.
+"""
+
+from __future__ import annotations
+
+import os
+import struct
+import uuid
+from dataclasses import dataclass
+from pathlib import Path
+
+_MAGIC = b"Microsoft C/C++ MSF 7.00\r\n\x1aDS\0\0\0"
+_MAX_BYTES = 512 * 1024 * 1024
+_MAX_STREAMS = 1_000_000
+
+
+class PdbFormatError(ValueError):
+    """An invalid or unsupported MSF/PDB representation."""
+
+
+def _unpack(fmt: str, data: bytes, offset: int = 0):
+    size = struct.calcsize(fmt)
+    if offset < 0 or offset + size > len(data):
+        raise PdbFormatError("truncated MSF structure")
+    return struct.unpack_from(fmt, data, offset)
+
+
+@dataclass(frozen=True)
+class PdbIdentity:
+    version: int
+    signature: int
+    age: int
+    guid: uuid.UUID
+
+
+class PdbMsf:
+    """An in-memory bounded MSF 7 container, with no PE identity assertion."""
+
+    def __init__(self, path: Path):
+        with path.open("rb") as stream:
+            size = os.fstat(stream.fileno()).st_size
+            if size > _MAX_BYTES:
+                raise PdbFormatError("PDB exceeds the supported 512 MiB bound")
+            self._data = stream.read(size + 1)
+        if len(self._data) != size:
+            raise PdbFormatError("PDB changed while being read")
+        if self._data[:32] != _MAGIC:
+            raise PdbFormatError("not an MSF 7 PDB")
+        block_size, _, count, directory_size, _, directory_map = _unpack("<6I", self._data, 32)
+        if block_size not in (512, 1024, 2048, 4096) or len(self._data) != count * block_size:
+            raise PdbFormatError("invalid MSF block geometry")
+        self._block_size = block_size
+        directory_blocks = (directory_size + block_size - 1) // block_size
+        # This bounded implementation supports one block-map block only.
+        if directory_blocks > block_size // 4 or directory_size > len(self._data):
+            raise PdbFormatError("unsupported MSF directory extent")
+        blocks = _unpack(f"<{directory_blocks}I", self._data, directory_map * block_size)
+        directory = self._read_blocks(blocks, directory_size)
+        nstreams = _unpack("<I", directory)[0]
+        if nstreams > _MAX_STREAMS or 4 + 4 * nstreams > len(directory):
+            raise PdbFormatError("invalid MSF stream count")
+        sizes = _unpack(f"<{nstreams}I", directory, 4)
+        entries: list[tuple[tuple[int, ...], int] | None] = []
+        offset = 4 + 4 * nstreams
+        for size in sizes:
+            if size == 0xFFFFFFFF:
+                entries.append(None)
+                continue
+            if size > len(self._data):
+                raise PdbFormatError("invalid MSF stream size")
+            nblocks = (size + block_size - 1) // block_size
+            blocks = _unpack(f"<{nblocks}I", directory, offset)
+            offset += nblocks * 4
+            if any(b >= count for b in blocks):
+                raise PdbFormatError("MSF stream block outside file")
+            entries.append((blocks, size))
+        self._streams = tuple(entries)
+
+    def _read_blocks(self, blocks: tuple[int, ...], size: int) -> bytes:
+        if any((b + 1) * self._block_size > len(self._data) for b in blocks):
+            raise PdbFormatError("MSF block outside file")
+        return b"".join(
+            self._data[b * self._block_size : (b + 1) * self._block_size] for b in blocks
+        )[:size]
+
+    @property
+    def stream_count(self) -> int:
+        return len(self._streams)
+
+    def stream(self, index: int) -> bytes:
+        if not 0 <= index < len(self._streams):
+            raise PdbFormatError(f"missing PDB stream {index}")
+        entry = self._streams[index]
+        if entry is None:
+            raise PdbFormatError(f"missing PDB stream {index}")
+        blocks, size = entry
+        return self._read_blocks(blocks, size)
+
+    def identity(self) -> PdbIdentity:
+        """Read the PDB information stream; the caller must match PE GUID/age."""
+        data = self.stream(1)
+        version, signature, age = _unpack("<III", data)
+        if len(data) < 28 or not any(data[12:28]):
+            raise PdbFormatError("invalid or missing PDB GUID")
+        return PdbIdentity(version, signature, age, uuid.UUID(bytes_le=data[12:28]))

docs/pdb-msf-reader.md

@@ -0,0 +1,18 @@
+# Experimental PDB MSF transport
+
+`decbench.utils.pdb_msf` reads streams and the information-stream identity from
+bounded MSF 7 files. It is a standalone building block; no evaluator, default
+metric, cache, or backend uses it. It does not locate a PDB or match one to a PE.
+Before treating PDB contents as ground truth, a future integration must match
+the PE CodeView record's GUID and age and define its own evidence contract.
+
+The reader accepts 512, 1024, 2048, and 4096 byte blocks, a file up to 512 MiB,
+at most one million stream entries, and a directory whose block map fits in one
+block. Unsupported formats and malformed structures raise `PdbFormatError`;
+file-system errors propagate. This is not a comprehensive PDB validator, and
+the identity fields alone do not authenticate the file. Types, symbols, source
+locations, function ranges, and PE pairing are outside this module's scope.
+
+Tests include synthetic containers and a compiler-produced PE/PDB fixture,
+whose information is compared with `llvm-pdbutil` when the tools are available.
+The fixture is cross-compiled; it is not a native Windows execution test.

tests/test_pdb_msf.py

@@ -0,0 +1,107 @@
+"""PDB stream extraction only; this does not qualify a PE/PDB pair."""
+
+import re
+import shutil
+import struct
+import subprocess
+import uuid
+from pathlib import Path
+
+import pytest
+
+from decbench.utils import pdb_msf
+
+
+def _minimal_msf() -> bytes:
+    block_size = 512
+    data = bytearray(5 * block_size)
+    data[:32] = pdb_msf._MAGIC
+    # block 1 maps block 2, the directory; info stream occupies block 3.
+    struct.pack_into("<6I", data, 32, block_size, 0, 5, 16, 0, 1)
+    struct.pack_into("<I", data, block_size, 2)
+    struct.pack_into("<IIII", data, 2 * block_size, 2, 0, 28, 3)
+    guid = uuid.UUID("12345678-1234-5678-9abc-def012345678")
+    struct.pack_into("<III", data, 3 * block_size, 20000404, 19, 1)
+    data[3 * block_size + 12 : 3 * block_size + 28] = guid.bytes_le
+    return bytes(data)
+
+
+def test_minimal_streams_and_identity(tmp_path: Path) -> None:
+    path = tmp_path / "x.pdb"
+    path.write_bytes(_minimal_msf())
+    msf = pdb_msf.PdbMsf(path)
+    assert msf.stream_count == 2
+    assert msf.stream(0) == b""
+    assert str(msf.identity().guid) == "12345678-1234-5678-9abc-def012345678"
+    assert msf.identity().age == 1
+    with pytest.raises(pdb_msf.PdbFormatError):
+        msf.stream(2)
+
+
+def test_rejects_bad_geometry_truncation_and_out_of_bounds(tmp_path: Path) -> None:
+    path = tmp_path / "x.pdb"
+    original = _minimal_msf()
+    for mutated in (b"", original[:-1], b"bad" + original[3:]):
+        path.write_bytes(mutated)
+        with pytest.raises(pdb_msf.PdbFormatError):
+            pdb_msf.PdbMsf(path)
+    for offset, value in ((32, 123), (40, 10), (52, 99), (512, 99), (1036, 99)):
+        data = bytearray(original)
+        struct.pack_into("<I", data, offset, value)
+        path.write_bytes(data)
+        with pytest.raises(pdb_msf.PdbFormatError):
+            pdb_msf.PdbMsf(path)
+
+
+def test_rejects_oversized_and_zero_guid(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    path = tmp_path / "x.pdb"
+    data = bytearray(_minimal_msf())
+    path.write_bytes(data)
+    monkeypatch.setattr(pdb_msf, "_MAX_BYTES", 50)
+    with pytest.raises(pdb_msf.PdbFormatError):
+        pdb_msf.PdbMsf(path)
+    monkeypatch.setattr(pdb_msf, "_MAX_BYTES", len(data))
+    data[3 * 512 + 12 : 3 * 512 + 28] = bytes(16)
+    path.write_bytes(data)
+    with pytest.raises(pdb_msf.PdbFormatError):
+        pdb_msf.PdbMsf(path).identity()
+
+
+def test_generated_pdb_matches_llvm_pdbutil(tmp_path: Path) -> None:
+    compiler, linker, inspector = (
+        shutil.which(name) for name in ("clang-cl", "lld-link", "llvm-pdbutil")
+    )
+    if not all((compiler, linker, inspector)):
+        pytest.skip("clang-cl, lld-link and llvm-pdbutil are required")
+    source, obj, pdb, exe = (
+        tmp_path / name for name in ("main.c", "main.obj", "main.pdb", "main.exe")
+    )
+    source.write_text("int main(void) { return 0; }\n")
+    subprocess.run(
+        [compiler, "/c", "/Z7", "/Od", "/GS-", str(source), f"/Fo{obj}"],
+        check=True,
+        capture_output=True,
+    )
+    subprocess.run(
+        [
+            linker,
+            "/entry:main",
+            "/nodefaultlib",
+            "/subsystem:console",
+            "/debug:full",
+            f"/pdb:{pdb}",
+            f"/out:{exe}",
+            str(obj),
+        ],
+        check=True,
+        capture_output=True,
+    )
+    summary = subprocess.check_output([inspector, "dump", "-summary", str(pdb)], text=True)
+    guid = re.search(r"GUID:\s*\{([0-9A-Fa-f-]+)\}", summary)
+    age = re.search(r"Age:\s*(\d+)", summary)
+    streams = re.search(r"Number of streams:\s*(\d+)", summary)
+    assert guid and age and streams
+    msf = pdb_msf.PdbMsf(pdb)
+    assert msf.identity().guid == uuid.UUID(guid.group(1))
+    assert msf.identity().age == int(age.group(1))
+    assert msf.stream_count == int(streams.group(1))