From 9b67aad1135763bde9195c7ba9a589dd93de9f8d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Jul 2026 12:34:09 -0400 Subject: [PATCH] fix(copilot): mirror transactional Python hooks --- sdks/python/README.md | 2 + sdks/python/elydora/plugins/_transaction.py | 58 +- sdks/python/elydora/plugins/copilot.py | 225 +++----- .../elydora/plugins/copilot_contract.py | 355 ++++++++---- .../elydora/plugins/copilot_installation.py | 277 ++++++++++ sdks/python/elydora/plugins/copilot_io.py | 431 ++++++++++++--- sdks/python/elydora/plugins/copilot_schema.py | 276 ++++++++++ sdks/python/elydora/plugins/registry.py | 2 +- sdks/python/tests/copilot_support.py | 216 ++++++++ sdks/python/tests/test_cli_security.py | 5 +- sdks/python/tests/test_copilot_official.py | 100 ++++ sdks/python/tests/test_copilot_plugin.py | 518 +++++++++--------- sdks/python/tests/test_copilot_transaction.py | 238 ++++++++ 13 files changed, 2120 insertions(+), 583 deletions(-) create mode 100644 sdks/python/elydora/plugins/copilot_installation.py create mode 100644 sdks/python/elydora/plugins/copilot_schema.py create mode 100644 sdks/python/tests/copilot_support.py create mode 100644 sdks/python/tests/test_copilot_official.py create mode 100644 sdks/python/tests/test_copilot_transaction.py diff --git a/sdks/python/README.md b/sdks/python/README.md index dd96948..0799c48 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -82,6 +82,8 @@ Claude Code installation writes exact matchless `PreToolUse`, `PostToolUse`, and Codex installation writes exact `PreToolUse` and `PostToolUse` command groups to `$CODEX_HOME/hooks.json` (`~/.codex/hooks.json` by default) and preserves the complete native event payload. A configured `CODEX_HOME` follows Codex's existing-directory canonicalization rule. User TOML, project, plugin, and managed sources remain unchanged and continue loading additively. The hook file, generated runtimes, runtime config, and private key commit as one rollback-capable update. Run `/hooks` after installation and approve both Elydora definition hashes. +GitHub Copilot CLI installation writes exact `preToolUse`, `postToolUse`, and `postToolUseFailure` handlers to `$COPILOT_HOME/hooks/elydora-audit.json` (`~/.copilot/hooks/elydora-audit.json` by default). Empty home overrides select the default. Policy, repository, inline settings, cross-tool Claude, and plugin hook sources remain unchanged and continue loading additively. Elydora evaluates the effective `disableAllHooks` value across Copilot's documented settings precedence before writing. The managed hook file, exact legacy migration, generated runtimes, runtime config, and private key commit through one rollback-capable transaction. Managed commands preserve the complete native camelCase payload and propagate freeze or revocation through exit code `2`; Copilot timeouts retain their documented fail-open behavior. Restart active Copilot CLI sessions after installation. + Cursor installation writes native global `preToolUse`, `postToolUse`, and `postToolUseFailure` handlers to `~/.cursor/hooks.json`. Elydora forwards Cursor's native success and failure payloads, preserves unrelated user hooks, and leaves project and enterprise sources unchanged. The generated guard and audit handlers use a 10-second fail-closed boundary, exact freeze exit-code propagation, protected runtime credentials, and one rollback-capable transaction for all runtime and hook configuration files. Gemini CLI installation writes named, matchless `BeforeTool` and `AfterTool` command hooks to `$GEMINI_CLI_HOME/.gemini/settings.json` (`~/.gemini/settings.json` by default). Empty home overrides select the OS home; relative and literal-tilde values retain Gemini CLI's native path behavior. JSON comments and unrelated settings remain intact. Workspace, system-default, system-override, and extension hook sources remain unchanged and continue loading additively. User settings, generated runtimes, runtime config, and private key commit through one rollback-capable transaction. Managed handlers preserve the complete native snake_case payload, emit valid JSON on successful execution, propagate freeze and revocation through exit code `2`, and use encoded PowerShell commands on Windows. `hooksConfig.enabled` and `hooksConfig.disabled` remain authoritative. Run `/hooks list` after installation to inspect effective hooks from every source. diff --git a/sdks/python/elydora/plugins/_transaction.py b/sdks/python/elydora/plugins/_transaction.py index d9cb3dc..a8de7af 100644 --- a/sdks/python/elydora/plugins/_transaction.py +++ b/sdks/python/elydora/plugins/_transaction.py @@ -5,7 +5,7 @@ from __future__ import annotations from dataclasses import dataclass import os import tempfile -from typing import List, Optional, Sequence +from typing import List, Optional, Sequence, cast from uuid import uuid4 from ._managed_files import ( @@ -30,6 +30,14 @@ class FileChange: maximum_bytes: int +@dataclass(frozen=True) +class FilePrecondition: + file_path: str + label: str + original: Optional[FileSnapshot] + maximum_bytes: int = MAX_SOURCE_BYTES + + @dataclass class _StagedChange: change: FileChange @@ -39,6 +47,22 @@ class _StagedChange: committed: bool = False +_EXPECTED_SNAPSHOT_UNSET = object() + + +def _same_snapshot( + current: Optional[FileSnapshot], + expected: Optional[FileSnapshot], +) -> bool: + if current is None or expected is None: + return current is expected + return ( + current.contents == expected.contents + and current.device == expected.device + and current.inode == expected.inode + ) + + def read_optional( file_path: str, label: str, @@ -55,6 +79,7 @@ def source_change( next_source: Optional[str], mode: int, maximum_bytes: int = MAX_SOURCE_BYTES, + expected_snapshot: object = _EXPECTED_SNAPSHOT_UNSET, ) -> Optional[FileChange]: if original == next_source: return None @@ -69,6 +94,10 @@ def source_change( current = None if snapshot is None else snapshot.contents if current != original: raise OSError(f"{label} changed before staging: {file_path}") + if expected_snapshot is not _EXPECTED_SNAPSHOT_UNSET: + expected = cast(Optional[FileSnapshot], expected_snapshot) + if not _same_snapshot(snapshot, expected): + raise OSError(f"{label} changed before staging: {file_path}") return FileChange( file_path, label, @@ -170,6 +199,23 @@ def _assert_unchanged(change: FileChange) -> None: raise OSError(f"{change.label} changed during installation: {change.file_path}") +def _assert_preconditions( + preconditions: Sequence[FilePrecondition], + operation: str, +) -> None: + for condition in preconditions: + current = read_physical_file( + condition.file_path, + condition.label, + condition.maximum_bytes, + ) + if not _same_snapshot(current, condition.original): + raise OSError( + f"{condition.label} changed during {operation}: " + f"{condition.file_path}" + ) + + def _stage(change: FileChange) -> _StagedChange: _assert_unchanged(change) directory = os.path.dirname(change.file_path) @@ -302,7 +348,11 @@ def _cleanup(staged: _StagedChange) -> None: _remove_optional(staged.rollback_path) -def write_changes(changes: Sequence[FileChange], label: str) -> None: +def write_changes( + changes: Sequence[FileChange], + label: str, + preconditions: Sequence[FilePrecondition] = (), +) -> None: filtered = [change for change in changes if change.original != change.next_source] if not filtered: return @@ -311,10 +361,14 @@ def write_changes(changes: Sequence[FileChange], label: str) -> None: raise ValueError(f"{label} contains duplicate file targets") staged: List[_StagedChange] = [] try: + _assert_preconditions(preconditions, label) for change in filtered: staged.append(_stage(change)) + _assert_preconditions(preconditions, label) for item in staged: + _assert_preconditions(preconditions, label) _commit(item) + _assert_preconditions(preconditions, label) except Exception as error: recovery_errors = [] for item in reversed(staged): diff --git a/sdks/python/elydora/plugins/copilot.py b/sdks/python/elydora/plugins/copilot.py index 32da3ac..4dac6d4 100644 --- a/sdks/python/elydora/plugins/copilot.py +++ b/sdks/python/elydora/plugins/copilot.py @@ -1,188 +1,129 @@ -"""GitHub Copilot CLI user-hook integration.""" +"""GitHub Copilot CLI native user-hook integration.""" from __future__ import annotations -import json import os -from typing import List, Optional +from typing import List -from ._transaction import ( - FileChange, - file_change, - require_runtime, - write_changes, -) from .base import AgentPlugin, InstallConfig, PluginStatus from .copilot_contract import ( AGENT_KEY, - AUDIT_SCRIPT, - CopilotHooks, CopilotSources, + RenderedDocument, RuntimeContract, build_handler, remove_managed_hooks, render_document, runtime_contracts, ) -from .copilot_io import read_sources, rendered_change, runtime_files_exist -from .hook_template import generate_hook_script +from .copilot_installation import ( + commit_copilot_installation, + commit_copilot_uninstall, + preflight_copilot_installation, + prepare_copilot_installation, + prepare_copilot_uninstall, +) +from .copilot_io import read_sources, runtime_files_exist -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") - - -def _home_dir() -> str: - return os.path.expanduser("~") - - -def _json_source(value: dict) -> str: - return json.dumps(value, indent=2) + "\n" - - -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 _rendered_changes( +def _render_installation( sources: CopilotSources, - user_hooks: CopilotHooks, - agent_id: str = "", -) -> List[FileChange]: + guard_path: str, + audit_path: str, +) -> List[RenderedDocument]: + user_hooks = remove_managed_hooks(sources.user.hooks) + user_hooks["preToolUse"] = [ + *user_hooks.get("preToolUse", []), + build_handler(guard_path), + ] + user_hooks["postToolUse"] = [ + *user_hooks.get("postToolUse", []), + build_handler(audit_path), + ] + user_hooks["postToolUseFailure"] = [ + *user_hooks.get("postToolUseFailure", []), + build_handler(audit_path), + ] rendered = [render_document(sources.user, user_hooks)] if sources.legacy is not None: rendered.append(render_document( sources.legacy, - remove_managed_hooks( - sources.legacy.hooks, - ELYDORA_DIR, - agent_id, - ), + remove_managed_hooks(sources.legacy.hooks), )) - return [ - change - for item in rendered - for change in [rendered_change(item)] - if change is not None - ] + return rendered -def _contracts(sources: CopilotSources) -> List[RuntimeContract]: - contracts = runtime_contracts(sources.user.hooks, ELYDORA_DIR) - if sources.legacy is not None: - contracts.extend(runtime_contracts(sources.legacy.hooks, ELYDORA_DIR)) - unique = {} - for contract in contracts: - unique[os.path.normcase(contract.agent_id)] = contract - return list(unique.values()) - - -def _configured_path( +def _render_uninstall( sources: CopilotSources, - contracts: List[RuntimeContract], -) -> str: - user_ids = { - os.path.normcase(contract.agent_id) - for contract in runtime_contracts(sources.user.hooks, ELYDORA_DIR) - } - if any(os.path.normcase(contract.agent_id) in user_ids for contract in contracts): - return sources.user.file_path + agent_id: str, +) -> List[RenderedDocument]: + rendered = [render_document( + sources.user, + remove_managed_hooks(sources.user.hooks, agent_id), + )] if sources.legacy is not None: - return sources.legacy.file_path - return sources.user.file_path + rendered.append(render_document( + sources.legacy, + remove_managed_hooks(sources.legacy.hooks, agent_id), + )) + return rendered + + +def _merged_contracts(sources: CopilotSources) -> List[RuntimeContract]: + contracts = runtime_contracts(sources.user.hooks) + if sources.legacy is not None: + contracts.extend(runtime_contracts(sources.legacy.hooks)) + unique = { + os.path.normcase(contract.agent_id): contract + for contract in contracts + } + return list(unique.values()) class CopilotPlugin(AgentPlugin): """Install Elydora into GitHub Copilot CLI's global user hooks.""" + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + sources = read_sources() + preflight_copilot_installation(config, sources) + def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - sources = read_sources(_home_dir()) - guard_path = config.get("guard_script_path", "") - require_runtime(guard_path, "Elydora guard runtime") - - agent_directory = os.path.join(ELYDORA_DIR, agent_id) - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - user_hooks = remove_managed_hooks(sources.user.hooks, ELYDORA_DIR) - user_hooks["preToolUse"] = [ - *user_hooks.get("preToolUse", []), - build_handler(guard_path), - ] - user_hooks["postToolUse"] = [ - *user_hooks.get("postToolUse", []), - build_handler(audit_path), - ] - - 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"), + sources = read_sources() + paths = preflight_copilot_installation(config, sources) + rendered = _render_installation( + sources, + paths.guard_path, + paths.audit_path, ) - candidates: List[Optional[FileChange]] = [ - file_change( - os.path.join(agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, agent_id)), - 0o600, - ), - file_change( - os.path.join(agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - ), - file_change( - audit_path, - "Elydora audit runtime", - audit_script, - 0o700, - ), - ] - changes = [change for change in candidates if change is not None] - changes.extend(_rendered_changes(sources, user_hooks)) - write_changes(changes, "Install GitHub Copilot hooks") + prepared = prepare_copilot_installation(config, sources, rendered) + commit_copilot_installation(prepared) print(f"GitHub Copilot CLI hooks: {sources.user.file_path}") + print( + "GitHub Copilot CLI: restart active sessions to load updated hooks." + ) def uninstall(self, agent_id: str = "") -> None: - sources = read_sources(_home_dir()) - user_hooks = remove_managed_hooks( - sources.user.hooks, - ELYDORA_DIR, - agent_id, - ) - changes = _rendered_changes(sources, user_hooks, agent_id) - write_changes(changes, "Uninstall GitHub Copilot hooks") + sources = read_sources() + rendered = _render_uninstall(sources, agent_id) + commit_copilot_uninstall(prepare_copilot_uninstall(rendered)) def status(self) -> PluginStatus: - sources = read_sources(_home_dir()) - contracts = _contracts(sources) - if not contracts: - return PluginStatus( - installed=False, - agent=AGENT_KEY, - details="Not installed", - ) - installed = runtime_files_exist(contracts) - config_path = _configured_path(sources, contracts) + sources = read_sources() + contracts = _merged_contracts(sources) + configured = sources.disabled_by is None and bool(contracts) + installed = configured and runtime_files_exist(contracts) details = ( - f"Config: {config_path}" + f"Config: {sources.user.file_path}" if installed - else f"Configured at {config_path}; runtime scripts missing" + else ( + f"Configured at {sources.user.file_path}; " + "managed contract incomplete" + ) ) return PluginStatus( installed=installed, agent=AGENT_KEY, - details=details, + details=details if contracts else "Not installed", ) diff --git a/sdks/python/elydora/plugins/copilot_contract.py b/sdks/python/elydora/plugins/copilot_contract.py index c9a5663..b9b1fd2 100644 --- a/sdks/python/elydora/plugins/copilot_contract.py +++ b/sdks/python/elydora/plugins/copilot_contract.py @@ -1,4 +1,4 @@ -"""GitHub Copilot hook contract, ownership, and rendering.""" +"""GitHub Copilot CLI hook ownership and rendering.""" from __future__ import annotations @@ -8,7 +8,11 @@ import os import re import shlex import sys -from typing import Any, Dict, List, Optional, Set +from typing import Any, Dict, List, Optional, Tuple + +from ._managed_files import FileSnapshot +from ._strict_json import parse_json_object +from .copilot_schema import CopilotHooks, validate_hooks AGENT_KEY = "copilot" @@ -18,7 +22,11 @@ CONFIG_FILE = "elydora-audit.json" HOOK_TIMEOUT_SECONDS = 10 JsonObject = Dict[str, Any] -CopilotHooks = Dict[str, List[JsonObject]] +MANAGED_EVENTS: Tuple[Tuple[str, str], ...] = ( + ("preToolUse", GUARD_SCRIPT), + ("postToolUse", AUDIT_SCRIPT), + ("postToolUseFailure", AUDIT_SCRIPT), +) @dataclass(frozen=True) @@ -27,13 +35,24 @@ class CopilotDocument: file_path: str root: JsonObject hooks: CopilotHooks + hooks_disabled: bool raw: Optional[str] = None + snapshot: Optional[FileSnapshot] = None + + +@dataclass(frozen=True) +class SourcePrecondition: + file_path: str + label: str + snapshot: Optional[FileSnapshot] @dataclass(frozen=True) class CopilotSources: user: CopilotDocument legacy: Optional[CopilotDocument] + disabled_by: Optional[str] + settings_preconditions: Tuple[SourcePrecondition, ...] @dataclass(frozen=True) @@ -50,16 +69,36 @@ class RuntimeContract: audit_path: str -def _same_path(left: str, right: str) -> bool: +@dataclass(frozen=True) +class _ParsedArgument: + value: str + next_index: int + + +@dataclass(frozen=True) +class _ManagedEntry: + agent_id: str + script_path: str + + +def runtime_root() -> str: + return os.path.join(os.path.expanduser("~"), ".elydora") + + +def same_path(left: str, right: str) -> bool: return os.path.normcase(os.path.abspath(left)) == os.path.normcase( os.path.abspath(right) ) -def _same_agent_id(left: str, right: str) -> bool: +def same_agent_id(left: str, right: str) -> bool: return os.path.normcase(left) == os.path.normcase(right) +def _quote_posix(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + def _quote_powershell(value: str) -> str: return "'" + value.replace("'", "''") + "'" @@ -67,7 +106,7 @@ def _quote_powershell(value: str) -> str: def build_handler(script_path: str) -> JsonObject: return { "type": "command", - "bash": shlex.join([sys.executable, script_path]), + "bash": f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}", "powershell": ( f"& {_quote_powershell(sys.executable)} " f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" @@ -76,109 +115,205 @@ def build_handler(script_path: str) -> JsonObject: } -def _generated_script_path(handler: JsonObject) -> Optional[str]: - if set(handler) != {"type", "bash", "powershell", "timeoutSec"}: +def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: + if start >= len(command) or command[start] != "'": return None - if handler.get("type") != "command": + 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_generated_bash(command: Any) -> Optional[Tuple[str, str]]: + if not isinstance(command, str): return None - if handler.get("timeoutSec") != HOOK_TIMEOUT_SECONDS: + executable = _read_posix_argument(command, 0) + if executable is None or command[executable.next_index:executable.next_index + 1] != " ": return None - bash = handler.get("bash") - if not isinstance(bash, str): + 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_generated_powershell(command: Any) -> Optional[Tuple[str, str]]: + if not isinstance(command, str) or not command.startswith("& "): + return None + executable = _read_powershell_argument(command, 2) + if executable is None or command[executable.next_index:executable.next_index + 1] != " ": + return None + script = _read_powershell_argument(command, executable.next_index + 1) + if script is None or command[script.next_index:] != "; exit $LASTEXITCODE": + return None + return executable.value, script.value + + +def _exact_handler_keys(handler: JsonObject) -> bool: + return set(handler) == {"type", "bash", "powershell", "timeoutSec"} + + +def _is_python_executable(file_path: str) -> bool: + return os.path.isabs(file_path) and re.fullmatch( + r"(?:python|pypy)(?:[0-9]+(?:\.[0-9]+)*)?(?:\.exe)?", + os.path.basename(file_path), + re.IGNORECASE, + ) is not None + + +def _current_script_path(handler: JsonObject) -> Optional[str]: + if ( + not _exact_handler_keys(handler) + or handler.get("type") != "command" + or handler.get("timeoutSec") != HOOK_TIMEOUT_SECONDS + ): + return None + bash = _parse_generated_bash(handler.get("bash")) + powershell = _parse_generated_powershell(handler.get("powershell")) + if ( + bash is None + or powershell is None + or not _is_python_executable(bash[0]) + or not same_path(bash[0], powershell[0]) + or not same_path(bash[1], powershell[1]) + ): + return None + return bash[1] + + +def _prior_script_path(handler: JsonObject) -> Optional[str]: + if ( + not _exact_handler_keys(handler) + or handler.get("type") != "command" + or handler.get("timeoutSec") != HOOK_TIMEOUT_SECONDS + or not isinstance(handler.get("bash"), str) + ): + return None + bash_command = str(handler["bash"]) try: - arguments = shlex.split(bash) + arguments = shlex.split(bash_command) except ValueError: return None - if len(arguments) != 2 or not _same_path(arguments[0], sys.executable): + powershell = _parse_generated_powershell(handler.get("powershell")) + if ( + len(arguments) != 2 + or shlex.join(arguments) != bash_command + or powershell is None + or not _is_python_executable(arguments[0]) + or not same_path(arguments[0], powershell[0]) + or not same_path(arguments[1], powershell[1]) + ): return None - script_path = arguments[1] - expected_powershell = ( - f"& {_quote_powershell(sys.executable)} " - f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" - ) - return script_path if handler.get("powershell") == expected_powershell else None + return arguments[1] def _legacy_script_path(handler: JsonObject) -> Optional[str]: - if set(handler) != {"type", "bash", "powershell", "timeoutSec"}: + if ( + not _exact_handler_keys(handler) + or handler.get("type") != "command" + or handler.get("timeoutSec") != 5 + or not isinstance(handler.get("bash"), str) + or handler.get("bash") != handler.get("powershell") + ): return None - if handler.get("type") != "command" or handler.get("timeoutSec") != 5: - return None - bash = handler.get("bash") - powershell = handler.get("powershell") - if not isinstance(bash, str) or bash != powershell: - return None - command = re.fullmatch(r'"[^"]+"\s+(.+)', bash) - return command.group(1) if command else bash + command = str(handler["bash"]) + match = re.fullmatch(r'"[^"\r\n]+"\s+(.+)', command) + return match.group(1) if match else command -def _managed_agent_id( - handler: JsonObject, - script_name: str, - runtime_root: str, -) -> Optional[str]: - script_path = _generated_script_path(handler) or _legacy_script_path(handler) - if not script_path or os.path.basename(script_path) != script_name: +def _managed_entry( + handler: JsonObject, script_name: str +) -> Optional[_ManagedEntry]: + script_path = ( + _current_script_path(handler) + or _prior_script_path(handler) + or _legacy_script_path(handler) + ) + if ( + not script_path + or os.path.normcase(os.path.basename(script_path)) + != os.path.normcase(script_name) + ): return None agent_directory = os.path.dirname(script_path) - if not _same_path(os.path.dirname(agent_directory), runtime_root): + if not same_path(os.path.dirname(agent_directory), runtime_root()): return None agent_id = os.path.basename(agent_directory) - return agent_id if agent_id not in {"", ".", ".."} else None + if agent_id in ("", ".", ".."): + return None + return _ManagedEntry(agent_id, script_path) -def _validate_hooks(value: Any, label: str) -> CopilotHooks: - if value is None: - raise ValueError(f'{label} field "hooks" must be an object') - if not isinstance(value, dict): - raise ValueError(f'{label} field "hooks" must be an object') - hooks: CopilotHooks = {} - for event, handlers in value.items(): - if not isinstance(event, str): - raise ValueError(f"{label} hook event names must be strings") - if not isinstance(handlers, list): - raise ValueError(f'{label} field "hooks.{event}" must be an array') - if not all(isinstance(handler, dict) for handler in handlers): - raise ValueError( - f'{label} field "hooks.{event}" must contain objects' - ) - hooks[event] = list(handlers) - return hooks - - -def parse_document(file_path: str, raw: str, label: str) -> CopilotDocument: - try: - value = json.loads(raw) - except json.JSONDecodeError as error: - raise ValueError(f"Failed to parse {label} at {file_path}: {error}") from error - if not isinstance(value, dict): - raise ValueError(f"{label} at {file_path} must contain a JSON object") - if value.get("version") != 1: - raise ValueError(f"{label} at {file_path} must declare version 1") - hooks = _validate_hooks(value.get("hooks", {}), label) - return CopilotDocument(True, file_path, value, hooks, raw) +def parse_document( + file_path: str, + snapshot: FileSnapshot, + label: str, +) -> CopilotDocument: + document_label = f"{label} at {file_path}" + root = parse_json_object(snapshot.contents, document_label) + version = root.get("version") + if isinstance(version, bool) or version != 1: + raise ValueError(f"{document_label} must declare version 1") + if "disableAllHooks" in root and not isinstance( + root["disableAllHooks"], bool + ): + raise ValueError( + f'{document_label} field "disableAllHooks" must be a boolean' + ) + hooks = validate_hooks(root["hooks"], document_label) if "hooks" in root else {} + return CopilotDocument( + True, + file_path, + root, + hooks, + root.get("disableAllHooks") is True, + snapshot.contents, + snapshot, + ) def create_document(file_path: str) -> CopilotDocument: - return CopilotDocument(False, file_path, {}, {}) + return CopilotDocument(False, file_path, {}, {}, False) def remove_managed_hooks( - hooks: CopilotHooks, - runtime_root: str, - agent_id: str = "", + hooks: CopilotHooks, agent_id: str = "" ) -> CopilotHooks: - result = dict(hooks) - for event, script_name in ( - ("preToolUse", GUARD_SCRIPT), - ("postToolUse", AUDIT_SCRIPT), - ): + result = {event: list(handlers) for event, handlers in hooks.items()} + for event, script_name in MANAGED_EVENTS: handlers = [] for handler in result.get(event, []): - managed_id = _managed_agent_id(handler, script_name, runtime_root) - remove = managed_id is not None and ( - not agent_id or _same_agent_id(managed_id, agent_id) + managed = _managed_entry(handler, script_name) + remove = managed is not None and ( + not agent_id or same_agent_id(managed.agent_id, agent_id) ) if not remove: handlers.append(handler) @@ -190,7 +325,7 @@ def remove_managed_hooks( def _empty_owned_document(root: JsonObject, hooks: CopilotHooks) -> bool: - return not hooks and all(key in {"version", "hooks"} for key in root) + return not hooks and all(key in ("version", "hooks") for key in root) def render_document( @@ -206,37 +341,47 @@ def render_document( root["hooks"] = hooks else: root.pop("hooks", None) - next_source = json.dumps(root, indent=2) + "\n" - return RenderedDocument(document, next_source != document.raw, next_source) + next_source = json.dumps(root, indent=2, ensure_ascii=False) + "\n" + parse_document( + document.file_path, + FileSnapshot(next_source, 0, 0, 0), + "GitHub Copilot rendered hooks", + ) + return RenderedDocument( + document, + next_source != document.raw, + next_source, + ) -def _managed_ids( - handlers: List[JsonObject], - script_name: str, - runtime_root: str, -) -> Set[str]: - result: Set[str] = set() +def _managed_entries( + handlers: List[JsonObject], script_name: str +) -> Dict[str, List[_ManagedEntry]]: + result: Dict[str, List[_ManagedEntry]] = {} for handler in handlers: - agent_id = _managed_agent_id(handler, script_name, runtime_root) - if agent_id: - result.add(agent_id) + entry = _managed_entry(handler, script_name) + if entry is not None: + result.setdefault(os.path.normcase(entry.agent_id), []).append(entry) return result -def runtime_contracts( - hooks: CopilotHooks, - runtime_root: str, -) -> List[RuntimeContract]: - guards = _managed_ids(hooks.get("preToolUse", []), GUARD_SCRIPT, runtime_root) - audits = _managed_ids(hooks.get("postToolUse", []), AUDIT_SCRIPT, runtime_root) - contracts: List[RuntimeContract] = [] - for agent_id in sorted(guards): - if not any(_same_agent_id(agent_id, audit_id) for audit_id in audits): +def runtime_contracts(hooks: CopilotHooks) -> List[RuntimeContract]: + guards = _managed_entries(hooks.get("preToolUse", []), GUARD_SCRIPT) + successes = _managed_entries(hooks.get("postToolUse", []), AUDIT_SCRIPT) + failures = _managed_entries( + 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_path(success[0].script_path, failure[0].script_path): continue - agent_directory = os.path.join(runtime_root, agent_id) contracts.append(RuntimeContract( - agent_id, - os.path.join(agent_directory, GUARD_SCRIPT), - os.path.join(agent_directory, AUDIT_SCRIPT), + guard[0].agent_id, + guard[0].script_path, + success[0].script_path, )) return contracts diff --git a/sdks/python/elydora/plugins/copilot_installation.py b/sdks/python/elydora/plugins/copilot_installation.py new file mode 100644 index 0000000..0a63e41 --- /dev/null +++ b/sdks/python/elydora/plugins/copilot_installation.py @@ -0,0 +1,277 @@ +"""Transactional GitHub Copilot CLI hook installation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +import sys +from typing import List, Optional, Sequence, Tuple + +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, + FilePrecondition, + file_change, + source_change, + write_changes, +) +from .base import InstallConfig +from .copilot_contract import ( + AGENT_KEY, + AUDIT_SCRIPT, + GUARD_SCRIPT, + CopilotSources, + RenderedDocument, + runtime_root, + same_path, +) +from .copilot_io import ( + require_hooks_enabled, + validate_api_origin, + validate_private_key, + validate_runtime_tree, +) +from .copilot_schema import validate_javascript_regexes +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script + + +@dataclass(frozen=True) +class CopilotRuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +@dataclass(frozen=True) +class PreparedCopilotInstallation: + changes: Tuple[FileChange, ...] + preconditions: Tuple[FilePrecondition, ...] + + +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) -> CopilotRuntimePaths: + 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_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + "Elydora guard runtime must use the managed agent directory: " + f"{guard_path}" + ) + return CopilotRuntimePaths( + 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"GitHub Copilot CLI installation requires agent_name {AGENT_KEY}" + ) + if not os.path.isabs(sys.executable): + raise RuntimeError( + "GitHub Copilot CLI requires an absolute Python executable path" + ) + 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_copilot_installation( + config: InstallConfig, + sources: CopilotSources, +) -> CopilotRuntimePaths: + if not sources.user.file_path: + raise ValueError( + "GitHub Copilot CLI installation requires a user hook path" + ) + require_hooks_enabled(sources) + hook_sources = [sources.user.hooks] + if sources.legacy is not None: + hook_sources.append(sources.legacy.hooks) + validate_javascript_regexes(hook_sources) + _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 _document_change(rendered: RenderedDocument) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.file_path, + "GitHub Copilot hook source", + rendered.document.raw, + rendered.next_source, + 0o600, + MAX_SOURCE_BYTES, + rendered.document.snapshot, + ) + + +def _present( + changes: Sequence[Optional[FileChange]], +) -> Tuple[FileChange, ...]: + return tuple(change for change in changes if change is not None) + + +def _document_preconditions( + rendered: Sequence[RenderedDocument], +) -> Tuple[FilePrecondition, ...]: + return tuple( + FilePrecondition( + item.document.file_path, + "GitHub Copilot hook source", + item.document.snapshot, + MAX_SOURCE_BYTES, + ) + for item in rendered + if not item.changed + ) + + +def _settings_preconditions( + sources: CopilotSources, +) -> Tuple[FilePrecondition, ...]: + return tuple( + FilePrecondition( + item.file_path, + item.label, + item.snapshot, + MAX_SOURCE_BYTES, + ) + for item in sources.settings_preconditions + ) + + +def _validate_rendered_documents( + sources: CopilotSources, + rendered: Sequence[RenderedDocument], +) -> None: + expected = [sources.user] + if sources.legacy is not None: + expected.append(sources.legacy) + if len(rendered) != len(expected): + raise ValueError("GitHub Copilot rendered source set is incomplete") + for document in expected: + matches = [ + item for item in rendered + if same_path(item.document.file_path, document.file_path) + ] + if len(matches) != 1: + raise ValueError( + "GitHub Copilot rendered source set contains unexpected paths" + ) + + +def prepare_copilot_installation( + config: InstallConfig, + sources: CopilotSources, + rendered: Sequence[RenderedDocument], +) -> PreparedCopilotInstallation: + paths = preflight_copilot_installation(config, sources) + _validate_rendered_documents(sources, rendered) + guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) + audit_script = generate_hook_script( + org_id=config.get("org_id", ""), + agent_id=paths.agent_id, + kid=config.get("kid", ""), + base_url=config.get("base_url", "https://api.elydora.com"), + native_payload=True, + agent_name=AGENT_KEY, + ) + changes = _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 + ), + *(_document_change(item) for item in rendered), + )) + preconditions = ( + *_settings_preconditions(sources), + *_document_preconditions(rendered), + ) + return PreparedCopilotInstallation(changes, preconditions) + + +def commit_copilot_installation( + prepared: PreparedCopilotInstallation, +) -> None: + write_changes( + prepared.changes, + "Install GitHub Copilot hooks", + prepared.preconditions, + ) + + +def prepare_copilot_uninstall( + rendered: Sequence[RenderedDocument], +) -> Tuple[FileChange, ...]: + return _present(tuple(_document_change(item) for item in rendered)) + + +def commit_copilot_uninstall(changes: Sequence[FileChange]) -> None: + write_changes(changes, "Uninstall GitHub Copilot hooks") diff --git a/sdks/python/elydora/plugins/copilot_io.py b/sdks/python/elydora/plugins/copilot_io.py index 2781b8b..ae88f9e 100644 --- a/sdks/python/elydora/plugins/copilot_io.py +++ b/sdks/python/elydora/plugins/copilot_io.py @@ -1,96 +1,401 @@ -"""GitHub Copilot hook source and runtime I/O.""" +"""GitHub Copilot CLI source discovery and managed runtime validation.""" from __future__ import annotations -import json +import base64 +from dataclasses import dataclass import os -from typing import Any, Dict, List, Optional +from typing import Any, List, Optional, Tuple +import urllib.parse -from ._transaction import ( - FileChange, - read_optional, - regular_file_exists, - source_change, +from ._jsonc import parse_jsonc +from ._managed_files import ( + FileSnapshot, + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + MAX_SOURCE_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, ) +from ._strict_json import JsonObject, parse_json_object from .copilot_contract import ( AGENT_KEY, + AUDIT_SCRIPT, CONFIG_FILE, + GUARD_SCRIPT, CopilotDocument, CopilotSources, - RenderedDocument, RuntimeContract, + SourcePrecondition, create_document, parse_document, + runtime_root, + same_agent_id, + same_path, ) +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script -JsonObject = Dict[str, Any] +@dataclass(frozen=True) +class SettingsLayer: + file_path: str + label: str + jsonc: bool -def config_paths(home_dir: str) -> tuple[str, str]: - override = os.environ.get("COPILOT_HOME", "").strip() - copilot_home = override or os.path.join(home_dir, ".copilot") - user_path = os.path.join(copilot_home, "hooks", CONFIG_FILE) - legacy_path = os.path.join(os.getcwd(), ".github", "hooks", "hooks.json") - return user_path, legacy_path +@dataclass(frozen=True) +class CopilotPaths: + copilot_home: str + user_hooks_directory: str + user_hook_path: str + legacy_hook_path: str + settings_layers: Tuple[SettingsLayer, ...] + inspected_directories: Tuple[Tuple[str, str], ...] -def _read_document(file_path: str, label: str) -> Optional[CopilotDocument]: - raw = read_optional(file_path, label) - return None if raw is None else parse_document(file_path, raw, label) +@dataclass(frozen=True) +class _ParsedSettingsLayer: + layer: SettingsLayer + disable_all_hooks: Optional[bool] + snapshot: Optional[FileSnapshot] -def read_sources(home_dir: str) -> CopilotSources: - user_path, legacy_path = config_paths(home_dir) - user = _read_document(user_path, "GitHub Copilot user hooks") - legacy = _read_document(legacy_path, "GitHub Copilot legacy project hooks") - return CopilotSources(user or create_document(user_path), legacy) - - -def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: - if not rendered.changed: - return None - return source_change( - rendered.document.file_path, - "GitHub Copilot hook source", - rendered.document.raw, - rendered.next_source, - 0o600, +def resolve_copilot_paths(home_dir: Optional[str] = None) -> CopilotPaths: + home = home_dir if home_dir is not None else os.path.expanduser("~") + override = os.environ.get("COPILOT_HOME") + copilot_home = override if override else os.path.join(home, ".copilot") + project = os.getcwd() + github = os.path.join(project, ".github") + github_copilot = os.path.join(github, "copilot") + github_hooks = os.path.join(github, "hooks") + claude = os.path.join(project, ".claude") + user_hooks = os.path.join(copilot_home, "hooks") + return CopilotPaths( + copilot_home=copilot_home, + user_hooks_directory=user_hooks, + user_hook_path=os.path.join(user_hooks, CONFIG_FILE), + legacy_hook_path=os.path.join(github_hooks, "hooks.json"), + settings_layers=( + SettingsLayer( + os.path.join(copilot_home, "config.json"), + "legacy Copilot user config", + False, + ), + SettingsLayer( + os.path.join(copilot_home, "settings.json"), + "Copilot user settings", + True, + ), + SettingsLayer( + os.path.join(claude, "settings.json"), + "Claude repository settings", + True, + ), + SettingsLayer( + os.path.join(claude, "settings.local.json"), + "Claude local settings", + True, + ), + SettingsLayer( + os.path.join(github_copilot, "settings.json"), + "Copilot repository settings", + True, + ), + SettingsLayer( + os.path.join(github_copilot, "settings.local.json"), + "Copilot local settings", + True, + ), + ), + inspected_directories=( + (project, "Copilot working directory"), + (copilot_home, "COPILOT_HOME"), + (user_hooks, "Copilot user hooks directory"), + (github, "GitHub configuration directory"), + (github_hooks, "GitHub repository hooks directory"), + (github_copilot, "Copilot repository settings directory"), + (claude, "Claude repository settings directory"), + ), ) -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - raw = read_optional(file_path, "Elydora runtime config") - if raw is None: - return None - try: - value = json.loads(raw) - except json.JSONDecodeError as error: - raise ValueError( - f"Failed to parse Elydora runtime config at {file_path}: {error}" - ) from error +def _inspect_directories(locations: Tuple[Tuple[str, str], ...]) -> None: + for directory, label in locations: + physical_directory_exists(directory, label) + + +def _read_hook_document( + file_path: str, label: str +) -> Optional[CopilotDocument]: + snapshot = read_physical_file(file_path, label, MAX_SOURCE_BYTES) + return None if snapshot is None else parse_document(file_path, snapshot, label) + + +def _parse_settings(raw: str, layer: SettingsLayer) -> JsonObject: + if not raw.strip(): + return {} + label = f"{layer.label} at {layer.file_path}" + value = ( + parse_jsonc(raw, label, allow_trailing_commas=True) + if layer.jsonc + else parse_json_object(raw, label) + ) if not isinstance(value, dict): + raise ValueError(f"{label} must contain a JSON object") + return value + + +def _read_settings_layer(layer: SettingsLayer) -> _ParsedSettingsLayer: + snapshot = read_physical_file( + layer.file_path, layer.label, MAX_SOURCE_BYTES + ) + if snapshot is None: + return _ParsedSettingsLayer(layer, None, None) + root = _parse_settings(snapshot.contents, layer) + disabled = root.get("disableAllHooks") + if "disableAllHooks" in root and not isinstance(disabled, bool): raise ValueError( - f"Elydora runtime config at {file_path} must contain a JSON object" + f'{layer.label} at {layer.file_path} field "disableAllHooks" ' + "must be a boolean" + ) + return _ParsedSettingsLayer(layer, disabled, snapshot) + + +def _effective_disabled_source( + layers: List[_ParsedSettingsLayer], +) -> Optional[str]: + disabled_by: Optional[str] = None + for item in layers: + source = f"{item.layer.label} at {item.layer.file_path}" + if item.disable_all_hooks is True: + disabled_by = source + elif item.disable_all_hooks is False: + disabled_by = None + return disabled_by + + +def read_sources(home_dir: Optional[str] = None) -> CopilotSources: + paths = resolve_copilot_paths(home_dir) + _inspect_directories(paths.inspected_directories) + user = _read_hook_document( + paths.user_hook_path, "GitHub Copilot user hooks" + ) + legacy = _read_hook_document( + paths.legacy_hook_path, "GitHub Copilot legacy project hooks" + ) + layers = [_read_settings_layer(layer) for layer in paths.settings_layers] + user_document = user or create_document(paths.user_hook_path) + disabled_by = ( + f"GitHub Copilot user hooks at {paths.user_hook_path}" + if user_document.hooks_disabled + else _effective_disabled_source(layers) + ) + preconditions = tuple( + SourcePrecondition( + item.layer.file_path, + item.layer.label, + item.snapshot, + ) + for item in layers + ) + return CopilotSources( + user_document, + legacy, + disabled_by, + preconditions, + ) + + +def require_hooks_enabled(sources: CopilotSources) -> None: + if sources.disabled_by: + raise ValueError( + "GitHub Copilot hooks are disabled by " + f"{sources.disabled_by}; set disableAllHooks to false before installation" + ) + + +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_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, + contract: RuntimeContract, + 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_string(config.get("org_id"), "org_id", config_path) + _require_string(config.get("kid"), "kid", config_path) + agent_id = _require_string(config.get("agent_id"), "agent_id", config_path) + if ( + not same_agent_id(agent_id, contract.agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + f"Elydora runtime identity does not match Copilot hooks: {config_path}" + ) + if "token" in config: + _require_string(config.get("token"), "token", config_path) + base_url = _require_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_SCRIPT, "Elydora guard runtime"), + (AUDIT_SCRIPT, "Elydora audit runtime"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), + ) + artifact_exists = any( + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + configured_id = config.get("agent_id") + if ( + not isinstance(configured_id, str) + or not same_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime config identity does not match Copilot agent " + f"{agent_id}: {config_path}" + ) + + +def _valid_contract_paths(contract: RuntimeContract) -> bool: + agent_directory = os.path.dirname(contract.guard_path) + return ( + same_path(os.path.dirname(agent_directory), runtime_root()) + and same_path( + contract.guard_path, + os.path.join(agent_directory, GUARD_SCRIPT), + ) + and same_path( + contract.audit_path, + os.path.join(agent_directory, AUDIT_SCRIPT), + ) + ) + + +def _runtime_contract_exists(contract: RuntimeContract) -> bool: + if not _valid_contract_paths(contract): + return False + root = runtime_root() + agent_directory = os.path.dirname(contract.guard_path) + 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, config_path) + validate_private_key(key.contents, "Elydora private key") + expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id) + expected_audit = generate_hook_script( + org_id="", + agent_id=contract.agent_id, + kid="", + base_url="", + native_payload=True, + agent_name=AGENT_KEY, + ) + return guard.contents == expected_guard and audit.contents == expected_audit + + def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: - for contract in contracts: - agent_directory = os.path.dirname(contract.guard_path) - config_path = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path) - if ( - config is None - or config.get("agent_name") != AGENT_KEY - or config.get("agent_id") != contract.agent_id - ): - continue - if regular_file_exists( - contract.guard_path, "Elydora guard runtime" - ) and regular_file_exists( - contract.audit_path, "Elydora audit runtime" - ): - return True - return False + return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/sdks/python/elydora/plugins/copilot_schema.py b/sdks/python/elydora/plugins/copilot_schema.py new file mode 100644 index 0000000..cb5ba30 --- /dev/null +++ b/sdks/python/elydora/plugins/copilot_schema.py @@ -0,0 +1,276 @@ +"""GitHub Copilot CLI hook schema validation.""" + +from __future__ import annotations + +import math +import os +import json +import shutil +import subprocess # nosec B404 - validates official JavaScript regex syntax +from typing import Any, Dict, List +import urllib.parse + + +JsonObject = Dict[str, Any] +CopilotHooks = Dict[str, List[JsonObject]] + +SUPPORTED_EVENTS = frozenset({ + "agentStop", + "Stop", + "errorOccurred", + "ErrorOccurred", + "notification", + "Notification", + "permissionRequest", + "PermissionRequest", + "postToolUse", + "PostToolUse", + "postToolUseFailure", + "PostToolUseFailure", + "preCompact", + "PreCompact", + "preToolUse", + "PreToolUse", + "sessionEnd", + "SessionEnd", + "sessionStart", + "SessionStart", + "subagentStart", + "SubagentStart", + "subagentStop", + "SubagentStop", + "userPromptSubmitted", + "UserPromptSubmit", + "userPromptTransformed", +}) + +_REGEX_TIMEOUT_SECONDS = 10 +_REGEX_VALIDATOR = """import fs from "node:fs"; +const entries = JSON.parse(fs.readFileSync(0, "utf8")); +for (const entry of entries) { + try { + new RegExp("^(?:" + entry.pattern + ")$"); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + process.stderr.write(entry.label + ": " + message); + process.exit(1); + } +} +""" + + +def _field_label(label: str, field: str) -> str: + return f'{label} field "{field}"' + + +def _require_string(value: Any, label: str) -> str: + if not isinstance(value, str) or not value: + raise ValueError(f"{label} must be a non-empty string") + return value + + +def _validate_optional_string( + handler: JsonObject, field: str, label: str +) -> None: + if field in handler and not isinstance(handler[field], str): + raise ValueError(f"{_field_label(label, field)} must be a string") + + +def _validate_timeout(handler: JsonObject, label: str) -> None: + for field in ("timeout", "timeoutSec"): + if field not in handler: + continue + value = handler[field] + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not math.isfinite(value) + or value <= 0 + ): + raise ValueError( + f"{_field_label(label, field)} must be a positive number" + ) + + +def _validate_matcher(handler: JsonObject, event: str, label: str) -> None: + if "matcher" not in handler: + return + matcher = _require_string( + handler["matcher"], _field_label(label, "matcher") + ) + if event in ( + "preToolUse", + "PreToolUse", + "permissionRequest", + "PermissionRequest", + ) and matcher in ("*", "**"): + return + + +def _validate_string_map(value: Any, label: str) -> None: + if not isinstance(value, dict): + raise ValueError(f"{label} must be an object") + for key, item in value.items(): + if not isinstance(key, str): + raise ValueError(f"{label} keys must be strings") + _require_string(item, f"{label}.{key}") + + +def _validate_command( + handler: JsonObject, event: str, label: str +) -> None: + command_fields = tuple( + field for field in ("bash", "powershell", "command") if field in handler + ) + if not command_fields: + raise ValueError(f"{label} must define bash, powershell, or command") + for field in ("bash", "powershell", "command", "cwd"): + _validate_optional_string(handler, field, label) + if "env" in handler: + _validate_string_map(handler["env"], _field_label(label, "env")) + _validate_timeout(handler, label) + _validate_matcher(handler, event, label) + + +def _is_loopback(hostname: str) -> bool: + return hostname in ("localhost", "::1") or hostname.startswith("127.") + + +def _validate_http(handler: JsonObject, event: str, label: str) -> None: + raw_url = _require_string(handler.get("url"), _field_label(label, "url")) + try: + parsed = urllib.parse.urlsplit(raw_url) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError(f"{_field_label(label, 'url')} is invalid") from error + localhost_allowed = ( + os.environ.get("COPILOT_HOOK_ALLOW_LOCALHOST") == "1" + and parsed.scheme == "http" + and hostname is not None + and _is_loopback(hostname) + ) + if ( + (parsed.scheme != "https" and not localhost_allowed) + or not parsed.netloc + or hostname is None + ): + raise ValueError(f"{_field_label(label, 'url')} must use HTTPS") + if "headers" in handler: + _validate_string_map( + handler["headers"], _field_label(label, "headers") + ) + if "allowedEnvVars" in handler: + value = handler["allowedEnvVars"] + if not isinstance(value, list) or any( + not isinstance(item, str) or not item for item in value + ): + raise ValueError( + f"{_field_label(label, 'allowedEnvVars')} must be an array of strings" + ) + _validate_timeout(handler, label) + _validate_matcher(handler, event, label) + + +def _validate_prompt(handler: JsonObject, event: str, label: str) -> None: + if event not in ("sessionStart", "SessionStart"): + raise ValueError( + f"{label} prompt hooks are supported only for sessionStart" + ) + _require_string(handler.get("prompt"), _field_label(label, "prompt")) + + +def _validate_handler(handler: JsonObject, event: str, label: str) -> None: + handler_type = handler.get("type") + if handler_type is None: + handler_type = "command" + if handler_type == "command": + _validate_command(handler, event, label) + elif handler_type == "http": + _validate_http(handler, event, label) + elif handler_type == "prompt": + _validate_prompt(handler, event, label) + else: + raise ValueError( + f'{_field_label(label, "type")} is unsupported' + ) + + +def validate_hooks(value: Any, label: str) -> CopilotHooks: + if not isinstance(value, dict): + raise ValueError(f'{label} field "hooks" must be an object') + hooks: CopilotHooks = {} + for event, handlers in value.items(): + if not isinstance(event, str) or event not in SUPPORTED_EVENTS: + raise ValueError(f'{label} hook event "{event}" is unsupported') + if not isinstance(handlers, list): + raise ValueError(f'{label} field "hooks.{event}" must be an array') + validated = [] + for index, handler in enumerate(handlers): + item_label = f"{label} handler hooks.{event}[{index}]" + if not isinstance(handler, dict): + raise ValueError(f"{item_label} must be an object") + _validate_handler(handler, event, item_label) + validated.append(dict(handler)) + hooks[event] = validated + return hooks + + +def _matcher_entries(hooks: CopilotHooks) -> List[JsonObject]: + entries = [] + wildcard_events = { + "preToolUse", + "PreToolUse", + "permissionRequest", + "PermissionRequest", + } + for event, handlers in hooks.items(): + for index, handler in enumerate(handlers): + matcher = handler.get("matcher") + if not isinstance(matcher, str): + continue + if event in wildcard_events and matcher in ("*", "**"): + continue + entries.append({ + "label": f"GitHub Copilot hooks.{event}[{index}] matcher", + "pattern": matcher, + }) + return entries + + +def validate_javascript_regexes(sources: List[CopilotHooks]) -> None: + entries = [entry for hooks in sources for entry in _matcher_entries(hooks)] + if not entries: + return + node_path = shutil.which("node") + if node_path is None: + raise FileNotFoundError( + "Node.js runtime is required to validate GitHub Copilot hook matchers" + ) + try: + result = subprocess.run( # nosec B603 + [node_path, "--input-type=module", "--eval", _REGEX_VALIDATOR], + input=json.dumps(entries), + text=True, + capture_output=True, + check=False, + timeout=_REGEX_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as error: + raise TimeoutError( + "GitHub Copilot hook matcher validation timed out after " + f"{_REGEX_TIMEOUT_SECONDS} seconds" + ) from error + except OSError as error: + raise OSError(f"Run Node.js Copilot matcher validator: {error}") from error + if result.returncode == 0: + return + message = ( + result.stderr.strip() + or result.stdout.strip() + or f"Node.js exited with code {result.returncode}" + ) + raise ValueError( + "GitHub Copilot matcher must be a valid JavaScript regular expression: " + f"{message}" + ) diff --git a/sdks/python/elydora/plugins/registry.py b/sdks/python/elydora/plugins/registry.py index c0c47bf..33aa597 100644 --- a/sdks/python/elydora/plugins/registry.py +++ b/sdks/python/elydora/plugins/registry.py @@ -28,7 +28,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "copilot": { "name": "GitHub Copilot CLI", - "hook_event": "preToolUse/postToolUse", + "hook_event": "preToolUse/postToolUse/postToolUseFailure", "config_path": "~/.copilot/hooks/elydora-audit.json", }, "cursor": { diff --git a/sdks/python/tests/copilot_support.py b/sdks/python/tests/copilot_support.py new file mode 100644 index 0000000..36db225 --- /dev/null +++ b/sdks/python/tests/copilot_support.py @@ -0,0 +1,216 @@ +from __future__ import annotations + +import base64 +from dataclasses import dataclass +import json +import os +from pathlib import Path +import subprocess +import sys +from typing import Any + +import pytest + +from elydora.plugins import copilot +from elydora.plugins.base import InstallConfig + + +AGENT_ID = "agent-1" +VALID_PRIVATE_KEY = base64.urlsafe_b64encode(bytes([11]) * 32).rstrip( + b"=" +).decode("ascii") +MISSING = object() + + +@dataclass(frozen=True) +class CopilotFixture: + plugin: copilot.CopilotPlugin + config: InstallConfig + home_dir: Path + copilot_home: Path + project_dir: Path + agent_dir: Path + config_path: Path + legacy_path: Path + guard_path: Path + hook_path: Path + runtime_config_path: Path + private_key_path: Path + user_settings_path: Path + legacy_user_config_path: Path + claude_settings_path: Path + claude_local_settings_path: Path + repository_settings_path: Path + local_settings_path: Path + + +def write_json_or_text(path: Path, value: object) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + source = ( + value + if isinstance(value, str) + else json.dumps(value, indent=2, ensure_ascii=False) + "\n" + ) + path.write_text(source, encoding="utf-8") + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + user_config: object = MISSING, + legacy_config: object = MISSING, + user_settings: object = MISSING, + legacy_user_config: object = MISSING, + claude_settings: object = MISSING, + claude_local_settings: object = MISSING, + repository_settings: object = MISSING, + local_settings: object = MISSING, +) -> CopilotFixture: + home_dir = tmp_path / "home with spaces and 'quote %COPILOT%" + project_dir = tmp_path / "project with spaces" + copilot_home = home_dir / "custom Copilot 'home" + agent_dir = home_dir / ".elydora" / AGENT_ID + config_path = copilot_home / "hooks" / "elydora-audit.json" + legacy_path = project_dir / ".github" / "hooks" / "hooks.json" + user_settings_path = copilot_home / "settings.json" + legacy_user_config_path = copilot_home / "config.json" + claude_settings_path = project_dir / ".claude" / "settings.json" + claude_local_settings_path = ( + project_dir / ".claude" / "settings.local.json" + ) + repository_settings_path = ( + project_dir / ".github" / "copilot" / "settings.json" + ) + local_settings_path = ( + project_dir / ".github" / "copilot" / "settings.local.json" + ) + project_dir.mkdir(parents=True) + optional = ( + (config_path, user_config), + (legacy_path, legacy_config), + (user_settings_path, user_settings), + (legacy_user_config_path, legacy_user_config), + (claude_settings_path, claude_settings), + (claude_local_settings_path, claude_local_settings), + (repository_settings_path, repository_settings), + (local_settings_path, local_settings), + ) + for path, value in optional: + if value is not MISSING: + write_json_or_text(path, value) + + monkeypatch.chdir(project_dir) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.setenv("COPILOT_HOME", str(copilot_home)) + config: InstallConfig = { + "agent_id": AGENT_ID, + "agent_name": "copilot", + "org_id": "org-1", + "private_key": VALID_PRIVATE_KEY, + "kid": "kid-1", + "token": "token-1", + "base_url": "http://127.0.0.1:9", + "guard_script_path": str(agent_dir / "guard.py"), + } + return CopilotFixture( + plugin=copilot.CopilotPlugin(), + config=config, + home_dir=home_dir, + copilot_home=copilot_home, + project_dir=project_dir, + agent_dir=agent_dir, + config_path=config_path, + legacy_path=legacy_path, + guard_path=agent_dir / "guard.py", + hook_path=agent_dir / "hook.py", + runtime_config_path=agent_dir / "config.json", + private_key_path=agent_dir / "private.key", + user_settings_path=user_settings_path, + legacy_user_config_path=legacy_user_config_path, + claude_settings_path=claude_settings_path, + claude_local_settings_path=claude_local_settings_path, + repository_settings_path=repository_settings_path, + local_settings_path=local_settings_path, + ) + + +def managed_handler( + settings: dict[str, Any], event: str, script_name: str +) -> dict[str, Any]: + for handler in settings.get("hooks", {}).get(event, []): + if script_name in str(handler.get("bash")): + return handler + raise AssertionError(f"managed {event} handler not found") + + +def assert_native_handler(handler: dict[str, Any]) -> None: + assert set(handler) == {"type", "bash", "powershell", "timeoutSec"} + assert handler["type"] == "command" + assert handler["timeoutSec"] == 10 + assert handler["bash"].startswith("'") + assert Path(sys.executable).name in handler["bash"] + assert handler["powershell"].startswith("& '") + assert handler["powershell"].endswith("; exit $LASTEXITCODE") + + +def legacy_managed_config( + fixture: CopilotFixture, + extra_hooks: dict[str, Any] | None = None, +) -> dict[str, Any]: + return { + "version": 1, + "hooks": { + "preToolUse": [{ + "type": "command", + "bash": f'"{sys.executable}" {fixture.guard_path}', + "powershell": f'"{sys.executable}" {fixture.guard_path}', + "timeoutSec": 5, + }], + "postToolUse": [{ + "type": "command", + "bash": str(fixture.hook_path), + "powershell": str(fixture.hook_path), + "timeoutSec": 5, + }], + **(extra_hooks or {}), + }, + } + + +def run_hook( + handler: dict[str, Any], + fixture: CopilotFixture, + payload: str, +) -> subprocess.CompletedProcess[str]: + if os.name == "nt": + command = [ + "powershell.exe", + "-NoProfile", + "-NonInteractive", + "-Command", + handler["powershell"], + ] + else: + command = ["/bin/sh", "-c", handler["bash"]] + return subprocess.run( + command, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env={ + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + }, + input=payload, + text=True, + ) + + +def assert_runtime_absent(fixture: CopilotFixture) -> None: + assert fixture.guard_path.exists() is False + assert fixture.hook_path.exists() is False + assert fixture.runtime_config_path.exists() is False + assert fixture.private_key_path.exists() is False diff --git a/sdks/python/tests/test_cli_security.py b/sdks/python/tests/test_cli_security.py index 2496744..e1d0940 100644 --- a/sdks/python/tests/test_cli_security.py +++ b/sdks/python/tests/test_cli_security.py @@ -390,9 +390,9 @@ def test_legacy_plugins_persist_one_owner_only_private_key( runtime_root = case_root / ".elydora" guard_path = runtime_root / "agent-1" / "guard.py" guard_path.parent.mkdir(parents=True) - if module is not cursor: + if module not in (cursor, copilot): guard_path.write_text("pass\n", encoding="utf-8") - if module is cursor: + if module in (cursor, copilot): monkeypatch.setenv("HOME", str(case_root)) monkeypatch.setenv("USERPROFILE", str(case_root)) else: @@ -410,6 +410,7 @@ def test_legacy_plugins_persist_one_owner_only_private_key( monkeypatch.setattr(module, "PLUGIN_DIR", str(case_root / "plugins")) elif module is copilot: case_root.mkdir(parents=True, exist_ok=True) + monkeypatch.setenv("COPILOT_HOME", str(case_root / ".copilot")) monkeypatch.chdir(case_root) plugin_type().install( diff --git a/sdks/python/tests/test_copilot_official.py b/sdks/python/tests/test_copilot_official.py new file mode 100644 index 0000000..015d8a0 --- /dev/null +++ b/sdks/python/tests/test_copilot_official.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import shutil +import subprocess + +import pytest + +from copilot_support import prepare_fixture + + +COPILOT_ENTRY = os.environ.get("ELYDORA_COPILOT_ENTRY") +COPILOT_RUNTIME_ENTRY = os.environ.get("ELYDORA_COPILOT_RUNTIME_ENTRY") +NODE = shutil.which("node") + + +@pytest.mark.skipif( + not COPILOT_ENTRY or not COPILOT_RUNTIME_ENTRY or not NODE, + reason=( + "set ELYDORA_COPILOT_ENTRY and ELYDORA_COPILOT_RUNTIME_ENTRY " + "to official package files" + ), +) +def test_official_copilot_1_0_71_loads_all_three_managed_hooks( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + "COPILOT_HOME": str(fixture.copilot_home), + } + version = subprocess.run( + [str(NODE), str(COPILOT_ENTRY), "--version"], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + text=True, + ) + assert version.returncode == 0, version.stderr + assert "GitHub Copilot CLI 1.0.71." in version.stdout + + source = """ +import { createRequire } from 'node:module'; +const require = createRequire(import.meta.url); +const runtime = require(process.env.ELYDORA_COPILOT_RUNTIME_ENTRY); +const session = await runtime.hookSessionCreate({ + cwd: process.env.ELYDORA_PROJECT, + repoRoot: process.env.ELYDORA_PROJECT, + sessionId: 'elydora-python-official-test', + settingsJson: '{}', + userHooksDir: process.env.ELYDORA_HOOKS, + allowLocalhost: false, + allowHttpAuthHooks: false, + discoverPolicies: false, +}); +try { + const snapshot = JSON.parse(await runtime.hookSessionSnapshot(session.handle)); + console.log(JSON.stringify({ load: session.load, snapshot })); +} finally { + runtime.hookSessionDispose(session.handle); +} +""" + loaded = subprocess.run( + [str(NODE), "--input-type=module", "--eval", source], + capture_output=True, + check=False, + cwd=fixture.project_dir, + env={ + **environment, + "ELYDORA_COPILOT_RUNTIME_ENTRY": str(COPILOT_RUNTIME_ENTRY), + "ELYDORA_HOOKS": str(fixture.config_path.parent), + "ELYDORA_PROJECT": str(fixture.project_dir), + }, + text=True, + ) + assert loaded.returncode == 0, loaded.stderr + result = json.loads(loaded.stdout) + assert result["load"]["hookCount"] == 3 + assert result["load"]["errors"] == [] + assert result["load"]["warnings"] == [] + hooks = result["snapshot"]["hooks"] + assert sorted(hook["eventName"] for hook in hooks) == [ + "postToolUse", + "postToolUseFailure", + "preToolUse", + ] + for hook in hooks: + assert Path(hook["source"]).name == "elydora-audit.json" + spec = json.loads(hook["specJson"]) + assert spec["config"]["type"] == "command" + assert spec["config"]["timeoutSec"] == 10 + assert "python" in spec["config"]["bash"].lower() + assert spec["config"]["powershell"].startswith("& ") diff --git a/sdks/python/tests/test_copilot_plugin.py b/sdks/python/tests/test_copilot_plugin.py index c36f968..cdfd6bc 100644 --- a/sdks/python/tests/test_copilot_plugin.py +++ b/sdks/python/tests/test_copilot_plugin.py @@ -1,183 +1,39 @@ from __future__ import annotations -from dataclasses import dataclass import json import os from pathlib import Path -import subprocess -import sys -from typing import Any import pytest from elydora import cli -from elydora.plugins import _transaction, copilot -from elydora.plugins.base import InstallConfig +from elydora.plugins import copilot from elydora.plugins.registry import SUPPORTED_AGENTS - -AGENT_ID = "agent-1" -MISSING = object() +from copilot_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + assert_native_handler, + assert_runtime_absent, + legacy_managed_config, + managed_handler, + prepare_fixture, + run_hook, + write_json_or_text, +) -@dataclass(frozen=True) -class CopilotFixture: - plugin: copilot.CopilotPlugin - config: InstallConfig - home_dir: Path - copilot_home: Path - project_dir: Path - agent_dir: Path - config_path: Path - legacy_path: Path - guard_path: Path - hook_path: Path - runtime_config_path: Path - private_key_path: Path - - -def write_json_or_text(path: Path, value: object) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - source = value if isinstance(value, str) else json.dumps(value, indent=2) - path.write_text(source, encoding="utf-8") - - -def prepare_fixture( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - *, - user_config: object = MISSING, - legacy_config: object = MISSING, - create_guard: bool = True, -) -> CopilotFixture: - home_dir = tmp_path / "home with spaces" - project_dir = tmp_path / "project with spaces" - copilot_home = home_dir / "custom copilot" - agent_dir = home_dir / ".elydora" / AGENT_ID - config_path = copilot_home / "hooks" / "elydora-audit.json" - legacy_path = project_dir / ".github" / "hooks" / "hooks.json" - guard_path = agent_dir / "guard.py" - hook_path = agent_dir / "hook.py" - agent_dir.mkdir(parents=True) - project_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 user_config is not MISSING: - write_json_or_text(config_path, user_config) - if legacy_config is not MISSING: - write_json_or_text(legacy_path, legacy_config) - - monkeypatch.chdir(project_dir) - monkeypatch.setattr(copilot, "ELYDORA_DIR", str(agent_dir.parent)) - monkeypatch.setattr(copilot, "_home_dir", lambda: str(home_dir), raising=False) - monkeypatch.setenv("COPILOT_HOME", str(copilot_home)) - config: InstallConfig = { - "agent_id": AGENT_ID, - "agent_name": "copilot", - "org_id": "org-1", - "private_key": "test-key", - "kid": "kid-1", - "base_url": "https://api.elydora.test", - "guard_script_path": str(guard_path), - } - return CopilotFixture( - plugin=copilot.CopilotPlugin(), - config=config, - home_dir=home_dir, - copilot_home=copilot_home, - project_dir=project_dir, - agent_dir=agent_dir, - config_path=config_path, - legacy_path=legacy_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 handler in settings.get("hooks", {}).get(event, []): - if script_name in str(handler.get("bash")): - return handler - raise AssertionError(f"managed {event} handler not found") - - -def assert_native_handler(handler: dict[str, Any]) -> None: - assert set(handler) == {"type", "bash", "powershell", "timeoutSec"} - assert handler["type"] == "command" - assert handler["timeoutSec"] == 10 - assert Path(sys.executable).name in handler["bash"] - assert handler["powershell"].startswith("& ") - assert handler["powershell"].endswith("; exit $LASTEXITCODE") - - -def legacy_managed_config( - fixture: CopilotFixture, - extra_hooks: dict[str, Any] | None = None, -) -> dict[str, Any]: - return { - "version": 1, - "hooks": { - "preToolUse": [{ - "type": "command", - "bash": f'"{sys.executable}" {fixture.guard_path}', - "powershell": f'"{sys.executable}" {fixture.guard_path}', - "timeoutSec": 5, - }], - "postToolUse": [{ - "type": "command", - "bash": str(fixture.hook_path), - "powershell": str(fixture.hook_path), - "timeoutSec": 5, - }], - **(extra_hooks or {}), - }, - } - - -def run_hook( - handler: dict[str, Any], - home_dir: Path, - payload: str, -) -> subprocess.CompletedProcess[str]: - if os.name == "nt": - command = [ - "powershell.exe", - "-NoProfile", - "-NonInteractive", - "-Command", - handler["powershell"], - ] - else: - command = ["/bin/sh", "-c", handler["bash"]] - return subprocess.run( - command, - capture_output=True, - check=False, - env={**os.environ, "HOME": str(home_dir), "USERPROFILE": str(home_dir)}, - input=payload, - text=True, - ) - - -def test_copilot_is_registered_with_native_user_hooks() -> None: +def test_copilot_is_registered_with_complete_native_contract() -> None: assert SUPPORTED_AGENTS["copilot"] == { "name": "GitHub Copilot CLI", - "hook_event": "preToolUse/postToolUse", + "hook_event": "preToolUse/postToolUse/postToolUseFailure", "config_path": "~/.copilot/hooks/elydora-audit.json", } assert cli.PLUGIN_MAP["copilot"] is copilot.CopilotPlugin + assert copilot.CopilotPlugin.manages_guard_runtime is True -def test_install_preserves_user_hooks_migrates_legacy_and_is_idempotent( +def test_install_preserves_hooks_migrates_legacy_and_is_idempotent( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: @@ -188,15 +44,21 @@ def test_install_preserves_user_hooks_migrates_legacy_and_is_idempotent( "version": 1, "disableAllHooks": False, "hooks": { - "sessionStart": [{"type": "command", "command": "user-session"}], - "preToolUse": [{"type": "command", "command": "user-pre"}], + "sessionStart": [ + {"type": "command", "command": "user-session"} + ], + "preToolUse": [ + {"type": "command", "command": "user-pre"} + ], }, }, ) write_json_or_text( fixture.legacy_path, legacy_managed_config(fixture, { - "notification": [{"type": "command", "command": "user-notification"}], + "notification": [ + {"type": "command", "command": "user-notification"} + ], }), ) @@ -214,18 +76,32 @@ def test_install_preserves_user_hooks_migrates_legacy_and_is_idempotent( } assert len(settings["hooks"]["preToolUse"]) == 2 assert len(settings["hooks"]["postToolUse"]) == 1 + assert len(settings["hooks"]["postToolUseFailure"]) == 1 assert_native_handler(managed_handler(settings, "preToolUse", "guard.py")) assert_native_handler(managed_handler(settings, "postToolUse", "hook.py")) + assert_native_handler( + managed_handler(settings, "postToolUseFailure", "hook.py") + ) legacy = json.loads(fixture.legacy_path.read_text(encoding="utf-8")) assert legacy == { "version": 1, "hooks": { - "notification": [{"type": "command", "command": "user-notification"}], + "notification": [ + {"type": "command", "command": "user-notification"} + ], }, } - assert fixture.runtime_config_path.is_file() - assert fixture.private_key_path.read_text(encoding="utf-8") == "test-key" + runtime = json.loads( + fixture.runtime_config_path.read_text(encoding="utf-8") + ) + assert runtime["agent_id"] == AGENT_ID + assert runtime["agent_name"] == "copilot" + assert fixture.private_key_path.read_text(encoding="utf-8") == ( + VALID_PRIVATE_KEY + ) + assert fixture.guard_path.is_file() assert fixture.hook_path.is_file() + assert fixture.plugin.status()["installed"] is True def test_migration_removes_legacy_file_owned_entirely_by_elydora( @@ -242,43 +118,55 @@ def test_migration_removes_legacy_file_owned_entirely_by_elydora( assert_native_handler(managed_handler(settings, "preToolUse", "guard.py")) -def test_commands_block_and_forward_native_payload_byte_for_byte( +def test_commands_block_and_forward_success_and_failure_payloads( 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.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", + ) 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({ + base = { "sessionId": "session-1", "timestamp": 1, "cwd": str(fixture.project_dir), "toolName": "powershell", "toolArgs": {"command": "Get-ChildItem"}, - }, separators=(",", ":")) + } + pre_payload = json.dumps(base, separators=(",", ":")) guard = managed_handler(settings, "preToolUse", "guard.py") - guard_result = run_hook(guard, fixture.home_dir, pre_payload) + guard_result = run_hook(guard, fixture, pre_payload) assert guard_result.returncode == 2, guard_result.stderr assert "Agent is frozen by Elydora" in guard_result.stderr - post_payload = json.dumps({ - "sessionId": "session-1", - "timestamp": 2, - "cwd": str(fixture.project_dir), - "toolName": "powershell", - "toolArgs": {"command": "Get-ChildItem"}, - "toolResult": {"output": "ok"}, - }, separators=(",", ":")) - audit = managed_handler(settings, "postToolUse", "hook.py") - audit_result = run_hook(audit, fixture.home_dir, post_payload) - assert audit_result.returncode == 0, audit_result.stderr - assert capture_path.read_text(encoding="utf-8") == post_payload + success_payload = json.dumps( + {**base, "toolResult": {"resultType": "success", "textResultForLlm": "ok"}}, + separators=(",", ":"), + ) + success = managed_handler(settings, "postToolUse", "hook.py") + success_result = run_hook(success, fixture, success_payload) + assert success_result.returncode == 0, success_result.stderr + assert capture_path.read_text(encoding="utf-8") == success_payload + + failure_payload = json.dumps( + {**base, "error": "command failed"}, + separators=(",", ":"), + ) + failure = managed_handler(settings, "postToolUseFailure", "hook.py") + failure_result = run_hook(failure, fixture, failure_payload) + assert failure_result.returncode == 0, failure_result.stderr + assert capture_path.read_text(encoding="utf-8") == failure_payload def test_empty_home_override_uses_official_default( @@ -287,32 +175,192 @@ def test_empty_home_override_uses_official_default( ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) monkeypatch.setenv("COPILOT_HOME", "") + fixture.plugin.install(fixture.config) + default_path = fixture.home_dir / ".copilot" / "hooks" / "elydora-audit.json" settings = json.loads(default_path.read_text(encoding="utf-8")) assert_native_handler(managed_handler(settings, "preToolUse", "guard.py")) assert str(default_path) in fixture.plugin.status()["details"] -def test_status_requires_complete_pair_and_valid_runtime_identity( +def test_status_requires_complete_contract_and_exact_runtime_sources( 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"] + del settings["hooks"]["postToolUseFailure"] write_json_or_text(fixture.config_path, settings) assert fixture.plugin.status()["installed"] is False fixture.plugin.install(fixture.config) + fixture.guard_path.write_text("# tampered\n", encoding="utf-8") + assert fixture.plugin.status()["installed"] is False + + fixture.plugin.install(fixture.config) + valid_runtime = fixture.runtime_config_path.read_text(encoding="utf-8") fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") - with pytest.raises(ValueError, match="parse Elydora runtime config"): + with pytest.raises(ValueError, match="runtime config"): + fixture.plugin.status() + + fixture.runtime_config_path.write_text(valid_runtime, encoding="utf-8") + fixture.private_key_path.write_text("invalid", encoding="utf-8") + with pytest.raises(ValueError, match="private key"): fixture.plugin.status() -def test_uninstall_removes_exact_ownership_and_preserves_user_entries( +def test_settings_precedence_allows_later_false_and_preserves_jsonc( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + user_source = "{\n // user policy\n \"disableAllHooks\": true,\n}\n" + claude_source = "{\n \"disableAllHooks\": true,\n}\n" + repository_source = "{\n \"disableAllHooks\": false,\n}\n" + fixture = prepare_fixture( + monkeypatch, + tmp_path, + user_settings=user_source, + claude_local_settings=claude_source, + repository_settings=repository_source, + ) + + fixture.plugin.install(fixture.config) + + assert fixture.user_settings_path.read_text(encoding="utf-8") == user_source + assert fixture.claude_local_settings_path.read_text( + encoding="utf-8" + ) == claude_source + assert fixture.repository_settings_path.read_text( + encoding="utf-8" + ) == repository_source + assert fixture.plugin.status()["installed"] is True + + +@pytest.mark.parametrize( + ("kwargs", "source_name"), + [ + ({"repository_settings": {"disableAllHooks": True}}, "repository"), + ( + { + "claude_local_settings": {"disableAllHooks": True}, + "repository_settings": {"disableAllHooks": True}, + }, + "repository", + ), + ], +) +def test_effective_disabled_settings_block_before_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kwargs: dict[str, object], + source_name: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, **kwargs) + with pytest.raises(ValueError, match=source_name): + fixture.plugin.install(fixture.config) + assert fixture.config_path.exists() is False + assert_runtime_absent(fixture) + + +def test_managed_file_disable_flag_remains_authoritative( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + user_config={"version": 1, "disableAllHooks": True, "hooks": {}}, + local_settings={"disableAllHooks": False}, + ) + with pytest.raises(ValueError, match="user hooks"): + fixture.plugin.install(fixture.config) + assert_runtime_absent(fixture) + + +def test_matchers_use_official_javascript_regular_expression_syntax( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + user_config={ + "version": 1, + "hooks": { + "preToolUse": [{ + "type": "command", + "command": "user-pre", + "matcher": "(?shell)", + }], + }, + }, + ) + + fixture.plugin.install(fixture.config) + + settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) + assert settings["hooks"]["preToolUse"][0]["matcher"] == ( + "(?shell)" + ) + + +@pytest.mark.parametrize( + "existing", + [ + "{ malformed", + '{"version":1,"version":1,"hooks":{}}', + {"hooks": {}}, + {"version": True, "hooks": {}}, + {"version": 1, "hooks": None}, + {"version": 1, "hooks": {"unknownEvent": []}}, + {"version": 1, "hooks": {"preToolUse": None}}, + {"version": 1, "hooks": {"preToolUse": [None]}}, + {"version": 1, "hooks": {"preToolUse": [{}]}}, + {"version": 1, "hooks": {"preToolUse": [{"command": 1}]}}, + {"version": 1, "hooks": {"preToolUse": [{"command": "x", "timeoutSec": 0}]}}, + {"version": 1, "hooks": {"preToolUse": [{"command": "x", "matcher": "["}]}}, + {"version": 1, "hooks": {"postToolUse": [{"type": "prompt", "prompt": "x"}]}}, + {"version": 1, "hooks": {"postToolUse": [{"type": "http", "url": "http://example.com"}]}}, + ], +) +def test_invalid_hook_documents_are_preserved_before_runtime_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + existing: object, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, user_config=existing) + original = fixture.config_path.read_text(encoding="utf-8") + with pytest.raises(ValueError): + fixture.plugin.install(fixture.config) + assert fixture.config_path.read_text(encoding="utf-8") == original + assert_runtime_absent(fixture) + + +@pytest.mark.parametrize( + "settings", + [ + "{ malformed", + '{"disableAllHooks":true,"disableAllHooks":false}', + '{"disableAllHooks":"yes"}', + ], +) +def test_invalid_settings_are_preserved_before_runtime_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + settings: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, user_settings=settings) + with pytest.raises(ValueError): + fixture.plugin.install(fixture.config) + assert fixture.user_settings_path.read_text(encoding="utf-8") == settings + assert_runtime_absent(fixture) + + +def test_uninstall_removes_exact_ownership_and_preserves_adjacent_handlers( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: @@ -328,8 +376,8 @@ def test_uninstall_removes_exact_ownership_and_preserves_user_entries( settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) settings["hooks"]["preToolUse"].append({ "type": "command", - "bash": f"{sys.executable} {fixture.agent_dir.parent / 'agent-10' / 'guard.py'}", - "powershell": "user-decoy", + "bash": "user-pre", + "powershell": "user-pre", "timeoutSec": 10, }) write_json_or_text(fixture.config_path, settings) @@ -340,9 +388,14 @@ def test_uninstall_removes_exact_ownership_and_preserves_user_entries( assert remaining["hooks"]["notification"] == [ {"type": "command", "command": "keep"} ] + assert remaining["hooks"]["preToolUse"] == [{ + "type": "command", + "bash": "user-pre", + "powershell": "user-pre", + "timeoutSec": 10, + }] assert "postToolUse" not in remaining["hooks"] - assert len(remaining["hooks"]["preToolUse"]) == 1 - assert "agent-10" in remaining["hooks"]["preToolUse"][0]["bash"] + assert "postToolUseFailure" not in remaining["hooks"] def test_uninstall_leaves_absent_sources_absent( @@ -355,85 +408,14 @@ def test_uninstall_leaves_absent_sources_absent( assert fixture.legacy_path.exists() is False -@pytest.mark.parametrize( - "existing", - [ - "{ malformed", - {"hooks": {}}, - {"version": 2, "hooks": {}}, - {"version": 1, "hooks": None}, - {"version": 1, "hooks": {"preToolUse": None}}, - {"version": 1, "hooks": {"preToolUse": [None]}}, - ], -) -def test_invalid_config_is_preserved_before_runtime_writes( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - existing: object, -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path, user_config=existing) - original = fixture.config_path.read_text(encoding="utf-8") - with pytest.raises((ValueError, json.JSONDecodeError)): - fixture.plugin.install(fixture.config) - assert fixture.config_path.read_text(encoding="utf-8") == original - assert fixture.runtime_config_path.exists() is False - assert fixture.private_key_path.exists() is False - assert fixture.hook_path.exists() is False - - -def test_missing_guard_is_rejected_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.runtime_config_path.exists() is False - assert fixture.private_key_path.exists() is False - assert fixture.hook_path.exists() is False - - -def test_transaction_rolls_back_every_file_after_commit_failure( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = prepare_fixture( - monkeypatch, - tmp_path, - user_config={"version": 1, "hooks": {"notification": []}}, - legacy_config={"version": 1, "hooks": {"sessionStart": []}}, - ) - user_before = fixture.config_path.read_text(encoding="utf-8") - legacy_before = fixture.legacy_path.read_text(encoding="utf-8") - original_commit = _transaction._commit - commits = 0 - - def fail_after_first_commit(staged: Any) -> None: - nonlocal commits - original_commit(staged) - commits += 1 - if commits == 1: - raise OSError("injected commit failure") - - monkeypatch.setattr(_transaction, "_commit", fail_after_first_commit) - with pytest.raises(OSError, match="injected commit failure"): - fixture.plugin.install(fixture.config) - assert fixture.config_path.read_text(encoding="utf-8") == user_before - assert fixture.legacy_path.read_text(encoding="utf-8") == legacy_before - assert fixture.runtime_config_path.exists() is False - assert fixture.private_key_path.exists() is False - assert fixture.hook_path.exists() is False - - -def test_atomic_writes_leave_no_transaction_files( +def test_preflight_validates_credentials_and_runtime_path_without_writes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - for directory in (fixture.agent_dir, fixture.config_path.parent): - assert all( - path.suffix not in {".tmp", ".rollback"} - for path in directory.iterdir() - ) + invalid = dict(fixture.config) + invalid["private_key"] = "invalid" + with pytest.raises(ValueError, match="private_key"): + fixture.plugin.preflight_install(invalid) # type: ignore[arg-type] + assert fixture.config_path.exists() is False + assert_runtime_absent(fixture) diff --git a/sdks/python/tests/test_copilot_transaction.py b/sdks/python/tests/test_copilot_transaction.py new file mode 100644 index 0000000..b4d6043 --- /dev/null +++ b/sdks/python/tests/test_copilot_transaction.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +from typing import Any + +import pytest + +from elydora.plugins import _transaction, copilot +from elydora.plugins.copilot_installation import prepare_copilot_installation +from elydora.plugins.copilot_io import read_sources + +from copilot_support import ( + AGENT_ID, + assert_runtime_absent, + legacy_managed_config, + prepare_fixture, + write_json_or_text, +) + + +def test_install_rolls_back_runtime_hooks_and_legacy_after_late_failure( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + user_config={"version": 1, "hooks": {"notification": []}}, + ) + write_json_or_text(fixture.legacy_path, legacy_managed_config(fixture)) + user_before = fixture.config_path.read_text(encoding="utf-8") + legacy_before = fixture.legacy_path.read_text(encoding="utf-8") + original_commit = _transaction._commit + commits = 0 + + def fail_after_fifth_commit(staged: Any) -> None: + nonlocal commits + original_commit(staged) + commits += 1 + if commits == 5: + raise OSError("injected late commit failure") + + monkeypatch.setattr(_transaction, "_commit", fail_after_fifth_commit) + with pytest.raises(OSError, match="injected late commit failure"): + fixture.plugin.install(fixture.config) + + assert fixture.config_path.read_text(encoding="utf-8") == user_before + assert fixture.legacy_path.read_text(encoding="utf-8") == legacy_before + assert_runtime_absent(fixture) + + +def test_install_detects_concurrent_hook_replacement_before_its_commit( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + user_config={"version": 1, "hooks": {"notification": []}}, + ) + concurrent = '{"version":1,"hooks":{"notification":[]},"external":true}\n' + original_commit = _transaction._commit + commits = 0 + + def replace_after_first_commit(staged: Any) -> None: + nonlocal commits + original_commit(staged) + commits += 1 + if commits == 1: + fixture.config_path.write_text(concurrent, encoding="utf-8") + + monkeypatch.setattr(_transaction, "_commit", replace_after_first_commit) + with pytest.raises(OSError, match="changed during installation"): + fixture.plugin.install(fixture.config) + + assert fixture.config_path.read_text(encoding="utf-8") == concurrent + assert_runtime_absent(fixture) + + +def test_install_rolls_back_when_effective_settings_change_mid_transaction( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + user_settings={"disableAllHooks": False}, + ) + original_commit = _transaction._commit + commits = 0 + + def disable_after_third_commit(staged: Any) -> None: + nonlocal commits + original_commit(staged) + commits += 1 + if commits == 3: + write_json_or_text( + fixture.user_settings_path, + {"disableAllHooks": True}, + ) + + monkeypatch.setattr(_transaction, "_commit", disable_after_third_commit) + with pytest.raises(OSError, match="user settings changed"): + fixture.plugin.install(fixture.config) + + assert json.loads( + fixture.user_settings_path.read_text(encoding="utf-8") + )["disableAllHooks"] is True + assert fixture.config_path.exists() is False + assert_runtime_absent(fixture) + + +def test_prepare_rejects_same_content_stale_hook_snapshot( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture( + monkeypatch, + tmp_path, + user_config={"version": 1, "hooks": {"notification": []}}, + ) + sources = read_sources() + paths = copilot.preflight_copilot_installation(fixture.config, sources) + rendered = copilot._render_installation( + sources, + paths.guard_path, + paths.audit_path, + ) + original = fixture.config_path.read_text(encoding="utf-8") + replacement = fixture.config_path.with_name("replacement.json") + replacement.write_text(original, encoding="utf-8") + os.replace(replacement, fixture.config_path) + + with pytest.raises(OSError, match="changed before staging"): + prepare_copilot_installation(fixture.config, sources, rendered) + assert_runtime_absent(fixture) + + +def test_install_preserves_orphaned_runtime_without_verifiable_identity( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.agent_dir.mkdir(parents=True) + fixture.hook_path.write_text("# orphaned\n", encoding="utf-8") + + with pytest.raises(ValueError, match="cannot be verified"): + fixture.plugin.install(fixture.config) + + assert fixture.hook_path.read_text(encoding="utf-8") == "# orphaned\n" + assert fixture.config_path.exists() is False + assert fixture.guard_path.exists() is False + assert fixture.runtime_config_path.exists() is False + assert fixture.private_key_path.exists() is False + + +def _symlink_or_skip( + target: Path, + link: Path, + *, + directory: bool, +) -> None: + try: + link.parent.mkdir(parents=True, exist_ok=True) + os.symlink(target, link, target_is_directory=directory) + except (NotImplementedError, OSError) as error: + pytest.skip(f"symbolic links unavailable: {error}") + + +@pytest.mark.parametrize("location", ["home", "hook", "runtime"]) +def test_install_rejects_linked_hook_and_runtime_paths( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + location: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + if location == "home": + target = tmp_path / "copilot-target" + target.mkdir() + linked = tmp_path / "linked-copilot-home" + _symlink_or_skip(target, linked, directory=True) + monkeypatch.setenv("COPILOT_HOME", str(linked)) + elif location == "hook": + target = tmp_path / "hook-target.json" + write_json_or_text(target, {"version": 1, "hooks": {}}) + _symlink_or_skip(target, fixture.config_path, directory=False) + else: + target = tmp_path / "runtime-target" + target.mkdir() + fixture.agent_dir.parent.mkdir(parents=True) + _symlink_or_skip(target, fixture.agent_dir, directory=True) + + with pytest.raises(OSError, match="physical"): + fixture.plugin.install(fixture.config) + + +def test_uninstall_restores_both_hook_documents_when_second_commit_fails( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + managed = fixture.config_path.read_text(encoding="utf-8") + fixture.legacy_path.parent.mkdir(parents=True, exist_ok=True) + fixture.legacy_path.write_text(managed, encoding="utf-8") + user_before = fixture.config_path.read_text(encoding="utf-8") + legacy_before = fixture.legacy_path.read_text(encoding="utf-8") + original_commit = _transaction._commit + commits = 0 + + def fail_after_second_commit(staged: Any) -> None: + nonlocal commits + original_commit(staged) + commits += 1 + if commits == 2: + raise OSError("injected uninstall failure") + + monkeypatch.setattr(_transaction, "_commit", fail_after_second_commit) + with pytest.raises(OSError, match="injected uninstall failure"): + fixture.plugin.uninstall(AGENT_ID) + + assert fixture.config_path.read_text(encoding="utf-8") == user_before + assert fixture.legacy_path.read_text(encoding="utf-8") == legacy_before + + +def test_installation_leaves_no_transaction_artifacts( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.plugin.install(fixture.config) + for directory in (fixture.agent_dir, fixture.config_path.parent): + assert all( + path.suffix not in {".tmp", ".rollback"} + for path in directory.iterdir() + )