fix Codex hook trust detection
This commit is contained in:
thanhnv
@@ -0,0 +1,196 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import contextlib
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
|
||||
|
||||
def module(name: str, path: Path):
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
value = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(value)
|
||||
return value
|
||||
|
||||
|
||||
INIT = module("casan_init_test", ROOT / "packages/casan-devkit/casan-init.py")
|
||||
|
||||
|
||||
def codex_hooks():
|
||||
hooks = {}
|
||||
for event in ("UserPromptSubmit", "PreToolUse", "PostToolUse", "Stop"):
|
||||
hooks[event] = [{
|
||||
"hooks": [{
|
||||
"type": "command",
|
||||
"command": (
|
||||
'python3 ".casan/casan-hook.py" '
|
||||
f"--client codex --event {event}"
|
||||
),
|
||||
"timeout": 15,
|
||||
}],
|
||||
}]
|
||||
return {"hooks": hooks}
|
||||
|
||||
|
||||
class FakeProcess:
|
||||
def __init__(self, response):
|
||||
self.stdin = io.StringIO()
|
||||
self.stdout = io.StringIO(json.dumps(response) + "\n")
|
||||
self.returncode = None
|
||||
|
||||
def poll(self):
|
||||
return self.returncode
|
||||
|
||||
def terminate(self):
|
||||
self.returncode = 0
|
||||
|
||||
def wait(self, timeout=None):
|
||||
self.returncode = 0
|
||||
return 0
|
||||
|
||||
def kill(self):
|
||||
self.returncode = -9
|
||||
|
||||
|
||||
class CasanInitTests(unittest.TestCase):
|
||||
def trust_response(self, target: str, statuses):
|
||||
source = str(Path(target, ".codex", "hooks.json").resolve())
|
||||
hooks = []
|
||||
for index, (event, status) in enumerate(zip(
|
||||
("user_prompt_submit", "pre_tool_use", "post_tool_use", "stop"),
|
||||
statuses,
|
||||
)):
|
||||
hooks.append({
|
||||
"key": f"{source}:{event}:0:0",
|
||||
"sourcePath": source,
|
||||
"command": (
|
||||
'python3 ".casan/casan-hook.py" '
|
||||
f"--client codex --event {event}"
|
||||
),
|
||||
"enabled": True,
|
||||
"trustStatus": status,
|
||||
"displayOrder": index,
|
||||
})
|
||||
return {
|
||||
"id": 2,
|
||||
"result": {
|
||||
"data": [{
|
||||
"cwd": target,
|
||||
"hooks": hooks,
|
||||
"warnings": [],
|
||||
"errors": [],
|
||||
}],
|
||||
},
|
||||
}
|
||||
|
||||
def test_codex_trust_uses_effective_app_server_state(self):
|
||||
with tempfile.TemporaryDirectory() as target:
|
||||
hooks_path = Path(target, ".codex", "hooks.json")
|
||||
hooks_path.parent.mkdir()
|
||||
hooks_path.write_text(json.dumps(codex_hooks()), encoding="utf-8")
|
||||
response = self.trust_response(target, ["trusted"] * 4)
|
||||
process = FakeProcess(response)
|
||||
|
||||
with patch.object(INIT.shutil, "which", return_value="/bin/codex"), \
|
||||
patch.object(INIT.subprocess, "Popen", return_value=process):
|
||||
result = INIT._codex_project_hook_trust(target)
|
||||
|
||||
self.assertTrue(result["verified"])
|
||||
self.assertEqual(result["status"], "trusted")
|
||||
self.assertEqual(result["trusted_hooks"], 4)
|
||||
self.assertFalse(result["review_required"])
|
||||
self.assertTrue(result["operational"])
|
||||
requests = [
|
||||
json.loads(line) for line in process.stdin.getvalue().splitlines()
|
||||
]
|
||||
self.assertEqual(requests[-1]["method"], "hooks/list")
|
||||
self.assertEqual(requests[-1]["params"]["cwds"], [target])
|
||||
|
||||
def test_codex_trust_detects_modified_hook(self):
|
||||
with tempfile.TemporaryDirectory() as target:
|
||||
hooks_path = Path(target, ".codex", "hooks.json")
|
||||
hooks_path.parent.mkdir()
|
||||
hooks_path.write_text(json.dumps(codex_hooks()), encoding="utf-8")
|
||||
response = self.trust_response(
|
||||
target, ["trusted", "modified", "trusted", "trusted"])
|
||||
|
||||
with patch.object(INIT.shutil, "which", return_value="/bin/codex"), \
|
||||
patch.object(
|
||||
INIT.subprocess, "Popen",
|
||||
return_value=FakeProcess(response),
|
||||
):
|
||||
result = INIT._codex_project_hook_trust(target)
|
||||
|
||||
self.assertTrue(result["verified"])
|
||||
self.assertEqual(result["status"], "modified")
|
||||
self.assertEqual(result["modified_hooks"], 1)
|
||||
self.assertTrue(result["review_required"])
|
||||
self.assertFalse(result["operational"])
|
||||
|
||||
def test_doctor_does_not_warn_when_codex_hooks_are_trusted(self):
|
||||
with tempfile.TemporaryDirectory() as target:
|
||||
root = Path(target)
|
||||
(root / ".casan").mkdir()
|
||||
(root / ".codex").mkdir()
|
||||
(root / ".casan/config.json").write_text(json.dumps({
|
||||
"project_id": "trusted-project",
|
||||
"enforcement_mode": "enforce",
|
||||
"integration_mode": "project_hook",
|
||||
"clients": ["codex"],
|
||||
}), encoding="utf-8")
|
||||
(root / ".casan/version.lock").write_text(
|
||||
json.dumps({"harness_hash": "sha256:harness"}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
(root / ".casan/casan-hook.py").write_text(
|
||||
"# test bootstrap\n", encoding="utf-8")
|
||||
(root / ".codex/hooks.json").write_text(
|
||||
json.dumps(codex_hooks()), encoding="utf-8")
|
||||
trusted = {
|
||||
"verified": True,
|
||||
"review_required": False,
|
||||
"operational": True,
|
||||
"status": "trusted",
|
||||
"disabled_hooks": 0,
|
||||
}
|
||||
args = argparse.Namespace(
|
||||
target=target, harness=None, client=None, json=True)
|
||||
|
||||
with patch.object(
|
||||
INIT, "resolve_project_harness", return_value="/harness"
|
||||
), patch.object(
|
||||
INIT, "compute_live",
|
||||
return_value=("sha256:harness", "live"),
|
||||
), patch.object(
|
||||
INIT, "_legacy_instruction_conflicts", return_value=[]
|
||||
), patch.object(
|
||||
INIT, "_smoke_bootstrap",
|
||||
return_value={"ok": True, "decision": "allow"},
|
||||
), patch.object(
|
||||
INIT, "_codex_project_hook_trust", return_value=trusted
|
||||
), patch.object(
|
||||
INIT.shutil, "which", return_value=None
|
||||
), contextlib.redirect_stdout(io.StringIO()) as output:
|
||||
returncode = INIT.cmd_doctor(args)
|
||||
|
||||
result = json.loads(output.getvalue())
|
||||
self.assertEqual(returncode, 0)
|
||||
self.assertEqual(result["status"], "ready")
|
||||
self.assertEqual(result["operational_status"], "ready")
|
||||
self.assertEqual(result["warnings"], [])
|
||||
self.assertEqual(result["required_actions"], [])
|
||||
self.assertFalse(
|
||||
result["client_checks"]["codex"]["trust_review_required"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user