#!/usr/bin/env python3 """Collect release/restart provenance and enforce audit-checkout gates.""" from __future__ import annotations import argparse import hashlib 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, load_observed_evidence, ) _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("--observed-evidence-file", 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: observed_evidence = load_observed_evidence(args.observed_evidence_file) 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, observed_evidence=observed_evidence, runtime_files=args.runtime_file, capture_commands=capture_commands, broker_log_path=args.broker_log_path, ) except (RuntimeError, TypeError, 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. record_bytes = (json.dumps(report, ensure_ascii=False, indent=2) + "\n").encode("utf-8") with output_path.open("xb") as fp: fp.write(record_bytes) output_path.chmod(0o444) print( json.dumps( { "schema_version": 1, "record_path": str(output_path.resolve()), "record_sha256": hashlib.sha256(record_bytes).hexdigest(), "gate_passed": bool(report["gate"]["passed"]), }, separators=(",", ":"), ) ) if dirty_checkout: return 2 return 0 if report["gate"]["passed"] else 1 if __name__ == "__main__": raise SystemExit(main())