feat(tools): read-time provenance footer for sensitive artifacts

core/harness/tools/files.py

@@ -12,11 +12,14 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import os
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any
 
 from core.agent_runtime.tools.base import Tool, tool_parameters
+from core.harness.permissions import is_sensitive_path
 from core.harness.tools.diagnostics import format_diagnostics, run_diagnostics
 from core.harness.tools.replace import ReplaceError, replace
 
@@ -32,6 +35,50 @@ def _resolve(workspace: str, file_path: str) -> Path:
     return (p if p.is_absolute() else base / p).resolve()
 
 
+def _provenance_footer(target: Path, workspace: str) -> str:
+    """Cross-session artifact provenance (2610.13227 §5.2 proxy).
+
+    When a read touches a sensitive artifact (MCP/hook/config chain), append
+    a read-time snapshot footer (mtime/size/sha256 head) so the agent sees
+    mutation facts before reusing the artifact. This is a proxy for
+    SkillRelayGuard's write-time provenance ledger: writer identity and
+    session origin are NOT recorded here — only detectable mutation facts.
+    Default on; ``DEEPCODE_PROVENANCE_CONTEXT=0`` disables. Never raises:
+    a failed stat degrades to fewer fields, a total failure omits the footer.
+    """
+    flag = os.environ.get("DEEPCODE_PROVENANCE_CONTEXT", "1").strip().lower()
+    if flag in ("0", "false", "off", "no"):
+        return ""
+    try:
+        if not is_sensitive_path(str(target), workspace):
+            return ""
+    except Exception:
+        return ""
+    parts: list[str] = []
+    try:
+        st = target.stat()
+        mtime = datetime.fromtimestamp(st.st_mtime, tz=timezone.utc)
+        parts.append(f"mtime={mtime.strftime('%Y-%m-%dT%H:%M:%SZ')}")
+        parts.append(f"size={st.st_size}B")
+    except OSError:
+        pass
+    try:
+        h = hashlib.sha256()
+        with target.open("rb") as f:
+            h.update(f.read(65536))
+        parts.append(f"sha256_64k={h.hexdigest()[:16]}")
+    except OSError:
+        pass
+    if not parts:
+        return ""
+    return (
+        "\n[provenance] sensitive artifact: "
+        + " ".join(parts)
+        + " (read-time snapshot; cross-session writes to MCP/hook configs"
+        " are code-execution vectors — 2610.13227 IP)"
+    )
+
+
 def _read_window(
     target: Path, offset: int, limit: int
 ) -> tuple[list[str], int, bool, bool]:
@@ -154,6 +201,7 @@ def _read(self, file_path: str, offset: int, limit: int) -> str:
             )
         elif more:
             result += f"\n… [more lines; use offset={next_line} to continue]"
+        result += _provenance_footer(target, self._workspace)
         return result
 
 

tests/test_file_provenance.py

@@ -0,0 +1,75 @@
+"""Regression tests: sensitive-artifact provenance footer on read (2610.13227 §5.2 proxy).
+
+The Read tool appends a read-time snapshot footer (mtime/size/sha256 head) when
+the target matches SENSITIVE_PATH_PATTERNS (credential stores on upstream main),
+so a later session sees mutation facts before reusing the artifact. Proxy, not
+write-time ledger: writer identity is not recorded. Default on; kill switch
+DEEPCODE_PROVENANCE_CONTEXT=0; footer never breaks a read.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from core.harness.tools.files import ReadTool, _provenance_footer
+
+
+@pytest.fixture()
+def workspace(tmp_path):
+    ws = tmp_path / "ws"
+    ws.mkdir()
+    return str(ws)
+
+
+def _read(workspace: str, file_path: str) -> str:
+    return asyncio.run(
+        ReadTool(workspace).execute(file_path=file_path, offset=1, limit=100)
+    )
+
+
+def test_sensitive_read_gets_footer(workspace):
+    p = f"{workspace}/.env"
+    with open(p, "w", encoding="utf-8") as f:
+        f.write("SECRET=1\n")
+    out = _read(workspace, ".env")
+    assert "[provenance]" in out
+    assert "sha256_64k=" in out
+    assert "mtime=" in out
+    assert "size=" in out
+    assert "2610.13227" in out
+
+
+def test_normal_read_has_no_footer(workspace):
+    p = f"{workspace}/notes.txt"
+    with open(p, "w", encoding="utf-8") as f:
+        f.write("hello")
+    out = _read(workspace, "notes.txt")
+    assert "[provenance]" not in out
+    assert "hello" in out
+
+
+def test_kill_switch_env(workspace, monkeypatch):
+    monkeypatch.setenv("DEEPCODE_PROVENANCE_CONTEXT", "0")
+    p = f"{workspace}/.env"
+    with open(p, "w", encoding="utf-8") as f:
+        f.write("SECRET=1\n")
+    out = _read(workspace, ".env")
+    assert "[provenance]" not in out
+
+
+def test_footer_is_helper_pure_on_missing_file(tmp_path):
+    # A vanished path must not raise: helper degrades to "" (footer is
+    # best-effort, never breaks the read contract).
+    out = _provenance_footer(tmp_path / "no-such.mcp.json", str(tmp_path))
+    assert out == ""
+
+
+def test_footer_empty_for_non_sensitive(workspace):
+    p = f"{workspace}/plain.md"
+    with open(p, "w", encoding="utf-8") as f:
+        f.write("x")
+    from pathlib import Path
+
+    assert _provenance_footer(Path(p), workspace) == ""