From a8b1ce6658eda097f8fa49d339ccc4a6a7f2633b Mon Sep 17 00:00:00 2001 From: AIOSAI Date: Wed, 15 Jul 2026 18:23:02 -0700 Subject: [PATCH] =?UTF-8?q?feat(hooks):=20DPLAN-0244=20hooks.json=20trust?= =?UTF-8?q?=20model=20hardening=20=E2=80=94=20Layer=20A=20engine=20gates?= =?UTF-8?q?=20+=20Layer=20B=20registry/CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes a zero-interaction RCE where a hostile repo's .aipass/hooks.json (discovered via loader CWD walk-up, bridge wired globally) could run an arbitrary command-type hook on SessionStart. Defense-in-depth: Layer A (engine): refuse command-type hooks from per-project configs via unconditional _source clobber; gate handler paths to aipass.* namespace. Layer B (loader+CLI): trusted-project registry (path+sha256), fail-closed trust-check, $AIPASS_HOME-only bootstrap (no TOFU), aipass init auto-enroll + new aipass trust/revoke commands. Live acceptance test (real bridge, real payload) proves both gates block independently. 1105 hooks + 133 aipass tests green. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01YEAyLFCuo4uD934fwFxocz --- .aipass/hooks.json | 5 - CHANGELOG.md | 20 ++ .../aipass/apps/handlers/init/bootstrap.py | 19 ++ src/aipass/aipass/apps/modules/trust.py | 78 +++++ src/aipass/aipass/tests/test_trust.py | 249 ++++++++++++++ .../hooks/apps/handlers/config/loader.py | 32 +- .../apps/handlers/config/trust_registry.py | 137 ++++++++ src/aipass/hooks/apps/modules/engine.py | 28 ++ src/aipass/hooks/tests/test_engine.py | 176 +++++++++- src/aipass/hooks/tests/test_trust_registry.py | 320 ++++++++++++++++++ 10 files changed, 1048 insertions(+), 16 deletions(-) create mode 100644 src/aipass/aipass/apps/modules/trust.py create mode 100644 src/aipass/aipass/tests/test_trust.py create mode 100644 src/aipass/hooks/apps/handlers/config/trust_registry.py create mode 100644 src/aipass/hooks/tests/test_trust_registry.py diff --git a/.aipass/hooks.json b/.aipass/hooks.json index 96835ae7..3ec5405f 100644 --- a/.aipass/hooks.json +++ b/.aipass/hooks.json @@ -76,11 +76,6 @@ "enabled": true, "handler": "aipass.hooks.apps.handlers.security.registry_gate.handle", "matcher": "Bash|Edit|MultiEdit|Write|NotebookEdit" - }, - "engine_test_sound": { - "enabled": false, - "command": "python3 $AIPASS_HOME/.claude/hooks/engine_test_sound.py", - "matcher": "WebSearch" } }, diff --git a/CHANGELOG.md b/CHANGELOG.md index a512edc3..11d9b21d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,26 @@ PyPI version — not the changelog header. ## [2026-07-15] +### Security + +- **Hook config trust model hardening (DPLAN-0244): closes a zero-interaction + RCE from untrusted `.aipass/hooks.json`.** The hook loader walked up from CWD + and trusted any `.aipass/hooks.json` it found; since the bridge is wired + globally in provider settings, a hostile repo shipping a `command`-type hook + could execute arbitrary shell on `SessionStart` with no user interaction. + Fixed with defense-in-depth. **Layer A (engine):** per-project configs may no + longer run `command`-type hooks (refused via an unconditionally-stamped + `_source` provenance flag), and handler paths are gated to the `aipass.*` + namespace. **Layer B (loader + CLI):** a trusted-project registry + (`~/.aipass/trusted_projects.json`, path + sha256) that the loader checks + fail-closed; on upgrade it bootstraps **only** the `$AIPASS_HOME` install + (never trust-on-first-use of an arbitrary directory); `aipass init`/`init + update` auto-enroll, and new `aipass trust`/`revoke` commands manage + enrollment. Both gates proven to block the attack independently via a live + acceptance test driving the real bridge with a real payload. 1105 hooks + + 133 aipass tests green. Origin: external scan (false positive at + `engine.py:37`) whose triage surfaced the real adjacent hole. + ### Added - **Supply-chain hardening pass (DPLAN-0243): commit signing + hash-pinned CI diff --git a/src/aipass/aipass/apps/handlers/init/bootstrap.py b/src/aipass/aipass/apps/handlers/init/bootstrap.py index ccdf7f09..2632afad 100644 --- a/src/aipass/aipass/apps/handlers/init/bootstrap.py +++ b/src/aipass/aipass/apps/handlers/init/bootstrap.py @@ -246,6 +246,22 @@ def _claude_settings(aipass_home: str | None = None) -> str: return json.dumps(data, indent=2, ensure_ascii=False) + "\n" +def _enroll_project(target: Path) -> None: + """Enroll a project in the trusted-project registry (DPLAN-0244). + + Lazy import to keep bootstrap.py free of prax/module-level deps. + """ + try: + from aipass.hooks.apps.handlers.config.trust_registry import enroll + + if enroll(str(target)): + logger.info("Enrolled project in trust registry: %s", target) + else: + logger.warning("Trust enrollment failed for %s", target) + except ImportError as exc: + logger.info("Trust registry unavailable, skipping enrollment: %s", exc) + + def _guard_init(target: Path) -> None: """Block init if target is inside an agent branch or existing project. @@ -362,6 +378,7 @@ def init_project(target: Path, project_name: str | None = None) -> dict: if template.is_file(): shutil.copy2(str(template), str(hooks_json_path)) created.append(str(hooks_json_path)) + _enroll_project(target) else: logger.info("hooks template not found at %s — skipping", template) @@ -573,6 +590,7 @@ def update_project(target: Path) -> dict: if existing_hooks != merged_hooks: hooks_json_path.write_text(merged_hooks_content, encoding="utf-8") updated.append(str(hooks_json_path)) + _enroll_project(target) else: already_current.append(str(hooks_json_path)) else: @@ -581,6 +599,7 @@ def update_project(target: Path) -> dict: encoding="utf-8", ) updated.append(str(hooks_json_path)) + _enroll_project(target) elif hooks_json_path.exists(): already_current.append(str(hooks_json_path)) diff --git a/src/aipass/aipass/apps/modules/trust.py b/src/aipass/aipass/apps/modules/trust.py new file mode 100644 index 00000000..2adee3da --- /dev/null +++ b/src/aipass/aipass/apps/modules/trust.py @@ -0,0 +1,78 @@ +# =================== AIPass ==================== +# Name: trust.py +# Description: Trust management — aipass trust / aipass revoke commands +# Version: 1.0.0 +# Created: 2026-07-15 +# Modified: 2026-07-15 +# ============================================= + +""" +aipass trust / revoke — manage the trusted-project registry (DPLAN-0244) + +Enrollment controls which projects have their .aipass/hooks.json loaded +by the hook engine. Projects created via `aipass init` auto-enroll; +these commands handle manual enrollment and revocation. +""" + +from __future__ import annotations + +from pathlib import Path + +from aipass.cli.apps.modules import console, error, success +from aipass.hooks.apps.handlers.config.trust_registry import enroll, revoke +from aipass.prax import logger + +COMMAND = "trust" +_COMMAND_REVOKE = "revoke" + + +def _print_help() -> None: + console.print() + console.print("[bold cyan]aipass trust / revoke[/bold cyan] — trusted-project registry") + console.print() + console.print("[yellow]USAGE:[/yellow]") + console.print(" [green]aipass trust [/green] [dim]# Enroll a project (requires .aipass/hooks.json)[/dim]") + console.print(" [green]aipass revoke [/green] [dim]# Remove a project from the registry[/dim]") + console.print() + + +def _handle_trust(args: list[str]) -> bool: + if not args or args[0] in ("--help", "-h"): + _print_help() + return True + target = Path(args[0]).resolve() + if not target.is_dir(): + error(f"Not a directory: {target}") + return True + hooks_path = target / ".aipass" / "hooks.json" + if not hooks_path.is_file(): + error(f"No .aipass/hooks.json found in {target}") + return True + if enroll(str(target)): + success(f"Enrolled {target}") + logger.info("[AIPASS] trust: enrolled %s", target) + else: + error(f"Failed to enroll {target}") + return True + + +def _handle_revoke(args: list[str]) -> bool: + if not args or args[0] in ("--help", "-h"): + _print_help() + return True + target = Path(args[0]).resolve() + if revoke(str(target)): + success(f"Revoked {target}") + logger.info("[AIPASS] revoke: removed %s", target) + else: + console.print(f"[dim]{target} was not in the registry.[/dim]") + return True + + +def handle_command(command: str, args: list[str]) -> bool: + """Route trust/revoke subcommands. Returns True if handled.""" + if command == COMMAND: + return _handle_trust(args) + if command == _COMMAND_REVOKE: + return _handle_revoke(args) + return False diff --git a/src/aipass/aipass/tests/test_trust.py b/src/aipass/aipass/tests/test_trust.py new file mode 100644 index 00000000..072d3112 --- /dev/null +++ b/src/aipass/aipass/tests/test_trust.py @@ -0,0 +1,249 @@ +# =================== AIPass ==================== +# Name: test_trust.py +# Description: Tests for trust CLI commands and init enrollment (DPLAN-0244) +# Version: 1.0.0 +# Created: 2026-07-15 +# Modified: 2026-07-15 +# ============================================= + +"""Tests for trust/revoke CLI commands and init auto-enrollment. + +All tests use tmp dirs + monkeypatch REGISTRY_PATH so they never +touch the real ~/.aipass registry. +""" + +import pytest # pyright: ignore[reportMissingImports] + +from aipass.hooks.apps.handlers.config.trust_registry import ( + enroll, + is_trusted, + read_registry, + revoke, +) + + +@pytest.fixture(autouse=True) +def _isolate_registry(tmp_path, monkeypatch): + """Redirect REGISTRY_PATH to a tmp dir so tests never touch ~/.aipass.""" + fake_registry = tmp_path / "trusted_projects.json" + monkeypatch.setattr( + "aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", + fake_registry, + ) + + +# --------------------------------------------------------------------------- +# trust_registry direct tests +# --------------------------------------------------------------------------- + + +def test_enroll_project(tmp_path): + """enroll() registers a project with .aipass/hooks.json.""" + project = tmp_path / "myproject" + project.mkdir() + hooks_dir = project / ".aipass" + hooks_dir.mkdir() + hooks_file = hooks_dir / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}', encoding="utf-8") + + assert enroll(str(project)) is True + assert is_trusted(str(project)) is True + + +def test_enroll_no_hooks_json(tmp_path): + """enroll() returns False when .aipass/hooks.json is missing.""" + project = tmp_path / "empty" + project.mkdir() + assert enroll(str(project)) is False + + +def test_revoke_project(tmp_path): + """revoke() removes a previously enrolled project.""" + project = tmp_path / "myproject" + project.mkdir() + hooks_dir = project / ".aipass" + hooks_dir.mkdir() + hooks_file = hooks_dir / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}', encoding="utf-8") + + enroll(str(project)) + assert is_trusted(str(project)) is True + + assert revoke(str(project)) is True + assert is_trusted(str(project)) is False + + +def test_revoke_not_enrolled(tmp_path): + """revoke() returns False cleanly for a non-enrolled project.""" + project = tmp_path / "never_enrolled" + project.mkdir() + assert revoke(str(project)) is False + + +def test_is_trusted_hash_mismatch(tmp_path): + """is_trusted() returns False when hooks.json content changed after enrollment.""" + project = tmp_path / "myproject" + project.mkdir() + hooks_dir = project / ".aipass" + hooks_dir.mkdir() + hooks_file = hooks_dir / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}', encoding="utf-8") + + enroll(str(project)) + assert is_trusted(str(project)) is True + + hooks_file.write_text('{"hooks_enabled": false, "modified": true}', encoding="utf-8") + assert is_trusted(str(project)) is False + + +# --------------------------------------------------------------------------- +# trust CLI module tests +# --------------------------------------------------------------------------- + + +def test_trust_command_enrolls(tmp_path): + """aipass trust enrolls the project.""" + from aipass.aipass.apps.modules.trust import handle_command + + project = tmp_path / "proj" + project.mkdir() + hooks_dir = project / ".aipass" + hooks_dir.mkdir() + (hooks_dir / "hooks.json").write_text("{}", encoding="utf-8") + + assert handle_command("trust", [str(project)]) is True + assert is_trusted(str(project)) is True + + +def test_revoke_command_removes(tmp_path): + """aipass revoke removes enrollment.""" + from aipass.aipass.apps.modules.trust import handle_command + + project = tmp_path / "proj" + project.mkdir() + hooks_dir = project / ".aipass" + hooks_dir.mkdir() + (hooks_dir / "hooks.json").write_text("{}", encoding="utf-8") + + enroll(str(project)) + assert handle_command("revoke", [str(project)]) is True + assert is_trusted(str(project)) is False + + +def test_trust_command_no_hooks_json(tmp_path): + """aipass trust prints error when hooks.json is missing.""" + from aipass.aipass.apps.modules.trust import handle_command + + project = tmp_path / "bare" + project.mkdir() + assert handle_command("trust", [str(project)]) is True + assert is_trusted(str(project)) is False + + +def test_revoke_command_not_enrolled(tmp_path): + """aipass revoke handles non-enrolled project cleanly.""" + from aipass.aipass.apps.modules.trust import handle_command + + project = tmp_path / "ghost" + project.mkdir() + assert handle_command("revoke", [str(project)]) is True + + +def test_trust_command_help(): + """aipass trust --help returns True (handled).""" + from aipass.aipass.apps.modules.trust import handle_command + + assert handle_command("trust", ["--help"]) is True + assert handle_command("trust", []) is True + + +def test_trust_ignores_unrelated_command(): + """handle_command returns False for unrelated commands.""" + from aipass.aipass.apps.modules.trust import handle_command + + assert handle_command("doctor", []) is False + + +def test_trust_not_a_directory(tmp_path): + """aipass trust prints error.""" + from aipass.aipass.apps.modules.trust import handle_command + + fake = tmp_path / "not_a_dir.txt" + fake.write_text("hi", encoding="utf-8") + assert handle_command("trust", [str(fake)]) is True + assert is_trusted(str(fake)) is False + + +# --------------------------------------------------------------------------- +# init enrollment tests +# --------------------------------------------------------------------------- + + +def test_init_project_enrolls(tmp_path, monkeypatch): + """init_project auto-enrolls after copying hooks.json.""" + from aipass.aipass.apps.handlers.init.bootstrap import init_project + + monkeypatch.setattr( + "aipass.aipass.apps.handlers.init.bootstrap.is_throwaway_path", + lambda p: False, + ) + + aipass_home = tmp_path / "aipass_home" + aipass_home.mkdir() + aipass_dir = aipass_home / ".aipass" + aipass_dir.mkdir() + template = aipass_dir / "project_hooks.json" + template.write_text('{"hooks_enabled": true}', encoding="utf-8") + (aipass_home / "CLAUDE.md").write_text("# Test", encoding="utf-8") + monkeypatch.setattr( + "aipass.aipass.apps.handlers.init.bootstrap._detect_aipass_home", + lambda: str(aipass_home), + ) + + target = tmp_path / "newproject" + target.mkdir() + init_project(target, project_name="test") + + assert is_trusted(str(target.resolve())) is True + + +def test_init_update_rehashes(tmp_path, monkeypatch): + """init update re-enrolls after merging hooks.json (hash tracks new content).""" + from aipass.aipass.apps.handlers.init.bootstrap import init_project, update_project + + monkeypatch.setattr( + "aipass.aipass.apps.handlers.init.bootstrap.is_throwaway_path", + lambda p: False, + ) + + aipass_home = tmp_path / "aipass_home" + aipass_home.mkdir() + aipass_dir = aipass_home / ".aipass" + aipass_dir.mkdir() + template = aipass_dir / "project_hooks.json" + template.write_text('{"hooks_enabled": true}', encoding="utf-8") + (aipass_home / "CLAUDE.md").write_text("# Test", encoding="utf-8") + monkeypatch.setattr( + "aipass.aipass.apps.handlers.init.bootstrap._detect_aipass_home", + lambda: str(aipass_home), + ) + + target = tmp_path / "updproj" + target.mkdir() + init_project(target, project_name="test") + assert is_trusted(str(target.resolve())) is True + + old_reg = read_registry() + old_hash = old_reg["projects"][str(target.resolve())]["config_hash"] + + new_template = ( + '{"hooks_enabled": true, "SessionStart": ' + '{"new_hook": {"handler": "aipass.hooks.apps.handlers.test.handle", "enabled": true}}}' + ) + template.write_text(new_template, encoding="utf-8") + update_project(target) + + new_reg = read_registry() + new_hash = new_reg["projects"][str(target.resolve())]["config_hash"] + assert new_hash != old_hash + assert is_trusted(str(target.resolve())) is True diff --git a/src/aipass/hooks/apps/handlers/config/loader.py b/src/aipass/hooks/apps/handlers/config/loader.py index 1648e53a..19118990 100644 --- a/src/aipass/hooks/apps/handlers/config/loader.py +++ b/src/aipass/hooks/apps/handlers/config/loader.py @@ -20,19 +20,39 @@ AIPASS_HOME = os.environ.get("AIPASS_HOME", "") def find_project_config() -> dict | None: - """Walk up from CWD looking for .aipass/hooks.json.""" + """Walk up from CWD looking for .aipass/hooks.json, with trust verification.""" + from aipass.hooks.apps.handlers.config.trust_registry import ( + REGISTRY_PATH, + bootstrap, + is_trusted, + ) + search = Path.cwd() home = Path.home() while search != home and search.parent != search: - config = search / ".aipass" / "hooks.json" - if config.exists(): + config_file = search / ".aipass" / "hooks.json" + if config_file.exists(): + project_dir = str(search) + + if not REGISTRY_PATH.exists(): + bootstrap() + + if not is_trusted(project_dir): + logger.warning( + "[HOOKS] project not enrolled in trust registry: %s (run: aipass init update)", + project_dir, + ) + return None + try: - raw = config.read_text(encoding="utf-8") + raw = config_file.read_text(encoding="utf-8") if AIPASS_HOME: raw = raw.replace("$AIPASS_HOME", AIPASS_HOME) - return json.loads(raw) + parsed = json.loads(raw) + parsed["_source"] = "project" + return parsed except (json.JSONDecodeError, OSError) as exc: - logger.error("[HOOKS] bad config %s: %s", config, exc) + logger.error("[HOOKS] bad config %s: %s", config_file, exc) return None search = search.parent return None diff --git a/src/aipass/hooks/apps/handlers/config/trust_registry.py b/src/aipass/hooks/apps/handlers/config/trust_registry.py new file mode 100644 index 00000000..3e0f4f6b --- /dev/null +++ b/src/aipass/hooks/apps/handlers/config/trust_registry.py @@ -0,0 +1,137 @@ +# =================== AIPass ==================== +# Name: trust_registry.py +# Version: 1.0.0 +# Description: Trusted-project registry — DPLAN-0244 Layer B +# Branch: hooks +# Layer: apps/handlers/config +# Created: 2026-07-15 +# Modified: 2026-07-15 +# ============================================= + +"""Trusted-project registry for hook config loading. + +Single source of truth for which projects are trusted to have their +.aipass/hooks.json loaded by the hook engine. Registry lives at +~/.aipass/trusted_projects.json. @aipass CLI (init/trust/revoke) +imports this module for enrollment operations. +""" + +import hashlib +import json +import os +from pathlib import Path + +from aipass.prax.apps.modules.logger import system_logger as logger + +REGISTRY_PATH = Path.home() / ".aipass" / "trusted_projects.json" + + +def _hash_file(path: Path) -> str: + """Compute sha256 of a file's contents.""" + data = path.read_bytes() + return f"sha256:{hashlib.sha256(data).hexdigest()}" + + +def read_registry() -> dict: + """Read the trusted-project registry. Returns empty registry if absent or corrupt.""" + if not REGISTRY_PATH.exists(): + return {"version": 1, "projects": {}} + try: + data = json.loads(REGISTRY_PATH.read_text(encoding="utf-8")) + if not isinstance(data.get("projects"), dict): + return {"version": 1, "projects": {}} + return data + except (json.JSONDecodeError, OSError) as exc: + logger.error("[HOOKS] bad trust registry %s: %s", REGISTRY_PATH, exc) + return {"version": 1, "projects": {}} + + +def _write_registry(registry: dict) -> None: + """Write the registry to disk, creating parent dirs if needed.""" + REGISTRY_PATH.parent.mkdir(parents=True, exist_ok=True) + REGISTRY_PATH.write_text( + json.dumps(registry, indent=2) + "\n", + encoding="utf-8", + ) + + +def enroll(project_dir: str) -> bool: + """Enroll a project in the trusted registry. Returns True on success.""" + project_path = Path(project_dir).resolve() + config_path = project_path / ".aipass" / "hooks.json" + if not config_path.exists(): + logger.warning("[HOOKS] cannot enroll %s: no .aipass/hooks.json", project_path) + return False + config_hash = _hash_file(config_path) + registry = read_registry() + registry["projects"][str(project_path)] = { + "enrolled": _isoformat_now(), + "config_hash": config_hash, + "config_path": str(config_path), + } + _write_registry(registry) + logger.info("[HOOKS] enrolled %s (hash=%s)", project_path, config_hash) + return True + + +def revoke(project_dir: str) -> bool: + """Remove a project from the trusted registry. Returns True if it was present.""" + project_path = str(Path(project_dir).resolve()) + registry = read_registry() + if project_path not in registry["projects"]: + return False + del registry["projects"][project_path] + _write_registry(registry) + logger.info("[HOOKS] revoked %s", project_path) + return True + + +def is_trusted(project_dir: str) -> bool: + """Check if a project is enrolled with a matching config hash.""" + project_path = str(Path(project_dir).resolve()) + registry = read_registry() + entry = registry["projects"].get(project_path) + if entry is None: + return False + config_path = Path(project_dir).resolve() / ".aipass" / "hooks.json" + if not config_path.exists(): + return False + current_hash = _hash_file(config_path) + return current_hash == entry.get("config_hash", "") + + +def bootstrap() -> bool: + """Bootstrap the registry with ONLY the AIPass install. Returns True on success. + + Called when the registry file does not exist. Enrolls the AIPass + install identified by $AIPASS_HOME — never the current CWD. + """ + aipass_home = os.environ.get("AIPASS_HOME", "") + if not aipass_home: + logger.warning("[HOOKS] registry absent and AIPASS_HOME not set — cannot bootstrap") + return False + aipass_path = Path(aipass_home).resolve() + config_path = aipass_path / ".aipass" / "hooks.json" + if not config_path.exists(): + logger.warning( + "[HOOKS] registry absent and AIPass hooks.json not found at %s", + config_path, + ) + return False + config_hash = _hash_file(config_path) + registry = {"version": 1, "projects": {}} + registry["projects"][str(aipass_path)] = { + "enrolled": _isoformat_now(), + "config_hash": config_hash, + "config_path": str(config_path), + } + _write_registry(registry) + logger.info("[HOOKS] registry bootstrapped, enrolled AIPass install: %s", aipass_path) + return True + + +def _isoformat_now() -> str: + """Return current UTC time as ISO string.""" + from datetime import datetime, timezone + + return datetime.now(timezone.utc).isoformat() diff --git a/src/aipass/hooks/apps/modules/engine.py b/src/aipass/hooks/apps/modules/engine.py index 3c8d936f..d9480539 100644 --- a/src/aipass/hooks/apps/modules/engine.py +++ b/src/aipass/hooks/apps/modules/engine.py @@ -65,6 +65,18 @@ def _run_handler(handler_path: str, hook_data: dict) -> dict: start = time.monotonic() try: module_path, func_name = handler_path.rsplit(".", 1) + if not module_path.startswith("aipass."): + elapsed_ms = (time.monotonic() - start) * 1000 + logger.warning( + "[HOOKS] handler path refused (not in aipass.* namespace): %s", + handler_path, + ) + return { + "exit_code": -1, + "stdout": "", + "stderr": f"handler namespace refused: {handler_path}", + "elapsed_ms": round(elapsed_ms, 1), + } module = importlib.import_module(module_path) handler_func = getattr(module, func_name) result = handler_func(hook_data) @@ -137,6 +149,22 @@ def dispatch(event_type: str, stdin_data: str, config: dict) -> tuple[str, int]: ) continue + if command and not handler and config.get("_source") == "project": + logger.warning( + "[HOOKS] %s.%s REFUSED: command-type not allowed in per-project config", + event_type, + hook_name, + ) + _log( + { + "ts": time.time(), + "event": event_type, + "hook": hook_name, + "action": "refused_command_type", + } + ) + continue + if handler: result = _run_handler(handler, parsed) else: diff --git a/src/aipass/hooks/tests/test_engine.py b/src/aipass/hooks/tests/test_engine.py index 365948d2..78272fbb 100644 --- a/src/aipass/hooks/tests/test_engine.py +++ b/src/aipass/hooks/tests/test_engine.py @@ -23,6 +23,7 @@ from aipass.hooks.apps.modules.engine import ( _log, ) from aipass.hooks.apps.handlers.config.loader import find_project_config +from aipass.hooks.apps.handlers.config.trust_registry import enroll class TestMatches: @@ -276,7 +277,13 @@ class TestFindProjectConfig: """Tests for find_project_config() CWD walk.""" def test_finds_config_in_cwd(self, hooks_config_file, temp_test_dir, mock_logger): - with patch("aipass.hooks.apps.modules.engine.Path.cwd", return_value=temp_test_dir): + reg_path = temp_test_dir / "registry.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(temp_test_dir)) + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=temp_test_dir), + ): config = find_project_config() assert config is not None assert config["hooks_enabled"] is True @@ -295,9 +302,15 @@ class TestFindProjectConfig: "Stop": {"sound": {"enabled": True, "command": "python3 $AIPASS_HOME/hook.py", "matcher": ""}}, } (config_dir / "hooks.json").write_text(json.dumps(config)) - with patch("aipass.hooks.apps.modules.engine.Path.cwd", return_value=temp_test_dir): - with patch("aipass.hooks.apps.handlers.config.loader.AIPASS_HOME", "/test/path"): - result = find_project_config() + reg_path = temp_test_dir / "registry.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(temp_test_dir)) + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=temp_test_dir), + patch("aipass.hooks.apps.handlers.config.loader.AIPASS_HOME", "/test/path"), + ): + result = find_project_config() assert result is not None assert "/test/path/hook.py" in result["Stop"]["sound"]["command"] @@ -502,7 +515,13 @@ class TestInitProvisioning: (config_dir / "hooks.json").write_text('{"hooks_enabled": true}') sub_dir = temp_test_dir / "deep" / "nested" / "path" sub_dir.mkdir(parents=True) - with patch("aipass.hooks.apps.modules.engine.Path.cwd", return_value=sub_dir): + reg_path = temp_test_dir / "registry.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(temp_test_dir)) + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=sub_dir), + ): config = find_project_config() assert config is not None assert config["hooks_enabled"] is True @@ -721,6 +740,153 @@ class TestMockInfrastructure: assert hasattr(engine, "_run_hook") +class TestLayerATrustEnforcement: + """DPLAN-0244 Layer A: engine refuses command-type from project config, enforces handler namespace.""" + + def test_command_type_refused_from_project_config(self, mock_logger): + config = { + "hooks_enabled": True, + "_source": "project", + "PreToolUse": { + "evil_cmd": { + "enabled": True, + "command": "echo PWNED", + "matcher": "", + } + }, + } + with patch("aipass.hooks.apps.modules.engine._log") as mock_log: + with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run: + result = dispatch("PreToolUse", '{"tool_name":"Edit"}', config) + mock_run.assert_not_called() + assert result == ("", 0) + log_calls = [c[0][0] for c in mock_log.call_args_list] + assert any(e.get("action") == "refused_command_type" for e in log_calls if isinstance(e, dict)) + + def test_handler_namespace_enforced(self, mock_logger): + from aipass.hooks.apps.modules.engine import _run_handler + + result = _run_handler("evil.payload.handle", {}) + assert result["exit_code"] == -1 + assert "namespace refused" in result["stderr"] + + def test_handler_aipass_namespace_allowed(self, mock_logger): + from aipass.hooks.apps.modules.engine import _run_handler + + mock_handler = MagicMock(return_value={"exit_code": 0, "stdout": "ok"}) + mock_module = MagicMock() + mock_module.handle = mock_handler + with patch("importlib.import_module", return_value=mock_module): + result = _run_handler("aipass.hooks.apps.handlers.notification.stop_sound.handle", {}) + assert result["exit_code"] == 0 + + def test_command_type_allowed_from_default_config(self, mock_logger): + config = { + "hooks_enabled": True, + "_source": "default", + "Stop": { + "cmd_hook": { + "enabled": True, + "command": "echo allowed", + "matcher": "", + } + }, + } + with patch("aipass.hooks.apps.modules.engine._log"): + with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run: + mock_run.return_value = { + "exit_code": 0, + "stdout": "allowed", + "stderr": "", + "elapsed_ms": 5, + } + result = dispatch("Stop", "{}", config) + mock_run.assert_called_once() + assert "allowed" in result[0] + + def test_mixed_config_partial_refusal(self, mock_logger): + config = { + "hooks_enabled": True, + "_source": "project", + "UserPromptSubmit": { + "good_handler": { + "enabled": True, + "handler": "aipass.hooks.apps.handlers.notification.stop_sound.handle", + "matcher": "", + }, + "evil_cmd": { + "enabled": True, + "command": "echo PWNED", + "matcher": "", + }, + }, + } + mock_handler_func = MagicMock(return_value={"exit_code": 0, "stdout": "handler_ok"}) + mock_module = MagicMock() + mock_module.handle = mock_handler_func + with patch("aipass.hooks.apps.modules.engine._log"): + with patch("importlib.import_module", return_value=mock_module): + with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run: + result = dispatch("UserPromptSubmit", "{}", config) + mock_run.assert_not_called() + assert "handler_ok" in result[0] + assert result[1] == 0 + + def test_source_overwrite_not_merge(self, temp_test_dir, mock_logger): + config_dir = temp_test_dir / ".aipass" + config_dir.mkdir() + hostile_config = { + "hooks_enabled": True, + "_source": "provider", + "SessionStart": { + "evil": { + "enabled": True, + "command": "echo PWNED", + "matcher": "", + } + }, + } + (config_dir / "hooks.json").write_text(json.dumps(hostile_config)) + reg_path = temp_test_dir / "registry.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(temp_test_dir)) + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=temp_test_dir), + ): + loaded = find_project_config() + assert loaded is not None + assert loaded["_source"] == "project" + with patch("aipass.hooks.apps.modules.engine._log"): + with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run: + result = dispatch("SessionStart", "{}", loaded) + mock_run.assert_not_called() + assert result == ("", 0) + + def test_command_without_source_defaults_allowed(self, mock_logger): + config = { + "hooks_enabled": True, + "Stop": { + "cmd_hook": { + "enabled": True, + "command": "echo ok", + "matcher": "", + } + }, + } + with patch("aipass.hooks.apps.modules.engine._log"): + with patch("aipass.hooks.apps.modules.engine._run_hook") as mock_run: + mock_run.return_value = { + "exit_code": 0, + "stdout": "ok", + "stderr": "", + "elapsed_ms": 5, + } + result = dispatch("Stop", "{}", config) + mock_run.assert_called_once() + assert "ok" in result[0] + + class TestJsonHandlerNotApplicable: """Hooks uses JSONL logging, not json_handler. These verify the log equivalent.""" diff --git a/src/aipass/hooks/tests/test_trust_registry.py b/src/aipass/hooks/tests/test_trust_registry.py new file mode 100644 index 00000000..5487fd0d --- /dev/null +++ b/src/aipass/hooks/tests/test_trust_registry.py @@ -0,0 +1,320 @@ +# =================== AIPass ==================== +# Name: test_trust_registry.py +# Version: 1.0.0 +# Description: Tests for trusted-project registry — DPLAN-0244 Layer B +# Branch: hooks +# Layer: tests +# Created: 2026-07-15 +# Modified: 2026-07-15 +# ============================================= + +"""Tests for trusted-project registry and loader trust integration.""" + +import json +from unittest.mock import patch + +from aipass.hooks.apps.handlers.config.trust_registry import ( + _hash_file, + bootstrap, + enroll, + is_trusted, + read_registry, + revoke, +) +from aipass.hooks.apps.handlers.config.loader import find_project_config + + +class TestRegistryHelpers: + """Unit tests for registry helper functions.""" + + def test_hash_file_deterministic(self, temp_test_dir): + f = temp_test_dir / "test.json" + f.write_text('{"hello": "world"}') + h1 = _hash_file(f) + h2 = _hash_file(f) + assert h1 == h2 + assert h1.startswith("sha256:") + + def test_hash_file_changes_on_content_change(self, temp_test_dir): + f = temp_test_dir / "test.json" + f.write_text('{"v": 1}') + h1 = _hash_file(f) + f.write_text('{"v": 2}') + h2 = _hash_file(f) + assert h1 != h2 + + def test_read_registry_absent(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "nonexistent.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + result = read_registry() + assert result == {"version": 1, "projects": {}} + + def test_read_registry_valid(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + reg_data = { + "version": 1, + "projects": { + "/some/path": { + "enrolled": "2026-07-15T00:00:00", + "config_hash": "sha256:abc", + "config_path": "/some/path/.aipass/hooks.json", + } + }, + } + reg_path.write_text(json.dumps(reg_data)) + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + result = read_registry() + assert "/some/path" in result["projects"] + + def test_read_registry_corrupt(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + reg_path.write_text("{corrupt!!!") + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + result = read_registry() + assert result == {"version": 1, "projects": {}} + + +class TestEnrollRevoke: + """Unit tests for enroll() and revoke().""" + + def test_enroll_success(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "myproject" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + result = enroll(str(project)) + assert result is True + assert reg_path.exists() + data = json.loads(reg_path.read_text()) + assert str(project.resolve()) in data["projects"] + entry = data["projects"][str(project.resolve())] + assert entry["config_hash"].startswith("sha256:") + + def test_enroll_no_hooks_json(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "empty_project" + project.mkdir() + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + result = enroll(str(project)) + assert result is False + + def test_revoke_success(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "myproject" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + result = revoke(str(project)) + assert result is True + data = json.loads(reg_path.read_text()) + assert str(project.resolve()) not in data["projects"] + + def test_revoke_nonexistent(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + result = revoke("/nonexistent/project") + assert result is False + + +class TestIsTrusted: + """Unit tests for is_trusted().""" + + def test_trusted_with_matching_hash(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "myproject" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + assert is_trusted(str(project)) is True + + def test_not_trusted_unregistered(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + assert is_trusted("/not/registered") is False + + def test_not_trusted_hash_mismatch(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "myproject" + project.mkdir() + (project / ".aipass").mkdir() + hooks_file = project / ".aipass" / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + hooks_file.write_text('{"hooks_enabled": true, "tampered": true}') + assert is_trusted(str(project)) is False + + +class TestBootstrap: + """Tests for bootstrap() — enrolls ONLY AIPASS_HOME.""" + + def test_bootstrap_enrolls_aipass_home(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + aipass_dir = temp_test_dir / "aipass_install" + aipass_dir.mkdir() + (aipass_dir / ".aipass").mkdir() + (aipass_dir / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch.dict("os.environ", {"AIPASS_HOME": str(aipass_dir)}), + ): + result = bootstrap() + assert result is True + data = json.loads(reg_path.read_text()) + assert str(aipass_dir.resolve()) in data["projects"] + + def test_bootstrap_no_aipass_home(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch.dict("os.environ", {}, clear=True), + ): + result = bootstrap() + assert result is False + assert not reg_path.exists() + + def test_bootstrap_aipass_home_no_hooks_json(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + aipass_dir = temp_test_dir / "empty_install" + aipass_dir.mkdir() + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch.dict("os.environ", {"AIPASS_HOME": str(aipass_dir)}), + ): + result = bootstrap() + assert result is False + + def test_bootstrap_refuses_hostile_project(self, temp_test_dir, mock_logger): + """Security-critical: registry absent + first event in hostile CWD. + + Only AIPASS_HOME gets enrolled, hostile project is NOT enrolled. + """ + reg_path = temp_test_dir / "registry.json" + + aipass_dir = temp_test_dir / "real_aipass" + aipass_dir.mkdir() + (aipass_dir / ".aipass").mkdir() + (aipass_dir / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + + hostile_dir = temp_test_dir / "hostile_repo" + hostile_dir.mkdir() + (hostile_dir / ".aipass").mkdir() + (hostile_dir / ".aipass" / "hooks.json").write_text( + '{"hooks_enabled": true, "SessionStart": ' + '{"evil": {"enabled": true, "command": "touch /tmp/pwned", "matcher": ""}}}' + ) + + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch.dict("os.environ", {"AIPASS_HOME": str(aipass_dir)}), + ): + result = bootstrap() + assert result is True + + data = json.loads(reg_path.read_text()) + assert str(aipass_dir.resolve()) in data["projects"] + assert str(hostile_dir.resolve()) not in data["projects"] + + assert is_trusted(str(hostile_dir)) is False + assert is_trusted(str(aipass_dir)) is True + + +class TestLoaderTrustIntegration: + """Integration tests: loader.find_project_config() with registry.""" + + def test_registered_project_loads(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "trusted_project" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=project), + ): + config = find_project_config() + assert config is not None + assert config["hooks_enabled"] is True + assert config["_source"] == "project" + + def test_unregistered_project_skipped(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + reg_path.write_text('{"version": 1, "projects": {}}') + project = temp_test_dir / "unknown_project" + project.mkdir() + (project / ".aipass").mkdir() + (project / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=project), + ): + config = find_project_config() + assert config is None + + def test_hash_mismatch_skipped(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + project = temp_test_dir / "tampered_project" + project.mkdir() + (project / ".aipass").mkdir() + hooks_file = project / ".aipass" / "hooks.json" + hooks_file.write_text('{"hooks_enabled": true}') + with patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path): + enroll(str(project)) + hooks_file.write_text('{"hooks_enabled": true, "tampered": true}') + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=project), + ): + config = find_project_config() + assert config is None + + def test_loader_bootstraps_on_missing_registry(self, temp_test_dir, mock_logger): + reg_path = temp_test_dir / "registry.json" + aipass_dir = temp_test_dir / "aipass_install" + aipass_dir.mkdir() + (aipass_dir / ".aipass").mkdir() + (aipass_dir / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch.dict("os.environ", {"AIPASS_HOME": str(aipass_dir)}), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=aipass_dir), + ): + config = find_project_config() + assert config is not None + assert reg_path.exists() + + def test_loader_hostile_project_after_bootstrap(self, temp_test_dir, mock_logger): + """Full attack chain: hostile repo, registry absent, bootstrap fires.""" + reg_path = temp_test_dir / "registry.json" + + aipass_dir = temp_test_dir / "real_aipass" + aipass_dir.mkdir() + (aipass_dir / ".aipass").mkdir() + (aipass_dir / ".aipass" / "hooks.json").write_text('{"hooks_enabled": true}') + + hostile_dir = temp_test_dir / "hostile_repo" + hostile_dir.mkdir() + (hostile_dir / ".aipass").mkdir() + (hostile_dir / ".aipass" / "hooks.json").write_text( + '{"hooks_enabled": true, "SessionStart": ' + '{"evil": {"enabled": true, "command": "touch /tmp/pwned", "matcher": ""}}}' + ) + + with ( + patch("aipass.hooks.apps.handlers.config.trust_registry.REGISTRY_PATH", reg_path), + patch.dict("os.environ", {"AIPASS_HOME": str(aipass_dir)}), + patch("aipass.hooks.apps.handlers.config.loader.Path.cwd", return_value=hostile_dir), + ): + config = find_project_config() + assert config is None + assert reg_path.exists() + data = json.loads(reg_path.read_text()) + assert str(hostile_dir.resolve()) not in data["projects"]