fix(ops): make release provenance gate fail closed

tests/scripts/test_release_provenance_gate.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from scripts import release_provenance_gate
+
+
+def test_documented_python_invocation_loads_project_without_pythonpath():
+    script = Path(release_provenance_gate.__file__).resolve()
+    env = os.environ.copy()
+    env.pop("PYTHONPATH", None)
+
+    completed = subprocess.run(
+        [sys.executable, str(script), "--help"],
+        cwd=script.parent,
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert completed.returncode == 0, completed.stderr
+    assert "--expected-oci-revision" in completed.stdout
+    assert script.stat().st_mode & 0o111 != 0
+
+
+def _init_repo(tmp_path: Path) -> tuple[Path, str]:
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
+    subprocess.run(["git", "config", "user.email", "test@example.com"], cwd=repo, check=True)
+    subprocess.run(["git", "config", "user.name", "Test"], cwd=repo, check=True)
+    (repo / "runtime.py").write_text("print('ok')\n", encoding="utf-8")
+    subprocess.run(["git", "add", "runtime.py"], cwd=repo, check=True)
+    subprocess.run(["git", "commit", "-m", "init"], cwd=repo, check=True, capture_output=True)
+    head = subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True
+    ).stdout.strip()
+    subprocess.run(["git", "checkout", "--detach", head], cwd=repo, check=True, capture_output=True)
+    return repo, head
+
+
+def _fake_read_only_commands(tmp_path: Path) -> tuple[Path, Path]:
+    bindir = tmp_path / "bin"
+    bindir.mkdir(parents=True)
+    trex = bindir / "trex-core"
+    rabbit = bindir / "rabbitmqctl"
+    trex.write_text("#!/bin/sh\nprintf '{}\\n'\n", encoding="utf-8")
+    rabbit.write_text("#!/bin/sh\nprintf 'consumer\\n'\n", encoding="utf-8")
+    trex.chmod(0o755)
+    rabbit.chmod(0o755)
+    return trex, rabbit
+
+
+def _argv(tmp_path: Path, repo: Path, head: str) -> list[str]:
+    trex, rabbit = _fake_read_only_commands(tmp_path)
+    broker_log = tmp_path / "broker.log"
+    broker_log.write_text("event=order_submit account=must-not-leak\n", encoding="utf-8")
+    return [
+        "--actor", "operator-a",
+        "--reason", "deploy release",
+        "--instance", "inst-a",
+        "--operation-id", "op-123",
+        "--phase", "pre",
+        "--command-class", "deploy",
+        "--checkout-dir", str(repo),
+        "--expected-oci-revision", head,
+        "--image-digest", "sha256:" + ("a" * 64),
+        "--runtime-file", "runtime.py",
+        "--broker-log-path", str(broker_log),
+        "--capture-command", f"pending_slots={trex} slots --status pending --instance inst-a --json",
+        "--capture-command", f"reconciliation_slots={trex} slots --status all --instance inst-a --json",
+        "--capture-command", f"queue_consumers={rabbit} list_consumers",
+    ]
+
+
+def test_cli_writes_unique_record_outside_checkout_and_keeps_checkout_clean(tmp_path, capsys):
+    repo, head = _init_repo(tmp_path)
+    argv = _argv(tmp_path, repo, head)
+
+    assert release_provenance_gate.main(argv) == 0
+    first = Path(capsys.readouterr().out.strip())
+    assert release_provenance_gate.main(argv) == 0
+    second = Path(capsys.readouterr().out.strip())
+
+    assert first != second
+    assert first.parent == tmp_path / "repo-release-provenance"
+    assert first.stat().st_mode & 0o222 == 0
+    assert second.stat().st_mode & 0o222 == 0
+    payload = json.loads(first.read_text(encoding="utf-8"))
+    assert payload["gate"]["passed"] is True
+    assert "must-not-leak" not in first.read_text(encoding="utf-8")
+    status = subprocess.run(
+        ["git", "status", "--porcelain"], cwd=repo, check=True, capture_output=True, text=True
+    ).stdout
+    assert status == ""
+
+
+def test_cli_dirty_checkout_is_fail_closed_and_isolated(tmp_path, capsys):
+    repo, head = _init_repo(tmp_path)
+    argv = _argv(tmp_path, repo, head)
+    (repo / "runtime.py").write_text("dirty\n", encoding="utf-8")
+
+    assert release_provenance_gate.main(argv) == 2
+    output = Path(capsys.readouterr().out.strip())
+    payload = json.loads(output.read_text(encoding="utf-8"))
+    assert output.stat().st_mode & 0o222 == 0
+    assert payload["gate"]["passed"] is False
+    assert payload["gate"]["deployment_blocked"] is True
+    assert payload["state_capture"]["command_outputs"] == {}
+    snapshot = Path(payload["dirty_checkout_isolation"]["isolate_dir"])
+    assert snapshot.parent == tmp_path / "repo-release-provenance-isolation"
+    assert (snapshot / "runtime.py").read_text() == "dirty\n"
+
+
+def test_cli_rejects_unsafe_command_and_checkout_local_output(tmp_path):
+    repo, head = _init_repo(tmp_path)
+    argv = _argv(tmp_path, repo, head)
+    unsafe = argv.index(next(x for x in argv if x.startswith("pending_slots=")))
+    argv[unsafe] = "pending_slots=trex-core cancel slot-1 --yes"
+    with pytest.raises(SystemExit):
+        release_provenance_gate.main(argv)
+
+    argv = _argv(tmp_path / "other", repo, head)
+    argv.extend(["--output-dir", str(repo / "artifacts")])
+    with pytest.raises(SystemExit):
+        release_provenance_gate.main(argv)