237 lines
7.5 KiB
Python
237 lines
7.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import time
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
|
import pytest
|
|
|
|
from cursor_support import (
|
|
ElydoraApiHandler,
|
|
VALID_PRIVATE_KEY,
|
|
managed_handler,
|
|
prepare_fixture,
|
|
run_handler,
|
|
server_base_url,
|
|
start_api_server,
|
|
write_json,
|
|
write_text,
|
|
)
|
|
from elydora.crypto import (
|
|
compute_chain_hash,
|
|
compute_payload_hash,
|
|
jcs_canonicalize,
|
|
)
|
|
from elydora.utils import base64url_decode
|
|
|
|
|
|
SUCCESS_EVENT = {
|
|
"conversation_id": "conversation-1",
|
|
"generation_id": "generation-1",
|
|
"hook_event_name": "postToolUse",
|
|
"tool_name": "Shell",
|
|
"tool_input": {"command": "Get-ChildItem"},
|
|
"tool_output": '{"exitCode":0,"stdout":"ok"}',
|
|
"tool_use_id": "call-1",
|
|
"cwd": "project",
|
|
"duration": 42,
|
|
}
|
|
FAILURE_EVENT = {
|
|
"conversation_id": "conversation-1",
|
|
"generation_id": "generation-1",
|
|
"hook_event_name": "postToolUseFailure",
|
|
"tool_name": "Shell",
|
|
"tool_input": {"command": "exit 1"},
|
|
"tool_use_id": "call-2",
|
|
"cwd": "project",
|
|
"error_message": "command failed",
|
|
"failure_type": "error",
|
|
"duration": 21,
|
|
"is_interrupt": False,
|
|
}
|
|
|
|
|
|
def _payload(value: object) -> bytes:
|
|
return (json.dumps(value, separators=(",", ":")) + "\n").encode()
|
|
|
|
|
|
def _assert_signed_operation(operation: dict, event: dict) -> None:
|
|
assert operation["payload"] == event
|
|
assert operation["payload_hash"] == compute_payload_hash(event)
|
|
assert operation["subject"] == {"session_id": event["conversation_id"]}
|
|
assert operation["action"] == {"tool": event["tool_name"]}
|
|
expected_chain = compute_chain_hash(
|
|
operation["prev_chain_hash"],
|
|
operation["payload_hash"],
|
|
operation["operation_id"],
|
|
operation["issued_at"],
|
|
)
|
|
assert operation["chain_hash"] == expected_chain
|
|
unsigned = {
|
|
key: value
|
|
for key, value in operation.items()
|
|
if key not in {"chain_hash", "signature"}
|
|
}
|
|
public_key = Ed25519PrivateKey.from_private_bytes(
|
|
base64url_decode(VALID_PRIVATE_KEY)
|
|
).public_key()
|
|
public_key.verify(
|
|
base64url_decode(operation["signature"]),
|
|
jcs_canonicalize(unsigned).encode(),
|
|
)
|
|
|
|
|
|
def test_handlers_enforce_status_and_submit_both_native_events(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
server = start_api_server()
|
|
try:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
fixture.config["base_url"] = server_base_url(server)
|
|
fixture.install()
|
|
settings = fixture.settings()
|
|
guard = managed_handler(settings, "preToolUse", "guard.py")
|
|
success = managed_handler(settings, "postToolUse", "hook.py")
|
|
failure = managed_handler(settings, "postToolUseFailure", "hook.py")
|
|
|
|
guard_result = run_handler(
|
|
guard,
|
|
_payload({"hook_event_name": "preToolUse"}),
|
|
)
|
|
success_result = run_handler(success, _payload(SUCCESS_EVENT))
|
|
failure_result = run_handler(failure, _payload(FAILURE_EVENT))
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
assert guard_result.returncode == 0
|
|
assert json.loads(guard_result.stdout) == {"permission": "allow"}
|
|
assert success_result.returncode == 0
|
|
assert json.loads(success_result.stdout) == {}
|
|
assert failure_result.returncode == 0
|
|
assert json.loads(failure_result.stdout) == {}
|
|
assert ElydoraApiHandler.get_paths == ["/v1/agents/agent-1"]
|
|
assert ElydoraApiHandler.authorizations == [
|
|
"Bearer token-1",
|
|
"Bearer token-1",
|
|
"Bearer token-1",
|
|
]
|
|
assert len(ElydoraApiHandler.operations) == 2
|
|
_assert_signed_operation(ElydoraApiHandler.operations[0], SUCCESS_EVENT)
|
|
_assert_signed_operation(ElydoraApiHandler.operations[1], FAILURE_EVENT)
|
|
assert (
|
|
ElydoraApiHandler.operations[1]["prev_chain_hash"]
|
|
== ElydoraApiHandler.operations[0]["chain_hash"]
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["frozen", "revoked"])
|
|
def test_guard_returns_native_denial_and_exit_two_from_cache(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
status: str,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
fixture.install()
|
|
write_json(
|
|
fixture.agent_dir / "status-cache.json",
|
|
{"status": status, "cached_at": time.time()},
|
|
)
|
|
guard = managed_handler(fixture.settings(), "preToolUse", "guard.py")
|
|
|
|
result = run_handler(guard, _payload({"hook_event_name": "preToolUse"}))
|
|
|
|
denial = json.loads(result.stdout)
|
|
assert result.returncode == 2
|
|
assert denial["permission"] == "deny"
|
|
assert status in denial["userMessage"]
|
|
assert b"Tool execution blocked" in result.stderr
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("target", "source", "event"),
|
|
[
|
|
("config.json", "{ malformed", "preToolUse"),
|
|
("config.json", "{ malformed", "postToolUse"),
|
|
("private.key", "invalid", "postToolUse"),
|
|
("chain-state.json", "{ malformed", "postToolUse"),
|
|
(
|
|
"status-cache.json",
|
|
json.dumps({"status": "active", "cached_at": 4_102_444_800}),
|
|
"preToolUse",
|
|
),
|
|
],
|
|
)
|
|
def test_runtime_integrity_failures_exit_one_and_are_observable(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
target: str,
|
|
source: str,
|
|
event: str,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
fixture.install()
|
|
write_text(fixture.agent_dir / target, source)
|
|
script = "guard.py" if event == "preToolUse" else "hook.py"
|
|
handler = managed_handler(fixture.settings(), event, script)
|
|
payload = {"hook_event_name": event}
|
|
|
|
result = run_handler(handler, _payload(payload))
|
|
|
|
assert result.returncode == 1
|
|
assert result.stdout == b""
|
|
assert b"Elydora" in result.stderr
|
|
if event != "preToolUse":
|
|
error_log = fixture.agent_dir / "error.log"
|
|
assert error_log.is_file()
|
|
assert error_log.read_text(encoding="utf-8")
|
|
|
|
|
|
@pytest.mark.parametrize("event", ["preToolUse", "postToolUse"])
|
|
def test_malformed_hook_input_exits_one(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
event: str,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
fixture.install()
|
|
script = "guard.py" if event == "preToolUse" else "hook.py"
|
|
handler = managed_handler(fixture.settings(), event, script)
|
|
|
|
result = run_handler(handler, b"{ malformed")
|
|
|
|
assert result.returncode == 1
|
|
assert result.stdout == b""
|
|
assert b"invalid JSON" in result.stderr
|
|
|
|
|
|
@pytest.mark.parametrize("event", ["preToolUse", "postToolUse"])
|
|
def test_api_failures_exit_one_and_audit_failure_is_logged(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
event: str,
|
|
) -> None:
|
|
server = start_api_server()
|
|
ElydoraApiHandler.get_status = 500
|
|
ElydoraApiHandler.post_status = 500
|
|
try:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
fixture.config["base_url"] = server_base_url(server)
|
|
fixture.install()
|
|
script = "guard.py" if event == "preToolUse" else "hook.py"
|
|
handler = managed_handler(fixture.settings(), event, script)
|
|
payload = {"hook_event_name": event}
|
|
result = run_handler(handler, _payload(payload))
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
assert result.returncode == 1
|
|
assert result.stdout == b""
|
|
assert b"HTTP 500" in result.stderr
|
|
if event == "postToolUse":
|
|
log = fixture.agent_dir / "error.log"
|
|
assert "HTTP 500" in log.read_text(encoding="utf-8")
|