diff --git a/AGENTS.md b/AGENTS.md index 6fa13f7..8cf1261 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -26,6 +26,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Model stable, legacy, and early-access hook generations as explicit contracts. Keep their activation requirements visible in CLI output and README guidance. - Resolve stable Kimi hooks through `$KIMI_CODE_HOME/config.toml` with `~/.kimi-code/config.toml` as the default, and treat an empty override as the default. Activate `~/.kimi/config.toml` only when the legacy home exists. Register exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` rules with ten-second commands; stable Kimi supports its current sixteen-event schema and legacy kimi-cli supports its thirteen-event schema. Preserve TOML comments and array style, then parse the complete rendered document before staging it. Preserve the native snake_case payload, propagate exit code `2`, and keep audit delivery fail-open with owner-only error logs. Use encoded PowerShell commands on Windows and exact two-argument shell commands on Unix. Commit every detected config, generated runtime, runtime config, and private key through one rollback-capable transaction. - Resolve Grok Build user hooks through `$GROK_HOME/hooks/*.json` with `~/.grok/hooks/*.json` as the default, and treat an empty override as the default. Keep Claude Code, Cursor, project `.grok/hooks`, plugin, and `hooks-paths` sources read-only. Register exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups with ten-second command handlers, preserve the complete native camelCase payload, and emit Grok's documented deny JSON plus exit code `2` for frozen or revoked agents. Use encoded PowerShell commands on Windows and exact two-argument shell commands on Unix. Commit Grok's hook file, generated runtimes, runtime config, and private key through one rollback-capable transaction. +- Resolve Claude Code user hooks through `$CLAUDE_CONFIG_DIR/settings.json` with `~/.claude/settings.json` when the variable is absent. Match Claude Code's path resolution exactly: relative and empty values resolve from the current working directory, and literal tildes remain path segments. Keep project, local, managed, plugin, skill, and agent hook sources read-only. Register exact matchless `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups with ten-second exec-form command handlers. Preserve the complete native snake_case payload and propagate freeze and revocation through exit code `2`. Commit Claude Code's user settings, generated runtimes, runtime config, and private key through one rollback-capable transaction. Reject installation while the user source sets `disableAllHooks`, then require `/hooks` and `claude doctor` verification for higher-scope policy effects. - Write Auggie hooks only to `~/.augment/settings.json`; keep system and workspace settings read-only. Generate `.cmd` wrappers on Windows and `.sh` wrappers on Unix because Auggie dispatches supported script paths, and express hook timeouts in milliseconds. - Validate Auggie matcher syntax during installation with Node.js `new RegExp`; keep status and uninstall independent from the JavaScript validator so recovery remains available offline. - Write Cline hooks only to `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default; keep Documents and workspace hook roots read-only. Preserve official input byte-for-byte, resolve Python runtimes from their absolute shebangs, and translate guard exit code `2` into Cline's JSON stdout cancellation control. diff --git a/README.md b/README.md index e009e2f..b60f5b5 100644 --- a/README.md +++ b/README.md @@ -78,6 +78,8 @@ elydora install \ Credential options may be omitted in an interactive terminal; the CLI then reads the private key and optional API token through hidden prompts. Credential files must contain one UTF-8 line of at most 64 KiB. Unix credential files require owner-only permissions such as `chmod 600`. The installed signing key is stored once at `~/.elydora//private.key` with mode `0600`, and generated audit hooks read that file at runtime. +Claude Code installation writes exact matchless `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups to `$CLAUDE_CONFIG_DIR/settings.json`, with `~/.claude/settings.json` selected when the variable is absent. Relative and empty overrides resolve from the current working directory, matching Claude Code's native path behavior. Each handler uses Claude Code's cross-platform exec form with the current Python executable, a single script argument, and a ten-second timeout. Managed runtimes preserve the complete native snake_case event payload and enforce frozen or revoked state through exit code `2`. Project, local, managed, plugin, skill, and agent hook sources remain unchanged and continue loading additively. User settings, generated runtimes, runtime config, and private key commit as one rollback-capable transaction. `disableAllHooks` blocks installation; managed hook policy and safe mode can suppress user hooks. Run `/hooks` and `claude doctor` after installation. + Codex installation writes exact `PreToolUse` and `PostToolUse` command groups to `$CODEX_HOME/hooks.json` (`~/.codex/hooks.json` by default) and preserves the complete native event payload. A configured `CODEX_HOME` follows Codex's existing-directory canonicalization rule. User TOML, project, plugin, and managed sources remain unchanged and continue loading additively. The hook file, generated runtimes, runtime config, and private key commit as one rollback-capable update. Run `/hooks` after installation and approve both Elydora definition hashes. Cursor installation writes native global `preToolUse`, `postToolUse`, and `postToolUseFailure` handlers to `~/.cursor/hooks.json`. Elydora forwards Cursor's native success and failure payloads, preserves unrelated user hooks, and leaves project and enterprise sources unchanged. The generated guard and audit handlers use a 10-second fail-closed boundary, exact freeze exit-code propagation, protected runtime credentials, and one rollback-capable transaction for all runtime and hook configuration files. diff --git a/elydora/plugins/claudecode.py b/elydora/plugins/claudecode.py index 6ec104c..66e60bd 100644 --- a/elydora/plugins/claudecode.py +++ b/elydora/plugins/claudecode.py @@ -1,210 +1,105 @@ -"""Claude Code plugin — merges PostToolUse hook into ~/.claude/settings.json.""" +"""Claude Code native user-hook integration.""" from __future__ import annotations -import json -import os -import sys - -from ._file_io import write_json_atomic, write_text_atomic from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script +from .claudecode_contract import ( + AGENT_KEY, + AUDIT_STATUS, + GUARD_STATUS, + ClaudeHooks, + build_claude_group, + claude_runtime_contracts, + remove_managed_claude_hooks, + render_claude_document, +) +from .claudecode_installation import ( + commit_claude_installation, + commit_claude_uninstall, + preflight_claude_installation, + prepare_claude_installation, + prepare_claude_uninstall, +) +from .claudecode_io import ( + claude_runtime_files_exist, + read_claude_document, +) -SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".claude", "settings.json") -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") +def _require_enabled_hooks(disabled: bool, file_path: str) -> None: + if disabled: + raise ValueError( + f"Claude Code hooks are disabled by disableAllHooks: {file_path}" + ) + + +def _installed_hooks( + hooks: ClaudeHooks, guard_path: str, audit_path: str +) -> ClaudeHooks: + cleaned = remove_managed_claude_hooks(hooks) + return { + **cleaned, + "PreToolUse": [ + *cleaned.get("PreToolUse", []), + build_claude_group(guard_path, GUARD_STATUS), + ], + "PostToolUse": [ + *cleaned.get("PostToolUse", []), + build_claude_group(audit_path, AUDIT_STATUS), + ], + "PostToolUseFailure": [ + *cleaned.get("PostToolUseFailure", []), + build_claude_group(audit_path, AUDIT_STATUS), + ], + } class ClaudeCodePlugin(AgentPlugin): - """Install/uninstall Elydora audit hook for Claude Code.""" + """Install Elydora into Claude Code's native global user settings.""" - @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: + document = read_claude_document() + _require_enabled_hooks(document.hooks_disabled, document.file_path) + preflight_claude_installation(config, document) def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - agent_name = config.get("agent_name", "") - - # 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", + document = read_claude_document() + _require_enabled_hooks(document.hooks_disabled, document.file_path) + paths = preflight_claude_installation(config, document) + rendered = render_claude_document( + document, + _installed_hooks(document.hooks, paths.guard_path, paths.audit_path), ) - - # 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", + changes = prepare_claude_installation(config, paths, rendered) + commit_claude_installation(changes) + print( + "Claude Code: global PreToolUse, PostToolUse, " + "and PostToolUseFailure hooks installed." ) - - # Write the hook script - 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"), - ) - hook_path = self._hook_path_for(agent_id) - write_text_atomic( - hook_path, - script, - 0o700, - "Elydora audit runtime", - ) - - guard_script_path = config.get("guard_script_path", "") - python_exe = sys.executable - - # Merge into Claude Code settings - settings = _load_json(SETTINGS_PATH) - hooks = settings.setdefault("hooks", {}) - - # --- PreToolUse (guard — freeze enforcement) --- - 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({ - "hooks": [ - { - "type": "command", - "command": f'"{python_exe}" {guard_script_path}', - } - ], - }) - - # --- PostToolUse (audit logging) --- - post_tool_use = hooks.setdefault("PostToolUse", []) - - # Remove any existing Elydora hook entry - post_tool_use[:] = [h for h in post_tool_use if not _is_elydora_hook(h)] - - post_tool_use.append({ - "hooks": [ - { - "type": "command", - "command": hook_path, - } - ], - }) - - _save_json(SETTINGS_PATH, settings) - print("Elydora hook installed for Claude Code.") - print(f" Hook script: {hook_path}") - print(f" Settings: {SETTINGS_PATH}") + print(" Claude Code verification: run /hooks and claude doctor.") def uninstall(self, agent_id: str = "") -> None: - # Remove hook entries from settings - 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 Claude Code.") + document = read_claude_document() + rendered = render_claude_document( + document, + remove_managed_claude_hooks(document.hooks, agent_id), + ) + commit_claude_uninstall(prepare_claude_uninstall(rendered)) 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 settings" - elif settings_configured: - details = "Configured in settings but hook script missing" - else: - details = "Not installed" - - return PluginStatus(installed=installed, agent="claudecode", details=details) - - -def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool: - # Collect all command strings from the entry - commands: list[str] = [] - inner_hooks = entry.get("hooks") - if isinstance(inner_hooks, list): - for hook in inner_hooks: - if isinstance(hook, dict) and isinstance(hook.get("command"), str): - commands.append(hook["command"]) - else: - command = entry.get("command") - if isinstance(command, str): - commands.append(command) - - for cmd in commands: - cmd_lower = cmd.lower() - if "elydora" not in cmd_lower: - continue - # If agent_id is specified, only match hooks for that specific agent - if agent_id and agent_id in cmd: - return True - if not agent_id: - return True - return False - - -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, "Claude Code settings") + document = read_claude_document() + contracts = claude_runtime_contracts(document.hooks) + configured = not document.hooks_disabled and bool(contracts) + installed = configured and claude_runtime_files_exist(contracts) + details = ( + f"Config: {document.file_path}" + if installed + else f"Configured at {document.file_path}; managed contract incomplete" + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details if configured else "Not installed", + ) diff --git a/elydora/plugins/claudecode_contract.py b/elydora/plugins/claudecode_contract.py new file mode 100644 index 0000000..19bf2ca --- /dev/null +++ b/elydora/plugins/claudecode_contract.py @@ -0,0 +1,486 @@ +"""Claude Code hook schema, ownership, and rendering.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import math +import os +import re +import sys +from typing import Any, Dict, List, Optional, Tuple +import urllib.parse + +from ._strict_json import JsonObject, parse_json_object + + +AGENT_KEY = "claudecode" +CONFIG_FILE = "settings.json" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_SECONDS = 10 +GUARD_STATUS = "Checking Elydora agent state" +AUDIT_STATUS = "Recording Elydora tool use" + +HOOK_EVENTS = frozenset({ + "SessionStart", "Setup", "InstructionsLoaded", "UserPromptSubmit", + "UserPromptExpansion", "MessageDisplay", "PreToolUse", + "PermissionRequest", "PostToolUse", "PostToolUseFailure", + "PostToolBatch", "PermissionDenied", "Notification", "SubagentStart", + "SubagentStop", "TaskCreated", "TaskCompleted", "Stop", "StopFailure", + "TeammateIdle", "ConfigChange", "CwdChanged", "FileChanged", + "WorktreeCreate", "WorktreeRemove", "PreCompact", "PostCompact", + "SessionEnd", "Elicitation", "ElicitationResult", +}) +COMMON_HANDLER_KEYS = frozenset({"if", "once", "statusMessage", "timeout"}) +HANDLER_KEYS = { + "command": COMMON_HANDLER_KEYS | { + "args", "async", "asyncRewake", "command", "shell", "type", + }, + "prompt": COMMON_HANDLER_KEYS | { + "continueOnBlock", "model", "prompt", "type", + }, + "agent": COMMON_HANDLER_KEYS | {"model", "prompt", "type"}, + "http": COMMON_HANDLER_KEYS | { + "allowedEnvVars", "headers", "type", "url", + }, + "mcp_tool": COMMON_HANDLER_KEYS | {"input", "server", "tool", "type"}, +} +_UNSET = object() + +ClaudeHooks = Dict[str, List[JsonObject]] + + +@dataclass(frozen=True) +class ClaudeDocument: + exists: bool + file_path: str + root: JsonObject + hooks: ClaudeHooks + hooks_disabled: bool + raw: Optional[str] = None + + +@dataclass(frozen=True) +class RenderedClaudeDocument: + document: ClaudeDocument + changed: bool + next_source: Optional[str] = None + + +@dataclass(frozen=True) +class ClaudeRuntimeContract: + agent_id: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class _RuntimeReference: + agent_id: str + script_path: str + + +def runtime_root() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def same_claude_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def same_claude_agent_id(left: str, right: str) -> bool: + return os.path.normcase(left) == os.path.normcase(right) + + +def _require_known_keys( + value: JsonObject, allowed: frozenset[str], label: str +) -> None: + extra = next((key for key in value if key not in allowed), None) + if extra is not None: + raise ValueError(f'{label} contains unsupported field "{extra}"') + + +def _require_non_empty_string(value: Any, field: str, label: str) -> str: + if not isinstance(value, str) or not value: + raise ValueError(f'{label} field "{field}" must be a non-empty string') + return value + + +def _optional_string(value: JsonObject, field: str, label: str) -> None: + if field in value and not isinstance(value[field], str): + raise ValueError(f'{label} field "{field}" must be a string') + + +def _optional_boolean(value: JsonObject, field: str, label: str) -> None: + if field in value and type(value[field]) is not bool: + raise ValueError(f'{label} field "{field}" must be a boolean') + + +def _validate_common_fields(value: JsonObject, label: str) -> None: + _optional_string(value, "if", label) + _optional_string(value, "statusMessage", label) + _optional_boolean(value, "once", label) + timeout = value.get("timeout") + if "timeout" in value and ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not math.isfinite(float(timeout)) + or float(timeout) <= 0 + ): + raise ValueError(f"{label} timeout must be a positive finite number") + + +def _string_array(value: Any, field: str, label: str) -> List[str]: + if not isinstance(value, list) or any( + not isinstance(item, str) for item in value + ): + raise ValueError(f'{label} field "{field}" must be an array of strings') + return value + + +def _validate_handler( + value: Any, event: str, group_index: int, handler_index: int +) -> JsonObject: + label = ( + f"Claude Code settings handler hooks.{event}[{group_index}]" + f".hooks[{handler_index}]" + ) + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + handler = dict(value) + handler_type = handler.get("type") + if not isinstance(handler_type, str) or handler_type not in HANDLER_KEYS: + raise ValueError(f'{label} has unsupported type "{handler_type}"') + _require_known_keys(handler, HANDLER_KEYS[handler_type], label) + _validate_common_fields(handler, label) + if handler_type == "command": + _require_non_empty_string(handler.get("command"), "command", label) + if "args" in handler: + _string_array(handler["args"], "args", label) + _optional_boolean(handler, "async", label) + _optional_boolean(handler, "asyncRewake", label) + if "shell" in handler and handler["shell"] not in ("bash", "powershell"): + raise ValueError( + f'{label} field "shell" must be "bash" or "powershell"' + ) + elif handler_type in ("prompt", "agent"): + _require_non_empty_string(handler.get("prompt"), "prompt", label) + _optional_string(handler, "model", label) + if handler_type == "prompt": + _optional_boolean(handler, "continueOnBlock", label) + elif handler_type == "http": + raw_url = _require_non_empty_string(handler.get("url"), "url", label) + try: + parsed = urllib.parse.urlsplit(raw_url) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError(f'{label} field "url" must be a valid URL') from error + if parsed.scheme not in ("http", "https") or hostname is None: + raise ValueError(f'{label} field "url" must use HTTP or HTTPS') + headers = handler.get("headers") + if "headers" in handler and ( + not isinstance(headers, dict) + or any(not isinstance(item, str) for item in headers.values()) + ): + raise ValueError(f'{label} field "headers" must map names to strings') + if "allowedEnvVars" in handler: + allowed = _string_array( + handler["allowedEnvVars"], "allowedEnvVars", label + ) + if any(not item for item in allowed): + raise ValueError( + f'{label} field "allowedEnvVars" must contain non-empty strings' + ) + else: + _require_non_empty_string(handler.get("server"), "server", label) + _require_non_empty_string(handler.get("tool"), "tool", label) + if "input" in handler and not isinstance(handler["input"], dict): + raise ValueError(f'{label} field "input" must be an object') + return handler + + +def _validate_group(value: Any, event: str, index: int) -> JsonObject: + label = f"Claude Code settings matcher group hooks.{event}[{index}]" + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + group = dict(value) + _require_known_keys(group, frozenset({"hooks", "matcher"}), label) + if "matcher" in group and not isinstance(group["matcher"], str): + raise ValueError(f"{label} matcher must be a string") + handlers = group.get("hooks") + if not isinstance(handlers, list): + raise ValueError(f"{label} must contain a hooks array") + group["hooks"] = [ + _validate_handler(handler, event, index, handler_index) + for handler_index, handler in enumerate(handlers) + ] + return group + + +def _read_hooks(value: Any) -> ClaudeHooks: + if value is _UNSET: + return {} + if not isinstance(value, dict): + raise ValueError('Claude Code settings field "hooks" must be an object') + hooks: ClaudeHooks = {} + for event, groups in value.items(): + if event not in HOOK_EVENTS: + raise ValueError( + f'Claude Code settings contains unsupported hook event "{event}"' + ) + if not isinstance(groups, list): + raise ValueError( + f'Claude Code settings field "hooks.{event}" must be an array' + ) + hooks[event] = [ + _validate_group(group, event, index) + for index, group in enumerate(groups) + ] + return hooks + + +def parse_claude_document(file_path: str, raw: str) -> ClaudeDocument: + root = parse_json_object(raw, f"Claude Code user settings at {file_path}") + disabled = root.get("disableAllHooks") + if "disableAllHooks" in root and type(disabled) is not bool: + raise ValueError( + 'Claude Code settings field "disableAllHooks" must be a boolean' + ) + return ClaudeDocument( + True, + file_path, + root, + _read_hooks(root.get("hooks", _UNSET)), + disabled is True, + raw, + ) + + +def create_claude_document(file_path: str) -> ClaudeDocument: + return ClaudeDocument(False, file_path, {}, {}, False) + + +def build_claude_group(script_path: str, status_message: str) -> JsonObject: + return { + "hooks": [{ + "type": "command", + "command": sys.executable, + "args": [script_path], + "timeout": HOOK_TIMEOUT_SECONDS, + "statusMessage": status_message, + }] + } + + +def _exact_managed_group(group: JsonObject) -> bool: + return set(group) == {"hooks"} + + +def _runtime_reference( + script_path: str, script_name: str +) -> Optional[_RuntimeReference]: + if not os.path.isabs(script_path) or os.path.basename(script_path) != script_name: + return None + agent_directory = os.path.dirname(script_path) + if not same_claude_path(os.path.dirname(agent_directory), runtime_root()): + return None + agent_id = os.path.basename(agent_directory) + if agent_id in ("", ".", ".."): + return None + return _RuntimeReference(agent_id, script_path) + + +def _legacy_reference( + command: Any, script_name: str +) -> Optional[_RuntimeReference]: + if not isinstance(command, str): + return None + if script_name == AUDIT_SCRIPT: + return _runtime_reference(command, script_name) + match = re.fullmatch(r'"([^"\r\n]+)" ([^\r\n]+)', command) + if ( + match is None + or not os.path.isabs(match.group(1)) + or not same_claude_path(match.group(1), sys.executable) + ): + return None + return _runtime_reference(match.group(2), script_name) + + +def _managed_reference( + handler: JsonObject, + script_name: str, + status_message: str, + include_legacy: bool = False, +) -> Optional[_RuntimeReference]: + if ( + set(handler) == {"args", "command", "statusMessage", "timeout", "type"} + and handler.get("type") == "command" + and type(handler.get("timeout")) is int + and handler.get("timeout") == HOOK_TIMEOUT_SECONDS + and handler.get("statusMessage") == status_message + and isinstance(handler.get("command"), str) + and os.path.isabs(str(handler["command"])) + and same_claude_path(str(handler["command"]), sys.executable) + and isinstance(handler.get("args"), list) + and len(handler["args"]) == 1 + and isinstance(handler["args"][0], str) + ): + return _runtime_reference(handler["args"][0], script_name) + if ( + include_legacy + and set(handler) == {"command", "type"} + and handler.get("type") == "command" + ): + return _legacy_reference(handler.get("command"), script_name) + return None + + +def _remove_from_groups( + groups: List[JsonObject], + script_name: str, + status_message: str, + agent_id: str, +) -> Tuple[List[JsonObject], bool]: + result: List[JsonObject] = [] + removed = False + for group in groups: + kept = [] + for handler in group["hooks"]: + reference = _managed_reference( + handler, script_name, status_message, True + ) + owned = reference is not None and ( + not agent_id + or same_claude_agent_id(reference.agent_id, agent_id) + ) + if owned: + removed = True + else: + kept.append(handler) + if kept: + result.append({**group, "hooks": kept}) + elif not _exact_managed_group(group): + result.append({**group, "hooks": []}) + return result, removed + + +def remove_managed_claude_hooks( + hooks: ClaudeHooks, agent_id: str = "" +) -> ClaudeHooks: + result = {event: list(groups) for event, groups in hooks.items()} + for event, script_name, status_message in ( + ("PreToolUse", GUARD_SCRIPT, GUARD_STATUS), + ("PostToolUse", AUDIT_SCRIPT, AUDIT_STATUS), + ("PostToolUseFailure", AUDIT_SCRIPT, AUDIT_STATUS), + ): + groups, removed = _remove_from_groups( + result.get(event, []), script_name, status_message, agent_id + ) + if not removed: + continue + if groups: + result[event] = groups + else: + result.pop(event, None) + return result + + +def _managed_event(event: str) -> Optional[Tuple[str, str]]: + if event == "PreToolUse": + return GUARD_SCRIPT, GUARD_STATUS + if event in ("PostToolUse", "PostToolUseFailure"): + return AUDIT_SCRIPT, AUDIT_STATUS + return None + + +def _entirely_managed(document: ClaudeDocument) -> bool: + if not document.exists or set(document.root) != {"hooks"} or not document.hooks: + return False + handlers = 0 + for event, groups in document.hooks.items(): + contract = _managed_event(event) + if contract is None or not groups: + return False + for group in groups: + if ( + not _exact_managed_group(group) + or not group["hooks"] + or any( + _managed_reference(handler, contract[0], contract[1], True) + is None + for handler in group["hooks"] + ) + ): + return False + handlers += len(group["hooks"]) + return handlers > 0 + + +def render_claude_document( + document: ClaudeDocument, hooks: ClaudeHooks +) -> RenderedClaudeDocument: + if hooks == document.hooks: + return RenderedClaudeDocument(document, False) + if not document.exists and not hooks: + return RenderedClaudeDocument(document, False) + if not hooks and _entirely_managed(document): + return RenderedClaudeDocument(document, True) + root = dict(document.root) + if hooks: + root["hooks"] = hooks + else: + root.pop("hooks", None) + next_source = json.dumps(root, indent=2, ensure_ascii=False) + "\n" + parse_claude_document(document.file_path, next_source) + return RenderedClaudeDocument( + document, next_source != document.raw, next_source + ) + + +def _references_for_event( + groups: List[JsonObject], script_name: str, status_message: str +) -> Dict[str, List[_RuntimeReference]]: + result: Dict[str, List[_RuntimeReference]] = {} + for group in groups: + if not _exact_managed_group(group): + continue + for handler in group["hooks"]: + reference = _managed_reference(handler, script_name, status_message) + if reference is not None: + result.setdefault( + os.path.normcase(reference.agent_id), [] + ).append(reference) + return result + + +def claude_runtime_contracts( + hooks: ClaudeHooks, +) -> List[ClaudeRuntimeContract]: + guards = _references_for_event( + hooks.get("PreToolUse", []), GUARD_SCRIPT, GUARD_STATUS + ) + successes = _references_for_event( + hooks.get("PostToolUse", []), AUDIT_SCRIPT, AUDIT_STATUS + ) + failures = _references_for_event( + hooks.get("PostToolUseFailure", []), AUDIT_SCRIPT, AUDIT_STATUS + ) + contracts = [] + for key, guard in guards.items(): + success = successes.get(key, []) + failure = failures.get(key, []) + if len(guard) != 1 or len(success) != 1 or len(failure) != 1: + continue + if not same_claude_path(success[0].script_path, failure[0].script_path): + continue + contracts.append(ClaudeRuntimeContract( + guard[0].agent_id, + guard[0].script_path, + success[0].script_path, + )) + return contracts diff --git a/elydora/plugins/claudecode_installation.py b/elydora/plugins/claudecode_installation.py new file mode 100644 index 0000000..1f1fdd4 --- /dev/null +++ b/elydora/plugins/claudecode_installation.py @@ -0,0 +1,167 @@ +"""Transactional Claude Code hook and runtime installation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +from typing import List, Optional + +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, write_changes +from .base import InstallConfig +from .claudecode_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + ClaudeDocument, + RenderedClaudeDocument, + runtime_root, + same_claude_path, +) +from .claudecode_io import ( + rendered_change, + validate_api_origin, + validate_private_key, + validate_runtime_tree, +) +from .hook_template import generate_guard_script, generate_hook_script + + +@dataclass(frozen=True) +class ClaudeRuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +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) -> ClaudeRuntimePaths: + 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) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + if not same_claude_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + "Elydora guard runtime must use the managed agent directory: " + f"{guard_path}" + ) + return ClaudeRuntimePaths(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"Claude Code 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") + _agent_paths(config) + + +def preflight_claude_installation( + config: InstallConfig, document: ClaudeDocument +) -> ClaudeRuntimePaths: + if not document.file_path: + raise ValueError("Claude Code installation requires a settings path") + _validate_install_config(config) + 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 _present(changes: List[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def prepare_claude_installation( + config: InstallConfig, + paths: ClaudeRuntimePaths, + rendered: RenderedClaudeDocument, +) -> List[FileChange]: + 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, + ) + return _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 + ), + rendered_change(rendered), + ]) + + +def commit_claude_installation(changes: List[FileChange]) -> None: + write_changes(changes, "Install Claude Code hooks") + + +def prepare_claude_uninstall( + rendered: RenderedClaudeDocument, +) -> List[FileChange]: + return _present([rendered_change(rendered)]) + + +def commit_claude_uninstall(changes: List[FileChange]) -> None: + write_changes(changes, "Uninstall Claude Code hooks") diff --git a/elydora/plugins/claudecode_io.py b/elydora/plugins/claudecode_io.py new file mode 100644 index 0000000..ca0b95b --- /dev/null +++ b/elydora/plugins/claudecode_io.py @@ -0,0 +1,246 @@ +"""Claude Code settings discovery and managed runtime I/O.""" + +from __future__ import annotations + +import base64 +import os +from typing import Any, List, Optional +import urllib.parse + +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + MAX_SOURCE_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, +) +from ._strict_json import JsonObject, parse_json_object +from ._transaction import FileChange, source_change, write_changes +from .claudecode_contract import ( + AGENT_KEY, + CONFIG_FILE, + ClaudeDocument, + ClaudeRuntimeContract, + RenderedClaudeDocument, + create_claude_document, + parse_claude_document, + runtime_root, + same_claude_agent_id, + same_claude_path, +) + + +def claude_config_directory() -> str: + if "CLAUDE_CONFIG_DIR" not in os.environ: + return os.path.join(os.path.expanduser("~"), ".claude") + return os.path.abspath(os.environ["CLAUDE_CONFIG_DIR"]) + + +def claude_settings_path() -> str: + return os.path.join(claude_config_directory(), CONFIG_FILE) + + +def read_claude_document() -> ClaudeDocument: + file_path = claude_settings_path() + physical_directory_exists( + os.path.dirname(file_path), "Claude Code configuration directory" + ) + snapshot = read_physical_file( + file_path, "Claude Code user settings", MAX_SOURCE_BYTES + ) + if snapshot is None: + return create_claude_document(file_path) + return parse_claude_document(file_path, snapshot.contents) + + +def rendered_change( + rendered: RenderedClaudeDocument, +) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.file_path, + "Claude Code user settings", + rendered.document.raw, + rendered.next_source, + 0o600, + MAX_SOURCE_BYTES, + ) + + +def write_claude_document(rendered: RenderedClaudeDocument) -> None: + change = rendered_change(rendered) + if change is not None: + write_changes([change], "Uninstall Claude Code hooks") + + +def validate_api_origin(value: str, label: str = "base_url") -> None: + try: + 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" + ) + + +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 _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}' + ) + _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_claude_agent_id(agent_id, expected_agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime identity does not match Claude Code hooks: " + f"{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 _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 validate_runtime_tree(agent_directory: str, agent_id: str) -> None: + root = runtime_root() + 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.py", "Elydora guard runtime"), + ("hook.py", "Elydora audit runtime"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), + ) + artifact_states = [ + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ] + artifact_exists = any(artifact_states) + 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_claude_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime config identity does not match Claude Code agent " + f"{agent_id}: {config_path}" + ) + + +def _runtime_contract_exists(contract: ClaudeRuntimeContract) -> bool: + root = runtime_root() + agent_directory = os.path.dirname(contract.guard_path) + if ( + not same_claude_path(os.path.dirname(agent_directory), root) + or not same_claude_path( + contract.audit_path, os.path.join(agent_directory, "hook.py") + ) + ): + return False + if not physical_directory_exists(root, "Elydora runtime directory"): + return False + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return False + config_path = os.path.join(agent_directory, "config.json") + key_path = os.path.join(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 config is None or key is None or guard is None or audit is None: + return False + _validate_runtime_config(config, contract.agent_id, config_path) + validate_private_key(key.contents, "Elydora private key") + return bool(guard.contents) and bool(audit.contents) + + +def claude_runtime_files_exist( + contracts: List[ClaudeRuntimeContract], +) -> bool: + return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index e7c790b..a0d1862 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -13,8 +13,8 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "claudecode": { "name": "Claude Code", - "hook_event": "PostToolUse", - "config_path": "~/.claude/settings.json", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", + "config_path": "$CLAUDE_CONFIG_DIR/settings.json", }, "codex": { "name": "OpenAI Codex", diff --git a/tests/claudecode_support.py b/tests/claudecode_support.py new file mode 100644 index 0000000..93a3d4a --- /dev/null +++ b/tests/claudecode_support.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import subprocess # nosec B404 +import sys +from threading import Thread +from typing import Any, Mapping + +import pytest + +from elydora.plugins.claudecode import ClaudeCodePlugin +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@dataclass +class ClaudeFixture: + root_dir: Path + home_dir: Path + project_dir: Path + config_path: Path + agent_dir: Path + guard_path: Path + hook_path: Path + runtime_config_path: Path + private_key_path: Path + plugin: ClaudeCodePlugin + config: dict[str, str] + + def install(self) -> None: + self.plugin.install(self.config) + + def settings(self) -> dict[str, Any]: + return json.loads(self.config_path.read_text(encoding="utf-8")) + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + operation_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls, status: Any, operation_status: int) -> None: + cls.agent_status = status + cls.operation_status = operation_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: + 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: + 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( + type(self).operation_status, + {"operation": {"accepted": True}}, + ) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server( + *, status: Any = "active", operation_status: int = 201 +) -> ThreadingHTTPServer: + ElydoraApiHandler.reset(status, operation_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") + 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 prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_settings: object = MISSING, + base_url: str = "http://127.0.0.1:9", + config_override: object = None, +) -> ClaudeFixture: + home_dir = tmp_path / "home with spaces and 'quote %CLAUDE_HOOK_EVENT%" + project_dir = tmp_path / "project with spaces" + project_dir.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.chdir(project_dir) + if config_override is MISSING: + monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) + config_directory = home_dir / ".claude" + else: + configured = ( + str(home_dir / "custom claude") + if config_override is None + else str(config_override) + ) + monkeypatch.setenv("CLAUDE_CONFIG_DIR", configured) + config_directory = Path(os.path.abspath(configured)) + config_path = config_directory / "settings.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" + if existing_settings is not MISSING: + write_json(config_path, existing_settings) + return ClaudeFixture( + root_dir=tmp_path, + home_dir=home_dir, + project_dir=project_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=ClaudeCodePlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "claudecode", + "private_key": VALID_PRIVATE_KEY, + "kid": "key-1", + "token": "token-1", + "base_url": base_url, + "guard_script_path": str(guard_path), + }, + ) + + +def managed_handler( + settings: Mapping[str, Any], event: str +) -> dict[str, Any]: + for group in settings["hooks"][event]: + if set(group) != {"hooks"}: + continue + for handler in group["hooks"]: + if ( + set(handler) + == {"type", "command", "args", "timeout", "statusMessage"} + and handler.get("type") == "command" + and handler.get("timeout") == 10 + ): + return handler + raise AssertionError(f"managed handler for {event!r} not found") + + +def assert_native_handler( + handler: Mapping[str, Any], script_path: Path, status_message: str +) -> None: + assert set(handler) == { + "type", "command", "args", "timeout", "statusMessage", + } + assert handler["type"] == "command" + assert handler["command"] == sys.executable + assert handler["args"] == [str(script_path)] + assert handler["timeout"] == 10 + assert handler["statusMessage"] == status_message + + +def run_handler( + handler: Mapping[str, Any], payload: bytes, fixture: ClaudeFixture +) -> subprocess.CompletedProcess[bytes]: + return subprocess.run( + [str(handler["command"]), *map(str, handler["args"])], + input=payload, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env={ + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + }, + ) + + +def legacy_handler(script_path: Path) -> dict[str, str]: + command = ( + f'"{sys.executable}" {script_path}' + if script_path.name == "guard.py" + else str(script_path) + ) + return {"type": "command", "command": command} + + +def symlink_or_skip( + target: Path, link: Path, *, directory: bool = False +) -> None: + try: + link.symlink_to(target, target_is_directory=directory) + except OSError as error: + pytest.skip(f"Symbolic links unavailable: {error}") diff --git a/tests/test_claudecode_official.py b/tests/test_claudecode_official.py new file mode 100644 index 0000000..9371443 --- /dev/null +++ b/tests/test_claudecode_official.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import os +from pathlib import Path +import subprocess # nosec B404 + +import pytest + +from claudecode_support import prepare_fixture + + +CLAUDE_BINARY = os.environ.get("ELYDORA_CLAUDE_BINARY") + + +@pytest.mark.skipif( + not CLAUDE_BINARY, + reason="set ELYDORA_CLAUDE_BINARY to the official Claude Code executable", +) +def test_official_claude_code_accepts_installed_settings( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + "CLAUDE_CONFIG_DIR": str(fixture.config_path.parent), + "DISABLE_AUTOUPDATER": "1", + "DISABLE_TELEMETRY": "1", + } + + version = subprocess.run( + [str(CLAUDE_BINARY), "--version"], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + text=True, + ) + doctor = subprocess.run( + [str(CLAUDE_BINARY), "doctor"], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + text=True, + ) + + assert version.returncode == 0, version.stderr + assert "Claude Code" in version.stdout + assert doctor.returncode == 0, doctor.stderr + output = (doctor.stdout + "\n" + doctor.stderr).lower() + assert "invalid settings" not in output + assert "settings validation" not in output + assert "failed to parse" not in output diff --git a/tests/test_claudecode_plugin.py b/tests/test_claudecode_plugin.py new file mode 100644 index 0000000..5457dc8 --- /dev/null +++ b/tests/test_claudecode_plugin.py @@ -0,0 +1,451 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess # nosec B404 +import sys + +import pytest + +from claudecode_support import ( + AGENT_ID, + MISSING, + VALID_PRIVATE_KEY, + assert_native_handler, + legacy_handler, + managed_handler, + prepare_fixture, + write_json, + write_text, +) +from elydora.plugins.claudecode_contract import AUDIT_STATUS, GUARD_STATUS +from elydora.plugins.registry import SUPPORTED_AGENTS + + +def assert_managed_triple(settings: dict, fixture: object) -> None: + assert_native_handler( + managed_handler(settings, "PreToolUse"), + fixture.guard_path, + GUARD_STATUS, + ) + assert_native_handler( + managed_handler(settings, "PostToolUse"), + fixture.hook_path, + AUDIT_STATUS, + ) + assert_native_handler( + managed_handler(settings, "PostToolUseFailure"), + fixture.hook_path, + AUDIT_STATUS, + ) + + +def test_claude_code_is_registered_with_the_official_contract() -> None: + assert SUPPORTED_AGENTS["claudecode"] == { + "name": "Claude Code", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", + "config_path": "$CLAUDE_CONFIG_DIR/settings.json", + } + + +def test_install_preserves_settings_and_writes_one_exact_triple( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + existing = { + "$schema": "https://json.schemastore.org/claude-code-settings.json", + "model": "sonnet", + "disableAllHooks": False, + "hooks": { + "Notification": [{ + "matcher": "permission_prompt", + "hooks": [{ + "type": "http", + "url": "https://example.test/hook", + "timeout": 1, + }], + }], + "PreToolUse": [{ + "matcher": "Bash", + "hooks": [{ + "type": "command", + "command": "existing-command", + "timeout": 5, + }], + }], + }, + } + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=existing + ) + + fixture.install() + output = capsys.readouterr().out + assert "run /hooks and claude doctor" in output + installed = fixture.settings() + assert installed["model"] == "sonnet" + assert installed["hooks"]["Notification"] == existing["hooks"]["Notification"] + assert installed["hooks"]["PreToolUse"][0] == existing["hooks"]["PreToolUse"][0] + assert_managed_triple(installed, fixture) + first_source = fixture.config_path.read_text(encoding="utf-8") + + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == first_source + + +@pytest.mark.parametrize( + ("config_override", "expected_kind"), + [ + (None, "custom"), + (MISSING, "default"), + ("relative claude", "relative"), + ("", "cwd"), + ("~", "literal_tilde"), + ], +) +def test_claude_config_dir_matches_native_path_resolution( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + config_override: object, + expected_kind: str, +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, config_override=config_override + ) + + fixture.install() + + expected = { + "custom": fixture.home_dir / "custom claude" / "settings.json", + "default": fixture.home_dir / ".claude" / "settings.json", + "relative": fixture.project_dir / "relative claude" / "settings.json", + "cwd": fixture.project_dir / "settings.json", + "literal_tilde": fixture.project_dir / "~" / "settings.json", + }[expected_kind] + assert fixture.config_path == expected + assert expected.is_file() + + +def test_malformed_settings_fail_before_runtime_creation( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings="{ malformed" + ) + + with pytest.raises(ValueError, match="parse Claude Code user settings"): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == "{ malformed" + assert not fixture.agent_dir.exists() + + +def test_documented_handler_types_and_fields_are_preserved( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user_hooks = { + "SessionStart": [{"hooks": [{ + "type": "prompt", + "prompt": "Review context", + "model": "haiku", + "continueOnBlock": True, + "if": "always", + "once": False, + "statusMessage": "Reviewing", + "timeout": 0.5, + }]}], + "PreToolUse": [{"matcher": "Bash", "hooks": [{ + "type": "command", + "command": "user-command", + "args": ["--safe"], + "async": False, + "asyncRewake": True, + "shell": "powershell", + }]}], + "Stop": [{"hooks": [{ + "type": "agent", "prompt": "Verify completion", "model": "sonnet", + }]}], + "Notification": [{"hooks": [{ + "type": "http", + "url": "https://example.test/hook", + "headers": {"Authorization": "Bearer ${TOKEN}"}, + "allowedEnvVars": ["TOKEN"], + }]}], + "PostToolUse": [{"hooks": [{ + "type": "mcp_tool", + "server": "audit", + "tool": "record", + "input": {"source": "claude"}, + }]}], + } + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings={"hooks": user_hooks} + ) + + fixture.install() + + installed = fixture.settings()["hooks"] + for event, groups in user_hooks.items(): + assert installed[event][0] == groups[0] + + +INVALID_SETTINGS = [ + ('{"hooks":{},"hooks":{}}', 'duplicate field "hooks"'), + (json.dumps({"disableAllHooks": "yes"}), "must be a boolean"), + (json.dumps({"hooks": None}), 'field "hooks" must be an object'), + (json.dumps({"hooks": {"MadeUp": []}}), "unsupported hook event"), + (json.dumps({"hooks": {"PreToolUse": None}}), "must be an array"), + (json.dumps({"hooks": {"PreToolUse": [None]}}), "must be an object"), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [], "label": "x"}]}}), + 'unsupported field "label"', + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"matcher": 1, "hooks": []}]}}), + "matcher must be a string", + ), + (json.dumps({"hooks": {"PreToolUse": [{}]}}), "hooks array"), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [None]}]}}), + "handler", + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [{"type": "file"}]}]}}), + "unsupported type", + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "x", "invented": True, + }]}]}}), + 'unsupported field "invented"', + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "", + }]}]}}), + "non-empty string", + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "x", "args": [1], + }]}]}}), + "array of strings", + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "x", "timeout": 0, + }]}]}}), + "positive finite number", + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [{ + "type": "http", "url": "https://example.test", "headers": {"A": 1}, + }]}]}}), + "map names to strings", + ), + ( + json.dumps({"hooks": {"PreToolUse": [{"hooks": [{ + "type": "mcp_tool", "server": "s", "tool": "t", "input": [], + }]}]}}), + "must be an object", + ), +] + + +@pytest.mark.parametrize(("source", "pattern"), INVALID_SETTINGS) +def test_invalid_official_hook_shapes_fail_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + source: str, + pattern: str, +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=source + ) + + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == source + assert not fixture.agent_dir.exists() + + +def test_disabled_user_hooks_block_installation( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings={"disableAllHooks": True} + ) + + with pytest.raises(ValueError, match="disabled by disableAllHooks"): + fixture.install() + + assert not fixture.agent_dir.exists() + assert fixture.settings() == {"disableAllHooks": True} + + +def test_exact_legacy_hooks_migrate_and_lookalikes_survive_uninstall( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + lookalike = legacy_handler(fixture.guard_path) + lookalike["command"] += " --inspect" + write_json(fixture.config_path, { + "hooks": { + "PreToolUse": [ + {"hooks": [legacy_handler(fixture.guard_path)]}, + {"hooks": [lookalike]}, + ], + "PostToolUse": [{"hooks": [legacy_handler(fixture.hook_path)]}], + } + }) + + fixture.install() + settings = fixture.settings() + assert_managed_triple(settings, fixture) + assert any( + group["hooks"][0].get("command") == lookalike["command"] + for group in settings["hooks"]["PreToolUse"] + ) + settings["hooks"]["PreToolUse"][-1]["hooks"].append({ + "type": "command", "command": "user-command", "timeout": 5, + }) + write_json(fixture.config_path, settings) + + fixture.plugin.uninstall(AGENT_ID) + + remaining = fixture.settings()["hooks"] + assert remaining["PreToolUse"] == [ + {"hooks": [lookalike]}, + {"hooks": [{"type": "command", "command": "user-command", "timeout": 5}]}, + ] + assert "PostToolUse" not in remaining + assert "PostToolUseFailure" not in remaining + + +def test_uninstall_preserves_user_settings_and_removes_managed_file( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user = prepare_fixture( + monkeypatch, + tmp_path / "user", + existing_settings={"model": "sonnet", "hooks": {"Notification": []}}, + ) + user.install() + user.plugin.uninstall(AGENT_ID) + assert user.settings() == { + "model": "sonnet", "hooks": {"Notification": []}, + } + + managed = prepare_fixture(monkeypatch, tmp_path / "managed") + managed.install() + managed.plugin.uninstall(AGENT_ID) + assert not managed.config_path.exists() + + +def test_status_requires_enabled_unique_triple_and_strict_runtime( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.plugin.status()["installed"] is True + + settings = fixture.settings() + del settings["hooks"]["PostToolUseFailure"] + write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + + fixture.install() + settings = fixture.settings() + settings["hooks"]["PostToolUseFailure"].append( + settings["hooks"]["PostToolUseFailure"][-1] + ) + write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + + fixture.install() + settings = fixture.settings() + settings["disableAllHooks"] = True + write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + + settings["disableAllHooks"] = False + write_json(fixture.config_path, settings) + write_text(fixture.private_key_path, "invalid") + with pytest.raises(ValueError, match="canonical 32-byte"): + fixture.plugin.status() + + +def test_project_and_local_settings_remain_unchanged( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + project_settings = fixture.project_dir / ".claude" / "settings.json" + local_settings = fixture.project_dir / ".claude" / "settings.local.json" + project_source = '{"hooks":{"PreToolUse":[]}}\n' + local_source = '{"model":"haiku"}\n' + write_text(project_settings, project_source) + write_text(local_settings, local_source) + + fixture.install() + + assert project_settings.read_text(encoding="utf-8") == project_source + assert local_settings.read_text(encoding="utf-8") == local_source + + +def test_cli_install_status_and_uninstall_end_to_end( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + key_path = tmp_path / "install-private.key" + token_path = tmp_path / "install-token.txt" + write_text(key_path, VALID_PRIVATE_KEY + "\n") + write_text(token_path, "token-1\n") + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + "CLAUDE_CONFIG_DIR": str(fixture.config_path.parent), + } + root = Path(__file__).parents[1] + base = [sys.executable, "-m", "elydora.cli"] + + install = subprocess.run( + [ + *base, "install", "--agent", "claudecode", + "--org_id", "org-1", "--agent_id", AGENT_ID, + "--private_key_file", str(key_path), "--token_file", str(token_path), + "--kid", "key-1", "--base_url", "http://127.0.0.1:9", + ], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert install.returncode == 0, install.stderr + assert "PostToolUseFailure hooks installed" in install.stdout + assert_managed_triple(fixture.settings(), fixture) + + status = subprocess.run( + [*base, "status"], capture_output=True, check=False, + cwd=root, env=environment, text=True, + ) + assert status.returncode == 0, status.stderr + assert "Claude Code" in status.stdout + assert "[installed]" in status.stdout + + uninstall = subprocess.run( + [*base, "uninstall", "--agent", "claudecode", "--agent_id", AGENT_ID], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert uninstall.returncode == 0, uninstall.stderr + assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() diff --git a/tests/test_claudecode_runtime.py b/tests/test_claudecode_runtime.py new file mode 100644 index 0000000..40925bb --- /dev/null +++ b/tests/test_claudecode_runtime.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from claudecode_support import ( + ElydoraApiHandler, + managed_handler, + prepare_fixture, + run_handler, + server_base_url, + start_api_server, +) + + +def official_payload(event: str, **overrides: object) -> dict[str, object]: + value: dict[str, object] = { + "session_id": "session-1", + "prompt_id": "302d811d-0d17-41ad-a359-d2cb618fd42b", + "transcript_path": "/tmp/session-1.jsonl", + "cwd": "/tmp/project", + "permission_mode": "default", + "effort": {"level": "high"}, + "hook_event_name": event, + "tool_name": "Bash", + "tool_input": {"command": "pytest", "description": "Run tests"}, + "tool_use_id": "toolu_01ABC123", + } + value.update(overrides) + return value + + +def encoded(value: object) -> bytes: + return json.dumps(value, separators=(",", ":")).encode() + + +def test_runtimes_preserve_success_and_failure_payloads( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server() + try: + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + fixture.install() + settings = fixture.settings() + guard = managed_handler(settings, "PreToolUse") + success_audit = managed_handler(settings, "PostToolUse") + failure_audit = managed_handler(settings, "PostToolUseFailure") + + pre = official_payload("PreToolUse") + guard_result = run_handler(guard, encoded(pre), fixture) + success = official_payload( + "PostToolUse", + tool_response={ + "stdout": "tests passed", + "stderr": "", + "interrupted": False, + "isImage": False, + }, + ) + success_result = run_handler(success_audit, encoded(success), fixture) + failure = official_payload( + "PostToolUseFailure", + error="Command exited with non-zero status code 1", + is_interrupt=False, + duration_ms=4187, + ) + failure_result = run_handler(failure_audit, encoded(failure), fixture) + finally: + server.shutdown() + server.server_close() + + assert guard_result.returncode == 0 + assert guard_result.stdout == b"" + assert guard_result.stderr == b"" + assert success_result.returncode == 0 + assert failure_result.returncode == 0 + assert [request["method"] for request in ElydoraApiHandler.requests] == [ + "GET", "POST", "POST", + ] + success_operation = ElydoraApiHandler.requests[1]["json"] + failure_operation = ElydoraApiHandler.requests[2]["json"] + assert success_operation["payload"] == success + assert failure_operation["payload"] == failure + assert success_operation["subject"] == {"session_id": "session-1"} + assert success_operation["action"] == {"tool": "Bash"} + assert ElydoraApiHandler.requests[1]["authorization"] == "Bearer token-1" + assert failure_operation["prev_chain_hash"] == success_operation["chain_hash"] + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_returns_official_exit_code_two( + 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_handler( + managed_handler(fixture.settings(), "PreToolUse"), + encoded(official_payload("PreToolUse")), + fixture, + ) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 2 + assert result.stdout == b"" + assert status.encode() in result.stderr.lower() + assert b"Tool execution blocked" in result.stderr + + +def test_runtime_failures_are_observable_and_fail_open( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + settings = fixture.settings() + guard = managed_handler(settings, "PreToolUse") + audit = managed_handler(settings, "PostToolUse") + + guard_result = run_handler( + guard, encoded(official_payload("PreToolUse")), fixture + ) + audit_result = run_handler( + audit, + encoded(official_payload("PostToolUse", tool_response={"stdout": ""})), + fixture, + ) + malformed = run_handler(audit, b"{ malformed", fixture) + + assert guard_result.returncode == 0 + assert b"Failed to resolve agent status" in guard_result.stderr + assert audit_result.returncode == 0 + assert malformed.returncode == 0 + log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8") + assert "invalid JSON" in log + assert "refused" in log.lower() or "urlopen" in log.lower() diff --git a/tests/test_claudecode_transaction.py b/tests/test_claudecode_transaction.py new file mode 100644 index 0000000..2c882a2 --- /dev/null +++ b/tests/test_claudecode_transaction.py @@ -0,0 +1,283 @@ +from __future__ import annotations + +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 +import sys + +import pytest + +from claudecode_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + prepare_fixture, + symlink_or_skip, + write_json, + write_text, +) +from elydora.plugins import _transaction, claudecode_installation +from elydora.plugins.claudecode_contract import ( + AUDIT_STATUS, + GUARD_STATUS, + build_claude_group, + remove_managed_claude_hooks, + render_claude_document, +) +from elydora.plugins.claudecode_io import read_claude_document +from elydora.utils import base64url_encode + + +RUNTIME_NAMES = ("guard.py", "config.json", "private.key", "hook.py") + + +def prepare_installation(fixture: object) -> list[_transaction.FileChange]: + document = read_claude_document() + paths = claudecode_installation.preflight_claude_installation( + fixture.config, document + ) + cleaned = remove_managed_claude_hooks(document.hooks) + hooks = { + **cleaned, + "PreToolUse": [ + *cleaned.get("PreToolUse", []), + build_claude_group(paths.guard_path, GUARD_STATUS), + ], + "PostToolUse": [ + *cleaned.get("PostToolUse", []), + build_claude_group(paths.audit_path, AUDIT_STATUS), + ], + "PostToolUseFailure": [ + *cleaned.get("PostToolUseFailure", []), + build_claude_group(paths.audit_path, AUDIT_STATUS), + ], + } + return claudecode_installation.prepare_claude_installation( + fixture.config, paths, render_claude_document(document, hooks) + ) + + +def assert_no_transaction_files(root: Path) -> None: + assert not list(root.rglob("*.tmp")) + assert not list(root.rglob("*.rollback")) + + +def test_transaction_rolls_back_all_five_files( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + original = '{"model":"sonnet"}\n' + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=original + ) + real_replace = _transaction.os.replace + failed = False + + def fail_settings_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 Claude settings failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_settings_once) + + with pytest.raises(OSError, match="injected Claude settings failure"): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == original + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_concurrent_settings_change_is_detected_before_runtime_commits( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings={"model": "sonnet"} + ) + changes = prepare_installation(fixture) + concurrent = '{"model":"opus"}\n' + fixture.config_path.write_text(concurrent, encoding="utf-8") + + with pytest.raises(OSError, match="changed during installation"): + claudecode_installation.commit_claude_installation(changes) + + assert fixture.config_path.read_text(encoding="utf-8") == concurrent + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_recovery_preserves_original_after_committed_runtime_changes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + original = fixture.runtime_config_path.read_text(encoding="utf-8") + fixture.config["org_id"] = "org-updated" + fixture.config["private_key"] = base64url_encode(bytes(range(31, -1, -1))) + changes = prepare_installation(fixture) + real_replace = _transaction.os.replace + calls = 0 + + def change_first_then_fail(source: str, destination: str) -> None: + nonlocal calls + calls += 1 + if calls == 1: + real_replace(source, destination) + write_text(Path(destination), "external change\n") + return + raise OSError("injected later commit failure") + + monkeypatch.setattr(_transaction.os, "replace", change_first_then_fail) + + with pytest.raises(OSError, match="original content preserved at"): + claudecode_installation.commit_claude_installation(changes) + + assert fixture.runtime_config_path.read_text(encoding="utf-8") == "external change\n" + rollbacks = list(fixture.agent_dir.glob(".config.json.*.rollback")) + assert len(rollbacks) == 1 + assert rollbacks[0].read_text(encoding="utf-8") == original + + +@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_settings_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.config_path.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_path, { + "agent_id": "another-agent", "agent_name": "claudecode", + }) + original = fixture.runtime_config_path.read_text(encoding="utf-8") + + with pytest.raises(ValueError, match="identity does not match"): + fixture.install() + + assert fixture.runtime_config_path.read_text(encoding="utf-8") == original + assert not fixture.config_path.exists() + + +@pytest.mark.parametrize("kind", ["configuration", "settings", "runtime"]) +def test_linked_configuration_and_runtime_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 == "settings": + write_text(target, "{}\n") + fixture.config_path.parent.mkdir(parents=True) + symlink_or_skip(target, fixture.config_path) + else: + target.mkdir() + link = ( + fixture.config_path.parent + if kind == "configuration" + else fixture.agent_dir.parent + ) + if kind == "runtime" and link.exists(): + shutil.rmtree(link) + link.parent.mkdir(parents=True, exist_ok=True) + symlink_or_skip(target, link, directory=True) + + with pytest.raises(OSError, match="physical (file|directory)"): + fixture.install() + + +@pytest.mark.parametrize( + ("field", "value", "pattern"), + [ + ("agent_name", "codex", "requires agent_name claudecode"), + ("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.config_path.exists() + assert not fixture.agent_dir.exists() + + +def test_status_surfaces_malformed_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") + + with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.plugin.status() + + +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.home_dir) + + +def test_cli_preflight_preserves_malformed_settings( + tmp_path: Path, +) -> None: + home_dir = tmp_path / "cli-home" + settings_path = home_dir / ".claude" / "settings.json" + key_path = tmp_path / "private.key" + write_text(settings_path, "{ malformed") + write_text(key_path, VALID_PRIVATE_KEY) + environment = {**os.environ, "HOME": str(home_dir), "USERPROFILE": str(home_dir)} + environment.pop("CLAUDE_CONFIG_DIR", None) + + result = subprocess.run( + [ + sys.executable, "-m", "elydora.cli", "install", + "--agent", "claudecode", "--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 "parse Claude Code user settings" in result.stderr + assert settings_path.read_text(encoding="utf-8") == "{ malformed" + assert not (home_dir / ".elydora").exists() diff --git a/tests/test_cli_security.py b/tests/test_cli_security.py index e404840..a2e42d6 100644 --- a/tests/test_cli_security.py +++ b/tests/test_cli_security.py @@ -15,7 +15,6 @@ from elydora._cli_secrets import ( ) from elydora.plugins._file_io import write_text_atomic from elydora.plugins import ( - claudecode, copilot, cursor, gemini, @@ -374,7 +373,6 @@ def test_atomic_secret_write_preserves_existing_file_on_replace_failure( @pytest.mark.parametrize( "module, plugin_type, agent_name", [ - (claudecode, claudecode.ClaudeCodePlugin, "claudecode"), (copilot, copilot.CopilotPlugin, "copilot"), (cursor, cursor.CursorPlugin, "cursor"), (gemini, gemini.GeminiPlugin, "gemini"), diff --git a/tests/test_settings_hook_detection.py b/tests/test_settings_hook_detection.py index e790ac6..b6b66b4 100644 --- a/tests/test_settings_hook_detection.py +++ b/tests/test_settings_hook_detection.py @@ -4,7 +4,6 @@ from collections.abc import Callable import pytest -from elydora.plugins.claudecode import _is_elydora_hook as is_claude_hook from elydora.plugins.gemini import _is_elydora_hook as is_gemini_hook from elydora.plugins.letta import _is_elydora_hook as is_letta_hook @@ -12,7 +11,7 @@ from elydora.plugins.letta import _is_elydora_hook as is_letta_hook HookDetector = Callable[[dict, str], bool] -@pytest.mark.parametrize("detector", [is_claude_hook, is_gemini_hook, is_letta_hook]) +@pytest.mark.parametrize("detector", [is_gemini_hook, is_letta_hook]) def test_hook_detection_handles_only_string_commands(detector: HookDetector) -> None: assert detector({"command": 42}, "") is False assert detector({"hooks": [{"command": None}]}, "") is False