diff --git a/.chock/bin/claude_code.py b/.chock/bin/claude_code.py index 93fb51ad..f39b0fd3 100755 --- a/.chock/bin/claude_code.py +++ b/.chock/bin/claude_code.py @@ -741,6 +741,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -804,6 +806,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -872,7 +875,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -885,6 +888,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -941,14 +945,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2273,6 +2277,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': True} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2284,7 +2301,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/codex_cli.py b/.chock/bin/codex_cli.py index aeac58cd..f1e41dba 100755 --- a/.chock/bin/codex_cli.py +++ b/.chock/bin/codex_cli.py @@ -738,6 +738,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -801,6 +803,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -869,7 +872,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -882,6 +885,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -938,14 +942,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2204,6 +2208,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2215,7 +2232,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/cursor.py b/.chock/bin/cursor.py index a119469a..af70c81c 100755 --- a/.chock/bin/cursor.py +++ b/.chock/bin/cursor.py @@ -656,6 +656,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -719,6 +721,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -787,7 +790,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -800,6 +803,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -856,14 +860,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2122,6 +2126,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'beforeMCPExecution': True, 'beforeReadFile': False, 'beforeShellExecution': True, 'beforeTabFileRead': False, 'preToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(cursor_wire(raw), False) def _spoken(verdict): @@ -2133,7 +2150,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/devin.py b/.chock/bin/devin.py index 6b99643b..e93a7a95 100755 --- a/.chock/bin/devin.py +++ b/.chock/bin/devin.py @@ -710,6 +710,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -773,6 +775,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -841,7 +844,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -854,6 +857,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -910,14 +914,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2176,6 +2180,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PermissionRequest': False, 'PreToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2187,7 +2204,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/gemini_cli.py b/.chock/bin/gemini_cli.py index 552639d2..1afd4d41 100755 --- a/.chock/bin/gemini_cli.py +++ b/.chock/bin/gemini_cli.py @@ -710,6 +710,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -773,6 +775,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -841,7 +844,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -854,6 +857,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -910,14 +914,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2176,6 +2180,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'BeforeTool': True} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2187,7 +2204,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/grok.py b/.chock/bin/grok.py index cdb01e04..470de4f2 100755 --- a/.chock/bin/grok.py +++ b/.chock/bin/grok.py @@ -693,6 +693,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -756,6 +758,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -824,7 +827,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -837,6 +840,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -893,14 +897,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2159,6 +2163,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': False} +_CHOCK_EVENT_KEYS = ('hookEventName', 'hook_event_name', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2170,7 +2187,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/tabnine.py b/.chock/bin/tabnine.py index f6c5beb2..6d7f192c 100755 --- a/.chock/bin/tabnine.py +++ b/.chock/bin/tabnine.py @@ -693,6 +693,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -756,6 +758,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -824,7 +827,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -837,6 +840,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -893,14 +897,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2159,6 +2163,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'BeforeTool': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2170,7 +2187,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/vscode_copilot.py b/.chock/bin/vscode_copilot.py index 5a3e1907..c7c3a84a 100755 --- a/.chock/bin/vscode_copilot.py +++ b/.chock/bin/vscode_copilot.py @@ -589,6 +589,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -652,6 +654,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -720,7 +723,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -733,6 +736,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -789,14 +793,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2055,6 +2059,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = _chock_copilot_respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': True, 'preToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2066,7 +2083,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/.chock/bin/windsurf.py b/.chock/bin/windsurf.py index 35ab986e..586e0904 100755 --- a/.chock/bin/windsurf.py +++ b/.chock/bin/windsurf.py @@ -607,6 +607,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -670,6 +672,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -738,7 +741,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -751,6 +754,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -807,14 +811,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2073,6 +2077,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'pre_mcp_tool_use': False, 'pre_run_command': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2084,7 +2101,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/changelog.d/guard-can-ask-env.md b/changelog.d/guard-can-ask-env.md new file mode 100644 index 00000000..e317ebab --- /dev/null +++ b/changelog.d/guard-can-ask-env.md @@ -0,0 +1,7 @@ +- **Guards learn whether the client can prompt (`CHOCK_CAN_ASK`).** A guard run by a vendored hook + runtime now sees `CHOCK_CAN_ASK=1` when the event that fired turns an ask (exit 3) into a prompt, and + `0` when it turns it into a block or the runtime cannot name the event: 1 at Claude Code, Copilot and + Gemini CLI pre-tool and Cursor `beforeShellExecution`; 0 at Cursor `preToolUse`, Codex, Devin, Grok, + Tabnine and Windsurf. It is unset everywhere else (eval replay, direct calls) and never inherited from + the caller's environment; a guard treats unset as "can ask". Opt-in: no guard reads it yet, so no + verdict changes. diff --git a/spec/script-backed-gates.md b/spec/script-backed-gates.md index 02c43aaf..6becac3f 100644 --- a/spec/script-backed-gates.md +++ b/spec/script-backed-gates.md @@ -137,6 +137,13 @@ eval runner, always, whether or not the command parses. - **`CHOCK_TOOL`.** The tool that ran the command, from the hook payload's tool name: `bash` (Bash), `powershell` (PowerShell, pwsh), `shell` (sh, shell, run_shell_command, Cursor's beforeShellExecution, any other name containing "shell"), else `unknown`. +- **`CHOCK_CAN_ASK`.** Set by the vendored hook runtime: `1` when the event that fired shows an + ask (exit 3) as a prompt, `0` when the client turns it into a block. Taken from the vendor data's + `honours_escalate` for that wire event: `1` at Claude Code `PreToolUse`, Copilot `PreToolUse`, + Gemini CLI `BeforeTool`, Cursor `beforeShellExecution`/`beforeMCPExecution`; `0` at every other + pre-tool event (Cursor `preToolUse`, Codex, Devin, Grok, Tabnine, Windsurf) and at an event the + runtime cannot name. Unset everywhere else (eval replay, a direct call) and never inherited from + the caller's environment. A guard must treat unset as "can ask". - **Exit codes.** `0` allows. `3` asks; the first output line is the prompt. (`4` warns only for a `script` gate and a script-backed git hook, above; a command guard exiting 4 is not checked.) `1` blocks only when the guard gave a reason: non-empty stderr (stdout also read) that is not a crash diff --git a/src/chock/gate/data/dispatch.py.tmpl b/src/chock/gate/data/dispatch.py.tmpl index 6f38df2d..fd6b7459 100644 --- a/src/chock/gate/data/dispatch.py.tmpl +++ b/src/chock/gate/data/dispatch.py.tmpl @@ -1,3 +1,16 @@ +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = __CAN_ASK__ +_CHOCK_EVENT_KEYS = __EVENT_KEYS__ + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(__WIRE_NAME__, False) def _spoken(verdict): @@ -9,7 +22,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/src/chock/gate/guard_runner.py b/src/chock/gate/guard_runner.py index e8e35d7d..8df44880 100644 --- a/src/chock/gate/guard_runner.py +++ b/src/chock/gate/guard_runner.py @@ -26,6 +26,7 @@ TOOL_UNKNOWN = "unknown" #: Set to 1 in the guard's environment when argv is a plain whitespace split, not a shell parse. ARGV_FALLBACK_ENV = "CHOCK_ARGV_FALLBACK" +CAN_ASK_ENV = "CHOCK_CAN_ASK" # "1"/"0": the client prompts on a guard's ask; unset when the caller cannot say _POSIX_BASH = ("bash", "/bin/bash", "/usr/bin/bash") #: Where Git for Windows installs when `git` is not on PATH to say where it is. @@ -87,6 +88,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(os.environ) + env.pop(CAN_ASK_ENV, None) # only a caller that knows the client may say; never inherited if os.name != _WINDOWS: return env home = Path(interpreter).parent @@ -163,7 +165,7 @@ def _no_interpreter(guard: Path, deadline: float) -> tuple[str, str]: return GUARD_ERRORED, reason -def run_guard_detailed(guard: Path, command: str, tool: str = "") -> tuple[str, str]: +def run_guard_detailed(guard: Path, command: str, tool: str = "", *, can_ask: bool | None = None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -178,6 +180,7 @@ def run_guard_detailed(guard: Path, command: str, tool: str = "") -> tuple[str, env = {**interpreter_env(interpreter), "CHOCK_RAW_COMMAND": command, "CHOCK_TOOL": normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = "1" + env.update({} if can_ask is None else {CAN_ASK_ENV: "1" if can_ask else "0"}) proc = subprocess.run( # noqa: S603 -- running the guard script against the command is the feature [interpreter, str(guard), *args], capture_output=True, @@ -260,7 +263,7 @@ def log_outcome(guard: Path, tool: str, *, verdict: str) -> None: return -def evaluate(argv: list[str], command: str, tool: str = "") -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str = "", *, can_ask: bool | None = None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: @@ -272,7 +275,7 @@ def evaluate(argv: list[str], command: str, tool: str = "") -> tuple[str, str] | f"chock guard {guard} is missing, so this command cannot be checked. " "Run `chock sync --repo .` to reinstall the policy's guards.", ) - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: "block", GUARD_ASKED: "ask", GUARD_CLEAN: "allow"} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) diff --git a/src/chock/gate/runtime_bundle.py b/src/chock/gate/runtime_bundle.py index a103b247..6d66ab7b 100644 --- a/src/chock/gate/runtime_bundle.py +++ b/src/chock/gate/runtime_bundle.py @@ -90,6 +90,7 @@ def _extract(module) -> str: _DISPATCH = _DATA_DIR.joinpath("dispatch.py.tmpl").read_text(encoding="utf-8") _DISPATCH_BRANCH_TOKEN = "# __SESSION_START_BRANCH__\n" # noqa: S105 -- a template marker, not a credential +_WIRE_NAME_OF = {"cursor": "cursor_wire(raw)"} # a family that infers the wire name from shape: the same way here _UNREADABLE = _DATA_DIR.joinpath("unreadable_payload.py.tmpl").read_text(encoding="utf-8") _WIRE_RAW_TOKENS = {"__PRE_TOOL_RAW__": contract.PRE_TOOL, "__STOP_RAW__": contract.STOP} @@ -167,6 +168,24 @@ def _unreadable_source(agent: str) -> str: _STOP_WARN_RESPOND_CALL = "return _chock_stop_warn_respond(degrade(decision, event), event)" +def pre_tool_can_ask(agent: str) -> dict[str, bool]: + """Per wire event `agent` maps to pre_tool: whether that gate shows an escalate as a prompt.""" + vendor = entry(agent) + gates = vendor.get("verdicts", {}).get("gates", {}) + pre_tool = [name for name, canonical in vendor.get("events", {}).items() if canonical == contract.PRE_TOOL] + return {name: bool((gates.get(name) or {}).get("honours_escalate")) for name in pre_tool} + + +def _can_ask_tokens(agent: str) -> dict[str, str]: + """The dispatcher's can-ask map, the keys naming a payload's event, and how this bundle reads the name.""" + vendor = entry(agent) + return { + "__CAN_ASK__": repr(pre_tool_can_ask(agent)), + "__EVENT_KEYS__": repr(tuple(dict.fromkeys((*vendor["claims"].get("event_key", ()), *_EVENT_KEYS)))), + "__WIRE_NAME__": _WIRE_NAME_OF.get(vendor.get("family"), "_chock_wire_name(raw)"), + } + + def _handler_source(agent: str) -> str: """The full handler-block body for `agent`: extracted guard logic, optionally extracted""" parts = [ @@ -202,7 +221,10 @@ def _handler_source(agent: str) -> str: parts.append(_SESSION_START_ORCHESTRATION) branch = _SESSION_START_BRANCH if agent in _SESSION_START_AGENTS else "" parts.append(_unreadable_source(agent)) - parts.append(_DISPATCH.replace(_DISPATCH_BRANCH_TOKEN, branch)) + dispatch = _DISPATCH.replace(_DISPATCH_BRANCH_TOKEN, branch) + for token, value in _can_ask_tokens(agent).items(): + dispatch = dispatch.replace(token, value) + parts.append(dispatch) if agent in _COPILOT_RESPOND_AGENTS: parts.append(_COPILOT_RESPOND) if agent in _WARN_RESPOND_AGENTS: diff --git a/tests/fixtures/runtime_goldens/antigravity.py b/tests/fixtures/runtime_goldens/antigravity.py index bf333b71..72ca2602 100644 --- a/tests/fixtures/runtime_goldens/antigravity.py +++ b/tests/fixtures/runtime_goldens/antigravity.py @@ -676,6 +676,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -739,6 +741,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -807,7 +810,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -820,6 +823,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -876,14 +880,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2142,6 +2146,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2153,7 +2170,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/claude_code.py b/tests/fixtures/runtime_goldens/claude_code.py index 93fb51ad..f39b0fd3 100644 --- a/tests/fixtures/runtime_goldens/claude_code.py +++ b/tests/fixtures/runtime_goldens/claude_code.py @@ -741,6 +741,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -804,6 +806,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -872,7 +875,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -885,6 +888,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -941,14 +945,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2273,6 +2277,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': True} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2284,7 +2301,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/codex_cli.py b/tests/fixtures/runtime_goldens/codex_cli.py index aeac58cd..f1e41dba 100644 --- a/tests/fixtures/runtime_goldens/codex_cli.py +++ b/tests/fixtures/runtime_goldens/codex_cli.py @@ -738,6 +738,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -801,6 +803,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -869,7 +872,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -882,6 +885,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -938,14 +942,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2204,6 +2208,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2215,7 +2232,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/cursor.py b/tests/fixtures/runtime_goldens/cursor.py index a119469a..af70c81c 100644 --- a/tests/fixtures/runtime_goldens/cursor.py +++ b/tests/fixtures/runtime_goldens/cursor.py @@ -656,6 +656,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -719,6 +721,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -787,7 +790,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -800,6 +803,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -856,14 +860,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2122,6 +2126,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'beforeMCPExecution': True, 'beforeReadFile': False, 'beforeShellExecution': True, 'beforeTabFileRead': False, 'preToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(cursor_wire(raw), False) def _spoken(verdict): @@ -2133,7 +2150,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/devin.py b/tests/fixtures/runtime_goldens/devin.py index 6b99643b..e93a7a95 100644 --- a/tests/fixtures/runtime_goldens/devin.py +++ b/tests/fixtures/runtime_goldens/devin.py @@ -710,6 +710,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -773,6 +775,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -841,7 +844,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -854,6 +857,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -910,14 +914,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2176,6 +2180,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PermissionRequest': False, 'PreToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2187,7 +2204,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/gemini_cli.py b/tests/fixtures/runtime_goldens/gemini_cli.py index 552639d2..1afd4d41 100644 --- a/tests/fixtures/runtime_goldens/gemini_cli.py +++ b/tests/fixtures/runtime_goldens/gemini_cli.py @@ -710,6 +710,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -773,6 +775,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -841,7 +844,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -854,6 +857,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -910,14 +914,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2176,6 +2180,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'BeforeTool': True} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2187,7 +2204,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/grok.py b/tests/fixtures/runtime_goldens/grok.py index cdb01e04..470de4f2 100644 --- a/tests/fixtures/runtime_goldens/grok.py +++ b/tests/fixtures/runtime_goldens/grok.py @@ -693,6 +693,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -756,6 +758,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -824,7 +827,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -837,6 +840,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -893,14 +897,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2159,6 +2163,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': False} +_CHOCK_EVENT_KEYS = ('hookEventName', 'hook_event_name', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2170,7 +2187,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/tabnine.py b/tests/fixtures/runtime_goldens/tabnine.py index f6c5beb2..6d7f192c 100644 --- a/tests/fixtures/runtime_goldens/tabnine.py +++ b/tests/fixtures/runtime_goldens/tabnine.py @@ -693,6 +693,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -756,6 +758,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -824,7 +827,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -837,6 +840,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -893,14 +897,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2159,6 +2163,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'BeforeTool': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2170,7 +2187,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/vscode_copilot.py b/tests/fixtures/runtime_goldens/vscode_copilot.py index 5a3e1907..c7c3a84a 100644 --- a/tests/fixtures/runtime_goldens/vscode_copilot.py +++ b/tests/fixtures/runtime_goldens/vscode_copilot.py @@ -589,6 +589,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -652,6 +654,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -720,7 +723,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -733,6 +736,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -789,14 +793,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2055,6 +2059,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = _chock_copilot_respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'PreToolUse': True, 'preToolUse': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2066,7 +2083,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/fixtures/runtime_goldens/windsurf.py b/tests/fixtures/runtime_goldens/windsurf.py index 35ab986e..586e0904 100644 --- a/tests/fixtures/runtime_goldens/windsurf.py +++ b/tests/fixtures/runtime_goldens/windsurf.py @@ -607,6 +607,8 @@ def allowed_seconds(expired): ARGV_FALLBACK_ENV = 'CHOCK_ARGV_FALLBACK' +CAN_ASK_ENV = 'CHOCK_CAN_ASK' + _POSIX_BASH = ('bash', '/bin/bash', '/usr/bin/bash') _GIT_FOR_WINDOWS = ('C:\\Program Files\\Git', 'C:\\Program Files (x86)\\Git') @@ -670,6 +672,7 @@ def bash_candidates() -> list[str]: def interpreter_env(interpreter: str) -> dict[str, str]: """The environment a guard runs in: on Windows, Git's usr/bin ahead on PATH for sed and grep.""" env = dict(_chock_os.environ) + env.pop(CAN_ASK_ENV, None) if _chock_os.name != _WINDOWS: return env home = _chock_Path(interpreter).parent @@ -738,7 +741,7 @@ def _no_interpreter(guard: _chock_Path, deadline: float) -> tuple[str, str]: print(f'chock: {reason}', file=sys.stderr) return (GUARD_ERRORED, reason) -def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[str, str]: +def run_guard_detailed(guard: _chock_Path, command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str]: """`run_guard`'s verdict plus the guard's own first line, which an ask carries to the user.""" args, fallback = split_command(command) if not args: @@ -751,6 +754,7 @@ def run_guard_detailed(guard: _chock_Path, command: str, tool: str='') -> tuple[ env = {**interpreter_env(interpreter), 'CHOCK_RAW_COMMAND': command, 'CHOCK_TOOL': normalize_tool(tool)} if fallback: env[ARGV_FALLBACK_ENV] = '1' + env.update({} if can_ask is None else {CAN_ASK_ENV: '1' if can_ask else '0'}) proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=time_left(deadline, [interpreter, str(guard)]), check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {ENGINE_BUDGET_SECONDS}s, not checked', file=sys.stderr) @@ -807,14 +811,14 @@ def log_outcome(guard: _chock_Path, tool: str, *, verdict: str) -> None: except Exception: return -def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | None: +def evaluate(argv: list[str], command: str, tool: str='', *, can_ask: bool | None=None) -> tuple[str, str] | None: """Run the guard named on `argv` (`--guard `) against `command`.""" guard = guard_path_from_argv(argv) if guard is None: return None if not guard.exists(): return (VERDICT_DENY, f"chock guard {guard} is missing, so this command cannot be checked. Run `chock sync --repo .` to reinstall the policy's guards.") - verdict, message = run_guard_detailed(guard, command, tool) + verdict, message = run_guard_detailed(guard, command, tool, can_ask=can_ask) logged = {GUARD_BLOCKED: 'block', GUARD_ASKED: 'ask', GUARD_CLEAN: 'allow'} if verdict in logged: log_outcome(guard, tool, verdict=logged[verdict]) @@ -2073,6 +2077,19 @@ def _chock_unreadable(): decision = Decision.escalate(reason) if action == "ask" else Decision.deny(reason) text, code = respond(degrade(decision, event), event) return (text, code) if _chock_refuses(text, code) else ("", 2) +#: Per wire event mapped to pre_tool: whether the client shows a guard's ask as a prompt. +_CHOCK_CAN_ASK = {'pre_mcp_tool_use': False, 'pre_run_command': False} +_CHOCK_EVENT_KEYS = ('hook_event_name', 'hookEventName', 'agent_action_name') + + +def _chock_wire_name(raw): + names = (raw.get(key) for key in _CHOCK_EVENT_KEYS) if isinstance(raw, dict) else () + return next((name for name in names if isinstance(name, str)), None) + + +def _chock_can_ask(raw): + # An event this bundle cannot name, or one the vendor data does not list, cannot prompt. + return _CHOCK_CAN_ASK.get(_chock_wire_name(raw), False) def _spoken(verdict): @@ -2084,7 +2101,7 @@ def _spoken(verdict): def _judge(event): if event.event == "pre_tool" and event.command: - verdict = evaluate(sys.argv[1:], event.command, event.tool or "") + verdict = evaluate(sys.argv[1:], event.command, event.tool or "", can_ask=_chock_can_ask(event.raw)) if verdict is not None: outcome, reason = verdict return Decision.escalate(reason) if outcome == ESCALATE else Decision.deny(reason) diff --git a/tests/test_guard_can_ask.py b/tests/test_guard_can_ask.py new file mode 100644 index 00000000..bdc15306 --- /dev/null +++ b/tests/test_guard_can_ask.py @@ -0,0 +1,138 @@ +"""CHOCK_CAN_ASK: the dispatcher tells a guard whether its client prompts on an ask; nobody else does.""" + +from __future__ import annotations + +import json +import os +import stat +import subprocess +import sys +from pathlib import Path + +import pytest + +from chock.eval import execute +from chock.gate import guard_runner, runtime_bundle + +DUMP = 'printf "%s" "${CHOCK_CAN_ASK-unset}" > "$OUT"; exit 0' +ASK = 'echo "confirm first"; exit 3' +BLOCK = 'echo "no"; exit 1' +CLEAN = "exit 0" + +#: The plan's table (org-plan plan/one-oh-gaps.md C1b): each vendor's value at its shell pre-tool event. +PLANNED = {"claude_code": "1", "vscode_copilot": "1", "cursor": "1", "codex_cli": "0", "devin": "0"} + +_HOOK = {"hook_event_name": "PreToolUse", "tool_name": "Bash", "tool_input": {"command": "ls"}} +_CURSOR = {"command": "ls", "cwd": ".", "conversation_id": "c", "workspace_roots": ["."]} +#: One shell pre_tool payload per bundle that has one, with the wire event it names. +PAYLOADS = { + "claude_code": ("PreToolUse", _HOOK), + "codex_cli": ("PreToolUse", _HOOK), + "cursor": ("beforeShellExecution", {**_CURSOR, "hook_event_name": "beforeShellExecution"}), + "devin": ("PreToolUse", _HOOK), + "gemini_cli": ("BeforeTool", {**_HOOK, "hook_event_name": "BeforeTool"}), + "tabnine": ("BeforeTool", {**_HOOK, "hook_event_name": "BeforeTool"}), + "vscode_copilot": ("PreToolUse", _HOOK), + "windsurf": ("pre_run_command", {"agent_action_name": "pre_run_command", "tool_info": {"command_line": "ls"}}), +} +#: Cursor events beyond the shell one: the value follows the event that fired, not the agent. +CURSOR_EVENTS = [ + ("preToolUse", {**_HOOK, "hook_event_name": "preToolUse", "conversation_id": "c", "workspace_roots": ["."]}, "0"), + ("unnamed shell shape", _CURSOR, "1"), +] + + +def make_guard(tmp_path: Path, body: str) -> Path: + guard = tmp_path / "g.sh" + guard.write_text(f"#!/usr/bin/env bash\n{body}\n", encoding="utf-8", newline="\n") + guard.chmod(guard.stat().st_mode | stat.S_IEXEC) + return guard + + +def seen(tmp_path: Path) -> str: + return (tmp_path / "seen.txt").read_text(encoding="utf-8") + + +@pytest.fixture +def dumping_guard(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + monkeypatch.setenv("OUT", str(tmp_path / "seen.txt")) + # An inherited value must never reach the guard: only the dispatcher may say. + monkeypatch.setenv(guard_runner.CAN_ASK_ENV, "inherited") + return make_guard(tmp_path, DUMP) + + +def run_bundle(agent: str, payload: dict, guard: Path, tmp_path: Path) -> subprocess.CompletedProcess[bytes]: + runtime = tmp_path / f"{agent}.py" + runtime.write_text(runtime_bundle.render(agent), encoding="utf-8") + return subprocess.run( + [sys.executable, str(runtime), "--guard", str(guard)], + input=json.dumps(payload).encode(), + capture_output=True, + cwd=tmp_path, + env=os.environ.copy(), + check=False, + ) + + +@pytest.mark.parametrize("agent", sorted(PLANNED)) +def test_each_planned_vendor_resolves_to_the_planned_value(agent: str) -> None: + event = PAYLOADS[agent][0] + assert str(int(runtime_bundle.pre_tool_can_ask(agent)[event])) == PLANNED[agent] + + +@pytest.mark.parametrize("agent", sorted(PAYLOADS)) +def test_the_dispatcher_hands_each_vendors_guard_its_value(agent, tmp_path, dumping_guard) -> None: + event, payload = PAYLOADS[agent] + + run_bundle(agent, payload, dumping_guard, tmp_path) + + expected = str(int(runtime_bundle.pre_tool_can_ask(agent)[event])) + assert seen(tmp_path) == PLANNED.get(agent, expected) == expected + + +@pytest.mark.parametrize(("event", "payload", "expected"), CURSOR_EVENTS, ids=[e[0] for e in CURSOR_EVENTS]) +def test_cursor_follows_the_event_that_fired(event, payload, expected, tmp_path, dumping_guard) -> None: + run_bundle("cursor", payload, dumping_guard, tmp_path) + + assert seen(tmp_path) == expected + + +@pytest.mark.parametrize("agent", sorted(PAYLOADS)) +def test_an_unknown_event_name_cannot_ask(agent) -> None: + namespace: dict = {} + exec(runtime_bundle.render(agent), namespace) # noqa: S102 -- loading the rendered bundle is the test + unknown = {"hook_event_name": "NotAnEvent", "hookEventName": "NotAnEvent", "agent_action_name": "NotAnEvent"} + + assert namespace["_chock_can_ask"](unknown) is False + + +@pytest.mark.parametrize("can_ask", [True, False]) +def test_evaluate_sets_one_or_zero(can_ask, tmp_path, dumping_guard) -> None: + guard_runner.evaluate(["--guard", str(dumping_guard)], "ls", "Bash", can_ask=can_ask) + + assert seen(tmp_path) == ("1" if can_ask else "0") + + +def test_unset_outside_the_dispatcher(tmp_path, dumping_guard) -> None: + guard_runner.evaluate(["--guard", str(dumping_guard)], "ls", "Bash") + assert seen(tmp_path) == "unset" + + guard_runner.run_guard(dumping_guard, "ls") + assert seen(tmp_path) == "unset" + + +def test_unset_in_eval_replay(tmp_path, dumping_guard) -> None: + execute._run_guard(tmp_path, dumping_guard, "ls") + + assert seen(tmp_path) == "unset" + + +@pytest.mark.parametrize("body", [ASK, BLOCK, CLEAN]) +def test_no_guard_verdict_changes_with_the_value(body, tmp_path) -> None: + guard = make_guard(tmp_path, body) + argv = ["--guard", str(guard)] + + legacy = guard_runner.evaluate(argv, "ls", "Bash") + + assert guard_runner.evaluate(argv, "ls", "Bash", can_ask=True) == legacy + assert guard_runner.evaluate(argv, "ls", "Bash", can_ask=False) == legacy diff --git a/tests/test_pretooluse_logging.py b/tests/test_pretooluse_logging.py index 633817e9..56c8f86f 100644 --- a/tests/test_pretooluse_logging.py +++ b/tests/test_pretooluse_logging.py @@ -47,7 +47,7 @@ def evaluate(guard: Path, command: str, tool: str = "Bash") -> tuple[str, str] | def test_block_is_recorded(tmp_path: Path, monkeypatch) -> None: guard = make_guard(tmp_path, BLOCKING_GUARD) - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_BLOCKED, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_BLOCKED, "")) assert evaluate(guard, "rm -rf /") == (guard_runner.VERDICT_DENY, "Blocked by chock policy: block-destructive") @@ -63,7 +63,7 @@ def test_block_is_recorded(tmp_path: Path, monkeypatch) -> None: def test_allow_is_recorded(tmp_path: Path, monkeypatch) -> None: guard = make_guard(tmp_path, CLEAN_GUARD) - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_CLEAN, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_CLEAN, "")) assert evaluate(guard, "ls -la") is None assert read_log(tmp_path)[0]["verdict"] == "allow" @@ -89,11 +89,11 @@ def test_unchecked_guard_is_not_recorded_as_a_pass(tmp_path: Path, monkeypatch) """The distinction the outcome words exist for: nothing was checked.""" guard = make_guard(tmp_path, BROKEN_GUARD) - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_UNCHECKED, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_UNCHECKED, "")) assert evaluate(guard, "ls -la") is None assert read_log(tmp_path) == [] - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_ERRORED, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_ERRORED, "")) outcome, reason = evaluate(guard, "ls -la") assert outcome == guard_runner.VERDICT_ESCALATE assert reason, "a confirmation request must say why it is being asked for" @@ -103,7 +103,7 @@ def test_unchecked_guard_is_not_recorded_as_a_pass(tmp_path: Path, monkeypatch) def test_command_never_reaches_the_log(tmp_path: Path, monkeypatch) -> None: """Commands carry credentials far more often than files do.""" guard = make_guard(tmp_path, BLOCKING_GUARD) - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_BLOCKED, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_BLOCKED, "")) evaluate(guard, SECRET_COMMAND) @@ -114,7 +114,7 @@ def test_command_never_reaches_the_log(tmp_path: Path, monkeypatch) -> None: def test_env_switch_disables_logging(tmp_path: Path, monkeypatch) -> None: guard = make_guard(tmp_path, BLOCKING_GUARD) - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_BLOCKED, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_BLOCKED, "")) monkeypatch.setenv(GATE_LOG_ENV, "0") assert evaluate(guard, "rm -rf /") == (guard_runner.VERDICT_DENY, "Blocked by chock policy: block-destructive") @@ -123,7 +123,7 @@ def test_env_switch_disables_logging(tmp_path: Path, monkeypatch) -> None: def test_logging_failure_does_not_change_the_verdict(tmp_path: Path, monkeypatch) -> None: guard = make_guard(tmp_path, BLOCKING_GUARD) - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_BLOCKED, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_BLOCKED, "")) class Exploding: @staticmethod @@ -140,7 +140,7 @@ def test_guard_outside_an_chock_repo_is_not_recorded(tmp_path: Path, monkeypatch guard = tmp_path / "loose" / "implementations" / "block-destructive.sh" guard.parent.mkdir(parents=True, exist_ok=True) guard.write_text(BLOCKING_GUARD, encoding="utf-8", newline="\n") - monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_: (guard_runner.GUARD_BLOCKED, "")) + monkeypatch.setattr(guard_runner, "run_guard_detailed", lambda *_, **__: (guard_runner.GUARD_BLOCKED, "")) assert evaluate(guard, "rm -rf /") == (guard_runner.VERDICT_DENY, "Blocked by chock policy: block-destructive") assert read_log(tmp_path) == []