diff --git a/core/harness/hooks/__init__.py b/core/harness/hooks/__init__.py index bba58e39..451bb1e9 100644 --- a/core/harness/hooks/__init__.py +++ b/core/harness/hooks/__init__.py @@ -10,7 +10,7 @@ hooks pays nothing. """ -from core.harness.hooks.discovery import Handler, DiscoveryResult, discover_hooks +from core.harness.hooks.discovery import DiscoveryResult, Handler, discover_hooks from core.harness.hooks.engine import ( ContextOutcome, HooksEngine, @@ -19,15 +19,18 @@ PreToolUseOutcome, StopOutcome, ) +from core.harness.hooks.trust import HookTrustStore, declaration_digest __all__ = [ - "Handler", + "ContextOutcome", "DiscoveryResult", - "discover_hooks", + "Handler", + "HookTrustStore", "HooksEngine", - "PreToolUseOutcome", + "PermissionRequestOutcome", "PostToolUseOutcome", - "ContextOutcome", + "PreToolUseOutcome", "StopOutcome", - "PermissionRequestOutcome", + "declaration_digest", + "discover_hooks", ] diff --git a/core/harness/hooks/discovery.py b/core/harness/hooks/discovery.py index c98164a8..2ab67d9b 100644 --- a/core/harness/hooks/discovery.py +++ b/core/harness/hooks/discovery.py @@ -30,6 +30,7 @@ import json from dataclasses import dataclass, field from pathlib import Path +from typing import TYPE_CHECKING from core.harness.hooks.events import ( HOOK_EVENT_NAMES, @@ -37,6 +38,9 @@ validate_matcher, ) +if TYPE_CHECKING: # avoid a runtime import cycle (trust imports discovery). + from core.harness.hooks.trust import HookTrustStore + _DEFAULT_TIMEOUT_SEC = 600 _SESSION_END_DEFAULT_TIMEOUT_SEC = 2 _SESSION_END_MAX_TIMEOUT_SEC = 60 @@ -74,7 +78,12 @@ def _hook_source_files(workspace: str, home: str | None) -> list[tuple[Path, str ] -def discover_hooks(workspace: str, home: str | None = None) -> DiscoveryResult: +def discover_hooks( + workspace: str, + home: str | None = None, + *, + trust_store: HookTrustStore | None = None, +) -> DiscoveryResult: """Discover all configured hook handlers for ``workspace``. Returns a :class:`DiscoveryResult`; an empty ``handlers`` list means the @@ -84,7 +93,20 @@ def discover_hooks(workspace: str, home: str | None = None) -> DiscoveryResult: handlers: list[Handler] = [] warnings: list[str] = [] order = 0 - for path, source in _hook_source_files(workspace, home): + sources = _hook_source_files(workspace, home) + if trust_store is not None: + from core.harness.hooks.trust import declaration_digest + + digest = declaration_digest(workspace, home=home) + if not trust_store.is_approved(workspace, digest): + warnings.append( + "project hook declarations are not approved (or changed since " + "approval); project hooks were skipped" + ) + sources = [ + (path, source) for path, source in sources if source != "project" + ] + for path, source in sources: events = _load_hook_events(path, warnings) if not events: continue diff --git a/core/harness/hooks/engine.py b/core/harness/hooks/engine.py index 71bdc19c..830d151c 100644 --- a/core/harness/hooks/engine.py +++ b/core/harness/hooks/engine.py @@ -109,9 +109,16 @@ def discover( *, model: str = "", permission_mode: str = "default", + trust_store=None, ): - """Build an engine for ``workspace``, or ``None`` if no hooks configured.""" - result = discover_hooks(workspace, home=home) + """Build an engine for ``workspace``, or ``None`` if no hooks configured. + + ``trust_store`` is an optional + :class:`~core.harness.hooks.trust.HookTrustStore`; when given, project + hook files are discovered only if their declaration digest is + approved, so a changed ``hooks.json`` stops running until re-approved. + """ + result = discover_hooks(workspace, home=home, trust_store=trust_store) if not result.handlers: return None, result.warnings engine = cls( diff --git a/core/harness/hooks/trust.py b/core/harness/hooks/trust.py new file mode 100644 index 00000000..3cf8183d --- /dev/null +++ b/core/harness/hooks/trust.py @@ -0,0 +1,211 @@ +"""Project hook declaration trust — sha256 digest approval store (ZCode). + +ZCode's rule: a project's ``hooks.json`` is *workspace-supplied code*. Trusting +a project once must not silently trust it forever — if the declaration changes, +the approval is void and the operator must re-approve the new content. + +The existing engine gate is a boolean ``project_trusted``. This module layers a +content digest on top: + +* :func:`declaration_digest` hashes the canonical declaration content, so + whitespace/key-order churn does not invalidate an approval while any real + command/matcher/event change does. +* :class:`HookTrustStore` persists approved digests keyed by workspace in a + small JSON file under ``deepcode_home()``. It is intentionally dependency-free + (no SQLite) so hook wiring stays cheap. +* The gate stays **opt-in**: with no store configured, behavior is the existing + boolean ``project_trusted`` path, byte-for-byte. + +An approval is per (workspace, digest). A changed declaration produces a new +digest, so the old approval no longer matches and project hooks revert to +blocked until the operator re-approves. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import tempfile +import threading +from pathlib import Path + +logger = logging.getLogger(__name__) + +_STORE_FILENAME = "hook_trust.json" +_STORE_VERSION = 1 + + +def declaration_digest( + workspace: str, + *, + home: str | None = None, + paths: tuple[Path, ...] | None = None, +) -> str: + """sha256 digest over the canonical project hook declarations. + + ``paths`` lets callers pass the exact discovered project files; when omitted + the two project sources are used (user-level config is the operator's own + machine and is never gated, so it never feeds the digest). + + Only *declaration* content is hashed — event name, matcher, command, timeout, + priority, type — so cosmetic edits (reordering keys, reindenting) keep the + approval while any semantic change produces a new digest. Files that are + absent or unreadable contribute a stable marker so a file appearing/disappearing + changes the digest. + """ + from core.harness.hooks.discovery import _hook_source_files + + if paths is None: + paths = tuple( + path + for path, source in _hook_source_files(workspace, home) + if source == "project" + ) + hasher = hashlib.sha256() + for path in sorted(paths, key=str): + hasher.update(str(path).encode("utf-8")) + hasher.update(b"\x00") + hasher.update(_canonical_declaration(path)) + hasher.update(b"\x00") + return hasher.hexdigest() + + +def _canonical_declaration(path: Path) -> bytes: + """Canonical, order-stable bytes for one hooks file's declarations.""" + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + # Absent/unreadable: a stable per-path marker that differs from real content. + return b"" + hooks = data.get("hooks") if isinstance(data, dict) else None + if not isinstance(hooks, dict): + return b"" + reduced: dict[str, list] = {} + for event_name in sorted(hooks): + groups = hooks[event_name] + if not isinstance(groups, list): + continue + reduced[event_name] = [ + _canonical_group(group) for group in groups if isinstance(group, dict) + ] + return json.dumps(reduced, sort_keys=True, separators=(",", ":")).encode("utf-8") + + +def _canonical_group(group: dict) -> dict: + """Reduce one matcher-group to its declaration-relevant fields.""" + out: dict[str, object] = {} + matcher = group.get("matcher") + if isinstance(matcher, str): + out["matcher"] = matcher + handlers = group.get("hooks") + if isinstance(handlers, list): + out["hooks"] = [ + { + key: handler[key] + for key in ("type", "command", "timeout", "async", "priority") + if isinstance(handler, dict) and key in handler + } + for handler in handlers + if isinstance(handler, dict) + ] + return out + + +class HookTrustStore: + """Persist approved project hook digests per workspace. + + Thread-safe and file-backed. Approvals are a set of digests per workspace so + a workspace can be re-approved across restarts without re-prompting until its + declaration actually changes. + """ + + def __init__(self, path: Path | None = None) -> None: + self._path = path or (self._default_dir() / _STORE_FILENAME) + self._lock = threading.Lock() + + @staticmethod + def _default_dir() -> Path: + from core.config import deepcode_home + + return deepcode_home() + + @property + def path(self) -> Path: + return self._path + + def is_approved(self, workspace: str, digest: str) -> bool: + """True when this exact declaration digest is approved for ``workspace``.""" + with self._lock: + return digest in self._read().get(_key(workspace), []) + + def approve(self, workspace: str, digest: str) -> None: + """Approve one declaration digest for ``workspace`` (idempotent).""" + with self._lock: + data = self._read() + approved = data.setdefault(_key(workspace), []) + if digest not in approved: + approved.append(digest) + self._write(data) + + def revoke(self, workspace: str, digest: str | None = None) -> None: + """Revoke one digest, or every approval for ``workspace`` when None.""" + with self._lock: + data = self._read() + key = _key(workspace) + if key not in data: + return + if digest is None: + del data[key] + else: + approved = [d for d in data[key] if d != digest] + if approved: + data[key] = approved + else: + del data[key] + self._write(data) + + def _read(self) -> dict[str, list[str]]: + try: + raw = json.loads(self._path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + if not isinstance(raw, dict) or raw.get("version") != _STORE_VERSION: + return {} + workspaces = raw.get("workspaces") + if not isinstance(workspaces, dict): + return {} + return { + key: [d for d in value if isinstance(d, str)] + for key, value in workspaces.items() + if isinstance(value, list) + } + + def _write(self, data: dict[str, list[str]]) -> None: + payload = {"version": _STORE_VERSION, "workspaces": data} + self._path.parent.mkdir(parents=True, exist_ok=True) + # Atomic replace so a crash mid-write never corrupts the store. + fd, tmp = tempfile.mkstemp( + dir=str(self._path.parent), prefix=".hook_trust.", suffix=".tmp" + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, indent=2, sort_keys=True) + os.replace(tmp, self._path) + except OSError: + logger.exception("failed to persist hook trust store at %s", self._path) + try: + os.unlink(tmp) + except OSError: + pass + + +def _key(workspace: str) -> str: + return str(Path(workspace).resolve()) + + +__all__ = [ + "HookTrustStore", + "declaration_digest", +] diff --git a/tests/test_hook_declaration_trust.py b/tests/test_hook_declaration_trust.py new file mode 100644 index 00000000..7d77a84b --- /dev/null +++ b/tests/test_hook_declaration_trust.py @@ -0,0 +1,251 @@ +"""Project hook declaration trust — digest approval store (ZCode rule). + +Contract under test: + +- trusting a project once must not silently trust it forever: the digest + covers the canonical DECLARATION content (event/matcher/command/timeout/ + priority/type), so any real change produces a new digest and the old + approval no longer matches; +- cosmetic churn (key reordering, reindenting, non-declaration keys) does + NOT invalidate an approval; +- a file appearing or disappearing changes the digest (stable markers for + absent files); +- the store is per (workspace, digest), persists across instances, and + treats a corrupt store file as empty instead of crashing; +- the gate is opt-in at the engine seam: with no store configured project + discovery is untouched (verified structurally — the engine takes + trust_store=None as its default). +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from core.harness.hooks.trust import HookTrustStore, declaration_digest + +_DECLARATION_V1 = { + "hooks": { + "PostToolUse": [ + { + "matcher": "Bash", + "hooks": [ + {"type": "command", "command": "python lint.py", "timeout": 30} + ], + } + ] + } +} + + +def _write(ws_hooks: Path, payload: dict) -> Path: + ws_hooks.parent.mkdir(parents=True, exist_ok=True) + ws_hooks.write_text(json.dumps(payload), encoding="utf-8") + return ws_hooks + + +def test_digest_is_stable_across_cosmetic_churn(tmp_path: Path) -> None: + path = _write(tmp_path / "hooks.json", _DECLARATION_V1) + before = declaration_digest(str(tmp_path), paths=(path,)) + + # Same semantics: key order shuffled, extra whitespace, non-declaration + # key added inside the handler, and an unrelated top-level key. + churned = { + "version": 2, + "hooks": { + "PostToolUse": [ + { + "hooks": [ + { + "timeout": 30, + "command": "python lint.py", + "description": "cosmetic, not hashed", + "type": "command", + } + ], + "matcher": "Bash", + } + ] + }, + } + path.write_text(json.dumps(churned, indent=4), encoding="utf-8") + after = declaration_digest(str(tmp_path), paths=(path,)) + assert after == before + + +def test_any_semantic_change_produces_a_new_digest(tmp_path: Path) -> None: + path = _write(tmp_path / "hooks.json", _DECLARATION_V1) + before = declaration_digest(str(tmp_path), paths=(path,)) + + for mutation in ( + {"command": "python lint.py && rm -rf /"}, # command swapped + {"timeout": 5}, # timeout changed + ): + payload = json.loads(json.dumps(_DECLARATION_V1)) + payload["hooks"]["PostToolUse"][0]["hooks"][0].update(mutation) + _write(tmp_path / "hooks.json", payload) + assert declaration_digest(str(tmp_path), paths=(path,)) != before + + # A matcher change is semantic too. + payload = json.loads(json.dumps(_DECLARATION_V1)) + payload["hooks"]["PostToolUse"][0]["matcher"] = "Edit" + _write(tmp_path / "hooks.json", payload) + assert declaration_digest(str(tmp_path), paths=(path,)) != before + + +def test_a_file_appearing_or_disappearing_changes_the_digest(tmp_path: Path) -> None: + first = _write(tmp_path / "a" / "hooks.json", _DECLARATION_V1) + second = tmp_path / "b" / "hooks.json" # absent + + with_neither = declaration_digest(str(tmp_path), paths=(first, second)) + _write(second, _DECLARATION_V1) + with_both = declaration_digest(str(tmp_path), paths=(first, second)) + assert with_neither != with_both + + # Absent files contribute a stable marker: two runs over the same tree + # agree, so the digest does not flap across restarts. + assert declaration_digest(str(tmp_path), paths=(first, second)) == with_both + + +def test_declared_but_unreadable_content_differs_from_real_content( + tmp_path: Path, +) -> None: + real = _write(tmp_path / "real.json", _DECLARATION_V1) + broken = tmp_path / "broken.json" + broken.write_text("{not json", encoding="utf-8") + missing = tmp_path / "missing.json" + + real_digest = declaration_digest(str(tmp_path), paths=(real,)) + broken_digest = declaration_digest(str(tmp_path), paths=(broken,)) + missing_digest = declaration_digest(str(tmp_path), paths=(missing,)) + assert broken_digest != real_digest + assert missing_digest != real_digest + assert missing_digest == declaration_digest(str(tmp_path), paths=(missing,)) + + +# --------------------------------------------------------------------------- +# Store +# --------------------------------------------------------------------------- + + +def test_approval_is_per_digest_and_persists_across_instances( + tmp_path: Path, +) -> None: + path = tmp_path / "hook_trust.json" + store = HookTrustStore(path) + workspace = str(tmp_path / "ws") + digest = "d" * 64 + + assert not store.is_approved(workspace, digest) + store.approve(workspace, digest) + store.approve(workspace, digest) # idempotent + assert store.is_approved(workspace, digest) + + # A fresh instance over the same file sees the approval (atomic write). + assert HookTrustStore(path).is_approved(workspace, digest) + + # Another workspace and another digest are independent. + assert not store.is_approved(str(tmp_path / "other-ws"), digest) + assert not store.is_approved(workspace, "e" * 64) + + +def test_changed_declaration_reapproval_cycle(tmp_path: Path) -> None: + ws = tmp_path / "ws" + hooks = _write(ws / "hooks.json", _DECLARATION_V1) + store = HookTrustStore(tmp_path / "hook_trust.json") + + old_digest = declaration_digest(str(ws), paths=(hooks,)) + store.approve(str(ws), old_digest) + + # Declaration changes -> new digest -> old approval no longer matches, + # so project hooks are skipped until re-approved. + payload = json.loads(json.dumps(_DECLARATION_V1)) + payload["hooks"]["PostToolUse"][0]["hooks"][0]["command"] = "python lint2.py" + _write(ws / "hooks.json", payload) + new_digest = declaration_digest(str(ws), paths=(hooks,)) + assert new_digest != old_digest + assert not store.is_approved(str(ws), new_digest) + + store.approve(str(ws), new_digest) + assert store.is_approved(str(ws), new_digest) + # The old approval row remains harmless history; the gate checks the + # CURRENT declaration's digest, which is the new one. + assert store.is_approved(str(ws), old_digest) + + +def test_revoke_single_and_revoke_all(tmp_path: Path) -> None: + store = HookTrustStore(tmp_path / "hook_trust.json") + workspace = str(tmp_path / "ws") + store.approve(workspace, "a" * 64) + store.approve(workspace, "b" * 64) + + store.revoke(workspace, "a" * 64) + assert not store.is_approved(workspace, "a" * 64) + assert store.is_approved(workspace, "b" * 64) + + store.revoke(workspace) + assert not store.is_approved(workspace, "b" * 64) + # Revoking an unknown workspace/digest is a no-op, not an error. + store.revoke(workspace) + store.revoke(workspace, "c" * 64) + + +def test_corrupt_store_file_is_treated_as_empty(tmp_path: Path) -> None: + path = tmp_path / "hook_trust.json" + path.write_text("{corrupted", encoding="utf-8") + store = HookTrustStore(path) + assert not store.is_approved(str(tmp_path / "ws"), "d" * 64) + + # Approving over a corrupt file repairs the store. + store.approve(str(tmp_path / "ws"), "d" * 64) + assert HookTrustStore(path).is_approved(str(tmp_path / "ws"), "d" * 64) + + # A version bump invalidates old approvals instead of misreading them. + path.write_text( + json.dumps({"version": 99, "workspaces": {"ws": ["d" * 64]}}), encoding="utf-8" + ) + assert not store.is_approved(str(tmp_path / "ws"), "d" * 64) + + +def test_engine_gate_is_opt_in_via_its_default_parameter() -> None: + """Structural check: the discovery seam defaults trust_store to None (off).""" + import inspect + + from core.harness.hooks.engine import HooksEngine + + parameter = inspect.signature(HooksEngine.discover).parameters.get("trust_store") + assert parameter is not None + assert parameter.default is None + + +def test_discover_hooks_gate_skips_project_sources_until_approved( + tmp_path: Path, +) -> None: + """End-to-end gate: unapproved -> project hooks skipped + warning.""" + from core.harness.hooks.discovery import discover_hooks + + ws = tmp_path / "ws" + home = tmp_path / "home" # empty user tree keeps discovery deterministic + home.mkdir() + _write(ws / ".deepcode" / "hooks.json", _DECLARATION_V1) + store = HookTrustStore(tmp_path / "hook_trust.json") + + # No store: project hooks discovered as usual (feature dormant). + baseline = discover_hooks(str(ws), home=str(home)) + assert any(h.source == "project" for h in baseline.handlers) + + # Store present but unapproved: project sources skipped + warning. + gated = discover_hooks(str(ws), home=str(home), trust_store=store) + assert not any(h.source == "project" for h in gated.handlers) + assert any("not approved" in warning for warning in gated.warnings) + + # Approve the CURRENT digest: project hooks are discovered again. + store.approve(str(ws), declaration_digest(str(ws), home=str(home))) + approved = discover_hooks(str(ws), home=str(home), trust_store=store) + assert any(h.source == "project" for h in approved.handlers) + assert not any("not approved" in warning for warning in approved.warnings)