diff --git a/AGENTS.md b/AGENTS.md index 47fe3e3..6fa13f7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -25,7 +25,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a - Write Cursor hooks only to `~/.cursor/hooks.json`; keep project and enterprise sources read-only. Preserve user hooks, migrate the prior versionless Elydora contract, and own exact native `preToolUse`, `postToolUse`, and `postToolUseFailure` commands. Forward native success and failure payloads, retain PowerShell exit code `2`, set every handler to `failClosed` with a bounded timeout, and commit the guard, audit script, runtime identity, private key, and hook configuration as one rollback-capable transaction. - Model stable, legacy, and early-access hook generations as explicit contracts. Keep their activation requirements visible in CLI output and README guidance. - Resolve stable Kimi hooks through `$KIMI_CODE_HOME/config.toml` with `~/.kimi-code/config.toml` as the default, and treat an empty override as the default. Activate `~/.kimi/config.toml` only when the legacy home exists. Register exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` rules with ten-second commands; stable Kimi supports its current sixteen-event schema and legacy kimi-cli supports its thirteen-event schema. Preserve TOML comments and array style, then parse the complete rendered document before staging it. Preserve the native snake_case payload, propagate exit code `2`, and keep audit delivery fail-open with owner-only error logs. Use encoded PowerShell commands on Windows and exact two-argument shell commands on Unix. Commit every detected config, generated runtime, runtime config, and private key through one rollback-capable transaction. -- Keep Grok Build writes inside its native global `$GROK_HOME/hooks/*.json` contract. Treat Claude Code and Cursor compatibility files plus project `.grok/hooks` as read-only integration sources. +- Resolve Grok Build user hooks through `$GROK_HOME/hooks/*.json` with `~/.grok/hooks/*.json` as the default, and treat an empty override as the default. Keep Claude Code, Cursor, project `.grok/hooks`, plugin, and `hooks-paths` sources read-only. Register exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` groups with ten-second command handlers, preserve the complete native camelCase payload, and emit Grok's documented deny JSON plus exit code `2` for frozen or revoked agents. Use encoded PowerShell commands on Windows and exact two-argument shell commands on Unix. Commit Grok's hook file, generated runtimes, runtime config, and private key through one rollback-capable transaction. - Write Auggie hooks only to `~/.augment/settings.json`; keep system and workspace settings read-only. Generate `.cmd` wrappers on Windows and `.sh` wrappers on Unix because Auggie dispatches supported script paths, and express hook timeouts in milliseconds. - Validate Auggie matcher syntax during installation with Node.js `new RegExp`; keep status and uninstall independent from the JavaScript validator so recovery remains available offline. - Write Cline hooks only to `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default; keep Documents and workspace hook roots read-only. Preserve official input byte-for-byte, resolve Python runtimes from their absolute shebangs, and translate guard exit code `2` into Cline's JSON stdout cancellation control. diff --git a/README.md b/README.md index 852bc1b..e009e2f 100644 --- a/README.md +++ b/README.md @@ -86,7 +86,7 @@ Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/ 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. -Grok Build installation writes native global hooks to `$GROK_HOME/hooks/elydora-audit.json` (default `~/.grok/hooks/elydora-audit.json`). Project hooks still follow Grok's `/hooks-trust` workflow; Elydora leaves project, Claude Code, and Cursor compatibility files unchanged. +Grok Build installation writes native global `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` hooks to `$GROK_HOME/hooks/elydora-audit.json` (default `~/.grok/hooks/elydora-audit.json`). The hook file, generated runtimes, runtime config, and private key commit as one rollback-capable transaction. Managed commands preserve Grok's native camelCase JSON, return `{"decision":"deny","reason":"..."}` with exit code `2` for frozen or revoked agents, and use encoded PowerShell on Windows. Project hooks follow Grok's `/hooks-trust` workflow; Elydora leaves project, plugin, `hooks-paths`, Claude Code, and Cursor compatibility sources unchanged. Run `grok inspect --json` to inspect the effective hook inventory. Auggie installation writes user-level hooks to `~/.augment/settings.json` and creates the `.cmd` or `.sh` wrappers required by its command runner. System and workspace settings remain unchanged. Sessions started with `--augment-cache-dir` load settings from that alternate directory. diff --git a/elydora/plugins/_strict_json.py b/elydora/plugins/_strict_json.py index 80f1bf7..c94f503 100644 --- a/elydora/plugins/_strict_json.py +++ b/elydora/plugins/_strict_json.py @@ -3,7 +3,7 @@ from __future__ import annotations import json -from typing import Any, Dict, List, Tuple +from typing import Any, Dict, List, NoReturn, Tuple JsonObject = Dict[str, Any] @@ -13,6 +13,10 @@ class _DuplicateKey(ValueError): pass +class _InvalidConstant(ValueError): + pass + + def _unique_object(pairs: List[Tuple[str, Any]]) -> JsonObject: result: JsonObject = {} for key, value in pairs: @@ -22,10 +26,18 @@ def _unique_object(pairs: List[Tuple[str, Any]]) -> JsonObject: return result +def _reject_constant(value: str) -> NoReturn: + raise _InvalidConstant(f'invalid numeric constant "{value}"') + + 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: + value = json.loads( + raw, + object_pairs_hook=_unique_object, + parse_constant=_reject_constant, + ) + except (_DuplicateKey, _InvalidConstant, 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") diff --git a/elydora/plugins/_transaction.py b/elydora/plugins/_transaction.py index e881f1d..d9cb3dc 100644 --- a/elydora/plugins/_transaction.py +++ b/elydora/plugins/_transaction.py @@ -9,6 +9,7 @@ from typing import List, Optional, Sequence from uuid import uuid4 from ._managed_files import ( + FileSnapshot, MAX_SOURCE_BYTES, ensure_physical_directory, physical_file_exists, @@ -34,6 +35,7 @@ class _StagedChange: change: FileChange temporary_path: Optional[str] rollback_path: Optional[str] + committed_snapshot: Optional[FileSnapshot] = None committed: bool = False @@ -56,6 +58,13 @@ def source_change( ) -> Optional[FileChange]: if original == next_source: return None + if ( + next_source is not None + and len(next_source.encode("utf-8")) > maximum_bytes + ): + raise ValueError( + f"{label} exceeds {maximum_bytes} bytes: {file_path}" + ) snapshot = read_physical_file(file_path, label, maximum_bytes) current = None if snapshot is None else snapshot.contents if current != original: @@ -212,17 +221,80 @@ def _commit(staged: _StagedChange) -> None: raise OSError(f"Missing staged file for {staged.change.label}") os.replace(staged.temporary_path, staged.change.file_path) staged.committed = True + current = read_physical_file( + staged.change.file_path, + staged.change.label, + staged.change.maximum_bytes, + ) + if staged.change.next_source is None: + if current is not None: + raise OSError( + f"{staged.change.label} remained after removal: " + f"{staged.change.file_path}" + ) + return + if current is None or current.contents != staged.change.next_source: + raise OSError( + f"{staged.change.label} changed immediately after commit: " + f"{staged.change.file_path}" + ) + staged.committed_snapshot = current + + +def _assert_committed_unchanged(staged: _StagedChange) -> None: + current = read_physical_file( + staged.change.file_path, + staged.change.label, + staged.change.maximum_bytes, + ) + if staged.change.next_source is None: + if current is not None: + raise OSError( + f"{staged.change.label} changed during transaction recovery: " + f"{staged.change.file_path}" + ) + return + committed = staged.committed_snapshot + if current is None or committed is None or ( + current.contents != committed.contents + or current.device != committed.device + or current.inode != committed.inode + ): + raise OSError( + f"{staged.change.label} changed during transaction recovery: " + f"{staged.change.file_path}" + ) + + +def _preserve_rollback(staged: _StagedChange, error: Exception) -> OSError: + if staged.rollback_path is None: + return error if isinstance(error, OSError) else OSError(str(error)) + rollback_path = staged.rollback_path + staged.rollback_path = None + return OSError( + f"{error}; original content preserved at {rollback_path}" + ) def _rollback(staged: _StagedChange) -> None: if not staged.committed: return - if staged.change.original is None: - _remove_optional(staged.change.file_path) - return - if staged.rollback_path is None: - raise OSError(f"Missing rollback data for {staged.change.label}") - os.replace(staged.rollback_path, staged.change.file_path) + try: + _assert_committed_unchanged(staged) + except Exception as error: + raise _preserve_rollback(staged, error) from error + try: + if staged.change.next_source is None or staged.change.original is not None: + if staged.rollback_path is None: + raise OSError( + f"Missing rollback data for {staged.change.label}" + ) + os.replace(staged.rollback_path, staged.change.file_path) + staged.rollback_path = None + else: + _remove_optional(staged.change.file_path) + except Exception as error: + raise _preserve_rollback(staged, error) from error def _cleanup(staged: _StagedChange) -> None: @@ -248,7 +320,7 @@ def write_changes(changes: Sequence[FileChange], label: str) -> None: for item in reversed(staged): try: _rollback(item) - except OSError as rollback_error: + except Exception as rollback_error: recovery_errors.append(str(rollback_error)) for item in staged: try: diff --git a/elydora/plugins/cursor.py b/elydora/plugins/cursor.py index ef0e977..0770700 100644 --- a/elydora/plugins/cursor.py +++ b/elydora/plugins/cursor.py @@ -178,6 +178,7 @@ class CursorPlugin(AgentPlugin): agent_id, success_output='{"permission":"allow"}\n', fail_closed=True, + deny_protocol="cursor", ) changes = _present([ runtime_change( diff --git a/elydora/plugins/grok.py b/elydora/plugins/grok.py index acdbe3f..0c6834d 100644 --- a/elydora/plugins/grok.py +++ b/elydora/plugins/grok.py @@ -2,425 +2,88 @@ from __future__ import annotations -from dataclasses import dataclass -import math -import os -import shlex -# subprocess provides argument quoting only; this module starts no process. -import subprocess # nosec B404 -import sys -from typing import Any, Dict, List, Optional, Tuple - -from ._file_io import ( - read_json as _read_json, - regular_file_exists as _regular_file_exists, - remove_file as _remove_file, - require_runtime as _require_runtime, - write_json_atomic as _write_json_atomic, - write_text_atomic as _write_text_atomic, -) from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script +from .grok_contract import ( + AGENT_KEY, + GrokHooks, + build_grok_group, + grok_runtime_contracts, + remove_managed_grok_hooks, + render_grok_document, +) +from .grok_installation import ( + commit_grok_installation, + commit_grok_uninstall, + preflight_grok_installation, + prepare_grok_installation, + prepare_grok_uninstall, +) +from .grok_io import grok_runtime_files_exist, read_grok_document -AGENT_KEY = "grok" -GUARD_SCRIPT = "guard.py" -AUDIT_SCRIPT = "hook.py" -HOOK_TIMEOUT_SECONDS = 10 -CONFIG_FILE = "elydora-audit.json" -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") - -JsonObject = Dict[str, Any] -GrokHooks = Dict[str, List[JsonObject]] - - -@dataclass(frozen=True) -class GrokDocument: - exists: bool - config_path: str - root: JsonObject - hooks: GrokHooks - - -@dataclass(frozen=True) -class RuntimeContract: - agent_id: str - guard_path: str - audit_path: str - - -def _home_dir() -> str: - return os.path.expanduser("~") - - -def _resolve_config_path() -> str: - grok_home = os.environ.get("GROK_HOME") or os.path.join(_home_dir(), ".grok") - return os.path.join(grok_home, "hooks", CONFIG_FILE) - - -def _build_command(script_path: str) -> str: - arguments = [sys.executable, script_path] - return ( - subprocess.list2cmdline(arguments) - if os.name == "nt" - else shlex.join(arguments) - ) - - -def _build_handler(script_path: str) -> JsonObject: +def _installed_hooks( + hooks: GrokHooks, guard_path: str, audit_path: str +) -> GrokHooks: + cleaned = remove_managed_grok_hooks(hooks) return { - "type": "command", - "command": _build_command(script_path), - "timeout": HOOK_TIMEOUT_SECONDS, + **cleaned, + "PreToolUse": [ + *cleaned.get("PreToolUse", []), + build_grok_group(guard_path), + ], + "PostToolUse": [ + *cleaned.get("PostToolUse", []), + build_grok_group(audit_path), + ], + "PostToolUseFailure": [ + *cleaned.get("PostToolUseFailure", []), + build_grok_group(audit_path), + ], } -def _parse_generated_command(command: str) -> Optional[Tuple[str, str]]: - try: - arguments = shlex.split(command, posix=os.name != "nt") - except ValueError: - return None - if os.name == "nt": - arguments = [ - value[1:-1] - if len(value) >= 2 and value.startswith('"') and value.endswith('"') - else value - for value in arguments - ] - if len(arguments) != 2 or not arguments[0] or not arguments[1]: - return None - return arguments[0], arguments[1] - - -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 _managed_agent_id(handler: JsonObject, script_name: str) -> Optional[str]: - if ( - handler.get("type") != "command" - or handler.get("timeout") != HOOK_TIMEOUT_SECONDS - or not isinstance(handler.get("command"), str) - ): - return None - parsed = _parse_generated_command(str(handler["command"])) - if parsed is None: - return None - script_path = parsed[1] - if os.path.normcase(os.path.basename(script_path)) != os.path.normcase(script_name): - return None - agent_dir = os.path.dirname(script_path) - if not _same_path(os.path.dirname(agent_dir), ELYDORA_DIR): - return None - agent_id = os.path.basename(agent_dir) - return agent_id if agent_id not in {"", ".", ".."} else None - - -def _validate_handler( - value: Any, - event: str, - group_index: int, - handler_index: int, -) -> JsonObject: - label = ( - f"Grok hooks config handler hooks.{event}[{group_index}]" - f".hooks[{handler_index}]" - ) - if not isinstance(value, dict): - raise ValueError(f"{label} must be an object") - handler = dict(value) - handler_type = handler.get("type") - if handler_type not in ("command", "http"): - raise ValueError(f'{label} has unsupported type "{handler_type}"') - if handler_type == "command" and ( - not isinstance(handler.get("command"), str) or not handler["command"] - ): - raise ValueError(f"{label} requires a non-empty command") - if handler_type == "http" and ( - not isinstance(handler.get("url"), str) or not handler["url"] - ): - raise ValueError(f"{label} requires a non-empty url") - timeout = handler.get("timeout") - if timeout is not None and ( - isinstance(timeout, bool) - or not isinstance(timeout, (int, float)) - or not math.isfinite(timeout) - or timeout <= 0 - ): - raise ValueError(f"{label} timeout must be a positive number") - return handler - - -def _validate_group(value: Any, event: str, group_index: int) -> JsonObject: - label = f"Grok hooks config group hooks.{event}[{group_index}]" - if not isinstance(value, dict): - raise ValueError(f"{label} must be an object") - group = dict(value) - matcher = group.get("matcher") - if "matcher" in group and not isinstance(matcher, str): - raise ValueError(f"{label} matcher must be a string") - handlers = group.get("hooks") - if not isinstance(handlers, list): - raise ValueError(f"{label} must contain a hooks array") - group["hooks"] = [ - _validate_handler(handler, event, group_index, handler_index) - for handler_index, handler in enumerate(handlers) - ] - return group - - -def _read_hooks(root: JsonObject) -> GrokHooks: - if "hooks" not in root: - return {} - raw_hooks = root["hooks"] - if not isinstance(raw_hooks, dict): - raise ValueError('Grok hooks config field "hooks" must be an object') - hooks: GrokHooks = {} - for event, groups in raw_hooks.items(): - if not isinstance(groups, list): - raise ValueError( - f'Grok hooks config field "hooks.{event}" must be an array' - ) - hooks[event] = [ - _validate_group(group, event, group_index) - for group_index, group in enumerate(groups) - ] - return hooks - - -def _read_config() -> GrokDocument: - config_path = _resolve_config_path() - root = _read_json(config_path, "Grok hooks config") - if root is None: - return GrokDocument(False, config_path, {}, {}) - return GrokDocument(True, config_path, root, _read_hooks(root)) - - -def _remove_managed( - groups: List[JsonObject], - script_name: str, - agent_id: str = "", -) -> Tuple[List[JsonObject], bool]: - result: List[JsonObject] = [] - changed = False - for group in groups: - if "matcher" in group: - result.append(group) - continue - handlers = group["hooks"] - kept: List[JsonObject] = [] - group_changed = False - for handler in handlers: - managed_id = _managed_agent_id(handler, script_name) - remove = managed_id is not None and ( - not agent_id or _same_agent_id(managed_id, agent_id) - ) - if remove: - changed = True - group_changed = True - else: - kept.append(handler) - if not group_changed: - result.append(group) - elif kept: - result.append({**group, "hooks": kept}) - return result, changed - - -def _remove_managed_hooks( - hooks: GrokHooks, - agent_id: str = "", -) -> Tuple[GrokHooks, bool]: - result = dict(hooks) - changed = False - for event, script_name in ( - ("PreToolUse", GUARD_SCRIPT), - ("PostToolUse", AUDIT_SCRIPT), - ): - groups, event_changed = _remove_managed( - result.get(event, []), script_name, agent_id - ) - if not event_changed: - continue - changed = True - if groups: - result[event] = groups - else: - result.pop(event, None) - return result, changed - - -def _managed_ids(groups: List[JsonObject], script_name: str) -> Dict[str, str]: - result: Dict[str, str] = {} - for group in groups: - if "matcher" in group: - continue - for handler in group["hooks"]: - agent_id = _managed_agent_id(handler, script_name) - if agent_id: - result[os.path.normcase(agent_id)] = agent_id - return result - - -def _runtime_contracts(hooks: GrokHooks) -> List[RuntimeContract]: - guards = _managed_ids(hooks.get("PreToolUse", []), GUARD_SCRIPT) - audits = _managed_ids(hooks.get("PostToolUse", []), AUDIT_SCRIPT) - contracts: List[RuntimeContract] = [] - for normalized_id in sorted(set(guards) & set(audits)): - agent_id = guards[normalized_id] - contracts.append(RuntimeContract( - agent_id=agent_id, - guard_path=os.path.join(ELYDORA_DIR, agent_id, GUARD_SCRIPT), - audit_path=os.path.join(ELYDORA_DIR, agent_id, AUDIT_SCRIPT), - )) - return contracts - - -def _runtime_scripts_exist(contracts: List[RuntimeContract]) -> bool: - 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}") from error - - for contract in contracts: - entry = next( - ( - candidate - for candidate in entries - if _same_agent_id(candidate.name, contract.agent_id) - ), - None, - ) - if entry is None: - continue - try: - if not entry.is_dir(follow_symlinks=False): - continue - except OSError as error: - raise OSError(f"Read Elydora runtime entry at {entry.path}: {error}") from error - runtime_config_path = os.path.join(entry.path, "config.json") - runtime_config = _read_json(runtime_config_path, "Elydora runtime config") - if runtime_config is None: - continue - agent_name = runtime_config.get("agent_name") - if not isinstance(agent_name, str): - raise ValueError( - f'Elydora runtime config at {runtime_config_path} field ' - '"agent_name" must be a string' - ) - if agent_name != AGENT_KEY: - continue - guard_path = os.path.join(entry.path, GUARD_SCRIPT) - audit_path = os.path.join(entry.path, AUDIT_SCRIPT) - return _regular_file_exists( - guard_path, "Elydora guard runtime" - ) and _regular_file_exists(audit_path, "Elydora audit runtime") - return False - - class GrokPlugin(AgentPlugin): """Install Elydora into Grok Build's native global user hooks.""" + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + document = read_grok_document() + preflight_grok_installation(config, document) + def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - source = _read_config() - guard_path = config.get("guard_script_path", "") - _require_runtime(guard_path, "Elydora guard runtime") - agent_dir = os.path.join(ELYDORA_DIR, agent_id) - audit_path = os.path.join(agent_dir, AUDIT_SCRIPT) - cleaned, _ = _remove_managed_hooks(source.hooks) - hooks: GrokHooks = { - **cleaned, - "PreToolUse": [ - *cleaned.get("PreToolUse", []), - {"hooks": [_build_handler(guard_path)]}, - ], - "PostToolUse": [ - *cleaned.get("PostToolUse", []), - {"hooks": [_build_handler(audit_path)]}, - ], - } - 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": AGENT_KEY, - } - 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"), + document = read_grok_document() + paths = preflight_grok_installation(config, document) + rendered = render_grok_document( + document, + _installed_hooks(document.hooks, paths.guard_path, paths.audit_path), ) - _write_json_atomic( - os.path.join(agent_dir, "config.json"), - runtime_config, - 0o600, - "Elydora runtime config", + changes = prepare_grok_installation(config, paths, rendered) + commit_grok_installation(changes) + print( + "Grok Build: global PreToolUse, PostToolUse, " + "and PostToolUseFailure hooks installed." ) - _write_text_atomic( - os.path.join(agent_dir, "private.key"), - config.get("private_key", ""), - 0o600, - "Elydora private key", - ) - _write_text_atomic(audit_path, audit_script, 0o700, "Elydora audit runtime") - _write_json_atomic( - source.config_path, - {**source.root, "hooks": hooks}, - 0o600, - "Grok hooks config", - ) - print("Grok Build: global PreToolUse and PostToolUse hooks installed.") def uninstall(self, agent_id: str = "") -> None: - source = _read_config() - if not source.exists: - return - hooks, changed = _remove_managed_hooks(source.hooks, agent_id) - if not changed: - return - root = dict(source.root) - if hooks: - root["hooks"] = hooks - else: - root.pop("hooks", None) - if root: - _write_json_atomic( - source.config_path, - root, - 0o600, - "Grok hooks config", - ) - else: - _remove_file(source.config_path, "Grok hooks config") + document = read_grok_document() + rendered = render_grok_document( + document, + remove_managed_grok_hooks(document.hooks, agent_id), + ) + commit_grok_uninstall(prepare_grok_uninstall(rendered)) def status(self) -> PluginStatus: - source = _read_config() - contracts = _runtime_contracts(source.hooks) - if not contracts: - return PluginStatus( - installed=False, - agent=AGENT_KEY, - details="Not installed", - ) - installed = _runtime_scripts_exist(contracts) + document = read_grok_document() + contracts = grok_runtime_contracts(document.hooks) + installed = bool(contracts) and grok_runtime_files_exist(contracts) details = ( - f"Config: {source.config_path}" + f"Config: {document.config_path}" if installed - else f"Configured at {source.config_path}; runtime scripts missing" + else f"Configured at {document.config_path}; managed contract incomplete" + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details if contracts else "Not installed", ) - return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) diff --git a/elydora/plugins/grok_command.py b/elydora/plugins/grok_command.py new file mode 100644 index 0000000..66d3fae --- /dev/null +++ b/elydora/plugins/grok_command.py @@ -0,0 +1,232 @@ +"""Grok hook command rendering and exact runtime ownership.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +import ntpath +import os +import re +import shlex +import subprocess # nosec B404 - used only for legacy command rendering +import sys +from typing import Optional, Tuple + + +@dataclass(frozen=True) +class GrokRuntimeReference: + agent_id: str + script_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_grok_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def same_grok_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_grok_command(script_path: str) -> str: + if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): + raise ValueError( + "Grok 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[_ParsedArgument]: + if start >= len(command) or command[start] != "'": + return None + apostrophe = "'\"'\"'" + value = "" + index = start + 1 + while index < len(command): + if command.startswith(apostrophe, index): + value += "'" + index += len(apostrophe) + continue + if command[index] == "'": + return _ParsedArgument(value, index + 1) + value += command[index] + index += 1 + return None + + +def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: + executable = _read_posix_argument(command, 0) + if executable is None or command[executable.next_index : executable.next_index + 1] != " ": + return None + script = _read_posix_argument(command, executable.next_index + 1) + if script is None or script.next_index != len(command): + return None + return executable.value, script.value + + +def _read_powershell_argument( + command: str, start: int +) -> Optional[_ParsedArgument]: + if start >= len(command) or command[start] != "'": + return None + value = "" + index = start + 1 + while index < len(command): + if command[index] != "'": + value += command[index] + index += 1 + continue + if index + 1 < len(command) and command[index + 1] == "'": + value += "'" + index += 2 + continue + return _ParsedArgument(value, index + 1) + return None + + +def _parse_powershell_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: str) -> Optional[Tuple[str, str]]: + 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_posix_command(command: str) -> Optional[Tuple[str, str]]: + try: + arguments = shlex.split(command, posix=True) + except ValueError: + return None + if len(arguments) != 2 or shlex.join(arguments) != command: + return None + return arguments[0], arguments[1] + + +def _parse_legacy_windows_command(command: str) -> Optional[Tuple[str, str]]: + match = re.fullmatch( + r'(?:(?:"([^"\r\n]+)")|([^\s"\r\n]+)) ' + r'(?:(?:"([^"\r\n]+)")|([^\s"\r\n]+))', + command, + ) + if match is None: + return None + arguments = [match.group(1) or match.group(2), match.group(3) or match.group(4)] + if subprocess.list2cmdline(arguments) != command: + 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.IGNORECASE, + ) + is not None + ) + + +def grok_runtime_reference( + command: str, script_name: str +) -> Optional[GrokRuntimeReference]: + parsed = ( + _parse_posix_command(command) + or _parse_windows_command(command) + or _parse_legacy_posix_command(command) + or _parse_legacy_windows_command(command) + ) + if parsed is None: + return None + executable, script_path = parsed + if ( + not os.path.isabs(executable) + or not os.path.isabs(script_path) + or not _is_python_executable(executable) + or os.path.basename(script_path) != script_name + ): + return None + agent_directory = os.path.dirname(script_path) + if not same_grok_path(os.path.dirname(agent_directory), runtime_root()): + return None + agent_id = os.path.basename(agent_directory) + if agent_id in {"", ".", ".."}: + return None + return GrokRuntimeReference(agent_id, script_path) diff --git a/elydora/plugins/grok_contract.py b/elydora/plugins/grok_contract.py new file mode 100644 index 0000000..4076dcf --- /dev/null +++ b/elydora/plugins/grok_contract.py @@ -0,0 +1,296 @@ +"""Grok hook schema, exact ownership, and JSON rendering.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +from typing import Any, Dict, List, Optional, Tuple + +from ._strict_json import JsonObject, parse_json_object +from .grok_command import ( + GrokRuntimeReference, + build_grok_command, + grok_runtime_reference, + same_grok_agent_id, + same_grok_path, +) + + +AGENT_KEY = "grok" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_SECONDS = 10 +MATCHER_REJECTING_EVENTS = frozenset( + {"SessionStart", "SessionEnd", "Stop", "UserPromptSubmit"} +) + +GrokHooks = Dict[str, List[JsonObject]] + + +@dataclass(frozen=True) +class GrokDocument: + exists: bool + config_path: str + root: JsonObject + hooks: GrokHooks + raw: Optional[str] = None + + +@dataclass(frozen=True) +class RenderedGrokDocument: + document: GrokDocument + changed: bool + next_source: Optional[str] = None + + +@dataclass(frozen=True) +class GrokRuntimeContract: + agent_id: str + guard_path: str + audit_path: str + + +def _validate_handler( + value: Any, + event: str, + group_index: int, + handler_index: int, +) -> JsonObject: + label = ( + f"Grok user hooks handler hooks.{event}[{group_index}]" + f".hooks[{handler_index}]" + ) + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + handler = dict(value) + handler_type = handler.get("type") + if handler_type not in ("command", "http"): + raise ValueError(f'{label} has unsupported type "{handler_type}"') + if handler_type == "command" and ( + not isinstance(handler.get("command"), str) or not handler["command"] + ): + raise ValueError(f"{label} requires a non-empty command") + if handler_type == "http" and ( + not isinstance(handler.get("url"), str) or not handler["url"] + ): + raise ValueError(f"{label} requires a non-empty url") + timeout = handler.get("timeout") + if timeout is not None and (type(timeout) is not int or timeout < 0): + raise ValueError(f"{label} timeout must be a non-negative integer") + environment = handler.get("env") + if environment is not None and ( + 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") + return handler + + +def _validate_group(value: Any, event: str, group_index: int) -> JsonObject: + label = f"Grok user hooks group hooks.{event}[{group_index}]" + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + group = dict(value) + matcher = group.get("matcher") + if "matcher" in group and not isinstance(matcher, str): + raise ValueError(f"{label} matcher must be a string") + if "matcher" in group and event in MATCHER_REJECTING_EVENTS: + raise ValueError(f"{label} cannot declare a matcher for {event}") + handlers = group.get("hooks") + if not isinstance(handlers, list): + raise ValueError(f"{label} must contain a hooks array") + group["hooks"] = [ + _validate_handler(handler, event, group_index, handler_index) + for handler_index, handler in enumerate(handlers) + ] + return group + + +def _read_hooks(root: JsonObject) -> GrokHooks: + if "hooks" not in root: + return {} + value = root["hooks"] + if not isinstance(value, dict): + raise ValueError('Grok user hooks field "hooks" must be an object') + hooks: GrokHooks = {} + for event, groups in value.items(): + if not isinstance(groups, list): + raise ValueError( + f'Grok user hooks field "hooks.{event}" must be an array' + ) + hooks[event] = [ + _validate_group(group, event, index) + for index, group in enumerate(groups) + ] + return hooks + + +def parse_grok_document(config_path: str, raw: str) -> GrokDocument: + root = parse_json_object(raw, f"Grok user hooks at {config_path}") + return GrokDocument(True, config_path, root, _read_hooks(root), raw) + + +def create_grok_document(config_path: str) -> GrokDocument: + return GrokDocument(False, config_path, {}, {}) + + +def build_grok_group(script_path: str) -> JsonObject: + return { + "hooks": [{ + "type": "command", + "command": build_grok_command(script_path), + "timeout": HOOK_TIMEOUT_SECONDS, + }] + } + + +def _exact_managed_group(group: JsonObject) -> bool: + return set(group) == {"hooks"} + + +def _managed_reference( + handler: JsonObject, script_name: str +) -> Optional[GrokRuntimeReference]: + if ( + set(handler) != {"type", "command", "timeout"} + or handler.get("type") != "command" + or type(handler.get("timeout")) is not int + or handler.get("timeout") != HOOK_TIMEOUT_SECONDS + or not isinstance(handler.get("command"), str) + ): + return None + return grok_runtime_reference(str(handler["command"]), script_name) + + +def _managed_contract(event: str) -> Optional[Tuple[str, str]]: + if event == "PreToolUse": + return event, GUARD_SCRIPT + if event in ("PostToolUse", "PostToolUseFailure"): + return event, AUDIT_SCRIPT + return None + + +def _remove_from_groups( + groups: List[JsonObject], script_name: str, agent_id: str +) -> List[JsonObject]: + result: List[JsonObject] = [] + for group in groups: + if not _exact_managed_group(group): + result.append(group) + continue + kept = [] + for handler in group["hooks"]: + reference = _managed_reference(handler, script_name) + remove = reference is not None and ( + not agent_id or same_grok_agent_id(reference.agent_id, agent_id) + ) + if not remove: + kept.append(handler) + if kept: + result.append({"hooks": kept}) + return result + + +def remove_managed_grok_hooks( + hooks: GrokHooks, agent_id: str = "" +) -> GrokHooks: + result = {event: list(groups) for event, groups in hooks.items()} + for event, script_name in ( + ("PreToolUse", GUARD_SCRIPT), + ("PostToolUse", AUDIT_SCRIPT), + ("PostToolUseFailure", AUDIT_SCRIPT), + ): + groups = _remove_from_groups( + result.get(event, []), script_name, agent_id + ) + if groups: + result[event] = groups + else: + result.pop(event, None) + return result + + +def _entirely_managed(document: GrokDocument) -> bool: + if not document.exists or set(document.root) != {"hooks"}: + return False + if not document.hooks: + return False + count = 0 + for event, groups in document.hooks.items(): + contract = _managed_contract(event) + if contract is None or not groups: + return False + for group in groups: + if ( + not _exact_managed_group(group) + or not group["hooks"] + or any( + _managed_reference(handler, contract[1]) is None + for handler in group["hooks"] + ) + ): + return False + count += len(group["hooks"]) + return count > 0 + + +def render_grok_document( + document: GrokDocument, hooks: GrokHooks +) -> RenderedGrokDocument: + if not document.exists and not hooks: + return RenderedGrokDocument(document, False) + if not hooks and _entirely_managed(document): + return RenderedGrokDocument(document, True) + root = dict(document.root) + if hooks: + root["hooks"] = hooks + else: + root.pop("hooks", None) + next_source = json.dumps(root, indent=2, ensure_ascii=False) + "\n" + parse_grok_document(document.config_path, next_source) + return RenderedGrokDocument( + document, + next_source != document.raw, + next_source, + ) + + +def _references_for_event( + groups: List[JsonObject], script_name: str +) -> Dict[str, List[GrokRuntimeReference]]: + result: Dict[str, List[GrokRuntimeReference]] = {} + for group in groups: + if not _exact_managed_group(group): + continue + for handler in group["hooks"]: + reference = _managed_reference(handler, script_name) + if reference is not None: + result.setdefault( + os.path.normcase(reference.agent_id), [] + ).append(reference) + return result + + +def grok_runtime_contracts(hooks: GrokHooks) -> List[GrokRuntimeContract]: + guards = _references_for_event(hooks.get("PreToolUse", []), GUARD_SCRIPT) + successes = _references_for_event( + hooks.get("PostToolUse", []), AUDIT_SCRIPT + ) + failures = _references_for_event( + hooks.get("PostToolUseFailure", []), AUDIT_SCRIPT + ) + contracts = [] + for key, guard in guards.items(): + success = successes.get(key, []) + failure = failures.get(key, []) + if len(guard) != 1 or len(success) != 1 or len(failure) != 1: + continue + if not same_grok_path(success[0].script_path, failure[0].script_path): + continue + contracts.append(GrokRuntimeContract( + agent_id=guard[0].agent_id, + guard_path=guard[0].script_path, + audit_path=success[0].script_path, + )) + return contracts diff --git a/elydora/plugins/grok_installation.py b/elydora/plugins/grok_installation.py new file mode 100644 index 0000000..5367642 --- /dev/null +++ b/elydora/plugins/grok_installation.py @@ -0,0 +1,178 @@ +"""Transactional Grok 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, source_change, write_changes +from .base import InstallConfig +from .grok_command import runtime_root, same_grok_path +from .grok_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + GrokDocument, + RenderedGrokDocument, +) +from .grok_io import ( + validate_api_origin, + validate_private_key, + validate_runtime_tree, +) +from .hook_template import generate_guard_script, generate_hook_script + + +@dataclass(frozen=True) +class GrokRuntimePaths: + 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) -> GrokRuntimePaths: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + agent_directory = resolve_agent_directory(runtime_root(), agent_id) + guard_path = os.path.join(agent_directory, GUARD_SCRIPT) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + if not same_grok_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + "Elydora guard runtime must use the managed agent directory: " + f"{guard_path}" + ) + return GrokRuntimePaths(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"Grok 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_grok_installation( + config: InstallConfig, document: GrokDocument +) -> GrokRuntimePaths: + if not document.config_path: + raise ValueError("Grok installation requires a hook configuration 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 rendered_change( + rendered: RenderedGrokDocument, +) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.config_path, + "Grok user hooks", + rendered.document.raw, + rendered.next_source, + 0o600, + MAX_SOURCE_BYTES, + ) + + +def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def prepare_grok_installation( + config: InstallConfig, + paths: GrokRuntimePaths, + rendered: RenderedGrokDocument, +) -> List[FileChange]: + guard_script = generate_guard_script( + AGENT_KEY, paths.agent_id, deny_protocol="grok" + ) + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=paths.agent_id, + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + native_payload=True, + agent_name=AGENT_KEY, + ) + return _present([ + _runtime_change( + paths.guard_path, "Elydora guard runtime", guard_script, 0o700 + ), + _runtime_change( + os.path.join(paths.agent_directory, "config.json"), + "Elydora runtime config", + _json_source(_runtime_config(config, paths.agent_id)), + 0o600, + MAX_CONFIG_BYTES, + ), + _runtime_change( + os.path.join(paths.agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + _runtime_change( + paths.audit_path, "Elydora audit runtime", audit_script, 0o700 + ), + rendered_change(rendered), + ]) + + +def commit_grok_installation(changes: List[FileChange]) -> None: + write_changes(changes, "Install Grok hooks") + + +def prepare_grok_uninstall(rendered: RenderedGrokDocument) -> List[FileChange]: + return _present([rendered_change(rendered)]) + + +def commit_grok_uninstall(changes: List[FileChange]) -> None: + write_changes(changes, "Uninstall Grok hooks") diff --git a/elydora/plugins/grok_io.py b/elydora/plugins/grok_io.py new file mode 100644 index 0000000..ed188cd --- /dev/null +++ b/elydora/plugins/grok_io.py @@ -0,0 +1,218 @@ +"""Grok user-hook discovery and managed runtime I/O.""" + +from __future__ import annotations + +import base64 +import os +from typing import Any, List, Optional +import urllib.parse + +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, +) +from ._strict_json import JsonObject, parse_json_object +from .grok_command import runtime_root, same_grok_agent_id, same_grok_path +from .grok_contract import ( + AGENT_KEY, + GrokDocument, + GrokRuntimeContract, + create_grok_document, + parse_grok_document, +) + + +CONFIG_FILE = "elydora-audit.json" + + +def grok_config_path() -> str: + configured = os.environ.get("GROK_HOME") + grok_home = ( + os.path.join(os.path.expanduser("~"), ".grok") + if configured is None or configured == "" + else os.path.abspath(configured) + ) + return os.path.join(grok_home, "hooks", CONFIG_FILE) + + +def read_grok_document() -> GrokDocument: + config_path = grok_config_path() + hooks_directory = os.path.dirname(config_path) + physical_directory_exists( + os.path.dirname(hooks_directory), "Grok home directory" + ) + physical_directory_exists(hooks_directory, "Grok hooks directory") + snapshot = read_physical_file(config_path, "Grok user hooks") + if snapshot is None: + return create_grok_document(config_path) + return parse_grok_document(config_path, snapshot.contents) + + +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_grok_agent_id(agent_id, expected_agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + f"Elydora runtime identity does not match Grok hooks: {config_path}" + ) + if "token" in config: + _require_non_empty_string(config.get("token"), "token", config_path) + base_url = _require_non_empty_string( + config.get("base_url"), "base_url", config_path + ) + validate_api_origin(base_url, "Elydora runtime config base_url") + + +def _read_runtime_config(file_path: str) -> Optional[JsonObject]: + snapshot = read_physical_file( + file_path, "Elydora runtime config", MAX_CONFIG_BYTES + ) + if snapshot is None: + return None + return parse_json_object( + snapshot.contents, f"Elydora runtime config at {file_path}" + ) + + +def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: + root = runtime_root() + if not physical_directory_exists(root, "Elydora runtime directory"): + return + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return + config_path = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path) + artifacts = ( + ("private.key", "Elydora private key"), + ("guard.py", "Elydora guard runtime"), + ("hook.py", "Elydora audit runtime"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), + ) + artifact_states = [ + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ] + artifact_exists = any(artifact_states) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + configured_id = config.get("agent_id") + if ( + not isinstance(configured_id, str) + or not same_grok_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + f"Elydora runtime config identity does not match Grok agent {agent_id}: " + f"{config_path}" + ) + + +def _runtime_contract_exists(contract: GrokRuntimeContract) -> bool: + root = runtime_root() + agent_directory = os.path.dirname(contract.guard_path) + if ( + not same_grok_path(os.path.dirname(agent_directory), root) + or not same_grok_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 grok_runtime_files_exist(contracts: List[GrokRuntimeContract]) -> bool: + return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/elydora/plugins/guard_template.py b/elydora/plugins/guard_template.py index 13fcb13..a1a8e77 100644 --- a/elydora/plugins/guard_template.py +++ b/elydora/plugins/guard_template.py @@ -13,6 +13,7 @@ def generate_guard_script( success_output: str = "", *, fail_closed: bool = False, + deny_protocol: str = "", ) -> str: """Return a self-contained status guard for a command-hook provider.""" return f'''#!{sys.executable} @@ -32,6 +33,7 @@ AGENT_NAME = {agent_name!r} AGENT_ID = {agent_id!r} SUCCESS_OUTPUT = {success_output!r} FAIL_CLOSED = {fail_closed!r} +DENY_PROTOCOL = {deny_protocol!r} ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") CONFIG_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "config.json") STATUS_CACHE_PATH = os.path.join(ELYDORA_DIR, AGENT_ID, "status-cache.json") @@ -49,13 +51,19 @@ def validate_status(value, source): def block_agent(status): state = "frozen" if status == "frozen" else "revoked" message = 'Agent "' + AGENT_NAME + '" is ' + state + " in Elydora." - if SUCCESS_OUTPUT: + if DENY_PROTOCOL == "cursor": sys.stdout.write(json.dumps({{ "permission": "deny", "userMessage": message, "agentMessage": message + " Tool execution is blocked.", }}) + "\\n") sys.stdout.flush() + elif DENY_PROTOCOL == "grok": + sys.stdout.write(json.dumps({{ + "decision": "deny", + "reason": message, + }}) + "\\n") + sys.stdout.flush() sys.stderr.write("[Elydora guard] " + message + " Tool execution blocked.\\n") raise SystemExit(2) diff --git a/elydora/plugins/registry.py b/elydora/plugins/registry.py index de8e72f..e7c790b 100644 --- a/elydora/plugins/registry.py +++ b/elydora/plugins/registry.py @@ -48,7 +48,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "grok": { "name": "Grok Build", - "hook_event": "PreToolUse/PostToolUse", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", "config_path": "~/.grok/hooks/elydora-audit.json", }, "kirocli": { diff --git a/tests/grok_support.py b/tests/grok_support.py new file mode 100644 index 0000000..c9b9980 --- /dev/null +++ b/tests/grok_support.py @@ -0,0 +1,246 @@ +from __future__ import annotations + +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import shlex +import subprocess # nosec B404 +import sys +from threading import Thread +from typing import Any, Mapping + +import pytest + +from elydora.plugins.grok import GrokPlugin +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@dataclass +class GrokFixture: + root_dir: Path + home_dir: Path + project_dir: Path + grok_home: Path + config_path: Path + agent_dir: Path + guard_path: Path + hook_path: Path + runtime_config_path: Path + private_key_path: Path + plugin: GrokPlugin + config: dict[str, str] + + def install(self) -> None: + self.plugin.install(self.config) + + def settings(self) -> dict[str, Any]: + return json.loads(self.config_path.read_text(encoding="utf-8")) + + +class GrokApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + operation_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls) -> None: + cls.agent_status = "active" + cls.operation_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(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), + }) + status = type(self).operation_status + response: object = ( + {"operation": {"accepted": True}} + if 200 <= status < 300 + else {"error": {"code": "UPSTREAM_FAILURE", "message": "failed"}} + ) + self._respond(status, response) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server( + *, status: str = "active", operation_status: int = 201 +) -> ThreadingHTTPServer: + GrokApiHandler.reset() + GrokApiHandler.agent_status = status + GrokApiHandler.operation_status = operation_status + server = ThreadingHTTPServer(("127.0.0.1", 0), GrokApiHandler) + 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 prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_config: object = MISSING, + explicit_grok_home: bool = True, + base_url: str = "http://127.0.0.1:9", +) -> GrokFixture: + home_dir = tmp_path / "home with spaces and 'quote %GROK_HOOK_EVENT%" + project_dir = tmp_path / "project with spaces" + grok_home = ( + home_dir / "custom grok" + if explicit_grok_home + else home_dir / ".grok" + ) + config_path = grok_home / "hooks" / "elydora-audit.json" + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / "guard.py" + hook_path = agent_dir / "hook.py" + project_dir.mkdir(parents=True) + if existing_config is not MISSING: + source = ( + str(existing_config) + if isinstance(existing_config, str) + else json.dumps(existing_config, indent=2) + "\n" + ) + write_text(config_path, source) + + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.setenv("GROK_HOOK_EVENT", "injected-command-fragment") + if explicit_grok_home: + monkeypatch.setenv("GROK_HOME", str(grok_home)) + else: + monkeypatch.delenv("GROK_HOME", raising=False) + + return GrokFixture( + root_dir=tmp_path, + home_dir=home_dir, + project_dir=project_dir, + grok_home=grok_home, + config_path=config_path, + agent_dir=agent_dir, + guard_path=guard_path, + hook_path=hook_path, + runtime_config_path=agent_dir / "config.json", + private_key_path=agent_dir / "private.key", + plugin=GrokPlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "grok", + "private_key": VALID_PRIVATE_KEY, + "kid": "kid-1", + "token": "token-1", + "base_url": base_url, + "guard_script_path": str(guard_path), + }, + ) + + +def managed_handler(settings: Mapping[str, Any], event: str) -> dict[str, Any]: + for group in reversed(settings.get("hooks", {}).get(event, [])): + if set(group) != {"hooks"} or not isinstance(group["hooks"], list): + continue + for handler in reversed(group["hooks"]): + if ( + set(handler) == {"type", "command", "timeout"} + and handler.get("type") == "command" + and handler.get("timeout") == 10 + ): + return handler + raise AssertionError(f"managed {event} handler not found") + + +def assert_managed_handler(handler: Mapping[str, Any]) -> None: + assert set(handler) == {"type", "command", "timeout"} + assert handler["type"] == "command" + assert handler["timeout"] == 10 + command = str(handler["command"]) + if os.name == "nt": + assert "powershell.exe" in command.lower() + assert " -EncodedCommand " in command + else: + assert command.startswith("'") + + +def assert_managed_triple(settings: Mapping[str, Any]) -> None: + for event in ("PreToolUse", "PostToolUse", "PostToolUseFailure"): + assert_managed_handler(managed_handler(settings, event)) + assert managed_handler(settings, "PostToolUse")["command"] == managed_handler( + settings, "PostToolUseFailure" + )["command"] + + +def run_hook( + command: str, + payload: str, + fixture: GrokFixture, + environment: Mapping[str, str] | None = None, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + 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), + **dict(environment or {}), + }, + text=True, + ) + + +def legacy_command(script_path: Path) -> str: + arguments = [sys.executable, str(script_path)] + return ( + subprocess.list2cmdline(arguments) + if os.name == "nt" + else shlex.join(arguments) + ) + + +def assert_no_transaction_files(root: Path) -> None: + assert not list(root.rglob("*.tmp")) + assert not list(root.rglob("*.rollback")) diff --git a/tests/test_grok_plugin.py b/tests/test_grok_plugin.py index 4ac0b3e..c9f470e 100644 --- a/tests/test_grok_plugin.py +++ b/tests/test_grok_plugin.py @@ -1,141 +1,33 @@ 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 import cli from elydora.plugins import grok -from elydora.plugins.base import InstallConfig from elydora.plugins.registry import SUPPORTED_AGENTS - - -AGENT_ID = "agent-1" -MISSING = object() - - -@dataclass(frozen=True) -class GrokFixture: - plugin: grok.GrokPlugin - config: InstallConfig - home_dir: Path - grok_home: Path - agent_dir: Path - config_path: Path - guard_path: Path - hook_path: Path - runtime_config_path: Path - private_key_path: Path - - -def prepare_fixture( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - *, - existing_config: object = MISSING, - explicit_grok_home: bool = True, - create_guard: bool = True, -) -> GrokFixture: - home_dir = tmp_path / "home with spaces" - agent_dir = home_dir / ".elydora" / AGENT_ID - grok_home = ( - home_dir / "custom grok" - if explicit_grok_home - else home_dir / ".grok" - ) - config_path = grok_home / "hooks" / "elydora-audit.json" - guard_path = agent_dir / "guard.py" - hook_path = agent_dir / "hook.py" - agent_dir.mkdir(parents=True) - if create_guard: - guard_path.write_text( - "import sys\nsys.stdin.read()\n" - "sys.stderr.write('Agent is frozen by Elydora.')\n" - "raise SystemExit(2)\n", - encoding="utf-8", - ) - if existing_config is not MISSING: - config_path.parent.mkdir(parents=True, exist_ok=True) - raw = ( - str(existing_config) - if isinstance(existing_config, str) - else json.dumps(existing_config, indent=2) - ) - config_path.write_text(raw, encoding="utf-8") - - monkeypatch.setattr(grok, "ELYDORA_DIR", str(agent_dir.parent)) - monkeypatch.setattr(grok, "_home_dir", lambda: str(home_dir)) - if explicit_grok_home: - monkeypatch.setenv("GROK_HOME", str(grok_home)) - else: - monkeypatch.delenv("GROK_HOME", raising=False) - - config: InstallConfig = { - "agent_id": AGENT_ID, - "agent_name": "grok", - "org_id": "org-1", - "private_key": "test-key", - "kid": "kid-1", - "base_url": "https://api.elydora.test", - "guard_script_path": str(guard_path), - } - return GrokFixture( - plugin=grok.GrokPlugin(), - config=config, - home_dir=home_dir, - grok_home=grok_home, - agent_dir=agent_dir, - config_path=config_path, - guard_path=guard_path, - hook_path=hook_path, - runtime_config_path=agent_dir / "config.json", - private_key_path=agent_dir / "private.key", - ) - - -def managed_handler( - settings: dict[str, Any], event: str, script_name: str -) -> dict[str, Any]: - for group in settings["hooks"][event]: - if "matcher" in group: - continue - for handler in group["hooks"]: - if script_name in str(handler.get("command")): - return handler - raise AssertionError(f"managed {event} handler not found") - - -def run_command( - command: str, - home_dir: Path, - payload: str, -) -> subprocess.CompletedProcess[str]: - return subprocess.run( - command, - shell=True, - capture_output=True, - check=False, - env={**os.environ, "HOME": str(home_dir), "USERPROFILE": str(home_dir)}, - input=payload, - text=True, - ) +from grok_support import ( + AGENT_ID, + assert_managed_triple, + assert_no_transaction_files, + legacy_command, + prepare_fixture, +) def test_grok_is_registered_in_sdk_and_cli() -> None: assert SUPPORTED_AGENTS["grok"] == { "name": "Grok Build", - "hook_event": "PreToolUse/PostToolUse", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", "config_path": "~/.grok/hooks/elydora-audit.json", } assert cli.PLUGIN_MAP["grok"] is grok.GrokPlugin -def test_install_preserves_native_config_and_is_idempotent( +def test_install_writes_exact_triple_and_preserves_valid_user_hooks( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], @@ -144,17 +36,15 @@ def test_install_preserves_native_config_and_is_idempotent( "schemaVersion": 1, "hooks": { "SessionStart": [{ - "matcher": "startup", "hooks": [{ "type": "http", "url": "https://example.test/hook", - "timeout": 5, - "headers": {"x": "keep"}, + "timeout": 0, }], "label": "keep group metadata", }], "PreToolUse": [{ - "matcher": "Bash", + "matcher": "Bash|run_terminal_command", "hooks": [{ "type": "command", "command": "existing-command", @@ -163,36 +53,25 @@ def test_install_preserves_native_config_and_is_idempotent( }], }, } - fixture = prepare_fixture( - monkeypatch, tmp_path, existing_config=existing - ) + fixture = prepare_fixture(monkeypatch, tmp_path, existing_config=existing) - fixture.plugin.install(fixture.config) - fixture.plugin.install(fixture.config) + fixture.install() + first_source = fixture.config_path.read_text(encoding="utf-8") + fixture.install() - assert "global PreToolUse and PostToolUse hooks installed" in capsys.readouterr().out - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert "PostToolUseFailure hooks installed" in capsys.readouterr().out + settings = fixture.settings() + assert fixture.config_path.read_text(encoding="utf-8") == first_source assert settings["schemaVersion"] == 1 assert settings["hooks"]["SessionStart"] == existing["hooks"]["SessionStart"] assert settings["hooks"]["PreToolUse"][0] == existing["hooks"]["PreToolUse"][0] assert len(settings["hooks"]["PreToolUse"]) == 2 - assert len(settings["hooks"]["PostToolUse"]) == 1 - assert set(settings["hooks"]["PreToolUse"][1]) == {"hooks"} - for event, script_name in ( - ("PreToolUse", "guard.py"), - ("PostToolUse", "hook.py"), - ): - handler = managed_handler(settings, event, script_name) - assert set(handler) == {"type", "command", "timeout"} - assert handler["type"] == "command" - assert handler["timeout"] == 10 - assert json.loads(fixture.runtime_config_path.read_text())["agent_name"] == "grok" - assert fixture.private_key_path.read_text(encoding="utf-8") == "test-key" - assert (fixture.home_dir / ".claude" / "settings.json").exists() is False - assert (fixture.home_dir / ".cursor" / "hooks.json").exists() is False - assert ( + assert_managed_triple(settings) + assert not (fixture.home_dir / ".claude" / "settings.json").exists() + assert not (fixture.home_dir / ".cursor" / "hooks.json").exists() + assert not ( fixture.home_dir / ".grok" / "hooks" / "elydora-audit.json" - ).exists() is False + ).exists() def test_empty_home_override_uses_official_default( @@ -201,271 +80,229 @@ def test_empty_home_override_uses_official_default( fixture = prepare_fixture( monkeypatch, tmp_path, explicit_grok_home=False ) - fixture.plugin.install(fixture.config) + fixture.install() monkeypatch.setenv("GROK_HOME", "") assert fixture.plugin.status()["installed"] is True + assert_managed_triple(fixture.settings()) -def test_commands_block_and_forward_official_payload_byte_for_byte( - 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( - "from pathlib import Path\nimport sys\n" - f"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")) - pre_payload = json.dumps({ - "hookEventName": "PreToolUse", - "sessionId": "session-1", - "cwd": str(fixture.home_dir), - "workspaceRoot": str(fixture.home_dir), - "toolName": "Bash", - "toolInput": {"command": "echo test"}, - }, separators=(",", ":")) - guard = managed_handler(settings, "PreToolUse", "guard.py") - guard_result = run_command(guard["command"], fixture.home_dir, pre_payload) - assert guard_result.returncode == 2 - assert "Agent is frozen by Elydora" in guard_result.stderr - - post_payload = json.dumps({ - "hookEventName": "PostToolUse", - "sessionId": "session-1", - "cwd": str(fixture.home_dir), - "workspaceRoot": str(fixture.home_dir), - "toolName": "Bash", - "toolInput": {"command": "echo test"}, - "toolResult": {"output": "test"}, - }, separators=(",", ":")) - audit = managed_handler(settings, "PostToolUse", "hook.py") - audit_result = run_command(audit["command"], fixture.home_dir, post_payload) - assert audit_result.returncode == 0 - assert capture_path.read_text(encoding="utf-8") == post_payload - - -def test_status_requires_complete_pair_and_both_runtime_files( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - assert fixture.plugin.status()["installed"] is True - - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - del settings["hooks"]["PostToolUse"] - fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") - assert fixture.plugin.status()["installed"] is False - - fixture.plugin.install(fixture.config) - fixture.guard_path.unlink() - status = fixture.plugin.status() - assert status["installed"] is False - assert "runtime scripts missing" in status["details"] - - -def test_uninstall_removes_exact_ownership_and_preserves_mixed_handlers( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - fixture = prepare_fixture( - monkeypatch, - tmp_path, - existing_config={"owner": "user", "hooks": {"Notification": []}}, - ) - fixture.plugin.install(fixture.config) - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - settings["hooks"]["PreToolUse"][-1]["hooks"].append({ - "type": "command", - "command": "user-command", - "timeout": 10, - }) - settings["hooks"]["PreToolUse"].append({"hooks": [{ - "type": "command", - "command": grok._build_command(str(fixture.agent_dir / "guard.py.backup")), - "timeout": 10, - }]}) - settings["hooks"]["PreToolUse"].append({"hooks": [{ - "type": "command", - "command": grok._build_command(str( - fixture.agent_dir.parent / "agent-10" / "guard.py" - )), - "timeout": 10, - }]}) - fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") - - uninstall_id = "AGENT-1" if os.name == "nt" else AGENT_ID - fixture.plugin.uninstall(uninstall_id) - - remaining = json.loads(fixture.config_path.read_text(encoding="utf-8")) - assert remaining["owner"] == "user" - assert remaining["hooks"]["Notification"] == [] - assert len(remaining["hooks"]["PreToolUse"]) == 3 - assert remaining["hooks"]["PreToolUse"][0]["hooks"] == [{ - "type": "command", - "command": "user-command", - "timeout": 10, - }] - raw = fixture.config_path.read_text(encoding="utf-8") - assert "guard.py.backup" in raw - assert "agent-10" in raw - assert "PostToolUse" not in remaining["hooks"] - - -def test_install_replaces_stale_handlers_for_every_agent( - 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")) - for event, script_name in ( - ("PreToolUse", "guard.py"), - ("PostToolUse", "hook.py"), - ): - settings["hooks"][event].append({"hooks": [{ - "type": "command", - "command": grok._build_command(str( - fixture.agent_dir.parent / "agent-old" / script_name - )), - "timeout": 10, - }]}) - fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") - - fixture.plugin.install(fixture.config) - - raw = fixture.config_path.read_text(encoding="utf-8") - current = json.loads(raw) - assert "agent-old" not in raw - assert len(current["hooks"]["PreToolUse"]) == 1 - assert len(current["hooks"]["PostToolUse"]) == 1 - - -def test_uninstall_preserves_untouched_empty_native_group( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - fixture = prepare_fixture( - monkeypatch, tmp_path, existing_config={"owner": "user"} - ) - fixture.plugin.install(fixture.config) - settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) - settings["hooks"]["PreToolUse"].insert( - 0, {"hooks": [], "label": "keep empty group"} - ) - 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 remaining["hooks"]["PreToolUse"] == [ - {"hooks": [], "label": "keep empty group"} - ] - assert "PostToolUse" not in remaining["hooks"] - - -def test_uninstall_removes_config_owned_entirely_by_elydora( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - - fixture.plugin.uninstall(AGENT_ID) - - assert fixture.config_path.exists() is False - - -def test_install_preserves_malformed_json_for_recovery( +def test_install_parses_hook_file_before_creating_runtime_files( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: fixture = prepare_fixture( monkeypatch, tmp_path, existing_config="{ malformed" ) - with pytest.raises(ValueError, match="parse Grok hooks config"): - fixture.plugin.install(fixture.config) + with pytest.raises(ValueError, match="parse Grok user hooks"): + fixture.install() 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 not fixture.agent_dir.exists() @pytest.mark.parametrize( - "existing", + ("existing", "pattern"), [ - {"hooks": None}, - {"hooks": {"PreToolUse": None}}, - {"hooks": {"PreToolUse": [None]}}, - {"hooks": {"PreToolUse": [{"matcher": None, "hooks": []}]}}, - {"hooks": {"PreToolUse": [{"matcher": 1, "hooks": []}]}}, - {"hooks": {"PreToolUse": [{"hooks": None}]}}, - {"hooks": {"PreToolUse": [{"hooks": [ - {"type": "command", "command": ""} - ]}]}}, - {"hooks": {"PreToolUse": [{"hooks": [ - {"type": "file", "command": "x"} - ]}]}}, - {"hooks": {"PreToolUse": [{"hooks": [ - {"type": {"future": True}, "command": "x"} - ]}]}}, - {"hooks": {"PreToolUse": [{"hooks": [ - {"type": "http", "url": ""} - ]}]}}, - {"hooks": {"PreToolUse": [{"hooks": [ - {"type": "command", "command": "x", "timeout": 0} - ]}]}}, - {"hooks": {"PreToolUse": [{"hooks": [ - {"type": "command", "command": "x", "timeout": float("inf")} - ]}]}}, + ('{"hooks":{},"hooks":{}}', 'duplicate field "hooks"'), + ({"hooks": None}, 'field "hooks" must be an object'), + ({"hooks": {"PreToolUse": None}}, "must be an array"), + ({"hooks": {"PreToolUse": [None]}}, "group.*must be an object"), + ( + {"hooks": {"PreToolUse": [{"matcher": 1, "hooks": []}]}}, + "matcher must be a string", + ), + ( + {"hooks": {"SessionStart": [{"matcher": "x", "hooks": []}]}}, + "cannot declare a matcher", + ), + ( + {"hooks": {"PreToolUse": [{"hooks": [None]}]}}, + "handler.*must be an object", + ), + ( + {"hooks": {"PreToolUse": [{"hooks": [{"type": "file"}]}]}}, + "unsupported type", + ), + ( + {"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "" + }]}]}}, + "non-empty command", + ), + ( + {"hooks": {"PostToolUse": [{"hooks": [{ + "type": "http", "url": "" + }]}]}}, + "non-empty url", + ), + ( + {"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "x", "timeout": -1 + }]}]}}, + "non-negative integer", + ), + ( + {"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "x", "timeout": 1.5 + }]}]}}, + "non-negative integer", + ), + ( + '{"hooks":{"PreToolUse":[{"hooks":[{"type":"command",' + '"command":"x","timeout":NaN}]}]}}', + "invalid numeric constant", + ), + ( + {"hooks": {"PreToolUse": [{"hooks": [{ + "type": "command", "command": "x", "env": {"A": 1} + }]}]}}, + "env must map names to strings", + ), ], ) def test_install_rejects_invalid_native_shapes_before_writes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, - existing: dict[str, Any], + existing: object, + pattern: str, ) -> None: fixture = prepare_fixture( monkeypatch, tmp_path, existing_config=existing ) original = fixture.config_path.read_text(encoding="utf-8") - with pytest.raises(ValueError): - fixture.plugin.install(fixture.config) + with pytest.raises(ValueError, match=pattern): + 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 not fixture.agent_dir.exists() -def test_install_rejects_missing_guard_before_creating_files( - 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( +def test_install_migrates_exact_legacy_commands_and_preserves_lookalikes( 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") + lookalike = f"{legacy_command(fixture.guard_path)} --inspect" + legacy = { + "hooks": { + "PreToolUse": [ + {"hooks": [{ + "type": "command", + "command": legacy_command(fixture.guard_path), + "timeout": 10, + }]}, + {"hooks": [{ + "type": "command", + "command": lookalike, + "timeout": 10, + }]}, + ], + "PostToolUse": [{"hooks": [{ + "type": "command", + "command": legacy_command(fixture.hook_path), + "timeout": 10, + }]}], + } + } + fixture.config_path.parent.mkdir(parents=True) + fixture.config_path.write_text(json.dumps(legacy), encoding="utf-8") - with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.install() + settings = fixture.settings() + assert_managed_triple(settings) + assert any( + handler.get("command") == lookalike + for group in settings["hooks"]["PreToolUse"] + for handler in group["hooks"] + ) + managed = settings["hooks"]["PreToolUse"][-1] + managed["hooks"].append({ + "type": "command", "command": "user-command", "timeout": 10 + }) + fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") + + fixture.plugin.uninstall(AGENT_ID) + remaining = fixture.settings() + assert remaining["hooks"]["PreToolUse"] == [ + {"hooks": [{"type": "command", "command": lookalike, "timeout": 10}]}, + {"hooks": [{"type": "command", "command": "user-command", "timeout": 10}]}, + ] + assert "PostToolUse" not in remaining["hooks"] + assert "PostToolUseFailure" not in remaining["hooks"] + + +def test_uninstall_preserves_user_config_and_removes_fully_managed_file( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user = prepare_fixture( + monkeypatch, + tmp_path / "user", + existing_config={"owner": "user", "hooks": {"Notification": []}}, + ) + user.install() + user.plugin.uninstall(AGENT_ID) + assert user.settings() == { + "owner": "user", + "hooks": {"Notification": []}, + } + + managed = prepare_fixture(monkeypatch, tmp_path / "managed") + managed.install() + managed.plugin.uninstall(AGENT_ID) + assert not managed.config_path.exists() + + +def test_status_requires_one_complete_triple_identity_and_private_key( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.plugin.status()["installed"] is True + + settings = fixture.settings() + del settings["hooks"]["PostToolUseFailure"] + fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") + assert fixture.plugin.status()["installed"] is False + + fixture.install() + settings = fixture.settings() + settings["hooks"]["PostToolUseFailure"].append( + settings["hooks"]["PostToolUseFailure"][-1] + ) + fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") + assert fixture.plugin.status()["installed"] is False + + fixture.install() + fixture.private_key_path.write_text("invalid", encoding="utf-8") + with pytest.raises(ValueError, match="private key.*canonical 32-byte"): fixture.plugin.status() -def test_atomic_writes_leave_no_temporary_files( +@pytest.mark.parametrize( + "runtime_name", ["guard.py", "hook.py", "config.json", "private.key"] +) +def test_status_requires_every_runtime_file( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, runtime_name: str +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + (fixture.agent_dir / runtime_name).unlink() + + assert fixture.plugin.status()["installed"] is False + + +def test_installation_leaves_no_transaction_files( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) + fixture.install() - for directory in (fixture.agent_dir, fixture.config_path.parent): - assert all(path.suffix != ".tmp" for path in directory.iterdir()) + 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 diff --git a/tests/test_grok_runtime.py b/tests/test_grok_runtime.py new file mode 100644 index 0000000..191f5ca --- /dev/null +++ b/tests/test_grok_runtime.py @@ -0,0 +1,158 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from grok_support import ( + GrokApiHandler, + managed_handler, + prepare_fixture, + run_hook, + server_base_url, + start_api_server, +) + + +PRE_PAYLOAD = { + "hookEventName": "pre_tool_use", + "sessionId": "session-1", + "cwd": "C:/project", + "workspaceRoot": "C:/project", + "toolName": "run_terminal_command", + "toolInput": {"command": "npm test"}, + "toolUseId": "tool-use-1", + "toolInputTruncated": False, + "timestamp": "2026-07-19T12:00:00Z", +} + + +def test_runtimes_enforce_active_state_and_preserve_native_payloads( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server() + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + try: + fixture.install() + settings = fixture.settings() + guard = managed_handler(settings, "PreToolUse") + success_audit = managed_handler(settings, "PostToolUse") + failure_audit = managed_handler(settings, "PostToolUseFailure") + + guard_result = run_hook( + str(guard["command"]), + json.dumps(PRE_PAYLOAD), + fixture, + {"GROK_HOOK_EVENT": "injected-command-fragment"}, + ) + assert guard_result.returncode == 0, guard_result.stderr + assert guard_result.stdout == "" + + success_payload = { + **PRE_PAYLOAD, + "hookEventName": "post_tool_use", + "toolResult": {"output": "tests passed"}, + "toolResultTruncated": False, + "durationMs": 125, + } + success = run_hook( + str(success_audit["command"]), + json.dumps(success_payload), + fixture, + ) + assert success.returncode == 0, success.stderr + + failure_payload = { + **PRE_PAYLOAD, + "hookEventName": "post_tool_use_failure", + "toolResult": {"error": "command failed", "exitCode": 1}, + "toolResultTruncated": False, + "durationMs": 40, + } + failure = run_hook( + str(failure_audit["command"]), + json.dumps(failure_payload), + fixture, + ) + assert failure.returncode == 0, failure.stderr + + operations = [ + request["json"] + for request in GrokApiHandler.requests + if request["method"] == "POST" + ] + assert len(operations) == 2 + assert operations[0]["payload"] == success_payload + assert operations[1]["payload"] == failure_payload + assert operations[0]["subject"]["session_id"] == "session-1" + assert operations[1]["action"]["tool"] == "run_terminal_command" + finally: + server.shutdown() + server.server_close() + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_emits_official_deny_json_and_exit_code_two( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, +) -> None: + server = start_api_server(status=status) + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + try: + fixture.install() + command = managed_handler( + fixture.settings(), "PreToolUse" + )["command"] + result = run_hook(str(command), json.dumps(PRE_PAYLOAD), fixture) + + assert result.returncode == 2 + assert status in result.stderr.lower() + decision = json.loads(result.stdout) + assert set(decision) == {"decision", "reason"} + assert decision["decision"] == "deny" + assert status in decision["reason"].lower() + finally: + server.shutdown() + server.server_close() + + +def test_audit_runtime_records_malformed_input_and_api_failures( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server(operation_status=503) + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + try: + fixture.install() + command = managed_handler( + fixture.settings(), "PostToolUse" + )["command"] + + malformed = run_hook(str(command), "{ malformed", fixture) + assert malformed.returncode == 0 + assert malformed.stderr == "" + + upstream = run_hook( + str(command), + json.dumps({ + **PRE_PAYLOAD, + "hookEventName": "post_tool_use", + "toolResult": {"output": "test"}, + }), + fixture, + ) + assert upstream.returncode == 0 + assert upstream.stderr == "" + log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8") + assert "invalid JSON" in log + assert "HTTP 503" in log + finally: + server.shutdown() + server.server_close() diff --git a/tests/test_grok_security.py b/tests/test_grok_security.py new file mode 100644 index 0000000..5ad4e3b --- /dev/null +++ b/tests/test_grok_security.py @@ -0,0 +1,221 @@ +from __future__ import annotations + +import os +from pathlib import Path +import subprocess # nosec B404 +import sys + +import pytest + +from grok_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + prepare_fixture, + write_text, +) + + +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}") + + +def test_install_rejects_linked_hook_file_before_runtime_creation( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + target = tmp_path / "hooks-target.json" + source = '{"owner":"protected"}\n' + write_text(target, source) + fixture.config_path.parent.mkdir(parents=True) + symlink_or_skip(target, fixture.config_path) + + with pytest.raises(OSError, match="not a physical file"): + fixture.install() + + assert target.read_text(encoding="utf-8") == source + assert not fixture.agent_dir.exists() + + +@pytest.mark.parametrize("kind", ["home", "hooks", "runtime"]) +def test_install_rejects_linked_directories_before_writes( + 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 == "home": + link = fixture.grok_home + (target / "hooks").mkdir() + elif kind == "hooks": + link = fixture.config_path.parent + fixture.grok_home.mkdir(parents=True) + else: + link = fixture.agent_dir.parent + link.parent.mkdir(parents=True, exist_ok=True) + symlink_or_skip(target, link, directory=True) + + with pytest.raises(OSError, match="physical directory"): + fixture.install() + + untouched = fixture.config_path if kind == "runtime" else fixture.agent_dir + assert not untouched.exists() + + +def test_status_rejects_linked_private_key( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + target = tmp_path / "private-key-target" + source = fixture.private_key_path.read_text(encoding="utf-8") + write_text(target, source) + fixture.private_key_path.unlink() + symlink_or_skip(target, fixture.private_key_path) + + with pytest.raises(OSError, match="not a physical file"): + fixture.plugin.status() + + assert target.read_text(encoding="utf-8") == source + + +def test_preflight_inspects_every_runtime_artifact( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.agent_dir.mkdir(parents=True) + write_text(fixture.guard_path, "orphaned guard\n", 0o700) + target = tmp_path / "linked-audit-target" + write_text(target, "linked audit\n", 0o700) + symlink_or_skip(target, fixture.hook_path) + + with pytest.raises(OSError, match="not a physical file"): + fixture.install() + + assert target.read_text(encoding="utf-8") == "linked audit\n" + assert not fixture.config_path.exists() + + +@pytest.mark.parametrize("kind", ["orphaned", "mismatched"]) +def test_install_rejects_unverifiable_runtime_identity( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.agent_dir.mkdir(parents=True) + if kind == "orphaned": + write_text(fixture.guard_path, "orphaned guard\n", 0o700) + else: + write_text( + fixture.runtime_config_path, + '{"agent_name":"grok","agent_id":"another-agent"}\n', + ) + + pattern = ( + "identity cannot be verified" + if kind == "orphaned" + else "identity does not match" + ) + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert not fixture.config_path.exists() + + +@pytest.mark.parametrize( + ("field", "value", "pattern"), + [ + ("agent_name", "codex", "requires agent_name grok"), + ("agent_id", "../escape", "Invalid agent ID"), + ("private_key", "invalid", "canonical 32-byte"), + ("token", "", "non-empty string"), + ( + "base_url", + "https://api.elydora.com/path?token=secret", + "query parameters", + ), + ("guard_script_path", "outside", "managed agent directory"), + ], +) +def test_install_validates_runtime_inputs_before_creating_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + field: str, + value: str, + pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.config[field] = ( + str(fixture.home_dir / "outside" / "guard.py") + if field == "guard_script_path" + else value + ) + + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() + + +def test_status_surfaces_malformed_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") + + with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.plugin.status() + + +def test_cli_preflight_preserves_malformed_config_before_runtime_creation( + tmp_path: Path, +) -> None: + home_dir = tmp_path / "cli-home" + config_path = home_dir / ".grok" / "hooks" / "elydora-audit.json" + key_path = tmp_path / "private.key" + write_text(config_path, "{ malformed") + write_text(key_path, VALID_PRIVATE_KEY) + result = subprocess.run( + [ + sys.executable, + "-m", + "elydora.cli", + "install", + "--agent", + "grok", + "--org_id", + "org-1", + "--agent_id", + AGENT_ID, + "--private_key_file", + str(key_path), + "--kid", + "kid-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), + "GROK_HOME": "", + }, + text=True, + ) + + assert result.returncode == 1 + assert "parse Grok user hooks" in result.stderr + assert config_path.read_text(encoding="utf-8") == "{ malformed" + assert not (home_dir / ".elydora").exists() diff --git a/tests/test_grok_transaction.py b/tests/test_grok_transaction.py new file mode 100644 index 0000000..fd16c63 --- /dev/null +++ b/tests/test_grok_transaction.py @@ -0,0 +1,248 @@ +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from elydora.plugins import _transaction, grok_installation +from elydora.plugins.grok_contract import ( + build_grok_group, + remove_managed_grok_hooks, + render_grok_document, +) +from elydora.plugins.grok_io import read_grok_document +from elydora.utils import base64url_encode +from grok_support import ( + AGENT_ID, + GrokFixture, + assert_no_transaction_files, + prepare_fixture, + write_text, +) + + +RUNTIME_NAMES = ("guard.py", "config.json", "private.key", "hook.py") + + +def prepare_installation(fixture: GrokFixture): + document = read_grok_document() + config = fixture.config + paths = grok_installation.preflight_grok_installation(config, document) + cleaned = remove_managed_grok_hooks(document.hooks) + hooks = { + **cleaned, + "PreToolUse": [ + *cleaned.get("PreToolUse", []), + build_grok_group(paths.guard_path), + ], + "PostToolUse": [ + *cleaned.get("PostToolUse", []), + build_grok_group(paths.audit_path), + ], + "PostToolUseFailure": [ + *cleaned.get("PostToolUseFailure", []), + build_grok_group(paths.audit_path), + ], + } + rendered = render_grok_document(document, hooks) + return grok_installation.prepare_grok_installation( + config, paths, rendered + ) + + +def snapshot_files(paths: list[Path]) -> dict[Path, str]: + return {path: path.read_text(encoding="utf-8") for path in paths} + + +def assert_snapshot(snapshot: dict[Path, str]) -> None: + for path, source in snapshot.items(): + assert path.read_text(encoding="utf-8") == source + + +def preserved_rollback(directory: Path, basename: str) -> Path: + matches = list(directory.glob(f".{basename}.*.rollback")) + assert len(matches) == 1, [path.name for path in directory.iterdir()] + return matches[0] + + +def test_install_restores_config_and_four_runtimes_when_final_commit_fails( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + original = '{"owner":"user"}\n' + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_config=original + ) + real_replace = _transaction.os.replace + failed = False + + def fail_config_once(source: str, destination: str) -> None: + nonlocal failed + if os.path.abspath(destination) == str(fixture.config_path) and not failed: + failed = True + raise OSError("injected Grok config failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_config_once) + + with pytest.raises(OSError, match="injected Grok config failure"): + fixture.install() + + assert fixture.config_path.read_text(encoding="utf-8") == original + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_install_rejects_concurrent_config_change_and_restores_runtimes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_config={"owner": "user"}, + ) + changes = prepare_installation(fixture) + concurrent = '{"owner":"concurrent"}\n' + fixture.config_path.write_text(concurrent, encoding="utf-8") + + with pytest.raises(OSError, match="changed during installation"): + grok_installation.commit_grok_installation(changes) + + assert fixture.config_path.read_text(encoding="utf-8") == concurrent + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_uninstall_preserves_config_when_commit_fails( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_config={"owner": "user"}, + ) + fixture.install() + before = fixture.config_path.read_text(encoding="utf-8") + document = read_grok_document() + rendered = render_grok_document( + document, + remove_managed_grok_hooks(document.hooks, AGENT_ID), + ) + changes = grok_installation.prepare_grok_uninstall(rendered) + + def fail_replace(_source: str, _destination: str) -> None: + raise OSError("injected Grok uninstall failure") + + monkeypatch.setattr(_transaction.os, "replace", fail_replace) + + with pytest.raises(OSError, match="injected Grok uninstall failure"): + grok_installation.commit_grok_uninstall(changes) + + assert fixture.config_path.read_text(encoding="utf-8") == before + assert_no_transaction_files(fixture.home_dir) + + +def test_recovery_preserves_original_after_committed_file_changes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + original = fixture.runtime_config_path.read_text(encoding="utf-8") + fixture.config["org_id"] = "org-updated" + changes = prepare_installation(fixture) + real_replace = _transaction.os.replace + calls = 0 + + def change_first_then_fail(source: str, destination: str) -> None: + nonlocal calls + calls += 1 + if calls == 1: + real_replace(source, destination) + write_text(Path(destination), "external change\n") + return + raise OSError("injected later commit failure") + + monkeypatch.setattr( + _transaction.os, "replace", change_first_then_fail + ) + + with pytest.raises(OSError, match="original content preserved at"): + grok_installation.commit_grok_installation(changes) + + assert fixture.runtime_config_path.read_text(encoding="utf-8") == ( + "external change\n" + ) + rollback = preserved_rollback(fixture.agent_dir, "config.json") + assert rollback.read_text(encoding="utf-8") == original + + +def test_recovery_preserves_original_after_restore_failure( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + original = fixture.runtime_config_path.read_text(encoding="utf-8") + fixture.config["org_id"] = "org-updated" + fixture.config["private_key"] = base64url_encode( + bytes(range(31, -1, -1)) + ) + changes = prepare_installation(fixture) + real_replace = _transaction.os.replace + commits = 0 + + def fail_commit_and_restore(source: str, destination: str) -> None: + nonlocal commits + if source.endswith(".rollback"): + raise OSError("injected rollback failure") + commits += 1 + if commits == 2: + raise OSError("injected later commit failure") + real_replace(source, destination) + + monkeypatch.setattr( + _transaction.os, "replace", fail_commit_and_restore + ) + + with pytest.raises(OSError, match="original content preserved at"): + grok_installation.commit_grok_installation(changes) + + rollback = preserved_rollback(fixture.agent_dir, "config.json") + assert rollback.read_text(encoding="utf-8") == original + + +def test_prepared_install_detects_change_before_commit( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + existing_config={"owner": "user"}, + ) + changes = prepare_installation(fixture) + concurrent = '{"owner":"concurrent-before-stage"}\n' + fixture.config_path.write_text(concurrent, encoding="utf-8") + + with pytest.raises(OSError, match="changed during installation"): + grok_installation.commit_grok_installation(changes) + + assert fixture.config_path.read_text(encoding="utf-8") == concurrent + assert_no_transaction_files(fixture.home_dir) + + +def test_transaction_rejects_oversized_source_before_staging( + tmp_path: Path, +) -> None: + target = tmp_path / "managed" / "config.json" + + with pytest.raises(ValueError, match="exceeds 4 bytes"): + _transaction.file_change( + str(target), + "managed test source", + "five!", + 0o600, + maximum_bytes=4, + ) + + assert not target.parent.exists()