ClaudeandClaude Opus 5.5
8064834c37
- test_refined_dfg::test_source_pin runs its mismatch negative controls first (now including window mode) so missing history cannot skip them. - test_re100: renamed to test_manifest_matches_its_pinned_commit, plus an explicit record that today's src differs from the pin. - Structural manifest checks run before any history-dependent check. - Real tamper test: a flipped digest raises via verify_manifest, and a path absent at the commit raises ValueError. - A blobless clone (tree present, blob missing) is reported as unavailable history, not as a missing path. - All history skips go through history_unavailable(): fatal when CELERIS_REQUIRE_HISTORY=1. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
84 lines
3.6 KiB
Python
84 lines
3.6 KiB
Python
"""Record runtime checkout identity separately from the reviewed source origin."""
|
|
|
|
import subprocess
|
|
|
|
|
|
def checkout_provenance(repo):
|
|
"""Require a Git checkout; never label changed-source runs as historical evidence."""
|
|
revision = subprocess.check_output(
|
|
["git", "-C", str(repo), "rev-parse", "HEAD"], text=True
|
|
).strip()
|
|
dirty = subprocess.check_output(
|
|
["git", "-C", str(repo), "status", "--porcelain", "--untracked-files=normal"],
|
|
text=True,
|
|
)
|
|
return dict(checkout_revision=revision, checkout_dirty=bool(dirty))
|
|
|
|
|
|
class HistoryUnavailable(RuntimeError):
|
|
"""The clone cannot show the recorded commit (no git, shallow clone, installed copy)."""
|
|
|
|
|
|
def historical_sha256(repo, commit, paths):
|
|
"""SHA-256 of each path as stored in `commit`, read with git, never from the working tree.
|
|
|
|
Raises HistoryUnavailable when git or the commit object is missing, so callers can
|
|
report an unchecked integrity record explicitly instead of passing or failing silently.
|
|
"""
|
|
import hashlib
|
|
git = ["git", "-C", str(repo)]
|
|
try:
|
|
subprocess.run([*git, "cat-file", "-e", f"{commit}^{{commit}}"], check=True,
|
|
capture_output=True)
|
|
except (OSError, subprocess.CalledProcessError) as error:
|
|
raise HistoryUnavailable(f"commit {commit} is not available in {repo}") from error
|
|
digests = {}
|
|
for path in paths:
|
|
listed = subprocess.run([*git, "ls-tree", "--name-only", commit, "--", path],
|
|
capture_output=True, text=True)
|
|
if listed.returncode:
|
|
raise HistoryUnavailable(f"tree of {commit} is not available in {repo}")
|
|
if listed.stdout.strip() != path:
|
|
raise ValueError(f"{path} is not part of {commit}")
|
|
shown = subprocess.run([*git, "show", f"{commit}:{path}"], capture_output=True)
|
|
if shown.returncode: # e.g. a blobless partial clone without network access
|
|
raise HistoryUnavailable(f"content of {path} at {commit} is not available in {repo}")
|
|
digests[path] = hashlib.sha256(shown.stdout).hexdigest()
|
|
return digests
|
|
|
|
|
|
def verify_manifest(repo, commit, digests):
|
|
"""Raise RuntimeError naming the first path whose recorded digest differs at `commit`."""
|
|
actual = historical_sha256(repo, commit, digests)
|
|
for path, digest in digests.items():
|
|
if actual[path] != digest:
|
|
raise RuntimeError(f"Historical manifest mismatch at {commit}: {path}")
|
|
return actual
|
|
|
|
|
|
def historical_tree(repo, commit, prefix="src"):
|
|
"""Tracked file names under `prefix` at `commit` (HistoryUnavailable if missing)."""
|
|
historical_sha256(repo, commit, [])
|
|
return subprocess.check_output(["git", "-C", str(repo), "ls-tree", "-r", "--name-only",
|
|
commit, "--", prefix], text=True).splitlines()
|
|
|
|
|
|
def historical_files(repo, commit, paths, dest):
|
|
"""Write each path as stored in `commit` under `dest` (HistoryUnavailable if missing)."""
|
|
from pathlib import Path
|
|
historical_sha256(repo, commit, [])
|
|
for path in paths:
|
|
target = Path(dest) / path
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{path}"],
|
|
check=True, capture_output=True).stdout)
|
|
|
|
|
|
def history_unavailable(testcase, error):
|
|
"""Skip with the reason, or fail when CELERIS_REQUIRE_HISTORY=1 (release gating)."""
|
|
import os
|
|
message = f"historical integrity NOT checked: {error}"
|
|
if os.environ.get("CELERIS_REQUIRE_HISTORY") == "1":
|
|
raise AssertionError(message)
|
|
testcase.skipTest(message)
|