diff --git a/AGENTS.md b/AGENTS.md index 799a019..ff5e495 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,6 +22,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Forward official hook JSON from STDIN without reshaping provider fields. - Use the provider's documented blocking mechanism. Command-hook providers that define exit code `2` must receive exit code `2` from the freeze guard. - Report installation as healthy only when a complete hook contract references both generated runtime scripts and both scripts exist. +- Write Cursor hooks only to `~/.cursor/hooks.json`; keep project and enterprise sources read-only. Preserve user hooks, migrate the prior versionless Elydora contract, use exact native command ownership, retain PowerShell exit code `2`, and set security-critical handlers to `failClosed` with a bounded timeout. - Model stable, legacy, and early-access hook generations as explicit contracts. Keep their activation requirements visible in CLI output and README guidance. - Select Kimi Code and legacy kimi-cli config targets from existing runtime evidence; a fresh installation targets current Kimi Code and keeps migration markers isolated to active runtimes. - Keep Grok Build writes inside its native global `$GROK_HOME/hooks/*.json` contract. Treat Claude Code and Cursor compatibility files plus project `.grok/hooks` as read-only integration sources. diff --git a/README.md b/README.md index d34c2f1..f7428a4 100644 --- a/README.md +++ b/README.md @@ -80,6 +80,8 @@ Credential options may be omitted in an interactive terminal; the CLI then reads Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex after installation and trust the Elydora `PreToolUse` and `PostToolUse` definitions. +Cursor installation writes native global `preToolUse` and `postToolUse` handlers to `~/.cursor/hooks.json`. Elydora preserves unrelated user hooks, leaves project and enterprise sources unchanged, and configures a 10-second fail-closed command boundary with exact freeze exit-code propagation. + 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. Kimi installation writes the strict hook contract to each detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the migrating Python CLI's `~/.kimi/config.toml`. A fresh installation targets current Kimi Code, avoiding cross-runtime migration markers. Both runtimes load the hooks globally; run `/hooks` to inspect them. diff --git a/elydora/cli.py b/elydora/cli.py index 3c34d84..315d435 100644 --- a/elydora/cli.py +++ b/elydora/cli.py @@ -204,21 +204,7 @@ def cmd_install(args: argparse.Namespace) -> None: print(f"Verified key pair (public key: {pub[:16]}...)") - # Create per-agent directory under ~/.elydora/{agent_id}/ - ensure_private_directory(_runtime_root()) - ensure_private_directory(agent_dir) - - # Generate and write the guard script guard_script_path = os.path.join(agent_dir, "guard.py") - guard_script = generate_guard_script(agent_name, args.agent_id) - write_text_atomic( - guard_script_path, - guard_script, - 0o700, - "Elydora guard runtime", - ) - print(f" Guard script: {guard_script_path}") - config: InstallConfig = { "org_id": args.org_id, "agent_id": args.agent_id, @@ -230,8 +216,24 @@ def cmd_install(args: argparse.Namespace) -> None: } if secrets.token: config["token"] = secrets.token + plugin.preflight_install(config) + + # Create per-agent directory under ~/.elydora/{agent_id}/ + ensure_private_directory(_runtime_root()) + ensure_private_directory(agent_dir) + + # Generate and write the guard script for adapters that use the shared runtime. + if not plugin.manages_guard_runtime: + guard_script = generate_guard_script(agent_name, args.agent_id) + write_text_atomic( + guard_script_path, + guard_script, + 0o700, + "Elydora guard runtime", + ) plugin.install(config) + print(f" Guard script: {guard_script_path}") def cmd_uninstall(args: argparse.Namespace) -> None: diff --git a/elydora/plugins/base.py b/elydora/plugins/base.py index 13e565e..5b513c9 100644 --- a/elydora/plugins/base.py +++ b/elydora/plugins/base.py @@ -32,6 +32,11 @@ class PluginStatus(TypedDict): class AgentPlugin(ABC): """Base class for all agent plugins.""" + manages_guard_runtime = False + + def preflight_install(self, config: InstallConfig) -> None: + """Validate provider sources before the CLI creates runtime files.""" + @abstractmethod def install(self, config: InstallConfig) -> None: """Install the Elydora hook for this agent.""" diff --git a/elydora/plugins/cursor.py b/elydora/plugins/cursor.py index c16a9ef..ff28169 100644 --- a/elydora/plugins/cursor.py +++ b/elydora/plugins/cursor.py @@ -1,181 +1,183 @@ -"""Cursor plugin — writes/merges postToolUse hook into ~/.cursor/hooks.json.""" +"""Cursor native user-hook integration.""" from __future__ import annotations import json import os -import sys +from typing import List, Optional -from ._file_io import write_json_atomic, write_text_atomic +from elydora._runtime_paths import resolve_agent_directory + +from ._transaction import FileChange from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script +from .cursor_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + build_handler, + remove_managed_hooks, + render_document, + runtime_contracts, + runtime_root, + same_path, +) +from .cursor_io import ( + read_document, + rendered_change, + physical_directory_exists, + physical_file_exists, + require_runtime_directory, + runtime_change, + runtime_files_exist, + validate_runtime_identity, + write_cursor_changes, +) +from .hook_template import generate_guard_script, generate_hook_script -SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".cursor", "hooks.json") -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") +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 _present(changes: List[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def _agent_paths(config: InstallConfig) -> tuple[str, str, str]: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + agent_directory = resolve_agent_directory(runtime_root(), agent_id) + guard_path = os.path.join(agent_directory, GUARD_SCRIPT) + configured_guard = config.get("guard_script_path", "") + if not same_path(configured_guard, guard_path): + raise ValueError( + "Elydora guard runtime must use the managed agent directory: " + f"{guard_path}" + ) + return agent_id, agent_directory, guard_path class CursorPlugin(AgentPlugin): - """Install/uninstall Elydora audit hook for Cursor.""" + """Install Elydora into Cursor's native global user hooks.""" - @staticmethod - def _hook_path_for(agent_id: str) -> str: - return os.path.join(ELYDORA_DIR, agent_id, "hook.py") + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + read_document() + agent_id, agent_directory, guard_path = _agent_paths(config) + if not physical_directory_exists(agent_directory): + return + validate_runtime_identity( + os.path.join(agent_directory, "config.json"), + agent_id, + ) + for file_path, label in ( + (guard_path, "Elydora guard runtime"), + (os.path.join(agent_directory, AUDIT_SCRIPT), "Elydora audit runtime"), + (os.path.join(agent_directory, "private.key"), "Elydora private key"), + ): + physical_file_exists(file_path, label) def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - agent_name = config.get("agent_name", "") + document = read_document() + agent_id, agent_directory, guard_path = _agent_paths(config) + require_runtime_directory(agent_directory) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - # Create per-agent directory - agent_dir = os.path.join(ELYDORA_DIR, agent_id) - os.makedirs(agent_dir, exist_ok=True) - - # Write config.json - config_data = { - "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_name, - } - config_path = os.path.join(agent_dir, "config.json") - write_json_atomic( - config_path, - config_data, - 0o600, - "Elydora runtime config", - ) - - # Write private key - private_key_path = os.path.join(agent_dir, "private.key") - write_text_atomic( - private_key_path, - config.get("private_key", ""), - 0o600, - "Elydora private key", - ) - - script = generate_hook_script( + runtime_config_path = os.path.join(agent_directory, "config.json") + validate_runtime_identity(runtime_config_path, agent_id) + hooks = remove_managed_hooks(document.hooks) + hooks["preToolUse"] = [ + *hooks.get("preToolUse", []), + build_handler(guard_path), + ] + hooks["postToolUse"] = [ + *hooks.get("postToolUse", []), + build_handler(audit_path), + ] + hook_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"), + success_output="{}\n", ) - hook_path = self._hook_path_for(agent_id) - write_text_atomic( - hook_path, - script, - 0o700, - "Elydora audit runtime", + guard_script = generate_guard_script( + AGENT_KEY, + agent_id, + success_output='{"permission":"allow"}\n', ) - - guard_script_path = config.get("guard_script_path", "") - python_exe = sys.executable - - settings = _load_json(SETTINGS_PATH) - hooks = settings.setdefault("hooks", {}) - - # --- preToolUse (guard — freeze enforcement, camelCase) --- - pre_tool_use = hooks.setdefault("preToolUse", []) - pre_tool_use[:] = [h for h in pre_tool_use if not _is_elydora_hook(h)] - if guard_script_path: - pre_tool_use.append({ - "command": f'"{python_exe}" {guard_script_path}', - }) - - # --- postToolUse (audit logging, camelCase) --- - post_tool_use = hooks.setdefault("postToolUse", []) - - post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)] - - post_tool_use.append({ - "command": hook_path, - }) - - _save_json(SETTINGS_PATH, settings) - print("Elydora hook installed for Cursor.") - print(f" Hook script: {hook_path}") - print(f" Settings: {SETTINGS_PATH}") + changes = _present([ + runtime_change( + guard_path, + "Elydora guard runtime", + guard_script, + 0o700, + ), + runtime_change( + runtime_config_path, + "Elydora runtime config", + _json_source(_runtime_config(config, agent_id)), + 0o600, + ), + runtime_change( + os.path.join(agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + ), + runtime_change( + audit_path, + "Elydora audit runtime", + hook_script, + 0o700, + ), + rendered_change(render_document(document, hooks)), + ]) + write_cursor_changes(changes, "Install Cursor hooks") + print(f" Cursor hooks: {document.file_path}") def uninstall(self, agent_id: str = "") -> None: - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - changed = False - - # Remove preToolUse entries - pre_tool_use = hooks.get("preToolUse", []) - pre_filtered = [h for h in pre_tool_use if not _is_elydora_hook(h, agent_id)] - if len(pre_filtered) != len(pre_tool_use): - hooks["preToolUse"] = pre_filtered - if not pre_filtered: - del hooks["preToolUse"] - changed = True - - # Remove postToolUse entries - post_tool_use = hooks.get("postToolUse", []) - post_filtered = [h for h in post_tool_use if not _is_elydora_hook(h, agent_id)] - if len(post_filtered) != len(post_tool_use): - hooks["postToolUse"] = post_filtered - if not post_filtered: - del hooks["postToolUse"] - changed = True - - if changed: - if not hooks: - del settings["hooks"] - _save_json(SETTINGS_PATH, settings) - - # Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir) - print("Elydora hook uninstalled from Cursor.") + document = read_document() + hooks = remove_managed_hooks(document.hooks, agent_id) + change = rendered_change(render_document(document, hooks)) + write_cursor_changes( + [] if change is None else [change], + "Uninstall Cursor hooks", + ) def status(self) -> PluginStatus: - # Scan ~/.elydora/*/hook.py for any installed hook - import glob as _glob - hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py") - hook_files = _glob.glob(hook_pattern) - hook_exists = len(hook_files) > 0 - - settings_configured = False - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - pre_tool_use = hooks.get("preToolUse", []) - post_tool_use = hooks.get("postToolUse", []) - pre_configured = any(_is_elydora_hook(h) for h in pre_tool_use) - post_configured = any(_is_elydora_hook(h) for h in post_tool_use) - settings_configured = pre_configured and post_configured - - installed = hook_exists and settings_configured - if installed: - details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}" - elif hook_exists: - details = "Hook script exists but not configured in hooks.json" - elif settings_configured: - details = "Configured in hooks.json but hook script missing" - else: - details = "Not installed" - - return PluginStatus(installed=installed, agent="cursor", details=details) - - -def _is_elydora_hook(hook: dict, agent_id: str = "") -> bool: - cmd = hook.get("command", "") - if "elydora" not in cmd.lower(): - return False - # If agent_id is specified, only match hooks for that specific agent - if agent_id: - return agent_id in cmd - return True - - -def _load_json(path: str) -> dict: - if not os.path.exists(path): - return {} - with open(path, "r", encoding="utf-8") as f: - return json.load(f) - - -def _save_json(path: str, data: dict) -> None: - write_json_atomic(path, data, 0o600, "Cursor hooks config") + document = read_document() + contracts = runtime_contracts(document.hooks) + if not contracts: + return PluginStatus( + installed=False, + agent=AGENT_KEY, + details="Not installed", + ) + installed = runtime_files_exist(contracts) + details = ( + f"Config: {document.file_path}" + if installed + else f"Configured at {document.file_path}; runtime files missing" + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details, + ) diff --git a/elydora/plugins/cursor_contract.py b/elydora/plugins/cursor_contract.py new file mode 100644 index 0000000..7d94400 --- /dev/null +++ b/elydora/plugins/cursor_contract.py @@ -0,0 +1,348 @@ +"""Cursor hook contract, ownership, and rendering.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +import re +import sys +from typing import Any, Dict, List, Optional, Tuple + + +AGENT_KEY = "cursor" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_SECONDS = 10 + +JsonObject = Dict[str, Any] +CursorHooks = Dict[str, List[JsonObject]] + + +@dataclass(frozen=True) +class CursorDocument: + exists: bool + file_path: str + root: JsonObject + hooks: CursorHooks + raw: Optional[str] = None + + +@dataclass(frozen=True) +class RenderedDocument: + document: CursorDocument + changed: bool + next_source: Optional[str] = None + + +@dataclass(frozen=True) +class RuntimeContract: + agent_id: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class _ParsedArgument: + value: str + next_index: int + + +class _DuplicateKey(ValueError): + pass + + +def runtime_root() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def same_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def _same_agent_id(left: str, right: str) -> bool: + return os.path.normcase(left) == os.path.normcase(right) + + +def _quote_posix(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def _quote_powershell(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def build_handler(script_path: str) -> JsonObject: + if os.name == "nt": + command = ( + f"& {_quote_powershell(sys.executable)} " + f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" + ) + else: + command = f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" + return { + "command": command, + "timeout": HOOK_TIMEOUT_SECONDS, + "failClosed": True, + } + + +def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: + if start >= len(command) or command[start] != "'": + return None + apostrophe = "'\"'\"'" + value = "" + index = start + 1 + while index < len(command): + if command.startswith(apostrophe, index): + value += "'" + index += len(apostrophe) + continue + if command[index] == "'": + return _ParsedArgument(value, index + 1) + value += command[index] + index += 1 + return None + + +def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: + executable = _read_posix_argument(command, 0) + if executable is None or command[executable.next_index:executable.next_index + 1] != " ": + return None + script = _read_posix_argument(command, executable.next_index + 1) + if script is None or script.next_index != len(command): + return None + return executable.value, script.value + + +def _read_powershell_argument( + command: str, + start: int, +) -> Optional[_ParsedArgument]: + if start >= len(command) or command[start] != "'": + return None + value = "" + index = start + 1 + while index < len(command): + if command[index] != "'": + value += command[index] + index += 1 + continue + if index + 1 < len(command) and command[index + 1] == "'": + value += "'" + index += 2 + continue + return _ParsedArgument(value, index + 1) + return None + + +def _parse_powershell_command(command: str) -> Optional[Tuple[str, str]]: + if not command.startswith("& "): + return None + executable = _read_powershell_argument(command, 2) + if executable is None or command[executable.next_index:executable.next_index + 1] != " ": + return None + script = _read_powershell_argument(command, executable.next_index + 1) + if script is None or command[script.next_index:] != "; exit $LASTEXITCODE": + return None + return executable.value, script.value + + +def _legacy_script_path(command: str) -> Optional[str]: + quoted = re.fullmatch(r'"([^"\r\n]+)" ([^\r\n]+)', command) + if quoted and same_path(quoted.group(1), sys.executable): + return quoted.group(2) + return command if os.path.isabs(command) else None + + +def _managed_script_path(handler: JsonObject) -> Optional[str]: + if ( + set(handler) == {"command", "timeout", "failClosed"} + and isinstance(handler.get("command"), str) + and type(handler.get("timeout")) is int + and handler.get("timeout") == HOOK_TIMEOUT_SECONDS + and handler.get("failClosed") is True + ): + parsed = ( + _parse_powershell_command(handler["command"]) + if os.name == "nt" + else _parse_posix_command(handler["command"]) + ) + if ( + parsed is not None + and os.path.isabs(parsed[0]) + and os.path.isabs(parsed[1]) + and same_path(parsed[0], sys.executable) + ): + return parsed[1] + if set(handler) != {"command"} or not isinstance(handler.get("command"), str): + return None + return _legacy_script_path(handler["command"]) + + +def _managed_agent_id( + handler: JsonObject, + script_name: str, +) -> Optional[str]: + script_path = _managed_script_path(handler) + if script_path is None or os.path.basename(script_path) != script_name: + return None + agent_directory = os.path.dirname(script_path) + if not same_path(os.path.dirname(agent_directory), runtime_root()): + return None + agent_id = os.path.basename(agent_directory) + return agent_id if agent_id not in {"", ".", ".."} else None + + +def _unique_object(pairs: List[Tuple[str, Any]]) -> JsonObject: + result: JsonObject = {} + for key, value in pairs: + if key in result: + raise _DuplicateKey(f'duplicate field "{key}"') + result[key] = value + return result + + +def parse_json_object(raw: str, label: str) -> JsonObject: + try: + value = json.loads(raw, object_pairs_hook=_unique_object) + except _DuplicateKey as error: + raise ValueError(f"Failed to parse {label}: {error}") from error + except json.JSONDecodeError as error: + raise ValueError(f"Failed to parse {label}: {error}") from error + if not isinstance(value, dict): + raise ValueError(f"{label} must contain a JSON object") + return value + + +def _read_hooks(value: Any, label: str) -> CursorHooks: + if value is None: + raise ValueError(f'{label} field "hooks" must be an object') + if not isinstance(value, dict): + raise ValueError(f'{label} field "hooks" must be an object') + hooks: CursorHooks = {} + for event, handlers in value.items(): + if not isinstance(handlers, list): + raise ValueError(f'{label} field "hooks.{event}" must be an array') + if not all(isinstance(handler, dict) for handler in handlers): + raise ValueError( + f'{label} field "hooks.{event}" must contain objects' + ) + hooks[event] = list(handlers) + return hooks + + +def _contains_managed_hook(hooks: CursorHooks) -> bool: + return any( + _managed_agent_id(handler, script_name) is not None + for event, script_name in ( + ("preToolUse", GUARD_SCRIPT), + ("postToolUse", AUDIT_SCRIPT), + ) + for handler in hooks.get(event, []) + ) + + +def parse_document(file_path: str, raw: str) -> CursorDocument: + label = f"Cursor user hooks at {file_path}" + root = parse_json_object(raw, label) + hooks = _read_hooks(root.get("hooks", {}), label) + version = root.get("version") + current = type(version) is int and version == 1 + legacy_owned = "version" not in root and _contains_managed_hook(hooks) + if not current and not legacy_owned: + raise ValueError(f"{label} must declare version 1") + return CursorDocument(True, file_path, root, hooks, raw) + + +def create_document(file_path: str) -> CursorDocument: + return CursorDocument(False, file_path, {}, {}) + + +def remove_managed_hooks( + hooks: CursorHooks, + agent_id: str = "", +) -> CursorHooks: + result = {event: list(handlers) for event, handlers in hooks.items()} + for event, script_name in ( + ("preToolUse", GUARD_SCRIPT), + ("postToolUse", AUDIT_SCRIPT), + ): + handlers = [] + for handler in result.get(event, []): + managed_id = _managed_agent_id(handler, script_name) + remove = managed_id is not None and ( + not agent_id or _same_agent_id(managed_id, agent_id) + ) + if not remove: + handlers.append(handler) + if handlers: + result[event] = handlers + else: + result.pop(event, None) + return result + + +def _entirely_managed(document: CursorDocument) -> bool: + if not document.exists or not set(document.root).issubset({"version", "hooks"}): + return False + if not document.hooks: + return False + handler_count = 0 + for event, handlers in document.hooks.items(): + script_name = { + "preToolUse": GUARD_SCRIPT, + "postToolUse": AUDIT_SCRIPT, + }.get(event) + if script_name is None or not handlers: + return False + handler_count += len(handlers) + if any(_managed_agent_id(handler, script_name) is None for handler in handlers): + return False + return handler_count > 0 + + +def render_document( + document: CursorDocument, + hooks: CursorHooks, +) -> RenderedDocument: + if not document.exists and not hooks: + return RenderedDocument(document, False) + if document.exists and hooks == document.hooks: + return RenderedDocument(document, False) + if not hooks and _entirely_managed(document): + return RenderedDocument(document, True) + root = {**document.root, "version": 1} + if hooks: + root["hooks"] = hooks + else: + root.pop("hooks", None) + next_source = json.dumps(root, indent=2) + "\n" + return RenderedDocument(document, next_source != document.raw, next_source) + + +def _managed_ids(handlers: List[JsonObject], script_name: str) -> Dict[str, str]: + result = {} + for handler in handlers: + agent_id = _managed_agent_id(handler, script_name) + if agent_id: + result[os.path.normcase(agent_id)] = agent_id + return result + + +def runtime_contracts(hooks: CursorHooks) -> List[RuntimeContract]: + guards = _managed_ids(hooks.get("preToolUse", []), GUARD_SCRIPT) + audits = _managed_ids(hooks.get("postToolUse", []), AUDIT_SCRIPT) + contracts = [] + for key, agent_id in guards.items(): + if key not in audits: + continue + agent_directory = os.path.join(runtime_root(), agent_id) + contracts.append(RuntimeContract( + agent_id, + os.path.join(agent_directory, GUARD_SCRIPT), + os.path.join(agent_directory, AUDIT_SCRIPT), + )) + return contracts diff --git a/elydora/plugins/cursor_io.py b/elydora/plugins/cursor_io.py new file mode 100644 index 0000000..7b94264 --- /dev/null +++ b/elydora/plugins/cursor_io.py @@ -0,0 +1,192 @@ +"""Cursor hook source and runtime I/O.""" + +from __future__ import annotations + +import os +import stat +from typing import Any, List, Optional + +from ._transaction import FileChange, source_change, write_changes +from .cursor_contract import ( + AGENT_KEY, + CursorDocument, + JsonObject, + RenderedDocument, + RuntimeContract, + create_document, + parse_document, + parse_json_object, +) + + +CONFIG_FILE = "hooks.json" + + +def config_path() -> str: + return os.path.join(os.path.expanduser("~"), ".cursor", CONFIG_FILE) + + +def _read_optional_physical(file_path: str, label: str) -> Optional[str]: + try: + metadata = os.lstat(file_path) + except FileNotFoundError: + return None + except OSError as error: + raise OSError(f"Inspect {label} at {file_path}: {error}") from error + if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"{label} path is not a physical file: {file_path}") + try: + with open(file_path, "r", encoding="utf-8", newline="") as file: + return file.read() + except OSError as error: + raise OSError(f"Read {label} at {file_path}: {error}") from error + + +def read_document() -> CursorDocument: + file_path = config_path() + raw = _read_optional_physical(file_path, "Cursor user hooks") + return create_document(file_path) if raw is None else parse_document(file_path, raw) + + +def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.file_path, + "Cursor user hooks", + rendered.document.raw, + rendered.next_source, + 0o600, + ) + + +def runtime_change( + file_path: str, + label: str, + next_source: str, + mode: int, +) -> Optional[FileChange]: + original = _read_optional_physical(file_path, label) + return source_change(file_path, label, original, next_source, mode) + + +def physical_file_exists(file_path: str, label: str) -> bool: + try: + metadata = os.lstat(file_path) + except FileNotFoundError: + return False + except OSError as error: + raise OSError(f"Inspect {label} at {file_path}: {error}") from error + if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"{label} path is not a physical file: {file_path}") + return True + + +def require_runtime(file_path: str, label: str) -> None: + if not file_path: + raise ValueError(f"{label} path is required") + if not physical_file_exists(file_path, label): + raise FileNotFoundError(f"{label} is missing: {file_path}") + + +def physical_directory_exists(directory: str) -> bool: + try: + metadata = os.lstat(directory) + except FileNotFoundError: + return False + except OSError as error: + raise OSError( + f"Inspect Elydora agent runtime directory at {directory}: {error}" + ) from error + if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError( + f"Elydora agent runtime path is not a physical directory: {directory}" + ) + return True + + +def require_runtime_directory(directory: str) -> None: + if not physical_directory_exists(directory): + raise FileNotFoundError( + f"Elydora agent runtime directory is missing: {directory}" + ) + + +def _read_runtime_config(file_path: str) -> Optional[JsonObject]: + raw = _read_optional_physical(file_path, "Elydora runtime config") + if raw is None: + return None + return parse_json_object(raw, f"Elydora runtime config at {file_path}") + + +def validate_runtime_identity( + file_path: str, + agent_id: str, +) -> None: + config = _read_runtime_config(file_path) + if config is None: + return + if config.get("agent_name") != AGENT_KEY or not _same_agent_id( + config.get("agent_id"), agent_id + ): + raise ValueError( + f"Elydora runtime config identity does not match Cursor agent {agent_id}: " + f"{file_path}" + ) + + +def _same_agent_id(value: Any, expected: str) -> bool: + return isinstance(value, str) and os.path.normcase(value) == os.path.normcase( + expected + ) + + +def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: + for contract in contracts: + agent_directory = os.path.dirname(contract.guard_path) + if not physical_directory_exists(agent_directory): + continue + config_path_value = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path_value) + if ( + config is None + or config.get("agent_name") != AGENT_KEY + or not _same_agent_id(config.get("agent_id"), contract.agent_id) + ): + continue + if physical_file_exists( + contract.guard_path, + "Elydora guard runtime", + ) and physical_file_exists( + contract.audit_path, + "Elydora audit runtime", + ) and physical_file_exists( + os.path.join(agent_directory, "private.key"), + "Elydora private key", + ): + return True + return False + + +def _ensure_physical_directory(directory: str, label: str) -> None: + try: + os.makedirs(directory, mode=0o700, exist_ok=True) + metadata = os.lstat(directory) + except OSError as error: + raise OSError(f"Prepare {label} directory at {directory}: {error}") from error + if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"{label} directory is not a physical directory: {directory}") + + +def write_cursor_changes(changes: List[FileChange], label: str) -> None: + effective = [change for change in changes if change.original != change.next_source] + if not effective: + return + directories = { + os.path.normcase(os.path.abspath(os.path.dirname(change.file_path))): + os.path.dirname(change.file_path) + for change in effective + } + for directory in directories.values(): + _ensure_physical_directory(directory, label) + write_changes(effective, label) diff --git a/elydora/plugins/hook_template.py b/elydora/plugins/hook_template.py index 96f8aaf..1e5f482 100644 --- a/elydora/plugins/hook_template.py +++ b/elydora/plugins/hook_template.py @@ -16,6 +16,7 @@ def generate_hook_script( agent_id: str, kid: str, base_url: str, + success_output: str = "", ) -> str: """Return a complete, self-contained Python hook script as a string. @@ -49,6 +50,7 @@ ORG_ID = {org_id!r} AGENT_ID = {agent_id!r} KID = {kid!r} BASE_URL = {base_url!r} +SUCCESS_OUTPUT = {success_output!r} ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") PRIVATE_KEY_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "private.key") @@ -341,10 +343,16 @@ def main(): if __name__ == "__main__": main() + if SUCCESS_OUTPUT: + sys.stdout.write(SUCCESS_OUTPUT) ''' -def generate_guard_script(agent_name: str, agent_id: str) -> str: +def generate_guard_script( + agent_name: str, + agent_id: str, + success_output: str = "", +) -> str: """Return a self-contained Python guard script for PreToolUse freeze enforcement. The script: @@ -367,6 +375,7 @@ import time AGENT_NAME = {agent_name!r} AGENT_ID = {agent_id!r} +SUCCESS_OUTPUT = {success_output!r} ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") CONFIG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "config.json") STATUS_CACHE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "status-cache.json") @@ -444,4 +453,6 @@ def main(): if __name__ == "__main__": main() + if SUCCESS_OUTPUT: + sys.stdout.write(SUCCESS_OUTPUT) ''' diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index 9537030..5d3ff94 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -33,7 +33,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "cursor": { "name": "Cursor", - "hook_event": "PostToolUse", + "hook_event": "PreToolUse/PostToolUse", "config_path": "~/.cursor/hooks.json", }, "droid": { diff --git a/tests/test_cli_security.py b/tests/test_cli_security.py index 8201319..27f661a 100644 --- a/tests/test_cli_security.py +++ b/tests/test_cli_security.py @@ -249,7 +249,13 @@ def test_cli_reads_credentials_from_files_before_installing( installed: list[dict[str, str]] = [] class Plugin: + manages_guard_runtime = False + + def preflight_install(self, config: dict[str, str]) -> None: + assert Path(config["guard_script_path"]).exists() is False + def install(self, config: dict[str, str]) -> None: + assert Path(config["guard_script_path"]).is_file() installed.append(config) monkeypatch.setattr(cli, "_get_plugin", lambda _name: Plugin()) @@ -281,6 +287,41 @@ def test_cli_reads_credentials_from_files_before_installing( assert guard_path.stat().st_mode & 0o777 == 0o700 +def test_cursor_cli_preflight_rejects_config_before_runtime_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + private_key_file = tmp_path / "private.key" + write_secret(private_key_file, PRIVATE_KEY) + home_dir = tmp_path / "home" + config_path = home_dir / ".cursor" / "hooks.json" + config_path.parent.mkdir(parents=True) + config_path.write_text("{ malformed", encoding="utf-8") + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + args = cli.build_parser().parse_args( + [ + "install", + "--agent", + "cursor", + "--org_id", + "org-1", + "--agent_id", + "agent-1", + "--private_key_file", + str(private_key_file), + "--kid", + "key-1", + ] + ) + + with pytest.raises(ValueError, match="parse Cursor user hooks"): + cli.cmd_install(args) + + assert config_path.read_text(encoding="utf-8") == "{ malformed" + assert (home_dir / ".elydora").exists() is False + + def test_generated_hook_reads_private_key_from_private_file( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, @@ -353,7 +394,11 @@ def test_legacy_plugins_persist_one_owner_only_private_key( guard_path = runtime_root / "agent-1" / "guard.py" guard_path.parent.mkdir(parents=True) guard_path.write_text("pass\n", encoding="utf-8") - monkeypatch.setattr(module, "ELYDORA_DIR", str(runtime_root)) + if module is cursor: + monkeypatch.setenv("HOME", str(case_root)) + monkeypatch.setenv("USERPROFILE", str(case_root)) + else: + monkeypatch.setattr(module, "ELYDORA_DIR", str(runtime_root)) if hasattr(module, "SETTINGS_PATH"): monkeypatch.setattr( diff --git a/tests/test_cursor_plugin.py b/tests/test_cursor_plugin.py new file mode 100644 index 0000000..a884b4d --- /dev/null +++ b/tests/test_cursor_plugin.py @@ -0,0 +1,500 @@ +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass +import json +import os +from pathlib import Path +import subprocess +import sys +import time +from typing import Any, Mapping + +import pytest + +from elydora.plugins import cursor +from elydora.plugins.cursor import CursorPlugin +from elydora.plugins.hook_template import generate_guard_script +from elydora.plugins.registry import SUPPORTED_AGENTS + + +AGENT_ID = "agent-1" + + +@dataclass +class CursorFixture: + home_dir: Path + config_path: Path + agent_dir: Path + guard_path: Path + hook_path: Path + runtime_config_path: Path + private_key_path: Path + plugin: CursorPlugin + config: dict[str, str] + + def install(self) -> None: + self.plugin.install(self.config) + + +def _write_json(path: Path, value: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + source = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n" + path.write_text(source, encoding="utf-8") + + +def _prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_config: object | None = None, + create_guard: bool = True, +) -> CursorFixture: + home_dir = tmp_path / "home with spaces and 'quote" + config_path = home_dir / ".cursor" / "hooks.json" + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / "guard.py" + hook_path = agent_dir / "hook.py" + runtime_config_path = agent_dir / "config.json" + private_key_path = agent_dir / "private.key" + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.setattr(cursor, "SETTINGS_PATH", str(config_path), raising=False) + monkeypatch.setattr( + cursor, + "ELYDORA_DIR", + str(home_dir / ".elydora"), + raising=False, + ) + agent_dir.mkdir(parents=True) + if create_guard: + guard_path.write_text( + generate_guard_script("cursor", AGENT_ID), + encoding="utf-8", + ) + if existing_config is not None: + _write_json(config_path, existing_config) + return CursorFixture( + home_dir=home_dir, + config_path=config_path, + agent_dir=agent_dir, + guard_path=guard_path, + hook_path=hook_path, + runtime_config_path=runtime_config_path, + private_key_path=private_key_path, + plugin=CursorPlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "cursor", + "private_key": "private-key", + "kid": "key-1", + "token": "token-1", + "base_url": "https://api.elydora.com", + "guard_script_path": str(guard_path), + }, + ) + + +def _managed_handler(settings: Mapping[str, Any], event: str, script: str) -> dict: + handlers = settings["hooks"][event] + return next(handler for handler in handlers if script in handler.get("command", "")) + + +def _assert_native_handler(handler: dict) -> None: + assert set(handler) == {"command", "timeout", "failClosed"} + assert handler["timeout"] == 10 + assert handler["failClosed"] is True + assert sys.executable.lower() in handler["command"].lower() + if os.name == "nt": + assert handler["command"].startswith("& '") + assert handler["command"].endswith("; exit $LASTEXITCODE") + else: + assert handler["command"].startswith("'") + + +def _run_handler( + handler: Mapping[str, Any], payload: bytes, + environment: Mapping[str, str] | None = None, +) -> subprocess.CompletedProcess[bytes]: + if os.name == "nt": + arguments = [ + "powershell.exe", "-NoProfile", "-NonInteractive", + "-Command", handler["command"], + ] + else: + arguments = ["/bin/sh", "-c", handler["command"]] + return subprocess.run( + arguments, + input=payload, + capture_output=True, + check=False, + env={**os.environ, **(environment or {})}, + ) + + +def _symlink_or_skip(target: Path, link: Path) -> None: + try: + link.symlink_to(target) + except OSError as error: + pytest.skip(f"Symbolic links unavailable: {error}") + + +def test_cursor_registry_declares_both_native_events() -> None: + assert SUPPORTED_AGENTS["cursor"] == { + "name": "Cursor", "hook_event": "PreToolUse/PostToolUse", + "config_path": "~/.cursor/hooks.json", + } +def test_install_preserves_user_hooks_migrates_legacy_and_is_idempotent( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path) + _write_json( + fixture.config_path, + { + "description": "user-owned", + "hooks": { + "sessionStart": [{"command": "user-session"}], + "preToolUse": [ + {"command": "user-pre"}, + {"command": f'"{sys.executable}" {fixture.guard_path}'}, + ], + "postToolUse": [{"command": str(fixture.hook_path)}], + }, + }, + ) + + fixture.install() + fixture.install() + + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert settings["version"] == 1 + assert settings["description"] == "user-owned" + assert settings["hooks"]["sessionStart"] == [{"command": "user-session"}] + assert settings["hooks"]["preToolUse"][0] == {"command": "user-pre"} + assert len(settings["hooks"]["preToolUse"]) == 2 + assert len(settings["hooks"]["postToolUse"]) == 1 + _assert_native_handler(_managed_handler(settings, "preToolUse", "guard.py")) + _assert_native_handler(_managed_handler(settings, "postToolUse", "hook.py")) +def test_handlers_block_frozen_agent_and_forward_official_input_byte_for_byte( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path) + fixture.install() + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + guard = _managed_handler(settings, "preToolUse", "guard.py") + audit = _managed_handler(settings, "postToolUse", "hook.py") + _write_json( + fixture.agent_dir / "status-cache.json", + {"status": "active", "cached_at": time.time()}, + ) + assert json.loads(_run_handler(guard, b"{}\n").stdout) == {"permission": "allow"} + _write_json( + fixture.agent_dir / "status-cache.json", + {"status": "frozen", "cached_at": time.time()}, + ) + pre_payload = ( + json.dumps( + { + "conversation_id": "conversation-1", + "generation_id": "generation-1", + "hook_event_name": "preToolUse", + "tool_name": "Shell", + "tool_input": {"command": "Get-ChildItem"}, + "tool_use_id": "call-1", + "cwd": str(tmp_path), + }, + separators=(",", ":"), + ) + + "\n" + ).encode() + guard_result = _run_handler(guard, pre_payload) + assert guard_result.returncode == 2 + assert b"frozen by Elydora" in guard_result.stderr + capture_path = tmp_path / "captured-event.json" + fixture.hook_path.write_text( + "import os, pathlib, sys\n" + "pathlib.Path(os.environ['ELYDORA_CAPTURE']).write_bytes(sys.stdin.buffer.read())\n" + "print('{}')\n", + encoding="utf-8", + ) + post_payload = ( + json.dumps( + { + "conversation_id": "conversation-1", + "generation_id": "generation-1", + "hook_event_name": "postToolUse", + "tool_name": "Shell", + "tool_input": {"command": "Get-ChildItem"}, + "tool_output": '{"exitCode":0,"stdout":"ok"}', + "tool_use_id": "call-1", + "cwd": str(tmp_path), + "duration": 42, + }, + separators=(",", ":"), + ) + + "\n" + ).encode() + audit_result = _run_handler(audit, post_payload, {"ELYDORA_CAPTURE": str(capture_path)}) + assert audit_result.returncode == 0 + assert json.loads(audit_result.stdout) == {} + assert capture_path.read_bytes() == post_payload + + +def test_status_requires_exact_pair_identity_and_physical_runtimes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.plugin.status()["installed"] is True + + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + _managed_handler(settings, "preToolUse", "guard.py")["failClosed"] = False + _write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + + fixture.install() + hook_source = fixture.hook_path.read_text(encoding="utf-8") + fixture.hook_path.unlink() + assert fixture.plugin.status()["installed"] is False + fixture.hook_path.write_text(hook_source, encoding="utf-8") + + _write_json( + fixture.runtime_config_path, + {"agent_id": "another-agent", "agent_name": "cursor"}, + ) + assert fixture.plugin.status()["installed"] is False + fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") + with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.plugin.status() + + +def test_uninstall_uses_exact_ownership_and_preserves_user_entries( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture( + monkeypatch, + tmp_path, + existing_config={ + "version": 1, + "hooks": {"sessionStart": [{"command": "keep"}]}, + }, + ) + fixture.install() + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + other_guard = deepcopy(_managed_handler(settings, "preToolUse", "guard.py")) + other_audit = deepcopy(_managed_handler(settings, "postToolUse", "hook.py")) + other_guard["command"] = other_guard["command"].replace(AGENT_ID, "agent-10") + other_audit["command"] = other_audit["command"].replace(AGENT_ID, "agent-10") + settings["hooks"]["preToolUse"].extend( + [ + other_guard, + { + "command": "echo elydora agent-1 guard.py", + "timeout": 10, + "failClosed": True, + }, + ] + ) + settings["hooks"]["postToolUse"].append(other_audit) + _write_json(fixture.config_path, settings) + + fixture.plugin.uninstall(AGENT_ID) + + remaining = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert remaining["hooks"]["sessionStart"] == [{"command": "keep"}] + assert len(remaining["hooks"]["preToolUse"]) == 2 + assert len(remaining["hooks"]["postToolUse"]) == 1 + assert "agent-10" in remaining["hooks"]["preToolUse"][0]["command"] + + +@pytest.mark.parametrize( + "existing_config", + [ + "{ malformed", + "[]\n", + '{"hooks":{}}\n', + '{"version":2,"hooks":{}}\n', + '{"version":1,"hooks":null}\n', + '{"version":1,"hooks":{"preToolUse":null}}\n', + '{"version":1,"hooks":{"preToolUse":[null]}}\n', + '{"version":1,"version":1,"hooks":{}}\n', + '{"version":1,"hooks":{},}\n', + ], +) +def test_invalid_config_is_preserved_before_runtime_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + existing_config: str, +) -> None: + fixture = _prepare_fixture( + monkeypatch, + tmp_path, + existing_config=existing_config, + ) + original_guard = fixture.guard_path.read_text(encoding="utf-8") + + with pytest.raises(ValueError): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == existing_config + assert fixture.guard_path.read_text(encoding="utf-8") == original_guard + assert fixture.hook_path.exists() is False + assert fixture.runtime_config_path.exists() is False + assert fixture.private_key_path.exists() is False + + +def test_install_creates_missing_guard_and_rejects_unmanaged_path( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path, create_guard=False) + fixture.install() + assert fixture.guard_path.is_file() + + fixture = _prepare_fixture(monkeypatch, tmp_path / "unmanaged", create_guard=False) + unmanaged = fixture.home_dir / "unmanaged-guard.py" + unmanaged.write_text("pass\n", encoding="utf-8") + fixture.config["guard_script_path"] = str(unmanaged) + with pytest.raises(ValueError, match="managed agent directory"): + fixture.install() + assert fixture.config_path.exists() is False + + +@pytest.mark.parametrize("runtime_name", ["guard.py", "hook.py"]) +def test_symbolic_link_runtime_is_rejected_before_config_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + runtime_name: str, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path) + runtime_path = fixture.agent_dir / runtime_name + target = fixture.home_dir / f"{runtime_name}.target" + target.write_text("pass\n", encoding="utf-8") + if runtime_path.exists(): + runtime_path.unlink() + _symlink_or_skip(target, runtime_path) + + with pytest.raises(OSError, match="physical file"): + fixture.install() + + assert fixture.config_path.exists() is False + assert target.read_text(encoding="utf-8") == "pass\n" + + +def test_symbolic_link_config_is_rejected_and_target_is_preserved( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path) + target = fixture.home_dir / "cursor-hooks.target.json" + original = '{"version":1,"hooks":{}}\n' + target.write_text(original, encoding="utf-8") + fixture.config_path.parent.mkdir(parents=True) + _symlink_or_skip(target, fixture.config_path) + + with pytest.raises(OSError, match="physical file"): + fixture.install() + + assert target.read_text(encoding="utf-8") == original + assert fixture.config_path.is_symlink() + + +@pytest.mark.parametrize("runtime_name", ["config.json", "private.key", "guard.py", "hook.py"]) +def test_status_rejects_symbolic_link_runtime_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + runtime_name: str, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path) + fixture.install() + runtime_path = fixture.agent_dir / runtime_name + contents = runtime_path.read_bytes() + target = fixture.home_dir / f"status-{runtime_name}.target" + target.write_bytes(contents) + runtime_path.unlink() + _symlink_or_skip(target, runtime_path) + + with pytest.raises(OSError, match="physical file"): + fixture.plugin.status() + + +def test_uninstall_removes_entirely_managed_config_and_keeps_absence( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture(monkeypatch, tmp_path) + fixture.install() + fixture.plugin.uninstall(AGENT_ID) + assert fixture.config_path.exists() is False + fixture.plugin.uninstall(AGENT_ID) + assert fixture.config_path.exists() is False + user_source = '{"version":1,"hooks":{}}\n' + fixture.config_path.write_text(user_source, encoding="utf-8") + fixture.plugin.uninstall(AGENT_ID) + assert fixture.config_path.read_text(encoding="utf-8") == user_source + + +def test_transaction_rolls_back_runtime_files_when_config_commit_fails( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = _prepare_fixture( + monkeypatch, + tmp_path, + existing_config={"version": 1, "hooks": {"sessionStart": []}}, + create_guard=False, + ) + original = fixture.config_path.read_text(encoding="utf-8") + real_replace = os.replace + failed = False + def fail_config_once(source: str, destination: str) -> None: + nonlocal failed + if os.path.abspath(destination) == str(fixture.config_path) and not failed: + failed = True + raise OSError("injected config commit failure") + real_replace(source, destination) + monkeypatch.setattr(os, "replace", fail_config_once) + with pytest.raises(OSError, match="injected config commit failure"): + fixture.install() + assert fixture.config_path.read_text(encoding="utf-8") == original + assert fixture.guard_path.exists() is False + assert fixture.hook_path.exists() is False + assert fixture.runtime_config_path.exists() is False + assert fixture.private_key_path.exists() is False + assert not list(fixture.home_dir.rglob("*.tmp")) + assert not list(fixture.home_dir.rglob("*.rollback")) + + +def test_atomic_config_write_detects_concurrent_source_change( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + from elydora.plugins._transaction import write_changes + from elydora.plugins.cursor_contract import render_document + from elydora.plugins.cursor_io import read_document, rendered_change + + fixture = _prepare_fixture( + monkeypatch, + tmp_path, + existing_config={ + "version": 1, + "hooks": {"sessionStart": [{"command": "original"}]}, + }, + ) + document = read_document() + rendered = render_document(document, {}) + change = rendered_change(rendered) + assert change is not None + concurrent = '{"version":1,"hooks":{"sessionStart":[]}}\n' + fixture.config_path.write_text(concurrent, encoding="utf-8") + + with pytest.raises(OSError, match="changed during installation"): + write_changes([change], "Update Cursor hooks") + + assert fixture.config_path.read_text(encoding="utf-8") == concurrent + assert not list(fixture.config_path.parent.glob("*.tmp"))