diff --git a/AGENTS.md b/AGENTS.md index 392dbda..0ffa768 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -30,7 +30,7 @@ This repository publishes the open-source server, console, integration catalog, - 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 each SDK's 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`. - All three SDKs commit Auggie settings, both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Their status checks require exact runtime identity, canonical private keys, physical files, and exact wrapper sources. Installation validates matcher syntax with JavaScript `RegExp`; status and uninstall stay available for offline recovery. Verify the effective configuration with `auggie tools list`. -- Treat Cline 3 CLI file hooks as the SDK 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 and use supported event filename extensions. 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, and exact wrapper sources. Translate guard exit code `2` into one-line JSON stdout with `cancel: true`; hook errors, timeouts, and invalid control JSON fail open. Model Cline IDE as a separate variant with runtime `--hooks-dir`, Documents, and `.clinerules/hooks` roots, exact `.ps1` filenames on Windows, and executable extensionless filenames on macOS and Linux. +- Treat Cline 3 CLI file hooks as the SDK 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 and use supported event filename extensions. 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, and exact generated runtime and wrapper sources. Node wrappers use `process.execPath`; Python wrappers resolve the runtime executable from an absolute shebang. Translate guard exit code `2` into one-line JSON stdout with `cancel: true`; hook errors, timeouts, and invalid control JSON fail open. Model Cline IDE as a separate variant with runtime `--hooks-dir`, Documents, and `.clinerules/hooks` roots, exact `.ps1` filenames on Windows, and executable extensionless filenames on macOS and Linux. - Resolve Codex user hooks through `$CODEX_HOME/hooks.json` with `~/.codex/hooks.json` as the default, matching Codex's existing-directory canonicalization rule. Model user JSON, inline TOML layers, project files, managed `requirements.toml`, and plugin-bundled hooks as additive stable sources. SDK adapters write only the user JSON source, register exact `PreToolUse` and `PostToolUse` match-all command groups, preserve the complete native payload, propagate freeze and revocation through exit code `2`, keep guard lookup and audit delivery fail-open with observable errors, commit user hooks and all four runtime artifacts through one rollback-capable transaction, and require `/hooks` trust review for both definition hashes. - Write GitHub Copilot CLI hooks only to `$COPILOT_HOME/hooks/elydora-audit.json` with `~/.copilot/hooks/elydora-audit.json` as the default. Migrate exact Elydora-owned entries from project `.github/hooks/hooks.json`, preserve native camelCase input, require `disableAllHooks` to be false, and commit runtime plus hook documents in one rollback-capable transaction. - Model Cursor as one CLI-and-IDE hook contract. Preserve user, project, and enterprise sources in documented priority order; forward `tool_name`, `tool_input`, and `conversation_id`; use exit code `2` for denial; record `failClosed` as the per-script failure-mode control. diff --git a/sdks/python/README.md b/sdks/python/README.md index c93af73..dd96948 100644 --- a/sdks/python/README.md +++ b/sdks/python/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/sdks/python/elydora/plugins/cline.py b/sdks/python/elydora/plugins/cline.py index 35ed079..a6cbb07 100644 --- a/sdks/python/elydora/plugins/cline.py +++ b/sdks/python/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/sdks/python/elydora/plugins/cline_contract.py b/sdks/python/elydora/plugins/cline_contract.py index 81be90c..44e5e7a 100644 --- a/sdks/python/elydora/plugins/cline_contract.py +++ b/sdks/python/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/sdks/python/elydora/plugins/cline_installation.py b/sdks/python/elydora/plugins/cline_installation.py new file mode 100644 index 0000000..9fec152 --- /dev/null +++ b/sdks/python/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/sdks/python/elydora/plugins/cline_io.py b/sdks/python/elydora/plugins/cline_io.py index b678d4d..a3b5a87 100644 --- a/sdks/python/elydora/plugins/cline_io.py +++ b/sdks/python/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/sdks/python/tests/cline_support.py b/sdks/python/tests/cline_support.py new file mode 100644 index 0000000..a841242 --- /dev/null +++ b/sdks/python/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/sdks/python/tests/test_cline_official.py b/sdks/python/tests/test_cline_official.py new file mode 100644 index 0000000..fd9e8e8 --- /dev/null +++ b/sdks/python/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/sdks/python/tests/test_cline_plugin.py b/sdks/python/tests/test_cline_plugin.py index 6c175e7..a7faec1 100644 --- a/sdks/python/tests/test_cline_plugin.py +++ b/sdks/python/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/sdks/python/tests/test_cline_transaction.py b/sdks/python/tests/test_cline_transaction.py new file mode 100644 index 0000000..79cfa8d --- /dev/null +++ b/sdks/python/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()