484 lines
16 KiB
Python
484 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
from dataclasses import dataclass
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from queue import Queue
|
|
import shutil
|
|
import subprocess # nosec B404
|
|
import sys
|
|
from threading import Thread
|
|
from typing import Any, Optional
|
|
|
|
import pytest
|
|
|
|
from elydora import cli
|
|
from elydora.plugins import cline
|
|
from elydora.plugins import cline_contract, cline_io
|
|
from elydora.plugins.base import InstallConfig
|
|
from elydora.plugins.registry import SUPPORTED_AGENTS
|
|
|
|
|
|
AGENT_ID = "agent-1"
|
|
MISSING = object()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ClineFixture:
|
|
plugin: cline.ClinePlugin
|
|
config: InstallConfig
|
|
home_dir: Path
|
|
workspace_dir: Path
|
|
cline_dir: Path
|
|
hooks_dir: Path
|
|
agent_dir: Path
|
|
guard_path: Path
|
|
audit_path: Path
|
|
guard_wrapper: Path
|
|
audit_wrapper: Path
|
|
runtime_config: Path
|
|
|
|
|
|
def _python_script(body: str) -> str:
|
|
return f"#!{sys.executable}\n{body}"
|
|
|
|
|
|
def prepare_fixture(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
*,
|
|
auto_install: bool = True,
|
|
create_guard: bool = True,
|
|
existing_audit: Any = MISSING,
|
|
existing_guard: Any = MISSING,
|
|
guard_source: Optional[str] = None,
|
|
) -> ClineFixture:
|
|
home_dir = tmp_path / "home with spaces"
|
|
workspace_dir = tmp_path / "workspace"
|
|
cline_dir = tmp_path / "custom-cline-home"
|
|
hooks_dir = cline_dir / "hooks"
|
|
agent_dir = home_dir / ".elydora" / AGENT_ID
|
|
guard_path = agent_dir / "guard.py"
|
|
audit_path = agent_dir / "hook.py"
|
|
guard_wrapper = hooks_dir / "PreToolUse.mjs"
|
|
audit_wrapper = hooks_dir / "PostToolUse.mjs"
|
|
workspace_dir.mkdir(parents=True)
|
|
agent_dir.mkdir(parents=True)
|
|
if create_guard:
|
|
guard_path.write_text(
|
|
guard_source
|
|
or _python_script(
|
|
"import sys\n"
|
|
"sys.stdin.read()\n"
|
|
"sys.stderr.write('Agent is frozen by Elydora.\\n')\n"
|
|
"raise SystemExit(2)\n"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
if existing_guard is not MISSING:
|
|
hooks_dir.mkdir(parents=True, exist_ok=True)
|
|
guard_wrapper.write_text(str(existing_guard), encoding="utf-8")
|
|
if existing_audit is not MISSING:
|
|
hooks_dir.mkdir(parents=True, exist_ok=True)
|
|
audit_wrapper.write_text(str(existing_audit), encoding="utf-8")
|
|
|
|
monkeypatch.setenv("HOME", str(home_dir))
|
|
monkeypatch.setenv("USERPROFILE", str(home_dir))
|
|
monkeypatch.setenv("CLINE_DIR", str(cline_dir))
|
|
private_key = base64.urlsafe_b64encode(bytes([1]) * 32).rstrip(b"=").decode()
|
|
config: InstallConfig = {
|
|
"agent_id": AGENT_ID,
|
|
"agent_name": "cline",
|
|
"org_id": "org-1",
|
|
"private_key": private_key,
|
|
"kid": "kid-1",
|
|
"base_url": "https://api.elydora.test",
|
|
"guard_script_path": str(guard_path),
|
|
}
|
|
fixture = ClineFixture(
|
|
plugin=cline.ClinePlugin(),
|
|
config=config,
|
|
home_dir=home_dir,
|
|
workspace_dir=workspace_dir,
|
|
cline_dir=cline_dir,
|
|
hooks_dir=hooks_dir,
|
|
agent_dir=agent_dir,
|
|
guard_path=guard_path,
|
|
audit_path=audit_path,
|
|
guard_wrapper=guard_wrapper,
|
|
audit_wrapper=audit_wrapper,
|
|
runtime_config=agent_dir / "config.json",
|
|
)
|
|
if auto_install:
|
|
fixture.plugin.install(config)
|
|
return fixture
|
|
|
|
|
|
def run_wrapper(
|
|
wrapper: Path,
|
|
fixture: ClineFixture,
|
|
payload: dict[str, Any],
|
|
) -> subprocess.CompletedProcess[str]:
|
|
node = shutil.which("node")
|
|
assert node is not None
|
|
return subprocess.run( # nosec B603
|
|
[node, str(wrapper)],
|
|
cwd=fixture.workspace_dir,
|
|
env={
|
|
**os.environ,
|
|
"HOME": str(fixture.home_dir),
|
|
"USERPROFILE": str(fixture.home_dir),
|
|
},
|
|
input=json.dumps(payload),
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
timeout=10,
|
|
)
|
|
|
|
|
|
def test_cline_is_registered_in_the_sdk_and_cli() -> None:
|
|
assert SUPPORTED_AGENTS["cline"] == {
|
|
"name": "Cline",
|
|
"hook_event": "PreToolUse/PostToolUse",
|
|
"config_path": "~/.cline/hooks/PreToolUse.mjs",
|
|
}
|
|
assert cli.PLUGIN_MAP["cline"] is cline.ClinePlugin
|
|
|
|
|
|
def test_install_writes_only_native_global_hooks_and_is_idempotent(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
guard_source = fixture.guard_wrapper.read_text(encoding="utf-8")
|
|
audit_source = fixture.audit_wrapper.read_text(encoding="utf-8")
|
|
|
|
fixture.plugin.install(fixture.config)
|
|
|
|
assert fixture.guard_wrapper.read_text(encoding="utf-8") == guard_source
|
|
assert fixture.audit_wrapper.read_text(encoding="utf-8") == audit_source
|
|
assert guard_source.startswith("#!/usr/bin/env node\n// @elydora-cline-hook ")
|
|
assert list(fixture.hooks_dir.glob("*.tmp")) == []
|
|
assert not (fixture.home_dir / "Documents" / "Cline" / "Hooks").exists()
|
|
assert not (fixture.workspace_dir / ".cline" / "hooks").exists()
|
|
assert not (fixture.workspace_dir / ".clinerules" / "hooks").exists()
|
|
|
|
|
|
def test_install_uses_official_default_without_cline_dir(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path, auto_install=False)
|
|
monkeypatch.delenv("CLINE_DIR")
|
|
|
|
fixture.plugin.install(fixture.config)
|
|
|
|
default_hooks = fixture.home_dir / ".cline" / "hooks"
|
|
assert (default_hooks / "PreToolUse.mjs").is_file()
|
|
assert (default_hooks / "PostToolUse.mjs").is_file()
|
|
assert not fixture.guard_wrapper.exists()
|
|
|
|
|
|
def test_wrappers_translate_freezes_and_forward_payload_byte_for_byte(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
capture_path = tmp_path / "captured-event.json"
|
|
fixture.audit_path.write_text(
|
|
_python_script(
|
|
"from pathlib import Path\n"
|
|
"import os, sys\n"
|
|
f"Path({str(capture_path)!r}).write_text("
|
|
"json.dumps({'cwd': os.getcwd(), 'input': sys.stdin.read()}), "
|
|
"encoding='utf-8')\n"
|
|
).replace("import os, sys", "import json, os, sys"),
|
|
encoding="utf-8",
|
|
)
|
|
pre_payload = {
|
|
"clineVersion": "3.0.46",
|
|
"hookName": "tool_call",
|
|
"taskId": "task-1",
|
|
"tool_call": {"name": "read_file", "input": {"path": "README.md"}},
|
|
}
|
|
guard = run_wrapper(fixture.guard_wrapper, fixture, pre_payload)
|
|
assert guard.returncode == 0
|
|
assert "Agent is frozen by Elydora" in guard.stderr
|
|
marker, control = guard.stdout.strip().split("\t", 1)
|
|
assert marker == "HOOK_CONTROL"
|
|
assert json.loads(control) == {
|
|
"cancel": True,
|
|
"errorMessage": "Agent is frozen by Elydora.",
|
|
}
|
|
|
|
post_payload = {
|
|
"clineVersion": "3.0.46",
|
|
"hookName": "tool_result",
|
|
"taskId": "task-1",
|
|
"tool_result": {"name": "read_file", "input": {"path": "README.md"}},
|
|
}
|
|
audit = run_wrapper(fixture.audit_wrapper, fixture, post_payload)
|
|
assert audit.returncode == 0
|
|
captured = json.loads(capture_path.read_text(encoding="utf-8"))
|
|
assert captured == {"cwd": str(fixture.workspace_dir), "input": json.dumps(post_payload)}
|
|
|
|
|
|
def test_wrappers_keep_passes_quiet_and_surface_runtime_failures(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
passing = prepare_fixture(
|
|
monkeypatch,
|
|
tmp_path / "passing",
|
|
guard_source=_python_script("import sys\nsys.stdin.read()\n"),
|
|
)
|
|
result = run_wrapper(passing.guard_wrapper, passing, {})
|
|
assert result.returncode == 0
|
|
assert result.stdout == ""
|
|
|
|
failing = prepare_fixture(monkeypatch, tmp_path / "failing")
|
|
failing.audit_path.write_text(
|
|
_python_script(
|
|
"import sys\n"
|
|
"sys.stderr.write('audit failed\\n')\n"
|
|
"raise SystemExit(7)\n"
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
result = run_wrapper(failing.audit_wrapper, failing, {})
|
|
assert result.returncode == 1
|
|
assert "audit failed" in result.stderr
|
|
assert "exited with code 7" in result.stderr
|
|
|
|
|
|
def test_wrapper_rejects_a_non_absolute_runtime_shebang(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(
|
|
monkeypatch,
|
|
tmp_path,
|
|
guard_source="#!python.exe\nimport sys\nsys.stdin.read()\n",
|
|
)
|
|
|
|
result = run_wrapper(fixture.guard_wrapper, fixture, {})
|
|
|
|
assert result.returncode == 1
|
|
assert result.stdout == ""
|
|
assert "runtime shebang executable must be absolute" in result.stderr
|
|
|
|
|
|
class _OperationHandler(BaseHTTPRequestHandler):
|
|
operations: Queue[dict[str, Any]]
|
|
|
|
def do_POST(self) -> None: # noqa: N802
|
|
length = int(self.headers["Content-Length"])
|
|
body = json.loads(self.rfile.read(length).decode("utf-8"))
|
|
self.operations.put(body)
|
|
self.send_response(201)
|
|
self.end_headers()
|
|
self.wfile.write(b"{}")
|
|
|
|
def log_message(self, format: str, *args: Any) -> None:
|
|
return
|
|
|
|
|
|
def test_audit_runtime_maps_official_nested_fields_into_the_operation(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
operations: Queue[dict[str, Any]] = Queue()
|
|
_OperationHandler.operations = operations
|
|
server = ThreadingHTTPServer(("127.0.0.1", 0), _OperationHandler)
|
|
thread = Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
fixture = prepare_fixture(monkeypatch, tmp_path, auto_install=False)
|
|
fixture.config["base_url"] = f"http://127.0.0.1:{server.server_port}"
|
|
try:
|
|
fixture.plugin.install(fixture.config)
|
|
payload = {
|
|
"hookName": "tool_result",
|
|
"taskId": "task-1",
|
|
"tool_result": {
|
|
"name": "read_file",
|
|
"input": {"path": "README.md"},
|
|
"output": "ok",
|
|
},
|
|
}
|
|
result = run_wrapper(fixture.audit_wrapper, fixture, payload)
|
|
assert result.returncode == 0
|
|
operation = operations.get(timeout=3)
|
|
assert operation["payload"] == {
|
|
"tool_name": "read_file",
|
|
"tool_input": {"path": "README.md"},
|
|
"session_id": "task-1",
|
|
}
|
|
assert operation["action"] == {"tool": "read_file"}
|
|
assert operation["subject"] == {"session_id": "task-1"}
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=3)
|
|
|
|
|
|
def test_status_requires_intact_hooks_and_runtimes(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
assert fixture.plugin.status() == {
|
|
"installed": True,
|
|
"agent": "cline",
|
|
"details": f"Config: {fixture.hooks_dir}",
|
|
}
|
|
fixture.guard_path.unlink()
|
|
assert fixture.plugin.status()["installed"] is False
|
|
fixture.guard_path.write_text(_python_script("pass\n"), encoding="utf-8")
|
|
fixture.audit_wrapper.unlink()
|
|
assert fixture.plugin.status() == {
|
|
"installed": False,
|
|
"agent": "cline",
|
|
"details": "Not installed",
|
|
}
|
|
|
|
|
|
def test_status_surfaces_corrupt_hooks_and_runtime_metadata(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
fixture.guard_wrapper.write_text(
|
|
fixture.guard_wrapper.read_text(encoding="utf-8") + "\n// tampered\n",
|
|
encoding="utf-8",
|
|
)
|
|
with pytest.raises(ValueError, match="managed template"):
|
|
fixture.plugin.status()
|
|
|
|
fixture.plugin.install(fixture.config)
|
|
fixture.runtime_config.write_text("{ malformed", encoding="utf-8")
|
|
with pytest.raises(ValueError, match="parse Elydora runtime config"):
|
|
fixture.plugin.status()
|
|
|
|
|
|
@pytest.mark.parametrize("collision", ["guard", "audit"])
|
|
def test_install_rejects_user_filename_collisions_before_writes(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
collision: str,
|
|
) -> None:
|
|
existing_guard = "// user PreToolUse hook\n" if collision == "guard" else MISSING
|
|
existing_audit = "// user PostToolUse hook\n" if collision == "audit" else MISSING
|
|
fixture = prepare_fixture(
|
|
monkeypatch,
|
|
tmp_path,
|
|
auto_install=False,
|
|
existing_guard=existing_guard,
|
|
existing_audit=existing_audit,
|
|
)
|
|
with pytest.raises(ValueError, match="owned by another integration"):
|
|
fixture.plugin.install(fixture.config)
|
|
assert fixture.audit_path.exists() is False
|
|
assert fixture.runtime_config.exists() is False
|
|
if collision == "guard":
|
|
assert fixture.guard_wrapper.read_text(encoding="utf-8") == existing_guard
|
|
assert fixture.audit_wrapper.exists() is False
|
|
else:
|
|
assert fixture.audit_wrapper.read_text(encoding="utf-8") == existing_audit
|
|
assert fixture.guard_wrapper.exists() is False
|
|
|
|
|
|
def test_install_preserves_corrupt_metadata_for_recovery(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
corrupt = "#!/usr/bin/env node\n// @elydora-cline-hook invalid\n"
|
|
fixture = prepare_fixture(
|
|
monkeypatch,
|
|
tmp_path,
|
|
auto_install=False,
|
|
existing_guard=corrupt,
|
|
)
|
|
with pytest.raises(ValueError, match="parse Elydora Cline hook metadata"):
|
|
fixture.plugin.install(fixture.config)
|
|
assert fixture.guard_wrapper.read_text(encoding="utf-8") == corrupt
|
|
assert fixture.audit_wrapper.exists() is False
|
|
assert fixture.audit_path.exists() is False
|
|
|
|
|
|
def test_install_rejects_missing_guard_before_creating_files(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(
|
|
monkeypatch,
|
|
tmp_path,
|
|
auto_install=False,
|
|
create_guard=False,
|
|
)
|
|
with pytest.raises(FileNotFoundError, match="guard runtime is missing"):
|
|
fixture.plugin.install(fixture.config)
|
|
assert fixture.guard_wrapper.exists() is False
|
|
assert fixture.audit_wrapper.exists() is False
|
|
assert fixture.audit_path.exists() is False
|
|
assert fixture.runtime_config.exists() is False
|
|
|
|
|
|
def test_uninstall_removes_exact_ownership_and_preserves_other_hooks(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
user_hook = fixture.hooks_dir / "PreToolUse.py"
|
|
user_hook.write_text("# user hook\n", encoding="utf-8")
|
|
|
|
fixture.plugin.uninstall("agent-10")
|
|
assert fixture.guard_wrapper.is_file()
|
|
assert fixture.audit_wrapper.is_file()
|
|
fixture.plugin.uninstall(AGENT_ID)
|
|
assert fixture.guard_wrapper.exists() is False
|
|
assert fixture.audit_wrapper.exists() is False
|
|
assert user_hook.read_text(encoding="utf-8") == "# user hook\n"
|
|
|
|
|
|
def test_hook_pair_restores_the_first_hook_when_the_second_commit_fails(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
) -> None:
|
|
fixture = prepare_fixture(monkeypatch, tmp_path)
|
|
original_guard = fixture.guard_wrapper.read_text(encoding="utf-8")
|
|
original_audit = fixture.audit_wrapper.read_text(encoding="utf-8")
|
|
guard_state = cline_io.read_hook_file(str(fixture.guard_wrapper))
|
|
audit_state = cline_io.read_hook_file(str(fixture.audit_wrapper))
|
|
replacement_guard = cline_contract.build_wrapper(
|
|
cline_contract.build_metadata("guard", "agent-2", str(fixture.guard_path))
|
|
)
|
|
replacement_audit = cline_contract.build_wrapper(
|
|
cline_contract.build_metadata("audit", "agent-2", str(fixture.audit_path))
|
|
)
|
|
real_replace = os.replace
|
|
failure_injected = False
|
|
|
|
def fail_audit_commit(source: Any, destination: Any) -> None:
|
|
nonlocal failure_injected
|
|
if not failure_injected and Path(destination) == fixture.audit_wrapper:
|
|
failure_injected = True
|
|
raise OSError("simulated audit commit failure")
|
|
real_replace(source, destination)
|
|
|
|
monkeypatch.setattr(os, "replace", fail_audit_commit)
|
|
|
|
with pytest.raises(OSError, match="Write Cline hook pair"):
|
|
cline_io.write_hook_pair(
|
|
cline_io.PendingWrite(guard_state, replacement_guard),
|
|
cline_io.PendingWrite(audit_state, replacement_audit),
|
|
)
|
|
|
|
assert failure_injected is True
|
|
assert fixture.guard_wrapper.read_text(encoding="utf-8") == original_guard
|
|
assert fixture.audit_wrapper.read_text(encoding="utf-8") == original_audit
|
|
assert list(fixture.hooks_dir.glob("*.tmp")) == []
|