From 0ff19bee7357a0aea63c9fbc07ffae248f516314 Mon Sep 17 00:00:00 2001
From: Sollan Systems <khall0239@gmail.com>
Date: Wed, 15 Jul 2026 07:53:42 -0400
Subject: [PATCH] fix(kernel): close verify_evidence TOCTOU via fd-pinned
 open-then-verify (#67)

Replace the containment-check-then-read-by-name flow with os.open(resolved,
O_RDONLY|O_NOFOLLOW|O_NONBLOCK) -> fstat -> S_ISREG -> fdopen hashing. A leaf
symlink swap between check and read now fails ELOOP (missing_evidence_path)
instead of hashing attacker-controlled content outside the workspace; an
already-open fd is immune to later directory-entry changes. O_NONBLOCK stops a
FIFO swap from hanging the open. Flags degrade via getattr on platforms
without them. The strict xfail pinning the vulnerability flips to a normal
passing test.

Known residuals (by design, documented in the PR): intermediate-directory
component races (needs non-stdlib openat2) and hardlink swaps (regular files,
O_NOFOLLOW inert).
---
 loop/evidence.py                    | 19 ++++++++++++++++---
 scripts/test_adversarial_process.py | 18 +++++++++---------
 scripts/test_evidence.py            |  4 ++--
 3 files changed, 27 insertions(+), 14 deletions(-)

diff --git a/loop/evidence.py b/loop/evidence.py
index f6a7b29..bc4ed28 100644
--- a/loop/evidence.py
+++ b/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:
diff --git a/scripts/test_adversarial_process.py b/scripts/test_adversarial_process.py
index 682618a..b4651cf 100644
--- a/scripts/test_adversarial_process.py
+++ b/scripts/test_adversarial_process.py
@@ -179,25 +179,24 @@ 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)
@@ -205,3 +204,4 @@ def swap_after_containment(path: Path) -> bool:
         outside.unlink(missing_ok=True)
 
     assert report["ok"] is False
+    assert "missing_evidence_path" in {issue["code"] for issue in report["issues"]}
diff --git a/scripts/test_evidence.py b/scripts/test_evidence.py
index 097cfdd..ec35fa1 100644
--- a/scripts/test_evidence.py
+++ b/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)