diff --git a/engine/hooks/_sdk/runtime.py b/engine/hooks/_sdk/runtime.py index d1ea4351..d4db3c5c 100644 --- a/engine/hooks/_sdk/runtime.py +++ b/engine/hooks/_sdk/runtime.py @@ -25,7 +25,7 @@ def run_hook( event = json.load(sys.stdin) except json.JSONDecodeError as exc: _write_findings_file([]) - print(f"catstack-hook-error {hook}: JSONDecodeError: {exc}", file=sys.stderr) + print(f"catstack-hook-error {hook}: JSONDecodeError: hook payload is not JSON: {exc}", file=sys.stderr) stdout_text, _stderr_text, _exit_code = render( harness, hook_event_name or "", diff --git a/engine/hooks/categorical-scope-guard/README.md b/engine/hooks/categorical-scope-guard/README.md index 3eb5b2dc..2aabdff3 100644 --- a/engine/hooks/categorical-scope-guard/README.md +++ b/engine/hooks/categorical-scope-guard/README.md @@ -91,9 +91,12 @@ Do one of two things: ## Three outcomes, and the fail direction -- **HIT** — exit 2 with the message above. +- **HIT** — finding `categorical-scope-guard.partial-status-filter`. The + registry default is `stop`, so the shared runtime exits 2 with the + message above unless a local mode override lowers it to a warning. - **CLEAN** — exit 0. -- **UNCHECKED** — exit 2. **This hook fails closed.** It blocks, and says +- **UNCHECKED** — finding `categorical-scope-guard.unchecked`. **This hook + fails closed.** With the default `stop` mode it blocks, and says `UNCHECKED`, when: - the transcript path is absent, the file is missing, a line is malformed JSON (a torn final line is tolerated), the window runs past the 64 MB @@ -132,8 +135,10 @@ reason and let the user answer. ## Files -- `detect.py` — command parser, human-turn reader, `decide()`. -- `claude_pretooluse.py` — the entrypoint. +- `detect.py` — command parser, human-turn reader, `decide()`, and + `detect(event)`. +- `claude_pretooluse.py`, `cursor_pretooluse.py`, `codex_pretooluse.py` — + thin runtime entrypoints. - `claude.hook.json`, `install_claude_hook.py` — the settings merge that `install.sh` runs. - `tests/test_hooks.py`, `tests/fixtures/` — the fixtures are real commands diff --git a/engine/hooks/categorical-scope-guard/claude_pretooluse.py b/engine/hooks/categorical-scope-guard/claude_pretooluse.py index 673c734c..a3724044 100644 --- a/engine/hooks/categorical-scope-guard/claude_pretooluse.py +++ b/engine/hooks/categorical-scope-guard/claude_pretooluse.py @@ -1,55 +1,19 @@ #!/usr/bin/env python3 -"""Claude PreToolUse on Bash: block a status-narrowed mutation of a target the -live human turn quantified with all / every / each. Exits 2 with the reason on -stderr for HIT and for UNCHECKED; exits 0 for CLEAN. -""" +"""Claude Code PreToolUse entrypoint for categorical-scope-guard.""" from __future__ import annotations -import json import os import sys -sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) -from detect import CLEAN, decide_payload - -SHELL_LIKE_TOOL_NAMES = ( - "Bash", "bash", "shell", "Shell", "exec", "exec_command", - "run_terminal_cmd", "local_shell", "run_command", "shell_call", -) - - -def _tool_name(payload: dict) -> str: - return str( - payload.get("tool_name") - or payload.get("toolName") - or payload.get("tool") - or payload.get("name") - or "" - ) +from detect import detect # noqa: E402 +from runtime import run_hook # noqa: E402 def main() -> None: - raw = sys.stdin.read() - try: - payload = json.loads(raw) - except (json.JSONDecodeError, OSError) as exc: - sys.stderr.write(f"categorical-scope-guard: hook payload is not JSON, nothing to classify: {exc}\n") - return - if not isinstance(payload, dict) or _tool_name(payload) not in SHELL_LIKE_TOOL_NAMES: - return - try: - verdict = decide_payload(payload) - except Exception as exc: - sys.stderr.write( - f"categorical-scope-guard: UNCHECKED -- the detector failed ({exc!r}) while classifying a " - "status-filtered mutation. Blocked rather than passed; drop the status filter or rephrase the command.\n" - ) - sys.exit(2) - if verdict.outcome == CLEAN: - return - sys.stderr.write(verdict.message + "\n") - sys.exit(2) + run_hook("categorical-scope-guard", "claude", detect, "PreToolUse") if __name__ == "__main__": diff --git a/engine/hooks/categorical-scope-guard/codex_pretooluse.py b/engine/hooks/categorical-scope-guard/codex_pretooluse.py new file mode 100644 index 00000000..e2f2a14d --- /dev/null +++ b/engine/hooks/categorical-scope-guard/codex_pretooluse.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +"""Codex PreToolUse entrypoint for categorical-scope-guard.""" +from __future__ import annotations + +import os +import sys + +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) + +from detect import detect # noqa: E402 +from runtime import run_hook # noqa: E402 + + +def main() -> None: + run_hook("categorical-scope-guard", "codex", detect, "PreToolUse") + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/categorical-scope-guard/cursor_pretooluse.py b/engine/hooks/categorical-scope-guard/cursor_pretooluse.py new file mode 100644 index 00000000..bf6714e8 --- /dev/null +++ b/engine/hooks/categorical-scope-guard/cursor_pretooluse.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +"""Cursor PreToolUse entrypoint for categorical-scope-guard.""" +from __future__ import annotations + +import os +import sys + +sys.path.insert(0, os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "_sdk")) + +from detect import detect # noqa: E402 +from runtime import run_hook # noqa: E402 + + +def main() -> None: + run_hook("categorical-scope-guard", "cursor", detect, "PreToolUse") + + +if __name__ == "__main__": + main() diff --git a/engine/hooks/categorical-scope-guard/detect.py b/engine/hooks/categorical-scope-guard/detect.py index bc0f1912..b37566b4 100644 --- a/engine/hooks/categorical-scope-guard/detect.py +++ b/engine/hooks/categorical-scope-guard/detect.py @@ -14,14 +14,24 @@ from __future__ import annotations import sys +import hashlib import json import os import re from dataclasses import dataclass, field +from pathlib import Path + +SDK_DIR = Path(__file__).resolve().parents[1] / "_sdk" +if str(SDK_DIR) not in sys.path: + sys.path.insert(0, str(SDK_DIR)) + +from finding import Finding # noqa: E402 HIT = "hit" CLEAN = "clean" UNCHECKED = "unchecked" +RULE_PARTIAL_STATUS_FILTER = "categorical-scope-guard.partial-status-filter" +RULE_UNCHECKED = "categorical-scope-guard.unchecked" LIVE_TURNS = 4 MAX_SCAN_BYTES = 64 * 1024 * 1024 @@ -872,3 +882,54 @@ def decide_payload(payload: dict) -> Verdict: return Verdict(CLEAN) path = payload.get("transcript_path") or payload.get("transcriptPath") or "" return decide(command, lambda: read_live_window(path if isinstance(path, str) else "")) + + +SHELL_LIKE_TOOL_NAMES = ( + "Bash", "bash", "shell", "Shell", "exec", "exec_command", + "run_terminal_cmd", "local_shell", "run_command", "shell_call", +) + + +def _tool_name(payload: dict) -> str: + return str( + payload.get("tool_name") + or payload.get("toolName") + or payload.get("tool") + or payload.get("name") + or "" + ) + + +def _command_subject(payload: dict) -> str: + tool_input = payload.get("tool_input") or payload.get("toolInput") or {} + command = tool_input.get("command") if isinstance(tool_input, dict) else "" + if not isinstance(command, str): + command = "" + return "command:" + hashlib.sha256(command.encode("utf-8")).hexdigest()[:16] + + +def _finding(rule_id: str, payload: dict, message: str) -> Finding: + return Finding( + rule_id=rule_id, + subject=_command_subject(payload), + message=message, + evidence=message, + ) + + +def detect(event: dict) -> list[Finding]: + if _tool_name(event) not in SHELL_LIKE_TOOL_NAMES: + return [] + try: + verdict = decide_payload(event) + except Exception as exc: + print(f"catstack-hook-error categorical-scope-guard: {type(exc).__name__}: {exc}", file=sys.stderr) + message = ( + f"categorical-scope-guard: UNCHECKED -- the detector failed ({exc!r}) while classifying a " + "status-filtered mutation. Blocked rather than passed; drop the status filter or rephrase the command." + ) + return [_finding(RULE_UNCHECKED, event, message)] + if verdict.outcome == CLEAN: + return [] + rule_id = RULE_PARTIAL_STATUS_FILTER if verdict.outcome == HIT else RULE_UNCHECKED + return [_finding(rule_id, event, verdict.message)] diff --git a/engine/hooks/categorical-scope-guard/tests/test_hooks_sdk_mode.py b/engine/hooks/categorical-scope-guard/tests/test_hooks_sdk_mode.py new file mode 100644 index 00000000..b61c0c3b --- /dev/null +++ b/engine/hooks/categorical-scope-guard/tests/test_hooks_sdk_mode.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +HERE = os.path.dirname(os.path.abspath(__file__)) +HOOK_DIR = os.path.dirname(HERE) +FIXTURES = os.path.join(HERE, "fixtures") +ENTRYPOINT = os.path.join(HOOK_DIR, "claude_pretooluse.py") + + +def fixture(name: str) -> str: + with open(os.path.join(FIXTURES, name), encoding="utf-8") as handle: + return handle.read() + + +def human(text: str) -> dict[str, object]: + return {"type": "user", "message": {"role": "user", "content": text}} + + +def write_transcript(entries: list[dict[str, object]]) -> str: + handle = tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False, encoding="utf-8") + with handle: + for entry in entries: + handle.write(json.dumps(entry) + "\n") + return handle.name + + +def run_entrypoint(payload: dict[str, object], env: dict[str, str]) -> subprocess.CompletedProcess[str]: + merged_env = os.environ.copy() + merged_env.update(env) + return subprocess.run( + [sys.executable, ENTRYPOINT], + input=json.dumps(payload), + capture_output=True, + text=True, + env=merged_env, + ) + + +class SdkModeTest(unittest.TestCase): + def test_mode_override_warn_changes_block_to_warning(self) -> None: + path = write_transcript([human("can you make all tasks use claude and local executor")]) + try: + result = run_entrypoint( + { + "tool_name": "Bash", + "hook_event_name": "PreToolUse", + "transcript_path": path, + "tool_input": {"command": fixture("update_tasks_status_in_pending_queued.txt")}, + }, + {"CATSTACK_HOOK_MODE_CATEGORICAL_SCOPE_GUARD": "warn"}, + ) + finally: + os.unlink(path) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertEqual("", result.stderr) + rendered = json.loads(result.stdout) + self.assertIn( + "categorical-scope-guard", + rendered["hookSpecificOutput"]["additionalContext"], + ) + + def test_each_finding_writes_one_event_row_with_rule_id(self) -> None: + path = write_transcript([human("can you make all tasks use claude and local executor")]) + try: + with tempfile.TemporaryDirectory() as tmp: + result = run_entrypoint( + { + "tool_name": "Bash", + "hook_event_name": "PreToolUse", + "session_id": "categorical-scope-guard-sdk-mode", + "transcript_path": path, + "tool_input": {"command": fixture("update_tasks_status_in_pending_queued.txt")}, + }, + { + "CATSTACK_HOOK_METRICS_DIR": tmp, + "CATSTACK_HOOK_MODE_CATEGORICAL_SCOPE_GUARD": "warn", + }, + ) + rows = [ + json.loads(line) + for file in Path(tmp).glob("events-*.jsonl") + for line in file.read_text(encoding="utf-8").splitlines() + ] + finally: + os.unlink(path) + + self.assertEqual(0, result.returncode, result.stderr) + self.assertEqual(1, len(rows)) + self.assertEqual("categorical-scope-guard", rows[0]["hook"]) + self.assertEqual("categorical-scope-guard.partial-status-filter", rows[0]["rule_id"]) + self.assertEqual("warn", rows[0]["mode"]) + self.assertEqual("override", rows[0]["mode_source"]) + self.assertEqual("warned", rows[0]["action"]) + + +if __name__ == "__main__": + unittest.main()