trex-core/scripts/release_provenance_gate.py

#!/usr/bin/env python3
"""Collect release/restart provenance and enforce audit-checkout gates."""

from __future__ import annotations

import argparse
import json
import re
import shlex
import sys
import uuid
from datetime import UTC, datetime
from pathlib import Path

if __package__ in {None, ""}:
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from trex_core.adapters.persistence.ops_checks import (
    collect_release_provenance,
    isolate_dirty_checkout_files,
)

_COMMAND_CLASSES = ("deploy", "recovery", "manual-stop", "manual-start", "recreate")
_REQUIRED_CAPTURE_KEYS = frozenset(
    {"pending_slots", "reconciliation_slots", "queue_consumers"}
)
_SAFE_ID = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,127}")


def _record_path(
    output_dir: Path,
    *,
    operation_id: str,
    phase: str,
    instance: str,
    command_class: str,
) -> Path:
    for value in (operation_id, phase, instance, command_class):
        if not _SAFE_ID.fullmatch(value):
            raise ValueError("record filename components must be safe identifiers")
    stamp = datetime.now(tz=UTC).strftime("%Y%m%dT%H%M%S.%fZ")
    nonce = uuid.uuid4().hex[:12]
    return output_dir / (
        f"{stamp}--{operation_id}--{phase}--{instance}--{command_class}--{nonce}.json"
    )


def _parse_capture_command(raw: str) -> tuple[str, list[str]]:
    if "=" not in raw:
        raise ValueError("capture command must be KEY=command")
    key, command = raw.split("=", 1)
    key = key.strip()
    parts = shlex.split(command.strip())
    if not key or not parts:
        raise ValueError("capture command must be KEY=command")
    return key, parts


def _validate_capture_command(key: str, command: list[str], *, instance: str) -> None:
    executable = Path(command[0]).name
    args = command[1:]
    expected = {
        "pending_slots": ["slots", "--status", "pending", "--instance", instance, "--json"],
        "reconciliation_slots": ["slots", "--status", "all", "--instance", instance, "--json"],
        "queue_consumers": ["list_consumers"],
    }
    if key not in expected:
        raise ValueError(f"unsupported capture key: {key}")
    if key == "queue_consumers":
        valid = executable == "rabbitmqctl" and args == expected[key]
    else:
        valid = executable == "trex-core" and args == expected[key]
    if not valid:
        raise ValueError(f"capture command is not the approved read-only form for {key}")


def _outside_checkout(path: Path, checkout: Path, *, label: str) -> Path:
    resolved = path.resolve()
    try:
        resolved.relative_to(checkout.resolve())
    except ValueError:
        return resolved
    raise ValueError(f"{label} must be outside the audit checkout")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--actor", required=True)
    parser.add_argument("--reason", required=True)
    parser.add_argument("--instance", required=True)
    parser.add_argument("--operation-id", required=True)
    parser.add_argument("--phase", required=True, choices=("pre", "post"))
    parser.add_argument("--command-class", required=True, choices=_COMMAND_CLASSES)
    parser.add_argument("--checkout-dir", default=".")
    parser.add_argument("--expected-oci-revision", required=True)
    parser.add_argument("--image-digest", required=True)
    parser.add_argument("--runtime-file", action="append", required=True)
    parser.add_argument("--broker-log-path", default="/tmp/trex-core.log")
    parser.add_argument("--capture-command", action="append", required=True)
    parser.add_argument("--output-dir")
    parser.add_argument("--isolate-dir")
    args = parser.parse_args(argv)

    try:
        capture_commands: dict[str, list[str]] = {}
        for raw in args.capture_command:
            key, command = _parse_capture_command(raw)
            if key in capture_commands:
                raise ValueError(f"duplicate capture key: {key}")
            _validate_capture_command(key, command, instance=args.instance)
            capture_commands[key] = command
        if set(capture_commands) != _REQUIRED_CAPTURE_KEYS:
            missing = sorted(_REQUIRED_CAPTURE_KEYS - set(capture_commands))
            extra = sorted(set(capture_commands) - _REQUIRED_CAPTURE_KEYS)
            raise ValueError(f"capture command keys mismatch: missing={missing}, extra={extra}")

        checkout = Path(args.checkout_dir).resolve()
        output_default = checkout.parent / f"{checkout.name}-release-provenance"
        isolate_default = checkout.parent / f"{checkout.name}-release-provenance-isolation"
        output_dir = _outside_checkout(
            Path(args.output_dir) if args.output_dir else output_default,
            checkout,
            label="output directory",
        )
        isolate_dir = _outside_checkout(
            Path(args.isolate_dir) if args.isolate_dir else isolate_default,
            checkout,
            label="isolation directory",
        )

        report = collect_release_provenance(
            actor=args.actor,
            reason=args.reason,
            instance=args.instance,
            command_class=args.command_class,
            operation_id=args.operation_id,
            phase=args.phase,
            checkout_dir=checkout,
            expected_oci_revision=args.expected_oci_revision,
            image_digest=args.image_digest,
            runtime_files=args.runtime_file,
            capture_commands=capture_commands,
            broker_log_path=args.broker_log_path,
        )
    except (RuntimeError, ValueError) as exc:
        parser.error(str(exc))

    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = _record_path(
        output_dir,
        operation_id=args.operation_id,
        phase=args.phase,
        instance=args.instance,
        command_class=args.command_class,
    )

    dirty_checkout = bool(report["checkout"]["dirty_paths"])
    if dirty_checkout:
        report["dirty_checkout_isolation"] = isolate_dirty_checkout_files(
            checkout_dir=checkout,
            isolate_dir=isolate_dir,
        )

    # Exclusive create prevents a provenance record from being overwritten.
    with output_path.open("x", encoding="utf-8") as fp:
        fp.write(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
    output_path.chmod(0o444)
    print(str(output_path))
    if dirty_checkout:
        return 2
    return 0 if report["gate"]["passed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())