fix(kernel): close verify_evidence TOCTOU via fd-pinned open-then-verify (#67)

loop/evidence.py

@@ -9,7 +9,9 @@
 
 import hashlib
 import json
+import os
 import re
+import stat
 from pathlib import Path
 from typing import Any, Mapping
 
@@ -151,18 +153,29 @@ def verify_evidence(evidence: Mapping[str, Any], *, workspace_root: str | Path)
                 "issues": [ContractIssue("workspace_escape", f"evidence path escapes workspace: {uri}")]}
     checks["within_workspace"] = True
     try:
-        is_file = resolved.is_file()
+        fd = os.open(
+            resolved,
+            os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0),
+        )
     except OSError:
         checks["path_exists"] = False
         return {"ok": False, "checks": checks,
                 "issues": [ContractIssue("missing_evidence_path", f"evidence path is unavailable: {uri}")]}
-    if not is_file:
+    try:
+        file_stat = os.fstat(fd)
+    except OSError:
+        os.close(fd)
+        checks["path_exists"] = False
+        return {"ok": False, "checks": checks,
+                "issues": [ContractIssue("missing_evidence_path", f"evidence path is unavailable: {uri}")]}
+    if not stat.S_ISREG(file_stat.st_mode):
+        os.close(fd)
         return {"ok": False, "checks": checks,
                 "issues": [ContractIssue("not_a_file", f"evidence path is not a file: {uri}")]}
 
     digest = hashlib.sha256()
     try:
-        with resolved.open("rb") as source:
+        with os.fdopen(fd, "rb") as source:
             while chunk := source.read(64 * 1024):
                 digest.update(chunk)
     except OSError:

scripts/test_adversarial_process.py

@@ -179,29 +179,29 @@ def test_sqlite_raw_file_tamper_bypassing_sql_interface_is_not_detected(tmp_path
     assert SQLiteEventStore(path).read("run")[0]["payload"] == {"workspace": "X"}
 
 
-@pytest.mark.xfail(
-    strict=True,
-    reason="issue #67: verify_evidence TOCTOU — containment is not rechecked before the hash read",
-)
 def test_symlink_swap_between_containment_check_and_hash_read_escapes_workspace(tmp_path, monkeypatch) -> None:
     inside = tmp_path / "proof.txt"
     outside = tmp_path.parent / f"{tmp_path.name}-outside-proof.txt"
     inside.write_bytes(b"inside")
     outside.write_bytes(b"outside")
     record = _evidence("proof.txt", b"outside")
-    real_is_file = Path.is_file
+    real_open = os.open
+    swapped = False
 
-    def swap_after_containment(path: Path) -> bool:
-        if path == inside:
+    def swap_after_containment(path, flags, *args, **kwargs):
+        nonlocal swapped
+        if path == inside and not swapped:
             inside.unlink()
             os.symlink(outside, inside)
-        return real_is_file(path)
+            swapped = True
+        return real_open(path, flags, *args, **kwargs)
 
-    monkeypatch.setattr(Path, "is_file", swap_after_containment)
+    monkeypatch.setattr(os, "open", swap_after_containment)
 
     try:
         report = verify_evidence(record, workspace_root=tmp_path)
     finally:
         outside.unlink(missing_ok=True)
 
     assert report["ok"] is False
+    assert "missing_evidence_path" in {issue["code"] for issue in report["issues"]}

scripts/test_evidence.py

@@ -219,10 +219,10 @@ def test_verify_evidence_handles_artifact_io_failure_without_raising(tmp_path, m
     path = tmp_path / "proof.txt"
     path.write_text("evidence", encoding="utf-8")
 
-    def unavailable(self, *args, **kwargs):
+    def unavailable(*args, **kwargs):
         raise PermissionError("unavailable")
 
-    monkeypatch.setattr(Path, "open", unavailable)
+    monkeypatch.setattr(os, "open", unavailable)
     report = verify_evidence(evidence(uri="proof.txt"), workspace_root=tmp_path)
     assert "missing_evidence_path" in issue_codes(report)