"""Behavior checks with synthetic data; never read real credentials.""" import contextlib import importlib.util import io import json from pathlib import Path import tempfile import unittest import sys from unittest.mock import patch sys.dont_write_bytecode = True class GuardTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="guard-test-") self.addCleanup(self.temp.cleanup) self.root = Path(self.temp.name) source = Path(__file__).with_name("pre_tool_use_guard.py") spec = importlib.util.spec_from_file_location("guard_under_test", source) self.guard = importlib.util.module_from_spec(spec) with patch.object(Path, "home", return_value=self.root): spec.loader.exec_module(self.guard) self.secret = self.root / ".ssh" / "id_ed25519" self.secret.parent.mkdir() self.secret.write_text("synthetic fixture, no actual key") def payload(self, command=None, **changes): data = {"session_id": "test-session", "tool_name": "Bash", "cwd": str(self.root), "tool_input": {"command": command}} data.update(changes) return data def run_guard(self, data, host="codex"): output = io.StringIO() with contextlib.redirect_stdout(output): self.guard.evaluate(data, host=host) return json.loads(output.getvalue()) if output.getvalue() else {} def test_literals_are_not_file_reads(self): for command in [f"printf '%s' '{self.secret}'", f"echo '{self.secret}'", f"rg '{self.secret}' ./README.md"]: self.assertEqual(self.run_guard(self.payload(command)), {}) def test_patch_body_is_not_shell(self): command = ("*** Begin Patch\n*** Add File: article.md\n" f"+cat {self.secret}\n*** End Patch") self.assertEqual(self.run_guard(self.payload(command, tool_name="apply_patch")), {}) def test_patch_vault_write_still_denied(self): command = f"*** Begin Patch\n*** Add File: {self.secret}\n+text\n*** End Patch" result = self.run_guard(self.payload(command, tool_name="apply_patch")) self.assertEqual(result["hookSpecificOutput"]["permissionDecision"], "deny") self.assertNotIn("request_id", result["hookSpecificOutput"]["permissionDecisionReason"]) def test_read_requires_approval_and_stores_no_command(self): result = self.run_guard(self.payload(f"cat {self.secret}")) decision = result["hookSpecificOutput"] self.assertEqual(decision["permissionDecision"], "deny") self.assertIn("request_id=", decision["permissionDecisionReason"]) db = self.guard._approval_database() with db: row = db.execute("SELECT * FROM requests").fetchone() self.assertNotIn(str(self.secret), str(tuple(row))) db.close() def request_id(self, payload): import re result = self.run_guard(payload) return re.search(r"request_id=([a-f0-9]{32})", result["hookSpecificOutput"]["permissionDecisionReason"])[1] def test_approval_is_consumed_once(self): data = self.payload(f"cat {self.secret}") request_id = self.request_id(data) self.guard.approve_request(request_id) self.assertEqual(self.run_guard(data), {}) self.assertIn("request_id=", self.run_guard(data)["hookSpecificOutput"]["permissionDecisionReason"]) with self.assertRaises(ValueError): self.guard.approve_request(request_id) def test_changed_input_and_other_session_cannot_consume(self): data = self.payload(f"cat {self.secret}") self.guard.approve_request(self.request_id(data)) for changed in [self.payload(f"head {self.secret}"), self.payload(f"cat {self.secret}", session_id="other-session"), self.payload(f"cat {self.secret}", cwd=str(self.root / "other"))]: self.assertEqual(self.run_guard(changed)["hookSpecificOutput"]["permissionDecision"], "deny") self.assertEqual(self.run_guard(data), {}) def test_expired_approval_cannot_be_consumed(self): data = self.payload(f"cat {self.secret}") self.guard.approve_request(self.request_id(data)) with patch.object(self.guard.time, "time", return_value=self.guard.time.time() + 901): self.assertEqual(self.run_guard(data)["hookSpecificOutput"]["permissionDecision"], "deny") def test_missing_session_does_not_create_ticket(self): data = self.payload(f"cat {self.secret}", session_id=None) decision = self.run_guard(data)["hookSpecificOutput"] self.assertEqual(decision["permissionDecision"], "deny") self.assertNotIn("request_id=", decision["permissionDecisionReason"]) def test_claude_uses_native_ask(self): decision = self.run_guard(self.payload(f"cat {self.secret}"), "claude")["hookSpecificOutput"] self.assertEqual(decision["permissionDecision"], "ask") def test_read_redirection_and_copy_source_require_confirmation(self): for command in [f"wc -c < {self.secret}", f"cp {self.secret} ./copy"]: data = self.payload(tool_input={"command": command, "workdir": str(self.root)}) self.assertIn("request_id=", self.run_guard(data)["hookSpecificOutput"]["permissionDecisionReason"]) def test_public_key_remains_allowed(self): public = self.secret.with_suffix(".pub") public.write_text("public fixture") self.assertEqual(self.run_guard(self.payload(f"cat {public}")), {}) def test_write_and_delete_are_not_approval_requests(self): for command in [f"echo text > {self.secret}", f"rm -rf {self.secret}"]: reason = self.run_guard(self.payload(command))["hookSpecificOutput"]["permissionDecisionReason"] self.assertNotIn("request_id=", reason) def test_file_tool_sensitive_read(self): data = self.payload(tool_name="Read", tool_input={"file_path": str(self.secret)}) self.assertIn("request_id=", self.run_guard(data)["hookSpecificOutput"]["permissionDecisionReason"]) def test_broken_input_is_rejected(self): for data in [[], {"tool_input": "broken"}]: self.assertEqual(self.run_guard(data)["hookSpecificOutput"]["permissionDecision"], "deny") def test_literals_and_quoted_heredoc_do_not_become_operations(self): for command in [f"printf '%s' tee {self.secret.parent / 'config'}", f"printf '%s' '>' {self.secret.parent / 'config'}", f"cat <<'EOF' > article.md\ncat {self.secret}\nEOF\n"]: self.assertEqual(self.run_guard(self.payload(command)), {}) def test_all_destructive_effects_precede_approval(self): for command in [f"mv {self.secret} ./backup", f"cat {self.secret}; rm {self.secret.parent / 'config'}"]: reason = self.run_guard(self.payload(command))["hookSpecificOutput"]["permissionDecisionReason"] self.assertNotIn("request_id=", reason) for patch_text in [f"*** Delete File: {self.root / '.zshrc'}", f"*** Update File: {self.secret}\n*** Move to: backup"]: result = self.run_guard(self.payload("*** Begin Patch\n" + patch_text + "\n*** End Patch", tool_name="apply_patch")) self.assertNotIn("request_id=", result["hookSpecificOutput"]["permissionDecisionReason"]) def test_effective_tool_workdir_is_bound_and_checked(self): data = self.payload(tool_input={"command": "cat id_ed25519", "workdir": str(self.secret.parent)}) self.assertIn("request_id=", self.run_guard(data)["hookSpecificOutput"]["permissionDecisionReason"]) def test_late_approval_does_not_extend_request_lifetime(self): now = self.guard.time.time() data = self.payload(f"cat {self.secret}") with patch.object(self.guard.time, "time", return_value=now): request_id = self.request_id(data) with patch.object(self.guard.time, "time", return_value=now + 890): self.guard.approve_request(request_id) with patch.object(self.guard.time, "time", return_value=now + 901): self.assertIn("request_id=", self.run_guard(data)["hookSpecificOutput"]["permissionDecisionReason"]) def test_state_failure_explicitly_denies(self): with patch.object(self.guard, "_approval_database", side_effect=OSError("synthetic failure")): decision = self.run_guard(self.payload(f"cat {self.secret}"))["hookSpecificOutput"] self.assertEqual(decision["permissionDecision"], "deny") self.assertNotIn("request_id=", decision["permissionDecisionReason"]) def test_concurrent_approval_and_consumption_cannot_resurrect_ticket(self): from concurrent.futures import ThreadPoolExecutor data = self.payload(f"cat {self.secret}") request_id = self.request_id(data) def approve(_): try: self.guard.approve_request(request_id) return True except ValueError: return False with ThreadPoolExecutor(max_workers=8) as pool: self.assertEqual(sum(pool.map(approve, range(12))), 1) with patch.object(self.guard, "_deny", return_value=99): with ThreadPoolExecutor(max_workers=8) as pool: results = list(pool.map(lambda _: self.guard._read_permission(data, str(self.root), "synthetic read", "codex"), range(12))) self.assertEqual(results.count(0), 1) self.assertEqual(results.count(99), 11) with self.assertRaises(ValueError): self.guard.approve_request(request_id) def test_substitutions_and_nested_shell_are_inspected(self): for command in [f"printf '%s' \"$(cat {self.secret})\"", f"bash -c 'cat {self.secret}'"]: data = self.payload(tool_input={"command": command, "workdir": str(self.root)}) self.assertIn("request_id=", self.run_guard(data)["hookSpecificOutput"]["permissionDecisionReason"]) def test_missing_host_workdir_requires_absolute_simple_read(self): data = self.payload("cat id_ed25519", cwd=str(self.secret.parent)) decision = self.run_guard(data)["hookSpecificOutput"] self.assertEqual(decision["permissionDecision"], "deny") self.assertNotIn("request_id=", decision["permissionDecisionReason"]) self.assertIn("request_id=", self.run_guard(self.payload(f"cat {self.secret}"))["hookSpecificOutput"]["permissionDecisionReason"]) def test_reader_options_do_not_hide_inputs_or_outputs(self): source = self.root / "input.txt" source.write_text("synthetic input\n") for command in [f"cat -n {self.secret}", f"cat -e {self.secret}", f"head -n 2 {self.secret}", f"grep -nf{self.secret} {source}"]: with self.subTest(command=command): result = self.run_guard(self.payload(command))["hookSpecificOutput"] self.assertEqual(result["permissionDecision"], "deny") self.assertIn("request_id=", result["permissionDecisionReason"]) for output_option in [f"-o {self.secret}", f"-o{self.secret}", f"--output {self.secret}", f"--output={self.secret}"]: with self.subTest(output_option=output_option): result = self.run_guard(self.payload(f"sort {output_option} {source}"))["hookSpecificOutput"] self.assertEqual(result["permissionDecision"], "deny") self.assertNotIn("request_id=", result["permissionDecisionReason"]) for command in [f"uniq {source} {self.secret}", f"xxd {source} {self.secret}", f"less -o {self.secret} {source}", f"base64 -o {self.secret} {source}", f"sed --in-place 's/a/b/' {self.secret}"]: with self.subTest(command=command): result = self.run_guard(self.payload(command))["hookSpecificOutput"] self.assertEqual(result["permissionDecision"], "deny") self.assertNotIn("request_id=", result["permissionDecisionReason"]) result = self.run_guard(self.payload(f"sort -o {source} {self.secret}"))["hookSpecificOutput"] self.assertNotIn("request_id=", result["permissionDecisionReason"], "missing workdir must not treat a reader with output as a pure read") def test_compound_redirections_keep_write_priority(self): for command in [f"(printf '%s' fixture) > {self.secret}", f"{{ cat {self.secret}; }} > {self.secret.parent / 'config'}"]: with self.subTest(command=command): result = self.run_guard(self.payload(command))["hookSpecificOutput"] self.assertEqual(result["permissionDecision"], "deny") self.assertNotIn("request_id=", result["permissionDecisionReason"]) def test_static_option_roles_do_not_bypass_guard(self): source = self.root / "input.txt" source.write_text("synthetic input\n") cases = [(f"rg -e --files -e '^' {self.secret}", True)] for name in ("cp", "mv", "install", "ln"): for target in (f"-t{self.secret.parent}", f"-vt{self.secret.parent}", f"-t {self.secret.parent}", f"--target-directory={self.secret.parent}"): cases.append((f"{name} {target} {source}", False)) cases.extend([(f"sort --out={self.secret} {source}", False), (f"sort --out={self.secret.parent / 'config'} {self.secret}", False), (f"cp --target-dir={self.secret.parent} {source}", False)]) for command, should_request in cases: for host in ("codex", "claude"): with self.subTest(command=command, host=host): decision = self.run_guard(self.payload(command), host)["hookSpecificOutput"] self.assertEqual(decision["permissionDecision"], "ask" if host == "claude" and should_request else "deny") self.assertEqual("request_id=" in decision["permissionDecisionReason"], host == "codex" and should_request) self.assertEqual(self.run_guard(self.payload(f"rg --files {self.secret.parent}")), {}) self.assertIn("request_id=", self.run_guard(self.payload(f"rg -- --files {self.secret}"))["hookSpecificOutput"]["permissionDecisionReason"]) def test_pure_read_ticket_uses_parsed_roles(self): for command in [f"rg -e '.*' {self.secret}", f"rg -e --pre -e '^' {self.secret}"]: self.assertIn("request_id=", self.run_guard(self.payload(command))["hookSpecificOutput"]["permissionDecisionReason"]) for command in [f"rg --pre cat '^' {self.secret}", f"sed 'p' {self.secret}", f"cat {self.secret.parent}/*", f"yq --split-exp expr {self.secret}"]: decision = self.run_guard(self.payload(command))["hookSpecificOutput"] self.assertEqual(decision["permissionDecision"], "deny") self.assertNotIn("request_id=", decision["permissionDecisionReason"]) if __name__ == "__main__": unittest.main()