From 9fbfb75f451bf7ae11defe5b9197de9c2eac3a53 Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Mon, 14 Sep 2026 12:50:00 -0400 Subject: [PATCH 1/6] hol-guard-pretool --- src/hooks/hol_guard_pretool.py | 82 +++++++++++++++++++++++++ tests/test_hol_guard_pretool.py | 103 ++++++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+) create mode 100644 src/hooks/hol_guard_pretool.py create mode 100644 tests/test_hol_guard_pretool.py diff --git a/src/hooks/hol_guard_pretool.py b/src/hooks/hol_guard_pretool.py new file mode 100644 index 000000000..ff2a8d81d --- /dev/null +++ b/src/hooks/hol_guard_pretool.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import json +import subprocess +import sys +from typing import Any + +BLOCK_EXIT = 2 +TIMEOUT_SECONDS = 10.0 + + +def _command_from_input(payload: Any) -> str | 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 + return command + + +def _guard_allows(command: str) -> tuple[bool, str]: + try: + result = subprocess.run( + ["hol-guard", "command", "test", command, "--json"], + capture_output=True, + text=True, + timeout=TIMEOUT_SECONDS, + check=False, + ) + except FileNotFoundError: + return False, "guard_unavailable" + except subprocess.TimeoutExpired: + return False, "guard_timeout" + + 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]: + command = _command_from_input(payload) + if command is None: + return False, "guard_invalid_input" + return _guard_allows(command) + + +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()) diff --git a/tests/test_hol_guard_pretool.py b/tests/test_hol_guard_pretool.py new file mode 100644 index 000000000..b5ea49c9d --- /dev/null +++ b/tests/test_hol_guard_pretool.py @@ -0,0 +1,103 @@ +import json +import subprocess +from types import SimpleNamespace + +from src.hooks import hol_guard_pretool as guard + + +def _input(command: object = "git status") -> dict[str, object]: + return { + "hook_event": "PreToolUse", + "tool_name": "Bash", + "tool_input": {"command": command}, + } + + +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_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"] From 676c96b2859f0f1dd9e566d40d39b1c0f98c8b32 Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:41:03 -0400 Subject: [PATCH 2/6] fix Guard hook failure handling and cwd --- src/hooks/hol_guard_pretool.py | 21 +++++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) diff --git a/src/hooks/hol_guard_pretool.py b/src/hooks/hol_guard_pretool.py index ff2a8d81d..16515287c 100644 --- a/src/hooks/hol_guard_pretool.py +++ b/src/hooks/hol_guard_pretool.py @@ -9,7 +9,7 @@ TIMEOUT_SECONDS = 10.0 -def _command_from_input(payload: Any) -> str | None: +def _command_context(payload: Any) -> tuple[str, str | None] | None: if not isinstance(payload, dict): return None if payload.get("hook_event") != "PreToolUse": @@ -22,10 +22,13 @@ def _command_from_input(payload: Any) -> str | None: command = tool_input.get("command") if not isinstance(command, str) or not command.strip(): return None - return command + 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) -> tuple[bool, str]: +def _guard_allows(command: str, cwd: str | None = None) -> tuple[bool, str]: try: result = subprocess.run( ["hol-guard", "command", "test", command, "--json"], @@ -33,11 +36,16 @@ def _guard_allows(command: str) -> tuple[bool, str]: 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" @@ -58,10 +66,11 @@ def _guard_allows(command: str) -> tuple[bool, str]: def evaluate(payload: Any) -> tuple[bool, str]: - command = _command_from_input(payload) - if command is None: + context = _command_context(payload) + if context is None: return False, "guard_invalid_input" - return _guard_allows(command) + command, cwd = context + return _guard_allows(command, cwd) def main() -> int: From 2ec5e76af38dc1200dfe367f6729b62a802650de Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:41:43 -0400 Subject: [PATCH 3/6] pass effective Bash cwd to pre-tool hooks --- src/services/tool_execution/tool_hooks.py | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/src/services/tool_execution/tool_hooks.py b/src/services/tool_execution/tool_hooks.py index aeb007c7c..7ca6b8bfa 100644 --- a/src/services/tool_execution/tool_hooks.py +++ b/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"): @@ -177,7 +196,7 @@ async def run_post_tool_use_hooks( yield { "message": create_attachment_message({ "type": "hook_blocking_error", - "hook_name": f"PostToolUse:{tool.name}", + "hook_name": f"PostToolUseFailure:{tool.name}", "tool_use_id": tool_use_id, "hook_event": "PostToolUse", "blocking_error": result["blocking_error"], From 335b0a9c514e72256d1075b60d75884c78041404 Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:42:17 -0400 Subject: [PATCH 4/6] test Guard hook fail-closed paths and cwd --- tests/test_hol_guard_pretool.py | 123 +++++++++++++++++++++++++++++++- 1 file changed, 121 insertions(+), 2 deletions(-) diff --git a/tests/test_hol_guard_pretool.py b/tests/test_hol_guard_pretool.py index b5ea49c9d..83dc26472 100644 --- a/tests/test_hol_guard_pretool.py +++ b/tests/test_hol_guard_pretool.py @@ -1,15 +1,28 @@ 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") -> dict[str, object]: +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": {"command": command}, + "tool_input": tool_input, } @@ -84,6 +97,22 @@ def raise_missing(*args, **kwargs): 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") @@ -101,3 +130,93 @@ def fake_run(argv, **kwargs): 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 From 675494dc12cfad4f50fc320e2d1c8c433baaf67d Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:42:58 -0400 Subject: [PATCH 5/6] fix post-tool hook label --- src/services/tool_execution/tool_hooks.py | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/src/services/tool_execution/tool_hooks.py b/src/services/tool_execution/tool_hooks.py index 7ca6b8bfa..7029df915 100644 --- a/src/services/tool_execution/tool_hooks.py +++ b/src/services/tool_execution/tool_hooks.py @@ -149,9 +149,6 @@ async def run_pre_tool_use_hooks( if result.get("message"): yield {"type": "message", "message": {"message": result["message"]}} - # ``abort_controller`` is non-optional on ``ToolContext``; - # the truthiness guard used to paper over the field-is-None - # hazard class. if tool_use_context.abort_controller.signal.aborted: yield { "type": "message", @@ -196,7 +193,7 @@ async def run_post_tool_use_hooks( yield { "message": create_attachment_message({ "type": "hook_blocking_error", - "hook_name": f"PostToolUseFailure:{tool.name}", + "hook_name": f"PostToolUse:{tool.name}", "tool_use_id": tool_use_id, "hook_event": "PostToolUse", "blocking_error": result["blocking_error"], @@ -325,11 +322,6 @@ async def resolve_hook_permission_decision( "behavior": "deny", "message": f"Permission handler failed for {tool.name}", } - # FAIL CLOSED. TS cannot express a missing handler (canUseTool is a - # required field, query.ts:191), and the production lane's - # handlerless ask path denies the same way (handler.py:42-51) — an - # allow here would make every tool call permitted the moment a - # future caller forgets to wire can_use_tool. return { "behavior": "deny", "message": ( @@ -426,18 +418,10 @@ async def resolve_hook_permission_decision( return decision if hasattr(decision, "behavior"): return {"behavior": decision.behavior, "message": getattr(decision, "message", None)} - # critic M1 — unrecognized decision shape → fail CLOSED (deny), - # matching the no-hook branch's philosophy (:314-319). This - # branch used to fall through to allow. logger.debug("can_use_tool returned unrecognized shape in ask path") except Exception as e: logger.debug("can_use_tool error in ask path: %s", e) - # critic M1 — the hook-'ask' branch fails CLOSED on any adapter - # exception, a missing adapter, or an unrecognized shape. Previously - # this returned {"behavior": "allow"} — a fail-OPEN asymmetric with the - # no-hook branch (which denies on all three). TS also fails closed here - # (a throwing canUseTool propagates and aborts the tool call). return { "behavior": "deny", "message": f"Permission resolution failed for {tool.name}", From 5483dcdbde9f65a0d996c9326fed55311ad4823e Mon Sep 17 00:00:00 2001 From: Michael Kantor <6068672+kantorcodes@users.noreply.github.com> Date: Tue, 15 Sep 2026 14:44:09 -0400 Subject: [PATCH 6/6] keep Guard cwd patch focused --- src/services/tool_execution/tool_hooks.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/services/tool_execution/tool_hooks.py b/src/services/tool_execution/tool_hooks.py index 7029df915..d89bd91dd 100644 --- a/src/services/tool_execution/tool_hooks.py +++ b/src/services/tool_execution/tool_hooks.py @@ -149,6 +149,9 @@ async def run_pre_tool_use_hooks( if result.get("message"): yield {"type": "message", "message": {"message": result["message"]}} + # ``abort_controller`` is non-optional on ``ToolContext``; + # the truthiness guard used to paper over the field-is-None + # hazard class. if tool_use_context.abort_controller.signal.aborted: yield { "type": "message", @@ -322,6 +325,11 @@ async def resolve_hook_permission_decision( "behavior": "deny", "message": f"Permission handler failed for {tool.name}", } + # FAIL CLOSED. TS cannot express a missing handler (canUseTool is a + # required field, query.ts:191), and the production lane's + # handlerless ask path denies the same way (handler.py:42-51) — an + # allow here would make every tool call permitted the moment a + # future caller forgets to wire can_use_tool. return { "behavior": "deny", "message": ( @@ -418,10 +426,18 @@ async def resolve_hook_permission_decision( return decision if hasattr(decision, "behavior"): return {"behavior": decision.behavior, "message": getattr(decision, "message", None)} + # critic M1 — unrecognized decision shape → fail CLOSED (deny), + # matching the no-hook branch's philosophy (:314-319). This + # branch used to fall through to allow. logger.debug("can_use_tool returned unrecognized shape in ask path") except Exception as e: logger.debug("can_use_tool error in ask path: %s", e) + # critic M1 — the hook-'ask' branch fails CLOSED on any adapter + # exception, a missing adapter, or an unrecognized shape. Previously + # this returned {"behavior": "allow"} — a fail-OPEN asymmetric with the + # no-hook branch (which denies on all three). TS also fails closed here + # (a throwing canUseTool propagates and aborts the tool call). return { "behavior": "deny", "message": f"Permission resolution failed for {tool.name}",