From 97d1525227682dfa8191dbfe63bb5a24c9b22adb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Jul 2026 11:06:10 -0400 Subject: [PATCH] fix(cline): make Python hook installation transactional --- AGENTS.md | 2 +- README.md | 2 +- elydora/plugins/cline.py | 126 ++---- elydora/plugins/cline_contract.py | 12 +- elydora/plugins/cline_installation.py | 232 ++++++++++ elydora/plugins/cline_io.py | 332 ++++++++------ tests/cline_support.py | 226 +++++++++ tests/test_cline_official.py | 81 ++++ tests/test_cline_plugin.py | 630 ++++++++++++-------------- tests/test_cline_transaction.py | 304 +++++++++++++ 10 files changed, 1375 insertions(+), 572 deletions(-) create mode 100644 elydora/plugins/cline_installation.py create mode 100644 tests/cline_support.py create mode 100644 tests/test_cline_official.py create mode 100644 tests/test_cline_transaction.py diff --git a/AGENTS.md b/AGENTS.md index 02250f3..0e1966b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,7 +30,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Resolve Gemini CLI user hooks through `$GEMINI_CLI_HOME/.gemini/settings.json` with `~/.gemini/settings.json` when the variable is absent or empty. Preserve relative and literal-tilde home values as Gemini CLI does. Keep workspace, system defaults, system overrides, and extension hooks read-only. Preserve JSON comments while rejecting trailing commas and duplicate keys. Register exact matchless `BeforeTool` and `AfterTool` groups named `elydora-guard` and `elydora-audit` with ten-second command handlers. Preserve the complete native snake_case payload, emit valid JSON on exit code `0`, and propagate freeze and revocation through exit code `2`. Use encoded PowerShell commands on Windows. Commit Gemini CLI user settings, generated runtimes, runtime config, and private key through one rollback-capable transaction. Respect `hooksConfig.enabled` and `hooksConfig.disabled`, then require `/hooks list` verification for higher-scope policy effects. - Resolve Auggie user hooks through `~/.augment/settings.json`; keep system, workspace, local workspace, and alternate `--augment-cache-dir` settings read-only. Validate the shipped Auggie 0.33 hook schema: `PreToolUse`, `PostToolUse`, `Stop`, `SessionStart`, `SessionEnd`, `Notification`, and `PromptSubmit`; regex matchers belong to tool events, command handlers use string-array arguments and positive millisecond timeouts, and metadata flags are booleans. Generate `.cmd` wrappers on Windows and `.sh` wrappers on Unix, preserve the complete native snake_case payload, and propagate freeze and revocation through exit code `2`. - Validate Auggie matcher syntax during installation with Node.js `new RegExp`; keep status and uninstall independent from the JavaScript validator so recovery remains available offline. Commit Auggie settings, both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Require exact runtime identity, canonical private keys, physical files, and exact wrapper sources during status checks, then verify the effective configuration with `auggie tools list`. -- Write Cline hooks only to `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default; keep Documents and workspace hook roots read-only. Preserve official input byte-for-byte, resolve Python runtimes from their absolute shebangs, and translate guard exit code `2` into Cline's JSON stdout cancellation control. +- Treat Cline 3 CLI file hooks as the adapter contract. Resolve hooks additively from `~/Documents/Cline/Hooks`, `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default, `.clinerules/hooks`, and `.cline/hooks`; write only the `$CLINE_DIR` source. Preserve the complete hybrid CLI payload, resolve generated Python runtimes from absolute shebangs, and emit one-line JSON stdout with `cancel: true` for guard exit code `2`. Commit both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Status requires physical files, canonical private keys, exact runtime identity, exact generated sources, and exact wrapper sources. Hook errors, timeouts, and invalid controls fail open in Cline while remaining observable. - Resolve Codex user hooks through `$CODEX_HOME/hooks.json` with `~/.codex/hooks.json` as the default, matching Codex's existing-directory canonicalization rule. Preserve additive user TOML, project, plugin, and managed sources. Register exact `PreToolUse` and `PostToolUse` match-all command groups with ten-second handlers, preserve the complete native payload, propagate freeze and revocation through exit code `2`, keep guard lookup and audit delivery fail-open with observable errors, and commit user hooks plus all four runtime artifacts through one rollback-capable transaction. Require `/hooks` approval for both definition hashes. - Write Factory Droid hooks only to user-level `~/.factory/hooks.json`, its active legacy fallback, or the `hooks` object in `~/.factory/settings.json`; keep project and organization sources read-only. Preserve JSONC syntax, select sources per event, store direct event maps in hook files, keep settings hooks nested, validate JavaScript matchers before installation, use exact command ownership, and instruct users to review changes through `/hooks`. - Resolve Qwen Code user settings through explicit `QWEN_HOME`, `~/.qwen/.env`, then `~/.env`; keep workspace settings read-only. Preserve comments while rejecting trailing commas and duplicate keys, store hooks under the nested `hooks` object, treat `disableAllHooks` as unhealthy, validate JavaScript matchers before installation, use millisecond timeouts and explicit host shells, preserve PowerShell native exit codes, and instruct users to review changes through `/hooks`. diff --git a/README.md b/README.md index b11fc52..7666895 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,7 @@ Cursor installation writes native global `preToolUse`, `postToolUse`, and `postT Gemini CLI installation writes named, matchless `BeforeTool` and `AfterTool` command hooks to `$GEMINI_CLI_HOME/.gemini/settings.json` (`~/.gemini/settings.json` by default). Empty home overrides select the OS home; relative and literal-tilde values retain Gemini CLI's native path behavior. JSON comments and unrelated settings remain intact. Workspace, system-default, system-override, and extension hook sources remain unchanged and continue loading additively. User settings, generated runtimes, runtime config, and private key commit through one rollback-capable transaction. Managed handlers preserve the complete native snake_case payload, emit valid JSON on successful execution, propagate freeze and revocation through exit code `2`, and use encoded PowerShell commands on Windows. `hooksConfig.enabled` and `hooksConfig.disabled` remain authoritative. Run `/hooks list` after installation to inspect effective hooks from every source. -Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Elydora leaves the Documents and workspace hook roots unchanged. The wrappers preserve Cline's official payload and translate a frozen guard into JSON stdout cancellation control. +Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Documents, `.clinerules/hooks`, and workspace `.cline/hooks` remain unchanged and continue loading additively. Both hook files, generated Python runtimes, runtime config, and private key commit through one rollback-capable transaction. The wrappers preserve Cline 3's complete native payload, resolve Python through each runtime's absolute shebang, and emit the documented one-line JSON cancellation control. Kimi installation writes exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` rules to every detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the legacy Python CLI's `~/.kimi/config.toml`. A fresh installation targets stable Kimi Code; an existing legacy home activates the migration contract. Empty `KIMI_CODE_HOME` values use the default. The selected TOML files, generated runtimes, runtime config, and private key commit as one rollback-capable transaction. Managed commands preserve Kimi's native snake_case event JSON, propagate freeze and revocation through exit code `2`, and use encoded PowerShell on Windows. Run `/hooks` to inspect the global rules. diff --git a/elydora/plugins/cline.py b/elydora/plugins/cline.py index 35ed079..a6cbb07 100644 --- a/elydora/plugins/cline.py +++ b/elydora/plugins/cline.py @@ -2,107 +2,69 @@ from __future__ import annotations -import os +from typing import Tuple -from ._file_io import write_json_atomic, write_text_atomic from .base import AgentPlugin, InstallConfig, PluginStatus from .cline_contract import ( AGENT_KEY, - AUDIT_SCRIPT, HookFile, - build_metadata, - build_wrapper, - elydora_dir, + HookPaths, resolve_hook_files, runtime_contract, ) -from .cline_io import ( - PendingWrite, - read_hook_file, - remove_owned_hooks, - require_available_hook_file, - require_runtime, - runtime_files_exist, - write_hook_pair, +from .cline_installation import ( + commit_cline_installation, + commit_cline_uninstall, + preflight_cline_installation, + prepare_cline_installation, + prepare_cline_uninstall, ) -from .hook_template import generate_hook_script +from .cline_io import ( + read_hook_file, + require_available_hook_file, + runtime_files_exist, +) + + +def _read_hook_pair() -> Tuple[HookPaths, HookFile, HookFile]: + paths = resolve_hook_files() + return ( + paths, + read_hook_file(paths.guard_path), + read_hook_file(paths.audit_path), + ) + + +def _require_available_pair(guard_file: HookFile, audit_file: HookFile) -> None: + require_available_hook_file(guard_file) + require_available_hook_file(audit_file) class ClinePlugin(AgentPlugin): """Install Elydora into Cline's global file-hook directory.""" + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + _paths, guard_file, audit_file = _read_hook_pair() + _require_available_pair(guard_file, audit_file) + preflight_cline_installation(config, guard_file, audit_file) + def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - paths = resolve_hook_files() - guard_state = read_hook_file(paths.guard_path) - audit_state = read_hook_file(paths.audit_path) - require_available_hook_file(guard_state) - require_available_hook_file(audit_state) - - guard_path = config.get("guard_script_path", "") - require_runtime(guard_path, "Elydora guard runtime") - agent_directory = os.path.join(elydora_dir(), agent_id) - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - guard_metadata = build_metadata("guard", agent_id, guard_path) - audit_metadata = build_metadata("audit", agent_id, audit_path) - guard_source = build_wrapper(guard_metadata) - audit_source = build_wrapper(audit_metadata) - runtime_contract( - HookFile(True, paths.guard_path, guard_source, guard_metadata), - HookFile(True, paths.audit_path, audit_source, audit_metadata), - ) - - runtime_config = { - "org_id": config.get("org_id", ""), - "agent_id": agent_id, - "kid": config.get("kid", ""), - "base_url": config.get("base_url", "https://api.elydora.com"), - "token": config.get("token", ""), - "agent_name": AGENT_KEY, - } - audit_script = generate_hook_script( - org_id=config.get("org_id", ""), - agent_id=agent_id, - kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), - ) - write_json_atomic( - os.path.join(agent_directory, "config.json"), - runtime_config, - 0o600, - "Elydora runtime config", - ) - write_text_atomic( - os.path.join(agent_directory, "private.key"), - config.get("private_key", ""), - 0o600, - "Elydora private key", - ) - write_text_atomic( - audit_path, - audit_script, - 0o700, - "Elydora audit runtime", - ) - write_hook_pair( - PendingWrite(guard_state, guard_source), - PendingWrite(audit_state, audit_source), - ) + _paths, guard_file, audit_file = _read_hook_pair() + _require_available_pair(guard_file, audit_file) + changes = prepare_cline_installation(config, guard_file, audit_file) + commit_cline_installation(changes) print("Cline: user-level PreToolUse and PostToolUse hooks installed.") def uninstall(self, agent_id: str = "") -> None: - paths = resolve_hook_files() - guard_state = read_hook_file(paths.guard_path) - audit_state = read_hook_file(paths.audit_path) - remove_owned_hooks((guard_state, audit_state), agent_id) + _paths, guard_file, audit_file = _read_hook_pair() + changes = prepare_cline_uninstall((guard_file, audit_file), agent_id) + commit_cline_uninstall(changes) def status(self) -> PluginStatus: - paths = resolve_hook_files() - guard_state = read_hook_file(paths.guard_path) - audit_state = read_hook_file(paths.audit_path) - contract = runtime_contract(guard_state, audit_state) + paths, guard_file, audit_file = _read_hook_pair() + contract = runtime_contract(guard_file, audit_file) if contract is None: return PluginStatus( installed=False, @@ -115,7 +77,7 @@ class ClinePlugin(AgentPlugin): if installed else ( f"Configured at {paths.hooks_directory}; " - "runtime scripts missing" + "managed runtime incomplete" ) ) return PluginStatus( diff --git a/elydora/plugins/cline_contract.py b/elydora/plugins/cline_contract.py index 81be90c..44e5e7a 100644 --- a/elydora/plugins/cline_contract.py +++ b/elydora/plugins/cline_contract.py @@ -8,6 +8,8 @@ import json import os from typing import Any, Dict, Optional +from elydora._runtime_paths import resolve_agent_directory + AGENT_KEY = "cline" GUARD_FILE_NAME = "PreToolUse.mjs" @@ -177,7 +179,7 @@ async function main() { if (hookKind === 'guard' && result.code === 2) { const errorMessage = result.stderr.trim() || 'Agent is frozen by Elydora.'; - process.stdout.write('HOOK_CONTROL\\t' + JSON.stringify({ cancel: true, errorMessage }) + '\\n'); + process.stdout.write(JSON.stringify({ cancel: true, errorMessage }) + '\\n'); return; } if (result.signal) throw new Error('runtime terminated by signal ' + result.signal); @@ -251,11 +253,6 @@ def assert_wrapper_integrity(file: HookFile) -> None: ) -def _validate_agent_segment(agent_id: str) -> None: - if agent_id in {".", ".."} or os.path.basename(agent_id) != agent_id: - raise ValueError("Elydora Cline hook metadata contains an invalid agentId") - - def runtime_contract( guard_file: HookFile, audit_file: HookFile, @@ -270,8 +267,7 @@ def runtime_contract( raise ValueError("Elydora Cline hook files contain mismatched event metadata") if not same_agent_id(guard.agent_id, audit.agent_id): raise ValueError("Elydora Cline hook files reference different agents") - _validate_agent_segment(guard.agent_id) - agent_directory = os.path.join(elydora_dir(), guard.agent_id) + agent_directory = resolve_agent_directory(elydora_dir(), guard.agent_id) expected_guard = os.path.join(agent_directory, GUARD_SCRIPT) expected_audit = os.path.join(agent_directory, AUDIT_SCRIPT) if not _same_path(guard.runtime_path, expected_guard) or not _same_path( diff --git a/elydora/plugins/cline_installation.py b/elydora/plugins/cline_installation.py new file mode 100644 index 0000000..9fec152 --- /dev/null +++ b/elydora/plugins/cline_installation.py @@ -0,0 +1,232 @@ +"""Transactional Cline hook and runtime installation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +import sys +from typing import List, Optional, Sequence + +from elydora._runtime_paths import resolve_agent_directory + +from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES +from ._transaction import FileChange, file_change, source_change, write_changes +from .base import InstallConfig +from .cline_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + HookFile, + assert_wrapper_integrity, + build_metadata, + build_wrapper, + elydora_dir, + resolve_hook_files, + runtime_contract, + same_agent_id, +) +from .cline_io import ( + validate_api_origin, + validate_hook_tree, + validate_private_key, + validate_runtime_tree, +) +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script + + +@dataclass(frozen=True) +class ClineRuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +def _same_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def _runtime_config(config: InstallConfig, agent_id: str) -> dict: + value = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), + "agent_name": AGENT_KEY, + } + token = config.get("token") + if token: + value["token"] = token + return value + + +def _json_source(value: dict) -> str: + return json.dumps(value, indent=2) + "\n" + + +def _agent_paths(config: InstallConfig) -> ClineRuntimePaths: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + agent_directory = resolve_agent_directory(elydora_dir(), agent_id) + guard_path = os.path.join(agent_directory, GUARD_SCRIPT) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + if not _same_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + f"Elydora guard runtime must use the managed agent directory: {guard_path}" + ) + return ClineRuntimePaths(agent_id, agent_directory, guard_path, audit_path) + + +def _validate_install_config(config: InstallConfig) -> None: + for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): + value = config.get(field) + if not isinstance(value, str) or not value: + raise ValueError(f"{field} is required") + if config.get("agent_name") != AGENT_KEY: + raise ValueError(f"Cline installation requires agent_name {AGENT_KEY}") + validate_private_key(config["private_key"]) + validate_api_origin(config["base_url"]) + if "token" in config: + token = config["token"] + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string when provided") + if not os.path.isabs(sys.executable): + raise RuntimeError("Cline requires an absolute Python executable path") + _agent_paths(config) + + +def _validate_hook_files(guard_file: HookFile, audit_file: HookFile) -> None: + expected = resolve_hook_files() + if not _same_path(guard_file.file_path, expected.guard_path) or not _same_path( + audit_file.file_path, expected.audit_path + ): + raise ValueError("Cline installation received unexpected hook paths") + validate_hook_tree(expected.hooks_directory) + + +def preflight_cline_installation( + config: InstallConfig, + guard_file: HookFile, + audit_file: HookFile, +) -> ClineRuntimePaths: + _validate_install_config(config) + _validate_hook_files(guard_file, audit_file) + paths = _agent_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id) + return paths + + +def _runtime_change( + file_path: str, + label: str, + next_source: str, + mode: int, + maximum_bytes: int = MAX_SOURCE_BYTES, +) -> Optional[FileChange]: + return file_change(file_path, label, next_source, mode, maximum_bytes) + + +def _hook_change( + file: HookFile, label: str, next_source: Optional[str] +) -> Optional[FileChange]: + original = file.source if file.exists else None + if file.exists and original is None: + raise RuntimeError(f"{label} snapshot is missing source") + return source_change( + file.file_path, + label, + original, + next_source, + 0o700, + ) + + +def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def prepare_cline_installation( + config: InstallConfig, + guard_file: HookFile, + audit_file: HookFile, +) -> List[FileChange]: + paths = preflight_cline_installation(config, guard_file, audit_file) + guard_metadata = build_metadata("guard", paths.agent_id, paths.guard_path) + audit_metadata = build_metadata("audit", paths.agent_id, paths.audit_path) + guard_wrapper = build_wrapper(guard_metadata) + audit_wrapper = build_wrapper(audit_metadata) + runtime_contract( + HookFile(True, guard_file.file_path, guard_wrapper, guard_metadata), + HookFile(True, audit_file.file_path, audit_wrapper, audit_metadata), + ) + guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=paths.agent_id, + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + native_payload=True, + agent_name=AGENT_KEY, + ) + changes = _present( + ( + _runtime_change( + paths.guard_path, "Elydora guard runtime", guard_script, 0o700 + ), + _runtime_change( + os.path.join(paths.agent_directory, "config.json"), + "Elydora runtime config", + _json_source(_runtime_config(config, paths.agent_id)), + 0o600, + MAX_CONFIG_BYTES, + ), + _runtime_change( + os.path.join(paths.agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + _runtime_change( + paths.audit_path, "Elydora audit runtime", audit_script, 0o700 + ), + _hook_change(guard_file, "Cline PreToolUse hook", guard_wrapper), + _hook_change(audit_file, "Cline PostToolUse hook", audit_wrapper), + ) + ) + validate_runtime_tree(paths.agent_directory, paths.agent_id) + validate_hook_tree(os.path.dirname(guard_file.file_path)) + return changes + + +def commit_cline_installation(changes: Sequence[FileChange]) -> None: + write_changes(changes, "Install Cline hooks") + + +def prepare_cline_uninstall( + files: Sequence[HookFile], agent_id: str = "" +) -> List[FileChange]: + owned = [ + file + for file in files + if file.metadata is not None + and (not agent_id or same_agent_id(file.metadata.agent_id, agent_id)) + ] + changes: List[Optional[FileChange]] = [] + for file in owned: + metadata = file.metadata + assert metadata is not None + assert_wrapper_integrity(file) + changes.append( + _hook_change(file, f"Cline {metadata.kind} hook", None) + ) + return _present(changes) + + +def commit_cline_uninstall(changes: Sequence[FileChange]) -> None: + write_changes(changes, "Uninstall Cline hooks") diff --git a/elydora/plugins/cline_io.py b/elydora/plugins/cline_io.py index b678d4d..a3b5a87 100644 --- a/elydora/plugins/cline_io.py +++ b/elydora/plugins/cline_io.py @@ -1,53 +1,57 @@ -"""Fail-fast Cline hook I/O and runtime validation.""" +"""Physical Cline hook and managed runtime validation.""" from __future__ import annotations -from dataclasses import dataclass +import base64 import os -import tempfile -from typing import List, Sequence, Tuple +from typing import Any, Optional +import urllib.parse -from ._file_io import ( - read_json, - regular_file_exists, - remove_file, - require_runtime, - write_text_atomic, +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, ) +from ._strict_json import JsonObject, parse_json_object from .cline_contract import ( AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, HookFile, RuntimeContract, + elydora_dir, parse_metadata, same_agent_id, ) +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script -@dataclass(frozen=True) -class PendingWrite: - state: HookFile - source: str +def _same_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) -@dataclass(frozen=True) -class StagedFile: - state: HookFile - temporary_path: str +def validate_hook_tree(hooks_directory: str) -> None: + physical_directory_exists( + os.path.dirname(hooks_directory), "Cline configuration directory" + ) + physical_directory_exists(hooks_directory, "Cline hooks directory") def read_hook_file(file_path: str) -> HookFile: - try: - with open(file_path, "r", encoding="utf-8") as file: - source = file.read() - except FileNotFoundError: + validate_hook_tree(os.path.dirname(file_path)) + snapshot = read_physical_file(file_path, "Cline hook") + if snapshot is None: return HookFile(False, file_path) - except OSError as error: - raise OSError(f"Read Cline hook at {file_path}: {error}") from error return HookFile( exists=True, file_path=file_path, - source=source, - metadata=parse_metadata(file_path, source), + source=snapshot.contents, + metadata=parse_metadata(file_path, snapshot.contents), ) @@ -59,136 +63,194 @@ def require_available_hook_file(file: HookFile) -> None: ) -def _remove_temporary(path: str) -> None: +def validate_api_origin(value: str, label: str = "base_url") -> None: try: - os.remove(path) - except FileNotFoundError: - return - - -def _stage_file(write: PendingWrite) -> StagedFile: - directory = os.path.dirname(write.state.file_path) - try: - os.makedirs(directory, mode=0o700, exist_ok=True) - except OSError as error: - raise OSError(f"Create Cline hooks directory at {directory}: {error}") from error - descriptor = -1 - temporary_path = "" - try: - descriptor, temporary_path = tempfile.mkstemp( - prefix=f".{os.path.basename(write.state.file_path)}.", - suffix=".tmp", - dir=directory, - text=True, + parsed = urllib.parse.urlsplit(value) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error + invalid_character = "\\" in value or any( + character.isspace() or ord(character) < 32 for character in value + ) + if ( + parsed.scheme not in ("http", "https") + or not parsed.netloc + or hostname is None + or invalid_character + ): + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") + if ( + parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ValueError( + f"{label} must exclude credentials, query parameters, and fragments" ) - with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as file: - descriptor = -1 - file.write(write.source) - file.flush() - os.fsync(file.fileno()) - os.chmod(temporary_path, 0o700) - return StagedFile(write.state, temporary_path) - except Exception as error: - if descriptor >= 0: - try: - os.close(descriptor) - except OSError as close_error: - _remove_temporary(temporary_path) - raise OSError( - f"Stage Cline hook at {write.state.file_path}: {error}; " - f"close failed: {close_error}" - ) from error - _remove_temporary(temporary_path) - raise OSError( - f"Stage Cline hook at {write.state.file_path}: {error}" + + +def validate_private_key(value: str, label: str = "private_key") -> None: + try: + padded = value + "=" * ((4 - len(value) % 4) % 4) + seed = base64.b64decode( + padded.replace("-", "+").replace("_", "/"), validate=True + ) + canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") + except (ValueError, UnicodeEncodeError) as error: + raise ValueError( + f"{label} must be a canonical 32-byte base64url value" ) from error + if len(seed) != 32 or canonical != value: + raise ValueError(f"{label} must be a canonical 32-byte base64url value") -def _rollback_file(state: HookFile) -> None: - if state.exists: - if state.source is None: - raise RuntimeError("Existing Cline hook rollback state is missing source") - write_text_atomic( - state.file_path, - state.source, - 0o700, - "Cline hook rollback", +def _require_non_empty_string(value: Any, field: str, config_path: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"Elydora runtime config {field} is invalid: {config_path}") + return value + + +def _validate_runtime_config( + config: JsonObject, expected_agent_id: str, config_path: str +) -> None: + supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} + extra = next((key for key in config if key not in supported), None) + if extra is not None: + raise ValueError( + f'Elydora runtime config has unsupported field "{extra}": {config_path}' ) - else: - remove_file(state.file_path, "Cline hook rollback") + _require_non_empty_string(config.get("org_id"), "org_id", config_path) + _require_non_empty_string(config.get("kid"), "kid", config_path) + agent_id = _require_non_empty_string( + config.get("agent_id"), "agent_id", config_path + ) + if ( + not same_agent_id(agent_id, expected_agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + f"Elydora runtime identity does not match Cline hooks: {config_path}" + ) + if "token" in config: + _require_non_empty_string(config.get("token"), "token", config_path) + base_url = _require_non_empty_string( + config.get("base_url"), "base_url", config_path + ) + validate_api_origin(base_url, "Elydora runtime config base_url") -def write_hook_pair(guard: PendingWrite, audit: PendingWrite) -> None: - staged: List[StagedFile] = [] - try: - staged.append(_stage_file(guard)) - staged.append(_stage_file(audit)) - except Exception as error: - cleanup_errors: List[str] = [] - for item in staged: - try: - _remove_temporary(item.temporary_path) - except OSError as cleanup_error: - cleanup_errors.append(str(cleanup_error)) - suffix = f"; cleanup failed: {'; '.join(cleanup_errors)}" if cleanup_errors else "" - raise OSError(f"Stage Cline hook pair: {error}{suffix}") from error - - committed: List[StagedFile] = [] - try: - for item in staged: - os.replace(item.temporary_path, item.state.file_path) - committed.append(item) - except Exception as error: - recovery_errors: List[str] = [] - for item in reversed(committed): - try: - _rollback_file(item.state) - except (OSError, RuntimeError) as rollback_error: - recovery_errors.append(str(rollback_error)) - for item in staged[len(committed):]: - try: - _remove_temporary(item.temporary_path) - except OSError as cleanup_error: - recovery_errors.append(str(cleanup_error)) - suffix = f"; recovery failed: {'; '.join(recovery_errors)}" if recovery_errors else "" - raise OSError(f"Write Cline hook pair: {error}{suffix}") from error +def _read_runtime_config(file_path: str) -> Optional[JsonObject]: + snapshot = read_physical_file( + file_path, "Elydora runtime config", MAX_CONFIG_BYTES + ) + if snapshot is None: + return None + return parse_json_object( + snapshot.contents, f"Elydora runtime config at {file_path}" + ) -def remove_owned_hooks(files: Sequence[HookFile], agent_id: str = "") -> None: - for file in files: - owned_agent_id = file.metadata.agent_id if file.metadata else "" - if not owned_agent_id or ( - agent_id and not same_agent_id(owned_agent_id, agent_id) - ): - continue - remove_file(file.file_path, "Cline hook") +def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: + root = elydora_dir() + if not physical_directory_exists(root, "Elydora runtime directory"): + return + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return + config_path = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path) + artifacts = ( + ("private.key", "Elydora private key"), + (GUARD_SCRIPT, "Elydora guard runtime"), + (AUDIT_SCRIPT, "Elydora audit runtime"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), + ) + artifact_exists = any( + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + configured_id = config.get("agent_id") + if ( + not isinstance(configured_id, str) + or not same_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime config identity does not match Cline agent " + f"{agent_id}: {config_path}" + ) + + +def _valid_contract_paths(contract: RuntimeContract) -> bool: + return ( + _same_path(os.path.dirname(contract.agent_directory), elydora_dir()) + and _same_path( + contract.guard_path, + os.path.join(contract.agent_directory, GUARD_SCRIPT), + ) + and _same_path( + contract.audit_path, + os.path.join(contract.agent_directory, AUDIT_SCRIPT), + ) + ) def runtime_files_exist(contract: RuntimeContract) -> bool: - config_path = os.path.join(contract.agent_directory, "config.json") - config = read_json(config_path, "Elydora runtime config") - if config is None: + if not _valid_contract_paths(contract): return False - config_agent_id = config.get("agent_id") - if ( - config.get("agent_name") != AGENT_KEY - or not isinstance(config_agent_id, str) - or not same_agent_id(config_agent_id, contract.agent_id) + root = elydora_dir() + if not physical_directory_exists(root, "Elydora runtime directory"): + return False + if not physical_directory_exists( + contract.agent_directory, "Elydora agent runtime directory" ): return False - files: Tuple[Tuple[str, str], ...] = ( - (contract.guard_path, "Elydora guard runtime"), - (contract.audit_path, "Elydora audit runtime"), + config_path = os.path.join(contract.agent_directory, "config.json") + key_path = os.path.join(contract.agent_directory, "private.key") + config = _read_runtime_config(config_path) + key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) + guard = read_physical_file(contract.guard_path, "Elydora guard runtime") + audit = read_physical_file(contract.audit_path, "Elydora audit runtime") + if any(item is None for item in (config, key, guard, audit)): + return False + assert config is not None + assert key is not None + assert guard is not None + assert audit is not None + _validate_runtime_config(config, contract.agent_id, config_path) + validate_private_key(key.contents, "Elydora private key") + expected_audit = generate_hook_script( + org_id="", + agent_id=contract.agent_id, + kid="", + base_url="", + native_payload=True, + agent_name=AGENT_KEY, + ) + return ( + guard.contents == generate_guard_script(AGENT_KEY, contract.agent_id) + and audit.contents == expected_audit ) - return all(regular_file_exists(path, label) for path, label in files) __all__ = [ - "PendingWrite", "read_hook_file", - "remove_owned_hooks", "require_available_hook_file", - "require_runtime", "runtime_files_exist", - "write_hook_pair", + "validate_api_origin", + "validate_hook_tree", + "validate_private_key", + "validate_runtime_tree", ] diff --git a/tests/cline_support.py b/tests/cline_support.py new file mode 100644 index 0000000..a841242 --- /dev/null +++ b/tests/cline_support.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import base64 +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 +import sys +from threading import Thread +from typing import Any, Mapping + +import pytest + +from elydora.plugins import cline_contract as contract +from elydora.plugins.base import InstallConfig +from elydora.plugins.cline import ClinePlugin + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64.urlsafe_b64encode(bytes(range(32))).rstrip(b"=").decode() + + +@dataclass +class ClineFixture: + root_dir: Path + home_dir: Path + project_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 + private_key: Path + plugin: ClinePlugin + config: InstallConfig + + def install(self) -> None: + self.plugin.install(self.config) + + def environment(self) -> dict[str, str]: + return { + **os.environ, + "HOME": str(self.home_dir), + "USERPROFILE": str(self.home_dir), + "CLINE_DIR": str(self.cline_dir), + } + + def managed_paths(self) -> tuple[Path, ...]: + return ( + self.guard_path, + self.runtime_config, + self.private_key, + self.audit_path, + self.guard_wrapper, + self.audit_wrapper, + ) + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls, status: Any) -> None: + cls.agent_status = status + 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: # noqa: N802 + type(self).requests.append( + { + "method": "GET", + "path": self.path, + "authorization": self.headers.get("Authorization"), + } + ) + self._respond(200, {"agent": {"status": type(self).agent_status}}) + + def do_POST(self) -> None: # noqa: N802 + 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(201, {"operation": {"accepted": True}}) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server(*, status: Any = "active") -> ThreadingHTTPServer: + ElydoraApiHandler.reset(status) + 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", newline="") + 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 python_script(body: str) -> str: + return f"#!{sys.executable}\n{body}" + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_guard: object = MISSING, + existing_audit: object = MISSING, + base_url: str = "http://127.0.0.1:9", +) -> ClineFixture: + home_dir = tmp_path / "home with spaces and 'quote %CLINE%" + project_dir = tmp_path / "project with spaces" + cline_dir = tmp_path / "custom-cline-home" + hooks_dir = cline_dir / "hooks" + agent_dir = home_dir / ".elydora" / AGENT_ID + project_dir.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.setenv("CLINE_DIR", str(cline_dir)) + monkeypatch.chdir(project_dir) + guard_wrapper = hooks_dir / contract.GUARD_FILE_NAME + audit_wrapper = hooks_dir / contract.AUDIT_FILE_NAME + if existing_guard is not MISSING: + write_text(guard_wrapper, str(existing_guard), 0o700) + if existing_audit is not MISSING: + write_text(audit_wrapper, str(existing_audit), 0o700) + guard_path = agent_dir / contract.GUARD_SCRIPT + audit_path = agent_dir / contract.AUDIT_SCRIPT + return ClineFixture( + root_dir=tmp_path, + home_dir=home_dir, + project_dir=project_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", + private_key=agent_dir / "private.key", + plugin=ClinePlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "cline", + "private_key": VALID_PRIVATE_KEY, + "kid": "key-1", + "token": "token-1", + "base_url": base_url, + "guard_script_path": str(guard_path), + }, + ) + + +def run_wrapper( + wrapper: Path, + fixture: ClineFixture, + payload: bytes, +) -> subprocess.CompletedProcess[bytes]: + node = shutil.which("node") + assert node is not None + return subprocess.run( # nosec B603 + [node, str(wrapper)], + cwd=fixture.project_dir, + env=fixture.environment(), + input=payload, + capture_output=True, + check=False, + timeout=10, + ) + + +def assert_no_transaction_files(root: Path) -> None: + assert not list(root.rglob("*.tmp")) + assert not list(root.rglob("*.rollback")) + + +def snapshot_installation(fixture: ClineFixture) -> Mapping[Path, bytes]: + return {path: path.read_bytes() for path in fixture.managed_paths()} + + +def assert_snapshot(snapshot: Mapping[Path, bytes]) -> None: + for path, source in snapshot.items(): + assert path.read_bytes() == source + + +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}") diff --git a/tests/test_cline_official.py b/tests/test_cline_official.py new file mode 100644 index 0000000..fd9e8e8 --- /dev/null +++ b/tests/test_cline_official.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 + +import pytest + +from cline_support import prepare_fixture + + +CLINE_CORE_ENTRY = os.environ.get("ELYDORA_CLINE_CORE_ENTRY") +CLINE_ENTRY = os.environ.get("ELYDORA_CLINE_ENTRY") + + +@pytest.mark.skipif( + not (CLINE_CORE_ENTRY and CLINE_ENTRY), + reason=( + "set ELYDORA_CLINE_CORE_ENTRY and ELYDORA_CLINE_ENTRY " + "to official Cline files" + ), +) +def test_official_cline_loader_discovers_both_managed_hooks( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + node = shutil.which("node") + assert node is not None + assert CLINE_CORE_ENTRY is not None + assert CLINE_ENTRY is not None + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + version = subprocess.run( # nosec B603 + [node, CLINE_ENTRY, "--version"], + cwd=fixture.project_dir, + env=fixture.environment(), + capture_output=True, + check=False, + text=True, + timeout=10, + ) + assert version.returncode == 0, version.stderr + assert version.stdout.strip() == "3.0.46" + + source = """ +import { pathToFileURL } from 'node:url'; +const { listHookConfigFiles } = await import( + pathToFileURL(process.env.ELYDORA_CLINE_CORE_ENTRY).href +); +console.log(JSON.stringify(listHookConfigFiles(process.env.ELYDORA_WORKSPACE))); +""" + environment = { + **fixture.environment(), + "ELYDORA_CLINE_CORE_ENTRY": CLINE_CORE_ENTRY, + "ELYDORA_WORKSPACE": str(fixture.project_dir), + } + loaded = subprocess.run( # nosec B603 + [node, "--input-type=module", "--eval", source], + cwd=fixture.project_dir, + env=environment, + capture_output=True, + check=False, + text=True, + timeout=10, + ) + + assert loaded.returncode == 0, loaded.stderr + assert json.loads(loaded.stdout) == [ + { + "fileName": "PostToolUse", + "hookEventName": "tool_result", + "path": str(fixture.audit_wrapper), + }, + { + "fileName": "PreToolUse", + "hookEventName": "tool_call", + "path": str(fixture.guard_wrapper), + }, + ] diff --git a/tests/test_cline_plugin.py b/tests/test_cline_plugin.py index 6c175e7..a7faec1 100644 --- a/tests/test_cline_plugin.py +++ b/tests/test_cline_plugin.py @@ -1,143 +1,66 @@ 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 cline_support import ( + AGENT_ID, + ElydoraApiHandler, + MISSING, + assert_no_transaction_files, + prepare_fixture, + python_script, + run_wrapper, + server_base_url, + start_api_server, + write_text, +) 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() +def encoded(value: object) -> bytes: + return json.dumps(value, separators=(",", ":"), ensure_ascii=False).encode() -@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), +def official_payload(event: str) -> dict[str, object]: + nested = { + "id": "tool-call-1", + "name": "read_file", + "input": {"path": "README.md", "nested": {"preserve": True}}, } - 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, - ) + value: dict[str, object] = { + "clineVersion": "3.0.46", + "hookName": event, + "timestamp": "2026-07-19T12:00:00.000Z", + "taskId": "task-1", + "workspaceRoots": ["/workspace"], + "userId": "user-1", + "futureField": {"survives": ["exactly", 2]}, + } + if event == "tool_call": + value["tool_call"] = nested + value["preToolUse"] = { + "toolName": nested["name"], + "parameters": nested["input"], + } + else: + value["tool_result"] = { + **nested, + "output": "ok", + "durationMs": 12, + } + value["postToolUse"] = { + "toolName": nested["name"], + "parameters": nested["input"], + "result": "ok", + "success": True, + "executionTimeMs": 12, + } + return value def test_cline_is_registered_in_the_sdk_and_cli() -> None: @@ -147,35 +70,46 @@ def test_cline_is_registered_in_the_sdk_and_cli() -> None: "config_path": "~/.cline/hooks/PreToolUse.mjs", } assert cli.PLUGIN_MAP["cline"] is cline.ClinePlugin + assert cline.ClinePlugin.manages_guard_runtime is True -def test_install_writes_only_native_global_hooks_and_is_idempotent( +def test_install_commits_all_six_files_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.install() + before = {path: path.read_bytes() for path in fixture.managed_paths()} - fixture.plugin.install(fixture.config) + fixture.install() - 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 {path: path.read_bytes() for path in fixture.managed_paths()} == before + assert fixture.guard_wrapper.read_text(encoding="utf-8").startswith( + "#!/usr/bin/env node\n// @elydora-cline-hook " + ) + assert json.loads(fixture.runtime_config.read_text(encoding="utf-8")) == { + "org_id": "org-1", + "agent_id": AGENT_ID, + "kid": "key-1", + "base_url": "http://127.0.0.1:9", + "agent_name": "cline", + "token": "token-1", + } 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() + assert not (fixture.project_dir / ".cline" / "hooks").exists() + assert not (fixture.project_dir / ".clinerules" / "hooks").exists() + assert_no_transaction_files(fixture.home_dir) + assert_no_transaction_files(fixture.cline_dir) 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) + fixture = prepare_fixture(monkeypatch, tmp_path) monkeypatch.delenv("CLINE_DIR") - fixture.plugin.install(fixture.config) + fixture.install() default_hooks = fixture.home_dir / ".cline" / "hooks" assert (default_hooks / "PreToolUse.mjs").is_file() @@ -183,184 +117,234 @@ def test_install_uses_official_default_without_cline_dir( assert not fixture.guard_wrapper.exists() -def test_wrappers_translate_freezes_and_forward_payload_byte_for_byte( +def test_wrappers_preserve_bytes_and_emit_pure_json_control( 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( + fixture.install() + guard_capture = tmp_path / "guard-input.json" + audit_capture = tmp_path / "audit-input.json" + write_text( + fixture.guard_path, + 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", + "import sys\n" + f"Path({str(guard_capture)!r}).write_bytes(sys.stdin.buffer.read())\n" + "sys.stderr.write('Agent is frozen by Elydora.\\n')\n" + "raise SystemExit(2)\n" + ), + 0o700, ) - 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) + write_text( + fixture.audit_path, + python_script( + "from pathlib import Path\n" + "import sys\n" + f"Path({str(audit_capture)!r}).write_bytes(sys.stdin.buffer.read())\n" + ), + 0o700, + ) + pre_raw = encoded(official_payload("tool_call")) + post_raw = encoded(official_payload("tool_result")) + + guard = run_wrapper(fixture.guard_wrapper, fixture, pre_raw) + audit = run_wrapper(fixture.audit_wrapper, fixture, post_raw) + 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) == { + assert json.loads(guard.stdout) == { "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 b"HOOK_CONTROL" not in guard.stdout + assert guard_capture.read_bytes() == pre_raw 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)} + assert audit.stdout == b"" + assert audit_capture.read_bytes() == post_raw + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_generated_guard_translates_blocking_status_to_cline_control( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, +) -> None: + server = start_api_server(status=status) + try: + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + fixture.install() + result = run_wrapper( + fixture.guard_wrapper, + fixture, + encoded(official_payload("tool_call")), + ) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0 + assert json.loads(result.stdout)["cancel"] is True + assert status.encode() in result.stderr.lower() 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"), + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + write_text( + fixture.guard_path, + python_script("import sys\nsys.stdin.buffer.read()\n"), + 0o700, ) - result = run_wrapper(passing.guard_wrapper, passing, {}) - assert result.returncode == 0 - assert result.stdout == "" + passing = run_wrapper(fixture.guard_wrapper, fixture, b"{}") + assert passing.returncode == 0 + assert passing.stdout == b"" - failing = prepare_fixture(monkeypatch, tmp_path / "failing") - failing.audit_path.write_text( - _python_script( + write_text( + fixture.audit_path, + python_script( "import sys\n" "sys.stderr.write('audit failed\\n')\n" "raise SystemExit(7)\n" ), - encoding="utf-8", + 0o700, ) - 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 + failing = run_wrapper(fixture.audit_wrapper, fixture, b"{}") + assert failing.returncode == 1 + assert b"audit failed" in failing.stderr + assert b"exited with code 7" in failing.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", + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + write_text( + fixture.guard_path, + "#!python.exe\nimport sys\nsys.stdin.buffer.read()\n", + 0o700, ) - result = run_wrapper(fixture.guard_wrapper, fixture, {}) + result = run_wrapper(fixture.guard_wrapper, fixture, b"{}") assert result.returncode == 1 - assert result.stdout == "" - assert "runtime shebang executable must be absolute" in result.stderr + assert result.stdout == b"" + assert b"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( +def test_audit_submits_the_complete_native_event( 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}" + server = start_api_server() + payload = official_payload("tool_result") 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"} + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + fixture.install() + result = run_wrapper(fixture.audit_wrapper, fixture, encoded(payload)) 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", + assert result.returncode == 0 + request = next( + item for item in ElydoraApiHandler.requests if item["method"] == "POST" ) - with pytest.raises(ValueError, match="managed template"): - fixture.plugin.status() + operation = request["json"] + assert operation["payload"] == payload + assert operation["action"] == {"tool": "read_file"} + assert operation["subject"] == {"session_id": "task-1"} + assert request["authorization"] == "Bearer token-1" - fixture.plugin.install(fixture.config) - fixture.runtime_config.write_text("{ malformed", encoding="utf-8") - with pytest.raises(ValueError, match="parse Elydora runtime config"): + +@pytest.mark.parametrize( + ("target", "expected"), + [ + ("guard", None), + ("audit", None), + ("config", "parse Elydora runtime config"), + ("key", "private key.*canonical 32-byte"), + ("wrapper", "managed template"), + ], +) +def test_status_requires_exact_physical_runtime_identity( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + target: str, + expected: str | None, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.plugin.status()["installed"] is True + if target == "guard": + write_text(fixture.guard_path, "tampered\n", 0o700) + elif target == "audit": + write_text(fixture.audit_path, "tampered\n", 0o700) + elif target == "config": + write_text(fixture.runtime_config, "{ malformed") + elif target == "key": + write_text(fixture.private_key, "invalid") + else: + source = fixture.guard_wrapper.read_text(encoding="utf-8") + write_text(fixture.guard_wrapper, source + "// tampered\n", 0o700) + + if expected is None: + assert fixture.plugin.status()["installed"] is False + else: + with pytest.raises(ValueError, match=expected): + fixture.plugin.status() + + +@pytest.mark.parametrize( + ("source", "pattern"), + [ + ('{"agent_name":"cline","agent_name":"cline"}', "duplicate field"), + ( + json.dumps( + { + "org_id": "org-1", + "agent_id": "other-agent", + "kid": "key-1", + "base_url": "https://api.elydora.com", + "agent_name": "cline", + } + ), + "identity does not match", + ), + ( + json.dumps( + { + "org_id": "org-1", + "agent_id": AGENT_ID, + "kid": "key-1", + "base_url": "https://api.elydora.com", + "agent_name": "cline", + "hidden": True, + } + ), + "unsupported field", + ), + ], +) +def test_status_surfaces_invalid_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + source: str, + pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + write_text(fixture.runtime_config, source) + + with pytest.raises(ValueError, match=pattern): fixture.plugin.status() @@ -370,25 +354,21 @@ def test_install_rejects_user_filename_collisions_before_writes( 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 + guard = "// user PreToolUse hook\n" if collision == "guard" else MISSING + 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, + existing_guard=guard, + existing_audit=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 + fixture.install() + + assert not fixture.agent_dir.exists() + path = fixture.guard_wrapper if collision == "guard" else fixture.audit_wrapper + assert path.read_text(encoding="utf-8").startswith("// user") def test_install_preserves_corrupt_metadata_for_recovery( @@ -396,88 +376,48 @@ def test_install_preserves_corrupt_metadata_for_recovery( 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, - ) + fixture = prepare_fixture(monkeypatch, tmp_path, existing_guard=corrupt) + with pytest.raises(ValueError, match="parse Elydora Cline hook metadata"): - fixture.plugin.install(fixture.config) + fixture.install() + assert fixture.guard_wrapper.read_text(encoding="utf-8") == corrupt - assert fixture.audit_wrapper.exists() is False - assert fixture.audit_path.exists() is False + assert not fixture.agent_dir.exists() -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( +def test_uninstall_removes_exact_ownership_and_preserves_adjacent_hooks( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() user_hook = fixture.hooks_dir / "PreToolUse.py" - user_hook.write_text("# user hook\n", encoding="utf-8") + write_text(user_hook, "# user hook\n") 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 not fixture.guard_wrapper.exists() + assert not fixture.audit_wrapper.exists() assert user_hook.read_text(encoding="utf-8") == "# user hook\n" -def test_hook_pair_restores_the_first_hook_when_the_second_commit_fails( +def test_runtime_failures_remain_observable_and_fail_open( 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)) + fixture.install() + guard = run_wrapper( + fixture.guard_wrapper, + fixture, + encoded(official_payload("tool_call")), ) - replacement_audit = cline_contract.build_wrapper( - cline_contract.build_metadata("audit", "agent-2", str(fixture.audit_path)) - ) - real_replace = os.replace - failure_injected = False + audit = run_wrapper(fixture.audit_wrapper, fixture, b"{ malformed") - 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")) == [] + assert guard.returncode == 0 + assert b"Failed to resolve agent status" in guard.stderr + assert audit.returncode == 0 + assert b"invalid JSON" in (fixture.agent_dir / "error.log").read_bytes() diff --git a/tests/test_cline_transaction.py b/tests/test_cline_transaction.py new file mode 100644 index 0000000..79cfa8d --- /dev/null +++ b/tests/test_cline_transaction.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import os +from pathlib import Path +import subprocess # nosec B404 +import sys + +import pytest + +from cline_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + assert_no_transaction_files, + assert_snapshot, + prepare_fixture, + snapshot_installation, + symlink_or_skip, + write_json, + write_text, +) +from elydora.plugins import _transaction, cline_installation +from elydora.plugins.cline_contract import resolve_hook_files +from elydora.plugins.cline_io import read_hook_file + + +def prepare_installation(fixture: object) -> list[_transaction.FileChange]: + paths = resolve_hook_files() + guard = read_hook_file(paths.guard_path) + audit = read_hook_file(paths.audit_path) + return cline_installation.prepare_cline_installation( + fixture.config, guard, audit + ) + + +def test_transaction_rolls_back_all_six_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + real_replace = _transaction.os.replace + failed = False + + def fail_audit_once(source: str, destination: str) -> None: + nonlocal failed + if os.path.abspath(destination) == str(fixture.audit_wrapper) and not failed: + failed = True + raise OSError("injected Cline audit hook failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_audit_once) + with pytest.raises(OSError, match="injected Cline audit hook failure"): + fixture.install() + + assert failed is True + assert all(not path.exists() for path in fixture.managed_paths()) + assert_no_transaction_files(fixture.home_dir) + assert_no_transaction_files(fixture.cline_dir) + + +def test_concurrent_hook_replacement_is_detected_before_final_commit( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + changes = prepare_installation(fixture) + concurrent = "// concurrently replaced\n" + real_replace = _transaction.os.replace + injected = False + + def inject_replacement(source: str, destination: str) -> None: + nonlocal injected + if not injected: + injected = True + write_text(fixture.audit_wrapper, concurrent, 0o700) + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", inject_replacement) + with pytest.raises(OSError, match="changed during installation"): + cline_installation.commit_cline_installation(changes) + + assert injected is True + assert fixture.audit_wrapper.read_text(encoding="utf-8") == concurrent + for path in fixture.managed_paths()[:-1]: + assert not path.exists() + assert_no_transaction_files(fixture.home_dir) + assert_no_transaction_files(fixture.cline_dir) + + +def test_stale_hook_snapshot_is_rejected_before_staging( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + paths = resolve_hook_files() + guard = read_hook_file(paths.guard_path) + audit = read_hook_file(paths.audit_path) + concurrent = "// stale snapshot replacement\n" + write_text(fixture.audit_wrapper, concurrent, 0o700) + + with pytest.raises(OSError, match="changed before staging"): + cline_installation.prepare_cline_installation( + fixture.config, guard, audit + ) + + assert fixture.audit_wrapper.read_text(encoding="utf-8") == concurrent + assert not fixture.agent_dir.exists() + assert_no_transaction_files(fixture.root_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_fail_before_hook_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 not fixture.guard_wrapper.exists() + assert not fixture.audit_wrapper.exists() + + +def test_mismatched_runtime_identity_fails_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + write_json( + fixture.runtime_config, + {"agent_id": "another-agent", "agent_name": "cline"}, + ) + original = fixture.runtime_config.read_text(encoding="utf-8") + + with pytest.raises(ValueError, match="identity does not match"): + fixture.install() + + assert fixture.runtime_config.read_text(encoding="utf-8") == original + assert not fixture.guard_wrapper.exists() + + +@pytest.mark.parametrize( + "kind", ["configuration", "hooks", "runtime", "hook"] +) +def test_linked_configuration_runtime_and_hook_paths_are_rejected( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + target = tmp_path / f"{kind}-target" + if kind == "configuration": + target.mkdir() + symlink_or_skip(target, fixture.cline_dir, directory=True) + elif kind == "hooks": + target.mkdir() + fixture.cline_dir.mkdir() + symlink_or_skip(target, fixture.hooks_dir, directory=True) + elif kind == "runtime": + target.mkdir() + fixture.home_dir.mkdir() + symlink_or_skip( + target, fixture.home_dir / ".elydora", directory=True + ) + else: + fixture.install() + write_text(target, "external hook\n") + fixture.guard_wrapper.unlink() + symlink_or_skip(target, fixture.guard_wrapper) + + with pytest.raises(OSError, match="physical (file|directory)"): + fixture.install() + + +@pytest.mark.parametrize( + ("field", "value", "pattern"), + [ + ("agent_name", "codex", "requires agent_name cline"), + ("agent_id", "../escape", "Invalid agent ID"), + ("private_key", "invalid", "canonical 32-byte"), + ("token", "", "non-empty string"), + ("base_url", "https://api.elydora.com?q=1", "query parameters"), + ("guard_script_path", "outside", "managed agent directory"), + ], +) +def test_install_validates_runtime_inputs_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + field: str, + value: str, + pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config[field] = value + + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert not fixture.agent_dir.exists() + assert not fixture.hooks_dir.exists() + + +def test_uninstall_restores_both_hooks_when_second_removal_fails( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + before = snapshot_installation(fixture) + paths = resolve_hook_files() + changes = cline_installation.prepare_cline_uninstall( + ( + read_hook_file(paths.guard_path), + read_hook_file(paths.audit_path), + ), + AGENT_ID, + ) + real_replace = _transaction.os.replace + failed = False + + def fail_audit_removal(source: str, destination: str) -> None: + nonlocal failed + if os.path.abspath(source) == str(fixture.audit_wrapper) and not failed: + failed = True + raise OSError("injected Cline uninstall failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_audit_removal) + with pytest.raises(OSError, match="injected Cline uninstall failure"): + cline_installation.commit_cline_uninstall(changes) + + assert failed is True + assert_snapshot(before) + assert_no_transaction_files(fixture.root_dir) + + +def test_installation_leaves_no_transaction_artifacts( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert_no_transaction_files(fixture.root_dir) + + +def test_cli_preflight_preserves_a_hook_collision(tmp_path: Path) -> None: + home_dir = tmp_path / "cli-home" + hook_path = home_dir / ".cline" / "hooks" / "PreToolUse.mjs" + key_path = tmp_path / "private.key" + write_text(hook_path, "// user hook\n", 0o700) + write_text(key_path, VALID_PRIVATE_KEY) + environment = { + key: value + for key, value in { + **os.environ, + "HOME": str(home_dir), + "USERPROFILE": str(home_dir), + }.items() + if key != "CLINE_DIR" + } + result = subprocess.run( # nosec B603 + [ + sys.executable, + "-m", + "elydora.cli", + "install", + "--agent", + "cline", + "--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=environment, + text=True, + ) + + assert result.returncode == 1 + assert "owned by another integration" in result.stderr + assert hook_path.read_text(encoding="utf-8") == "// user hook\n" + assert not (home_dir / ".elydora").exists()