fix(python): sync transactional Codex hooks
This commit is contained in:
ZacharyZhang-NY committed
1 parent
f501df5077
commit
1d2e08de97
16 files changed
+2306
-870
No files matched your search
@@ -0,0 +1,233 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess # nosec B404
|
||||
import sys
|
||||
from threading import Thread
|
||||
from typing import Any, Mapping
|
||||
|
||||
import pytest
|
||||
|
||||
from elydora.plugins.codex import CodexPlugin
|
||||
from elydora.utils import base64url_encode
|
||||
|
||||
|
||||
AGENT_ID = "agent-1"
|
||||
MISSING = object()
|
||||
VALID_PRIVATE_KEY = base64url_encode(bytes(range(32)))
|
||||
|
||||
|
||||
@dataclass
|
||||
class CodexFixture:
|
||||
root_dir: Path
|
||||
home_dir: Path
|
||||
project_dir: Path
|
||||
config_path: Path
|
||||
agent_dir: Path
|
||||
guard_path: Path
|
||||
hook_path: Path
|
||||
runtime_config_path: Path
|
||||
private_key_path: Path
|
||||
plugin: CodexPlugin
|
||||
config: dict[str, str]
|
||||
|
||||
def install(self) -> None:
|
||||
self.plugin.install(self.config)
|
||||
|
||||
def settings(self, path: Path | None = None) -> dict[str, Any]:
|
||||
return json.loads((path or self.config_path).read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
class ElydoraApiHandler(BaseHTTPRequestHandler):
|
||||
agent_status: Any = "active"
|
||||
get_status = 200
|
||||
post_status = 201
|
||||
requests: list[dict[str, Any]] = []
|
||||
|
||||
@classmethod
|
||||
def reset(cls) -> None:
|
||||
cls.agent_status = "active"
|
||||
cls.get_status = 200
|
||||
cls.post_status = 201
|
||||
cls.requests = []
|
||||
|
||||
def _respond(self, status: int, value: object) -> None:
|
||||
body = json.dumps(value).encode()
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_GET(self) -> None:
|
||||
type(self).requests.append(
|
||||
{
|
||||
"method": "GET",
|
||||
"path": self.path,
|
||||
"authorization": self.headers.get("Authorization"),
|
||||
}
|
||||
)
|
||||
self._respond(
|
||||
type(self).get_status,
|
||||
{"agent": {"status": type(self).agent_status}},
|
||||
)
|
||||
|
||||
def do_POST(self) -> None:
|
||||
length = int(self.headers.get("Content-Length", "0"))
|
||||
raw = self.rfile.read(length)
|
||||
type(self).requests.append(
|
||||
{
|
||||
"method": "POST",
|
||||
"path": self.path,
|
||||
"authorization": self.headers.get("Authorization"),
|
||||
"raw": raw,
|
||||
"json": json.loads(raw),
|
||||
}
|
||||
)
|
||||
self._respond(type(self).post_status, {"operation": {"accepted": True}})
|
||||
|
||||
def log_message(self, _format: str, *_args: object) -> None:
|
||||
return
|
||||
|
||||
|
||||
def start_api_server() -> ThreadingHTTPServer:
|
||||
ElydoraApiHandler.reset()
|
||||
server = ThreadingHTTPServer(("127.0.0.1", 0), ElydoraApiHandler)
|
||||
Thread(target=server.serve_forever, daemon=True).start()
|
||||
return server
|
||||
|
||||
|
||||
def server_base_url(server: ThreadingHTTPServer) -> str:
|
||||
return f"http://127.0.0.1:{server.server_port}"
|
||||
|
||||
|
||||
def write_text(path: Path, source: str, mode: int = 0o600) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(source, encoding="utf-8")
|
||||
os.chmod(path, mode)
|
||||
|
||||
|
||||
def write_json(path: Path, value: object, mode: int = 0o600) -> None:
|
||||
source = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n"
|
||||
write_text(path, source, mode)
|
||||
|
||||
|
||||
def prepare_fixture(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
*,
|
||||
existing_config: object = MISSING,
|
||||
base_url: str = "http://127.0.0.1:9",
|
||||
) -> CodexFixture:
|
||||
home_dir = tmp_path / "home with spaces and 'quote %ELYDORA_HOOK_PATH%"
|
||||
project_dir = tmp_path / "project with spaces"
|
||||
config_path = home_dir / ".codex" / "hooks.json"
|
||||
agent_dir = home_dir / ".elydora" / AGENT_ID
|
||||
guard_path = agent_dir / "guard.py"
|
||||
hook_path = agent_dir / "hook.py"
|
||||
runtime_config_path = agent_dir / "config.json"
|
||||
private_key_path = agent_dir / "private.key"
|
||||
monkeypatch.setenv("HOME", str(home_dir))
|
||||
monkeypatch.setenv("USERPROFILE", str(home_dir))
|
||||
monkeypatch.setenv("CODEX_HOME", "")
|
||||
monkeypatch.setenv("ELYDORA_HOOK_PATH", str(tmp_path / "expanded"))
|
||||
project_dir.mkdir(parents=True)
|
||||
agent_dir.mkdir(parents=True)
|
||||
if existing_config is not MISSING:
|
||||
write_json(config_path, existing_config)
|
||||
return CodexFixture(
|
||||
root_dir=tmp_path,
|
||||
home_dir=home_dir,
|
||||
project_dir=project_dir,
|
||||
config_path=config_path,
|
||||
agent_dir=agent_dir,
|
||||
guard_path=guard_path,
|
||||
hook_path=hook_path,
|
||||
runtime_config_path=runtime_config_path,
|
||||
private_key_path=private_key_path,
|
||||
plugin=CodexPlugin(),
|
||||
config={
|
||||
"org_id": "org-1",
|
||||
"agent_id": AGENT_ID,
|
||||
"agent_name": "codex",
|
||||
"private_key": VALID_PRIVATE_KEY,
|
||||
"kid": "key-1",
|
||||
"token": "token-1",
|
||||
"base_url": base_url,
|
||||
"guard_script_path": str(guard_path),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def managed_handler(
|
||||
settings: Mapping[str, Any],
|
||||
event: str,
|
||||
status_message: str,
|
||||
) -> dict[str, Any]:
|
||||
for group in settings["hooks"][event]:
|
||||
for handler in group["hooks"]:
|
||||
if handler.get("statusMessage") == status_message:
|
||||
return handler
|
||||
raise AssertionError(f"handler {status_message!r} not found")
|
||||
|
||||
|
||||
def assert_native_handler(handler: Mapping[str, Any], status_message: str) -> None:
|
||||
assert set(handler) == {
|
||||
"type",
|
||||
"command",
|
||||
"commandWindows",
|
||||
"timeout",
|
||||
"statusMessage",
|
||||
}
|
||||
assert handler["type"] == "command"
|
||||
assert handler["timeout"] == 10
|
||||
assert handler["statusMessage"] == status_message
|
||||
assert str(handler["command"]).startswith("'")
|
||||
assert str(handler["commandWindows"]).startswith('"')
|
||||
assert " -EncodedCommand " in str(handler["commandWindows"])
|
||||
|
||||
|
||||
def run_handler(
|
||||
handler: Mapping[str, Any],
|
||||
payload: bytes,
|
||||
fixture: CodexFixture,
|
||||
) -> subprocess.CompletedProcess[bytes]:
|
||||
command_key = "commandWindows" if os.name == "nt" else "command"
|
||||
return subprocess.run(
|
||||
str(handler[command_key]),
|
||||
shell=True,
|
||||
input=payload,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
cwd=fixture.project_dir,
|
||||
env={
|
||||
**os.environ,
|
||||
"HOME": str(fixture.home_dir),
|
||||
"USERPROFILE": str(fixture.home_dir),
|
||||
"CODEX_HOME": "",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def legacy_handler(script_path: Path, status_message: str) -> dict[str, Any]:
|
||||
import shlex
|
||||
|
||||
arguments = [sys.executable, str(script_path)]
|
||||
return {
|
||||
"type": "command",
|
||||
"command": f"{shlex.quote(arguments[0])} {shlex.quote(arguments[1])}",
|
||||
"commandWindows": subprocess.list2cmdline(arguments),
|
||||
"timeout": 10,
|
||||
"statusMessage": status_message,
|
||||
}
|
||||
|
||||
|
||||
def symlink_or_skip(target: Path, link: Path, *, directory: bool = False) -> None:
|
||||
try:
|
||||
link.symlink_to(target, target_is_directory=directory)
|
||||
except OSError as error:
|
||||
pytest.skip(f"Symbolic links unavailable: {error}")
|
||||
@@ -1,125 +1,31 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from elydora.plugins import codex
|
||||
from elydora.plugins.base import InstallConfig
|
||||
from codex_support import (
|
||||
AGENT_ID,
|
||||
VALID_PRIVATE_KEY,
|
||||
assert_native_handler,
|
||||
legacy_handler,
|
||||
managed_handler,
|
||||
prepare_fixture,
|
||||
symlink_or_skip,
|
||||
write_json,
|
||||
write_text,
|
||||
)
|
||||
from elydora.plugins.codex_contract import (
|
||||
AUDIT_STATUS,
|
||||
GUARD_STATUS,
|
||||
build_handler,
|
||||
)
|
||||
from elydora.plugins.registry import SUPPORTED_AGENTS
|
||||
|
||||
|
||||
AGENT_ID = "agent-1"
|
||||
MISSING = object()
|
||||
GUARD_STATUS = "Checking Elydora agent state"
|
||||
AUDIT_STATUS = "Recording Elydora tool use"
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CodexFixture:
|
||||
plugin: codex.CodexPlugin
|
||||
config: InstallConfig
|
||||
home_dir: Path
|
||||
agent_dir: Path
|
||||
config_path: Path
|
||||
guard_path: Path
|
||||
hook_path: Path
|
||||
runtime_config_path: Path
|
||||
private_key_path: Path
|
||||
|
||||
|
||||
def write_json_or_text(path: Path, value: Any) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
content = value if isinstance(value, str) else json.dumps(value, indent=2)
|
||||
path.write_text(content, encoding="utf-8")
|
||||
|
||||
|
||||
def prepare_fixture(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
*,
|
||||
existing_settings: Any = MISSING,
|
||||
create_guard: bool = True,
|
||||
) -> CodexFixture:
|
||||
home_dir = tmp_path / "home with spaces"
|
||||
elydora_dir = home_dir / ".elydora"
|
||||
agent_dir = elydora_dir / AGENT_ID
|
||||
config_path = home_dir / ".codex" / "hooks.json"
|
||||
agent_dir.mkdir(parents=True)
|
||||
|
||||
guard_path = agent_dir / "guard.py"
|
||||
if create_guard:
|
||||
guard_path.write_text(
|
||||
"import sys\nsys.stderr.write('Agent is frozen by Elydora.')\n"
|
||||
"raise SystemExit(2)\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
if existing_settings is not MISSING:
|
||||
write_json_or_text(config_path, existing_settings)
|
||||
|
||||
monkeypatch.setattr(codex, "ELYDORA_DIR", str(elydora_dir))
|
||||
monkeypatch.setattr(codex, "CONFIG_PATH", str(config_path))
|
||||
config: InstallConfig = {
|
||||
"agent_id": AGENT_ID,
|
||||
"agent_name": "codex",
|
||||
"org_id": "org-1",
|
||||
"private_key": "test-key",
|
||||
"kid": "kid-1",
|
||||
"base_url": "https://api.elydora.test",
|
||||
"guard_script_path": str(guard_path),
|
||||
}
|
||||
return CodexFixture(
|
||||
plugin=codex.CodexPlugin(),
|
||||
config=config,
|
||||
home_dir=home_dir,
|
||||
agent_dir=agent_dir,
|
||||
config_path=config_path,
|
||||
guard_path=guard_path,
|
||||
hook_path=agent_dir / "hook.py",
|
||||
runtime_config_path=agent_dir / "config.json",
|
||||
private_key_path=agent_dir / "private.key",
|
||||
)
|
||||
|
||||
|
||||
def find_handler(
|
||||
settings: dict[str, Any],
|
||||
event: str,
|
||||
status_message: str,
|
||||
) -> dict[str, Any]:
|
||||
for group in settings["hooks"][event]:
|
||||
for handler in group["hooks"]:
|
||||
if handler.get("statusMessage") == status_message:
|
||||
return handler
|
||||
raise AssertionError(f"handler {status_message!r} not found")
|
||||
|
||||
|
||||
def run_command(
|
||||
command: str,
|
||||
home_dir: Path,
|
||||
payload: dict[str, Any],
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
env = {
|
||||
**os.environ,
|
||||
"HOME": str(home_dir),
|
||||
"USERPROFILE": str(home_dir),
|
||||
}
|
||||
return subprocess.run(
|
||||
command,
|
||||
shell=True,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
env=env,
|
||||
input=json.dumps(payload),
|
||||
text=True,
|
||||
)
|
||||
|
||||
|
||||
def test_codex_registry_points_at_global_hooks_contract() -> None:
|
||||
def test_codex_registry_points_at_native_user_hooks() -> None:
|
||||
assert SUPPORTED_AGENTS["codex"] == {
|
||||
"name": "OpenAI Codex",
|
||||
"hook_event": "PreToolUse/PostToolUse",
|
||||
@@ -127,285 +33,291 @@ def test_codex_registry_points_at_global_hooks_contract() -> None:
|
||||
}
|
||||
|
||||
|
||||
def test_install_preserves_existing_hooks_and_is_idempotent(
|
||||
def test_codex_follows_and_canonicalizes_codex_home(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
target = tmp_path / "custom Codex home"
|
||||
configured = tmp_path / "codex-home-link"
|
||||
target.mkdir()
|
||||
try:
|
||||
configured.symlink_to(target, target_is_directory=True)
|
||||
except OSError:
|
||||
configured = target
|
||||
monkeypatch.setenv("CODEX_HOME", str(configured))
|
||||
|
||||
fixture.install()
|
||||
|
||||
expected = Path(os.path.realpath(target)) / "hooks.json"
|
||||
assert expected.is_file()
|
||||
assert fixture.config_path.exists() is False
|
||||
status = fixture.plugin.status()
|
||||
assert status["installed"] is True
|
||||
assert status["details"] == f"Config: {expected}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kind", ["missing", "file"])
|
||||
def test_codex_rejects_invalid_codex_home_before_writes(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
kind: str,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
configured = tmp_path / f"{kind}-codex-home"
|
||||
if kind == "file":
|
||||
write_text(configured, "not a directory")
|
||||
|
||||
monkeypatch.setenv("CODEX_HOME", str(configured))
|
||||
with pytest.raises(OSError, match="CODEX_HOME"):
|
||||
fixture.install()
|
||||
|
||||
assert list(fixture.agent_dir.iterdir()) == []
|
||||
assert fixture.config_path.exists() is False
|
||||
|
||||
|
||||
def test_install_preserves_sources_migrates_legacy_and_is_idempotent(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
capsys: pytest.CaptureFixture[str],
|
||||
) -> None:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
existing_settings={
|
||||
"description": "Workspace hooks",
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
write_json(
|
||||
fixture.config_path,
|
||||
{
|
||||
"description": "User hooks",
|
||||
"metadata": {"owner": "user"},
|
||||
"hooks": {
|
||||
"SessionStart": [{
|
||||
"hooks": [{"type": "command", "command": "existing-command"}],
|
||||
}],
|
||||
"SessionStart": [
|
||||
{"matcher": "*", "hooks": [{"type": "command", "command": "keep"}]}
|
||||
],
|
||||
"PreToolUse": [
|
||||
{
|
||||
"matcher": "*",
|
||||
"hooks": [legacy_handler(fixture.guard_path, GUARD_STATUS)],
|
||||
}
|
||||
],
|
||||
"PostToolUse": [
|
||||
{
|
||||
"matcher": "*",
|
||||
"hooks": [legacy_handler(fixture.hook_path, AUDIT_STATUS)],
|
||||
}
|
||||
],
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
fixture.plugin.install(fixture.config)
|
||||
fixture.plugin.install(fixture.config)
|
||||
fixture.install()
|
||||
first = fixture.config_path.read_text(encoding="utf-8")
|
||||
fixture.install()
|
||||
second = fixture.config_path.read_text(encoding="utf-8")
|
||||
|
||||
assert "run /hooks to review and trust" in capsys.readouterr().out
|
||||
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
|
||||
assert settings["description"] == "Workspace hooks"
|
||||
assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "existing-command"
|
||||
assert first == second
|
||||
assert "approve both Elydora command hooks" in capsys.readouterr().out
|
||||
settings = fixture.settings()
|
||||
assert settings["description"] == "User hooks"
|
||||
assert settings["metadata"] == {"owner": "user"}
|
||||
assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "keep"
|
||||
assert len(settings["hooks"]["PreToolUse"]) == 1
|
||||
assert len(settings["hooks"]["PostToolUse"]) == 1
|
||||
assert settings["hooks"]["PreToolUse"][0]["matcher"] == "*"
|
||||
guard = find_handler(settings, "PreToolUse", GUARD_STATUS)
|
||||
assert guard["type"] == "command"
|
||||
assert guard["timeout"] == 10
|
||||
assert isinstance(guard["command"], str)
|
||||
assert isinstance(guard["commandWindows"], str)
|
||||
assert_native_handler(
|
||||
managed_handler(settings, "PreToolUse", GUARD_STATUS),
|
||||
GUARD_STATUS,
|
||||
)
|
||||
assert_native_handler(
|
||||
managed_handler(settings, "PostToolUse", AUDIT_STATUS),
|
||||
AUDIT_STATUS,
|
||||
)
|
||||
runtime = json.loads(fixture.runtime_config_path.read_text(encoding="utf-8"))
|
||||
assert runtime["agent_name"] == "codex"
|
||||
assert runtime["token"] == "token-1"
|
||||
assert fixture.private_key_path.read_text(encoding="utf-8") == VALID_PRIVATE_KEY
|
||||
assert fixture.guard_path.is_file()
|
||||
assert fixture.hook_path.is_file()
|
||||
|
||||
|
||||
def test_commands_block_frozen_agents_and_forward_official_payload(
|
||||
@pytest.mark.parametrize(
|
||||
("source", "error_pattern"),
|
||||
[
|
||||
('{"hooks":{},"hooks":{}}', "duplicate field"),
|
||||
('{"hooks":{},}', "parse Codex user hooks"),
|
||||
('{"hooks":{/*comment*/}}', "parse Codex user hooks"),
|
||||
("[]", "must contain a JSON object"),
|
||||
('{"hooks":null}', 'field "hooks" must be an object'),
|
||||
('{"hooks":{"PreToolUse":null}}', 'hooks.PreToolUse" must be an array'),
|
||||
(
|
||||
'{"hooks":{"PreToolUse":[{"hooks":null}]}}',
|
||||
"must contain a hooks array",
|
||||
),
|
||||
(
|
||||
'{"hooks":{"PreToolUse":[{"hooks":[null]}]}}',
|
||||
"must contain a hooks array",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_install_preserves_malformed_and_ambiguous_json(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
source: str,
|
||||
error_pattern: str,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path, existing_config=source)
|
||||
|
||||
with pytest.raises(ValueError, match=error_pattern):
|
||||
fixture.install()
|
||||
|
||||
assert fixture.config_path.read_text(encoding="utf-8") == source
|
||||
assert list(fixture.agent_dir.iterdir()) == []
|
||||
|
||||
|
||||
def test_status_requires_exact_pair_identity_and_runtime_secrets(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.plugin.install(fixture.config)
|
||||
capture_path = tmp_path / "captured-event.json"
|
||||
fixture.hook_path.write_text(
|
||||
"import pathlib, sys\n"
|
||||
f"pathlib.Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
|
||||
command_key = "commandWindows" if os.name == "nt" else "command"
|
||||
payload = {
|
||||
"hook_event_name": "PreToolUse",
|
||||
"session_id": "session-1",
|
||||
"turn_id": "turn-1",
|
||||
"transcript_path": None,
|
||||
"cwd": str(fixture.home_dir),
|
||||
"model": "gpt-5",
|
||||
"permission_mode": "default",
|
||||
"tool_name": "Bash",
|
||||
"tool_use_id": "call-1",
|
||||
"tool_input": {"command": "echo test"},
|
||||
}
|
||||
fixture.install()
|
||||
assert fixture.plugin.status()["installed"] is True
|
||||
|
||||
guard = find_handler(settings, "PreToolUse", GUARD_STATUS)
|
||||
guard_result = run_command(guard[command_key], fixture.home_dir, payload)
|
||||
assert guard_result.returncode == 2
|
||||
assert "Agent is frozen by Elydora" in guard_result.stderr
|
||||
fixture.private_key_path.unlink()
|
||||
status = fixture.plugin.status()
|
||||
assert status["installed"] is False
|
||||
assert "runtime files missing" in status["details"]
|
||||
|
||||
payload["hook_event_name"] = "PostToolUse"
|
||||
payload["tool_response"] = {"output": "test"}
|
||||
audit = find_handler(settings, "PostToolUse", AUDIT_STATUS)
|
||||
audit_result = run_command(audit[command_key], fixture.home_dir, payload)
|
||||
assert audit_result.returncode == 0
|
||||
assert json.loads(capture_path.read_text(encoding="utf-8")) == payload
|
||||
fixture.install()
|
||||
settings = fixture.settings()
|
||||
guard = managed_handler(settings, "PreToolUse", GUARD_STATUS)
|
||||
guard["extra"] = True
|
||||
write_json(fixture.config_path, settings)
|
||||
assert fixture.plugin.status()["installed"] is False
|
||||
|
||||
fixture.install()
|
||||
runtime = json.loads(fixture.runtime_config_path.read_text(encoding="utf-8"))
|
||||
runtime["agent_id"] = "another-agent"
|
||||
write_json(fixture.runtime_config_path, runtime)
|
||||
assert fixture.plugin.status()["installed"] is False
|
||||
|
||||
def test_status_requires_both_runtimes_and_uninstall_preserves_other_hooks(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
existing_handler = {"type": "command", "command": "existing-command"}
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
existing_settings={
|
||||
"hooks": {"PreToolUse": [{"hooks": [existing_handler]}]},
|
||||
},
|
||||
)
|
||||
fixture.plugin.install(fixture.config)
|
||||
assert fixture.plugin.status() == {
|
||||
"installed": True,
|
||||
"agent": "codex",
|
||||
"details": f"Config: {fixture.config_path}",
|
||||
}
|
||||
|
||||
fixture.guard_path.unlink()
|
||||
assert fixture.plugin.status() == {
|
||||
"installed": False,
|
||||
"agent": "codex",
|
||||
"details": f"Configured at {fixture.config_path}; runtime scripts missing",
|
||||
}
|
||||
|
||||
fixture.plugin.uninstall(AGENT_ID)
|
||||
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
|
||||
assert settings["hooks"]["PreToolUse"] == [{"hooks": [existing_handler]}]
|
||||
assert settings["hooks"]["PostToolUse"] == []
|
||||
|
||||
|
||||
def test_install_preserves_user_handlers_with_matching_status_text(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
user_guard = {
|
||||
"type": "command",
|
||||
"command": "user-guard",
|
||||
"statusMessage": GUARD_STATUS,
|
||||
}
|
||||
user_audit = {
|
||||
"type": "command",
|
||||
"command": "user-audit",
|
||||
"statusMessage": AUDIT_STATUS,
|
||||
}
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
existing_settings={
|
||||
"hooks": {
|
||||
"PreToolUse": [{"matcher": "Bash", "hooks": [user_guard]}],
|
||||
"PostToolUse": [{"matcher": "Bash", "hooks": [user_audit]}],
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
fixture.plugin.install(fixture.config)
|
||||
|
||||
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
|
||||
assert settings["hooks"]["PreToolUse"][0]["hooks"][0] == user_guard
|
||||
assert settings["hooks"]["PostToolUse"][0]["hooks"][0] == user_audit
|
||||
assert len(settings["hooks"]["PreToolUse"]) == 2
|
||||
assert len(settings["hooks"]["PostToolUse"]) == 2
|
||||
|
||||
|
||||
def test_install_rejects_missing_guard_before_runtime_or_config_writes(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path, create_guard=False)
|
||||
|
||||
with pytest.raises(FileNotFoundError, match="guard runtime is missing"):
|
||||
fixture.plugin.install(fixture.config)
|
||||
|
||||
assert fixture.config_path.exists() is False
|
||||
assert fixture.hook_path.exists() is False
|
||||
assert fixture.runtime_config_path.exists() is False
|
||||
assert fixture.private_key_path.exists() is False
|
||||
|
||||
|
||||
def test_status_surfaces_malformed_referenced_runtime_metadata(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.plugin.install(fixture.config)
|
||||
fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8")
|
||||
|
||||
write_text(fixture.runtime_config_path, "{ malformed")
|
||||
with pytest.raises(ValueError, match="parse Elydora runtime config"):
|
||||
fixture.plugin.status()
|
||||
|
||||
|
||||
def test_status_surfaces_malformed_matcher_groups(
|
||||
def test_uninstall_removes_exact_ownership_and_preserves_lookalikes(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.plugin.install(fixture.config)
|
||||
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
|
||||
settings["hooks"]["PreToolUse"][0]["hooks"] = None
|
||||
fixture.config_path.write_text(json.dumps(settings), encoding="utf-8")
|
||||
fixture.install()
|
||||
settings = fixture.settings()
|
||||
agent_ten = fixture.agent_dir.parent / "agent-10"
|
||||
other_guard = build_handler(str(agent_ten / "guard.py"), GUARD_STATUS)
|
||||
other_audit = build_handler(str(agent_ten / "hook.py"), AUDIT_STATUS)
|
||||
lookalike = {
|
||||
"type": "command",
|
||||
"command": f"inspect {fixture.guard_path}",
|
||||
"commandWindows": f"inspect {fixture.guard_path}",
|
||||
"timeout": 10,
|
||||
"statusMessage": GUARD_STATUS,
|
||||
}
|
||||
managed_guard = dict(managed_handler(settings, "PreToolUse", GUARD_STATUS))
|
||||
settings["hooks"]["SessionStart"] = [
|
||||
{"matcher": "*", "hooks": [{"type": "command", "command": "keep"}]}
|
||||
]
|
||||
settings["hooks"]["PreToolUse"].extend(
|
||||
[
|
||||
{"matcher": "*", "hooks": [other_guard]},
|
||||
{"matcher": "*", "hooks": [lookalike]},
|
||||
{"matcher": "Bash", "hooks": [managed_guard]},
|
||||
]
|
||||
)
|
||||
settings["hooks"]["PostToolUse"].append({"matcher": "*", "hooks": [other_audit]})
|
||||
write_json(fixture.config_path, settings)
|
||||
|
||||
with pytest.raises(ValueError, match="matcher group must contain a hooks array"):
|
||||
fixture.plugin.status()
|
||||
fixture.plugin.uninstall(AGENT_ID)
|
||||
|
||||
remaining = fixture.settings()
|
||||
assert remaining["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "keep"
|
||||
assert len(remaining["hooks"]["PreToolUse"]) == 3
|
||||
assert len(remaining["hooks"]["PostToolUse"]) == 1
|
||||
assert remaining["hooks"]["PreToolUse"][1]["hooks"][0] == lookalike
|
||||
assert remaining["hooks"]["PreToolUse"][2] == {"matcher": "Bash", "hooks": []}
|
||||
|
||||
|
||||
def test_uninstall_removes_config_owned_entirely_by_elydora(
|
||||
def test_uninstall_removes_an_entirely_managed_file(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.plugin.install(fixture.config)
|
||||
fixture.install()
|
||||
|
||||
fixture.plugin.uninstall(AGENT_ID)
|
||||
fixture.plugin.uninstall(AGENT_ID)
|
||||
|
||||
assert fixture.config_path.exists() is False
|
||||
|
||||
|
||||
def test_uninstall_matches_the_exact_agent_runtime_directory(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.plugin.install(fixture.config)
|
||||
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
|
||||
other_agent_dir = fixture.agent_dir.parent / "agent-10"
|
||||
|
||||
for event, status in (
|
||||
("PreToolUse", GUARD_STATUS),
|
||||
("PostToolUse", AUDIT_STATUS),
|
||||
):
|
||||
handler = find_handler(settings, event, status)
|
||||
other_handler = {
|
||||
**handler,
|
||||
"command": handler["command"].replace(
|
||||
str(fixture.agent_dir), str(other_agent_dir)
|
||||
),
|
||||
"commandWindows": handler["commandWindows"].replace(
|
||||
str(fixture.agent_dir), str(other_agent_dir)
|
||||
),
|
||||
}
|
||||
settings["hooks"][event].append(
|
||||
{"matcher": "*", "hooks": [other_handler]}
|
||||
)
|
||||
fixture.config_path.write_text(json.dumps(settings), encoding="utf-8")
|
||||
|
||||
fixture.plugin.uninstall(AGENT_ID)
|
||||
|
||||
remaining = json.loads(fixture.config_path.read_text(encoding="utf-8"))
|
||||
assert len(remaining["hooks"]["PreToolUse"]) == 1
|
||||
assert len(remaining["hooks"]["PostToolUse"]) == 1
|
||||
assert str(other_agent_dir) in find_handler(
|
||||
remaining, "PreToolUse", GUARD_STATUS
|
||||
)["command"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("existing_settings", "error_pattern"),
|
||||
"overrides",
|
||||
[
|
||||
({"hooks": None}, 'field "hooks" must be an object'),
|
||||
({"hooks": {"PreToolUse": None}}, 'field "hooks.PreToolUse" must be an array'),
|
||||
(
|
||||
{"hooks": {"PreToolUse": [{"hooks": None}]}},
|
||||
"matcher group must contain a hooks array",
|
||||
),
|
||||
{"agent_name": "cursor"},
|
||||
{"org_id": ""},
|
||||
{"agent_id": "../escape"},
|
||||
{"private_key": "invalid"},
|
||||
{"token": ""},
|
||||
{"base_url": "file:///tmp/elydora"},
|
||||
{"base_url": "https://user:secret@api.elydora.com"},
|
||||
{"base_url": "https://api.elydora.com?tenant=one"},
|
||||
{"base_url": "https://api.elydora.com\\evil"},
|
||||
{"base_url": "https://api.elydora.com/invalid path"},
|
||||
{"base_url": "https://api.elydora.com:invalid"},
|
||||
{"guard_script_path": "unmanaged.py"},
|
||||
],
|
||||
)
|
||||
def test_install_preserves_invalid_shapes_before_runtime_writes(
|
||||
def test_install_validates_identity_credentials_and_api_origin(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
existing_settings: dict[str, Any],
|
||||
error_pattern: str,
|
||||
overrides: dict[str, str],
|
||||
) -> None:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
existing_settings=existing_settings,
|
||||
)
|
||||
original = fixture.config_path.read_text(encoding="utf-8")
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.config.update(overrides)
|
||||
|
||||
with pytest.raises(ValueError, match=error_pattern):
|
||||
fixture.plugin.install(fixture.config)
|
||||
with pytest.raises(ValueError):
|
||||
fixture.install()
|
||||
|
||||
assert fixture.config_path.read_text(encoding="utf-8") == original
|
||||
assert fixture.hook_path.exists() is False
|
||||
assert fixture.runtime_config_path.exists() is False
|
||||
assert fixture.private_key_path.exists() is False
|
||||
assert fixture.config_path.exists() is False
|
||||
assert list(fixture.agent_dir.iterdir()) == []
|
||||
|
||||
|
||||
def test_install_preserves_malformed_config_before_runtime_writes(
|
||||
@pytest.mark.parametrize("kind", ["hooks", "config", "key", "guard", "audit"])
|
||||
def test_linked_hook_and_runtime_files_are_rejected(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
kind: str,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
existing_settings="{ malformed",
|
||||
)
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
if kind == "hooks":
|
||||
target = tmp_path / "hooks-target.json"
|
||||
original = '{"hooks":{}}\n'
|
||||
write_text(target, original)
|
||||
fixture.config_path.parent.mkdir(parents=True)
|
||||
symlink_or_skip(target, fixture.config_path)
|
||||
with pytest.raises(OSError, match="physical file"):
|
||||
fixture.install()
|
||||
assert target.read_text(encoding="utf-8") == original
|
||||
return
|
||||
|
||||
with pytest.raises(ValueError, match="parse Codex hooks config"):
|
||||
fixture.plugin.install(fixture.config)
|
||||
fixture.install()
|
||||
file_path = {
|
||||
"config": fixture.runtime_config_path,
|
||||
"key": fixture.private_key_path,
|
||||
"guard": fixture.guard_path,
|
||||
"audit": fixture.hook_path,
|
||||
}[kind]
|
||||
target = tmp_path / f"{kind}-target"
|
||||
target.write_bytes(file_path.read_bytes())
|
||||
file_path.unlink()
|
||||
symlink_or_skip(target, file_path)
|
||||
|
||||
assert fixture.config_path.read_text(encoding="utf-8") == "{ malformed"
|
||||
assert fixture.hook_path.exists() is False
|
||||
assert fixture.runtime_config_path.exists() is False
|
||||
assert fixture.private_key_path.exists() is False
|
||||
with pytest.raises(OSError, match="physical file"):
|
||||
fixture.plugin.status()
|
||||
|
||||
assert target.read_bytes()
|
||||
@@ -0,0 +1,330 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from codex_support import (
|
||||
ElydoraApiHandler,
|
||||
managed_handler,
|
||||
prepare_fixture,
|
||||
run_handler,
|
||||
server_base_url,
|
||||
start_api_server,
|
||||
symlink_or_skip,
|
||||
write_json,
|
||||
write_text,
|
||||
)
|
||||
from elydora.plugins.codex_contract import AUDIT_STATUS, GUARD_STATUS
|
||||
|
||||
|
||||
def payload(event: str, *, failed: bool = False) -> dict[str, object]:
|
||||
value: dict[str, object] = {
|
||||
"hook_event_name": event,
|
||||
"session_id": "session-1",
|
||||
"turn_id": "turn-1",
|
||||
"transcript_path": None,
|
||||
"cwd": "C:/workspace",
|
||||
"model": "gpt-5",
|
||||
"permission_mode": "default",
|
||||
"tool_name": "Bash",
|
||||
"tool_use_id": "call-1",
|
||||
"tool_input": {"command": "echo test"},
|
||||
}
|
||||
if event == "PostToolUse":
|
||||
value["tool_response"] = (
|
||||
{"success": False, "error": "command failed", "exit_code": 1}
|
||||
if failed
|
||||
else {"success": True, "output": "test"}
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
def encoded(value: object) -> bytes:
|
||||
return json.dumps(value, separators=(",", ":")).encode()
|
||||
|
||||
|
||||
def test_runtimes_enforce_active_state_and_preserve_native_event_json(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
server = start_api_server()
|
||||
try:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
base_url=server_base_url(server),
|
||||
)
|
||||
fixture.install()
|
||||
settings = fixture.settings()
|
||||
guard = managed_handler(settings, "PreToolUse", GUARD_STATUS)
|
||||
audit = managed_handler(settings, "PostToolUse", AUDIT_STATUS)
|
||||
|
||||
pre = payload("PreToolUse")
|
||||
first = payload("PostToolUse")
|
||||
failed = payload("PostToolUse", failed=True)
|
||||
guard_result = run_handler(guard, encoded(pre), fixture)
|
||||
first_result = run_handler(audit, encoded(first), fixture)
|
||||
failed_result = run_handler(audit, encoded(failed), fixture)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert guard_result.returncode == 0
|
||||
assert guard_result.stdout == b""
|
||||
assert guard_result.stderr == b""
|
||||
assert first_result.returncode == 0
|
||||
assert failed_result.returncode == 0
|
||||
requests = ElydoraApiHandler.requests
|
||||
assert [request["method"] for request in requests] == ["GET", "POST", "POST"]
|
||||
assert [request["authorization"] for request in requests] == [
|
||||
"Bearer token-1",
|
||||
"Bearer token-1",
|
||||
"Bearer token-1",
|
||||
]
|
||||
assert requests[1]["json"]["payload"] == first
|
||||
assert requests[2]["json"]["payload"] == failed
|
||||
assert requests[2]["json"]["prev_chain_hash"] == requests[1]["json"]["chain_hash"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("status", ["frozen", "revoked"])
|
||||
def test_guard_propagates_exit_two_for_blocked_agents(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
status: str,
|
||||
) -> None:
|
||||
server = start_api_server()
|
||||
ElydoraApiHandler.agent_status = status
|
||||
try:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
base_url=server_base_url(server),
|
||||
)
|
||||
fixture.install()
|
||||
guard = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS)
|
||||
result = run_handler(guard, encoded(payload("PreToolUse")), fixture)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert result.returncode == 2
|
||||
assert status.encode() in result.stderr
|
||||
assert b"Tool execution blocked" in result.stderr
|
||||
|
||||
|
||||
def test_guard_uses_a_protected_cached_blocking_status(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.install()
|
||||
write_json(
|
||||
fixture.agent_dir / "status-cache.json",
|
||||
{"status": "frozen", "cached_at": time.time()},
|
||||
)
|
||||
guard = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS)
|
||||
|
||||
result = run_handler(guard, encoded(payload("PreToolUse")), fixture)
|
||||
|
||||
assert result.returncode == 2
|
||||
assert b"Tool execution blocked" in result.stderr
|
||||
|
||||
|
||||
def test_fail_open_guard_reports_input_config_and_status_failures(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
server = start_api_server()
|
||||
try:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
base_url=server_base_url(server),
|
||||
)
|
||||
fixture.install()
|
||||
guard = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS)
|
||||
runtime_config = json.loads(
|
||||
fixture.runtime_config_path.read_text(encoding="utf-8")
|
||||
)
|
||||
|
||||
malformed_input = run_handler(guard, b"{ malformed", fixture)
|
||||
non_object_input = run_handler(guard, b"[]", fixture)
|
||||
write_text(fixture.runtime_config_path, "{ malformed")
|
||||
malformed_config = run_handler(
|
||||
guard,
|
||||
encoded(payload("PreToolUse")),
|
||||
fixture,
|
||||
)
|
||||
write_json(fixture.runtime_config_path, runtime_config)
|
||||
(fixture.agent_dir / "status-cache.json").unlink()
|
||||
ElydoraApiHandler.agent_status = "unknown"
|
||||
invalid_status = run_handler(
|
||||
guard,
|
||||
encoded(payload("PreToolUse")),
|
||||
fixture,
|
||||
)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert malformed_input.returncode == 0
|
||||
assert b"invalid JSON" in malformed_input.stderr
|
||||
assert b"fail-open" in malformed_input.stderr
|
||||
assert non_object_input.returncode == 0
|
||||
assert b"JSON object" in non_object_input.stderr
|
||||
assert b"fail-open" in non_object_input.stderr
|
||||
assert malformed_config.returncode == 0
|
||||
assert b"Failed to read agent config" in malformed_config.stderr
|
||||
assert invalid_status.returncode == 0
|
||||
assert b"invalid agent status" in invalid_status.stderr
|
||||
|
||||
|
||||
def test_generated_hooks_report_an_unsafe_runtime_origin(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.install()
|
||||
config = json.loads(fixture.runtime_config_path.read_text(encoding="utf-8"))
|
||||
config["base_url"] = "https://api.elydora.com\\evil"
|
||||
write_json(fixture.runtime_config_path, config)
|
||||
(fixture.agent_dir / "status-cache.json").unlink(missing_ok=True)
|
||||
settings = fixture.settings()
|
||||
|
||||
guard_result = run_handler(
|
||||
managed_handler(settings, "PreToolUse", GUARD_STATUS),
|
||||
encoded(payload("PreToolUse")),
|
||||
fixture,
|
||||
)
|
||||
audit_result = run_handler(
|
||||
managed_handler(settings, "PostToolUse", AUDIT_STATUS),
|
||||
encoded(payload("PostToolUse")),
|
||||
fixture,
|
||||
)
|
||||
|
||||
assert guard_result.returncode == 0
|
||||
assert b"absolute HTTP or HTTPS URL" in guard_result.stderr
|
||||
assert audit_result.returncode == 0
|
||||
assert "absolute HTTP or HTTPS URL" in (fixture.agent_dir / "error.log").read_text(
|
||||
encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("target", "source", "message"),
|
||||
[
|
||||
("input", "{ malformed", "invalid JSON"),
|
||||
("config.json", "{ malformed", "agent config"),
|
||||
("private.key", "invalid", "Private key"),
|
||||
("chain-state.json", "{ malformed", "Chain state"),
|
||||
],
|
||||
)
|
||||
def test_fail_open_audit_records_runtime_failures(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
target: str,
|
||||
source: str,
|
||||
message: str,
|
||||
) -> None:
|
||||
server = start_api_server()
|
||||
try:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
base_url=server_base_url(server),
|
||||
)
|
||||
fixture.install()
|
||||
audit = managed_handler(fixture.settings(), "PostToolUse", AUDIT_STATUS)
|
||||
if target != "input":
|
||||
write_text(fixture.agent_dir / target, source)
|
||||
runtime_input = (
|
||||
source.encode() if target == "input" else encoded(payload("PostToolUse"))
|
||||
)
|
||||
result = run_handler(audit, runtime_input, fixture)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert result.returncode == 0
|
||||
error_log = fixture.agent_dir / "error.log"
|
||||
assert error_log.is_file()
|
||||
assert message.lower() in error_log.read_text(encoding="utf-8").lower()
|
||||
|
||||
|
||||
def test_fail_open_audit_records_api_failures(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
server = start_api_server()
|
||||
ElydoraApiHandler.post_status = 500
|
||||
try:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
base_url=server_base_url(server),
|
||||
)
|
||||
fixture.install()
|
||||
audit = managed_handler(fixture.settings(), "PostToolUse", AUDIT_STATUS)
|
||||
result = run_handler(audit, encoded(payload("PostToolUse")), fixture)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert result.returncode == 0
|
||||
assert "HTTP 500" in (fixture.agent_dir / "error.log").read_text(encoding="utf-8")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target", ["status-cache.json", "chain-state.json", "error.log"]
|
||||
)
|
||||
def test_runtimes_reject_linked_cache_chain_and_error_state(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
target: str,
|
||||
) -> None:
|
||||
server = start_api_server()
|
||||
try:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
base_url=server_base_url(server),
|
||||
)
|
||||
fixture.install()
|
||||
target_path = tmp_path / f"{target}.target"
|
||||
source = {
|
||||
"status-cache.json": json.dumps(
|
||||
{"status": "active", "cached_at": time.time()}
|
||||
),
|
||||
"chain-state.json": json.dumps({"prev_chain_hash": "A" * 43}),
|
||||
"error.log": "preserve\n",
|
||||
}[target]
|
||||
write_text(target_path, source)
|
||||
link = fixture.agent_dir / target
|
||||
symlink_or_skip(target_path, link)
|
||||
if target == "status-cache.json":
|
||||
handler = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS)
|
||||
runtime_input = encoded(payload("PreToolUse"))
|
||||
else:
|
||||
handler = managed_handler(fixture.settings(), "PostToolUse", AUDIT_STATUS)
|
||||
runtime_input = (
|
||||
b"{ malformed"
|
||||
if target == "error.log"
|
||||
else encoded(payload("PostToolUse"))
|
||||
)
|
||||
result = run_handler(handler, runtime_input, fixture)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert result.returncode == 0
|
||||
assert target_path.read_text(encoding="utf-8") == source
|
||||
if target == "error.log":
|
||||
assert b"Failed to write error log" in result.stderr
|
||||
else:
|
||||
assert (
|
||||
b"physical file" in result.stderr
|
||||
or (fixture.agent_dir / "error.log").is_file()
|
||||
)
|
||||
@@ -0,0 +1,234 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import subprocess # nosec B404
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
from codex_support import (
|
||||
AGENT_ID,
|
||||
VALID_PRIVATE_KEY,
|
||||
prepare_fixture,
|
||||
symlink_or_skip,
|
||||
write_json,
|
||||
write_text,
|
||||
)
|
||||
from elydora.plugins import codex
|
||||
|
||||
|
||||
def assert_no_transaction_files(root: Path) -> None:
|
||||
assert not list(root.rglob("*.tmp"))
|
||||
assert not list(root.rglob("*.rollback"))
|
||||
|
||||
|
||||
def test_transaction_rolls_back_all_five_files_when_hooks_commit_fails(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
existing_config={"hooks": {"SessionStart": []}},
|
||||
)
|
||||
original = fixture.config_path.read_text(encoding="utf-8")
|
||||
real_replace = os.replace
|
||||
failed = False
|
||||
|
||||
def fail_hooks_once(source: str, destination: str) -> None:
|
||||
nonlocal failed
|
||||
if os.path.abspath(destination) == str(fixture.config_path) and not failed:
|
||||
failed = True
|
||||
raise OSError("injected hooks commit failure")
|
||||
real_replace(source, destination)
|
||||
|
||||
monkeypatch.setattr(os, "replace", fail_hooks_once)
|
||||
|
||||
with pytest.raises(OSError, match="injected hooks commit failure"):
|
||||
fixture.install()
|
||||
|
||||
assert fixture.config_path.read_text(encoding="utf-8") == original
|
||||
for path in (
|
||||
fixture.guard_path,
|
||||
fixture.hook_path,
|
||||
fixture.runtime_config_path,
|
||||
fixture.private_key_path,
|
||||
):
|
||||
assert path.exists() is False
|
||||
assert_no_transaction_files(fixture.home_dir)
|
||||
|
||||
|
||||
def test_install_detects_a_concurrent_hooks_change_before_staging(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
existing_config={"hooks": {"SessionStart": []}},
|
||||
)
|
||||
concurrent = '{"hooks":{"SessionStart":[{"hooks":[]}]}}\n'
|
||||
real_rendered_change = codex.rendered_change
|
||||
|
||||
def mutate_before_change(rendered: object):
|
||||
write_text(fixture.config_path, concurrent)
|
||||
return real_rendered_change(rendered) # type: ignore[arg-type]
|
||||
|
||||
monkeypatch.setattr(codex, "rendered_change", mutate_before_change)
|
||||
|
||||
with pytest.raises(OSError, match="changed before staging"):
|
||||
fixture.install()
|
||||
|
||||
assert fixture.config_path.read_text(encoding="utf-8") == concurrent
|
||||
assert list(fixture.agent_dir.iterdir()) == []
|
||||
assert_no_transaction_files(fixture.home_dir)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"runtime_name",
|
||||
[
|
||||
"private.key",
|
||||
"guard.py",
|
||||
"hook.py",
|
||||
"chain-state.json",
|
||||
"status-cache.json",
|
||||
"error.log",
|
||||
],
|
||||
)
|
||||
def test_orphan_runtime_artifacts_are_rejected_before_hooks_writes(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
runtime_name: str,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
artifact = fixture.agent_dir / runtime_name
|
||||
write_text(artifact, "orphan\n")
|
||||
|
||||
with pytest.raises(ValueError, match="identity cannot be verified"):
|
||||
fixture.install()
|
||||
|
||||
assert artifact.read_text(encoding="utf-8") == "orphan\n"
|
||||
assert fixture.config_path.exists() is False
|
||||
|
||||
|
||||
def test_mismatched_runtime_identity_is_rejected_before_writes(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
write_json(
|
||||
fixture.runtime_config_path,
|
||||
{"agent_id": "another-agent", "agent_name": "codex"},
|
||||
)
|
||||
original = fixture.runtime_config_path.read_text(encoding="utf-8")
|
||||
|
||||
with pytest.raises(ValueError, match="identity does not match"):
|
||||
fixture.install()
|
||||
|
||||
assert fixture.runtime_config_path.read_text(encoding="utf-8") == original
|
||||
assert fixture.config_path.exists() is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kind", ["codex", "runtime"])
|
||||
def test_linked_hooks_and_runtime_directories_are_rejected(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
kind: str,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
target = tmp_path / f"{kind}-target"
|
||||
target.mkdir()
|
||||
if kind == "codex":
|
||||
link = fixture.config_path.parent
|
||||
else:
|
||||
link = fixture.agent_dir.parent
|
||||
shutil.rmtree(link)
|
||||
link.parent.mkdir(parents=True, exist_ok=True)
|
||||
symlink_or_skip(target, link, directory=True)
|
||||
|
||||
with pytest.raises(OSError, match="physical directory"):
|
||||
fixture.install()
|
||||
|
||||
assert list(target.iterdir()) == []
|
||||
assert fixture.config_path.exists() is False
|
||||
|
||||
|
||||
def test_uninstall_rejects_a_linked_default_hooks_directory(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.install()
|
||||
codex_directory = fixture.config_path.parent
|
||||
target = tmp_path / "codex-hooks-target"
|
||||
codex_directory.rename(target)
|
||||
symlink_or_skip(target, codex_directory, directory=True)
|
||||
|
||||
with pytest.raises(OSError, match="physical directory"):
|
||||
fixture.plugin.uninstall(AGENT_ID)
|
||||
|
||||
assert (target / "hooks.json").is_file()
|
||||
|
||||
|
||||
def test_install_protects_credentials_and_cleans_transaction_files(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = prepare_fixture(monkeypatch, tmp_path)
|
||||
fixture.install()
|
||||
|
||||
assert_no_transaction_files(fixture.home_dir)
|
||||
if os.name != "nt":
|
||||
for path in (
|
||||
fixture.config_path,
|
||||
fixture.runtime_config_path,
|
||||
fixture.private_key_path,
|
||||
):
|
||||
assert path.stat().st_mode & 0o777 == 0o600
|
||||
|
||||
|
||||
def test_cli_preflight_preserves_malformed_hooks_before_runtime_creation(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
home_dir = tmp_path / "cli-home"
|
||||
hooks_path = home_dir / ".codex" / "hooks.json"
|
||||
key_path = tmp_path / "private.key"
|
||||
write_text(hooks_path, "{ malformed")
|
||||
write_text(key_path, VALID_PRIVATE_KEY)
|
||||
result = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
"-m",
|
||||
"elydora.cli",
|
||||
"install",
|
||||
"--agent",
|
||||
"codex",
|
||||
"--org_id",
|
||||
"org-1",
|
||||
"--agent_id",
|
||||
AGENT_ID,
|
||||
"--private_key_file",
|
||||
str(key_path),
|
||||
"--kid",
|
||||
"key-1",
|
||||
"--base_url",
|
||||
"https://api.elydora.com",
|
||||
],
|
||||
capture_output=True,
|
||||
check=False,
|
||||
cwd=Path(__file__).parents[1],
|
||||
env={
|
||||
**os.environ,
|
||||
"HOME": str(home_dir),
|
||||
"USERPROFILE": str(home_dir),
|
||||
"CODEX_HOME": "",
|
||||
},
|
||||
text=True,
|
||||
)
|
||||
|
||||
assert result.returncode == 1
|
||||
assert "parse Codex user hooks" in result.stderr
|
||||
assert hooks_path.read_text(encoding="utf-8") == "{ malformed"
|
||||
assert (home_dir / ".elydora").exists() is False
|
||||
@@ -44,7 +44,9 @@ def create_guard(tmp_path: Path, status: str) -> tuple[Path, ThreadingHTTPServer
|
||||
"agent_id": AGENT_ID,
|
||||
"base_url": f"http://127.0.0.1:{server.server_port}",
|
||||
}
|
||||
(agent_dir / "config.json").write_text(json.dumps(config), encoding="utf-8")
|
||||
config_path = agent_dir / "config.json"
|
||||
config_path.write_text(json.dumps(config), encoding="utf-8")
|
||||
os.chmod(config_path, 0o600)
|
||||
|
||||
script_path = tmp_path / "guard.py"
|
||||
script_path.write_text(
|
||||
@@ -54,7 +56,11 @@ def create_guard(tmp_path: Path, status: str) -> tuple[Path, ThreadingHTTPServer
|
||||
return script_path, server
|
||||
|
||||
|
||||
def run_guard(script_path: Path, home_dir: Path) -> subprocess.CompletedProcess[str]:
|
||||
def run_guard(
|
||||
script_path: Path,
|
||||
home_dir: Path,
|
||||
payload: str = "{}",
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
env = {
|
||||
**os.environ,
|
||||
"HOME": str(home_dir),
|
||||
@@ -65,7 +71,7 @@ def run_guard(script_path: Path, home_dir: Path) -> subprocess.CompletedProcess[
|
||||
capture_output=True,
|
||||
check=False,
|
||||
env=env,
|
||||
input="",
|
||||
input=payload,
|
||||
text=True,
|
||||
)
|
||||
|
||||
@@ -102,6 +108,7 @@ def test_cached_frozen_status_uses_blocking_exit_code(tmp_path: Path) -> None:
|
||||
json.dumps({"status": "frozen", "cached_at": time.time()}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
os.chmod(cache_path, 0o600)
|
||||
try:
|
||||
result = run_guard(script_path, tmp_path)
|
||||
finally:
|
||||
@@ -110,3 +117,29 @@ def test_cached_frozen_status_uses_blocking_exit_code(tmp_path: Path) -> None:
|
||||
|
||||
assert result.returncode == 2
|
||||
assert "Tool execution blocked" in result.stderr
|
||||
|
||||
|
||||
def test_fail_open_guard_reports_malformed_input(tmp_path: Path) -> None:
|
||||
script_path, server = create_guard(tmp_path, "active")
|
||||
try:
|
||||
result = run_guard(script_path, tmp_path, "{ malformed")
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert result.returncode == 0
|
||||
assert "invalid JSON" in result.stderr
|
||||
assert "fail-open" in result.stderr
|
||||
|
||||
|
||||
def test_fail_open_guard_reports_non_object_input(tmp_path: Path) -> None:
|
||||
script_path, server = create_guard(tmp_path, "active")
|
||||
try:
|
||||
result = run_guard(script_path, tmp_path, "[]")
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
|
||||
assert result.returncode == 0
|
||||
assert "JSON object" in result.stderr
|
||||
assert "fail-open" in result.stderr
|
||||
Reference in new issue
Block a user