DataOps-platform/scripts/validate_domain_replication.py

#!/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