trex/trex-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ops): verify release provenance evidence pairs

Browse Source
This commit is contained in:
Hermes Agent
2026-09-20 11:47:05 +09:00
parent cd2342f2c6
commit db2bdd7d4a
15 changed files with 1054 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/adapters/persistence/test_ops_checks.py
+69 -11
View File
@@ -130,8 +130,15 @@ def _provenance_kwargs(repo: Path, head: str) -> dict:
"operation_id": "op-1",
"phase": "pre",
"checkout_dir": repo,
"expected_oci_revision": head,
"image_digest": "sha256:" + ("a" * 64),
"observed_evidence": {
"schema_version": 1,
"target_id": "service-a",
"target_kind": "compose-service",
"image_id": "sha256:" + ("b" * 64),
"image_digest": "sha256:" + ("a" * 64),
"oci_revision": head,
"observed_at": "2026-09-20T02:00:00Z",
},
"runtime_files": ["tracked.txt"],
"capture_commands": {
"pending_slots": [
@@ -151,7 +158,7 @@ def test_collect_release_provenance_passes_on_clean_checkout(tmp_path):
report = collect_release_provenance(**_provenance_kwargs(repo, head))
assert report["checkout"]["git_clean"] is True
assert report["checkout"]["detached_head"] is True
assert report["artifact_identity"]["oci_revision_matches_head"] is True
assert report["observed_evidence"]["source"] == "operator_observed"
assert report["runtime_file_verification"][0]["matches_expected_revision"] is True
assert report["gate"]["passed"] is True
assert report["operation_id"] == "op-1"
@@ -181,7 +188,7 @@ def test_collect_release_provenance_blocks_on_revision_or_runtime_mismatch(tmp_p
kwargs = _provenance_kwargs(repo, old_head)
report = collect_release_provenance(**kwargs)
assert report["checkout"]["git_clean"] is True
assert report["artifact_identity"]["oci_revision_matches_head"] is False
assert report["checkout"]["head_revision"] != report["observed_evidence"]["oci_revision"]
assert report["runtime_file_verification"][0]["matches_expected_revision"] is False
assert report["gate"]["passed"] is False
@@ -192,13 +199,58 @@ def test_isolate_dirty_checkout_files_copies_dirty_and_untracked(tmp_path):
(repo / "new.txt").write_text("new\n", encoding="utf-8")
isolated = tmp_path / "isolated"
manifest = isolate_dirty_checkout_files(checkout_dir=repo, isolate_dir=isolated)
assert set(manifest["dirty_paths"]) == {"tracked.txt", "new.txt"}
assert set(manifest["copied_paths"]) == {"tracked.txt", "new.txt"}
entries = {entry["path"]: entry for entry in manifest["entries"]}
assert set(entries) == {"tracked.txt", "new.txt"}
assert all(entry["source_type"] == "regular" for entry in entries.values())
assert all(entry["matches"] is True for entry in entries.values())
snapshot = Path(manifest["isolate_dir"])
assert snapshot.parent == isolated
assert (snapshot / "tracked.txt").read_text(encoding="utf-8") == "dirty\n"
assert (snapshot / "new.txt").read_text(encoding="utf-8") == "new\n"
assert (snapshot / "manifest.json").is_file()
assert (snapshot / "manifest.json").stat().st_mode & 0o222 == 0
@pytest.mark.parametrize("target_kind", ["absolute", "relative_escape"])
def test_isolate_dirty_checkout_files_rejects_symlink_escape_without_mutating_source(
tmp_path, target_kind
):
repo, _ = _init_git_repo(tmp_path)
outside = tmp_path / "outside.txt"
outside.write_text("outside\n", encoding="utf-8")
link = repo / "escape-link"
target = str(outside) if target_kind == "absolute" else "../outside.txt"
link.symlink_to(target)
isolated = tmp_path / "isolated"
with pytest.raises(ValueError, match="symlink target escapes checkout"):
isolate_dirty_checkout_files(checkout_dir=repo, isolate_dir=isolated)
assert link.is_symlink()
assert link.readlink() == Path(target)
assert outside.read_text(encoding="utf-8") == "outside\n"
if isolated.exists():
assert not list(isolated.glob("*/manifest.json"))
def test_isolate_dirty_checkout_files_preserves_internal_relative_symlink(tmp_path):
repo, _ = _init_git_repo(tmp_path)
nested = repo / "nested"
nested.mkdir()
(nested / "target.txt").write_text("target\n", encoding="utf-8")
link = repo / "internal-link"
link.symlink_to("nested/target.txt")
manifest = isolate_dirty_checkout_files(
checkout_dir=repo, isolate_dir=tmp_path / "isolated"
)
entry = next(item for item in manifest["entries"] if item["path"] == "internal-link")
copied = Path(manifest["isolate_dir"]) / "internal-link"
assert entry["source_type"] == "symlink"
assert entry["copied_type"] == "symlink"
assert entry["matches"] is True
assert copied.readlink() == Path("nested/target.txt")
assert link.readlink() == Path("nested/target.txt")
def test_collect_release_provenance_rejects_runtime_path_escape(tmp_path):
@@ -226,8 +278,15 @@ def test_collect_release_provenance_reports_git_failure_context(tmp_path):
operation_id="op-1",
phase="pre",
checkout_dir=not_repo,
expected_oci_revision="0" * 40,
image_digest="sha256:" + ("a" * 64),
observed_evidence={
"schema_version": 1,
"target_id": "service-a",
"target_kind": "compose-service",
"image_id": "sha256:" + ("b" * 64),
"image_digest": "sha256:" + ("a" * 64),
"oci_revision": "0" * 40,
"observed_at": "2026-09-20T02:00:00Z",
},
runtime_files=["tracked.txt"],
capture_commands={
"pending_slots": [
@@ -291,8 +350,7 @@ def test_empty_broker_log_is_explicitly_captured_as_no_matching_events(tmp_path)
@pytest.mark.parametrize(
("field", "value"),
[
("expected_oci_revision", None),
("image_digest", "not-a-digest"),
("observed_evidence", {}),
("runtime_files", []),
("capture_commands", {}),
("instance", "../escape"),
tests/adapters/persistence/test_release_provenance_contract.py
+188
View File
@@ -0,0 +1,188 @@
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import pytest
from trex_core.adapters.persistence.ops_checks import (
load_observed_evidence,
verify_release_provenance_pair,
)
DIGEST_A = "sha256:" + "a" * 64
DIGEST_B = "sha256:" + "b" * 64
REVISION = "c" * 40
def _evidence() -> dict:
return {
"schema_version": 1,
"target_id": "service-a",
"target_kind": "compose-service",
"image_id": DIGEST_A,
"image_digest": DIGEST_B,
"oci_revision": REVISION,
"observed_at": "2026-09-20T02:00:00Z",
}
def test_load_observed_evidence_accepts_exact_v1(tmp_path: Path):
path = tmp_path / "evidence.json"
path.write_text(json.dumps(_evidence()), encoding="utf-8")
assert load_observed_evidence(path) == _evidence()
@pytest.mark.parametrize(
"raw",
[
'{"schema_version":1,"schema_version":1}',
json.dumps({**_evidence(), "unknown": True}),
json.dumps({**_evidence(), "target_id": ""}),
json.dumps({**_evidence(), "schema_version": True}),
json.dumps({**_evidence(), "image_id": "SHA256:" + "A" * 64}),
json.dumps({**_evidence(), "observed_at": "2026-09-20T02:00:00+09:00"}),
json.dumps({**_evidence(), "target_id": "token=must-not-leak"}),
],
)
def test_load_observed_evidence_rejects_invalid_or_sensitive_input(tmp_path: Path, raw: str):
path = tmp_path / "evidence.json"
path.write_text(raw, encoding="utf-8")
with pytest.raises(ValueError):
load_observed_evidence(path)
def _record(*, phase: str, operation_id: str = "op-1", gate_passed: bool = True) -> dict:
return {
"schema_version": 1,
"timestamp_utc": "2026-09-20T02:00:00Z",
"actor": "operator-a",
"reason": "release",
"instance": "instance-a",
"command_class": "deploy" if phase == "pre" else "recreate",
"operation_id": operation_id,
"phase": phase,
"observed_evidence": {**_evidence(), "source": "operator_observed"},
"checkout": {
"path": "/audit/checkout",
"head_revision": REVISION,
"detached_head": True,
"git_status_porcelain": [],
"git_clean": True,
"dirty_paths": [],
},
"runtime_file_verification": [
{
"path": "runtime.py",
"path_within_checkout": True,
"exists": True,
"current_sha256": "d" * 64,
"expected_revision": REVISION,
"expected_sha256": "d" * 64,
"matches_expected_revision": True,
}
],
"state_capture": {
"command_outputs": {
key: {
"command": ["read-only"],
"returncode": 0,
"stdout": {},
"stdout_format": "json",
"stderr": "",
}
for key in ("pending_slots", "reconciliation_slots", "queue_consumers")
},
"broker_side_effect_logs": {
"path": "/tmp/broker.log",
"exists": True,
"capture_status": "captured_empty",
"file_sha256": "e" * 64,
"scanned_lines": 0,
"matches": [],
},
},
"gate": {
"passed": gate_passed,
"deployment_blocked": not gate_passed,
"runtime_files_verified": True,
"state_captures_succeeded": True,
"broker_log_captured": True,
},
}
def _write_record(path: Path, record: dict) -> str:
path.write_text(json.dumps(record, sort_keys=True) + "\n", encoding="utf-8")
return hashlib.sha256(path.read_bytes()).hexdigest()
def test_pair_validator_binds_explicit_paths_and_expected_digests(tmp_path: Path):
pre = tmp_path / "pre.json"
post = tmp_path / "post.json"
pre_sha = _write_record(pre, _record(phase="pre"))
post_sha = _write_record(post, _record(phase="post"))
output = tmp_path / "manifests" / "pair.json"
manifest = verify_release_provenance_pair(
pre_path=pre,
pre_sha256=pre_sha,
post_path=post,
post_sha256=post_sha,
output_path=output,
)
assert manifest["pre"]["record_path"] == str(pre.resolve())
assert manifest["post"]["record_sha256"] == post_sha
assert output.stat().st_mode & 0o222 == 0
@pytest.mark.parametrize("mutation", ["tamper", "failed", "mismatch"])
def test_pair_validator_fails_without_manifest(tmp_path: Path, mutation: str):
pre = tmp_path / "pre.json"
post = tmp_path / "post.json"
pre_sha = _write_record(pre, _record(phase="pre"))
post_record = _record(phase="post")
if mutation == "failed":
post_record["gate"]["passed"] = False
post_record["gate"]["deployment_blocked"] = True
if mutation == "mismatch":
post_record["operation_id"] = "op-2"
post_sha = _write_record(post, post_record)
if mutation == "tamper":
post.write_text(post.read_text() + " ", encoding="utf-8")
output = tmp_path / "pair.json"
with pytest.raises(ValueError):
verify_release_provenance_pair(
pre_path=pre,
pre_sha256=pre_sha,
post_path=post,
post_sha256=post_sha,
output_path=output,
)
assert not output.exists()
def test_pair_validator_does_not_overwrite_existing_output(tmp_path: Path):
pre = tmp_path / "pre.json"
post = tmp_path / "post.json"
pre_sha = _write_record(pre, _record(phase="pre"))
post_sha = _write_record(post, _record(phase="post"))
output = tmp_path / "pair.json"
output.write_text("keep", encoding="utf-8")
with pytest.raises(FileExistsError):
verify_release_provenance_pair(
pre_path=pre,
pre_sha256=pre_sha,
post_path=post,
post_sha256=post_sha,
output_path=output,
)
assert output.read_text(encoding="utf-8") == "keep"