fix(ops): harden provenance gate path and command parsing

Co-authored-by: genichin <4282227+genichin@users.noreply.github.com>

scripts/release_provenance_gate.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import shlex
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -26,7 +27,7 @@ def _parse_capture_command(raw: str) -> tuple[str, list[str]]:
         raise ValueError("capture command must be KEY=command")
     key, command = raw.split("=", 1)
     key = key.strip()
-    parts = [x for x in command.strip().split(" ") if x]
+    parts = shlex.split(command.strip())
     if not key or not parts:
         raise ValueError("capture command must be KEY=command")
     return key, parts

tests/adapters/persistence/test_ops_checks.py

@@ -160,3 +160,22 @@ def test_isolate_dirty_checkout_files_copies_dirty_and_untracked(tmp_path):
     assert (isolated / "tracked.txt").read_text(encoding="utf-8") == "dirty\n"
     assert (isolated / "new.txt").read_text(encoding="utf-8") == "new\n"
     assert (isolated / "manifest.json").is_file()
+
+
+def test_collect_release_provenance_rejects_runtime_path_escape(tmp_path):
+    repo, head = _init_git_repo(tmp_path)
+    report = collect_release_provenance(
+        actor="operator-a",
+        reason="deploy",
+        instance="inst-a",
+        command_class="manual-start",
+        checkout_dir=repo,
+        expected_oci_revision=head,
+        runtime_files=["../outside.txt"],
+        capture_commands={},
+        broker_log_path=None,
+    )
+    runtime = report["runtime_file_verification"][0]
+    assert runtime["path_within_checkout"] is False
+    assert runtime["matches_expected_revision"] is False
+    assert report["gate"]["passed"] is False

trex_core/adapters/persistence/ops_checks.py

@@ -97,6 +97,15 @@ def _sha256_bytes(payload: bytes) -> str:
     return hashlib.sha256(payload).hexdigest()
 
 
+def _resolve_under(root: Path, relpath: str) -> Path | None:
+    candidate = (root / relpath).resolve()
+    try:
+        candidate.relative_to(root)
+        return candidate
+    except ValueError:
+        return None
+
+
 def _capture_command(command: list[str], checkout_dir: Path) -> dict:
     try:
         done = subprocess.run(
@@ -150,8 +159,9 @@ def collect_release_provenance(
     ref_revision = expected_oci_revision or head
     runtime_reports = []
     for relpath in runtime_files or []:
-        abs_path = (checkout / relpath).resolve()
-        exists = abs_path.is_file()
+        abs_path = _resolve_under(checkout, relpath)
+        path_valid = abs_path is not None
+        exists = bool(path_valid and abs_path.is_file())
         current_bytes = abs_path.read_bytes() if exists else b""
         current_sha = _sha256_bytes(current_bytes) if exists else None
         expected_bytes = _git_blob_bytes(checkout, ref_revision, relpath)
@@ -159,11 +169,14 @@ def collect_release_provenance(
         runtime_reports.append(
             {
                 "path": relpath,
+                "path_within_checkout": path_valid,
                 "exists": exists,
                 "current_sha256": current_sha,
                 "expected_revision": ref_revision,
                 "expected_sha256": expected_sha,
-                "matches_expected_revision": bool(exists and expected_bytes is not None and current_bytes == expected_bytes),
+                "matches_expected_revision": bool(
+                    path_valid and exists and expected_bytes is not None and current_bytes == expected_bytes
+                ),
             }
         )
     command_outputs = {}
@@ -219,10 +232,14 @@ def isolate_dirty_checkout_files(
     dirty_paths = _porcelain_paths(status)
     copied: list[str] = []
     for rel in dirty_paths:
-        src = (checkout / rel).resolve()
-        if not src.exists() or src.is_dir():
+        src = _resolve_under(checkout, rel)
+        if src is None or not src.exists() or src.is_dir():
+            continue
+        target = (dst / rel).resolve()
+        try:
+            target.relative_to(dst)
+        except ValueError:
             continue
-        target = dst / rel
         target.parent.mkdir(parents=True, exist_ok=True)
         shutil.copy2(src, target)
         copied.append(rel)