From 2f223dca7319359a5d336ffdd79af2a541cba400 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Jul 2026 08:19:11 -0400 Subject: [PATCH] fix(python): mirror official Gemini hooks --- sdks/python/README.md | 2 + sdks/python/elydora/plugins/gemini.py | 267 ++++--------- sdks/python/elydora/plugins/gemini_command.py | 206 ++++++++++ sdks/python/elydora/plugins/gemini_config.py | 181 +++++++++ .../python/elydora/plugins/gemini_contract.py | 344 +++++++++++++++++ .../elydora/plugins/gemini_installation.py | 161 ++++++++ sdks/python/elydora/plugins/gemini_io.py | 250 ++++++++++++ sdks/python/elydora/plugins/registry.py | 4 +- sdks/python/tests/gemini_support.py | 279 ++++++++++++++ sdks/python/tests/test_cli_security.py | 2 - sdks/python/tests/test_gemini_official.py | 64 +++ sdks/python/tests/test_gemini_plugin.py | 363 ++++++++++++++++++ sdks/python/tests/test_gemini_runtime.py | 165 ++++++++ sdks/python/tests/test_gemini_transaction.py | 254 ++++++++++++ .../tests/test_settings_hook_detection.py | 21 +- 15 files changed, 2359 insertions(+), 204 deletions(-) create mode 100644 sdks/python/elydora/plugins/gemini_command.py create mode 100644 sdks/python/elydora/plugins/gemini_config.py create mode 100644 sdks/python/elydora/plugins/gemini_contract.py create mode 100644 sdks/python/elydora/plugins/gemini_installation.py create mode 100644 sdks/python/elydora/plugins/gemini_io.py create mode 100644 sdks/python/tests/gemini_support.py create mode 100644 sdks/python/tests/test_gemini_official.py create mode 100644 sdks/python/tests/test_gemini_plugin.py create mode 100644 sdks/python/tests/test_gemini_runtime.py create mode 100644 sdks/python/tests/test_gemini_transaction.py diff --git a/sdks/python/README.md b/sdks/python/README.md index b5caeb1..0ebdc66 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -84,6 +84,8 @@ Codex installation writes exact `PreToolUse` and `PostToolUse` command groups to 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. +Gemini CLI installation writes named, matchless `BeforeTool` and `AfterTool` command hooks to `$GEMINI_CLI_HOME/.gemini/settings.json` (`~/.gemini/settings.json` by default). Empty home overrides select the OS home; relative and literal-tilde values retain Gemini CLI's native path behavior. JSON comments and unrelated settings remain intact. Workspace, system-default, system-override, and extension hook sources remain unchanged and continue loading additively. User settings, generated runtimes, runtime config, and private key commit through one rollback-capable transaction. Managed handlers preserve the complete native snake_case payload, emit valid JSON on successful execution, propagate freeze and revocation through exit code `2`, and use encoded PowerShell commands on Windows. `hooksConfig.enabled` and `hooksConfig.disabled` remain authoritative. Run `/hooks list` after installation to inspect effective hooks from every source. + Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Elydora leaves the Documents and workspace hook roots unchanged. The wrappers preserve Cline's official payload and translate a frozen guard into JSON stdout cancellation control. Kimi installation writes exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` rules to every detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the legacy Python CLI's `~/.kimi/config.toml`. A fresh installation targets stable Kimi Code; an existing legacy home activates the migration contract. Empty `KIMI_CODE_HOME` values use the default. The selected TOML files, generated runtimes, runtime config, and private key commit as one rollback-capable transaction. Managed commands preserve Kimi's native snake_case event JSON, propagate freeze and revocation through exit code `2`, and use encoded PowerShell on Windows. Run `/hooks` to inspect the global rules. diff --git a/sdks/python/elydora/plugins/gemini.py b/sdks/python/elydora/plugins/gemini.py index 0922623..0c61eb0 100644 --- a/sdks/python/elydora/plugins/gemini.py +++ b/sdks/python/elydora/plugins/gemini.py @@ -1,206 +1,101 @@ -"""Gemini CLI plugin — merges AfterTool hook into ~/.gemini/settings.json.""" +"""Gemini CLI native global user-hook integration.""" from __future__ import annotations -import json -import os -import sys - -from ._file_io import write_json_atomic, write_text_atomic from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script +from .gemini_config import GeminiDocument, render_gemini_document +from .gemini_contract import ( + AGENT_KEY, + AUDIT_HOOK_NAME, + GUARD_HOOK_NAME, + build_gemini_group, + disabled_managed_gemini_entries, + gemini_runtime_contracts, + managed_gemini_hooks_enabled, +) +from .gemini_installation import ( + commit_gemini_installation, + preflight_gemini_installation, + prepare_gemini_installation, +) +from .gemini_io import ( + gemini_runtime_files_exist, + read_gemini_document, + write_gemini_document, +) -SETTINGS_PATH = os.path.join(os.path.expanduser("~"), ".gemini", "settings.json") -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") +def _require_enabled_hooks(document: GeminiDocument) -> None: + if not document.hook_controls.enabled: + raise ValueError( + "Gemini CLI hooks are disabled by hooksConfig.enabled: " + f"{document.file_path}" + ) + disabled = disabled_managed_gemini_entries(document.hook_controls) + if disabled: + raise ValueError( + "Gemini CLI hooks are disabled by hooksConfig.disabled: " + + ", ".join(disabled) + ) class GeminiPlugin(AgentPlugin): - """Install/uninstall Elydora audit hook for Gemini CLI.""" + """Install Elydora into Gemini CLI's native user settings.""" - @staticmethod - def _hook_path_for(agent_id: str) -> str: - return os.path.join(ELYDORA_DIR, agent_id, "hook.py") + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + document = read_gemini_document() + _require_enabled_hooks(document) + preflight_gemini_installation(config, document) def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - agent_name = config.get("agent_name", "") - - # Create per-agent directory - agent_dir = os.path.join(ELYDORA_DIR, agent_id) - os.makedirs(agent_dir, exist_ok=True) - - # Write config.json - config_data = { - "org_id": config.get("org_id", ""), - "agent_id": agent_id, - "kid": config.get("kid", ""), - "base_url": config.get("base_url", "https://api.elydora.com"), - "token": config.get("token", ""), - "agent_name": agent_name, - } - config_path = os.path.join(agent_dir, "config.json") - write_json_atomic( - config_path, - config_data, - 0o600, - "Elydora runtime config", + document = read_gemini_document() + _require_enabled_hooks(document) + paths = preflight_gemini_installation(config, document) + rendered = render_gemini_document( + document, + None, + { + "BeforeTool": build_gemini_group( + paths.guard_path, GUARD_HOOK_NAME + ), + "AfterTool": build_gemini_group( + paths.audit_path, AUDIT_HOOK_NAME + ), + }, ) - - # Write private key - private_key_path = os.path.join(agent_dir, "private.key") - write_text_atomic( - private_key_path, - config.get("private_key", ""), - 0o600, - "Elydora private key", - ) - - script = generate_hook_script( - org_id=config.get("org_id", ""), - agent_id=agent_id, - kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), - ) - hook_path = self._hook_path_for(agent_id) - write_text_atomic( - hook_path, - script, - 0o700, - "Elydora audit runtime", - ) - - guard_script_path = config.get("guard_script_path", "") - python_exe = sys.executable - - settings = _load_json(SETTINGS_PATH) - hooks = settings.setdefault("hooks", {}) - - # --- BeforeTool (guard — freeze enforcement) --- - before_tool = hooks.setdefault("BeforeTool", []) - before_tool[:] = [h for h in before_tool if not _is_elydora_hook(h)] - if guard_script_path: - before_tool.append({ - "hooks": [ - { - "type": "command", - "command": f'"{python_exe}" {guard_script_path}', - } - ], - }) - - # --- AfterTool (audit logging) --- - after_tool = hooks.setdefault("AfterTool", []) - - after_tool[:] = [h for h in after_tool if not _is_elydora_hook(h)] - - after_tool.append({ - "hooks": [ - { - "type": "command", - "command": hook_path, - } - ], - }) - - _save_json(SETTINGS_PATH, settings) - print("Elydora hook installed for Gemini CLI.") - print(f" Hook script: {hook_path}") - print(f" Settings: {SETTINGS_PATH}") + changes = prepare_gemini_installation(config, paths, rendered) + commit_gemini_installation(changes) + print(f"Gemini CLI hooks installed at {document.file_path}") + print("Gemini CLI verification: run /hooks list.") def uninstall(self, agent_id: str = "") -> None: - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - changed = False - - # Remove BeforeTool entries - before_tool = hooks.get("BeforeTool", []) - before_filtered = [h for h in before_tool if not _is_elydora_hook(h, agent_id)] - if len(before_filtered) != len(before_tool): - hooks["BeforeTool"] = before_filtered - if not before_filtered: - del hooks["BeforeTool"] - changed = True - - # Remove AfterTool entries - after_tool = hooks.get("AfterTool", []) - after_filtered = [h for h in after_tool if not _is_elydora_hook(h, agent_id)] - if len(after_filtered) != len(after_tool): - hooks["AfterTool"] = after_filtered - if not after_filtered: - del hooks["AfterTool"] - changed = True - - if changed: - if not hooks: - del settings["hooks"] - _save_json(SETTINGS_PATH, settings) - - # Hook script removal is handled by cli.py cmd_uninstall (rmtree of agent dir) - print("Elydora hook uninstalled from Gemini CLI.") + document = read_gemini_document() + write_gemini_document( + render_gemini_document(document, agent_id or None, {}) + ) def status(self) -> PluginStatus: - # Scan ~/.elydora/*/hook.py for any installed hook - import glob as _glob - hook_pattern = os.path.join(ELYDORA_DIR, "*", "hook.py") - hook_files = _glob.glob(hook_pattern) - hook_exists = len(hook_files) > 0 - - settings_configured = False - if os.path.exists(SETTINGS_PATH): - settings = _load_json(SETTINGS_PATH) - hooks = settings.get("hooks", {}) - before_tool = hooks.get("BeforeTool", []) - after_tool = hooks.get("AfterTool", []) - before_configured = any(_is_elydora_hook(h) for h in before_tool) - after_configured = any(_is_elydora_hook(h) for h in after_tool) - settings_configured = before_configured and after_configured - - installed = hook_exists and settings_configured + document = read_gemini_document() + contracts = gemini_runtime_contracts(document.hooks) + controls_enabled = managed_gemini_hooks_enabled( + document.hook_controls + ) + configured = controls_enabled and bool(contracts) + installed = configured and gemini_runtime_files_exist(contracts) if installed: - details = f"Found {len(hook_files)} agent(s): {', '.join(hook_files)}" - elif hook_exists: - details = "Hook script exists but not configured in settings" - elif settings_configured: - details = "Configured in settings but hook script missing" + details = f"Config: {document.file_path}" + elif contracts and not controls_enabled: + details = f"Configured hooks are disabled: {document.file_path}" + elif contracts: + details = ( + f"Configured at {document.file_path}; managed contract incomplete" + ) else: details = "Not installed" - - return PluginStatus(installed=installed, agent="gemini", details=details) - - -def _is_elydora_hook(entry: dict, agent_id: str = "") -> bool: - # Collect all command strings from the entry - commands: list[str] = [] - inner_hooks = entry.get("hooks") - if isinstance(inner_hooks, list): - for hook in inner_hooks: - if isinstance(hook, dict) and isinstance(hook.get("command"), str): - commands.append(hook["command"]) - else: - command = entry.get("command") - if isinstance(command, str): - commands.append(command) - - for cmd in commands: - cmd_lower = cmd.lower() - if "elydora" not in cmd_lower: - continue - # If agent_id is specified, only match hooks for that specific agent - if agent_id and agent_id in cmd: - return True - if not agent_id: - return True - return False - - -def _load_json(path: str) -> dict: - if not os.path.exists(path): - return {} - with open(path, "r", encoding="utf-8") as f: - return json.load(f) - - -def _save_json(path: str, data: dict) -> None: - write_json_atomic(path, data, 0o600, "Gemini settings") + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details, + ) diff --git a/sdks/python/elydora/plugins/gemini_command.py b/sdks/python/elydora/plugins/gemini_command.py new file mode 100644 index 0000000..c98097c --- /dev/null +++ b/sdks/python/elydora/plugins/gemini_command.py @@ -0,0 +1,206 @@ +"""Gemini CLI hook command construction and exact runtime ownership.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +import ntpath +import os +import re +import sys +from typing import Optional, Tuple + + +@dataclass(frozen=True) +class GeminiRuntimeReference: + agent_id: str + script_path: str + + +def same_gemini_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def same_gemini_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") + 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"& {_quote_powershell(_windows_powershell_path())} " + "-NoLogo -NoProfile -NonInteractive -EncodedCommand " + f"{encoded}" + ) + + +def build_gemini_command(script_path: str) -> str: + if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): + raise ValueError( + "Gemini CLI hook commands require absolute executable and script paths" + ) + if os.name == "nt": + return _windows_command(script_path) + return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" + + +def _read_posix_argument(command: str, start: int) -> Optional[Tuple[str, int]]: + 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 value, index + 1 + value += command[index] + index += 1 + return None + + +def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: + executable = _read_posix_argument(command, 0) + if executable is None or executable[1] >= len(command): + return None + if command[executable[1]] != " ": + return None + script = _read_posix_argument(command, executable[1] + 1) + if script is None or script[1] != len(command): + return None + return executable[0], script[0] + + +def _read_powershell_argument( + command: str, start: int +) -> Optional[Tuple[str, int]]: + 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 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 executable[1] >= len(source): + return None + if source[executable[1]] != " ": + return None + script = _read_powershell_argument(source, executable[1] + 1) + if script is None or source[script[1] :] != "; exit $LASTEXITCODE": + return None + return executable[0], script[0] + + +def _parse_windows_command(command: str) -> Optional[Tuple[str, str]]: + if not command.startswith("& "): + return None + powershell = _read_powershell_argument(command, 2) + if ( + powershell is None + or not ntpath.isabs(powershell[0]) + or ntpath.basename(powershell[0]).lower() != "powershell.exe" + ): + return None + prefix = " -NoLogo -NoProfile -NonInteractive -EncodedCommand " + if not command.startswith(prefix, powershell[1]): + return None + encoded = command[powershell[1] + len(prefix) :] + if re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", encoded) is None: + return None + try: + payload = base64.b64decode(encoded, validate=True) + if base64.b64encode(payload).decode("ascii") != encoded: + return None + if len(payload) % 2: + return None + source = payload.decode("utf-16le") + except (ValueError, UnicodeDecodeError): + return None + return _parse_powershell_source(source) + + +def _parse_legacy_command( + command: str, script_name: str +) -> Optional[Tuple[str, str]]: + if script_name == "guard.py": + match = re.fullmatch(r'"([^"\r\n]+)" ([^\r\n]+)', command) + if match is None or not same_gemini_path(match.group(1), sys.executable): + return None + return match.group(1), match.group(2) + if "\r" in command or "\n" in command: + return None + return sys.executable, command + + +def gemini_runtime_reference( + command: str, + script_name: str, + include_legacy: bool = False, +) -> Optional[GeminiRuntimeReference]: + parsed = _parse_posix_command(command) or _parse_windows_command(command) + if parsed is None and include_legacy: + parsed = _parse_legacy_command(command, script_name) + if ( + parsed is None + or not same_gemini_path(parsed[0], sys.executable) + or not os.path.isabs(parsed[1]) + or os.path.basename(parsed[1]) != script_name + ): + return None + agent_directory = os.path.dirname(parsed[1]) + runtime_root = os.path.join(os.path.expanduser("~"), ".elydora") + if not same_gemini_path(os.path.dirname(agent_directory), runtime_root): + return None + agent_id = os.path.basename(agent_directory) + if agent_id in {"", ".", ".."}: + return None + return GeminiRuntimeReference(agent_id, parsed[1]) diff --git a/sdks/python/elydora/plugins/gemini_config.py b/sdks/python/elydora/plugins/gemini_config.py new file mode 100644 index 0000000..11a2357 --- /dev/null +++ b/sdks/python/elydora/plugins/gemini_config.py @@ -0,0 +1,181 @@ +"""Gemini CLI JSONC parsing and source-preserving hook edits.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, Optional + +from ._jsonc import JsoncEditor +from .gemini_contract import ( + MANAGED_EVENTS, + GeminiHookControls, + GeminiHooks, + JsonObject, + managed_gemini_removals, + read_gemini_hook_controls, + read_gemini_hooks, +) + + +OWNED_FILE_MARKER = "// Managed by Elydora" + + +@dataclass(frozen=True) +class GeminiDocument: + file_path: str + exists: bool + raw: str + root: JsonObject + hooks: GeminiHooks + hook_controls: GeminiHookControls + has_hooks_container: bool + owned_file: bool + + +@dataclass(frozen=True) +class RenderedGeminiDocument: + document: GeminiDocument + changed: bool + next_source: Optional[str] + + +def _label(file_path: str) -> str: + return f"Gemini CLI user settings at {file_path}" + + +def parse_gemini_document( + *, exists: bool, file_path: str, raw: str +) -> GeminiDocument: + editor = JsoncEditor( + raw, + _label(file_path), + allow_trailing_commas=False, + ) + if not isinstance(editor.value, dict): + raise ValueError(f"{_label(file_path)} must contain a JSON object") + root = editor.value + return GeminiDocument( + file_path=file_path, + exists=exists, + raw=raw, + root=root, + hooks=(read_gemini_hooks(root["hooks"]) if "hooks" in root else {}), + hook_controls=( + read_gemini_hook_controls(root["hooksConfig"]) + if "hooksConfig" in root + else GeminiHookControls(True, ()) + ), + has_hooks_container="hooks" in root, + owned_file=raw.startswith(OWNED_FILE_MARKER), + ) + + +def create_gemini_document(file_path: str) -> GeminiDocument: + return parse_gemini_document( + exists=False, + file_path=file_path, + raw=f"{OWNED_FILE_MARKER}\n{{}}\n", + ) + + +def _current_document( + document: GeminiDocument, editor: JsoncEditor +) -> GeminiDocument: + return parse_gemini_document( + exists=document.exists, + file_path=document.file_path, + raw=editor.raw, + ) + + +def _remove_managed( + editor: JsoncEditor, + document: GeminiDocument, + agent_id: Optional[str], +) -> None: + removals = managed_gemini_removals(document.hooks, agent_id) + for event in MANAGED_EVENTS: + event_removals = sorted( + (item for item in removals if item.event == event), + key=lambda item: item.group_index, + reverse=True, + ) + for removal in event_removals: + group_path = ("hooks", event, removal.group_index) + if removal.remove_group: + editor.delete(group_path) + continue + for handler_index in sorted( + removal.handler_indexes, reverse=True + ): + editor.delete((*group_path, "hooks", handler_index)) + if event_removals: + current = _current_document(document, editor) + if event in current.hooks and not current.hooks[event]: + editor.delete(("hooks", event)) + current = _current_document(document, editor) + if current.has_hooks_container and not current.hooks: + editor.delete(("hooks",)) + + +def _append_group( + editor: JsoncEditor, + document: GeminiDocument, + event: str, + group: JsonObject, +) -> None: + current = _current_document(document, editor) + if not current.has_hooks_container: + editor.add_property((), "hooks", {event: [group]}) + elif event in current.hooks: + editor.append(("hooks", event), group) + else: + editor.add_property(("hooks",), event, [group]) + + +def _already_installed( + document: GeminiDocument, additions: Dict[str, JsonObject] +) -> bool: + if set(additions) != set(MANAGED_EVENTS): + return False + removals = managed_gemini_removals(document.hooks) + for event in MANAGED_EVENTS: + expected = additions[event] + event_removals = [item for item in removals if item.event == event] + exact_groups = [ + group for group in document.hooks.get(event, []) if group == expected + ] + if ( + len(event_removals) != 1 + or not event_removals[0].remove_group + or len(exact_groups) != 1 + ): + return False + return True + + +def render_gemini_document( + document: GeminiDocument, + agent_id: Optional[str], + additions: Dict[str, JsonObject], +) -> RenderedGeminiDocument: + if agent_id is None and _already_installed(document, additions): + return RenderedGeminiDocument(document, False, document.raw) + editor = JsoncEditor( + document.raw, + _label(document.file_path), + allow_trailing_commas=False, + ) + _remove_managed(editor, document, agent_id) + for event in MANAGED_EVENTS: + group = additions.get(event) + if group is not None: + _append_group(editor, document, event, group) + current = _current_document(document, editor) + if not additions and document.owned_file and not current.root: + return RenderedGeminiDocument(document, True, None) + return RenderedGeminiDocument( + document, + editor.raw != document.raw, + editor.raw, + ) diff --git a/sdks/python/elydora/plugins/gemini_contract.py b/sdks/python/elydora/plugins/gemini_contract.py new file mode 100644 index 0000000..caac42a --- /dev/null +++ b/sdks/python/elydora/plugins/gemini_contract.py @@ -0,0 +1,344 @@ +"""Gemini CLI hook schema, controls, and exact ownership rules.""" + +from __future__ import annotations + +from dataclasses import dataclass +import math +import os +from typing import Any, Dict, List, Optional, Tuple + +from .gemini_command import ( + GeminiRuntimeReference, + build_gemini_command, + gemini_runtime_reference, + same_gemini_agent_id, +) + + +AGENT_KEY = "gemini" +CONFIG_FILE = "settings.json" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +GUARD_HOOK_NAME = "elydora-guard" +AUDIT_HOOK_NAME = "elydora-audit" +HOOK_TIMEOUT_MILLISECONDS = 10_000 +MANAGED_EVENTS = ("BeforeTool", "AfterTool") + +KNOWN_EVENTS = { + "BeforeTool", + "AfterTool", + "BeforeAgent", + "Notification", + "AfterAgent", + "SessionStart", + "SessionEnd", + "PreCompress", + "BeforeModel", + "AfterModel", + "BeforeToolSelection", +} + +JsonObject = Dict[str, Any] +GeminiHooks = Dict[str, List[JsonObject]] + + +@dataclass(frozen=True) +class GeminiHookControls: + enabled: bool + disabled: Tuple[str, ...] + + +@dataclass(frozen=True) +class GeminiRuntimeContract: + agent_id: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class ManagedRemoval: + event: str + group_index: int + handler_indexes: Tuple[int, ...] + remove_group: bool + + +def elydora_dir() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def _optional_string(value: JsonObject, field: str, label: str) -> None: + if field in value and not isinstance(value[field], str): + raise ValueError(f'{label} field "{field}" must be a string') + + +def _validate_timeout(value: JsonObject, label: str) -> None: + if "timeout" not in value: + return + timeout = value["timeout"] + if ( + isinstance(timeout, bool) + or not isinstance(timeout, (int, float)) + or not math.isfinite(float(timeout)) + or timeout < 0 + ): + raise ValueError(f"{label} timeout must be a non-negative finite number") + + +def _validate_environment(value: JsonObject, label: str) -> None: + if "env" not in value: + return + environment = value["env"] + if not isinstance(environment, dict) or any( + not isinstance(item, str) for item in environment.values() + ): + raise ValueError(f"{label} env must map names to strings") + + +def _validate_handler( + value: Any, event: str, group_index: int, handler_index: int +) -> JsonObject: + label = ( + f"Gemini CLI settings handler hooks.{event}[{group_index}]" + f".hooks[{handler_index}]" + ) + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + handler = dict(value) + if handler.get("type") != "command": + raise ValueError( + f'{label} has unsupported type "{handler.get("type")}"' + ) + for field in ("name", "description", "source"): + _optional_string(handler, field, label) + _validate_timeout(handler, label) + _validate_environment(handler, label) + if not isinstance(handler.get("command"), str) or not handler["command"]: + raise ValueError(f"{label} requires a non-empty command") + return handler + + +def _validate_group(value: Any, event: str, index: int) -> JsonObject: + label = f"Gemini CLI settings group hooks.{event}[{index}]" + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + group = dict(value) + if "matcher" in group and not isinstance(group["matcher"], str): + raise ValueError(f"{label} matcher must be a string") + if "sequential" in group and not isinstance(group["sequential"], bool): + raise ValueError(f"{label} sequential must be a boolean") + handlers = group.get("hooks") + if not isinstance(handlers, list): + raise ValueError(f"{label} must contain a hooks array") + group["hooks"] = [ + _validate_handler(handler, event, index, handler_index) + for handler_index, handler in enumerate(handlers) + ] + return group + + +def read_gemini_hooks(value: Any) -> GeminiHooks: + if not isinstance(value, dict): + raise ValueError('Gemini CLI settings field "hooks" must be an object') + hooks: GeminiHooks = {} + for event, groups in value.items(): + if not isinstance(groups, list): + raise ValueError( + f'Gemini CLI settings field "hooks.{event}" must be an array' + ) + hooks[event] = ( + [ + _validate_group(group, event, index) + for index, group in enumerate(groups) + ] + if event in KNOWN_EVENTS + else list(groups) + ) + return hooks + + +def read_gemini_hook_controls(value: Any) -> GeminiHookControls: + if not isinstance(value, dict): + raise ValueError( + 'Gemini CLI settings field "hooksConfig" must be an object' + ) + supported = {"enabled", "disabled", "notifications"} + extra = next((field for field in value if field not in supported), None) + if extra is not None: + raise ValueError( + "Gemini CLI settings field \"hooksConfig\" contains unsupported " + f'field "{extra}"' + ) + if "enabled" in value and not isinstance(value["enabled"], bool): + raise ValueError( + 'Gemini CLI settings field "hooksConfig.enabled" must be a boolean' + ) + if "notifications" in value and not isinstance( + value["notifications"], bool + ): + raise ValueError( + "Gemini CLI settings field \"hooksConfig.notifications\" must be " + "a boolean" + ) + disabled = value.get("disabled", []) + if not isinstance(disabled, list) or any( + not isinstance(item, str) for item in disabled + ): + raise ValueError( + "Gemini CLI settings field \"hooksConfig.disabled\" must be an " + "array of strings" + ) + return GeminiHookControls(value.get("enabled") is not False, tuple(disabled)) + + +def managed_gemini_hooks_enabled(controls: GeminiHookControls) -> bool: + return ( + controls.enabled + and GUARD_HOOK_NAME not in controls.disabled + and AUDIT_HOOK_NAME not in controls.disabled + ) + + +def disabled_managed_gemini_entries( + controls: GeminiHookControls, +) -> Tuple[str, ...]: + return tuple( + entry + for entry in controls.disabled + if entry in {GUARD_HOOK_NAME, AUDIT_HOOK_NAME} + or gemini_runtime_reference(entry, GUARD_SCRIPT, True) is not None + or gemini_runtime_reference(entry, AUDIT_SCRIPT, True) is not None + ) + + +def build_gemini_group(script_path: str, name: str) -> JsonObject: + return { + "hooks": [{ + "type": "command", + "name": name, + "command": build_gemini_command(script_path), + "timeout": HOOK_TIMEOUT_MILLISECONDS, + }] + } + + +def _exact_managed_group(group: JsonObject) -> bool: + return set(group) == {"hooks"} + + +def _current_managed_reference( + handler: JsonObject, script_name: str, hook_name: str +) -> Optional[GeminiRuntimeReference]: + if ( + set(handler) == {"command", "name", "timeout", "type"} + and handler.get("type") == "command" + and handler.get("name") == hook_name + and handler.get("timeout") == HOOK_TIMEOUT_MILLISECONDS + and isinstance(handler.get("command"), str) + ): + return gemini_runtime_reference(handler["command"], script_name) + return None + + +def _legacy_managed_reference( + handler: JsonObject, script_name: str +) -> Optional[GeminiRuntimeReference]: + if ( + set(handler) == {"command", "type"} + and handler.get("type") == "command" + and isinstance(handler.get("command"), str) + ): + return gemini_runtime_reference(handler["command"], script_name, True) + return None + + +def _managed_reference( + handler: JsonObject, + script_name: str, + hook_name: str, + include_legacy: bool, +) -> Optional[GeminiRuntimeReference]: + current = _current_managed_reference(handler, script_name, hook_name) + if current is not None or not include_legacy: + return current + return _legacy_managed_reference(handler, script_name) + + +def managed_gemini_removals( + hooks: GeminiHooks, agent_id: Optional[str] = None +) -> List[ManagedRemoval]: + removals = [] + for event, script_name, hook_name in ( + ("BeforeTool", GUARD_SCRIPT, GUARD_HOOK_NAME), + ("AfterTool", AUDIT_SCRIPT, AUDIT_HOOK_NAME), + ): + for group_index, group in enumerate(hooks.get(event, [])): + handlers = group.get("hooks", []) + indexes = tuple( + index + for index, handler in enumerate(handlers) + if isinstance(handler, dict) + and ( + reference := _managed_reference( + handler, script_name, hook_name, True + ) + ) + is not None + and ( + agent_id is None + or same_gemini_agent_id(reference.agent_id, agent_id) + ) + ) + if indexes: + removals.append( + ManagedRemoval( + event, + group_index, + indexes, + _exact_managed_group(group) + and len(indexes) == len(handlers), + ) + ) + return removals + + +def _references_for_event( + groups: List[JsonObject], script_name: str, hook_name: str +) -> Dict[str, List[GeminiRuntimeReference]]: + result: Dict[str, List[GeminiRuntimeReference]] = {} + for group in groups: + if not _exact_managed_group(group): + continue + for handler in group["hooks"]: + reference = _current_managed_reference( + handler, script_name, hook_name + ) + if reference is None: + continue + key = os.path.normcase(reference.agent_id) + result.setdefault(key, []).append(reference) + return result + + +def gemini_runtime_contracts( + hooks: GeminiHooks, +) -> List[GeminiRuntimeContract]: + guards = _references_for_event( + hooks.get("BeforeTool", []), GUARD_SCRIPT, GUARD_HOOK_NAME + ) + audits = _references_for_event( + hooks.get("AfterTool", []), AUDIT_SCRIPT, AUDIT_HOOK_NAME + ) + contracts = [] + for key, guard in guards.items(): + audit = audits.get(key, []) + if len(guard) != 1 or len(audit) != 1: + continue + contracts.append( + GeminiRuntimeContract( + guard[0].agent_id, + guard[0].script_path, + audit[0].script_path, + ) + ) + return contracts diff --git a/sdks/python/elydora/plugins/gemini_installation.py b/sdks/python/elydora/plugins/gemini_installation.py new file mode 100644 index 0000000..36b4c49 --- /dev/null +++ b/sdks/python/elydora/plugins/gemini_installation.py @@ -0,0 +1,161 @@ +"""Transactional Gemini CLI hook and runtime installation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +from typing import List, Optional + +from elydora._runtime_paths import resolve_agent_directory + +from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES +from ._transaction import FileChange, file_change, write_changes +from .base import InstallConfig +from .gemini_command import same_gemini_path +from .gemini_config import GeminiDocument, RenderedGeminiDocument +from .gemini_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + elydora_dir, +) +from .gemini_io import ( + rendered_change, + validate_api_origin, + validate_private_key, + validate_runtime_tree, +) +from .hook_template import generate_guard_script, generate_hook_script + + +@dataclass(frozen=True) +class GeminiRuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +def _runtime_config(config: InstallConfig, agent_id: str) -> dict: + value = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), + "agent_name": AGENT_KEY, + } + token = config.get("token") + if token: + value["token"] = token + return value + + +def _json_source(value: dict) -> str: + return json.dumps(value, indent=2) + "\n" + + +def _agent_paths(config: InstallConfig) -> GeminiRuntimePaths: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + agent_directory = resolve_agent_directory(elydora_dir(), agent_id) + guard_path = os.path.join(agent_directory, GUARD_SCRIPT) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + if not same_gemini_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + "Elydora guard runtime must use the managed agent directory: " + f"{guard_path}" + ) + return GeminiRuntimePaths(agent_id, agent_directory, guard_path, audit_path) + + +def _validate_install_config(config: InstallConfig) -> None: + for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): + value = config.get(field) + if not isinstance(value, str) or not value: + raise ValueError(f"{field} is required") + if config.get("agent_name") != AGENT_KEY: + raise ValueError( + f"Gemini CLI installation requires agent_name {AGENT_KEY}" + ) + validate_private_key(config["private_key"]) + validate_api_origin(config["base_url"]) + if "token" in config: + token = config["token"] + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string when provided") + _agent_paths(config) + + +def preflight_gemini_installation( + config: InstallConfig, document: GeminiDocument +) -> GeminiRuntimePaths: + if not document.file_path: + raise ValueError("Gemini CLI installation requires a settings path") + _validate_install_config(config) + paths = _agent_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id) + return paths + + +def _runtime_change( + file_path: str, + label: str, + next_source: str, + mode: int, + maximum_bytes: int = MAX_SOURCE_BYTES, +) -> Optional[FileChange]: + return file_change(file_path, label, next_source, mode, maximum_bytes) + + +def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def prepare_gemini_installation( + config: InstallConfig, + paths: GeminiRuntimePaths, + rendered: RenderedGeminiDocument, +) -> List[FileChange]: + guard_script = generate_guard_script( + AGENT_KEY, + paths.agent_id, + success_output="{}\n", + ) + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=paths.agent_id, + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + success_output="{}\n", + native_payload=True, + agent_name=AGENT_KEY, + ) + return _present([ + _runtime_change( + paths.guard_path, "Elydora guard runtime", guard_script, 0o700 + ), + _runtime_change( + os.path.join(paths.agent_directory, "config.json"), + "Elydora runtime config", + _json_source(_runtime_config(config, paths.agent_id)), + 0o600, + MAX_CONFIG_BYTES, + ), + _runtime_change( + os.path.join(paths.agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + _runtime_change( + paths.audit_path, "Elydora audit runtime", audit_script, 0o700 + ), + rendered_change(rendered), + ]) + + +def commit_gemini_installation(changes: List[FileChange]) -> None: + write_changes(changes, "Install Gemini CLI hooks") diff --git a/sdks/python/elydora/plugins/gemini_io.py b/sdks/python/elydora/plugins/gemini_io.py new file mode 100644 index 0000000..6ba5b51 --- /dev/null +++ b/sdks/python/elydora/plugins/gemini_io.py @@ -0,0 +1,250 @@ +"""Gemini CLI settings discovery and strict managed runtime inspection.""" + +from __future__ import annotations + +import base64 +import os +from typing import Any, List, Optional +import urllib.parse + +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + MAX_SOURCE_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, +) +from ._strict_json import JsonObject, parse_json_object +from ._transaction import FileChange, source_change, write_changes +from .gemini_command import same_gemini_agent_id, same_gemini_path +from .gemini_config import ( + GeminiDocument, + RenderedGeminiDocument, + create_gemini_document, + parse_gemini_document, +) +from .gemini_contract import ( + AGENT_KEY, + CONFIG_FILE, + GeminiRuntimeContract, + elydora_dir, +) + + +def gemini_configuration_directory() -> str: + configured_home = os.environ.get("GEMINI_CLI_HOME") + home = configured_home if configured_home else os.path.expanduser("~") + return os.path.join(home, ".gemini") + + +def gemini_settings_path() -> str: + return os.path.join(gemini_configuration_directory(), CONFIG_FILE) + + +def read_gemini_document() -> GeminiDocument: + file_path = gemini_settings_path() + physical_directory_exists( + os.path.dirname(file_path), "Gemini CLI configuration directory" + ) + snapshot = read_physical_file( + file_path, "Gemini CLI user settings", MAX_SOURCE_BYTES + ) + if snapshot is None: + return create_gemini_document(file_path) + return parse_gemini_document( + exists=True, + file_path=file_path, + raw=snapshot.contents, + ) + + +def rendered_change( + rendered: RenderedGeminiDocument, +) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.file_path, + "Gemini CLI user settings", + rendered.document.raw if rendered.document.exists else None, + rendered.next_source, + 0o600, + MAX_SOURCE_BYTES, + ) + + +def write_gemini_document(rendered: RenderedGeminiDocument) -> None: + change = rendered_change(rendered) + if change is not None: + write_changes([change], "Uninstall Gemini CLI hooks") + + +def validate_api_origin(value: str, label: str = "base_url") -> None: + try: + parsed = urllib.parse.urlsplit(value) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error + invalid_character = "\\" in value or any( + character.isspace() or ord(character) < 32 for character in value + ) + if ( + parsed.scheme not in ("http", "https") + or not parsed.netloc + or hostname is None + or invalid_character + ): + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") + if ( + parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ValueError( + f"{label} must exclude credentials, query parameters, and fragments" + ) + + +def validate_private_key(value: str, label: str = "private_key") -> None: + try: + padded = value + "=" * ((4 - len(value) % 4) % 4) + seed = base64.b64decode( + padded.replace("-", "+").replace("_", "/"), validate=True + ) + canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") + except (ValueError, UnicodeEncodeError) as error: + raise ValueError( + f"{label} must be a canonical 32-byte base64url value" + ) from error + if len(seed) != 32 or canonical != value: + raise ValueError(f"{label} must be a canonical 32-byte base64url value") + + +def _require_non_empty_string( + value: Any, field: str, config_path: str +) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError( + f"Elydora runtime config {field} is invalid: {config_path}" + ) + return value + + +def _validate_runtime_config( + config: JsonObject, expected_agent_id: str, config_path: str +) -> None: + supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} + extra = next((key for key in config if key not in supported), None) + if extra is not None: + raise ValueError( + f'Elydora runtime config has unsupported field "{extra}": {config_path}' + ) + _require_non_empty_string(config.get("org_id"), "org_id", config_path) + _require_non_empty_string(config.get("kid"), "kid", config_path) + agent_id = _require_non_empty_string( + config.get("agent_id"), "agent_id", config_path + ) + if ( + not same_gemini_agent_id(agent_id, expected_agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime identity does not match Gemini CLI hooks: " + f"{config_path}" + ) + if "token" in config: + _require_non_empty_string(config.get("token"), "token", config_path) + base_url = _require_non_empty_string( + config.get("base_url"), "base_url", config_path + ) + validate_api_origin(base_url, "Elydora runtime config base_url") + + +def _read_runtime_config(file_path: str) -> Optional[JsonObject]: + snapshot = read_physical_file( + file_path, "Elydora runtime config", MAX_CONFIG_BYTES + ) + if snapshot is None: + return None + return parse_json_object( + snapshot.contents, f"Elydora runtime config at {file_path}" + ) + + +def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: + root = elydora_dir() + if not physical_directory_exists(root, "Elydora runtime directory"): + return + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return + config_path = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path) + artifacts = ( + ("private.key", "Elydora private key"), + ("guard.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_exists = any( + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + configured_id = config.get("agent_id") + if ( + not isinstance(configured_id, str) + or not same_gemini_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime config identity does not match Gemini CLI agent " + f"{agent_id}: {config_path}" + ) + + +def _runtime_contract_exists(contract: GeminiRuntimeContract) -> bool: + root = elydora_dir() + agent_directory = os.path.dirname(contract.guard_path) + if ( + not same_gemini_path(os.path.dirname(agent_directory), root) + or not same_gemini_path( + contract.audit_path, os.path.join(agent_directory, "hook.py") + ) + ): + return False + if not physical_directory_exists(root, "Elydora runtime directory"): + return False + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return False + config_path = os.path.join(agent_directory, "config.json") + key_path = os.path.join(agent_directory, "private.key") + config = _read_runtime_config(config_path) + key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) + guard = read_physical_file(contract.guard_path, "Elydora guard runtime") + audit = read_physical_file(contract.audit_path, "Elydora audit runtime") + if config is None or key is None or guard is None or audit is None: + return False + _validate_runtime_config(config, contract.agent_id, config_path) + validate_private_key(key.contents, "Elydora private key") + return bool(guard.contents) and bool(audit.contents) + + +def gemini_runtime_files_exist( + contracts: List[GeminiRuntimeContract], +) -> bool: + return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/sdks/python/elydora/plugins/registry.py b/sdks/python/elydora/plugins/registry.py index a0d1862..c0c47bf 100644 --- a/sdks/python/elydora/plugins/registry.py +++ b/sdks/python/elydora/plugins/registry.py @@ -43,8 +43,8 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "gemini": { "name": "Gemini CLI", - "hook_event": "AfterTool", - "config_path": "~/.gemini/settings.json", + "hook_event": "BeforeTool/AfterTool", + "config_path": "$GEMINI_CLI_HOME/.gemini/settings.json", }, "grok": { "name": "Grok Build", diff --git a/sdks/python/tests/gemini_support.py b/sdks/python/tests/gemini_support.py new file mode 100644 index 0000000..96f93e7 --- /dev/null +++ b/sdks/python/tests/gemini_support.py @@ -0,0 +1,279 @@ +from __future__ import annotations + +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import ntpath +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._jsonc import parse_jsonc +from elydora.plugins.gemini import GeminiPlugin +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@dataclass +class GeminiFixture: + 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: GeminiPlugin + config: dict[str, str] + + def install(self) -> None: + self.plugin.install(self.config) + + def source(self) -> str: + with self.config_path.open("r", encoding="utf-8", newline="") as file: + return file.read() + + def settings(self) -> dict[str, Any]: + value = parse_jsonc( + self.source(), + f"Gemini CLI test settings at {self.config_path}", + allow_trailing_commas=False, + ) + assert isinstance(value, dict) + return value + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + operation_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls, status: Any, operation_status: int) -> None: + cls.agent_status = status + cls.operation_status = operation_status + cls.requests = [] + + def _respond(self, status: int, value: object) -> None: + body = json.dumps(value).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + type(self).requests.append({ + "method": "GET", + "path": self.path, + "authorization": self.headers.get("Authorization"), + }) + self._respond(200, {"agent": {"status": type(self).agent_status}}) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length) + type(self).requests.append({ + "method": "POST", + "path": self.path, + "authorization": self.headers.get("Authorization"), + "raw": raw, + "json": json.loads(raw), + }) + self._respond( + type(self).operation_status, + {"operation": {"accepted": True}}, + ) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server( + *, status: Any = "active", operation_status: int = 201 +) -> ThreadingHTTPServer: + ElydoraApiHandler.reset(status, operation_status) + server = ThreadingHTTPServer(("127.0.0.1", 0), ElydoraApiHandler) + Thread(target=server.serve_forever, daemon=True).start() + return server + + +def server_base_url(server: ThreadingHTTPServer) -> str: + return f"http://127.0.0.1:{server.server_port}" + + +def write_text(path: Path, source: str, mode: int = 0o600) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source, encoding="utf-8", newline="") + os.chmod(path, mode) + + +def write_json(path: Path, value: object, mode: int = 0o600) -> None: + source = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n" + write_text(path, source, mode) + + +def _settings_root( + home_dir: Path, project_dir: Path, override: object +) -> Path: + if override is MISSING or override == "": + return home_dir + configured = str(home_dir) if override is None else str(override) + value = Path(configured) + return value if value.is_absolute() else project_dir / value + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_settings: object = MISSING, + base_url: str = "http://127.0.0.1:9", + gemini_home_override: object = None, +) -> GeminiFixture: + home_dir = tmp_path / "home with spaces and 'quote $GEMINI_CWD %GEMINI_CWD%" + project_dir = tmp_path / "project with spaces" + project_dir.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.chdir(project_dir) + if gemini_home_override is MISSING: + monkeypatch.delenv("GEMINI_CLI_HOME", raising=False) + else: + configured = ( + str(home_dir) + if gemini_home_override is None + else str(gemini_home_override) + ) + monkeypatch.setenv("GEMINI_CLI_HOME", configured) + config_path = ( + _settings_root(home_dir, project_dir, gemini_home_override) + / ".gemini" + / "settings.json" + ) + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / "guard.py" + hook_path = agent_dir / "hook.py" + runtime_config_path = agent_dir / "config.json" + private_key_path = agent_dir / "private.key" + if existing_settings is not MISSING: + write_json(config_path, existing_settings) + return GeminiFixture( + 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=GeminiPlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "gemini", + "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, name: str +) -> dict[str, Any]: + for group in settings["hooks"][event]: + if set(group) != {"hooks"}: + continue + for handler in group["hooks"]: + if ( + set(handler) == {"type", "name", "command", "timeout"} + and handler.get("type") == "command" + and handler.get("name") == name + and handler.get("timeout") == 10_000 + ): + return handler + raise AssertionError(f"managed handler for {event!r} not found") + + +def assert_managed_handler( + handler: Mapping[str, Any], name: str +) -> None: + assert set(handler) == {"type", "name", "command", "timeout"} + assert handler["type"] == "command" + assert handler["name"] == name + assert handler["timeout"] == 10_000 + assert isinstance(handler["command"], str) + + +def run_handler( + handler: Mapping[str, Any], payload: bytes, fixture: GeminiFixture +) -> subprocess.CompletedProcess[bytes]: + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + "GEMINI_CLI_HOME": str(fixture.home_dir), + } + if os.name == "nt": + powershell = ntpath.join( + os.environ.get("SystemRoot", r"C:\Windows"), + "System32", + "WindowsPowerShell", + "v1.0", + "powershell.exe", + ) + source = ( + f'{handler["command"]}; ' + "if ($LASTEXITCODE -ne 0) { exit $LASTEXITCODE }" + ) + command = [ + powershell, + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + source, + ] + else: + command = ["/bin/bash", "-c", str(handler["command"])] + return subprocess.run( # nosec B603 + command, + input=payload, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + ) + + +def legacy_handler(script_path: Path) -> dict[str, str]: + command = ( + f'"{sys.executable}" {script_path}' + if script_path.name == "guard.py" + else str(script_path) + ) + return {"type": "command", "command": command} + + +def symlink_or_skip( + target: Path, link: Path, *, directory: bool = False +) -> None: + try: + link.symlink_to(target, target_is_directory=directory) + except OSError as error: + pytest.skip(f"Symbolic links unavailable: {error}") diff --git a/sdks/python/tests/test_cli_security.py b/sdks/python/tests/test_cli_security.py index a2e42d6..2496744 100644 --- a/sdks/python/tests/test_cli_security.py +++ b/sdks/python/tests/test_cli_security.py @@ -17,7 +17,6 @@ from elydora.plugins._file_io import write_text_atomic from elydora.plugins import ( copilot, cursor, - gemini, kiroide, letta, opencode, @@ -375,7 +374,6 @@ def test_atomic_secret_write_preserves_existing_file_on_replace_failure( [ (copilot, copilot.CopilotPlugin, "copilot"), (cursor, cursor.CursorPlugin, "cursor"), - (gemini, gemini.GeminiPlugin, "gemini"), (kiroide, kiroide.KiroIdePlugin, "kiroide"), (letta, letta.LettaPlugin, "letta"), (opencode, opencode.OpenCodePlugin, "opencode"), diff --git a/sdks/python/tests/test_gemini_official.py b/sdks/python/tests/test_gemini_official.py new file mode 100644 index 0000000..e5767e2 --- /dev/null +++ b/sdks/python/tests/test_gemini_official.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 + +import pytest + +from gemini_support import prepare_fixture + + +GEMINI_ENTRY = os.environ.get("ELYDORA_GEMINI_ENTRY") + + +@pytest.mark.skipif( + not GEMINI_ENTRY, + reason="set ELYDORA_GEMINI_ENTRY to the official Gemini CLI entry file", +) +def test_official_gemini_cli_accepts_installed_user_hooks( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + node = shutil.which("node") + assert node is not None + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + "GEMINI_CLI_HOME": str(fixture.home_dir), + "GEMINI_API_KEY": "official-loader-test-key", + "GEMINI_TELEMETRY_ENABLED": "false", + "OTEL_SDK_DISABLED": "true", + } + + version = subprocess.run( # nosec B603 + [node, str(GEMINI_ENTRY), "--version"], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + text=True, + ) + loaded = subprocess.run( # nosec B603 + [node, str(GEMINI_ENTRY), "--skip-trust", "--list-extensions"], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + text=True, + ) + + assert version.returncode == 0, version.stderr + assert "0.51.0" in version.stdout + assert loaded.returncode == 0, loaded.stderr + output = (loaded.stdout + "\n" + loaded.stderr).lower() + for message in ( + "invalid settings", + "settings validation", + "failed to parse", + "hook configuration error", + ): + assert message not in output diff --git a/sdks/python/tests/test_gemini_plugin.py b/sdks/python/tests/test_gemini_plugin.py new file mode 100644 index 0000000..f70f299 --- /dev/null +++ b/sdks/python/tests/test_gemini_plugin.py @@ -0,0 +1,363 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import subprocess # nosec B404 +import sys + +import pytest + +from gemini_support import ( + AGENT_ID, + MISSING, + VALID_PRIVATE_KEY, + assert_managed_handler, + legacy_handler, + managed_handler, + prepare_fixture, + write_json, + write_text, +) +from elydora.plugins.registry import SUPPORTED_AGENTS + + +GUARD_NAME = "elydora-guard" +AUDIT_NAME = "elydora-audit" + + +def assert_managed_pair(settings: dict, fixture: object) -> None: + guard = managed_handler(settings, "BeforeTool", GUARD_NAME) + audit = managed_handler(settings, "AfterTool", AUDIT_NAME) + assert_managed_handler(guard, GUARD_NAME) + assert_managed_handler(audit, AUDIT_NAME) + assert str(fixture.guard_path) not in ("", guard["command"]) + assert str(fixture.hook_path) not in ("", audit["command"]) + + +def test_gemini_is_registered_with_the_official_contract() -> None: + assert SUPPORTED_AGENTS["gemini"] == { + "name": "Gemini CLI", + "hook_event": "BeforeTool/AfterTool", + "config_path": "$GEMINI_CLI_HOME/.gemini/settings.json", + } + + +def test_install_preserves_jsonc_and_writes_one_exact_pair( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + existing = "\r\n".join([ + "{", + " // Keep this user preference.", + ' "theme": "GitHub",', + ' "hooks": {', + ' "FutureEvent": [null],', + ' "BeforeTool": [{ "matcher": "read_file", "hooks": ' + '[{ "type": "command", "command": "user-hook" }] }]', + " }", + "}", + "", + ]) + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=existing + ) + project_settings = fixture.project_dir / ".gemini" / "settings.json" + system_settings = tmp_path / "system-settings.json" + write_text(project_settings, '{ "owner": "project" }\n') + write_text(system_settings, '{ "owner": "system" }\n') + monkeypatch.setenv("GEMINI_CLI_SYSTEM_SETTINGS_PATH", str(system_settings)) + + fixture.install() + output = capsys.readouterr().out + assert "run /hooks list" in output + installed = fixture.settings() + assert "Keep this user preference" in fixture.source() + assert "\r\n" in fixture.source() + assert installed["theme"] == "GitHub" + assert installed["hooks"]["FutureEvent"] == [None] + assert installed["hooks"]["BeforeTool"][0]["hooks"][0]["command"] == "user-hook" + assert_managed_pair(installed, fixture) + first_source = fixture.source() + + fixture.install() + + assert fixture.source() == first_source + assert project_settings.read_text(encoding="utf-8") == '{ "owner": "project" }\n' + assert system_settings.read_text(encoding="utf-8") == '{ "owner": "system" }\n' + + +@pytest.mark.parametrize( + ("override", "kind"), + [ + (MISSING, "default"), + ("", "empty"), + ("relative gemini", "relative"), + ("~", "literal_tilde"), + ], +) +def test_gemini_home_matches_official_path_resolution( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + override: object, + kind: str, +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, gemini_home_override=override + ) + + fixture.install() + + expected = { + "default": fixture.home_dir / ".gemini" / "settings.json", + "empty": fixture.home_dir / ".gemini" / "settings.json", + "relative": fixture.project_dir / "relative gemini" / ".gemini" / "settings.json", + "literal_tilde": fixture.project_dir / "~" / ".gemini" / "settings.json", + }[kind] + assert fixture.config_path == expected + assert expected.is_file() + + +INVALID_SETTINGS = [ + ("{ malformed", "parse Gemini CLI user settings"), + ("[]", "must contain a JSON object"), + ('{ "theme": true, }', "trailing commas are not allowed"), + ('{ "hooks": {}, "hooks": {} }', 'duplicate field "hooks"'), + ('{ "hooks": null }', 'field "hooks" must be an object'), + ('{ "hooks": { "BeforeTool": null } }', "must be an array"), + ('{ "hooks": { "BeforeTool": [null] } }', "group.*must be an object"), + ( + '{ "hooks": { "BeforeTool": [{ "matcher": 1, "hooks": [] }] } }', + "matcher must be a string", + ), + ( + '{ "hooks": { "BeforeTool": [{ "sequential": 1, "hooks": [] }] } }', + "sequential must be a boolean", + ), + ('{ "hooks": { "BeforeTool": [{}] } }', "hooks array"), + ('{ "hooks": { "BeforeTool": [{ "hooks": [null] }] } }', "handler"), + ( + '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "http" }] }] } }', + "unsupported type", + ), + ( + '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "command", ' + '"command": "" }] }] } }', + "non-empty command", + ), + ( + '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "command", ' + '"command": "x", "timeout": -1 }] }] } }', + "non-negative finite number", + ), + ( + '{ "hooks": { "BeforeTool": [{ "hooks": [{ "type": "command", ' + '"command": "x", "env": { "A": 1 } }] }] } }', + "env must map names to strings", + ), + ('{ "hooksConfig": null }', "hooksConfig.*must be an object"), + ('{ "hooksConfig": { "future": true } }', 'unsupported field "future"'), + ('{ "hooksConfig": { "enabled": "yes" } }', "enabled.*must be a boolean"), + ('{ "hooksConfig": { "disabled": [1] } }', "array of strings"), + ('{ "hooksConfig": { "notifications": 1 } }', "notifications.*boolean"), +] + + +@pytest.mark.parametrize(("source", "pattern"), INVALID_SETTINGS) +def test_invalid_official_settings_fail_before_every_write( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + source: str, + pattern: str, +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=source + ) + + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert fixture.source() == source + assert not fixture.agent_dir.exists() + + +@pytest.mark.parametrize( + ("settings", "pattern"), + [ + ({"hooksConfig": {"enabled": False}}, "hooksConfig.enabled"), + ({"hooksConfig": {"disabled": [GUARD_NAME]}}, GUARD_NAME), + ({"hooksConfig": {"disabled": [AUDIT_NAME]}}, AUDIT_NAME), + ], +) +def test_canonical_hook_controls_block_installation( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + settings: dict, + pattern: str, +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=settings + ) + with pytest.raises(ValueError, match=pattern): + fixture.install() + assert not fixture.agent_dir.exists() + + +def test_disabled_legacy_command_blocks_installation( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + write_json(fixture.config_path, { + "hooksConfig": { + "disabled": [legacy_handler(fixture.guard_path)["command"]] + } + }) + + with pytest.raises(ValueError, match="hooksConfig.disabled"): + fixture.install() + + assert not fixture.agent_dir.exists() + + +def test_exact_legacy_hooks_migrate_and_lookalikes_survive_uninstall( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + lookalike = legacy_handler(fixture.guard_path) + lookalike["command"] += " --inspect" + write_json(fixture.config_path, { + "hooks": { + "BeforeTool": [ + {"hooks": [legacy_handler(fixture.guard_path)]}, + {"hooks": [lookalike]}, + ], + "AfterTool": [{"hooks": [legacy_handler(fixture.hook_path)]}], + } + }) + + fixture.install() + settings = fixture.settings() + assert_managed_pair(settings, fixture) + assert any( + group["hooks"][0].get("command") == lookalike["command"] + for group in settings["hooks"]["BeforeTool"] + ) + settings["hooks"]["BeforeTool"][-1]["hooks"].append({ + "type": "command", "command": "user-command" + }) + write_json(fixture.config_path, settings) + + fixture.plugin.uninstall(AGENT_ID) + + remaining = fixture.settings()["hooks"] + assert remaining["BeforeTool"] == [ + {"hooks": [lookalike]}, + {"hooks": [{"type": "command", "command": "user-command"}]}, + ] + assert "AfterTool" not in remaining + + +def test_uninstall_preserves_user_settings_and_removes_owned_file( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user = prepare_fixture( + monkeypatch, + tmp_path / "user", + existing_settings={"theme": "GitHub", "hooks": {"Notification": []}}, + ) + user.install() + user.plugin.uninstall(AGENT_ID) + assert user.settings() == { + "theme": "GitHub", "hooks": {"Notification": []} + } + + owned = prepare_fixture(monkeypatch, tmp_path / "owned") + owned.install() + owned.plugin.uninstall(AGENT_ID) + assert not owned.config_path.exists() + + +def test_status_requires_enabled_unique_pair_and_strict_runtime( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.plugin.status()["installed"] is True + + settings = fixture.settings() + settings["hooks"]["AfterTool"].append(settings["hooks"]["AfterTool"][-1]) + write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + + fixture.install() + settings = fixture.settings() + settings["hooksConfig"] = {"disabled": [AUDIT_NAME]} + write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + + settings["hooksConfig"] = {"disabled": []} + write_json(fixture.config_path, settings) + write_text(fixture.private_key_path, "invalid") + with pytest.raises(ValueError, match="canonical 32-byte"): + fixture.plugin.status() + + +def test_cli_install_status_and_uninstall_end_to_end( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + key_path = tmp_path / "install-private.key" + token_path = tmp_path / "install-token.txt" + write_text(key_path, VALID_PRIVATE_KEY + "\n") + write_text(token_path, "token-1\n") + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + "GEMINI_CLI_HOME": str(fixture.home_dir), + } + root = Path(__file__).parents[1] + base = [sys.executable, "-m", "elydora.cli"] + + install = subprocess.run( # nosec B603 + [ + *base, "install", "--agent", "gemini", + "--org_id", "org-1", "--agent_id", AGENT_ID, + "--private_key_file", str(key_path), + "--token_file", str(token_path), + "--kid", "key-1", "--base_url", "http://127.0.0.1:9", + ], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert install.returncode == 0, install.stderr + assert "run /hooks list" in install.stdout + assert_managed_pair(fixture.settings(), fixture) + + status = subprocess.run( # nosec B603 + [*base, "status"], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert status.returncode == 0, status.stderr + assert "Gemini CLI" in status.stdout + assert "[installed]" in status.stdout + + uninstall = subprocess.run( # nosec B603 + [*base, "uninstall", "--agent", "gemini", "--agent_id", AGENT_ID], + capture_output=True, + check=False, + cwd=root, + env=environment, + text=True, + ) + assert uninstall.returncode == 0, uninstall.stderr + assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() diff --git a/sdks/python/tests/test_gemini_runtime.py b/sdks/python/tests/test_gemini_runtime.py new file mode 100644 index 0000000..7c7e06d --- /dev/null +++ b/sdks/python/tests/test_gemini_runtime.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from gemini_support import ( + ElydoraApiHandler, + managed_handler, + prepare_fixture, + run_handler, + server_base_url, + start_api_server, +) + + +GUARD_NAME = "elydora-guard" +AUDIT_NAME = "elydora-audit" + + +def official_payload(event: str, **overrides: object) -> dict[str, object]: + value: dict[str, object] = { + "session_id": "session-1", + "transcript_path": "/tmp/session-1.jsonl", + "cwd": "/tmp/project", + "hook_event_name": event, + "timestamp": "2026-07-19T00:00:00.000Z", + "tool_name": "run_shell_command", + "tool_input": {"command": "pytest"}, + } + if event == "AfterTool": + value["tool_response"] = {"output": "passed", "error": None} + value.update(overrides) + return value + + +def encoded(value: object) -> bytes: + return json.dumps(value, separators=(",", ":")).encode() + + +def test_guard_accepts_active_agents_with_valid_json_stdout( + 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() + result = run_handler( + managed_handler(fixture.settings(), "BeforeTool", GUARD_NAME), + encoded(official_payload("BeforeTool")), + fixture, + ) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout) == {} + assert [request["method"] for request in ElydoraApiHandler.requests] == ["GET"] + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_propagates_official_exit_code_two( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, +) -> None: + server = start_api_server(status=status) + try: + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + fixture.install() + result = run_handler( + managed_handler(fixture.settings(), "BeforeTool", GUARD_NAME), + encoded(official_payload("BeforeTool")), + fixture, + ) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 2 + assert result.stdout == b"" + assert status.encode() in result.stderr.lower() + + +def test_audit_preserves_complete_native_after_tool_payload( + 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() + payload = official_payload( + "AfterTool", + mcp_context={"server_name": "filesystem", "tool_name": "write_file"}, + original_request_name="write_file", + future_provider_field={"preserved": True}, + ) + result = run_handler( + managed_handler(fixture.settings(), "AfterTool", AUDIT_NAME), + encoded(payload), + fixture, + ) + finally: + server.shutdown() + server.server_close() + + assert result.returncode == 0, result.stderr + assert json.loads(result.stdout) == {} + operation = next( + request["json"] + for request in ElydoraApiHandler.requests + if request["method"] == "POST" + ) + assert operation["payload"] == payload + assert operation["subject"] == {"session_id": "session-1"} + assert operation["action"] == {"tool": "run_shell_command"} + + +def test_runtime_failures_stay_observable_and_fail_open( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + settings = fixture.settings() + guard = managed_handler(settings, "BeforeTool", GUARD_NAME) + audit = managed_handler(settings, "AfterTool", AUDIT_NAME) + + invalid_guard = run_handler(guard, b"{ malformed", fixture) + audit_result = run_handler( + audit, encoded(official_payload("AfterTool")), fixture + ) + + assert invalid_guard.returncode == 0 + assert json.loads(invalid_guard.stdout) == {} + assert b"invalid JSON" in invalid_guard.stderr + assert audit_result.returncode == 0 + assert json.loads(audit_result.stdout) == {} + log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8") + assert "[elydora-hook]" in log + assert "refused" in log.lower() or "urlerror" in log.lower() + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits apply on POSIX") +def test_runtime_artifacts_use_private_modes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + for path, expected in ( + (fixture.runtime_config_path, 0o600), + (fixture.private_key_path, 0o600), + (fixture.guard_path, 0o700), + (fixture.hook_path, 0o700), + (fixture.config_path, 0o600), + ): + assert path.stat().st_mode & 0o777 == expected diff --git a/sdks/python/tests/test_gemini_transaction.py b/sdks/python/tests/test_gemini_transaction.py new file mode 100644 index 0000000..8d77ad5 --- /dev/null +++ b/sdks/python/tests/test_gemini_transaction.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 +import sys + +import pytest + +from gemini_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + prepare_fixture, + symlink_or_skip, + write_json, + write_text, +) +from elydora.plugins import _transaction, gemini_installation +from elydora.plugins.gemini_config import render_gemini_document +from elydora.plugins.gemini_contract import ( + AUDIT_HOOK_NAME, + GUARD_HOOK_NAME, + build_gemini_group, +) +from elydora.plugins.gemini_io import read_gemini_document + + +RUNTIME_NAMES = ("guard.py", "config.json", "private.key", "hook.py") + + +def prepare_installation(fixture: object) -> list[_transaction.FileChange]: + document = read_gemini_document() + paths = gemini_installation.preflight_gemini_installation( + fixture.config, document + ) + rendered = render_gemini_document( + document, + None, + { + "BeforeTool": build_gemini_group( + paths.guard_path, GUARD_HOOK_NAME + ), + "AfterTool": build_gemini_group( + paths.audit_path, AUDIT_HOOK_NAME + ), + }, + ) + return gemini_installation.prepare_gemini_installation( + fixture.config, paths, rendered + ) + + +def assert_no_transaction_files(root: Path) -> None: + assert not list(root.rglob("*.tmp")) + assert not list(root.rglob("*.rollback")) + + +def test_transaction_rolls_back_all_five_files( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + original = '{"theme":"GitHub"}\n' + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=original + ) + real_replace = _transaction.os.replace + failed = False + + def fail_settings_once(source: str, destination: str) -> None: + nonlocal failed + if os.path.abspath(destination) == str(fixture.config_path) and not failed: + failed = True + raise OSError("injected Gemini settings failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_settings_once) + + with pytest.raises(OSError, match="injected Gemini settings failure"): + fixture.install() + + assert fixture.source() == original + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_concurrent_settings_change_is_detected_before_runtime_commit( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings={"theme": "GitHub"} + ) + changes = prepare_installation(fixture) + concurrent = '{"theme":"Atom One","hooks":{"Notification":[]}}\n' + write_text(fixture.config_path, concurrent) + + with pytest.raises(OSError, match="changed during installation"): + gemini_installation.commit_gemini_installation(changes) + + assert fixture.source() == concurrent + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + 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_fail_before_settings_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + runtime_name: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + artifact = fixture.agent_dir / runtime_name + write_text(artifact, "orphan\n") + + with pytest.raises(ValueError, match="identity cannot be verified"): + fixture.install() + + assert artifact.read_text(encoding="utf-8") == "orphan\n" + assert not fixture.config_path.exists() + + +def test_mismatched_runtime_identity_fails_before_writes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + write_json(fixture.runtime_config_path, { + "agent_id": "another-agent", "agent_name": "gemini" + }) + original = fixture.runtime_config_path.read_text(encoding="utf-8") + + with pytest.raises(ValueError, match="identity does not match"): + fixture.install() + + assert fixture.runtime_config_path.read_text(encoding="utf-8") == original + assert not fixture.config_path.exists() + + +@pytest.mark.parametrize("kind", ["configuration", "settings", "runtime"]) +def test_linked_configuration_and_runtime_paths_are_rejected( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + target = tmp_path / f"{kind}-target" + if kind == "settings": + write_text(target, "{}\n") + fixture.config_path.parent.mkdir(parents=True) + symlink_or_skip(target, fixture.config_path) + else: + target.mkdir() + link = ( + fixture.config_path.parent + if kind == "configuration" + else fixture.agent_dir.parent + ) + if kind == "runtime" and link.exists(): + shutil.rmtree(link) + link.parent.mkdir(parents=True, exist_ok=True) + symlink_or_skip(target, link, directory=True) + + with pytest.raises(OSError, match="physical (file|directory)"): + fixture.install() + + +@pytest.mark.parametrize( + ("field", "value", "pattern"), + [ + ("agent_name", "codex", "requires agent_name gemini"), + ("agent_id", "../escape", "Invalid agent ID"), + ("private_key", "invalid", "canonical 32-byte"), + ("token", "", "non-empty string"), + ("base_url", "https://api.elydora.com?q=1", "query parameters"), + ("guard_script_path", "outside", "managed agent directory"), + ], +) +def test_install_validates_runtime_inputs_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + field: str, + value: str, + pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config[field] = value + + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() + + +def test_status_surfaces_malformed_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + write_text(fixture.runtime_config_path, "{ malformed") + + with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.plugin.status() + + +def test_installation_leaves_no_transaction_artifacts( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert_no_transaction_files(fixture.home_dir) + + +def test_cli_preflight_preserves_malformed_settings(tmp_path: Path) -> None: + home_dir = tmp_path / "cli-home" + settings_path = home_dir / ".gemini" / "settings.json" + key_path = tmp_path / "private.key" + write_text(settings_path, "{ malformed") + write_text(key_path, VALID_PRIVATE_KEY) + environment = { + **os.environ, + "HOME": str(home_dir), + "USERPROFILE": str(home_dir), + "GEMINI_CLI_HOME": str(home_dir), + } + + result = subprocess.run( # nosec B603 + [ + sys.executable, "-m", "elydora.cli", "install", + "--agent", "gemini", "--org_id", "org-1", + "--agent_id", AGENT_ID, "--private_key_file", str(key_path), + "--kid", "key-1", "--base_url", "https://api.elydora.com", + ], + capture_output=True, + check=False, + cwd=Path(__file__).parents[1], + env=environment, + text=True, + ) + + assert result.returncode == 1 + assert "parse Gemini CLI user settings" in result.stderr + assert settings_path.read_text(encoding="utf-8") == "{ malformed" + assert not (home_dir / ".elydora").exists() diff --git a/sdks/python/tests/test_settings_hook_detection.py b/sdks/python/tests/test_settings_hook_detection.py index b6b66b4..9633a7f 100644 --- a/sdks/python/tests/test_settings_hook_detection.py +++ b/sdks/python/tests/test_settings_hook_detection.py @@ -1,22 +1,15 @@ from __future__ import annotations -from collections.abc import Callable - -import pytest - -from elydora.plugins.gemini import _is_elydora_hook as is_gemini_hook from elydora.plugins.letta import _is_elydora_hook as is_letta_hook -HookDetector = Callable[[dict, str], bool] - - -@pytest.mark.parametrize("detector", [is_gemini_hook, is_letta_hook]) -def test_hook_detection_handles_only_string_commands(detector: HookDetector) -> None: - assert detector({"command": 42}, "") is False - assert detector({"hooks": [{"command": None}]}, "") is False - assert detector({"hooks": [{"command": "run .elydora/agent-1/hook.py"}]}, "") is True - assert detector( +def test_hook_detection_handles_only_string_commands() -> None: + assert is_letta_hook({"command": 42}, "") is False + assert is_letta_hook({"hooks": [{"command": None}]}, "") is False + assert is_letta_hook( + {"hooks": [{"command": "run .elydora/agent-1/hook.py"}]}, "" + ) is True + assert is_letta_hook( {"hooks": [{"command": "run .elydora/agent-1/hook.py"}]}, "agent-2", ) is False