DataOps-platform/scripts/generate_p3_wp14_release_manifest.py

"""Generate the unsigned, dirty-worktree P3-WP14 local release manifest."""

from __future__ import annotations

import hashlib
import json
import os
import re
import stat
import subprocess
from dataclasses import dataclass
from pathlib import Path

from p3_wp14_secure_io import atomic_write_bytes, exclusive_lock, read_bytes_once, sha256_file_once


ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "docs/phase3/P3_WP14_RELEASE_MANIFEST.json"
TREE_ROOTS = ("app", "deployment/app", "migrations", "deployment/migrations")
FILE_PATHS = (
    "docs/architecture/OPENAPI.yaml",
    "deploy/docker/docker-compose.yml",
    "scripts/generate_openapi.py",
    "docs/acceptance/P3_WP14_UAT_CASES.json",
    "docs/acceptance/P3_WP14_UAT_PLAN_AND_REPORT.md",
    "docs/acceptance/P3_WP14_DEFECT_AND_GATE_LEDGER.md",
    "docs/acceptance/P3_WP14_FIVE_PARTY_SIGNOFF.md",
    "docs/runbooks/P3_WP14_ENTERPRISE_PILOT_HANDOVER.md",
    "docs/validation/P3_WP14_ENTERPRISE_PILOT_EVIDENCE.md",
    "docs/phase3/P3_WP14_264_MODULE_CLOSURE_AUDIT.json",
    "docs/phase3/P3_WP14_ENTERPRISE_PILOT_ACCEPTANCE_IMPLEMENTATION_PLAN.md",
    "docs/validation/P3_WP14_LOCAL_CONFIG_A.json",
    "docs/validation/P3_WP14_LOCAL_CONFIG_B.json",
    "docs/validation/P3_WP14_LOCAL_CONFIG_COMPARISON.json",
    "tests/test_phase3_wp14_acceptance_handover_contract.py",
    "scripts/validate_p3_wp14_local_configs.py",
    "scripts/generate_p3_wp14_release_manifest.py",
    "scripts/generate_p3_wp14_verification_ledger.py",
    "scripts/verify_p3_wp14_fresh_local_engineering.py",
    "scripts/p3_wp14_secure_io.py",
)
COMPOSE = "deploy/docker/docker-compose.yml"
COMMAND_TIMEOUT_SECONDS = 30
COMMAND_HOME = ROOT / "work/p3_wp14_manifest_command_home"
COMPOSE_EXECUTABLE = "/usr/local/bin/docker-compose"
MANIFEST_LOCK = ROOT / "work/p3_wp14_manifest_lock_scope/lock"
MAX_DOCKERIGNORE_BYTES = 64 * 1024
MAX_TREE_FILES = 20_000
MAX_TREE_TOTAL_BYTES = 512 * 1024 * 1024
MAX_TREE_DEPTH = 32
MAX_TREE_PATH_BYTES = 1024
COMPOSE_INPUTS = (
    ("build_context", "."),
    ("dockerfile", "deploy/docker/backend.Dockerfile"),
    ("dockerfile", "deploy/docker/mcp.Dockerfile"),
    ("dockerfile", "deploy/docker/runner.Dockerfile"),
    ("dockerfile", "deploy/docker/frontend.Dockerfile"),
    ("dockerfile_copy_input", "deploy/docker/nginx.conf"),
    ("requirements", "requirements.txt"),
    ("requirements", "deployment/requirements.txt"),
    ("mount_input", "deploy/docker/postgres/init"),
    ("mount_input", "database"),
    ("mount_input", "deploy/docker/datasources/postgres/init.sql"),
    ("mount_input", "deploy/docker/datasources/mysql/init.sql"),
    ("mount_input", "deploy/docker/kestra/application.yml"),
    ("mount_input", "deploy/docker/edge-mtls-nginx.conf"),
)
DOCKERFILES = tuple(path for kind, path in COMPOSE_INPUTS if kind == "dockerfile")


def _minimal_command_env() -> dict[str, str]:
    """Do not inherit GIT_DIR, GIT_WORK_TREE, compose, or credential overrides."""
    return {
        "PATH": os.environ.get("PATH", ""),
        "HOME": str(COMMAND_HOME),
        "DOCKER_CONFIG": str(COMMAND_HOME / "docker"),
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_CONFIG_GLOBAL": "/dev/null",
    }


def _run_command(command: list[str], command_env: dict[str, str], *, text: bool) -> subprocess.CompletedProcess:
    """Run only from the repository root with an intentionally minimal env."""
    return subprocess.run(
        command,
        cwd=ROOT,
        env=command_env,
        text=text,
        capture_output=True,
        check=True,
        timeout=COMMAND_TIMEOUT_SECONDS,
    )


def _git_top_level(command_env: dict[str, str]) -> str:
    top_level = _run_command(["git", "rev-parse", "--show-toplevel"], command_env, text=True).stdout.strip()
    if Path(top_level) != ROOT:
        raise RuntimeError("unexpected git top-level")
    return top_level


def _collect_git_status(command_env: dict[str, str], top_level: str) -> dict[str, object]:
    """Return independently recomputable porcelain counts and its exact digest."""
    porcelain = _run_command(["git", "status", "--porcelain=v1", "-z"], command_env, text=False).stdout
    if not isinstance(porcelain, bytes):
        raise RuntimeError("git porcelain was not bytes")
    records = porcelain.split(b"\0")
    entries: list[bytes] = []
    index = 0
    while index < len(records):
        record = records[index]
        if not record:
            index += 1
            continue
        if len(record) < 4 or record[2:3] != b" ":
            raise RuntimeError("invalid git porcelain v1 -z record")
        xy = record[:2]
        entries.append(xy)
        index += 1
        # With -z, rename/copy records carry a second NUL-delimited old path.
        if b"R" in xy or b"C" in xy:
            if index >= len(records) or not records[index]:
                raise RuntimeError("truncated git porcelain rename/copy record")
            index += 1
    return {
        "command": ["git", "status", "--porcelain=v1", "-z"],
        "top_level": top_level,
        "entry_count": len(entries),
        "tracked": sum(xy != b"??" for xy in entries),
        "untracked": sum(xy == b"??" for xy in entries),
        "staged": sum(xy[:1] not in {b" ", b"?"} for xy in entries),
        "sha256": hashlib.sha256(porcelain).hexdigest(),
    }


def _compose_services(command_env: dict[str, str]) -> list[str]:
    rendered = _run_command(
        [COMPOSE_EXECUTABLE, "--env-file", "/dev/null", "-f", COMPOSE, "config", "--format", "json"], command_env, text=True
    )
    return sorted(json.loads(rendered.stdout)["services"])


def _image_record(source: str, reference: str) -> dict[str, str]:
    immutable = "@sha256:" in reference
    return {
        "source": source,
        "reference": reference,
        "status": "IMMUTABLE_DIGEST" if immutable else "UNRESOLVED_MUTABLE_TAG",
        "gate": "PASS_LOCAL" if immutable else "BLOCKED_EXTERNAL",
    }


def _image_reference_closure(
    compose_content: str | None = None, dockerfiles: dict[str, str] | None = None,
) -> list[dict[str, str]]:
    """Audit every Compose image and Dockerfile FROM without contacting a registry."""
    compose = compose_content if compose_content is not None else read_bytes_once(ROOT / COMPOSE).decode("utf-8")
    records = [
        _image_record(f"compose:{number}", match.group(1).strip())
        for number, line in enumerate(compose.splitlines(), 1)
        if (match := re.match(r"\s*image:\s*([^#\s]+)", line))
    ]
    source_files = dockerfiles if dockerfiles is not None else {
        path: read_bytes_once(ROOT / path).decode("utf-8") for path in DOCKERFILES
    }
    for path, content in source_files.items():
        stages: set[str] = set()
        for number, line in enumerate(content.splitlines(), 1):
            match = re.match(r"\s*FROM\s+(?:--platform=\S+\s+)?(\S+)(?:\s+AS\s+(\S+))?", line, re.IGNORECASE)
            if not match:
                continue
            reference, alias = match.groups()
            if reference in stages:
                records.append({"source": f"dockerfile:{path}:{number}", "reference": reference, "status": "INTERNAL_BUILD_STAGE", "gate": "PASS_LOCAL"})
            else:
                records.append(_image_record(f"dockerfile:{path}:{number}", reference))
            if alias:
                stages.add(alias)
    return records


def _sha256(path: Path) -> str:
    return sha256_file_once(path)


def _glob_regex(pattern: str) -> re.Pattern[str]:
    """Compile Docker's slash-aware wildcard subset used by this repository."""
    pieces: list[str] = []
    index = 0
    while index < len(pattern):
        if pattern.startswith("**/", index):
            pieces.append("(?:.*/)?")
            index += 3
        elif pattern.startswith("**", index):
            pieces.append(".*")
            index += 2
        elif pattern[index] == "*":
            pieces.append("[^/]*")
            index += 1
        elif pattern[index] == "?":
            pieces.append("[^/]")
            index += 1
        else:
            pieces.append(re.escape(pattern[index]))
            index += 1
    return re.compile("^" + "".join(pieces) + "(?:/.*)?$")


@dataclass(frozen=True)
class DockerIgnoreRule:
    pattern: str
    include: bool
    regex: re.Pattern[str]


class DockerIgnoreMatcher:
    """Ordered `.dockerignore` matcher with audited rule source and negation."""

    def __init__(self, rules: tuple[DockerIgnoreRule, ...], source_sha256: str):
        self.rules = rules
        self.source_sha256 = source_sha256

    @classmethod
    def from_bytes(cls, content: bytes) -> "DockerIgnoreMatcher":
        rules: list[DockerIgnoreRule] = []
        for raw in content.decode("utf-8", errors="strict").splitlines():
            line = raw.strip()
            if not line or line.startswith("#"):
                continue
            include = line.startswith("!")
            pattern = line[1:] if include else line
            pattern = pattern.removeprefix("/").rstrip("/")
            if not pattern or pattern == ".":
                continue
            if any(character in pattern for character in "[]\\"):
                raise ValueError("unsupported dockerignore pattern; refusing approximation")
            if "/" not in pattern:
                pattern = "**/" + pattern
            rules.append(DockerIgnoreRule(pattern=pattern, include=include, regex=_glob_regex(pattern)))
        return cls(tuple(rules), hashlib.sha256(content).hexdigest())

    @classmethod
    def from_root(cls, root: Path) -> "DockerIgnoreMatcher":
        return cls.from_bytes(read_bytes_once(root / ".dockerignore", limit=MAX_DOCKERIGNORE_BYTES))

    def ignores(self, relative: str) -> bool:
        ignored = False
        for rule in self.rules:
            if rule.regex.fullmatch(relative):
                ignored = not rule.include
        return ignored

    def may_reinclude_descendant(self, relative: str) -> bool:
        """Conservatively retain an ignored directory when a negation exists."""
        # A later Dockerignore negation can re-include a child.  We deliberately
        # over-approximate rather than incorrectly pruning an includable path.
        return any(rule.include for rule in self.rules)


def _tree_digest(relative_root: str, *, root: Path | None = None, matcher: DockerIgnoreMatcher | None = None) -> str:
    root = ROOT if root is None else root
    context_matcher = matcher or (DockerIgnoreMatcher.from_root(root) if relative_root == "." else None)
    digest = hashlib.sha256(); digest.update(b"["); first_record = True
    entry_count = 0; total_bytes = 0
    def visit(directory: Path) -> None:
        nonlocal entry_count, total_bytes, first_record
        try:
            entries = sorted(os.scandir(directory), key=lambda entry: entry.name)
        except OSError as error:
            raise ValueError("tree traversal failed") from error
        for entry in entries:
            path = Path(entry.path)
            relative = path.relative_to(root).as_posix()
            if len(relative.encode("utf-8")) > MAX_TREE_PATH_BYTES or len(Path(relative).parts) > MAX_TREE_DEPTH:
                raise ValueError("tree path boundary exceeded")
            entry_count += 1
            if entry_count > MAX_TREE_FILES:
                raise ValueError("tree resource boundary exceeded")
            try:
                metadata = entry.stat(follow_symlinks=False)
            except OSError as error:
                raise ValueError("tree entry changed during traversal") from error
            ignored = bool(context_matcher and context_matcher.ignores(relative))
            if stat.S_ISDIR(metadata.st_mode):
                if not (ignored and context_matcher and not context_matcher.may_reinclude_descendant(relative)):
                    visit(path)
                continue
            derived = any(part == "__pycache__" for part in path.parts) or path.suffix == ".pyc"
            if not stat.S_ISREG(metadata.st_mode) or derived or ignored:
                continue
            total_bytes += metadata.st_size
            if total_bytes > MAX_TREE_TOTAL_BYTES:
                raise ValueError("tree resource boundary exceeded")
            record = json.dumps([relative, _sha256(path)], ensure_ascii=False, separators=(",", ":")).encode("utf-8")
            if not first_record: digest.update(b",")
            digest.update(record); first_record = False
    visit(root / relative_root)
    digest.update(b"]")
    return digest.hexdigest()


def _input_census(command_env: dict[str, str]) -> str:
    matcher = DockerIgnoreMatcher.from_root(ROOT)
    values = {
        "trees": {root: _tree_digest(root, matcher=matcher if root == "." else None) for root in (*TREE_ROOTS, ".")},
        "files": {path: _sha256(ROOT / path) for path in FILE_PATHS},
        "compose": _sha256(ROOT / COMPOSE),
        "dockerignore": matcher.source_sha256,
        "closure": {path: (_tree_digest(path, matcher=matcher if path == "." else None) if (ROOT / path).is_dir() else _sha256(ROOT / path)) for _kind, path in COMPOSE_INPUTS},
        "git_status": _collect_git_status(command_env, _git_top_level(command_env))["sha256"],
    }
    return hashlib.sha256(json.dumps(values, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()


def _main_locked() -> None:
    command_env = _minimal_command_env()
    before_census = _input_census(command_env)
    git_top = _git_top_level(command_env)
    root_context_matcher = DockerIgnoreMatcher.from_root(ROOT)
    artifact_sets = [
        {"kind": "tree", "root": root, "sha256": _tree_digest(root), "excluded_derived_paths": ["__pycache__", "*.pyc"]}
        for root in TREE_ROOTS
    ]
    artifact_sets.extend(
        {"kind": "file", "path": path, "sha256": _sha256(ROOT / path)}
        for path in FILE_PATHS
    )
    closure_inputs = []
    for kind, path in COMPOSE_INPUTS:
        absolute = ROOT / path
        if absolute.is_dir():
            digest = _tree_digest(path, matcher=root_context_matcher if path == "." else None)
        else:
            digest = _sha256(absolute)
        closure_inputs.append({"kind": kind, "path": path, "sha256": digest})
    compose_services = _compose_services(command_env)
    image_references = _image_reference_closure()
    git_status = _collect_git_status(command_env, git_top)
    payload = {
        "schema_version": "1.0",
        "work_package": "P3-WP14",
        "release_identity": {
            "state": "UNCOMMITTED_WORKTREE_EVIDENCE",
            "signed": False,
            "signature_status": "TBD_EXTERNAL",
            "git_base_commit": _run_command(["git", "rev-parse", "HEAD"], command_env, text=True).stdout.strip(),
            "git_base_tree": _run_command(["git", "rev-parse", "HEAD^{tree}"], command_env, text=True).stdout.strip(),
            "worktree_clean": git_status["entry_count"] == 0,
            "git_status": git_status,
            "declaration": "This is a local, unsigned snapshot from a known dirty shared worktree. It is not a clean, signed, pushed, deployed, or enterprise-approved release.",
        },
        "controlled_artifact_sets": artifact_sets,
        "compose_input_closure": {
            "compose_file": COMPOSE,
            "compose_sha256": _sha256(ROOT / COMPOSE),
            "dockerignore": {"path": ".dockerignore", "sha256": root_context_matcher.source_sha256},
            "render_command": [COMPOSE_EXECUTABLE, "--env-file", "/dev/null", "-f", COMPOSE, "config", "--format", "json"],
            "parsed_services": compose_services,
            "inputs": closure_inputs,
            "unresolved_variable_mounts": [
                "${DATAOPS_EDGE_MTLS_CERT_DIR:-./edge-mtls-certs}",
                "${EDGE_GATEWAY_SIGNING_PRIVATE_KEY_FILE:-./edge-secrets/edge-gateway-signing.key}",
            ],
        },
        "image_reference_closure": {
            "items": image_references,
            "immutable_signed_deployability_status": "BLOCKED_EXTERNAL" if any(item["status"] == "UNRESOLVED_MUTABLE_TAG" for item in image_references) else "TBD_EXTERNAL",
            "boundary": "No registry pull was performed. Any tag or variable reference without @sha256 is mutable and blocks immutable/signed deployability acceptance.",
        },
        "excluded_dependent_evidence": [
            "docs/phase3/P3_WP14_RELEASE_MANIFEST.json",
            "docs/phase3/P3_WP14_RELEASE_TRACEABILITY.json",
            "docs/validation/P3_WP14_LOCAL_CONFIG_RENDER.json",
            "docs/validation/P3_WP14_LOCAL_VERIFICATION_LEDGER.json",
            "docs/validation/P3_WP14_LOCAL_VERIFICATION.log",
        ],
        "boundary": "Dependent evidence is digest-checked by the traceability contract but excluded here to avoid self-referential hashes. No secret, enterprise endpoint, or external signature is recorded.",
    }
    if before_census != _input_census(command_env):
        raise RuntimeError("manifest input closure changed during generation")
    atomic_write_bytes(OUTPUT, json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") + b"\n")


def main() -> None:
    with exclusive_lock(MANIFEST_LOCK):
        _main_locked()


if __name__ == "__main__":
    main()