fix: require trusted recovery manifest digest

Author: salma2vec

examples/complete_edit_pilot.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import argparse
+import hmac
 import json
 import os
 import shutil
@@ -56,7 +57,10 @@ def _read_regular(path: Path) -> bytes:
 
 
 def verify_parent_evidence(
-    interrupted: Path, parent_manifest: Path, source_root: Path
+    interrupted: Path,
+    parent_manifest: Path,
+    expected_parent_manifest_sha256: str,
+    source_root: Path,
 ) -> dict[str, bytes]:
     """Verify independently supplied parent identity and exact immutable artifacts."""
     parent = interrupted.resolve(strict=True)
@@ -66,6 +70,8 @@ def verify_parent_evidence(
     if manifest_resolved.is_relative_to(parent):
         raise ValueError("Parent manifest must be supplied outside the parent artifact directory")
     manifest_bytes = _read_regular(parent_manifest)
+    if not hmac.compare_digest(digest_bytes(manifest_bytes), expected_parent_manifest_sha256):
+        raise ValueError("Parent manifest digest differs from the independently supplied digest")
     manifest = json.loads(manifest_bytes)
     required = {"schema", "parent_directory", "artifacts"}
     if not isinstance(manifest, dict) or set(manifest) != required:
@@ -124,14 +130,19 @@ def missing_cell(rows: list[dict], task_ids: list[str]) -> tuple[str, str, int,
 
 
 def run(
-    repo: Path, interrupted: Path, output: Path, tokenizer: Path, parent_manifest: Path
+    repo: Path,
+    interrupted: Path,
+    output: Path,
+    tokenizer: Path,
+    parent_manifest: Path,
+    parent_manifest_sha256: str,
 ) -> dict:
     root = Path(__file__).resolve().parents[1]
     if subprocess.check_output(["git", "status", "--porcelain"], cwd=root, text=True).strip():
         raise ValueError("Freeze recovery executor before inference")
     if output.exists() or (interrupted / "summary.json").exists():
         raise ValueError("Use fresh output and an incomplete original pilot")
-    parent = verify_parent_evidence(interrupted, parent_manifest, root)
+    parent = verify_parent_evidence(interrupted, parent_manifest, parent_manifest_sha256, root)
     protocol = json.loads(parent["protocol.json"])
     if digest_bytes(_read_regular(MANIFEST)) != protocol["manifest_sha256"]:
         raise ValueError("Frozen manifest changed")
@@ -194,9 +205,7 @@ def run(
             "restoration": asdict(restoration),
         }
         append_row(output / "controls.jsonl", control)
-        old_controls = [
-            json.loads(x) for x in parent["controls.jsonl"].decode().splitlines()
-        ]
+        old_controls = [json.loads(x) for x in parent["controls.jsonl"].decode().splitlines()]
         previous = next(c for c in old_controls if c["task_id"] == task_id)
         if not (
             baseline.passed
@@ -280,13 +289,21 @@ def run(
     for name in ("repo", "interrupted", "output", "tokenizer"):
         parser.add_argument("--" + name, type=Path, required=True)
     parser.add_argument("--parent-manifest", type=Path, required=True)
+    parser.add_argument("--parent-manifest-sha256", required=True)
     parser.add_argument("--allow-local-execution", action="store_true")
     args = parser.parse_args()
     if not args.allow_local_execution:
         parser.error("Explicit local execution acknowledgement is required")
     print(
         json.dumps(
-            run(args.repo, args.interrupted, args.output, args.tokenizer, args.parent_manifest),
+            run(
+                args.repo,
+                args.interrupted,
+                args.output,
+                args.tokenizer,
+                args.parent_manifest,
+                args.parent_manifest_sha256,
+            ),
             indent=2,
         )
     )

examples/results/edit-format-pilot/RECOVERY.md

@@ -7,8 +7,9 @@ task because available disk fell below its prespecified 2 GiB floor. The first
 discarded or repeated.
 
 The recovery executor completes only that unattempted condition. Before calling
-the model it verifies every original experiment source hash, the task manifest,
-the model digest, the unchanged parameters, and a unique 23-cell partial matrix.
+the model it verifies an independently recorded SHA-256 digest for the external
+parent manifest, every original experiment source hash, the task manifest, the
+model digest, the unchanged parameters, and a unique 23-cell partial matrix.
 It reconstructs the same prompt from the pinned mutant and repeats the scoring
 controls, requiring the original test identifiers. The disk floor remains in
 force. An error from the remaining model call counts as its outcome, without a
@@ -18,9 +19,16 @@ interrupted records remain unchanged.
 ```bash
 python -m examples.complete_edit_pilot --repo /path/to/boltons \
   --interrupted /path/to/interrupted-run --output /fresh/recovery-run \
-  --tokenizer /path/to/pinned-qwen-tokenizer.json --allow-local-execution
+  --tokenizer /path/to/pinned-qwen-tokenizer.json \
+  --parent-manifest /trusted/path/parent-manifest.json \
+  --parent-manifest-sha256 "$INDEPENDENTLY_RECORDED_SHA256" \
+  --allow-local-execution
 ```
 
+Obtain `INDEPENDENTLY_RECORDED_SHA256` from a trusted record made independently
+of the interrupted run and its parent manifest. Computing it from either input
+at recovery time does not establish the parent identity and must not be used.
+
 The combined result spans two process windows. It remains an exposed-task
 diagnostic, not a heldout comparison. A single validated repair shows that the
 interface can work; it does not establish a reliable repair system or a retrieval

src/cegraph/parser.py

@@ -261,8 +261,8 @@ def parse_python_file(repo_root: Path, file_path: Path) -> ParsedModule:
     if file_path.is_symlink() or not file_path.resolve().is_relative_to(repo_root.resolve()):
         raise ValueError("Source file must be contained in the repository without symlinks")
     before = file_path.lstat()
-    if not stat.S_ISREG(before.st_mode):
-        raise ValueError("Source file must be a regular file")
+    if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1:
+        raise ValueError("Source file must be a single-link regular file")
     flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
     try:
         descriptor = os.open(file_path, flags)
@@ -272,9 +272,10 @@ def parse_python_file(repo_root: Path, file_path: Path) -> ParsedModule:
         opened = os.fstat(descriptor)
         if (
             not stat.S_ISREG(opened.st_mode)
+            or opened.st_nlink != 1
             or (opened.st_dev, opened.st_ino) != (before.st_dev, before.st_ino)
         ):
-            raise ValueError("Source file changed or is not a regular file")
+            raise ValueError("Source file changed or is not a single-link regular file")
         with os.fdopen(descriptor, "r", encoding="utf-8") as handle:
             descriptor = -1
             source = handle.read()

tests/test_edit_pilot_recovery.py

@@ -43,16 +43,14 @@ def test_ambiguous_or_out_of_protocol_recovery_is_rejected(change):
         missing_cell(rows, ["a", "b", "c"])
 
 
-def parent_fixture(tmp_path: Path) -> tuple[Path, Path, Path]:
+def parent_fixture(tmp_path: Path) -> tuple[Path, Path, Path, str]:
     source_root = tmp_path / "source"
     source_root.mkdir()
     (source_root / "executor.py").write_text("value = 1\n")
     parent = tmp_path / "parent"
     parent.mkdir()
     protocol = {
-        "source_sha256": {
-            "executor.py": digest_bytes((source_root / "executor.py").read_bytes())
-        }
+        "source_sha256": {"executor.py": digest_bytes((source_root / "executor.py").read_bytes())}
     }
     artifacts = {
         "protocol.json": json.dumps(protocol).encode(),
@@ -71,14 +69,14 @@ def parent_fixture(tmp_path: Path) -> tuple[Path, Path, Path]:
             }
         )
     )
-    return source_root, parent, manifest
+    return source_root, parent, manifest, digest_bytes(manifest.read_bytes())
 
 
 def test_parent_evidence_requires_independent_manifest_and_contained_regular_files(
     tmp_path: Path,
 ) -> None:
-    source_root, parent, manifest = parent_fixture(tmp_path)
-    verified = verify_parent_evidence(parent, manifest, source_root)
+    source_root, parent, manifest, manifest_digest = parent_fixture(tmp_path)
+    verified = verify_parent_evidence(parent, manifest, manifest_digest, source_root)
     assert set(verified) == {
         "parent_manifest",
         "protocol.json",
@@ -88,23 +86,33 @@ def test_parent_evidence_requires_independent_manifest_and_contained_regular_fil
     inside_manifest = parent / "parent-manifest.json"
     inside_manifest.write_bytes(manifest.read_bytes())
     with pytest.raises(ValueError, match="outside the parent artifact directory"):
-        verify_parent_evidence(parent, inside_manifest, source_root)
+        verify_parent_evidence(parent, inside_manifest, manifest_digest, source_root)
 
     outside = tmp_path / "outside"
     outside.write_text("forged\n")
     (parent / "results.jsonl").unlink()
     (parent / "results.jsonl").symlink_to(outside)
     with pytest.raises(ValueError, match="regular file"):
-        verify_parent_evidence(parent, manifest, source_root)
+        verify_parent_evidence(parent, manifest, manifest_digest, source_root)
+
+
+def test_parent_evidence_rejects_manifest_changed_after_digest_is_supplied(
+    tmp_path: Path,
+) -> None:
+    source_root, parent, manifest, manifest_digest = parent_fixture(tmp_path)
+    manifest.write_text("{}")
+    with pytest.raises(ValueError, match="independently supplied digest"):
+        verify_parent_evidence(parent, manifest, manifest_digest, source_root)
 
 
 def test_parent_source_paths_cannot_escape_source_root(tmp_path: Path) -> None:
-    source_root, parent, manifest = parent_fixture(tmp_path)
+    source_root, parent, manifest, _ = parent_fixture(tmp_path)
     protocol = json.loads((parent / "protocol.json").read_text())
     protocol["source_sha256"] = {"../outside.py": "0" * 64}
     (parent / "protocol.json").write_text(json.dumps(protocol))
     payload = json.loads(manifest.read_text())
     payload["artifacts"]["protocol.json"] = digest_bytes((parent / "protocol.json").read_bytes())
     manifest.write_text(json.dumps(payload))
+    manifest_digest = digest_bytes(manifest.read_bytes())
     with pytest.raises(ValueError, match="contained relative path"):
-        verify_parent_evidence(parent, manifest, source_root)
+        verify_parent_evidence(parent, manifest, manifest_digest, source_root)

tests/test_graph.py

@@ -78,3 +78,12 @@ def swap_then_open(path, flags):
     monkeypatch.setattr("cegraph.parser.os.open", swap_then_open)
     with pytest.raises(ValueError, match="changed or became unsafe"):
         parse_python_file(tmp_path, source)
+
+
+def test_parser_rejects_hardlinked_source(tmp_path: Path) -> None:
+    source = tmp_path / "source.py"
+    source.write_text("value = 1\n")
+    os.link(source, tmp_path / "second-name.py")
+
+    with pytest.raises(ValueError, match="single-link regular file"):
+        parse_python_file(tmp_path, source)