feat(hooks): DPLAN-0244 hooks.json trust model hardening — Layer A engine gates + Layer B registry/CLI

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YEAyLFCuo4uD934fwFxocz
This commit is contained in:
AIOSAI
2026-07-15 18:23:02 -07:00
co-authored by Claude Fable 5
parent 807924241f
commit a8b1ce6658
10 changed files with 1048 additions and 16 deletions
-5
View File
@@ -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"
}
},
+20
View File
@@ -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
@@ -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))
+78
View File
@@ -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 <path>[/green] [dim]# Enroll a project (requires .aipass/hooks.json)[/dim]")
console.print(" [green]aipass revoke <path>[/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
+249
View File
@@ -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 <path> 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 <path> 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 <path> 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 <path> 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 <file> 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
@@ -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
@@ -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()
+28
View File
@@ -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:
+171 -5
View File
@@ -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."""
@@ -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"]