hol-guard-pretool

src/hooks/hol_guard_pretool.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from typing import Any
+
+BLOCK_EXIT = 2
+TIMEOUT_SECONDS = 10.0
+
+
+def _command_context(payload: Any) -> tuple[str, str | None] | None:
+    if not isinstance(payload, dict):
+        return None
+    if payload.get("hook_event") != "PreToolUse":
+        return None
+    if payload.get("tool_name") != "Bash":
+        return None
+    tool_input = payload.get("tool_input")
+    if not isinstance(tool_input, dict):
+        return None
+    command = tool_input.get("command")
+    if not isinstance(command, str) or not command.strip():
+        return None
+    cwd = tool_input.get("cwd")
+    if cwd is not None and (not isinstance(cwd, str) or not cwd.strip()):
+        return None
+    return command, cwd
+
+
+def _guard_allows(command: str, cwd: str | None = None) -> tuple[bool, str]:
+    try:
+        result = subprocess.run(
+            ["hol-guard", "command", "test", command, "--json"],
+            capture_output=True,
+            text=True,
+            timeout=TIMEOUT_SECONDS,
+            check=False,
+            cwd=cwd,
+        )
+    except FileNotFoundError:
+        return False, "guard_unavailable"
+    except subprocess.TimeoutExpired:
+        return False, "guard_timeout"
+    except OSError:
+        return False, "guard_error"
+    except UnicodeError:
+        return False, "guard_invalid_output"
+
+    if result.returncode != 0:
+        return False, "guard_error"
+
+    try:
+        payload = json.loads(result.stdout)
+    except (json.JSONDecodeError, TypeError):
+        return False, "guard_invalid_output"
+
+    if not isinstance(payload, dict):
+        return False, "guard_invalid_output"
+    classification = payload.get("classification")
+    if not isinstance(classification, dict):
+        return False, "guard_invalid_output"
+    if classification.get("explicitly_benign") is True and payload.get("minimum_action") == "allow":
+        return True, "guard_allow"
+    return False, "guard_block"
+
+
+def evaluate(payload: Any) -> tuple[bool, str]:
+    context = _command_context(payload)
+    if context is None:
+        return False, "guard_invalid_input"
+    command, cwd = context
+    return _guard_allows(command, cwd)
+
+
+def main() -> int:
+    try:
+        payload = json.load(sys.stdin)
+    except (json.JSONDecodeError, TypeError):
+        print("guard_invalid_input", file=sys.stderr)
+        return BLOCK_EXIT
+
+    allowed, reason = evaluate(payload)
+    if allowed:
+        return 0
+    print(reason, file=sys.stderr)
+    return BLOCK_EXIT
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/services/tool_execution/tool_hooks.py

@@ -36,6 +36,20 @@ class PreToolUseResult:
     stop_reason: str | None = None
 
 
+def _prepare_pre_tool_hook_input(
+    tool_use_context: ToolContext,
+    tool: Tool,
+    processed_input: dict[str, Any],
+) -> dict[str, Any]:
+    """Give Bash pre-tool hooks the directory Bash will actually use."""
+    if tool.name != "Bash" or "cwd" in processed_input:
+        return processed_input
+    effective_cwd = tool_use_context.cwd or tool_use_context.workspace_root
+    if effective_cwd is None:
+        return processed_input
+    return {**processed_input, "cwd": str(effective_cwd)}
+
+
 async def run_pre_tool_use_hooks(
     tool_use_context: ToolContext,
     tool: Tool,
@@ -48,10 +62,15 @@ async def run_pre_tool_use_hooks(
         if not has_hook_for_event("PreToolUse", tool_use_context):
             return
 
+        hook_input = _prepare_pre_tool_hook_input(
+            tool_use_context,
+            tool,
+            processed_input,
+        )
         async for result in execute_pre_tool_hooks(
             tool.name,
             tool_use_id,
-            processed_input,
+            hook_input,
             tool_use_context,
         ):
             if result.get("blocking_error"):

tests/test_hol_guard_pretool.py

@@ -0,0 +1,222 @@
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from src.hooks import hol_guard_pretool as guard
+from src.services.tool_execution.tool_hooks import _prepare_pre_tool_hook_input
+
+
+def _input(
+    command: object = "git status",
+    *,
+    cwd: str | None = None,
+) -> dict[str, object]:
+    tool_input: dict[str, object] = {"command": command}
+    if cwd is not None:
+        tool_input["cwd"] = cwd
+    return {
+        "hook_event": "PreToolUse",
+        "tool_name": "Bash",
+        "tool_input": tool_input,
+    }
+
+
+def _result(payload: object, returncode: int = 0) -> SimpleNamespace:
+    return SimpleNamespace(returncode=returncode, stdout=json.dumps(payload), stderr="")
+
+
+def test_explicit_benign_allow(monkeypatch):
+    monkeypatch.setattr(
+        guard.subprocess,
+        "run",
+        lambda *args, **kwargs: _result(
+            {"classification": {"explicitly_benign": True}, "minimum_action": "allow"}
+        ),
+    )
+    assert guard.evaluate(_input()) == (True, "guard_allow")
+
+
+def test_implicit_allow_blocks(monkeypatch):
+    monkeypatch.setattr(
+        guard.subprocess,
+        "run",
+        lambda *args, **kwargs: _result(
+            {"classification": {"explicitly_benign": False}, "minimum_action": "allow"}
+        ),
+    )
+    assert guard.evaluate(_input()) == (False, "guard_block")
+
+
+def test_review_blocks(monkeypatch):
+    monkeypatch.setattr(
+        guard.subprocess,
+        "run",
+        lambda *args, **kwargs: _result(
+            {"classification": {"explicitly_benign": False}, "minimum_action": "review"}
+        ),
+    )
+    assert guard.evaluate(_input()) == (False, "guard_block")
+
+
+def test_malformed_output_blocks(monkeypatch):
+    monkeypatch.setattr(
+        guard.subprocess,
+        "run",
+        lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout="{", stderr=""),
+    )
+    assert guard.evaluate(_input()) == (False, "guard_invalid_output")
+
+
+def test_nonzero_guard_exit_blocks(monkeypatch):
+    monkeypatch.setattr(
+        guard.subprocess,
+        "run",
+        lambda *args, **kwargs: SimpleNamespace(returncode=1, stdout="", stderr=""),
+    )
+    assert guard.evaluate(_input()) == (False, "guard_error")
+
+
+def test_timeout_blocks(monkeypatch):
+    def raise_timeout(*args, **kwargs):
+        raise subprocess.TimeoutExpired(cmd="hol-guard", timeout=guard.TIMEOUT_SECONDS)
+
+    monkeypatch.setattr(guard.subprocess, "run", raise_timeout)
+    assert guard.evaluate(_input()) == (False, "guard_timeout")
+
+
+def test_missing_guard_blocks(monkeypatch):
+    def raise_missing(*args, **kwargs):
+        raise FileNotFoundError
+
+    monkeypatch.setattr(guard.subprocess, "run", raise_missing)
+    assert guard.evaluate(_input()) == (False, "guard_unavailable")
+
+
+def test_permission_error_blocks(monkeypatch):
+    def raise_permission(*args, **kwargs):
+        raise PermissionError
+
+    monkeypatch.setattr(guard.subprocess, "run", raise_permission)
+    assert guard.evaluate(_input()) == (False, "guard_error")
+
+
+def test_decode_error_blocks(monkeypatch):
+    def raise_decode(*args, **kwargs):
+        raise UnicodeDecodeError("utf-8", b"\xff", 0, 1, "invalid start byte")
+
+    monkeypatch.setattr(guard.subprocess, "run", raise_decode)
+    assert guard.evaluate(_input()) == (False, "guard_invalid_output")
+
+
+def test_missing_command_blocks():
+    assert guard.evaluate(_input(None)) == (False, "guard_invalid_input")
+
+
+def test_command_is_passed_as_one_argv_item(monkeypatch):
+    seen = {}
+
+    def fake_run(argv, **kwargs):
+        seen["argv"] = argv
+        return _result(
+            {"classification": {"explicitly_benign": True}, "minimum_action": "allow"}
+        )
+
+    monkeypatch.setattr(guard.subprocess, "run", fake_run)
+    command = "echo $(touch /tmp/guard-argv-test)"
+    assert guard.evaluate(_input(command)) == (True, "guard_allow")
+    assert seen["argv"] == ["hol-guard", "command", "test", command, "--json"]
+
+
+def test_guard_runs_in_explicit_bash_cwd(monkeypatch, tmp_path):
+    seen = {}
+
+    def fake_run(argv, **kwargs):
+        seen["cwd"] = kwargs.get("cwd")
+        return _result(
+            {"classification": {"explicitly_benign": True}, "minimum_action": "allow"}
+        )
+
+    monkeypatch.setattr(guard.subprocess, "run", fake_run)
+    assert guard.evaluate(_input(cwd=str(tmp_path))) == (True, "guard_allow")
+    assert seen["cwd"] == str(tmp_path)
+
+
+def test_persisted_bash_cwd_is_added_to_hook_input(tmp_path):
+    workspace = tmp_path / "workspace"
+    prior_cwd = workspace / "nested"
+    workspace.mkdir()
+    prior_cwd.mkdir()
+    context = SimpleNamespace(cwd=prior_cwd, workspace_root=workspace)
+    tool = SimpleNamespace(name="Bash")
+
+    prepared = _prepare_pre_tool_hook_input(context, tool, {"command": "git status"})
+
+    assert prepared == {"command": "git status", "cwd": str(prior_cwd)}
+
+
+def test_explicit_bash_cwd_wins_over_persisted_context(tmp_path):
+    workspace = tmp_path / "workspace"
+    persisted = workspace / "persisted"
+    explicit = workspace / "explicit"
+    workspace.mkdir()
+    persisted.mkdir()
+    explicit.mkdir()
+    context = SimpleNamespace(cwd=persisted, workspace_root=workspace)
+    tool = SimpleNamespace(name="Bash")
+    original = {"command": "git status", "cwd": str(explicit)}
+
+    prepared = _prepare_pre_tool_hook_input(context, tool, original)
+
+    assert prepared is original
+    assert prepared["cwd"] == str(explicit)
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX executable permission semantics")
+def test_non_executable_guard_exits_with_blocking_code(tmp_path):
+    fake_guard = tmp_path / "hol-guard"
+    fake_guard.write_text("#!/bin/sh\nexit 0\n")
+    fake_guard.chmod(0o644)
+    env = os.environ.copy()
+    env["PATH"] = f"{tmp_path}{os.pathsep}{env.get('PATH', '')}"
+
+    result = subprocess.run(
+        [sys.executable, str(Path(guard.__file__))],
+        input=json.dumps(_input()),
+        capture_output=True,
+        text=True,
+        env=env,
+        check=False,
+    )
+
+    assert result.returncode == guard.BLOCK_EXIT
+    assert "guard_error" in result.stderr
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX executable script fixture")
+def test_invalid_utf8_guard_output_exits_with_blocking_code(tmp_path):
+    fake_guard = tmp_path / "hol-guard"
+    fake_guard.write_text(
+        "#!/usr/bin/env python3\n"
+        "import os\n"
+        "os.write(1, b'\\xff')\n"
+    )
+    fake_guard.chmod(0o755)
+    env = os.environ.copy()
+    env["PATH"] = f"{tmp_path}{os.pathsep}{env.get('PATH', '')}"
+
+    result = subprocess.run(
+        [sys.executable, str(Path(guard.__file__))],
+        input=json.dumps(_input()),
+        capture_output=True,
+        text=True,
+        env=env,
+        check=False,
+    )
+
+    assert result.returncode == guard.BLOCK_EXIT
+    assert "guard_invalid_output" in result.stderr