From 8d9d627b1c930909dc1399ae784cf608ae63ebeb Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Sat, 12 Sep 2026 13:37:19 +0000 Subject: [PATCH] fix(ops): harden provenance gate path and command parsing Co-authored-by: genichin <4282227+genichin@users.noreply.github.com> --- scripts/release_provenance_gate.py | 3 +- tests/adapters/persistence/test_ops_checks.py | 19 ++++++++++++ trex_core/adapters/persistence/ops_checks.py | 29 +++++++++++++++---- 3 files changed, 44 insertions(+), 7 deletions(-) diff --git a/scripts/release_provenance_gate.py b/scripts/release_provenance_gate.py index 39a86eb..e11a68b 100644 --- a/scripts/release_provenance_gate.py +++ b/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 diff --git a/tests/adapters/persistence/test_ops_checks.py b/tests/adapters/persistence/test_ops_checks.py index 76eafb7..aea382b 100644 --- a/tests/adapters/persistence/test_ops_checks.py +++ b/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 diff --git a/trex_core/adapters/persistence/ops_checks.py b/trex_core/adapters/persistence/ops_checks.py index 484facb..156486d 100644 --- a/trex_core/adapters/persistence/ops_checks.py +++ b/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)