diff --git a/AGENTS.md b/AGENTS.md index ff5e495..6dc91e2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,7 +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. +- Write Cursor hooks only to `~/.cursor/hooks.json`; keep project and enterprise sources read-only. Preserve user hooks, migrate the prior versionless Elydora contract, and own exact native `preToolUse`, `postToolUse`, and `postToolUseFailure` commands. Forward native success and failure payloads, retain PowerShell exit code `2`, set every handler to `failClosed` with a bounded timeout, and commit the guard, audit script, runtime identity, private key, and hook configuration as one rollback-capable transaction. - 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 f7428a4..0208b34 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,7 @@ 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. +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. 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. diff --git a/elydora/plugins/cursor.py b/elydora/plugins/cursor.py index ff28169..ef0e977 100644 --- a/elydora/plugins/cursor.py +++ b/elydora/plugins/cursor.py @@ -2,11 +2,14 @@ from __future__ import annotations +import base64 import json import os +import urllib.parse from typing import List, Optional from elydora._runtime_paths import resolve_agent_directory +from elydora.utils import base64url_encode from ._transaction import FileChange from .base import AgentPlugin, InstallConfig, PluginStatus @@ -29,6 +32,7 @@ from .cursor_io import ( require_runtime_directory, runtime_change, runtime_files_exist, + validate_config_directory, validate_runtime_identity, write_cursor_changes, ) @@ -72,28 +76,73 @@ def _agent_paths(config: InstallConfig) -> tuple[str, str, str]: return agent_id, agent_directory, guard_path +def _validate_install_config(config: InstallConfig) -> None: + for field in ("org_id", "agent_id", "kid", "private_key"): + 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"Cursor installation requires agent_name {AGENT_KEY}") + private_key = config["private_key"] + try: + padded = private_key + "=" * ((4 - len(private_key) % 4) % 4) + seed = base64.b64decode( + padded.replace("-", "+").replace("_", "/"), + validate=True, + ) + except (ValueError, UnicodeEncodeError) as error: + raise ValueError( + "private_key must be a canonical 32-byte base64url value" + ) from error + if len(seed) != 32 or base64url_encode(seed) != private_key: + raise ValueError("private_key must be a canonical 32-byte base64url value") + base_url = config.get("base_url", "https://api.elydora.com") + if not isinstance(base_url, str): + raise ValueError("base_url must be a string") + parsed = urllib.parse.urlsplit(base_url) + if parsed.scheme not in ("http", "https") or not parsed.netloc: + raise ValueError("base_url must be an absolute HTTP or HTTPS URL") + token = config.get("token", "") + if not isinstance(token, str): + raise ValueError("token must be a string") + + class CursorPlugin(AgentPlugin): """Install Elydora into Cursor's native global user hooks.""" manages_guard_runtime = True def preflight_install(self, config: InstallConfig) -> None: + _validate_install_config(config) read_document() + validate_config_directory() agent_id, agent_directory, guard_path = _agent_paths(config) + if not physical_directory_exists(runtime_root()): + return if not physical_directory_exists(agent_directory): return - validate_runtime_identity( + identity_exists = validate_runtime_identity( os.path.join(agent_directory, "config.json"), agent_id, ) + runtime_exists = False 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"), + (os.path.join(agent_directory, "chain-state.json"), "Elydora chain state"), + (os.path.join(agent_directory, "status-cache.json"), "Elydora status cache"), + (os.path.join(agent_directory, "error.log"), "Elydora error log"), ): - physical_file_exists(file_path, label) + runtime_exists = physical_file_exists(file_path, label) or runtime_exists + if runtime_exists and not identity_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) def install(self, config: InstallConfig) -> None: + self.preflight_install(config) document = read_document() agent_id, agent_directory, guard_path = _agent_paths(config) require_runtime_directory(agent_directory) @@ -110,17 +159,25 @@ class CursorPlugin(AgentPlugin): *hooks.get("postToolUse", []), build_handler(audit_path), ] + hooks["postToolUseFailure"] = [ + *hooks.get("postToolUseFailure", []), + 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", + fail_closed=True, + native_payload=True, + agent_name=AGENT_KEY, ) guard_script = generate_guard_script( AGENT_KEY, agent_id, success_output='{"permission":"allow"}\n', + fail_closed=True, ) changes = _present([ runtime_change( diff --git a/elydora/plugins/cursor_contract.py b/elydora/plugins/cursor_contract.py index 7d94400..cd849ed 100644 --- a/elydora/plugins/cursor_contract.py +++ b/elydora/plugins/cursor_contract.py @@ -240,6 +240,7 @@ def _contains_managed_hook(hooks: CursorHooks) -> bool: for event, script_name in ( ("preToolUse", GUARD_SCRIPT), ("postToolUse", AUDIT_SCRIPT), + ("postToolUseFailure", AUDIT_SCRIPT), ) for handler in hooks.get(event, []) ) @@ -269,6 +270,7 @@ def remove_managed_hooks( for event, script_name in ( ("preToolUse", GUARD_SCRIPT), ("postToolUse", AUDIT_SCRIPT), + ("postToolUseFailure", AUDIT_SCRIPT), ): handlers = [] for handler in result.get(event, []): @@ -295,6 +297,7 @@ def _entirely_managed(document: CursorDocument) -> bool: script_name = { "preToolUse": GUARD_SCRIPT, "postToolUse": AUDIT_SCRIPT, + "postToolUseFailure": AUDIT_SCRIPT, }.get(event) if script_name is None or not handlers: return False @@ -335,9 +338,10 @@ def _managed_ids(handlers: List[JsonObject], script_name: str) -> Dict[str, str] def runtime_contracts(hooks: CursorHooks) -> List[RuntimeContract]: guards = _managed_ids(hooks.get("preToolUse", []), GUARD_SCRIPT) audits = _managed_ids(hooks.get("postToolUse", []), AUDIT_SCRIPT) + failures = _managed_ids(hooks.get("postToolUseFailure", []), AUDIT_SCRIPT) contracts = [] for key, agent_id in guards.items(): - if key not in audits: + if key not in audits or key not in failures: continue agent_directory = os.path.join(runtime_root(), agent_id) contracts.append(RuntimeContract( diff --git a/elydora/plugins/cursor_io.py b/elydora/plugins/cursor_io.py index 7b94264..78c82ac 100644 --- a/elydora/plugins/cursor_io.py +++ b/elydora/plugins/cursor_io.py @@ -48,6 +48,18 @@ def read_document() -> CursorDocument: return create_document(file_path) if raw is None else parse_document(file_path, raw) +def validate_config_directory() -> None: + directory = os.path.dirname(config_path()) + try: + metadata = os.lstat(directory) + except FileNotFoundError: + return + except OSError as error: + raise OSError(f"Inspect Cursor hooks directory at {directory}: {error}") from error + if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"Cursor hooks directory is not a physical directory: {directory}") + + def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: if not rendered.changed: return None @@ -122,10 +134,10 @@ def _read_runtime_config(file_path: str) -> Optional[JsonObject]: def validate_runtime_identity( file_path: str, agent_id: str, -) -> None: +) -> bool: config = _read_runtime_config(file_path) if config is None: - return + return False if config.get("agent_name") != AGENT_KEY or not _same_agent_id( config.get("agent_id"), agent_id ): @@ -133,6 +145,7 @@ def validate_runtime_identity( f"Elydora runtime config identity does not match Cursor agent {agent_id}: " f"{file_path}" ) + return True def _same_agent_id(value: Any, expected: str) -> bool: diff --git a/elydora/plugins/guard_template.py b/elydora/plugins/guard_template.py new file mode 100644 index 0000000..8ae2cf6 --- /dev/null +++ b/elydora/plugins/guard_template.py @@ -0,0 +1,216 @@ +"""Generate self-contained agent status guards.""" + +from __future__ import annotations + +import sys + +from .runtime_reader_template import PROTECTED_RUNTIME_READER + + +def generate_guard_script( + agent_name: str, + agent_id: str, + success_output: str = "", + *, + fail_closed: bool = False, +) -> str: + """Return a self-contained status guard for a command-hook provider.""" + return f'''#!{sys.executable} +"""Elydora agent guard for {agent_name}. Generated by elydora-sdk.""" + +import json +import math +import os +import stat +import sys +import time +import urllib.error +import urllib.parse +import urllib.request + +AGENT_NAME = {agent_name!r} +AGENT_ID = {agent_id!r} +SUCCESS_OUTPUT = {success_output!r} +FAIL_CLOSED = {fail_closed!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") +CACHE_TTL_SECONDS = 60 + +{PROTECTED_RUNTIME_READER} + + +def validate_status(value, source): + if value not in ("active", "frozen", "revoked"): + raise ValueError(source + " contains an invalid agent status") + return value + + +def block_agent(status): + state = "frozen" if status == "frozen" else "revoked" + message = 'Agent "' + AGENT_NAME + '" is ' + state + " in Elydora." + if SUCCESS_OUTPUT: + sys.stdout.write(json.dumps({{ + "permission": "deny", + "userMessage": message, + "agentMessage": message + " Tool execution is blocked.", + }}) + "\\n") + sys.stdout.flush() + sys.stderr.write("[Elydora guard] " + message + " Tool execution blocked.\\n") + raise SystemExit(2) + + +def read_input(): + raw = sys.stdin.read() + if not raw.strip(): + if FAIL_CLOSED: + raise ValueError("Hook input is empty") + return + try: + value = json.loads(raw) + except (json.JSONDecodeError, ValueError) as error: + if FAIL_CLOSED: + raise ValueError("Hook input is invalid JSON: " + str(error)) from error + return + if FAIL_CLOSED and not isinstance(value, dict): + raise ValueError("Hook input must contain a JSON object") + + +def read_config(): + try: + raw = read_protected_file( + CONFIG_PATH, + "Agent config", + MAX_PROTECTED_CONFIG_BYTES, + ).decode("utf-8") + config = json.loads(raw) + except Exception as error: + raise RuntimeError("Failed to read agent config: " + str(error)) from error + if not isinstance(config, dict): + raise ValueError("Agent config must contain a JSON object") + if config.get("agent_id") != AGENT_ID: + raise ValueError("Agent config identity does not match the generated guard") + configured_name = config.get("agent_name") + if configured_name is not None and configured_name != AGENT_NAME: + raise ValueError("Agent config identity does not match the generated guard") + return config + + +def read_status_cache(): + try: + raw = read_protected_file( + STATUS_CACHE_PATH, + "Status cache", + MAX_PROTECTED_CONFIG_BYTES, + require_private_permissions=False, + ).decode("utf-8") + except FileNotFoundError: + return None + try: + cache = json.loads(raw) + except (json.JSONDecodeError, ValueError) as error: + raise ValueError("Status cache is invalid JSON: " + str(error)) from error + if not isinstance(cache, dict): + raise ValueError("Status cache must contain a JSON object") + cached_at = cache.get("cached_at") + if isinstance(cached_at, bool) or not isinstance(cached_at, (int, float)): + raise ValueError("Status cache has an invalid timestamp") + if not math.isfinite(cached_at): + raise ValueError("Status cache has an invalid timestamp") + age = time.time() - cached_at + if age < 0: + raise ValueError("Status cache timestamp is in the future") + return validate_status(cache.get("status"), "Status cache"), age + + +def write_status_cache(status): + token = os.urandom(8).hex() + temporary_path = STATUS_CACHE_PATH + "." + str(os.getpid()) + "." + token + ".tmp" + descriptor = -1 + try: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0) + descriptor = os.open(temporary_path, flags, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as file: + descriptor = -1 + json.dump({{"status": status, "cached_at": time.time()}}, file) + file.flush() + os.fsync(file.fileno()) + os.replace(temporary_path, STATUS_CACHE_PATH) + finally: + if descriptor >= 0: + os.close(descriptor) + try: + os.remove(temporary_path) + except FileNotFoundError: + pass + + +def resolve_remote_status(config): + base_url = config.get("base_url", "https://api.elydora.com") + if not isinstance(base_url, str): + raise ValueError("Agent config base_url must be a string") + parsed = urllib.parse.urlsplit(base_url) + if parsed.scheme not in ("http", "https"): + raise ValueError("Agent config base_url must use HTTP or HTTPS") + token = config.get("token", "") + if not isinstance(token, str): + raise ValueError("Agent config token must be a string") + url = base_url.rstrip("/") + "/v1/agents/" + urllib.parse.quote(AGENT_ID, safe="") + request = urllib.request.Request(url, method="GET") + request.add_header("Accept", "application/json") + if token: + request.add_header("Authorization", "Bearer " + token) + try: + with urllib.request.urlopen(request, timeout=3) as response: + value = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as error: + raise RuntimeError( + "Agent status API returned HTTP " + str(error.code) + ) from error + status = value.get("agent", {{}}).get("status") if isinstance(value, dict) else None + return validate_status(status, "Agent status API") + + +def main(): + read_input() + try: + config = read_config() + except Exception as error: + if FAIL_CLOSED: + raise + sys.stderr.write("[Elydora guard] " + str(error) + "\\n") + return + try: + cache = read_status_cache() + if cache is not None and cache[1] < CACHE_TTL_SECONDS: + if cache[0] != "active": + block_agent(cache[0]) + return + except Exception as error: + if FAIL_CLOSED: + raise + sys.stderr.write("[Elydora guard] " + str(error) + "\\n") + try: + status = resolve_remote_status(config) + write_status_cache(status) + if status != "active": + block_agent(status) + except SystemExit: + raise + except Exception as error: + if FAIL_CLOSED: + raise RuntimeError("Failed to resolve agent status: " + str(error)) from error + sys.stderr.write("[Elydora guard] Failed to resolve agent status: " + str(error) + "\\n") + + +if __name__ == "__main__": + try: + main() + except SystemExit: + raise + except Exception as error: + sys.stderr.write("[Elydora guard] " + repr(error) + "\\n") + raise SystemExit(1) + if SUCCESS_OUTPUT: + sys.stdout.write(SUCCESS_OUTPUT) +''' diff --git a/elydora/plugins/hook_template.py b/elydora/plugins/hook_template.py index 1e5f482..0727fea 100644 --- a/elydora/plugins/hook_template.py +++ b/elydora/plugins/hook_template.py @@ -1,14 +1,14 @@ -"""Generates self-contained Python hook scripts for AI agent integrations. - -The generated script contains all crypto and networking inline so it has zero -import dependencies beyond the Python stdlib + cryptography (already an -Elydora SDK dependency). -""" +"""Generate self-contained audit hook scripts.""" from __future__ import annotations import sys +from .guard_template import generate_guard_script +from .runtime_reader_template import PROTECTED_RUNTIME_READER + +__all__ = ["generate_guard_script", "generate_hook_script"] + def generate_hook_script( *, @@ -17,76 +17,63 @@ def generate_hook_script( kid: str, base_url: str, success_output: str = "", + fail_closed: bool = False, + native_payload: bool = False, + agent_name: str = "", ) -> str: - """Return a complete, self-contained Python hook script as a string. - - The script: - 1. Reads JSON from stdin (tool-use event) - 2. Constructs an Elydora Operation Record (EOR) - 3. Signs it with Ed25519 - 4. POSTs it fire-and-forget to the Elydora API - 5. Always exits 0 (never blocks the host agent) - - Chain state is persisted in ~/.elydora/{agent_id}/chain-state.json. - """ - shebang = sys.executable - return f'''#!{shebang} -"""Elydora audit hook — auto-generated, do not edit.""" + """Return a self-contained signed audit hook.""" + return f'''#!{sys.executable} +"""Elydora audit hook. Generated by elydora-sdk.""" import base64 import hashlib import json import math import os +import re import stat import struct import sys import time +import urllib.error +import urllib.parse +import urllib.request -# --------------------------------------------------------------------------- -# Configuration (baked at install time) -# --------------------------------------------------------------------------- ORG_ID = {org_id!r} AGENT_ID = {agent_id!r} +AGENT_NAME = {agent_name!r} KID = {kid!r} BASE_URL = {base_url!r} SUCCESS_OUTPUT = {success_output!r} - +FAIL_CLOSED = {fail_closed!r} +NATIVE_PAYLOAD = {native_payload!r} ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") +CONFIG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "config.json") PRIVATE_KEY_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "private.key") CHAIN_STATE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "chain-state.json") ERROR_LOG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "error.log") ZERO_CHAIN_HASH = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA" -MAX_PRIVATE_KEY_BYTES = 64 * 1024 -# --------------------------------------------------------------------------- -# Base64url helpers -# --------------------------------------------------------------------------- +{PROTECTED_RUNTIME_READER} -def base64url_encode(data: bytes) -> str: + +def base64url_encode(data): return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") -def base64url_decode(s: str) -> bytes: - s = s.replace("-", "+").replace("_", "/") - padding = 4 - len(s) % 4 - if padding != 4: - s += "=" * padding - return base64.b64decode(s) -# --------------------------------------------------------------------------- -# SHA-256 -# --------------------------------------------------------------------------- +def base64url_decode(value): + encoded = value.replace("-", "+").replace("_", "/") + encoded += "=" * ((4 - len(encoded) % 4) % 4) + return base64.b64decode(encoded, validate=True) + def sha256_base64url(data): if isinstance(data, str): data = data.encode("utf-8") return base64url_encode(hashlib.sha256(data).digest()) -# --------------------------------------------------------------------------- -# JCS canonicalization (RFC 8785) -# --------------------------------------------------------------------------- -def _jcs_number(value): +def jcs_number(value): if isinstance(value, bool): return "true" if value else "false" if isinstance(value, int): @@ -97,362 +84,278 @@ def _jcs_number(value): return "0" return json.dumps(value) + def jcs_canonicalize(value): if value is None: return "null" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, (int, float)): - return _jcs_number(value) + return jcs_number(value) if isinstance(value, str): return json.dumps(value, ensure_ascii=False) if isinstance(value, list): - return "[" + ",".join(jcs_canonicalize(v) for v in value) + "]" + return "[" + ",".join(jcs_canonicalize(item) for item in value) + "]" if isinstance(value, dict): - keys = sorted(value.keys()) - pairs = [] - for k in keys: - pairs.append(json.dumps(k, ensure_ascii=False) + ":" + jcs_canonicalize(value[k])) + pairs = [ + json.dumps(key, ensure_ascii=False) + ":" + jcs_canonicalize(value[key]) + for key in sorted(value) + ] return "{{" + ",".join(pairs) + "}}" return json.dumps(value) -# --------------------------------------------------------------------------- -# Hash helpers -# --------------------------------------------------------------------------- def compute_payload_hash(payload): return sha256_base64url(jcs_canonicalize(payload)) -def compute_chain_hash(prev_chain_hash, payload_hash, operation_id, issued_at): - input_str = f"{{prev_chain_hash}}|{{payload_hash}}|{{operation_id}}|{{issued_at}}" - return sha256_base64url(input_str) -# --------------------------------------------------------------------------- -# Ed25519 signing (uses cryptography library) -# --------------------------------------------------------------------------- +def compute_chain_hash(previous, payload_hash, operation_id, issued_at): + value = f"{{previous}}|{{payload_hash}}|{{operation_id}}|{{issued_at}}" + return sha256_base64url(value) -def sign_ed25519(private_key_b64url, data): - from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey - seed = base64url_decode(private_key_b64url) - key = Ed25519PrivateKey.from_private_bytes(seed) - return base64url_encode(key.sign(data)) def read_private_key(): - before = os.lstat(PRIVATE_KEY_PATH) - if not stat.S_ISREG(before.st_mode): - raise ValueError("Private key path is not a regular file: " + PRIVATE_KEY_PATH) - if before.st_size > MAX_PRIVATE_KEY_BYTES: - raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH) - if os.name != "nt" and before.st_mode & (stat.S_IRWXG | stat.S_IRWXO): - raise PermissionError("Private key file permissions are too broad: " + PRIVATE_KEY_PATH) + encoded = read_protected_file(PRIVATE_KEY_PATH, "Private key").decode("utf-8") + if not encoded or "\\r" in encoded or "\\n" in encoded or "\\0" in encoded: + raise ValueError("Private key file must contain exactly one line") + seed = base64url_decode(encoded) + if len(seed) != 32 or base64url_encode(seed) != encoded: + raise ValueError("Private key must be a canonical 32-byte base64url value") + return seed - flags = os.O_RDONLY - flags |= getattr(os, "O_BINARY", 0) - flags |= getattr(os, "O_CLOEXEC", 0) - flags |= getattr(os, "O_NOFOLLOW", 0) - descriptor = os.open(PRIVATE_KEY_PATH, flags) + +def sign_ed25519(seed, data): + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + return base64url_encode(Ed25519PrivateKey.from_private_bytes(seed).sign(data)) + + +def generate_uuidv7(): + timestamp = struct.pack(">Q", int(time.time() * 1000))[2:] + random_a = (struct.unpack(">H", os.urandom(2))[0] & 0x0FFF) | 0x7000 + random_b = bytearray(os.urandom(8)) + random_b[0] = (random_b[0] & 0x3F) | 0x80 + value = (timestamp + struct.pack(">H", random_a) + bytes(random_b)).hex() + return f"{{value[0:8]}}-{{value[8:12]}}-{{value[12:16]}}-{{value[16:20]}}-{{value[20:32]}}" + + +def log_error(error): + descriptor = -1 try: - file = os.fdopen(descriptor, "rb") - descriptor = -1 - with file: - after = os.fstat(file.fileno()) - if not stat.S_ISREG(after.st_mode): - raise ValueError("Private key path is not a regular file: " + PRIVATE_KEY_PATH) - if after.st_size > MAX_PRIVATE_KEY_BYTES: - raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH) - if os.name != "nt" and after.st_mode & (stat.S_IRWXG | stat.S_IRWXO): - raise PermissionError("Private key file permissions are too broad: " + PRIVATE_KEY_PATH) - if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino): - raise OSError("Private key file changed while opening: " + PRIVATE_KEY_PATH) - encoded = file.read(MAX_PRIVATE_KEY_BYTES + 1) + message = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + message += " [elydora-hook] " + repr(error) + "\\n" + flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT + flags |= getattr(os, "O_BINARY", 0) | getattr(os, "O_CLOEXEC", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(ERROR_LOG_PATH, flags, 0o600) + metadata = os.fstat(descriptor) + if not stat.S_ISREG(metadata.st_mode): + raise ValueError("Error log path is not a physical file: " + ERROR_LOG_PATH) + with os.fdopen(descriptor, "a", encoding="utf-8") as file: + descriptor = -1 + file.write(message) + if os.name != "nt": + os.chmod(ERROR_LOG_PATH, 0o600) + except Exception as log_failure: + sys.stderr.write("[Elydora audit] Failed to write error log: " + str(log_failure) + "\\n") finally: if descriptor >= 0: os.close(descriptor) - if len(encoded) > MAX_PRIVATE_KEY_BYTES: - raise ValueError("Private key file exceeds the size limit: " + PRIVATE_KEY_PATH) - value = encoded.decode("utf-8") - if not value or "\\r" in value or "\\n" in value or "\\0" in value: - raise ValueError("Private key file must contain exactly one line") - return value - -# --------------------------------------------------------------------------- -# UUIDv7 -# --------------------------------------------------------------------------- - -def generate_uuidv7(): - timestamp_ms = int(time.time() * 1000) - ts_bytes = struct.pack(">Q", timestamp_ms)[2:] - rand_a = struct.unpack(">H", os.urandom(2))[0] - rand_a = (rand_a & 0x0FFF) | 0x7000 - rand_b = bytearray(os.urandom(8)) - rand_b[0] = (rand_b[0] & 0x3F) | 0x80 - uuid_bytes = ts_bytes + struct.pack(">H", rand_a) + bytes(rand_b) - h = uuid_bytes.hex() - return f"{{h[0:8]}}-{{h[8:12]}}-{{h[12:16]}}-{{h[16:20]}}-{{h[20:32]}}" - -def generate_nonce(): - return base64url_encode(os.urandom(16)) - -# --------------------------------------------------------------------------- -# Chain state persistence -# --------------------------------------------------------------------------- def read_chain_state(): try: - with open(CHAIN_STATE_PATH, "r") as f: - state = json.load(f) - return state.get("prev_chain_hash", ZERO_CHAIN_HASH) or ZERO_CHAIN_HASH - except (FileNotFoundError, json.JSONDecodeError): + raw = read_protected_file( + CHAIN_STATE_PATH, + "Chain state", + MAX_PROTECTED_CONFIG_BYTES, + require_private_permissions=False, + ).decode("utf-8") + except FileNotFoundError: return ZERO_CHAIN_HASH + except Exception as error: + if FAIL_CLOSED: + raise RuntimeError("Failed to read chain state: " + str(error)) from error + log_error(error) + return ZERO_CHAIN_HASH + try: + state = json.loads(raw) + except (json.JSONDecodeError, ValueError) as error: + if FAIL_CLOSED: + raise RuntimeError("Chain state is invalid JSON: " + str(error)) from error + log_error(error) + return ZERO_CHAIN_HASH + previous = state.get("prev_chain_hash") if isinstance(state, dict) else None + if not isinstance(previous, str) or re.fullmatch(r"[A-Za-z0-9_-]{{43}}", previous) is None: + error = ValueError("Chain state contains an invalid prev_chain_hash") + if FAIL_CLOSED: + raise error + log_error(error) + return ZERO_CHAIN_HASH + return previous + def write_chain_state(chain_hash): - os.makedirs(os.path.dirname(CHAIN_STATE_PATH), exist_ok=True) - tmp_path = CHAIN_STATE_PATH + ".tmp" - with open(tmp_path, "w") as f: - json.dump({{"prev_chain_hash": chain_hash}}, f) + token = os.urandom(8).hex() + temporary_path = CHAIN_STATE_PATH + "." + str(os.getpid()) + "." + token + ".tmp" + descriptor = -1 try: - os.replace(tmp_path, CHAIN_STATE_PATH) - except OSError: - import shutil - shutil.move(tmp_path, CHAIN_STATE_PATH) - -def log_error(err): - try: - ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) - msg = f"{{ts}} [elydora-hook] {{err}}\\n" - os.makedirs(os.path.dirname(ERROR_LOG_PATH), exist_ok=True) - with open(ERROR_LOG_PATH, "a") as f: - f.write(msg) - except Exception: - pass - -# --------------------------------------------------------------------------- -# Main hook logic -# --------------------------------------------------------------------------- - -def main(): - try: - raw = sys.stdin.read() - if not raw.strip(): - return - + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0) + descriptor = os.open(temporary_path, flags, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as file: + descriptor = -1 + json.dump({{"prev_chain_hash": chain_hash}}, file) + file.flush() + os.fsync(file.fileno()) + os.replace(temporary_path, CHAIN_STATE_PATH) + finally: + if descriptor >= 0: + os.close(descriptor) try: - event = json.loads(raw) - except (json.JSONDecodeError, ValueError): - return + os.remove(temporary_path) + except FileNotFoundError: + pass - # Extract tool event data - tool_name = "unknown" - tool_input = {{}} - session_id = "unknown" - if isinstance(event, dict): - cline_event = event.get("tool_result") - if not isinstance(cline_event, dict): - cline_event = event.get("tool_call") - if not isinstance(cline_event, dict): - cline_event = {{}} - tool_name = event.get( - "tool_name", - event.get("toolName", event.get("name", cline_event.get("name", "unknown"))), - ) - tool_input = event.get( - "tool_input", - event.get("toolInput", event.get("input", cline_event.get("input", {{}}))), - ) - session_id = event.get( - "session_id", - event.get("sessionId", event.get("session", event.get("taskId", "unknown"))), - ) - - # Read chain state - prev_chain_hash = read_chain_state() - - # Build the EOR - operation_id = generate_uuidv7() - issued_at = int(time.time() * 1000) - nonce = generate_nonce() - - payload = {{ - "tool_name": tool_name, - "tool_input": tool_input, - "session_id": session_id, - }} - - payload_hash = compute_payload_hash(payload) - chain_hash = compute_chain_hash(prev_chain_hash, payload_hash, operation_id, issued_at) - - eor_without_sig = {{ - "op_version": "1.0", - "operation_id": operation_id, +def read_runtime_config(): + if not FAIL_CLOSED: + return {{ "org_id": ORG_ID, "agent_id": AGENT_ID, - "issued_at": issued_at, - "ttl_ms": 30000, - "nonce": nonce, - "operation_type": "ai.tool_use", - "subject": {{"session_id": session_id}}, - "action": {{"tool": tool_name}}, - "payload": payload, - "payload_hash": payload_hash, - "prev_chain_hash": prev_chain_hash, - "agent_pubkey_kid": KID, + "agent_name": AGENT_NAME, + "kid": KID, + "base_url": BASE_URL, + "token": "", }} + try: + raw = read_protected_file( + CONFIG_PATH, + "Agent config", + MAX_PROTECTED_CONFIG_BYTES, + ).decode("utf-8") + config = json.loads(raw) + except Exception as error: + raise RuntimeError("Failed to read agent config: " + str(error)) from error + if not isinstance(config, dict): + raise ValueError("Agent config must contain a JSON object") + if config.get("agent_id") != AGENT_ID or config.get("agent_name") != AGENT_NAME: + raise ValueError("Agent config identity does not match the generated audit hook") + for field in ("org_id", "agent_id", "kid"): + if not isinstance(config.get(field), str) or not config[field]: + raise ValueError("Agent config is missing " + field) + base_url = config.get("base_url", "https://api.elydora.com") + if not isinstance(base_url, str): + raise ValueError("Agent config base_url must be a string") + if urllib.parse.urlsplit(base_url).scheme not in ("http", "https"): + raise ValueError("Agent config base_url must use HTTP or HTTPS") + if not isinstance(config.get("token", ""), str): + raise ValueError("Agent config token must be a string") + return config - canonical = jcs_canonicalize(eor_without_sig) - signature = sign_ed25519(read_private_key(), canonical.encode("utf-8")) - eor = dict(eor_without_sig) - eor["chain_hash"] = chain_hash - eor["signature"] = signature +def read_event(): + raw = sys.stdin.read() + if not raw.strip(): + raise ValueError("Hook input is empty") + try: + event = json.loads(raw) + except (json.JSONDecodeError, ValueError) as error: + raise ValueError("Hook input is invalid JSON: " + str(error)) from error + if not isinstance(event, dict): + raise ValueError("Hook input must contain a JSON object") + return event - # POST to /v1/operations (5s timeout) - # Only update local chain state on success to prevent desync - try: - import urllib.request - import urllib.error - url = BASE_URL.rstrip("/") + "/v1/operations" - body = json.dumps(eor).encode("utf-8") - headers = {{"Content-Type": "application/json", "Accept": "application/json"}} - req = urllib.request.Request(url, data=body, headers=headers, method="POST") - resp = urllib.request.urlopen(req, timeout=5) - # Server accepted (2xx) — safe to advance local chain state + +def submit_operation(event, runtime): + nested = event.get("tool_result") + if not isinstance(nested, dict): + nested = event.get("tool_call") + if not isinstance(nested, dict): + nested = {{}} + tool_name = event.get("tool_name", event.get("toolName", event.get("name", nested.get("name", "unknown")))) + tool_input = event.get("tool_input", event.get("toolInput", event.get("input", nested.get("input", {{}})))) + session_id = event.get( + "conversation_id", + event.get("session_id", event.get("sessionId", event.get("session", event.get("taskId", "unknown")))), + ) + previous = read_chain_state() + operation_id = generate_uuidv7() + issued_at = int(time.time() * 1000) + payload = event if NATIVE_PAYLOAD else {{ + "tool_name": tool_name, + "tool_input": tool_input, + "session_id": session_id, + }} + payload_hash = compute_payload_hash(payload) + chain_hash = compute_chain_hash(previous, payload_hash, operation_id, issued_at) + unsigned = {{ + "op_version": "1.0", + "operation_id": operation_id, + "org_id": runtime["org_id"], + "agent_id": runtime["agent_id"], + "issued_at": issued_at, + "ttl_ms": 30000, + "nonce": base64url_encode(os.urandom(16)), + "operation_type": "ai.tool_use", + "subject": {{"session_id": session_id}}, + "action": {{"tool": tool_name}}, + "payload": payload, + "payload_hash": payload_hash, + "prev_chain_hash": previous, + "agent_pubkey_kid": runtime["kid"], + }} + signature = sign_ed25519( + read_private_key(), + jcs_canonicalize(unsigned).encode("utf-8"), + ) + operation = dict(unsigned) + operation.update({{"chain_hash": chain_hash, "signature": signature}}) + url = runtime.get("base_url", "https://api.elydora.com").rstrip("/") + "/v1/operations" + headers = {{"Content-Type": "application/json", "Accept": "application/json"}} + if runtime.get("token"): + headers["Authorization"] = "Bearer " + runtime["token"] + request = urllib.request.Request( + url, + data=json.dumps(operation).encode("utf-8"), + headers=headers, + method="POST", + ) + try: + with urllib.request.urlopen(request, timeout=5): write_chain_state(chain_hash) - except urllib.error.HTTPError as http_err: - if http_err.code == 400: - # Possible chain hash mismatch — try to resync from server response - try: - import re - err_body = json.loads(http_err.read().decode("utf-8")) - if isinstance(err_body, dict) and isinstance(err_body.get("error"), dict): - if err_body["error"].get("code") == "PREV_HASH_MISMATCH": - match = re.search(r'Expected prev_chain_hash "([^"]+)"', err_body["error"].get("message", "")) - if match: - write_chain_state(match.group(1)) - log_error("Chain hash resynced to server: " + match.group(1)) - except Exception: - pass # ignore parse errors - else: - log_error("API returned HTTP " + str(http_err.code)) - except Exception: - pass # Network error — don't advance chain state, will retry with same prev_chain_hash - - except Exception as e: - log_error(e) - -if __name__ == "__main__": - main() - if SUCCESS_OUTPUT: - sys.stdout.write(SUCCESS_OUTPUT) -''' - - -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: - 1. Reads agent config from ~/.elydora/{agent_id}/config.json - 2. Checks cached status from ~/.elydora/{agent_id}/status-cache.json (60s TTL) - 3. If cache is stale, fetches agent status from GET {base_url}/v1/agents/{agent_id} - 4. If status is "frozen", writes error to stderr and exits with blocking code 2 - 5. Fail-open: if API unreachable or config missing, exits 0 (allow) - 6. Uses only stdlib — no external deps - """ - shebang = sys.executable - return f'''#!{shebang} -"""Elydora agent guard (PreToolUse) for: {agent_name} -Generated by elydora-sdk — DO NOT EDIT.""" - -import json -import os -import sys -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") -CACHE_TTL_S = 60 # 60 seconds - -# Consume stdin so the parent process doesn't block on a full pipe -try: - sys.stdin.read() -except Exception: - pass + except urllib.error.HTTPError as error: + body = error.read().decode("utf-8") + if error.code == 400 and body: + try: + failure = json.loads(body) + except (json.JSONDecodeError, ValueError) as parse_error: + raise RuntimeError("Audit API returned invalid error JSON: " + str(parse_error)) from parse_error + details = failure.get("error", {{}}) if isinstance(failure, dict) else {{}} + if details.get("code") == "PREV_HASH_MISMATCH": + match = re.search( + r'Expected prev_chain_hash "([A-Za-z0-9_-]{{43}})"', + str(details.get("message", "")), + ) + if match: + write_chain_state(match.group(1)) + log_error("Chain hash resynced to server: " + match.group(1)) + raise RuntimeError("Audit API returned HTTP " + str(error.code)) from error def main(): - # Read agent config try: - with open(CONFIG_PATH, "r") as f: - config = json.load(f) - except Exception: - return # Can't read config — fail-open (allow) - - # Check cached status - try: - with open(STATUS_CACHE_PATH, "r") as f: - cache = json.load(f) - age = time.time() - cache.get("cached_at", 0) - if age < CACHE_TTL_S: - if cache.get("status") == "frozen": - sys.stderr.write( - 'Agent "' + AGENT_NAME + '" is frozen by Elydora. Tool execution blocked.\\n' - ) - sys.exit(2) - return # Cache is fresh — use cached result - except Exception: - pass # No cache or invalid — need to check API - - # Fetch agent status from API - base_url = config.get("base_url", "https://api.elydora.com").rstrip("/") - agent_id = config.get("agent_id", "") - - try: - import urllib.request - import urllib.error - - url = base_url + "/v1/agents/" + urllib.request.quote(agent_id, safe="") - req = urllib.request.Request(url, method="GET") - req.add_header("Accept", "application/json") - token = config.get("token", "") - if token: - req.add_header("Authorization", "Bearer " + token) - - resp = urllib.request.urlopen(req, timeout=3) - data = json.loads(resp.read().decode("utf-8")) - agent_status = "active" - if isinstance(data, dict) and isinstance(data.get("agent"), dict): - agent_status = data["agent"].get("status", "active") - - # Update cache - try: - os.makedirs(os.path.dirname(STATUS_CACHE_PATH), exist_ok=True) - with open(STATUS_CACHE_PATH, "w") as f: - json.dump({{"status": agent_status, "cached_at": time.time()}}, f) - except Exception: - pass # Ignore cache write failures - - if agent_status == "frozen": - sys.stderr.write( - 'Agent "' + AGENT_NAME + '" is frozen by Elydora. Tool execution blocked.\\n' - ) - sys.exit(2) - except SystemExit: - raise # Re-raise sys.exit - except Exception: - pass # API unreachable — fail-open (allow) + submit_operation(read_event(), read_runtime_config()) + except Exception as error: + log_error(error) + if FAIL_CLOSED: + raise if __name__ == "__main__": - main() + try: + main() + except Exception as error: + sys.stderr.write("[Elydora audit] " + repr(error) + "\\n") + raise SystemExit(1) if SUCCESS_OUTPUT: sys.stdout.write(SUCCESS_OUTPUT) ''' diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index 5d3ff94..4475cf7 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": "PreToolUse/PostToolUse", + "hook_event": "preToolUse/postToolUse/postToolUseFailure", "config_path": "~/.cursor/hooks.json", }, "droid": { diff --git a/elydora/plugins/runtime_reader_template.py b/elydora/plugins/runtime_reader_template.py new file mode 100644 index 0000000..fb6095d --- /dev/null +++ b/elydora/plugins/runtime_reader_template.py @@ -0,0 +1,44 @@ +"""Protected runtime-file reader embedded in generated hook scripts.""" + +PROTECTED_RUNTIME_READER = r''' +MAX_PROTECTED_SECRET_BYTES = 64 * 1024 +MAX_PROTECTED_CONFIG_BYTES = 512 * 1024 + + +def read_protected_file( + file_path, + label, + max_bytes=MAX_PROTECTED_SECRET_BYTES, + require_private_permissions=True, +): + before = os.lstat(file_path) + if not stat.S_ISREG(before.st_mode) or stat.S_ISLNK(before.st_mode): + raise ValueError(label + " path is not a physical file: " + file_path) + if before.st_size > max_bytes: + raise ValueError(label + " exceeds the size limit: " + file_path) + if require_private_permissions and os.name != "nt" and before.st_mode & (stat.S_IRWXG | stat.S_IRWXO): + raise PermissionError(label + " permissions are too broad: " + file_path) + + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_CLOEXEC", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(file_path, flags) + try: + after = os.fstat(descriptor) + if not stat.S_ISREG(after.st_mode): + raise ValueError(label + " path is not a physical file: " + file_path) + if after.st_size > max_bytes: + raise ValueError(label + " exceeds the size limit: " + file_path) + if require_private_permissions and os.name != "nt" and after.st_mode & (stat.S_IRWXG | stat.S_IRWXO): + raise PermissionError(label + " permissions are too broad: " + file_path) + if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino): + raise OSError(label + " changed while opening: " + file_path) + with os.fdopen(descriptor, "rb") as file: + descriptor = -1 + value = file.read(max_bytes + 1) + finally: + if descriptor >= 0: + os.close(descriptor) + if len(value) > max_bytes: + raise ValueError(label + " exceeds the size limit: " + file_path) + return value +''' diff --git a/tests/cursor_support.py b/tests/cursor_support.py new file mode 100644 index 0000000..bf6e659 --- /dev/null +++ b/tests/cursor_support.py @@ -0,0 +1,198 @@ +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.cursor import CursorPlugin +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@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 settings(self) -> dict[str, Any]: + return json.loads(self.config_path.read_text(encoding="utf-8")) + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status = "active" + get_status = 200 + post_status = 201 + get_paths: list[str] = [] + operations: list[dict[str, Any]] = [] + authorizations: list[str | None] = [] + + @classmethod + def reset(cls) -> None: + cls.agent_status = "active" + cls.get_status = 200 + cls.post_status = 201 + cls.get_paths = [] + cls.operations = [] + cls.authorizations = [] + + 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).get_paths.append(self.path) + type(self).authorizations.append(self.headers.get("Authorization")) + self._respond( + type(self).get_status, + {"agent": {"status": type(self).agent_status}}, + ) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + value = json.loads(self.rfile.read(length)) + assert isinstance(value, dict) + type(self).operations.append(value) + type(self).authorizations.append(self.headers.get("Authorization")) + self._respond(type(self).post_status, {"ok": True}) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server() -> ThreadingHTTPServer: + ElydoraApiHandler.reset() + 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_config: object | None = None, +) -> 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)) + agent_dir.mkdir(parents=True) + 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": VALID_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[str, Any]: + handlers = settings["hooks"][event] + return next(handler for handler in handlers if script in handler.get("command", "")) + + +def assert_native_handler(handler: dict[str, Any]) -> 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, *, 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_cli_security.py b/tests/test_cli_security.py index 27f661a..e404840 100644 --- a/tests/test_cli_security.py +++ b/tests/test_cli_security.py @@ -24,6 +24,7 @@ from elydora.plugins import ( opencode, ) from elydora.plugins.hook_template import generate_hook_script +from elydora.utils import base64url_decode PRIVATE_KEY = base64.urlsafe_b64encode(bytes([7]) * 32).rstrip(b"=").decode() @@ -341,7 +342,7 @@ def test_generated_hook_reads_private_key_from_private_file( exec(compile(script, "hook.py", "exec"), namespace) - assert namespace["read_private_key"]() == PRIVATE_KEY + assert namespace["read_private_key"]() == base64url_decode(PRIVATE_KEY) assert PRIVATE_KEY not in script assert "PRIVATE_KEY_PATH" in script @@ -393,7 +394,8 @@ def test_legacy_plugins_persist_one_owner_only_private_key( runtime_root = case_root / ".elydora" guard_path = runtime_root / "agent-1" / "guard.py" guard_path.parent.mkdir(parents=True) - guard_path.write_text("pass\n", encoding="utf-8") + if module is not cursor: + guard_path.write_text("pass\n", encoding="utf-8") if module is cursor: monkeypatch.setenv("HOME", str(case_root)) monkeypatch.setenv("USERPROFILE", str(case_root)) diff --git a/tests/test_cursor_plugin.py b/tests/test_cursor_plugin.py index a884b4d..2103b6c 100644 --- a/tests/test_cursor_plugin.py +++ b/tests/test_cursor_plugin.py @@ -1,156 +1,45 @@ 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 +from typing import Any import pytest -from elydora.plugins import cursor -from elydora.plugins.cursor import CursorPlugin -from elydora.plugins.hook_template import generate_guard_script +from cursor_support import ( + AGENT_ID, + assert_native_handler, + managed_handler, + prepare_fixture, + write_json, +) from elydora.plugins.registry import SUPPORTED_AGENTS -AGENT_ID = "agent-1" +EVENTS = ( + ("preToolUse", "guard.py"), + ("postToolUse", "hook.py"), + ("postToolUseFailure", "hook.py"), +) -@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: +def test_cursor_registry_declares_complete_native_contract() -> None: assert SUPPORTED_AGENTS["cursor"] == { - "name": "Cursor", "hook_event": "PreToolUse/PostToolUse", + "name": "Cursor", + "hook_event": "preToolUse/postToolUse/postToolUseFailure", "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 = prepare_fixture(monkeypatch, tmp_path) + write_json( fixture.config_path, { "description": "user-owned", @@ -168,92 +57,29 @@ def test_install_preserves_user_hooks_migrates_legacy_and_is_idempotent( fixture.install() fixture.install() - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + settings = fixture.settings() 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( + assert len(settings["hooks"]["postToolUseFailure"]) == 1 + for event, script in EVENTS: + assert_native_handler(managed_handler(settings, event, script)) + + +def test_status_requires_complete_contract_identity_and_physical_runtimes( 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 = 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) + settings = fixture.settings() + settings["hooks"].pop("postToolUseFailure") + write_json(fixture.config_path, settings) assert fixture.plugin.status()["installed"] is False fixture.install() @@ -262,21 +88,21 @@ def test_status_requires_exact_pair_identity_and_physical_runtimes( assert fixture.plugin.status()["installed"] is False fixture.hook_path.write_text(hook_source, encoding="utf-8") - _write_json( + 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") + write_json(fixture.runtime_config_path, "{ malformed") with pytest.raises(ValueError, match="parse Elydora runtime config"): fixture.plugin.status() -def test_uninstall_uses_exact_ownership_and_preserves_user_entries( +def test_uninstall_uses_exact_ownership_for_all_events( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: - fixture = _prepare_fixture( + fixture = prepare_fixture( monkeypatch, tmp_path, existing_config={ @@ -285,31 +111,28 @@ def test_uninstall_uses_exact_ownership_and_preserves_user_entries( }, ) 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 = fixture.settings() + for event, script in EVENTS: + other = deepcopy(managed_handler(settings, event, script)) + other["command"] = other["command"].replace(AGENT_ID, "agent-10") + settings["hooks"][event].append(other) + settings["hooks"]["preToolUse"].append( + { + "command": "echo elydora agent-1 guard.py", + "timeout": 10, + "failClosed": True, + } ) - settings["hooks"]["postToolUse"].append(other_audit) - _write_json(fixture.config_path, settings) + write_json(fixture.config_path, settings) fixture.plugin.uninstall(AGENT_ID) - remaining = json.loads(fixture.config_path.read_text(encoding="utf-8")) + remaining = fixture.settings() 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"] + assert len(remaining["hooks"]["postToolUseFailure"]) == 1 + assert "agent-10" in remaining["hooks"]["postToolUse"][0]["command"] @pytest.mark.parametrize( @@ -331,170 +154,102 @@ def test_invalid_config_is_preserved_before_runtime_writes( tmp_path: Path, existing_config: str, ) -> None: - fixture = _prepare_fixture( + 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 + for path in ( + fixture.guard_path, + fixture.hook_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert path.exists() is False -def test_install_creates_missing_guard_and_rejects_unmanaged_path( +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("org_id", "", "org_id is required"), + ("kid", "", "kid is required"), + ("agent_name", "codex", "requires agent_name cursor"), + ("private_key", "invalid", "canonical 32-byte base64url"), + ("base_url", "relative", "absolute HTTP or HTTPS URL"), + ("token", 42, "token must be a string"), + ], +) +def test_invalid_install_values_fail_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + field: str, + value: Any, + message: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config[field] = value # type: ignore[assignment] + + with pytest.raises(ValueError, match=message): + fixture.install() + + assert fixture.config_path.exists() is False + assert list(fixture.agent_dir.iterdir()) == [] + + +def test_install_creates_all_runtimes_and_rejects_unmanaged_guard( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: - fixture = _prepare_fixture(monkeypatch, tmp_path, create_guard=False) + fixture = prepare_fixture(monkeypatch, tmp_path) fixture.install() - assert fixture.guard_path.is_file() + for path in ( + fixture.guard_path, + fixture.hook_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert path.is_file() - fixture = _prepare_fixture(monkeypatch, tmp_path / "unmanaged", create_guard=False) - unmanaged = fixture.home_dir / "unmanaged-guard.py" + other = prepare_fixture(monkeypatch, tmp_path / "unmanaged") + unmanaged = other.home_dir / "unmanaged-guard.py" unmanaged.write_text("pass\n", encoding="utf-8") - fixture.config["guard_script_path"] = str(unmanaged) + other.config["guard_script_path"] = str(unmanaged) with pytest.raises(ValueError, match="managed agent directory"): - fixture.install() - assert fixture.config_path.exists() is False + other.install() + assert other.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( +def test_uninstall_removes_entirely_managed_config_and_preserves_absence( 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 = 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") + + source = '{"version":1,"hooks":{}}\n' + fixture.config_path.write_text(source, encoding="utf-8") fixture.plugin.uninstall(AGENT_ID) - assert fixture.config_path.read_text(encoding="utf-8") == user_source + assert fixture.config_path.read_text(encoding="utf-8") == source -def test_transaction_rolls_back_runtime_files_when_config_commit_fails( +def test_runtime_files_have_private_modes_on_posix( 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")) + if os.name == "nt": + pytest.skip("POSIX mode contract") + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.guard_path.stat().st_mode & 0o777 == 0o700 + assert fixture.hook_path.stat().st_mode & 0o777 == 0o700 + assert fixture.runtime_config_path.stat().st_mode & 0o777 == 0o600 + assert fixture.private_key_path.stat().st_mode & 0o777 == 0o600 diff --git a/tests/test_cursor_runtime.py b/tests/test_cursor_runtime.py new file mode 100644 index 0000000..8c63032 --- /dev/null +++ b/tests/test_cursor_runtime.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import json +from pathlib import Path +import time + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +import pytest + +from cursor_support import ( + ElydoraApiHandler, + VALID_PRIVATE_KEY, + managed_handler, + prepare_fixture, + run_handler, + server_base_url, + start_api_server, + write_json, + write_text, +) +from elydora.crypto import ( + compute_chain_hash, + compute_payload_hash, + jcs_canonicalize, +) +from elydora.utils import base64url_decode + + +SUCCESS_EVENT = { + "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": "project", + "duration": 42, +} +FAILURE_EVENT = { + "conversation_id": "conversation-1", + "generation_id": "generation-1", + "hook_event_name": "postToolUseFailure", + "tool_name": "Shell", + "tool_input": {"command": "exit 1"}, + "tool_use_id": "call-2", + "cwd": "project", + "error_message": "command failed", + "failure_type": "error", + "duration": 21, + "is_interrupt": False, +} + + +def _payload(value: object) -> bytes: + return (json.dumps(value, separators=(",", ":")) + "\n").encode() + + +def _assert_signed_operation(operation: dict, event: dict) -> None: + assert operation["payload"] == event + assert operation["payload_hash"] == compute_payload_hash(event) + assert operation["subject"] == {"session_id": event["conversation_id"]} + assert operation["action"] == {"tool": event["tool_name"]} + expected_chain = compute_chain_hash( + operation["prev_chain_hash"], + operation["payload_hash"], + operation["operation_id"], + operation["issued_at"], + ) + assert operation["chain_hash"] == expected_chain + unsigned = { + key: value + for key, value in operation.items() + if key not in {"chain_hash", "signature"} + } + public_key = Ed25519PrivateKey.from_private_bytes( + base64url_decode(VALID_PRIVATE_KEY) + ).public_key() + public_key.verify( + base64url_decode(operation["signature"]), + jcs_canonicalize(unsigned).encode(), + ) + + +def test_handlers_enforce_status_and_submit_both_native_events( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + server = start_api_server() + try: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config["base_url"] = server_base_url(server) + fixture.install() + settings = fixture.settings() + guard = managed_handler(settings, "preToolUse", "guard.py") + success = managed_handler(settings, "postToolUse", "hook.py") + failure = managed_handler(settings, "postToolUseFailure", "hook.py") + + guard_result = run_handler( + guard, + _payload({"hook_event_name": "preToolUse"}), + ) + success_result = run_handler(success, _payload(SUCCESS_EVENT)) + failure_result = run_handler(failure, _payload(FAILURE_EVENT)) + finally: + server.shutdown() + server.server_close() + + assert guard_result.returncode == 0 + assert json.loads(guard_result.stdout) == {"permission": "allow"} + assert success_result.returncode == 0 + assert json.loads(success_result.stdout) == {} + assert failure_result.returncode == 0 + assert json.loads(failure_result.stdout) == {} + assert ElydoraApiHandler.get_paths == ["/v1/agents/agent-1"] + assert ElydoraApiHandler.authorizations == [ + "Bearer token-1", + "Bearer token-1", + "Bearer token-1", + ] + assert len(ElydoraApiHandler.operations) == 2 + _assert_signed_operation(ElydoraApiHandler.operations[0], SUCCESS_EVENT) + _assert_signed_operation(ElydoraApiHandler.operations[1], FAILURE_EVENT) + assert ( + ElydoraApiHandler.operations[1]["prev_chain_hash"] + == ElydoraApiHandler.operations[0]["chain_hash"] + ) + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_returns_native_denial_and_exit_two_from_cache( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + write_json( + fixture.agent_dir / "status-cache.json", + {"status": status, "cached_at": time.time()}, + ) + guard = managed_handler(fixture.settings(), "preToolUse", "guard.py") + + result = run_handler(guard, _payload({"hook_event_name": "preToolUse"})) + + denial = json.loads(result.stdout) + assert result.returncode == 2 + assert denial["permission"] == "deny" + assert status in denial["userMessage"] + assert b"Tool execution blocked" in result.stderr + + +@pytest.mark.parametrize( + ("target", "source", "event"), + [ + ("config.json", "{ malformed", "preToolUse"), + ("config.json", "{ malformed", "postToolUse"), + ("private.key", "invalid", "postToolUse"), + ("chain-state.json", "{ malformed", "postToolUse"), + ( + "status-cache.json", + json.dumps({"status": "active", "cached_at": 4_102_444_800}), + "preToolUse", + ), + ], +) +def test_runtime_integrity_failures_exit_one_and_are_observable( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + target: str, + source: str, + event: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + write_text(fixture.agent_dir / target, source) + script = "guard.py" if event == "preToolUse" else "hook.py" + handler = managed_handler(fixture.settings(), event, script) + payload = {"hook_event_name": event} + + result = run_handler(handler, _payload(payload)) + + assert result.returncode == 1 + assert result.stdout == b"" + assert b"Elydora" in result.stderr + if event != "preToolUse": + error_log = fixture.agent_dir / "error.log" + assert error_log.is_file() + assert error_log.read_text(encoding="utf-8") + + +@pytest.mark.parametrize("event", ["preToolUse", "postToolUse"]) +def test_malformed_hook_input_exits_one( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + event: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + script = "guard.py" if event == "preToolUse" else "hook.py" + handler = managed_handler(fixture.settings(), event, script) + + result = run_handler(handler, b"{ malformed") + + assert result.returncode == 1 + assert result.stdout == b"" + assert b"invalid JSON" in result.stderr + + +@pytest.mark.parametrize("event", ["preToolUse", "postToolUse"]) +def test_api_failures_exit_one_and_audit_failure_is_logged( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + event: str, +) -> None: + server = start_api_server() + ElydoraApiHandler.get_status = 500 + ElydoraApiHandler.post_status = 500 + try: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config["base_url"] = server_base_url(server) + fixture.install() + script = "guard.py" if event == "preToolUse" else "hook.py" + handler = managed_handler(fixture.settings(), event, script) + payload = {"hook_event_name": event} + result = run_handler(handler, _payload(payload)) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 1 + assert result.stdout == b"" + assert b"HTTP 500" in result.stderr + if event == "postToolUse": + log = fixture.agent_dir / "error.log" + assert "HTTP 500" in log.read_text(encoding="utf-8") diff --git a/tests/test_cursor_security.py b/tests/test_cursor_security.py new file mode 100644 index 0000000..69790a7 --- /dev/null +++ b/tests/test_cursor_security.py @@ -0,0 +1,250 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from cursor_support import ( + AGENT_ID, + prepare_fixture, + symlink_or_skip, + write_json, + write_text, +) + + +@pytest.mark.parametrize( + "runtime_name", + [ + "guard.py", + "hook.py", + "private.key", + "chain-state.json", + "status-cache.json", + "error.log", + ], +) +def test_orphan_runtime_artifacts_are_rejected( + 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 fixture.config_path.exists() is False + + +def test_mismatched_runtime_identity_is_rejected_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": "cursor"}, + ) + 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 fixture.config_path.exists() is False + + +@pytest.mark.parametrize( + "runtime_name", + [ + "config.json", + "private.key", + "guard.py", + "hook.py", + "chain-state.json", + "status-cache.json", + "error.log", + ], +) +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) + target = fixture.home_dir / f"{runtime_name}.target" + source = ( + json.dumps({"agent_id": AGENT_ID, "agent_name": "cursor"}) + if runtime_name == "config.json" + else "target\n" + ) + write_text(target, source) + runtime_path = fixture.agent_dir / runtime_name + symlink_or_skip(target, runtime_path) + + with pytest.raises(OSError, match="physical file"): + fixture.install() + + assert target.read_text(encoding="utf-8") == source + assert fixture.config_path.exists() is False + + +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' + write_text(target, original) + 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() + + +def test_symbolic_link_config_directory_is_rejected( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + target = fixture.home_dir / "cursor-target" + target.mkdir() + symlink_or_skip(target, fixture.config_path.parent, directory=True) + + with pytest.raises(OSError, match="not a physical directory"): + fixture.install() + + assert list(target.iterdir()) == [] + + +def test_symbolic_link_agent_directory_is_rejected( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.agent_dir.rmdir() + target = fixture.home_dir / "agent-target" + target.mkdir() + symlink_or_skip(target, fixture.agent_dir, directory=True) + + with pytest.raises(OSError, match="not a physical directory"): + fixture.install() + + assert fixture.config_path.exists() is False + assert list(target.iterdir()) == [] + + +def test_symbolic_link_runtime_root_is_rejected( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.agent_dir.rmdir() + fixture.agent_dir.parent.rmdir() + target = fixture.home_dir / "runtime-target" + (target / AGENT_ID).mkdir(parents=True) + symlink_or_skip(target, fixture.agent_dir.parent, directory=True) + + with pytest.raises(OSError, match="not a physical directory"): + fixture.install() + + assert fixture.config_path.exists() is False + assert list((target / AGENT_ID).iterdir()) == [] + + +@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 + target = fixture.home_dir / f"status-{runtime_name}.target" + target.write_bytes(runtime_path.read_bytes()) + runtime_path.unlink() + symlink_or_skip(target, runtime_path) + + with pytest.raises(OSError, match="physical file"): + fixture.plugin.status() + + +def test_transaction_rolls_back_all_files_when_config_commit_fails( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_config={"version": 1, "hooks": {"sessionStart": []}}, + ) + 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 + for path in ( + fixture.guard_path, + fixture.hook_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert path.exists() is False + assert not list(fixture.home_dir.rglob("*.tmp")) + assert not list(fixture.home_dir.rglob("*.rollback")) + + +def test_transaction_detects_concurrent_config_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() + change = rendered_change(render_document(document, {})) + 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")) diff --git a/tests/test_guard_template.py b/tests/test_guard_template.py index 748a0a5..bf10725 100644 --- a/tests/test_guard_template.py +++ b/tests/test_guard_template.py @@ -7,6 +7,7 @@ from pathlib import Path import subprocess import sys from threading import Thread +import time from elydora.plugins.hook_template import generate_guard_script @@ -98,7 +99,7 @@ def test_cached_frozen_status_uses_blocking_exit_code(tmp_path: Path) -> None: script_path, server = create_guard(tmp_path, "active") cache_path = tmp_path / ".elydora" / AGENT_ID / "status-cache.json" cache_path.write_text( - json.dumps({"status": "frozen", "cached_at": 4_102_444_800}), + json.dumps({"status": "frozen", "cached_at": time.time()}), encoding="utf-8", ) try: