From f0c53132d1cea9b45d1337be5e4d91f6db8496ea Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 10:42:07 +0000 Subject: [PATCH] feat(guard): a policy implementation may be Python, not only shell Three coupled limits made a .py guard invisible end to end. Discovery in _guard_script matched .sh only, so a Python implementation was never found; run_guard hardcoded [bash, guard, *argv], so one handed to bash would fail; and Policy.guards globbed implementations/*.sh, so its eval suite silently demoted to tier 3 with no executable form. Each is a separate file and each hid the next. Discovery now walks GUARD_SUFFIXES, shell first, so a policy shipping both is deterministic rather than filesystem-ordered. run_guard picks the interpreter from the suffix -- this Python for .py, the existing bash probe otherwise -- and the "no usable bash" message becomes "no usable interpreter", since the condition it reports is no longer bash-specific. Plugin packaging needed no change: all four packagers already call _guard_script, so a Python guard now reaches plugin.json and the hooks file by the same path a shell one does. That was the point of fixing discovery rather than special-casing the emitters. What this does NOT change: a guard still receives shlex.split(command), so it judges a shell command. Guards that need file content -- the write-tool surface the adapters already extract `content` and `path` for, and which vendors.shell_matcher cannot emit a matcher for -- remain unavailable. That is a separate change to the input contract, not a follow-up to this one. Verification: pytest -q -> 1136 passed, 6 skipped pytest acceptance/ -> 21 passed ruff check . && ruff format --check . -> clean, 262 files chock check --only {matrix,mechanisms,validate,index,conflicts,evals,verify} chock check --only validate --mode {frontier-claude,frontier-devin} chock sync --repo . --check -> all PASS tools/check_literal_duplication.py -> no repeated literals Break-and-restore, each change reverted alone to prove a named test pins it: discovery -> test_discovery_finds_a_python_guard, test_the_plugin_packages_the_python_guard interpreter -> test_a_python_guard_blocks_and_allows eval glob -> test_a_python_guard_makes_the_suite_deterministic The vendored runtime is embedded in all ten adapters, so .chock/bin and the frozen runtime goldens move with guard_runner. Regenerated with CHOCK_REGEN_GOLDENS=1 and chock sync, not by hand; every adopter's next sync rewrites .chock/bin. Signed-off-by: Claude --- .chock/bin/antigravity.py | 16 +++- .chock/bin/claude_code.py | 16 +++- .chock/bin/codex_cli.py | 16 +++- .chock/bin/cursor.py | 16 +++- .chock/bin/devin.py | 16 +++- .chock/bin/gemini_cli.py | 16 +++- .chock/bin/grok.py | 16 +++- .chock/bin/tabnine.py | 16 +++- .chock/bin/vscode_copilot.py | 16 +++- .chock/bin/windsurf.py | 16 +++- src/chock/compile/emitters/in_agent.py | 9 +- src/chock/eval/suites.py | 5 +- src/chock/gate/guard_runner.py | 17 +++- tests/fixtures/runtime_goldens/antigravity.py | 16 +++- tests/fixtures/runtime_goldens/claude_code.py | 16 +++- tests/fixtures/runtime_goldens/codex_cli.py | 16 +++- tests/fixtures/runtime_goldens/cursor.py | 16 +++- tests/fixtures/runtime_goldens/devin.py | 16 +++- tests/fixtures/runtime_goldens/gemini_cli.py | 16 +++- tests/fixtures/runtime_goldens/grok.py | 16 +++- tests/fixtures/runtime_goldens/tabnine.py | 16 +++- .../runtime_goldens/vscode_copilot.py | 16 +++- tests/fixtures/runtime_goldens/windsurf.py | 16 +++- tests/test_python_guard.py | 89 +++++++++++++++++++ 24 files changed, 353 insertions(+), 87 deletions(-) create mode 100644 tests/test_python_guard.py diff --git a/.chock/bin/antigravity.py b/.chock/bin/antigravity.py index 8c285bf..d62bd8a 100755 --- a/.chock/bin/antigravity.py +++ b/.chock/bin/antigravity.py @@ -601,6 +601,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -640,6 +642,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -649,13 +657,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/claude_code.py b/.chock/bin/claude_code.py index b395abd..2c5e797 100755 --- a/.chock/bin/claude_code.py +++ b/.chock/bin/claude_code.py @@ -664,6 +664,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -703,6 +705,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -712,13 +720,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/codex_cli.py b/.chock/bin/codex_cli.py index 5f57668..fa89eb8 100755 --- a/.chock/bin/codex_cli.py +++ b/.chock/bin/codex_cli.py @@ -650,6 +650,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -689,6 +691,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -698,13 +706,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/cursor.py b/.chock/bin/cursor.py index 8bfc40d..80f9ceb 100755 --- a/.chock/bin/cursor.py +++ b/.chock/bin/cursor.py @@ -556,6 +556,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -595,6 +597,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -604,13 +612,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/devin.py b/.chock/bin/devin.py index b46bad5..76520a4 100755 --- a/.chock/bin/devin.py +++ b/.chock/bin/devin.py @@ -632,6 +632,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -671,6 +673,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -680,13 +688,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/gemini_cli.py b/.chock/bin/gemini_cli.py index eb78082..00e7101 100755 --- a/.chock/bin/gemini_cli.py +++ b/.chock/bin/gemini_cli.py @@ -632,6 +632,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -671,6 +673,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -680,13 +688,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/grok.py b/.chock/bin/grok.py index 678a2b9..63d0a80 100755 --- a/.chock/bin/grok.py +++ b/.chock/bin/grok.py @@ -615,6 +615,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -654,6 +656,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -663,13 +671,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/tabnine.py b/.chock/bin/tabnine.py index 5ddec24..580888b 100755 --- a/.chock/bin/tabnine.py +++ b/.chock/bin/tabnine.py @@ -615,6 +615,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -654,6 +656,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -663,13 +671,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/vscode_copilot.py b/.chock/bin/vscode_copilot.py index 12c383f..982e449 100755 --- a/.chock/bin/vscode_copilot.py +++ b/.chock/bin/vscode_copilot.py @@ -460,6 +460,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -499,6 +501,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -508,13 +516,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/.chock/bin/windsurf.py b/.chock/bin/windsurf.py index f06f012..b3b4a1c 100755 --- a/.chock/bin/windsurf.py +++ b/.chock/bin/windsurf.py @@ -531,6 +531,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -570,6 +572,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -579,13 +587,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/src/chock/compile/emitters/in_agent.py b/src/chock/compile/emitters/in_agent.py index ae310fa..685f411 100644 --- a/src/chock/compile/emitters/in_agent.py +++ b/src/chock/compile/emitters/in_agent.py @@ -19,11 +19,16 @@ } +#: Suffixes a guard implementation may carry, in the order discovery prefers them. +GUARD_SUFFIXES = (".sh", ".py") + + def _guard_script(policy_dir: Path, policy_id: str) -> str | None: """The policy's guard script name, by convention first, legacy map second.""" impl = policy_dir / "implementations" - if (impl / f"{policy_id}.sh").exists(): - return f"{policy_id}.sh" + for suffix in GUARD_SUFFIXES: + if (impl / f"{policy_id}{suffix}").exists(): + return f"{policy_id}{suffix}" legacy = GUARD_SCRIPTS.get(policy_id) if legacy and (impl / legacy).exists(): return legacy diff --git a/src/chock/eval/suites.py b/src/chock/eval/suites.py index 11d7ecc..d73ec70 100644 --- a/src/chock/eval/suites.py +++ b/src/chock/eval/suites.py @@ -7,6 +7,7 @@ import yaml +from chock.compile.emitters.in_agent import GUARD_SUFFIXES from chock.eval.model import Case from chock.manifest import load_manifest from chock.validation.loading import discover_artifacts @@ -67,7 +68,9 @@ def gate(self) -> dict[str, Any] | None: def guards(self) -> list[Path]: """Executable guard scripts shipped with the policy, if any.""" impl = self.dir / "implementations" - return sorted(impl.glob("*.sh")) if impl.is_dir() else [] + if not impl.is_dir(): + return [] + return sorted(p for p in impl.iterdir() if p.suffix in GUARD_SUFFIXES) @property def deterministic(self) -> bool: diff --git a/src/chock/gate/guard_runner.py b/src/chock/gate/guard_runner.py index a9ae218..533adb2 100644 --- a/src/chock/gate/guard_runner.py +++ b/src/chock/gate/guard_runner.py @@ -12,6 +12,8 @@ GUARD_VIOLATION = 1 +PYTHON_SUFFIX = ".py" + _BASH_CANDIDATES = ( "bash", r"C:\Program Files\Git\usr\bin\bash.exe", @@ -61,6 +63,13 @@ def find_bash(guard: Path) -> str | None: return None +def find_interpreter(guard: Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + + def run_guard(guard: Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -71,15 +80,15 @@ def run_guard(guard: Path, command: str) -> str: if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f"chock: no usable bash found, {guard.name} not checked", file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f"chock: no usable interpreter found, {guard.name} not checked", file=sys.stderr) return GUARD_UNCHECKED try: env = {**os.environ, "CHOCK_RAW_COMMAND": command} proc = subprocess.run( # noqa: S603 -- running the guard script against the command is the feature - [bash, str(guard), *args], + [interpreter, str(guard), *args], capture_output=True, text=True, encoding="utf-8", diff --git a/tests/fixtures/runtime_goldens/antigravity.py b/tests/fixtures/runtime_goldens/antigravity.py index 8c285bf..d62bd8a 100644 --- a/tests/fixtures/runtime_goldens/antigravity.py +++ b/tests/fixtures/runtime_goldens/antigravity.py @@ -601,6 +601,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -640,6 +642,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -649,13 +657,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/claude_code.py b/tests/fixtures/runtime_goldens/claude_code.py index b395abd..2c5e797 100644 --- a/tests/fixtures/runtime_goldens/claude_code.py +++ b/tests/fixtures/runtime_goldens/claude_code.py @@ -664,6 +664,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -703,6 +705,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -712,13 +720,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/codex_cli.py b/tests/fixtures/runtime_goldens/codex_cli.py index 5f57668..fa89eb8 100644 --- a/tests/fixtures/runtime_goldens/codex_cli.py +++ b/tests/fixtures/runtime_goldens/codex_cli.py @@ -650,6 +650,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -689,6 +691,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -698,13 +706,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/cursor.py b/tests/fixtures/runtime_goldens/cursor.py index 8bfc40d..80f9ceb 100644 --- a/tests/fixtures/runtime_goldens/cursor.py +++ b/tests/fixtures/runtime_goldens/cursor.py @@ -556,6 +556,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -595,6 +597,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -604,13 +612,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/devin.py b/tests/fixtures/runtime_goldens/devin.py index b46bad5..76520a4 100644 --- a/tests/fixtures/runtime_goldens/devin.py +++ b/tests/fixtures/runtime_goldens/devin.py @@ -632,6 +632,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -671,6 +673,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -680,13 +688,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/gemini_cli.py b/tests/fixtures/runtime_goldens/gemini_cli.py index eb78082..00e7101 100644 --- a/tests/fixtures/runtime_goldens/gemini_cli.py +++ b/tests/fixtures/runtime_goldens/gemini_cli.py @@ -632,6 +632,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -671,6 +673,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -680,13 +688,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/grok.py b/tests/fixtures/runtime_goldens/grok.py index 678a2b9..63d0a80 100644 --- a/tests/fixtures/runtime_goldens/grok.py +++ b/tests/fixtures/runtime_goldens/grok.py @@ -615,6 +615,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -654,6 +656,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -663,13 +671,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/tabnine.py b/tests/fixtures/runtime_goldens/tabnine.py index 5ddec24..580888b 100644 --- a/tests/fixtures/runtime_goldens/tabnine.py +++ b/tests/fixtures/runtime_goldens/tabnine.py @@ -615,6 +615,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -654,6 +656,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -663,13 +671,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/vscode_copilot.py b/tests/fixtures/runtime_goldens/vscode_copilot.py index 12c383f..982e449 100644 --- a/tests/fixtures/runtime_goldens/vscode_copilot.py +++ b/tests/fixtures/runtime_goldens/vscode_copilot.py @@ -460,6 +460,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -499,6 +501,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -508,13 +516,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/fixtures/runtime_goldens/windsurf.py b/tests/fixtures/runtime_goldens/windsurf.py index f06f012..b3b4a1c 100644 --- a/tests/fixtures/runtime_goldens/windsurf.py +++ b/tests/fixtures/runtime_goldens/windsurf.py @@ -531,6 +531,8 @@ def degrade(decision, event): # >>> agentseam handler >>> GUARD_VIOLATION = 1 +PYTHON_SUFFIX = '.py' + _BASH_CANDIDATES = ('bash', 'C:\\Program Files\\Git\\usr\\bin\\bash.exe', 'C:\\Program Files\\Git\\bin\\bash.exe', 'C:\\Program Files (x86)\\Git\\usr\\bin\\bash.exe', '/bin/bash', '/usr/bin/bash') GATE_LOG_ENV = 'CHOCK_GATE_LOG' @@ -570,6 +572,12 @@ def find_bash(guard: _chock_Path) -> str | None: return candidate return None +def find_interpreter(guard: _chock_Path) -> str | None: + """The interpreter that can run `guard`: this Python for `.py`, otherwise a usable bash.""" + if guard.suffix == PYTHON_SUFFIX: + return sys.executable or None + return find_bash(guard) + def run_guard(guard: _chock_Path, command: str) -> str: """`GUARD_BLOCKED` / `GUARD_CLEAN` when the guard ran, otherwise why it did not.""" try: @@ -579,13 +587,13 @@ def run_guard(guard: _chock_Path, command: str) -> str: return GUARD_UNCHECKED if not args: return GUARD_UNCHECKED - bash = find_bash(guard) - if bash is None: - print(f'chock: no usable bash found, {guard.name} not checked', file=sys.stderr) + interpreter = find_interpreter(guard) + if interpreter is None: + print(f'chock: no usable interpreter found, {guard.name} not checked', file=sys.stderr) return GUARD_UNCHECKED try: env = {**_chock_os.environ, 'CHOCK_RAW_COMMAND': command} - proc = _chock_subprocess.run([bash, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) + proc = _chock_subprocess.run([interpreter, str(guard), *args], capture_output=True, text=True, encoding='utf-8', errors='replace', env=env, timeout=_GUARD_TIMEOUT_SECONDS, check=False) except _chock_subprocess.TimeoutExpired: print(f'chock: guard timed out after {_GUARD_TIMEOUT_SECONDS}s, not checked', file=sys.stderr) return GUARD_ERRORED diff --git a/tests/test_python_guard.py b/tests/test_python_guard.py new file mode 100644 index 0000000..a50e204 --- /dev/null +++ b/tests/test_python_guard.py @@ -0,0 +1,89 @@ +"""A guard implementation may be Python, not only shell -- discovery, execution, evals, packaging.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import yaml + +from chock.compile.emitters.in_agent import GUARD_SUFFIXES, _guard_script +from chock.eval.suites import Policy +from chock.gate.guard_runner import GUARD_BLOCKED, GUARD_CLEAN, find_interpreter, run_guard +from chock.plugin.claude import claude_plugin_files + +POLICY_ID = "refuse-widget" + +_GUARD_SOURCE = '''#!/usr/bin/env python3 +"""Block any command mentioning the widget; the argv contract is the shell one.""" + +import sys + +if any("widget" in a for a in sys.argv[1:]): + print("widget commands are refused", file=sys.stderr) + raise SystemExit(1) +raise SystemExit(0) +''' + +_MANIFEST = { + "id": POLICY_ID, + "name": "Refuse Widget", + "version": "0.0.1", + "description": "trigger: widget commands. avoid: running them.", + "artifact": "rule", + "enforcement": "advise", + "rule": {"text": "block(widget): any_command\nprefer: the supported path\n"}, +} + + +def _policy(tmp_path: Path, suffix: str, *, source: str = _GUARD_SOURCE) -> Path: + policy_dir = tmp_path / ".agents" / "policies" / POLICY_ID + (policy_dir / "implementations").mkdir(parents=True) + (policy_dir / "manifest.yaml").write_text(yaml.safe_dump(_MANIFEST), encoding="utf-8") + (policy_dir / "implementations" / f"{POLICY_ID}{suffix}").write_text(source, encoding="utf-8") + return policy_dir + + +def test_discovery_finds_a_python_guard(tmp_path: Path) -> None: + """Without this a .py implementation is invisible: never wired, never packaged.""" + assert _guard_script(_policy(tmp_path, ".py"), POLICY_ID) == f"{POLICY_ID}.py" + + +def test_shell_wins_when_a_policy_ships_both(tmp_path: Path) -> None: + """Discovery order is fixed, so which guard runs never depends on filesystem order.""" + policy_dir = _policy(tmp_path, ".py") + (policy_dir / "implementations" / f"{POLICY_ID}.sh").write_text("exit 0\n", encoding="utf-8") + assert _guard_script(policy_dir, POLICY_ID) == f"{POLICY_ID}.sh" + assert GUARD_SUFFIXES.index(".sh") < GUARD_SUFFIXES.index(".py") + + +def test_interpreter_follows_the_suffix(tmp_path: Path) -> None: + """A .py guard handed to bash would fail; the suffix picks the interpreter.""" + assert find_interpreter(Path("x/implementations/p.py")) == sys.executable + shell = find_interpreter(_policy(tmp_path, ".sh", source="exit 0\n") / "implementations" / f"{POLICY_ID}.sh") + assert shell is None or "bash" in shell + + +def test_a_python_guard_blocks_and_allows(tmp_path: Path) -> None: + """End-to-end: the runner executes it and reads its exit code as a verdict.""" + guard = _policy(tmp_path, ".py") / "implementations" / f"{POLICY_ID}.py" + assert run_guard(guard, "make widget") == GUARD_BLOCKED + assert run_guard(guard, "make gadget") == GUARD_CLEAN + + +def test_a_python_guard_makes_the_suite_deterministic(tmp_path: Path) -> None: + """A shipped guard the eval runner cannot see would silently demote the policy to tier 3.""" + policy_dir = _policy(tmp_path, ".py") + suite = Policy(POLICY_ID, policy_dir, _MANIFEST) + assert [p.name for p in suite.guards] == [f"{POLICY_ID}.py"] + assert suite.deterministic + + +def test_the_plugin_packages_the_python_guard(tmp_path: Path) -> None: + """The packaged plugin is what other clients read, so the guard must reach it verbatim.""" + policy_dir = _policy(tmp_path, ".py") + files = claude_plugin_files(policy_dir, _MANIFEST, tmp_path) + packaged = {p.as_posix(): body for p, body in files.items()} + guard_rel = next(rel for rel in packaged if rel.endswith(f"{POLICY_ID}.py")) + assert packaged[guard_rel] == _GUARD_SOURCE + assert any(rel.endswith("hooks.json") for rel in packaged), "a shipped guard must be wired"