#!/usr/bin/env python3 """Validate a self-contained domain replication implementation package.""" from __future__ import annotations import argparse import csv import hashlib import io import json import os import re import stat import sys from pathlib import Path ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from app.core.governance.domain_replication import ( # noqa: E402 DomainReplicationError, evaluate_replication_package, ) _SHA256 = re.compile(r"^[0-9a-f]{64}$") def _json(content: bytes, label: str) -> dict: value = json.loads(content.decode("utf-8")) if not isinstance(value, dict): raise DomainReplicationError(f"JSON object required: {label}") return value def _open_directory_chain(path: Path) -> int: path = path.absolute() if not path.is_absolute(): raise DomainReplicationError("package directory must be absolute") fd = os.open("/", os.O_RDONLY | os.O_DIRECTORY) try: for component in path.parts[1:]: next_fd = os.open( component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd, ) os.close(fd) fd = next_fd details = os.fstat(fd) if not stat.S_ISDIR(details.st_mode) or details.st_nlink < 1: raise DomainReplicationError("package root is not a controlled directory") return fd except OSError as exc: os.close(fd) raise DomainReplicationError("package directory symlink or replacement was rejected") from exc def _relative_parts(relative: str) -> tuple[str, ...]: path = Path(relative) if path.is_absolute() or not path.parts or any(part in {"", ".", ".."} for part in path.parts): raise DomainReplicationError(f"package path escapes package directory: {relative}") return path.parts def _read_regular_once(root_fd: int, relative: str) -> bytes: parts = _relative_parts(relative) directory_fd = os.dup(root_fd) file_fd = -1 try: for component in parts[:-1]: next_fd = os.open( component, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory_fd, ) os.close(directory_fd) directory_fd = next_fd file_fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW, dir_fd=directory_fd) before = os.fstat(file_fd) if not stat.S_ISREG(before.st_mode): raise DomainReplicationError(f"package input must be a regular file: {relative}") if before.st_nlink != 1: raise DomainReplicationError(f"package input hardlink was rejected: {relative}") chunks: list[bytes] = [] while True: chunk = os.read(file_fd, 1024 * 1024) if not chunk: break chunks.append(chunk) if sum(map(len, chunks)) > 8 * 1024 * 1024: raise DomainReplicationError(f"package input exceeds size limit: {relative}") content = b"".join(chunks) after = os.fstat(file_fd) fingerprint = (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) if fingerprint != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns) or len(content) != before.st_size: raise DomainReplicationError(f"package input changed while being read: {relative}") return content except OSError as exc: raise DomainReplicationError(f"package input symlink or replacement was rejected: {relative}") from exc finally: if file_fd >= 0: os.close(file_fd) os.close(directory_fd) def _rows(content: bytes, label: str) -> list[dict[str, str]]: try: handle = io.StringIO(content.decode("utf-8"), newline="") return list(csv.DictReader(handle)) except UnicodeDecodeError as exc: raise DomainReplicationError(f"CSV must be UTF-8: {label}") from exc def validate_package(package_dir: Path, *, verify_acceptance_report: bool = True) -> dict: root_fd = _open_directory_chain(package_dir) try: blobs: dict[str, bytes] = {"manifest": _read_regular_once(root_fd, "manifest.json")} manifest = _json(blobs["manifest"], "manifest.json") files = manifest.get("files") if not isinstance(files, dict): raise DomainReplicationError("manifest.files must be an object") required_files = ("template", "evidence", "snapshot", "delta") if int(manifest.get("schema_version", 0)) == 2: required_files += ("execution_evidence",) optional_files = ("acceptance_report",) for key in (*required_files, *optional_files): binding = files.get(key) if key in optional_files and binding is None: continue if not isinstance(binding, dict): raise DomainReplicationError(f"manifest.files.{key} is required") relative = str(binding.get("path", "")) expected = str(binding.get("sha256", "")) if not _SHA256.fullmatch(expected): raise DomainReplicationError(f"manifest.files.{key}.sha256 is invalid") content = _read_regular_once(root_fd, relative) actual = hashlib.sha256(content).hexdigest() if actual != expected: raise DomainReplicationError( f"package file digest mismatch for {key}: expected={expected}, actual={actual}" ) blobs[key] = content report = evaluate_replication_package( manifest, _json(blobs["evidence"], "evidence"), template=_json(blobs["template"], "template"), snapshot_rows=_rows(blobs["snapshot"], "snapshot"), delta_rows=_rows(blobs["delta"], "delta"), execution_evidence=( _json(blobs["execution_evidence"], "execution_evidence") if "execution_evidence" in blobs else None ), ) if "acceptance_report" in blobs and verify_acceptance_report: rendered = json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n" if blobs["acceptance_report"].decode("utf-8") != rendered: raise DomainReplicationError( "acceptance report does not match the deterministic package result" ) return report finally: os.close(root_fd) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--package-dir", type=Path, required=True) parser.add_argument("--output", default="-") parser.add_argument("--refresh-acceptance-report", action="store_true") args = parser.parse_args(argv) report = validate_package( args.package_dir, verify_acceptance_report=not args.refresh_acceptance_report, ) rendered = json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n" if args.output == "-": sys.stdout.write(rendered) else: output = Path(args.output) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(rendered, encoding="utf-8") return 0 if __name__ == "__main__": try: raise SystemExit(main()) except (DomainReplicationError, OSError, ValueError, json.JSONDecodeError) as exc: print(f"ERROR: {exc}", file=sys.stderr) raise SystemExit(2) from exc