#!/usr/bin/env python3 """Validate corpus provenance, coverage, hashes, and generator reproducibility.""" from __future__ import annotations import hashlib import importlib.util import json import re import sys from pathlib import Path, PurePosixPath from typing import Any REQUIRED_CATEGORIES = { "digital-text", "links", "tables-forms", "vectors", "rtl", "cjk", "ocr-scanned", "rotated-mixed-size", "catalog-magazine", "malformed", "bounded-hostile", } SHA256 = re.compile(r"^[0-9a-f]{64}$") REDISTRIBUTABLE_LICENSES = {"CC0-1.0"} MAX_FIXTURE_BYTES = 64 * 1024 def fail(message: str) -> None: raise ValueError(message) def load_generator(root: Path): path = root / "generate_fixtures.py" spec = importlib.util.spec_from_file_location("corpus_generator", path) if spec is None or spec.loader is None: fail(f"cannot load generator: {path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def validate(manifest_path: Path) -> tuple[int, int]: root = manifest_path.resolve().parent data: Any = json.loads(manifest_path.read_text(encoding="utf-8")) if not isinstance(data, dict) or data.get("schema_version") != 1: fail("schema_version must be 1") generator_meta = data.get("generator") if not isinstance(generator_meta, dict) or set(generator_meta) != {"path", "version", "runtime"}: fail("generator metadata must contain exactly path, version, and runtime") if generator_meta["path"] != "generate_fixtures.py": fail("generator path must be package-local generate_fixtures.py") if data.get("corpus_license") not in REDISTRIBUTABLE_LICENSES: fail("corpus_license is not approved as redistributable") fixtures = data.get("fixtures") if not isinstance(fixtures, list) or not fixtures: fail("fixtures must be a non-empty array") ids: set[str] = set() paths: set[str] = set() categories: set[str] = set() total_bytes = 0 module = load_generator(root) if str(module.GENERATOR_VERSION) != str(generator_meta["version"]): fail("manifest generator version does not match generator source") generated_by_path: dict[str, bytes] = {} provenance_by_path: dict[str, str] = {} for name, factory in module.GENERATORS.items(): fixture_rel = f"fixtures/{name}" generated_by_path[fixture_rel] = factory() provenance_by_path[fixture_rel] = ( f"generated:{generator_meta['path']}@{generator_meta['version']}:{factory.__name__}" ) for index, item in enumerate(fixtures): where = f"fixtures[{index}]" required = {"id", "path", "license", "source", "sha256", "category", "expected_capabilities"} if not isinstance(item, dict) or set(item) != required: fail(f"{where} must contain exactly {sorted(required)}") if not isinstance(item["id"], str) or not item["id"] or item["id"] in ids: fail(f"{where}.id must be a unique non-empty string") ids.add(item["id"]) rel = PurePosixPath(item["path"]) if rel.is_absolute() or ".." in rel.parts or len(rel.parts) != 2 or rel.parts[0] != "fixtures": fail(f"{where}.path must be a direct relative child of fixtures/") if rel.suffix.lower() != ".pdf" or item["path"] in paths: fail(f"{where}.path must be a unique PDF path") paths.add(item["path"]) if item["license"] not in REDISTRIBUTABLE_LICENSES: fail(f"{where}.license is not approved as redistributable") expected_source = provenance_by_path.get(item["path"]) if not isinstance(item["source"], str) or item["source"] != expected_source: fail(f"{where}.source must exactly match generator version and factory: {expected_source}") if not isinstance(item["sha256"], str) or not SHA256.fullmatch(item["sha256"]): fail(f"{where}.sha256 must be a lowercase SHA-256 digest") if item["category"] not in REQUIRED_CATEGORIES: fail(f"{where}.category is not a required corpus category") categories.add(item["category"]) capabilities = item["expected_capabilities"] if not isinstance(capabilities, dict) or not capabilities: fail(f"{where}.expected_capabilities must be a non-empty object") fixture_root = root / "fixtures" fixture_path = root / Path(*rel.parts) if fixture_path.is_symlink(): fail(f"fixture must not be a symlink: {item['path']}") if fixture_path.parent != fixture_root or fixture_path.resolve().parent != fixture_root.resolve(): fail(f"fixture must resolve to a direct child of fixtures/: {item['path']}") if not fixture_path.is_file(): fail(f"missing fixture: {item['path']}") content = fixture_path.read_bytes() total_bytes += len(content) if len(content) > MAX_FIXTURE_BYTES: fail(f"fixture exceeds deterministic corpus bound ({MAX_FIXTURE_BYTES} bytes): {item['path']}") if not content.startswith(b"%PDF-") or not content.rstrip().endswith(b"%%EOF"): fail(f"fixture lacks PDF header or EOF marker: {item['path']}") actual_hash = hashlib.sha256(content).hexdigest() if actual_hash != item["sha256"]: fail(f"SHA-256 mismatch for {item['path']}: expected {item['sha256']}, got {actual_hash}") generated = generated_by_path.get(item["path"]) if generated is None: fail(f"manifest fixture has no generator entry: {item['path']}") if generated != content: fail(f"generator output is not byte-identical: {item['path']}") missing = REQUIRED_CATEGORIES - categories if missing: fail(f"missing required categories: {sorted(missing)}") disk_paths = {p.relative_to(root).as_posix() for p in (root / "fixtures").glob("*.pdf")} if disk_paths != paths: fail(f"manifest/disk fixture set mismatch: unlisted={sorted(disk_paths-paths)}, missing={sorted(paths-disk_paths)}") if set(generated_by_path) != paths: fail("manifest/generator fixture set mismatch") return len(fixtures), total_bytes def main(argv: list[str]) -> int: if len(argv) != 2: print(f"usage: {argv[0]} MANIFEST.json", file=sys.stderr) return 2 try: count, total = validate(Path(argv[1])) except (OSError, json.JSONDecodeError, ValueError) as error: print(f"manifest validation failed: {error}", file=sys.stderr) return 1 print(f"manifest valid: {count} fixtures, {len(REQUIRED_CATEGORIES)} categories, {total} bytes") return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv))