diff --git a/src/brigade/claude_hooks/runtime.py b/src/brigade/claude_hooks/runtime.py index 204c73a3..48762a95 100644 --- a/src/brigade/claude_hooks/runtime.py +++ b/src/brigade/claude_hooks/runtime.py @@ -38,6 +38,7 @@ "node_modules", } _SNAPSHOT_GIT_TIMEOUT_SECONDS = 3 +_UNAVAILABLE_FINGERPRINT = "unavailable" _BASH_WRITE_COMMANDS = { "apply_patch", "cp", @@ -1588,8 +1589,12 @@ def _bash_write_detected( ) -> bool: if not isinstance(baseline, str) or not baseline: return False + if baseline == _UNAVAILABLE_FINGERPRINT: + return True current = repo_worktree_fingerprint(target) - if current is None or current == baseline: + if current is None: + return True + if current == baseline: return False started = localio.parse_iso_datetime(started_at) if started is None: @@ -1757,10 +1762,9 @@ def handle_payload(event: str, payload: dict[str, Any]) -> dict[str, Any] | None tool_input: dict[str, Any] = raw_tool_input if isinstance(raw_tool_input, dict) else {} command = tool_input.get("command") baseline = repo_worktree_fingerprint(target) - if baseline is not None: - state["pending_bash_fingerprint"] = baseline - state["pending_bash_started_at"] = localio.utc_now_iso() - write_session_state(target, session_id, state) + state["pending_bash_fingerprint"] = baseline or _UNAVAILABLE_FINGERPRINT + state["pending_bash_started_at"] = localio.utc_now_iso() + write_session_state(target, session_id, state) if not is_raw_verification(command): return None state["verify_denied_count"] = int(state.get("verify_denied_count") or 0) + 1 diff --git a/tests/test_claude_hooks_runtime.py b/tests/test_claude_hooks_runtime.py index 10fba3c2..c983a202 100644 --- a/tests/test_claude_hooks_runtime.py +++ b/tests/test_claude_hooks_runtime.py @@ -412,7 +412,7 @@ def test_posttooluse_ignores_concurrent_session_write_during_read_only_bash(tmp_ ) -def test_posttooluse_snapshot_fails_open_when_state_cannot_be_inspected(tmp_path: Path, monkeypatch): +def test_posttooluse_snapshot_fails_closed_when_state_cannot_be_inspected(tmp_path: Path, monkeypatch): target = _wired_claude(tmp_path) session_id = "snapshot-unavailable" monkeypatch.setattr(runtime, "repo_worktree_fingerprint", lambda repo: None) @@ -424,11 +424,61 @@ def test_posttooluse_snapshot_fails_open_when_state_cannot_be_inspected(tmp_path tool_input={"command": f"{sys.executable} -c \"print('noop')\""}, ) assert runtime.handle_payload("PreToolUse", pretool) is None - assert runtime.read_session_state(target, session_id).get("pending_bash_fingerprint") is None + assert runtime.read_session_state(target, session_id).get("pending_bash_fingerprint") == "unavailable" succeeded = {**pretool, "hook_event_name": "PostToolUse"} assert runtime.handle_payload("PostToolUse", succeeded) is None - assert runtime.read_session_state(target, session_id)["write_observed"] is False + assert runtime.read_session_state(target, session_id)["write_observed"] is True + blocked = runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False)) + assert blocked["decision"] == "block" + + +def test_posttooluse_fails_closed_when_post_command_snapshot_is_unavailable(tmp_path: Path, monkeypatch): + target = _wired_claude(tmp_path) + session_id = "post-snapshot-unavailable" + fingerprint_calls = 0 + + def sequenced_fingerprint(repo: Path) -> str | None: + nonlocal fingerprint_calls + fingerprint_calls += 1 + return "baseline" if fingerprint_calls < 4 else None + + monkeypatch.setattr(runtime, "repo_worktree_fingerprint", sequenced_fingerprint) + pretool = _payload( + target, + "PreToolUse", + session_id=session_id, + tool_name="Bash", + tool_input={"command": f"{sys.executable} -c \"print('noop')\""}, + ) + + assert runtime.handle_payload("PreToolUse", pretool) is None + assert runtime.read_session_state(target, session_id)["pending_bash_fingerprint"] == "baseline" + assert runtime.handle_payload("PostToolUse", {**pretool, "hook_event_name": "PostToolUse"}) is None + + assert runtime.read_session_state(target, session_id)["write_observed"] is True + blocked = runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False)) + assert blocked["decision"] == "block" + + +@pytest.mark.parametrize("command", ["gh --version", "jq --version", "rg --version"]) +def test_posttooluse_unlisted_read_only_command_does_not_observe_write(tmp_path: Path, command: str): + target = _git_wired_claude(tmp_path) + session_id = f"read-only-{command.split()[0]}" + pretool = _payload( + target, + "PreToolUse", + session_id=session_id, + tool_name="Bash", + tool_input={"command": command}, + ) + + assert runtime.handle_payload("PreToolUse", pretool) is None + assert runtime.handle_payload("PostToolUse", {**pretool, "hook_event_name": "PostToolUse"}) is None + + state = runtime.read_session_state(target, session_id) + assert state["write_observed"] is False + assert "pending_bash_fingerprint" not in state assert ( runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False)) is None ) @@ -840,7 +890,7 @@ def fake_run(repo: Path, *git_args: str): assert runtime.repo_worktree_fingerprint(target) is None -def test_posttooluse_does_not_record_bash_write_when_hash_object_fails_for_untracked(tmp_path: Path, monkeypatch): +def test_posttooluse_fails_closed_when_untracked_state_check_fails(tmp_path: Path, monkeypatch): target = _git_wired_claude(tmp_path) session_id = "hash-object-fail" out_file = target / "new.txt" @@ -864,15 +914,14 @@ def fake_run(repo: Path, *git_args: str): assert runtime.handle_payload("PreToolUse", pretool) is None state = runtime.read_session_state(target, session_id) assert state["write_observed"] is False - assert "pending_bash_fingerprint" not in state + assert state["pending_bash_fingerprint"] == "unavailable" out_file.write_text("after") succeeded = {**pretool, "hook_event_name": "PostToolUse"} assert runtime.handle_payload("PostToolUse", succeeded) is None - assert runtime.read_session_state(target, session_id)["write_observed"] is False - assert ( - runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False)) is None - ) + assert runtime.read_session_state(target, session_id)["write_observed"] is True + blocked = runtime.handle_payload("Stop", _payload(target, "Stop", session_id=session_id, stop_hook_active=False)) + assert blocked["decision"] == "block" def test_posttooluse_records_bash_write_on_dirty_tracked_same_size_rewrite(tmp_path: Path):