diff --git a/sdks/python/README.md b/sdks/python/README.md index 388e2d9..8a39fc4 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -78,7 +78,7 @@ 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. -Codex performs a one-time trust review for user hooks. Run `/hooks` in Codex after installation and trust the Elydora `PreToolUse` and `PostToolUse` definitions. +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/sdks/python/elydora/plugins/_managed_files.py b/sdks/python/elydora/plugins/_managed_files.py new file mode 100644 index 0000000..6b176a7 --- /dev/null +++ b/sdks/python/elydora/plugins/_managed_files.py @@ -0,0 +1,104 @@ +"""Physical file and directory primitives for managed hook state.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import stat +from typing import Optional + + +MAX_SECRET_BYTES = 64 * 1024 +MAX_CONFIG_BYTES = 512 * 1024 +MAX_SOURCE_BYTES = 2 * 1024 * 1024 + + +@dataclass(frozen=True) +class FileSnapshot: + contents: str + device: int + inode: int + mode: int + + +def _inspect_regular_file(file_path: str, label: str) -> Optional[os.stat_result]: + try: + metadata = os.lstat(file_path) + except FileNotFoundError: + return None + except OSError as error: + raise OSError(f"Inspect {label} at {file_path}: {error}") from error + if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"{label} path is not a physical file: {file_path}") + return metadata + + +def read_physical_file( + file_path: str, + label: str, + maximum_bytes: int = MAX_SOURCE_BYTES, +) -> Optional[FileSnapshot]: + before = _inspect_regular_file(file_path, label) + if before is None: + return None + if before.st_size > maximum_bytes: + raise ValueError(f"{label} exceeds {maximum_bytes} bytes: {file_path}") + + flags = os.O_RDONLY | getattr(os, "O_BINARY", 0) | getattr(os, "O_CLOEXEC", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = -1 + try: + descriptor = os.open(file_path, flags) + after = os.fstat(descriptor) + if not stat.S_ISREG(after.st_mode): + raise OSError(f"{label} path is not a physical file: {file_path}") + if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino): + raise OSError(f"{label} changed while opening: {file_path}") + if after.st_size > maximum_bytes: + raise ValueError(f"{label} exceeds {maximum_bytes} bytes: {file_path}") + with os.fdopen(descriptor, "rb") as file: + descriptor = -1 + raw = file.read(maximum_bytes + 1) + except OSError as error: + raise OSError(f"Read {label} at {file_path}: {error}") from error + finally: + if descriptor >= 0: + os.close(descriptor) + + if len(raw) > maximum_bytes: + raise ValueError(f"{label} exceeds {maximum_bytes} bytes: {file_path}") + try: + contents = raw.decode("utf-8") + except UnicodeDecodeError as error: + raise ValueError(f"{label} at {file_path} must contain UTF-8 text") from error + return FileSnapshot( + contents=contents, + device=after.st_dev, + inode=after.st_ino, + mode=stat.S_IMODE(after.st_mode), + ) + + +def physical_file_exists(file_path: str, label: str) -> bool: + return _inspect_regular_file(file_path, label) is not None + + +def physical_directory_exists(directory: str, label: str) -> bool: + try: + metadata = os.lstat(directory) + except FileNotFoundError: + return False + except OSError as error: + raise OSError(f"Inspect {label} at {directory}: {error}") from error + if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): + raise OSError(f"{label} is not a physical directory: {directory}") + return True + + +def ensure_physical_directory(directory: str, label: str) -> None: + try: + os.makedirs(directory, mode=0o700, exist_ok=True) + except OSError as error: + raise OSError(f"Prepare {label} at {directory}: {error}") from error + if not physical_directory_exists(directory, label): + raise OSError(f"{label} is missing: {directory}") diff --git a/sdks/python/elydora/plugins/_strict_json.py b/sdks/python/elydora/plugins/_strict_json.py new file mode 100644 index 0000000..80f1bf7 --- /dev/null +++ b/sdks/python/elydora/plugins/_strict_json.py @@ -0,0 +1,32 @@ +"""Strict JSON parsing with duplicate-key rejection.""" + +from __future__ import annotations + +import json +from typing import Any, Dict, List, Tuple + + +JsonObject = Dict[str, Any] + + +class _DuplicateKey(ValueError): + pass + + +def _unique_object(pairs: List[Tuple[str, Any]]) -> JsonObject: + result: JsonObject = {} + for key, value in pairs: + if key in result: + raise _DuplicateKey(f'duplicate field "{key}"') + result[key] = value + return result + + +def parse_json_object(raw: str, label: str) -> JsonObject: + try: + value = json.loads(raw, object_pairs_hook=_unique_object) + except (_DuplicateKey, json.JSONDecodeError) as error: + raise ValueError(f"Failed to parse {label}: {error}") from error + if not isinstance(value, dict): + raise ValueError(f"{label} must contain a JSON object") + return value diff --git a/sdks/python/elydora/plugins/_transaction.py b/sdks/python/elydora/plugins/_transaction.py index b3a8486..e881f1d 100644 --- a/sdks/python/elydora/plugins/_transaction.py +++ b/sdks/python/elydora/plugins/_transaction.py @@ -4,11 +4,17 @@ from __future__ import annotations from dataclasses import dataclass import os -import stat import tempfile from typing import List, Optional, Sequence from uuid import uuid4 +from ._managed_files import ( + MAX_SOURCE_BYTES, + ensure_physical_directory, + physical_file_exists, + read_physical_file, +) + @dataclass(frozen=True) class FileChange: @@ -18,6 +24,9 @@ class FileChange: next_source: Optional[str] mode: int original_mode: Optional[int] + original_device: Optional[int] + original_inode: Optional[int] + maximum_bytes: int @dataclass @@ -28,23 +37,13 @@ class _StagedChange: committed: bool = False -def read_optional(file_path: str, label: str) -> Optional[str]: - try: - with open(file_path, "r", encoding="utf-8", newline="") as file: - return file.read() - except FileNotFoundError: - return None - except OSError as error: - raise OSError(f"Read {label} at {file_path}: {error}") from error - - -def _existing_mode(file_path: str) -> Optional[int]: - try: - return stat.S_IMODE(os.stat(file_path).st_mode) - except FileNotFoundError: - return None - except OSError as error: - raise OSError(f"Read file mode at {file_path}: {error}") from error +def read_optional( + file_path: str, + label: str, + maximum_bytes: int = MAX_SOURCE_BYTES, +) -> Optional[str]: + snapshot = read_physical_file(file_path, label, maximum_bytes) + return None if snapshot is None else snapshot.contents def source_change( @@ -53,16 +52,24 @@ def source_change( original: Optional[str], next_source: Optional[str], mode: int, + maximum_bytes: int = MAX_SOURCE_BYTES, ) -> Optional[FileChange]: if original == next_source: return None + snapshot = read_physical_file(file_path, label, maximum_bytes) + current = None if snapshot is None else snapshot.contents + if current != original: + raise OSError(f"{label} changed before staging: {file_path}") return FileChange( file_path, label, original, next_source, mode, - _existing_mode(file_path), + None if snapshot is None else snapshot.mode, + None if snapshot is None else snapshot.device, + None if snapshot is None else snapshot.inode, + maximum_bytes, ) @@ -71,13 +78,16 @@ def file_change( label: str, next_source: Optional[str], mode: int, + maximum_bytes: int = MAX_SOURCE_BYTES, ) -> Optional[FileChange]: + snapshot = read_physical_file(file_path, label, maximum_bytes) return source_change( file_path, label, - read_optional(file_path, label), + None if snapshot is None else snapshot.contents, next_source, mode, + maximum_bytes, ) @@ -137,20 +147,24 @@ def _write_staged( def _assert_unchanged(change: FileChange) -> None: - current = read_optional(change.file_path, change.label) - if current != change.original: + snapshot = read_physical_file( + change.file_path, + change.label, + change.maximum_bytes, + ) + current = None if snapshot is None else snapshot.contents + identity_changed = snapshot is not None and ( + snapshot.device != change.original_device + or snapshot.inode != change.original_inode + ) + if current != change.original or identity_changed: raise OSError(f"{change.label} changed during installation: {change.file_path}") def _stage(change: FileChange) -> _StagedChange: _assert_unchanged(change) directory = os.path.dirname(change.file_path) - try: - os.makedirs(directory, mode=0o700, exist_ok=True) - except OSError as error: - raise OSError( - f"Create directory for {change.label} at {directory}: {error}" - ) from error + ensure_physical_directory(directory, f"{change.label} directory") basename = os.path.basename(change.file_path) temporary_path = None rollback_path = None @@ -258,12 +272,7 @@ def write_changes(changes: Sequence[FileChange], label: str) -> None: def regular_file_exists(file_path: str, label: str) -> bool: - try: - return stat.S_ISREG(os.stat(file_path).st_mode) - except FileNotFoundError: - return False - except OSError as error: - raise OSError(f"Read {label} at {file_path}: {error}") from error + return physical_file_exists(file_path, label) def require_runtime(file_path: str, label: str) -> None: diff --git a/sdks/python/elydora/plugins/codex.py b/sdks/python/elydora/plugins/codex.py index 1c6e627..05f1654 100644 --- a/sdks/python/elydora/plugins/codex.py +++ b/sdks/python/elydora/plugins/codex.py @@ -1,415 +1,236 @@ -"""OpenAI Codex lifecycle-hook integration.""" +"""OpenAI Codex native user-hook integration.""" from __future__ import annotations +import base64 import json import os -import shlex -import stat -# subprocess provides argument quoting only; this module starts no process. -import subprocess # nosec B404 -import sys -import tempfile -from typing import Any, Dict, List, Optional +import urllib.parse +from typing import List, Optional +from elydora._runtime_paths import resolve_agent_directory +from elydora.utils import base64url_encode + +from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES +from ._transaction import FileChange from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script +from .codex_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + AUDIT_STATUS, + GUARD_SCRIPT, + GUARD_STATUS, + build_handler, + remove_managed_hooks, + render_document, + runtime_contracts, + runtime_root, + same_path, +) +from .codex_io import ( + read_document, + rendered_change, + runtime_change, + runtime_files_exist, + validate_hooks_directory, + validate_runtime_tree, + write_codex_changes, +) +from .hook_template import generate_guard_script, generate_hook_script -AGENT_KEY = "codex" -OWNED_DESCRIPTION = "Elydora audit and freeze enforcement" -GUARD_STATUS = "Checking Elydora agent state" -AUDIT_STATUS = "Recording Elydora tool use" -GUARD_SCRIPT = "guard.py" -AUDIT_SCRIPT = "hook.py" -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") -CONFIG_PATH = os.path.join(os.path.expanduser("~"), ".codex", "hooks.json") - -JsonObject = Dict[str, Any] - - -def _read_json(path: str, label: str) -> Optional[JsonObject]: - try: - with open(path, "r", encoding="utf-8") as file: - raw = file.read() - except FileNotFoundError: - return None - except OSError as error: - raise OSError(f"Read {label} at {path}: {error}") from error - - try: - value = json.loads(raw) - except json.JSONDecodeError as error: - raise ValueError(f"Failed to parse {label} at {path}: {error}") from error - if not isinstance(value, dict): - raise ValueError(f"{label} at {path} must contain a JSON object") +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 _cleanup_failed_write(path: str, label: str, cause: Exception) -> None: - try: - os.remove(path) - except FileNotFoundError: - return - except OSError as cleanup_error: - raise OSError( - f"Write {label} failed: {cause}; cleanup of {path} failed: {cleanup_error}" - ) from cause +def _json_source(value: dict) -> str: + return json.dumps(value, indent=2) + "\n" -def _write_text_atomic(path: str, content: str, mode: int, label: str) -> None: - directory = os.path.dirname(path) - try: - os.makedirs(directory, mode=0o700, exist_ok=True) - except OSError as error: - raise OSError(f"Create directory for {label} at {directory}: {error}") from error - - descriptor = -1 - temporary_path = "" - try: - descriptor, temporary_path = tempfile.mkstemp( - prefix=f".{os.path.basename(path)}.", - suffix=".tmp", - dir=directory, - text=True, - ) - with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as file: - descriptor = -1 - file.write(content) - file.flush() - os.fsync(file.fileno()) - os.chmod(temporary_path, mode) - os.replace(temporary_path, path) - except Exception as error: - if descriptor >= 0: - try: - os.close(descriptor) - except OSError as close_error: - _cleanup_failed_write(temporary_path, label, close_error) - raise OSError( - f"Write {label} at {path} failed: {error}; " - f"close failed: {close_error}" - ) from error - _cleanup_failed_write(temporary_path, label, error) - raise OSError(f"Write {label} at {path}: {error}") from error +def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] -def _write_json_atomic(path: str, value: JsonObject, mode: int, label: str) -> None: - encoded = json.dumps(value, indent=2) + "\n" - _write_text_atomic(path, encoded, mode, label) - - -def _remove_file(path: str, label: str) -> None: - try: - os.remove(path) - except FileNotFoundError: - return - except OSError as error: - raise OSError(f"Remove {label} at {path}: {error}") from error - - -def _regular_file_exists(path: str, label: str) -> bool: - try: - metadata = os.stat(path) - except FileNotFoundError: - return False - except OSError as error: - raise OSError(f"Read {label} at {path}: {error}") from error - return stat.S_ISREG(metadata.st_mode) - - -def _require_runtime(path: str, label: str) -> None: - if not path: - raise ValueError(f"{label} path is required") - if not _regular_file_exists(path, label): - raise FileNotFoundError(f"{label} is missing: {path}") - - -def _build_handler(script_path: str, status_message: str) -> JsonObject: - return { - "type": "command", - "command": f"{shlex.quote(sys.executable)} {shlex.quote(script_path)}", - "commandWindows": subprocess.list2cmdline([sys.executable, script_path]), - "timeout": 10, - "statusMessage": status_message, - } - - -def _hooks_object(settings: JsonObject) -> JsonObject: - if "hooks" not in settings: - return {} - hooks = settings["hooks"] - if not isinstance(hooks, dict): - raise ValueError('Codex hooks config field "hooks" must be an object') - return dict(hooks) - - -def _event_groups(hooks: JsonObject, event: str) -> List[JsonObject]: - if event not in hooks: - return [] - groups = hooks[event] - if not isinstance(groups, list) or not all( - isinstance(group, dict) for group in groups - ): - raise ValueError( - f'Codex hooks config field "hooks.{event}" must be an array of objects' - ) - return groups - - -def _is_managed_command(command: Any, script_name: str, agent_id: str = "") -> bool: - if not isinstance(command, str): - return False - normalized = os.path.normcase(command) - if ".elydora" not in normalized or script_name.lower() not in normalized.lower(): - return False +def _agent_paths(config: InstallConfig) -> tuple[str, str, str, str]: + agent_id = config.get("agent_id", "") if not agent_id: - return True - expected_path = os.path.normcase( - os.path.join(ELYDORA_DIR, agent_id, script_name) - ) - return expected_path in normalized + 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) + configured_guard = config.get("guard_script_path", "") + if not same_path(configured_guard, guard_path): + raise ValueError( + f"Elydora guard runtime must use the managed agent directory: {guard_path}" + ) + return agent_id, agent_directory, guard_path, audit_path -def _is_elydora_handler(handler: JsonObject, agent_id: str = "") -> bool: - status_message = handler.get("statusMessage") - if status_message == GUARD_STATUS: - script_name = GUARD_SCRIPT - elif status_message == AUDIT_STATUS: - script_name = AUDIT_SCRIPT - else: - return False - return any( - _is_managed_command(handler.get(key), script_name, agent_id) - for key in ("command", "commandWindows") - ) - - -def _without_elydora( - groups: List[JsonObject], - agent_id: str = "", -) -> List[JsonObject]: - filtered_groups: List[JsonObject] = [] - for group in groups: - handlers = group.get("hooks") - if not isinstance(handlers, list) or not all( - isinstance(handler, dict) for handler in handlers - ): - raise ValueError("Codex hook matcher group must contain a hooks array") - filtered = [ - handler - for handler in handlers - if not _is_elydora_handler(handler, agent_id) - ] - if filtered: - filtered_groups.append({**group, "hooks": filtered}) - return filtered_groups - - -def _find_handler( - groups: List[JsonObject], - status_message: str, -) -> Optional[JsonObject]: - for group in groups: - handlers = group.get("hooks") - if not isinstance(handlers, list) or not all( - isinstance(handler, dict) for handler in handlers - ): - raise ValueError("Codex hook matcher group must contain a hooks array") - for handler in handlers: - if ( - handler.get("statusMessage") == status_message - and _is_elydora_handler(handler) - ): - return handler - return None - - -def _command_references(handler: JsonObject, script_path: str) -> bool: - return any( - isinstance(handler.get(key), str) and script_path in handler[key] - for key in ("command", "commandWindows") - ) - - -def _is_owned_settings(settings: JsonObject, hooks: JsonObject) -> bool: - settings_keys = {"description", "hooks"} - hook_keys = {"PreToolUse", "PostToolUse"} - return ( - all(key in settings_keys for key in settings) - and settings.get("description") == OWNED_DESCRIPTION - and all(key in hook_keys for key in hooks) - and all(isinstance(value, list) and not value for value in hooks.values()) - ) - - -def _runtime_scripts_exist(guard: JsonObject, audit: JsonObject) -> bool: +def _validate_private_key(value: str) -> None: try: - with os.scandir(ELYDORA_DIR) as iterator: - entries = sorted(iterator, key=lambda entry: entry.name) - except FileNotFoundError: - return False - except OSError as error: - raise OSError( - f"Read Elydora runtime directory at {ELYDORA_DIR}: {error}" + padded = value + "=" * ((4 - len(value) % 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) != value: + raise ValueError("private_key must be a canonical 32-byte base64url value") - for entry in entries: - try: - is_directory = entry.is_dir(follow_symlinks=False) - except OSError as error: - raise OSError( - f"Read Elydora runtime entry at {entry.path}: {error}" - ) from error - if not is_directory: - continue - guard_path = os.path.join(entry.path, GUARD_SCRIPT) - hook_path = os.path.join(entry.path, AUDIT_SCRIPT) - if not ( - _command_references(guard, guard_path) - and _command_references(audit, hook_path) - ): - continue +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"Codex installation requires agent_name {AGENT_KEY}") + _validate_private_key(config["private_key"]) - config_path = os.path.join(entry.path, "config.json") - config = _read_json(config_path, "Elydora runtime config") - if config is None: - continue - agent_name = config.get("agent_name") - if not isinstance(agent_name, str): - raise ValueError( - f'Elydora runtime config at {config_path} field "agent_name" ' - "must be a string" - ) - if agent_name != AGENT_KEY: - continue - return _regular_file_exists( - guard_path, "Elydora guard runtime" - ) and _regular_file_exists(hook_path, "Elydora audit runtime") - return False + base_url = config["base_url"] + try: + parsed = urllib.parse.urlsplit(base_url) + hostname = parsed.hostname + parsed.port + has_credentials = parsed.username is not None or parsed.password is not None + except ValueError as error: + raise ValueError("base_url must be an absolute HTTP or HTTPS URL") from error + invalid_character = "\\" in base_url or any( + character.isspace() or ord(character) < 32 for character in base_url + ) + valid_origin = ( + parsed.scheme in ("http", "https") + and bool(parsed.netloc) + and hostname is not None + and not invalid_character + ) + if not valid_origin: + raise ValueError("base_url must be an absolute HTTP or HTTPS URL") + if has_credentials or parsed.query or parsed.fragment: + raise ValueError( + "base_url must exclude credentials, query parameters, and fragments" + ) + + 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) class CodexPlugin(AgentPlugin): - """Install Elydora into Codex user lifecycle hooks.""" + """Install Elydora into Codex's native global user hooks.""" + + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + _validate_install_config(config) + document = read_document() + validate_hooks_directory(document.file_path) + agent_id, agent_directory, _, _ = _agent_paths(config) + validate_runtime_tree(runtime_root(), agent_directory, agent_id) def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") + self.preflight_install(config) + document = read_document() + agent_id, agent_directory, guard_path, audit_path = _agent_paths(config) - existing_settings = _read_json(CONFIG_PATH, "Codex hooks config") - settings = ( - existing_settings - if existing_settings is not None - else {"description": OWNED_DESCRIPTION} - ) - hooks = _hooks_object(settings) - pre_tool_use = _without_elydora(_event_groups(hooks, "PreToolUse")) - post_tool_use = _without_elydora(_event_groups(hooks, "PostToolUse")) - - guard_path = config.get("guard_script_path", "") - _require_runtime(guard_path, "Elydora guard runtime") - agent_dir = os.path.join(ELYDORA_DIR, agent_id) - hook_path = os.path.join(agent_dir, AUDIT_SCRIPT) + hooks = remove_managed_hooks(document.hooks) hooks["PreToolUse"] = [ - *pre_tool_use, + *hooks.get("PreToolUse", []), { "matcher": "*", - "hooks": [_build_handler(guard_path, GUARD_STATUS)], + "hooks": [build_handler(guard_path, GUARD_STATUS)], }, ] hooks["PostToolUse"] = [ - *post_tool_use, + *hooks.get("PostToolUse", []), { "matcher": "*", - "hooks": [_build_handler(hook_path, AUDIT_STATUS)], + "hooks": [build_handler(audit_path, AUDIT_STATUS)], }, ] - runtime_config: JsonObject = { - "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": config.get("agent_name", AGENT_KEY) or AGENT_KEY, - } - hook_script = generate_hook_script( + guard_script = generate_guard_script(AGENT_KEY, agent_id) + audit_script = generate_hook_script( org_id=config.get("org_id", ""), agent_id=agent_id, kid=config.get("kid", ""), base_url=config.get("base_url", "https://api.elydora.com"), + native_payload=True, + agent_name=AGENT_KEY, ) - _write_json_atomic( - os.path.join(agent_dir, "config.json"), - runtime_config, - 0o600, - "Elydora runtime config", + changes = _present( + [ + runtime_change( + guard_path, + "Elydora guard runtime", + guard_script, + 0o700, + ), + runtime_change( + os.path.join(agent_directory, "config.json"), + "Elydora runtime config", + _json_source(_runtime_config(config, agent_id)), + 0o600, + MAX_CONFIG_BYTES, + ), + runtime_change( + os.path.join(agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + runtime_change( + audit_path, + "Elydora audit runtime", + audit_script, + 0o700, + ), + rendered_change(render_document(document, hooks)), + ] ) - _write_text_atomic( - os.path.join(agent_dir, "private.key"), - config.get("private_key", ""), - 0o600, - "Elydora private key", - ) - _write_text_atomic( - hook_path, - hook_script, - 0o700, - "Elydora audit runtime", - ) - _write_json_atomic( - CONFIG_PATH, - {**settings, "hooks": hooks}, - 0o600, - "Codex hooks config", - ) - print("Codex: run /hooks to review and trust the Elydora hooks.") + write_codex_changes(changes, "Install Codex hooks") + print(f" Codex hooks: {document.file_path}") + print(" Codex trust: run /hooks and approve both Elydora command hooks.") def uninstall(self, agent_id: str = "") -> None: - settings = _read_json(CONFIG_PATH, "Codex hooks config") - if settings is None: + document = read_document() + hooks = remove_managed_hooks(document.hooks, agent_id) + change = rendered_change(render_document(document, hooks)) + if change is None: return - hooks = _hooks_object(settings) - hooks["PreToolUse"] = _without_elydora( - _event_groups(hooks, "PreToolUse"), - agent_id, - ) - hooks["PostToolUse"] = _without_elydora( - _event_groups(hooks, "PostToolUse"), - agent_id, - ) - if _is_owned_settings(settings, hooks): - _remove_file(CONFIG_PATH, "Codex hooks config") - else: - _write_json_atomic( - CONFIG_PATH, - {**settings, "hooks": hooks}, - 0o600, - "Codex hooks config", - ) + validate_hooks_directory(document.file_path) + write_codex_changes([change], "Uninstall Codex hooks") def status(self) -> PluginStatus: - settings = _read_json(CONFIG_PATH, "Codex hooks config") - guard: Optional[JsonObject] = None - audit: Optional[JsonObject] = None - if settings is not None: - hooks = _hooks_object(settings) - guard = _find_handler(_event_groups(hooks, "PreToolUse"), GUARD_STATUS) - audit = _find_handler(_event_groups(hooks, "PostToolUse"), AUDIT_STATUS) - - if guard is None or audit is None: + document = read_document() + contracts = runtime_contracts(document.hooks) + if not contracts: return PluginStatus( installed=False, agent=AGENT_KEY, details="Not installed", ) - installed = _runtime_scripts_exist(guard, audit) + installed = runtime_files_exist(contracts) details = ( - f"Config: {CONFIG_PATH}" + f"Config: {document.file_path}" if installed - else f"Configured at {CONFIG_PATH}; runtime scripts missing" + else f"Configured at {document.file_path}; runtime files missing" ) return PluginStatus( installed=installed, diff --git a/sdks/python/elydora/plugins/codex_contract.py b/sdks/python/elydora/plugins/codex_contract.py new file mode 100644 index 0000000..1218090 --- /dev/null +++ b/sdks/python/elydora/plugins/codex_contract.py @@ -0,0 +1,448 @@ +"""Codex user-hook contract, exact ownership, and rendering.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +import json +import ntpath +import os +import re +import shlex +import subprocess # nosec B404 - used only for legacy argument rendering +import sys +from typing import Any, Dict, List, Optional, Tuple + +from ._strict_json import JsonObject, parse_json_object + + +AGENT_KEY = "codex" +CONFIG_FILE = "hooks.json" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_SECONDS = 10 +OWNED_DESCRIPTION = "Elydora audit and freeze enforcement" +GUARD_STATUS = "Checking Elydora agent state" +AUDIT_STATUS = "Recording Elydora tool use" + +CodexHooks = Dict[str, List[JsonObject]] + + +@dataclass(frozen=True) +class CodexDocument: + exists: bool + file_path: str + root: JsonObject + hooks: CodexHooks + raw: Optional[str] = None + + +@dataclass(frozen=True) +class RenderedDocument: + document: CodexDocument + changed: bool + next_source: Optional[str] = None + + +@dataclass(frozen=True) +class RuntimeContract: + agent_id: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class _ParsedArgument: + value: str + next_index: int + + +def runtime_root() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def same_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def _same_agent_id(left: str, right: str) -> bool: + return os.path.normcase(left) == os.path.normcase(right) + + +def _quote_posix(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def _quote_powershell(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def _windows_powershell_path() -> str: + configured = os.environ.get("SystemRoot") if os.name == "nt" else None + system_root = ( + configured + if configured + and ntpath.isabs(configured) + and re.search(r'["%\r\n]', configured) is None + else r"C:\Windows" + ) + return ntpath.join( + system_root, + "System32", + "WindowsPowerShell", + "v1.0", + "powershell.exe", + ) + + +def _windows_command(script_path: str) -> str: + source = ( + f"& {_quote_powershell(sys.executable)} " + f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" + ) + encoded = base64.b64encode(source.encode("utf-16le")).decode("ascii") + return ( + f'"{_windows_powershell_path()}" -NoLogo -NoProfile ' + f"-NonInteractive -EncodedCommand {encoded}" + ) + + +def build_handler(script_path: str, status_message: str) -> JsonObject: + return { + "type": "command", + "command": f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}", + "commandWindows": _windows_command(script_path), + "timeout": HOOK_TIMEOUT_SECONDS, + "statusMessage": status_message, + } + + +def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: + if start >= len(command) or command[start] != "'": + return None + apostrophe = "'\"'\"'" + value = "" + index = start + 1 + while index < len(command): + if command.startswith(apostrophe, index): + value += "'" + index += len(apostrophe) + continue + if command[index] == "'": + return _ParsedArgument(value, index + 1) + value += command[index] + index += 1 + return None + + +def _parse_posix_command(command: Any) -> Optional[Tuple[str, str]]: + if not isinstance(command, str): + return None + executable = _read_posix_argument(command, 0) + if ( + executable is None + or command[executable.next_index : executable.next_index + 1] != " " + ): + return None + script = _read_posix_argument(command, executable.next_index + 1) + if script is None or script.next_index != len(command): + return None + return executable.value, script.value + + +def _read_powershell_argument(command: str, start: int) -> Optional[_ParsedArgument]: + if start >= len(command) or command[start] != "'": + return None + value = "" + index = start + 1 + while index < len(command): + if command[index] != "'": + value += command[index] + index += 1 + continue + if index + 1 < len(command) and command[index + 1] == "'": + value += "'" + index += 2 + continue + return _ParsedArgument(value, index + 1) + return None + + +def _parse_powershell_source(source: str) -> Optional[Tuple[str, str]]: + if not source.startswith("& "): + return None + executable = _read_powershell_argument(source, 2) + if ( + executable is None + or source[executable.next_index : executable.next_index + 1] != " " + ): + return None + script = _read_powershell_argument(source, executable.next_index + 1) + if script is None or source[script.next_index :] != "; exit $LASTEXITCODE": + return None + return executable.value, script.value + + +def _parse_windows_command(command: Any) -> Optional[Tuple[str, str]]: + if not isinstance(command, str): + return None + match = re.fullmatch( + r'"([^"\r\n]+)" -NoLogo -NoProfile -NonInteractive ' + r"-EncodedCommand ([A-Za-z0-9+/]+={0,2})", + command, + ) + if ( + match is None + or not ntpath.isabs(match.group(1)) + or ntpath.basename(match.group(1)).lower() != "powershell.exe" + ): + return None + try: + raw = base64.b64decode(match.group(2), validate=True) + if base64.b64encode(raw).decode("ascii") != match.group(2): + return None + source = raw.decode("utf-16le") + except (UnicodeDecodeError, ValueError): + return None + return _parse_powershell_source(source) + + +def _parse_legacy_commands(handler: JsonObject) -> Optional[Tuple[str, str]]: + command = handler.get("command") + command_windows = handler.get("commandWindows") + if not isinstance(command, str) or not isinstance(command_windows, str): + return None + try: + arguments = shlex.split(command, posix=True) + except ValueError: + return None + if len(arguments) != 2: + return None + expected_posix = f"{shlex.quote(arguments[0])} {shlex.quote(arguments[1])}" + if command != expected_posix or command_windows != subprocess.list2cmdline( + arguments + ): + return None + return arguments[0], arguments[1] + + +def _is_python_executable(file_path: str) -> bool: + return ( + re.fullmatch( + r"python(?:[0-9]+(?:\.[0-9]+)*)?(?:\.exe)?", + os.path.basename(file_path), + re.I, + ) + is not None + ) + + +def _managed_script_path( + handler: JsonObject, + status_message: str, +) -> Optional[str]: + if ( + set(handler) + != {"type", "command", "commandWindows", "timeout", "statusMessage"} + or handler.get("type") != "command" + or type(handler.get("timeout")) is not int + or handler.get("timeout") != HOOK_TIMEOUT_SECONDS + or handler.get("statusMessage") != status_message + ): + return None + posix = _parse_posix_command(handler.get("command")) + windows = _parse_windows_command(handler.get("commandWindows")) + if posix is None or windows is None: + legacy = _parse_legacy_commands(handler) + posix = legacy + windows = legacy + if ( + posix is None + or windows is None + or not os.path.isabs(posix[0]) + or not os.path.isabs(posix[1]) + or not _is_python_executable(posix[0]) + or not _is_python_executable(windows[0]) + or not same_path(posix[0], windows[0]) + or not same_path(posix[1], windows[1]) + ): + return None + return posix[1] + + +def _managed_agent_id( + handler: JsonObject, + script_name: str, + status_message: str, +) -> Optional[str]: + script_path = _managed_script_path(handler, status_message) + if script_path is None or os.path.basename(script_path) != script_name: + return None + agent_directory = os.path.dirname(script_path) + if not same_path(os.path.dirname(agent_directory), runtime_root()): + return None + agent_id = os.path.basename(agent_directory) + return agent_id if agent_id not in {"", ".", ".."} else None + + +def _read_hooks(value: Any, label: str) -> CodexHooks: + if value is None or not isinstance(value, dict): + raise ValueError(f'{label} field "hooks" must be an object') + hooks: CodexHooks = {} + for event, groups in value.items(): + if not isinstance(groups, list): + raise ValueError(f'{label} field "hooks.{event}" must be an array') + parsed_groups = [] + for index, group in enumerate(groups): + if not isinstance(group, dict): + raise ValueError( + f"{label} matcher group hooks.{event}[{index}] must be an object" + ) + handlers = group.get("hooks") + if not isinstance(handlers, list) or not all( + isinstance(handler, dict) for handler in handlers + ): + raise ValueError( + f"{label} matcher group hooks.{event}[{index}] must contain a hooks array" + ) + parsed_groups.append(group) + hooks[event] = parsed_groups + return hooks + + +def parse_document(file_path: str, raw: str) -> CodexDocument: + label = f"Codex user hooks at {file_path}" + root = parse_json_object(raw, label) + hooks = _read_hooks(root.get("hooks", {}), label) + return CodexDocument(True, file_path, root, hooks, raw) + + +def create_document(file_path: str) -> CodexDocument: + return CodexDocument(False, file_path, {"description": OWNED_DESCRIPTION}, {}) + + +def _exact_matcher_group(group: JsonObject) -> bool: + return set(group) == {"matcher", "hooks"} and group.get("matcher") == "*" + + +def _remove_from_groups( + groups: List[JsonObject], + script_name: str, + status_message: str, + agent_id: str, +) -> List[JsonObject]: + result = [] + for group in groups: + handlers = [] + for handler in group["hooks"]: + managed_id = _managed_agent_id(handler, script_name, status_message) + remove = managed_id is not None and ( + not agent_id or _same_agent_id(managed_id, agent_id) + ) + if not remove: + handlers.append(handler) + if handlers or not _exact_matcher_group(group): + result.append({**group, "hooks": handlers}) + return result + + +def remove_managed_hooks(hooks: CodexHooks, agent_id: str = "") -> CodexHooks: + 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), + ): + groups = _remove_from_groups( + result.get(event, []), + script_name, + status_message, + agent_id, + ) + if groups: + result[event] = groups + else: + result.pop(event, None) + return result + + +def _entirely_managed(document: CodexDocument) -> bool: + if ( + not document.exists + or document.root.get("description") != OWNED_DESCRIPTION + or not set(document.root).issubset({"description", "hooks"}) + or not document.hooks + ): + return False + count = 0 + for event, groups in document.hooks.items(): + contract = { + "PreToolUse": (GUARD_SCRIPT, GUARD_STATUS), + "PostToolUse": (AUDIT_SCRIPT, AUDIT_STATUS), + }.get(event) + if contract is None or not groups: + return False + for group in groups: + handlers = group["hooks"] + if ( + not _exact_matcher_group(group) + or not handlers + or any( + _managed_agent_id(handler, contract[0], contract[1]) is None + for handler in handlers + ) + ): + return False + count += len(handlers) + return count > 0 + + +def render_document(document: CodexDocument, hooks: CodexHooks) -> RenderedDocument: + if not document.exists and not hooks: + return RenderedDocument(document, False) + if hooks == document.hooks: + return RenderedDocument(document, False) + if not hooks and _entirely_managed(document): + return RenderedDocument(document, True) + root = dict(document.root) + if hooks: + root["hooks"] = hooks + else: + root.pop("hooks", None) + next_source = json.dumps(root, indent=2) + "\n" + return RenderedDocument(document, next_source != document.raw, next_source) + + +def _managed_ids( + groups: List[JsonObject], + script_name: str, + status_message: str, +) -> Dict[str, str]: + result = {} + for group in groups: + if not _exact_matcher_group(group): + continue + for handler in group["hooks"]: + agent_id = _managed_agent_id(handler, script_name, status_message) + if agent_id: + result[os.path.normcase(agent_id)] = agent_id + return result + + +def runtime_contracts(hooks: CodexHooks) -> List[RuntimeContract]: + guards = _managed_ids(hooks.get("PreToolUse", []), GUARD_SCRIPT, GUARD_STATUS) + audits = _managed_ids(hooks.get("PostToolUse", []), AUDIT_SCRIPT, AUDIT_STATUS) + contracts = [] + for key, agent_id in guards.items(): + if key not in audits: + continue + agent_directory = os.path.join(runtime_root(), agent_id) + contracts.append( + RuntimeContract( + agent_id, + os.path.join(agent_directory, GUARD_SCRIPT), + os.path.join(agent_directory, AUDIT_SCRIPT), + ) + ) + return contracts diff --git a/sdks/python/elydora/plugins/codex_io.py b/sdks/python/elydora/plugins/codex_io.py new file mode 100644 index 0000000..6d06e3a --- /dev/null +++ b/sdks/python/elydora/plugins/codex_io.py @@ -0,0 +1,193 @@ +"""Codex hook source and managed runtime I/O.""" + +from __future__ import annotations + +import os +import stat +from typing import Any, List, Optional + +from ._managed_files import ( + MAX_CONFIG_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 .codex_contract import ( + AGENT_KEY, + CONFIG_FILE, + CodexDocument, + RenderedDocument, + RuntimeContract, + create_document, + parse_document, +) + + +def _same_agent_id(value: Any, expected: str) -> bool: + return isinstance(value, str) and os.path.normcase(value) == os.path.normcase( + expected + ) + + +def codex_home_path() -> str: + configured = os.environ.get("CODEX_HOME") + if configured is None or configured == "": + return os.path.join(os.path.expanduser("~"), ".codex") + try: + metadata = os.stat(configured) + except OSError as error: + raise OSError(f"Resolve CODEX_HOME at {configured}: {error}") from error + if not stat.S_ISDIR(metadata.st_mode): + raise OSError(f"CODEX_HOME is not a directory: {configured}") + try: + canonical = os.path.realpath(configured, strict=True) + except OSError as error: + raise OSError(f"Canonicalize CODEX_HOME at {configured}: {error}") from error + if not physical_directory_exists(canonical, "CODEX_HOME"): + raise OSError(f"CODEX_HOME is missing: {canonical}") + return canonical + + +def config_path() -> str: + return os.path.join(codex_home_path(), CONFIG_FILE) + + +def read_document() -> CodexDocument: + file_path = config_path() + snapshot = read_physical_file(file_path, "Codex user hooks", MAX_SOURCE_BYTES) + if snapshot is None: + return create_document(file_path) + return parse_document(file_path, snapshot.contents) + + +def validate_hooks_directory(file_path: str) -> None: + directory = os.path.dirname(file_path) + physical_directory_exists(directory, "Codex hooks directory") + + +def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.file_path, + "Codex user hooks", + rendered.document.raw, + rendered.next_source, + 0o600, + MAX_SOURCE_BYTES, + ) + + +def runtime_change( + file_path: str, + label: str, + next_source: str, + mode: int, + maximum_bytes: int = MAX_SOURCE_BYTES, +) -> Optional[FileChange]: + snapshot = read_physical_file(file_path, label, maximum_bytes) + return source_change( + file_path, + label, + None if snapshot is None else snapshot.contents, + next_source, + mode, + maximum_bytes, + ) + + +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_identity(agent_directory: str, agent_id: str) -> bool: + config_path_value = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path_value) + 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 False + if config.get("agent_name") != AGENT_KEY or not _same_agent_id( + config.get("agent_id"), + agent_id, + ): + raise ValueError( + f"Elydora runtime config identity does not match Codex agent {agent_id}: " + f"{config_path_value}" + ) + return True + + +def validate_runtime_tree( + runtime_root: str, agent_directory: str, agent_id: str +) -> None: + if not physical_directory_exists(runtime_root, "Elydora runtime directory"): + return + if not physical_directory_exists( + agent_directory, + "Elydora agent runtime directory", + ): + return + validate_runtime_identity(agent_directory, agent_id) + + +def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: + for contract in contracts: + agent_directory = os.path.dirname(contract.guard_path) + if not physical_directory_exists( + agent_directory, + "Elydora agent runtime directory", + ): + continue + config = _read_runtime_config(os.path.join(agent_directory, "config.json")) + if ( + config is None + or config.get("agent_name") != AGENT_KEY + or not _same_agent_id(config.get("agent_id"), contract.agent_id) + ): + continue + if all( + ( + physical_file_exists(contract.guard_path, "Elydora guard runtime"), + physical_file_exists(contract.audit_path, "Elydora audit runtime"), + physical_file_exists( + os.path.join(agent_directory, "private.key"), + "Elydora private key", + ), + ) + ): + return True + return False + + +def write_codex_changes(changes: List[FileChange], label: str) -> None: + write_changes(changes, label) diff --git a/sdks/python/elydora/plugins/cursor_contract.py b/sdks/python/elydora/plugins/cursor_contract.py index cd849ed..0f3557d 100644 --- a/sdks/python/elydora/plugins/cursor_contract.py +++ b/sdks/python/elydora/plugins/cursor_contract.py @@ -9,13 +9,14 @@ import re import sys from typing import Any, Dict, List, Optional, Tuple +from ._strict_json import JsonObject, parse_json_object + AGENT_KEY = "cursor" GUARD_SCRIPT = "guard.py" AUDIT_SCRIPT = "hook.py" HOOK_TIMEOUT_SECONDS = 10 -JsonObject = Dict[str, Any] CursorHooks = Dict[str, List[JsonObject]] @@ -48,10 +49,6 @@ class _ParsedArgument: next_index: int -class _DuplicateKey(ValueError): - pass - - def runtime_root() -> str: return os.path.join(os.path.expanduser("~"), ".elydora") @@ -109,7 +106,10 @@ def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: executable = _read_posix_argument(command, 0) - if executable is None or command[executable.next_index:executable.next_index + 1] != " ": + if ( + executable is None + or command[executable.next_index : executable.next_index + 1] != " " + ): return None script = _read_posix_argument(command, executable.next_index + 1) if script is None or script.next_index != len(command): @@ -142,10 +142,13 @@ def _parse_powershell_command(command: str) -> Optional[Tuple[str, str]]: if not command.startswith("& "): return None executable = _read_powershell_argument(command, 2) - if executable is None or command[executable.next_index:executable.next_index + 1] != " ": + if ( + executable is None + or command[executable.next_index : executable.next_index + 1] != " " + ): return None script = _read_powershell_argument(command, executable.next_index + 1) - if script is None or command[script.next_index:] != "; exit $LASTEXITCODE": + if script is None or command[script.next_index :] != "; exit $LASTEXITCODE": return None return executable.value, script.value @@ -196,27 +199,6 @@ def _managed_agent_id( return agent_id if agent_id not in {"", ".", ".."} else None -def _unique_object(pairs: List[Tuple[str, Any]]) -> JsonObject: - result: JsonObject = {} - for key, value in pairs: - if key in result: - raise _DuplicateKey(f'duplicate field "{key}"') - result[key] = value - return result - - -def parse_json_object(raw: str, label: str) -> JsonObject: - try: - value = json.loads(raw, object_pairs_hook=_unique_object) - except _DuplicateKey as error: - raise ValueError(f"Failed to parse {label}: {error}") from error - except json.JSONDecodeError as error: - raise ValueError(f"Failed to parse {label}: {error}") from error - if not isinstance(value, dict): - raise ValueError(f"{label} must contain a JSON object") - return value - - def _read_hooks(value: Any, label: str) -> CursorHooks: if value is None: raise ValueError(f'{label} field "hooks" must be an object') @@ -227,9 +209,7 @@ def _read_hooks(value: Any, label: str) -> CursorHooks: if not isinstance(handlers, list): raise ValueError(f'{label} field "hooks.{event}" must be an array') if not all(isinstance(handler, dict) for handler in handlers): - raise ValueError( - f'{label} field "hooks.{event}" must contain objects' - ) + raise ValueError(f'{label} field "hooks.{event}" must contain objects') hooks[event] = list(handlers) return hooks @@ -344,9 +324,11 @@ def runtime_contracts(hooks: CursorHooks) -> List[RuntimeContract]: if key not in audits or key not in failures: continue agent_directory = os.path.join(runtime_root(), agent_id) - contracts.append(RuntimeContract( - agent_id, - os.path.join(agent_directory, GUARD_SCRIPT), - os.path.join(agent_directory, AUDIT_SCRIPT), - )) + contracts.append( + RuntimeContract( + agent_id, + os.path.join(agent_directory, GUARD_SCRIPT), + os.path.join(agent_directory, AUDIT_SCRIPT), + ) + ) return contracts diff --git a/sdks/python/elydora/plugins/guard_template.py b/sdks/python/elydora/plugins/guard_template.py index 8ae2cf6..13fcb13 100644 --- a/sdks/python/elydora/plugins/guard_template.py +++ b/sdks/python/elydora/plugins/guard_template.py @@ -65,15 +65,29 @@ def read_input(): if not raw.strip(): if FAIL_CLOSED: raise ValueError("Hook input is empty") + sys.stderr.write( + "[Elydora guard] Hook input is empty; " + "enforcement continued in fail-open mode.\\n" + ) 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 + sys.stderr.write( + "[Elydora guard] Hook input is invalid JSON; " + "enforcement continued in fail-open mode: " + str(error) + "\\n" + ) + return + if not isinstance(value, dict): + if FAIL_CLOSED: + raise ValueError("Hook input must contain a JSON object") + sys.stderr.write( + "[Elydora guard] Hook input must contain a JSON object; " + "enforcement continued in fail-open mode.\\n" + ) return - if FAIL_CLOSED and not isinstance(value, dict): - raise ValueError("Hook input must contain a JSON object") def read_config(): @@ -102,7 +116,6 @@ def read_status_cache(): STATUS_CACHE_PATH, "Status cache", MAX_PROTECTED_CONFIG_BYTES, - require_private_permissions=False, ).decode("utf-8") except FileNotFoundError: return None @@ -124,34 +137,39 @@ def read_status_cache(): 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 + write_protected_json( + STATUS_CACHE_PATH, + "Status cache", + {{"status": status, "cached_at": time.time()}}, + ) 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") + try: + parsed = urllib.parse.urlsplit(base_url) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError( + "Agent config base_url must be an absolute HTTP or HTTPS URL" + ) from error + invalid_character = "\\\\" in base_url or any( + character.isspace() or ord(character) < 32 for character in base_url + ) + if ( + parsed.scheme not in ("http", "https") + or not parsed.netloc + or hostname is None + or invalid_character + ): + raise ValueError("Agent config base_url 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( + "Agent config base_url must exclude credentials, query parameters, and fragments" + ) token = config.get("token", "") if not isinstance(token, str): raise ValueError("Agent config token must be a string") diff --git a/sdks/python/elydora/plugins/hook_template.py b/sdks/python/elydora/plugins/hook_template.py index 0727fea..5d6838e 100644 --- a/sdks/python/elydora/plugins/hook_template.py +++ b/sdks/python/elydora/plugins/hook_template.py @@ -39,11 +39,8 @@ import urllib.error import urllib.parse import urllib.request -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} @@ -139,27 +136,16 @@ def generate_uuidv7(): def log_error(error): - descriptor = -1 try: 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) + append_protected_text(ERROR_LOG_PATH, "Error log", message) 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) + sys.stderr.write( + "[Elydora audit] Failed to write error log: " + + str(log_failure) + + "\\n" + ) def read_chain_state(): @@ -168,64 +154,30 @@ def read_chain_state(): 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 + raise RuntimeError("Failed to read chain state: " + str(error)) from error 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 + raise RuntimeError("Chain state is invalid JSON: " + str(error)) from error 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 + raise ValueError("Chain state contains an invalid prev_chain_hash") return previous def write_chain_state(chain_hash): - token = os.urandom(8).hex() - temporary_path = CHAIN_STATE_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({{"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: - os.remove(temporary_path) - except FileNotFoundError: - pass + write_protected_json( + CHAIN_STATE_PATH, + "Chain state", + {{"prev_chain_hash": chain_hash}}, + ) def read_runtime_config(): - if not FAIL_CLOSED: - return {{ - "org_id": ORG_ID, - "agent_id": AGENT_ID, - "agent_name": AGENT_NAME, - "kid": KID, - "base_url": BASE_URL, - "token": "", - }} try: raw = read_protected_file( CONFIG_PATH, @@ -237,7 +189,9 @@ def read_runtime_config(): 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: + if config.get("agent_id") != AGENT_ID: + raise ValueError("Agent config identity does not match the generated audit hook") + if AGENT_NAME and 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]: @@ -245,8 +199,28 @@ def read_runtime_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") - if urllib.parse.urlsplit(base_url).scheme not in ("http", "https"): - raise ValueError("Agent config base_url must use HTTP or HTTPS") + try: + parsed = urllib.parse.urlsplit(base_url) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError( + "Agent config base_url must be an absolute HTTP or HTTPS URL" + ) from error + invalid_character = "\\\\" in base_url or any( + character.isspace() or ord(character) < 32 for character in base_url + ) + if ( + parsed.scheme not in ("http", "https") + or not parsed.netloc + or hostname is None + or invalid_character + ): + raise ValueError("Agent config base_url 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( + "Agent config base_url must exclude credentials, query parameters, and fragments" + ) if not isinstance(config.get("token", ""), str): raise ValueError("Agent config token must be a string") return config diff --git a/sdks/python/elydora/plugins/runtime_reader_template.py b/sdks/python/elydora/plugins/runtime_reader_template.py index fb6095d..5c4d84e 100644 --- a/sdks/python/elydora/plugins/runtime_reader_template.py +++ b/sdks/python/elydora/plugins/runtime_reader_template.py @@ -1,35 +1,40 @@ -"""Protected runtime-file reader embedded in generated hook scripts.""" +"""Protected runtime-file primitives embedded in generated hook scripts.""" -PROTECTED_RUNTIME_READER = r''' +PROTECTED_RUNTIME_READER = r""" MAX_PROTECTED_SECRET_BYTES = 64 * 1024 MAX_PROTECTED_CONFIG_BYTES = 512 * 1024 +MAX_PROTECTED_LOG_BYTES = 2 * 1024 * 1024 + + +def validate_protected_metadata(metadata, file_path, label, max_bytes): + if not stat.S_ISREG(metadata.st_mode): + raise ValueError(label + " path is not a physical file: " + file_path) + if metadata.st_size > max_bytes: + raise ValueError(label + " exceeds the size limit: " + file_path) + if os.name != "nt" and metadata.st_mode & (stat.S_IRWXG | stat.S_IRWXO): + raise PermissionError(label + " permissions are too broad: " + file_path) + + +def inspect_protected_file(file_path, label, max_bytes): + metadata = os.lstat(file_path) + if stat.S_ISLNK(metadata.st_mode): + raise ValueError(label + " path is not a physical file: " + file_path) + validate_protected_metadata(metadata, file_path, label, max_bytes) + return metadata 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) - + before = inspect_protected_file(file_path, label, max_bytes) 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) + validate_protected_metadata(after, file_path, label, max_bytes) 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: @@ -41,4 +46,112 @@ def read_protected_file( if len(value) > max_bytes: raise ValueError(label + " exceeds the size limit: " + file_path) return value -''' + + +def open_protected_append(file_path, label, max_bytes): + for _attempt in range(3): + try: + before = inspect_protected_file(file_path, label, max_bytes) + except FileNotFoundError: + flags = os.O_WRONLY | os.O_APPEND | os.O_CREAT | os.O_EXCL + flags |= getattr(os, "O_BINARY", 0) | getattr(os, "O_CLOEXEC", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(file_path, flags, 0o600) + except FileExistsError: + continue + try: + validate_protected_metadata( + os.fstat(descriptor), + file_path, + label, + max_bytes, + ) + return descriptor + except Exception: + os.close(descriptor) + raise + + flags = os.O_WRONLY | os.O_APPEND + flags |= 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) + validate_protected_metadata(after, file_path, label, max_bytes) + if (before.st_dev, before.st_ino) != (after.st_dev, after.st_ino): + raise OSError(label + " changed while opening: " + file_path) + return descriptor + except Exception: + os.close(descriptor) + raise + raise OSError(label + " changed repeatedly while opening: " + file_path) + + +def append_protected_text( + file_path, + label, + text, + max_bytes=MAX_PROTECTED_LOG_BYTES, +): + encoded = text.encode("utf-8") + descriptor = open_protected_append(file_path, label, max_bytes) + try: + metadata = os.fstat(descriptor) + if metadata.st_size + len(encoded) > max_bytes: + raise ValueError(label + " exceeds the size limit: " + file_path) + offset = 0 + while offset < len(encoded): + written = os.write(descriptor, encoded[offset:]) + if written <= 0: + raise OSError(label + " write made no progress: " + file_path) + offset += written + os.fsync(descriptor) + if os.name != "nt": + os.fchmod(descriptor, 0o600) + finally: + os.close(descriptor) + + +def write_protected_json(file_path, label, value): + encoded = json.dumps(value, separators=(",", ":")).encode("utf-8") + if len(encoded) > MAX_PROTECTED_CONFIG_BYTES: + raise ValueError(label + " exceeds the size limit: " + file_path) + temporary_path = ( + file_path + + "." + + str(os.getpid()) + + "." + + os.urandom(8).hex() + + ".tmp" + ) + descriptor = -1 + try: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + flags |= getattr(os, "O_BINARY", 0) | getattr(os, "O_CLOEXEC", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(temporary_path, flags, 0o600) + with os.fdopen(descriptor, "wb") as file: + descriptor = -1 + file.write(encoded) + file.flush() + os.fsync(file.fileno()) + if os.name != "nt": + os.chmod(temporary_path, 0o600) + try: + inspect_protected_file( + file_path, + label, + MAX_PROTECTED_CONFIG_BYTES, + ) + except FileNotFoundError: + pass + os.replace(temporary_path, file_path) + finally: + if descriptor >= 0: + os.close(descriptor) + try: + os.remove(temporary_path) + except FileNotFoundError: + pass +""" diff --git a/sdks/python/tests/codex_support.py b/sdks/python/tests/codex_support.py new file mode 100644 index 0000000..42718d6 --- /dev/null +++ b/sdks/python/tests/codex_support.py @@ -0,0 +1,233 @@ +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.codex import CodexPlugin +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@dataclass +class CodexFixture: + 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: CodexPlugin + config: dict[str, str] + + def install(self) -> None: + self.plugin.install(self.config) + + def settings(self, path: Path | None = None) -> dict[str, Any]: + return json.loads((path or self.config_path).read_text(encoding="utf-8")) + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + get_status = 200 + post_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls) -> None: + cls.agent_status = "active" + cls.get_status = 200 + cls.post_status = 201 + 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( + type(self).get_status, + {"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).post_status, {"operation": {"accepted": 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 = MISSING, + base_url: str = "http://127.0.0.1:9", +) -> CodexFixture: + home_dir = tmp_path / "home with spaces and 'quote %ELYDORA_HOOK_PATH%" + project_dir = tmp_path / "project with spaces" + config_path = home_dir / ".codex" / "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.setenv("CODEX_HOME", "") + monkeypatch.setenv("ELYDORA_HOOK_PATH", str(tmp_path / "expanded")) + project_dir.mkdir(parents=True) + agent_dir.mkdir(parents=True) + if existing_config is not MISSING: + write_json(config_path, existing_config) + return CodexFixture( + 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=CodexPlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "codex", + "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, + status_message: str, +) -> dict[str, Any]: + for group in settings["hooks"][event]: + for handler in group["hooks"]: + if handler.get("statusMessage") == status_message: + return handler + raise AssertionError(f"handler {status_message!r} not found") + + +def assert_native_handler(handler: Mapping[str, Any], status_message: str) -> None: + assert set(handler) == { + "type", + "command", + "commandWindows", + "timeout", + "statusMessage", + } + assert handler["type"] == "command" + assert handler["timeout"] == 10 + assert handler["statusMessage"] == status_message + assert str(handler["command"]).startswith("'") + assert str(handler["commandWindows"]).startswith('"') + assert " -EncodedCommand " in str(handler["commandWindows"]) + + +def run_handler( + handler: Mapping[str, Any], + payload: bytes, + fixture: CodexFixture, +) -> subprocess.CompletedProcess[bytes]: + command_key = "commandWindows" if os.name == "nt" else "command" + return subprocess.run( + str(handler[command_key]), + shell=True, + input=payload, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env={ + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + "CODEX_HOME": "", + }, + ) + + +def legacy_handler(script_path: Path, status_message: str) -> dict[str, Any]: + import shlex + + arguments = [sys.executable, str(script_path)] + return { + "type": "command", + "command": f"{shlex.quote(arguments[0])} {shlex.quote(arguments[1])}", + "commandWindows": subprocess.list2cmdline(arguments), + "timeout": 10, + "statusMessage": status_message, + } + + +def symlink_or_skip(target: Path, link: Path, *, directory: bool = False) -> None: + try: + link.symlink_to(target, target_is_directory=directory) + except OSError as error: + pytest.skip(f"Symbolic links unavailable: {error}") diff --git a/sdks/python/tests/test_codex_plugin.py b/sdks/python/tests/test_codex_plugin.py index 71c3493..a8664d4 100644 --- a/sdks/python/tests/test_codex_plugin.py +++ b/sdks/python/tests/test_codex_plugin.py @@ -1,125 +1,31 @@ from __future__ import annotations -from dataclasses import dataclass import json import os from pathlib import Path -import subprocess -from typing import Any import pytest -from elydora.plugins import codex -from elydora.plugins.base import InstallConfig +from codex_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + assert_native_handler, + legacy_handler, + managed_handler, + prepare_fixture, + symlink_or_skip, + write_json, + write_text, +) +from elydora.plugins.codex_contract import ( + AUDIT_STATUS, + GUARD_STATUS, + build_handler, +) from elydora.plugins.registry import SUPPORTED_AGENTS -AGENT_ID = "agent-1" -MISSING = object() -GUARD_STATUS = "Checking Elydora agent state" -AUDIT_STATUS = "Recording Elydora tool use" - - -@dataclass(frozen=True) -class CodexFixture: - plugin: codex.CodexPlugin - config: InstallConfig - home_dir: Path - agent_dir: Path - config_path: Path - guard_path: Path - hook_path: Path - runtime_config_path: Path - private_key_path: Path - - -def write_json_or_text(path: Path, value: Any) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - content = value if isinstance(value, str) else json.dumps(value, indent=2) - path.write_text(content, encoding="utf-8") - - -def prepare_fixture( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - *, - existing_settings: Any = MISSING, - create_guard: bool = True, -) -> CodexFixture: - home_dir = tmp_path / "home with spaces" - elydora_dir = home_dir / ".elydora" - agent_dir = elydora_dir / AGENT_ID - config_path = home_dir / ".codex" / "hooks.json" - agent_dir.mkdir(parents=True) - - guard_path = agent_dir / "guard.py" - if create_guard: - guard_path.write_text( - "import sys\nsys.stderr.write('Agent is frozen by Elydora.')\n" - "raise SystemExit(2)\n", - encoding="utf-8", - ) - if existing_settings is not MISSING: - write_json_or_text(config_path, existing_settings) - - monkeypatch.setattr(codex, "ELYDORA_DIR", str(elydora_dir)) - monkeypatch.setattr(codex, "CONFIG_PATH", str(config_path)) - config: InstallConfig = { - "agent_id": AGENT_ID, - "agent_name": "codex", - "org_id": "org-1", - "private_key": "test-key", - "kid": "kid-1", - "base_url": "https://api.elydora.test", - "guard_script_path": str(guard_path), - } - return CodexFixture( - plugin=codex.CodexPlugin(), - config=config, - home_dir=home_dir, - agent_dir=agent_dir, - config_path=config_path, - guard_path=guard_path, - hook_path=agent_dir / "hook.py", - runtime_config_path=agent_dir / "config.json", - private_key_path=agent_dir / "private.key", - ) - - -def find_handler( - settings: dict[str, Any], - event: str, - status_message: str, -) -> dict[str, Any]: - for group in settings["hooks"][event]: - for handler in group["hooks"]: - if handler.get("statusMessage") == status_message: - return handler - raise AssertionError(f"handler {status_message!r} not found") - - -def run_command( - command: str, - home_dir: Path, - payload: dict[str, Any], -) -> subprocess.CompletedProcess[str]: - env = { - **os.environ, - "HOME": str(home_dir), - "USERPROFILE": str(home_dir), - } - return subprocess.run( - command, - shell=True, - capture_output=True, - check=False, - env=env, - input=json.dumps(payload), - text=True, - ) - - -def test_codex_registry_points_at_global_hooks_contract() -> None: +def test_codex_registry_points_at_native_user_hooks() -> None: assert SUPPORTED_AGENTS["codex"] == { "name": "OpenAI Codex", "hook_event": "PreToolUse/PostToolUse", @@ -127,285 +33,291 @@ def test_codex_registry_points_at_global_hooks_contract() -> None: } -def test_install_preserves_existing_hooks_and_is_idempotent( +def test_codex_follows_and_canonicalizes_codex_home( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + target = tmp_path / "custom Codex home" + configured = tmp_path / "codex-home-link" + target.mkdir() + try: + configured.symlink_to(target, target_is_directory=True) + except OSError: + configured = target + monkeypatch.setenv("CODEX_HOME", str(configured)) + + fixture.install() + + expected = Path(os.path.realpath(target)) / "hooks.json" + assert expected.is_file() + assert fixture.config_path.exists() is False + status = fixture.plugin.status() + assert status["installed"] is True + assert status["details"] == f"Config: {expected}" + + +@pytest.mark.parametrize("kind", ["missing", "file"]) +def test_codex_rejects_invalid_codex_home_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + configured = tmp_path / f"{kind}-codex-home" + if kind == "file": + write_text(configured, "not a directory") + + monkeypatch.setenv("CODEX_HOME", str(configured)) + with pytest.raises(OSError, match="CODEX_HOME"): + fixture.install() + + assert list(fixture.agent_dir.iterdir()) == [] + assert fixture.config_path.exists() is False + + +def test_install_preserves_sources_migrates_legacy_and_is_idempotent( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: - fixture = prepare_fixture( - monkeypatch, - tmp_path, - existing_settings={ - "description": "Workspace hooks", + fixture = prepare_fixture(monkeypatch, tmp_path) + write_json( + fixture.config_path, + { + "description": "User hooks", + "metadata": {"owner": "user"}, "hooks": { - "SessionStart": [{ - "hooks": [{"type": "command", "command": "existing-command"}], - }], + "SessionStart": [ + {"matcher": "*", "hooks": [{"type": "command", "command": "keep"}]} + ], + "PreToolUse": [ + { + "matcher": "*", + "hooks": [legacy_handler(fixture.guard_path, GUARD_STATUS)], + } + ], + "PostToolUse": [ + { + "matcher": "*", + "hooks": [legacy_handler(fixture.hook_path, AUDIT_STATUS)], + } + ], }, }, ) - fixture.plugin.install(fixture.config) - fixture.plugin.install(fixture.config) + fixture.install() + first = fixture.config_path.read_text(encoding="utf-8") + fixture.install() + second = fixture.config_path.read_text(encoding="utf-8") - assert "run /hooks to review and trust" in capsys.readouterr().out - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - assert settings["description"] == "Workspace hooks" - assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "existing-command" + assert first == second + assert "approve both Elydora command hooks" in capsys.readouterr().out + settings = fixture.settings() + assert settings["description"] == "User hooks" + assert settings["metadata"] == {"owner": "user"} + assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "keep" assert len(settings["hooks"]["PreToolUse"]) == 1 assert len(settings["hooks"]["PostToolUse"]) == 1 - assert settings["hooks"]["PreToolUse"][0]["matcher"] == "*" - guard = find_handler(settings, "PreToolUse", GUARD_STATUS) - assert guard["type"] == "command" - assert guard["timeout"] == 10 - assert isinstance(guard["command"], str) - assert isinstance(guard["commandWindows"], str) + assert_native_handler( + managed_handler(settings, "PreToolUse", GUARD_STATUS), + GUARD_STATUS, + ) + assert_native_handler( + managed_handler(settings, "PostToolUse", AUDIT_STATUS), + AUDIT_STATUS, + ) + runtime = json.loads(fixture.runtime_config_path.read_text(encoding="utf-8")) + assert runtime["agent_name"] == "codex" + assert runtime["token"] == "token-1" + assert fixture.private_key_path.read_text(encoding="utf-8") == VALID_PRIVATE_KEY + assert fixture.guard_path.is_file() + assert fixture.hook_path.is_file() -def test_commands_block_frozen_agents_and_forward_official_payload( +@pytest.mark.parametrize( + ("source", "error_pattern"), + [ + ('{"hooks":{},"hooks":{}}', "duplicate field"), + ('{"hooks":{},}', "parse Codex user hooks"), + ('{"hooks":{/*comment*/}}', "parse Codex user hooks"), + ("[]", "must contain a JSON object"), + ('{"hooks":null}', 'field "hooks" must be an object'), + ('{"hooks":{"PreToolUse":null}}', 'hooks.PreToolUse" must be an array'), + ( + '{"hooks":{"PreToolUse":[{"hooks":null}]}}', + "must contain a hooks array", + ), + ( + '{"hooks":{"PreToolUse":[{"hooks":[null]}]}}', + "must contain a hooks array", + ), + ], +) +def test_install_preserves_malformed_and_ambiguous_json( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + source: str, + error_pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, existing_config=source) + + with pytest.raises(ValueError, match=error_pattern): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == source + assert list(fixture.agent_dir.iterdir()) == [] + + +def test_status_requires_exact_pair_identity_and_runtime_secrets( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - capture_path = tmp_path / "captured-event.json" - fixture.hook_path.write_text( - "import pathlib, sys\n" - f"pathlib.Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n", - encoding="utf-8", - ) - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - command_key = "commandWindows" if os.name == "nt" else "command" - payload = { - "hook_event_name": "PreToolUse", - "session_id": "session-1", - "turn_id": "turn-1", - "transcript_path": None, - "cwd": str(fixture.home_dir), - "model": "gpt-5", - "permission_mode": "default", - "tool_name": "Bash", - "tool_use_id": "call-1", - "tool_input": {"command": "echo test"}, - } + fixture.install() + assert fixture.plugin.status()["installed"] is True - guard = find_handler(settings, "PreToolUse", GUARD_STATUS) - guard_result = run_command(guard[command_key], fixture.home_dir, payload) - assert guard_result.returncode == 2 - assert "Agent is frozen by Elydora" in guard_result.stderr + fixture.private_key_path.unlink() + status = fixture.plugin.status() + assert status["installed"] is False + assert "runtime files missing" in status["details"] - payload["hook_event_name"] = "PostToolUse" - payload["tool_response"] = {"output": "test"} - audit = find_handler(settings, "PostToolUse", AUDIT_STATUS) - audit_result = run_command(audit[command_key], fixture.home_dir, payload) - assert audit_result.returncode == 0 - assert json.loads(capture_path.read_text(encoding="utf-8")) == payload + fixture.install() + settings = fixture.settings() + guard = managed_handler(settings, "PreToolUse", GUARD_STATUS) + guard["extra"] = True + write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + fixture.install() + runtime = json.loads(fixture.runtime_config_path.read_text(encoding="utf-8")) + runtime["agent_id"] = "another-agent" + write_json(fixture.runtime_config_path, runtime) + assert fixture.plugin.status()["installed"] is False -def test_status_requires_both_runtimes_and_uninstall_preserves_other_hooks( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - existing_handler = {"type": "command", "command": "existing-command"} - fixture = prepare_fixture( - monkeypatch, - tmp_path, - existing_settings={ - "hooks": {"PreToolUse": [{"hooks": [existing_handler]}]}, - }, - ) - fixture.plugin.install(fixture.config) - assert fixture.plugin.status() == { - "installed": True, - "agent": "codex", - "details": f"Config: {fixture.config_path}", - } - - fixture.guard_path.unlink() - assert fixture.plugin.status() == { - "installed": False, - "agent": "codex", - "details": f"Configured at {fixture.config_path}; runtime scripts missing", - } - - fixture.plugin.uninstall(AGENT_ID) - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - assert settings["hooks"]["PreToolUse"] == [{"hooks": [existing_handler]}] - assert settings["hooks"]["PostToolUse"] == [] - - -def test_install_preserves_user_handlers_with_matching_status_text( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - user_guard = { - "type": "command", - "command": "user-guard", - "statusMessage": GUARD_STATUS, - } - user_audit = { - "type": "command", - "command": "user-audit", - "statusMessage": AUDIT_STATUS, - } - fixture = prepare_fixture( - monkeypatch, - tmp_path, - existing_settings={ - "hooks": { - "PreToolUse": [{"matcher": "Bash", "hooks": [user_guard]}], - "PostToolUse": [{"matcher": "Bash", "hooks": [user_audit]}], - }, - }, - ) - - fixture.plugin.install(fixture.config) - - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - assert settings["hooks"]["PreToolUse"][0]["hooks"][0] == user_guard - assert settings["hooks"]["PostToolUse"][0]["hooks"][0] == user_audit - assert len(settings["hooks"]["PreToolUse"]) == 2 - assert len(settings["hooks"]["PostToolUse"]) == 2 - - -def test_install_rejects_missing_guard_before_runtime_or_config_writes( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path, create_guard=False) - - with pytest.raises(FileNotFoundError, match="guard runtime is missing"): - fixture.plugin.install(fixture.config) - - assert fixture.config_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 - - -def test_status_surfaces_malformed_referenced_runtime_metadata( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") - + write_text(fixture.runtime_config_path, "{ malformed") with pytest.raises(ValueError, match="parse Elydora runtime config"): fixture.plugin.status() -def test_status_surfaces_malformed_matcher_groups( +def test_uninstall_removes_exact_ownership_and_preserves_lookalikes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - settings["hooks"]["PreToolUse"][0]["hooks"] = None - fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") + fixture.install() + settings = fixture.settings() + agent_ten = fixture.agent_dir.parent / "agent-10" + other_guard = build_handler(str(agent_ten / "guard.py"), GUARD_STATUS) + other_audit = build_handler(str(agent_ten / "hook.py"), AUDIT_STATUS) + lookalike = { + "type": "command", + "command": f"inspect {fixture.guard_path}", + "commandWindows": f"inspect {fixture.guard_path}", + "timeout": 10, + "statusMessage": GUARD_STATUS, + } + managed_guard = dict(managed_handler(settings, "PreToolUse", GUARD_STATUS)) + settings["hooks"]["SessionStart"] = [ + {"matcher": "*", "hooks": [{"type": "command", "command": "keep"}]} + ] + settings["hooks"]["PreToolUse"].extend( + [ + {"matcher": "*", "hooks": [other_guard]}, + {"matcher": "*", "hooks": [lookalike]}, + {"matcher": "Bash", "hooks": [managed_guard]}, + ] + ) + settings["hooks"]["PostToolUse"].append({"matcher": "*", "hooks": [other_audit]}) + write_json(fixture.config_path, settings) - with pytest.raises(ValueError, match="matcher group must contain a hooks array"): - fixture.plugin.status() + fixture.plugin.uninstall(AGENT_ID) + + remaining = fixture.settings() + assert remaining["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "keep" + assert len(remaining["hooks"]["PreToolUse"]) == 3 + assert len(remaining["hooks"]["PostToolUse"]) == 1 + assert remaining["hooks"]["PreToolUse"][1]["hooks"][0] == lookalike + assert remaining["hooks"]["PreToolUse"][2] == {"matcher": "Bash", "hooks": []} -def test_uninstall_removes_config_owned_entirely_by_elydora( +def test_uninstall_removes_an_entirely_managed_file( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) + fixture.install() + fixture.plugin.uninstall(AGENT_ID) fixture.plugin.uninstall(AGENT_ID) assert fixture.config_path.exists() is False -def test_uninstall_matches_the_exact_agent_runtime_directory( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - other_agent_dir = fixture.agent_dir.parent / "agent-10" - - for event, status in ( - ("PreToolUse", GUARD_STATUS), - ("PostToolUse", AUDIT_STATUS), - ): - handler = find_handler(settings, event, status) - other_handler = { - **handler, - "command": handler["command"].replace( - str(fixture.agent_dir), str(other_agent_dir) - ), - "commandWindows": handler["commandWindows"].replace( - str(fixture.agent_dir), str(other_agent_dir) - ), - } - settings["hooks"][event].append( - {"matcher": "*", "hooks": [other_handler]} - ) - fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") - - fixture.plugin.uninstall(AGENT_ID) - - remaining = json.loads(fixture.config_path.read_text(encoding="utf-8")) - assert len(remaining["hooks"]["PreToolUse"]) == 1 - assert len(remaining["hooks"]["PostToolUse"]) == 1 - assert str(other_agent_dir) in find_handler( - remaining, "PreToolUse", GUARD_STATUS - )["command"] - - @pytest.mark.parametrize( - ("existing_settings", "error_pattern"), + "overrides", [ - ({"hooks": None}, 'field "hooks" must be an object'), - ({"hooks": {"PreToolUse": None}}, 'field "hooks.PreToolUse" must be an array'), - ( - {"hooks": {"PreToolUse": [{"hooks": None}]}}, - "matcher group must contain a hooks array", - ), + {"agent_name": "cursor"}, + {"org_id": ""}, + {"agent_id": "../escape"}, + {"private_key": "invalid"}, + {"token": ""}, + {"base_url": "file:///tmp/elydora"}, + {"base_url": "https://user:secret@api.elydora.com"}, + {"base_url": "https://api.elydora.com?tenant=one"}, + {"base_url": "https://api.elydora.com\\evil"}, + {"base_url": "https://api.elydora.com/invalid path"}, + {"base_url": "https://api.elydora.com:invalid"}, + {"guard_script_path": "unmanaged.py"}, ], ) -def test_install_preserves_invalid_shapes_before_runtime_writes( +def test_install_validates_identity_credentials_and_api_origin( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, - existing_settings: dict[str, Any], - error_pattern: str, + overrides: dict[str, str], ) -> None: - fixture = prepare_fixture( - monkeypatch, - tmp_path, - existing_settings=existing_settings, - ) - original = fixture.config_path.read_text(encoding="utf-8") + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config.update(overrides) - with pytest.raises(ValueError, match=error_pattern): - fixture.plugin.install(fixture.config) + with pytest.raises(ValueError): + fixture.install() - assert fixture.config_path.read_text(encoding="utf-8") == original - assert fixture.hook_path.exists() is False - assert fixture.runtime_config_path.exists() is False - assert fixture.private_key_path.exists() is False + assert fixture.config_path.exists() is False + assert list(fixture.agent_dir.iterdir()) == [] -def test_install_preserves_malformed_config_before_runtime_writes( +@pytest.mark.parametrize("kind", ["hooks", "config", "key", "guard", "audit"]) +def test_linked_hook_and_runtime_files_are_rejected( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + kind: str, ) -> None: - fixture = prepare_fixture( - monkeypatch, - tmp_path, - existing_settings="{ malformed", - ) + fixture = prepare_fixture(monkeypatch, tmp_path) + if kind == "hooks": + target = tmp_path / "hooks-target.json" + original = '{"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 + return - with pytest.raises(ValueError, match="parse Codex hooks config"): - fixture.plugin.install(fixture.config) + fixture.install() + file_path = { + "config": fixture.runtime_config_path, + "key": fixture.private_key_path, + "guard": fixture.guard_path, + "audit": fixture.hook_path, + }[kind] + target = tmp_path / f"{kind}-target" + target.write_bytes(file_path.read_bytes()) + file_path.unlink() + symlink_or_skip(target, file_path) - assert fixture.config_path.read_text(encoding="utf-8") == "{ malformed" - assert fixture.hook_path.exists() is False - assert fixture.runtime_config_path.exists() is False - assert fixture.private_key_path.exists() is False + with pytest.raises(OSError, match="physical file"): + fixture.plugin.status() + + assert target.read_bytes() diff --git a/sdks/python/tests/test_codex_runtime.py b/sdks/python/tests/test_codex_runtime.py new file mode 100644 index 0000000..d02d8e1 --- /dev/null +++ b/sdks/python/tests/test_codex_runtime.py @@ -0,0 +1,330 @@ +from __future__ import annotations + +import json +from pathlib import Path +import time + +import pytest + +from codex_support import ( + ElydoraApiHandler, + managed_handler, + prepare_fixture, + run_handler, + server_base_url, + start_api_server, + symlink_or_skip, + write_json, + write_text, +) +from elydora.plugins.codex_contract import AUDIT_STATUS, GUARD_STATUS + + +def payload(event: str, *, failed: bool = False) -> dict[str, object]: + value: dict[str, object] = { + "hook_event_name": event, + "session_id": "session-1", + "turn_id": "turn-1", + "transcript_path": None, + "cwd": "C:/workspace", + "model": "gpt-5", + "permission_mode": "default", + "tool_name": "Bash", + "tool_use_id": "call-1", + "tool_input": {"command": "echo test"}, + } + if event == "PostToolUse": + value["tool_response"] = ( + {"success": False, "error": "command failed", "exit_code": 1} + if failed + else {"success": True, "output": "test"} + ) + return value + + +def encoded(value: object) -> bytes: + return json.dumps(value, separators=(",", ":")).encode() + + +def test_runtimes_enforce_active_state_and_preserve_native_event_json( + 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", GUARD_STATUS) + audit = managed_handler(settings, "PostToolUse", AUDIT_STATUS) + + pre = payload("PreToolUse") + first = payload("PostToolUse") + failed = payload("PostToolUse", failed=True) + guard_result = run_handler(guard, encoded(pre), fixture) + first_result = run_handler(audit, encoded(first), fixture) + failed_result = run_handler(audit, encoded(failed), fixture) + finally: + server.shutdown() + server.server_close() + + assert guard_result.returncode == 0 + assert guard_result.stdout == b"" + assert guard_result.stderr == b"" + assert first_result.returncode == 0 + assert failed_result.returncode == 0 + requests = ElydoraApiHandler.requests + assert [request["method"] for request in requests] == ["GET", "POST", "POST"] + assert [request["authorization"] for request in requests] == [ + "Bearer token-1", + "Bearer token-1", + "Bearer token-1", + ] + assert requests[1]["json"]["payload"] == first + assert requests[2]["json"]["payload"] == failed + assert requests[2]["json"]["prev_chain_hash"] == requests[1]["json"]["chain_hash"] + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_propagates_exit_two_for_blocked_agents( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, +) -> None: + server = start_api_server() + ElydoraApiHandler.agent_status = status + try: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + base_url=server_base_url(server), + ) + fixture.install() + guard = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS) + result = run_handler(guard, encoded(payload("PreToolUse")), fixture) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 2 + assert status.encode() in result.stderr + assert b"Tool execution blocked" in result.stderr + + +def test_guard_uses_a_protected_cached_blocking_status( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + write_json( + fixture.agent_dir / "status-cache.json", + {"status": "frozen", "cached_at": time.time()}, + ) + guard = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS) + + result = run_handler(guard, encoded(payload("PreToolUse")), fixture) + + assert result.returncode == 2 + assert b"Tool execution blocked" in result.stderr + + +def test_fail_open_guard_reports_input_config_and_status_failures( + 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() + guard = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS) + runtime_config = json.loads( + fixture.runtime_config_path.read_text(encoding="utf-8") + ) + + malformed_input = run_handler(guard, b"{ malformed", fixture) + non_object_input = run_handler(guard, b"[]", fixture) + write_text(fixture.runtime_config_path, "{ malformed") + malformed_config = run_handler( + guard, + encoded(payload("PreToolUse")), + fixture, + ) + write_json(fixture.runtime_config_path, runtime_config) + (fixture.agent_dir / "status-cache.json").unlink() + ElydoraApiHandler.agent_status = "unknown" + invalid_status = run_handler( + guard, + encoded(payload("PreToolUse")), + fixture, + ) + finally: + server.shutdown() + server.server_close() + + assert malformed_input.returncode == 0 + assert b"invalid JSON" in malformed_input.stderr + assert b"fail-open" in malformed_input.stderr + assert non_object_input.returncode == 0 + assert b"JSON object" in non_object_input.stderr + assert b"fail-open" in non_object_input.stderr + assert malformed_config.returncode == 0 + assert b"Failed to read agent config" in malformed_config.stderr + assert invalid_status.returncode == 0 + assert b"invalid agent status" in invalid_status.stderr + + +def test_generated_hooks_report_an_unsafe_runtime_origin( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + config = json.loads(fixture.runtime_config_path.read_text(encoding="utf-8")) + config["base_url"] = "https://api.elydora.com\\evil" + write_json(fixture.runtime_config_path, config) + (fixture.agent_dir / "status-cache.json").unlink(missing_ok=True) + settings = fixture.settings() + + guard_result = run_handler( + managed_handler(settings, "PreToolUse", GUARD_STATUS), + encoded(payload("PreToolUse")), + fixture, + ) + audit_result = run_handler( + managed_handler(settings, "PostToolUse", AUDIT_STATUS), + encoded(payload("PostToolUse")), + fixture, + ) + + assert guard_result.returncode == 0 + assert b"absolute HTTP or HTTPS URL" in guard_result.stderr + assert audit_result.returncode == 0 + assert "absolute HTTP or HTTPS URL" in (fixture.agent_dir / "error.log").read_text( + encoding="utf-8" + ) + + +@pytest.mark.parametrize( + ("target", "source", "message"), + [ + ("input", "{ malformed", "invalid JSON"), + ("config.json", "{ malformed", "agent config"), + ("private.key", "invalid", "Private key"), + ("chain-state.json", "{ malformed", "Chain state"), + ], +) +def test_fail_open_audit_records_runtime_failures( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + target: str, + source: str, + message: str, +) -> None: + server = start_api_server() + try: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + base_url=server_base_url(server), + ) + fixture.install() + audit = managed_handler(fixture.settings(), "PostToolUse", AUDIT_STATUS) + if target != "input": + write_text(fixture.agent_dir / target, source) + runtime_input = ( + source.encode() if target == "input" else encoded(payload("PostToolUse")) + ) + result = run_handler(audit, runtime_input, fixture) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0 + error_log = fixture.agent_dir / "error.log" + assert error_log.is_file() + assert message.lower() in error_log.read_text(encoding="utf-8").lower() + + +def test_fail_open_audit_records_api_failures( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + server = start_api_server() + ElydoraApiHandler.post_status = 500 + try: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + base_url=server_base_url(server), + ) + fixture.install() + audit = managed_handler(fixture.settings(), "PostToolUse", AUDIT_STATUS) + result = run_handler(audit, encoded(payload("PostToolUse")), fixture) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0 + assert "HTTP 500" in (fixture.agent_dir / "error.log").read_text(encoding="utf-8") + + +@pytest.mark.parametrize( + "target", ["status-cache.json", "chain-state.json", "error.log"] +) +def test_runtimes_reject_linked_cache_chain_and_error_state( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + target: str, +) -> None: + server = start_api_server() + try: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + base_url=server_base_url(server), + ) + fixture.install() + target_path = tmp_path / f"{target}.target" + source = { + "status-cache.json": json.dumps( + {"status": "active", "cached_at": time.time()} + ), + "chain-state.json": json.dumps({"prev_chain_hash": "A" * 43}), + "error.log": "preserve\n", + }[target] + write_text(target_path, source) + link = fixture.agent_dir / target + symlink_or_skip(target_path, link) + if target == "status-cache.json": + handler = managed_handler(fixture.settings(), "PreToolUse", GUARD_STATUS) + runtime_input = encoded(payload("PreToolUse")) + else: + handler = managed_handler(fixture.settings(), "PostToolUse", AUDIT_STATUS) + runtime_input = ( + b"{ malformed" + if target == "error.log" + else encoded(payload("PostToolUse")) + ) + result = run_handler(handler, runtime_input, fixture) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0 + assert target_path.read_text(encoding="utf-8") == source + if target == "error.log": + assert b"Failed to write error log" in result.stderr + else: + assert ( + b"physical file" in result.stderr + or (fixture.agent_dir / "error.log").is_file() + ) diff --git a/sdks/python/tests/test_codex_transaction.py b/sdks/python/tests/test_codex_transaction.py new file mode 100644 index 0000000..e83703d --- /dev/null +++ b/sdks/python/tests/test_codex_transaction.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 +import sys + +import pytest + +from codex_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + prepare_fixture, + symlink_or_skip, + write_json, + write_text, +) +from elydora.plugins import codex + + +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_when_hooks_commit_fails( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_config={"hooks": {"SessionStart": []}}, + ) + original = fixture.config_path.read_text(encoding="utf-8") + real_replace = os.replace + failed = False + + def fail_hooks_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 hooks commit failure") + real_replace(source, destination) + + monkeypatch.setattr(os, "replace", fail_hooks_once) + + with pytest.raises(OSError, match="injected hooks 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_no_transaction_files(fixture.home_dir) + + +def test_install_detects_a_concurrent_hooks_change_before_staging( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_config={"hooks": {"SessionStart": []}}, + ) + concurrent = '{"hooks":{"SessionStart":[{"hooks":[]}]}}\n' + real_rendered_change = codex.rendered_change + + def mutate_before_change(rendered: object): + write_text(fixture.config_path, concurrent) + return real_rendered_change(rendered) # type: ignore[arg-type] + + monkeypatch.setattr(codex, "rendered_change", mutate_before_change) + + with pytest.raises(OSError, match="changed before staging"): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == concurrent + assert list(fixture.agent_dir.iterdir()) == [] + assert_no_transaction_files(fixture.home_dir) + + +@pytest.mark.parametrize( + "runtime_name", + [ + "private.key", + "guard.py", + "hook.py", + "chain-state.json", + "status-cache.json", + "error.log", + ], +) +def test_orphan_runtime_artifacts_are_rejected_before_hooks_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 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": "codex"}, + ) + 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("kind", ["codex", "runtime"]) +def test_linked_hooks_and_runtime_directories_are_rejected( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + target = tmp_path / f"{kind}-target" + target.mkdir() + if kind == "codex": + link = fixture.config_path.parent + else: + link = fixture.agent_dir.parent + shutil.rmtree(link) + link.parent.mkdir(parents=True, exist_ok=True) + symlink_or_skip(target, link, directory=True) + + with pytest.raises(OSError, match="physical directory"): + fixture.install() + + assert list(target.iterdir()) == [] + assert fixture.config_path.exists() is False + + +def test_uninstall_rejects_a_linked_default_hooks_directory( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + codex_directory = fixture.config_path.parent + target = tmp_path / "codex-hooks-target" + codex_directory.rename(target) + symlink_or_skip(target, codex_directory, directory=True) + + with pytest.raises(OSError, match="physical directory"): + fixture.plugin.uninstall(AGENT_ID) + + assert (target / "hooks.json").is_file() + + +def test_install_protects_credentials_and_cleans_transaction_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + + assert_no_transaction_files(fixture.home_dir) + if os.name != "nt": + for path in ( + fixture.config_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert path.stat().st_mode & 0o777 == 0o600 + + +def test_cli_preflight_preserves_malformed_hooks_before_runtime_creation( + tmp_path: Path, +) -> None: + home_dir = tmp_path / "cli-home" + hooks_path = home_dir / ".codex" / "hooks.json" + key_path = tmp_path / "private.key" + write_text(hooks_path, "{ malformed") + write_text(key_path, VALID_PRIVATE_KEY) + result = subprocess.run( + [ + sys.executable, + "-m", + "elydora.cli", + "install", + "--agent", + "codex", + "--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={ + **os.environ, + "HOME": str(home_dir), + "USERPROFILE": str(home_dir), + "CODEX_HOME": "", + }, + text=True, + ) + + assert result.returncode == 1 + assert "parse Codex user hooks" in result.stderr + assert hooks_path.read_text(encoding="utf-8") == "{ malformed" + assert (home_dir / ".elydora").exists() is False diff --git a/sdks/python/tests/test_guard_template.py b/sdks/python/tests/test_guard_template.py index bf10725..e01f4bf 100644 --- a/sdks/python/tests/test_guard_template.py +++ b/sdks/python/tests/test_guard_template.py @@ -44,7 +44,9 @@ def create_guard(tmp_path: Path, status: str) -> tuple[Path, ThreadingHTTPServer "agent_id": AGENT_ID, "base_url": f"http://127.0.0.1:{server.server_port}", } - (agent_dir / "config.json").write_text(json.dumps(config), encoding="utf-8") + config_path = agent_dir / "config.json" + config_path.write_text(json.dumps(config), encoding="utf-8") + os.chmod(config_path, 0o600) script_path = tmp_path / "guard.py" script_path.write_text( @@ -54,7 +56,11 @@ def create_guard(tmp_path: Path, status: str) -> tuple[Path, ThreadingHTTPServer return script_path, server -def run_guard(script_path: Path, home_dir: Path) -> subprocess.CompletedProcess[str]: +def run_guard( + script_path: Path, + home_dir: Path, + payload: str = "{}", +) -> subprocess.CompletedProcess[str]: env = { **os.environ, "HOME": str(home_dir), @@ -65,7 +71,7 @@ def run_guard(script_path: Path, home_dir: Path) -> subprocess.CompletedProcess[ capture_output=True, check=False, env=env, - input="", + input=payload, text=True, ) @@ -102,6 +108,7 @@ def test_cached_frozen_status_uses_blocking_exit_code(tmp_path: Path) -> None: json.dumps({"status": "frozen", "cached_at": time.time()}), encoding="utf-8", ) + os.chmod(cache_path, 0o600) try: result = run_guard(script_path, tmp_path) finally: @@ -110,3 +117,29 @@ def test_cached_frozen_status_uses_blocking_exit_code(tmp_path: Path) -> None: assert result.returncode == 2 assert "Tool execution blocked" in result.stderr + + +def test_fail_open_guard_reports_malformed_input(tmp_path: Path) -> None: + script_path, server = create_guard(tmp_path, "active") + try: + result = run_guard(script_path, tmp_path, "{ malformed") + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0 + assert "invalid JSON" in result.stderr + assert "fail-open" in result.stderr + + +def test_fail_open_guard_reports_non_object_input(tmp_path: Path) -> None: + script_path, server = create_guard(tmp_path, "active") + try: + result = run_guard(script_path, tmp_path, "[]") + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0 + assert "JSON object" in result.stderr + assert "fail-open" in result.stderr