From 55a95e77d53a0dea0822bbe8ddc918528b303a38 Mon Sep 17 00:00:00 2001 From: raymondginger Date: Sun, 18 Oct 2026 18:34:37 +0800 Subject: [PATCH] feat(tools): append read-time provenance footer for sensitive artifacts Sensitive reads get a compact provenance note (mtime/size/sha256_64k snapshot, proxy-marked) so edited/binary/placeholder data stays identifiable at the moment it enters context. DEEPCODE_PROVENANCE_CONTEXT=0 disables; the footer is additive and fail-safe (best-effort, never aborts). --- core/harness/tools/files.py | 48 ++++++++++++++++++++++ tests/test_file_provenance.py | 75 +++++++++++++++++++++++++++++++++++ 2 files changed, 123 insertions(+) create mode 100644 tests/test_file_provenance.py diff --git a/core/harness/tools/files.py b/core/harness/tools/files.py index 1a0c8842..99ba6afc 100644 --- a/core/harness/tools/files.py +++ b/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 diff --git a/tests/test_file_provenance.py b/tests/test_file_provenance.py new file mode 100644 index 00000000..bd3f5fb2 --- /dev/null +++ b/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) == ""