fix(copilot): make Python hook installation transactional

This commit is contained in:
ZacharyZhang-NY committed 2026-07-20 12:29:26 -04:00
1 parent 97d1525227
commit 16ceac81c4
14 files changed
+2121 -583

No files matched your search

+1
View File
@@ -32,6 +32,7 @@ This repository owns the published `elydora` package, its CLI, synchronous and a
- Validate Auggie matcher syntax during installation with Node.js `new RegExp`; keep status and uninstall independent from the JavaScript validator so recovery remains available offline. Commit Auggie settings, both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Require exact runtime identity, canonical private keys, physical files, and exact wrapper sources during status checks, then verify the effective configuration with `auggie tools list`. - Validate Auggie matcher syntax during installation with Node.js `new RegExp`; keep status and uninstall independent from the JavaScript validator so recovery remains available offline. Commit Auggie settings, both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Require exact runtime identity, canonical private keys, physical files, and exact wrapper sources during status checks, then verify the effective configuration with `auggie tools list`.
- Treat Cline 3 CLI file hooks as the adapter contract. Resolve hooks additively from `~/Documents/Cline/Hooks`, `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default, `.clinerules/hooks`, and `.cline/hooks`; write only the `$CLINE_DIR` source. Preserve the complete hybrid CLI payload, resolve generated Python runtimes from absolute shebangs, and emit one-line JSON stdout with `cancel: true` for guard exit code `2`. Commit both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Status requires physical files, canonical private keys, exact runtime identity, exact generated sources, and exact wrapper sources. Hook errors, timeouts, and invalid controls fail open in Cline while remaining observable. - Treat Cline 3 CLI file hooks as the adapter contract. Resolve hooks additively from `~/Documents/Cline/Hooks`, `$CLINE_DIR/hooks` with `~/.cline/hooks` as the default, `.clinerules/hooks`, and `.cline/hooks`; write only the `$CLINE_DIR` source. Preserve the complete hybrid CLI payload, resolve generated Python runtimes from absolute shebangs, and emit one-line JSON stdout with `cancel: true` for guard exit code `2`. Commit both wrappers, generated runtimes, runtime config, and private key through one rollback-capable transaction. Status requires physical files, canonical private keys, exact runtime identity, exact generated sources, and exact wrapper sources. Hook errors, timeouts, and invalid controls fail open in Cline while remaining observable.
- Resolve Codex user hooks through `$CODEX_HOME/hooks.json` with `~/.codex/hooks.json` as the default, matching Codex's existing-directory canonicalization rule. Preserve additive user TOML, project, plugin, and managed sources. Register exact `PreToolUse` and `PostToolUse` match-all command groups with ten-second handlers, preserve the complete native payload, propagate freeze and revocation through exit code `2`, keep guard lookup and audit delivery fail-open with observable errors, and commit user hooks plus all four runtime artifacts through one rollback-capable transaction. Require `/hooks` approval for both definition hashes. - Resolve Codex user hooks through `$CODEX_HOME/hooks.json` with `~/.codex/hooks.json` as the default, matching Codex's existing-directory canonicalization rule. Preserve additive user TOML, project, plugin, and managed sources. Register exact `PreToolUse` and `PostToolUse` match-all command groups with ten-second handlers, preserve the complete native payload, propagate freeze and revocation through exit code `2`, keep guard lookup and audit delivery fail-open with observable errors, and commit user hooks plus all four runtime artifacts through one rollback-capable transaction. Require `/hooks` approval for both definition hashes.
- Resolve GitHub Copilot CLI user hooks through `$COPILOT_HOME/hooks/elydora-audit.json` with `~/.copilot/hooks/elydora-audit.json` as the default, and treat an empty override as the default. Keep policy, repository, inline settings, cross-tool Claude, and plugin hook sources read-only. Evaluate `disableAllHooks` through legacy user config, user settings, Claude repository and local settings, then GitHub repository and local settings; the managed hook file flag remains independently authoritative. Register exact `preToolUse`, `postToolUse`, and `postToolUseFailure` command handlers with ten-second timeouts, preserve the complete native camelCase payload, and propagate frozen or revoked state through exit code `2`. Commit the user hook file, exact legacy migration, generated runtimes, runtime config, and private key through one rollback-capable transaction. Require physical source paths, strict hook schemas, exact runtime sources, canonical private keys, and matching runtime identity during status checks, then verify discovery with the current official Copilot native hook loader and restart active CLI sessions.
- Write Factory Droid hooks only to user-level `~/.factory/hooks.json`, its active legacy fallback, or the `hooks` object in `~/.factory/settings.json`; keep project and organization sources read-only. Preserve JSONC syntax, select sources per event, store direct event maps in hook files, keep settings hooks nested, validate JavaScript matchers before installation, use exact command ownership, and instruct users to review changes through `/hooks`. - Write Factory Droid hooks only to user-level `~/.factory/hooks.json`, its active legacy fallback, or the `hooks` object in `~/.factory/settings.json`; keep project and organization sources read-only. Preserve JSONC syntax, select sources per event, store direct event maps in hook files, keep settings hooks nested, validate JavaScript matchers before installation, use exact command ownership, and instruct users to review changes through `/hooks`.
- Resolve Qwen Code user settings through explicit `QWEN_HOME`, `~/.qwen/.env`, then `~/.env`; keep workspace settings read-only. Preserve comments while rejecting trailing commas and duplicate keys, store hooks under the nested `hooks` object, treat `disableAllHooks` as unhealthy, validate JavaScript matchers before installation, use millisecond timeouts and explicit host shells, preserve PowerShell native exit codes, and instruct users to review changes through `/hooks`. - Resolve Qwen Code user settings through explicit `QWEN_HOME`, `~/.qwen/.env`, then `~/.env`; keep workspace settings read-only. Preserve comments while rejecting trailing commas and duplicate keys, store hooks under the nested `hooks` object, treat `disableAllHooks` as unhealthy, validate JavaScript matchers before installation, use millisecond timeouts and explicit host shells, preserve PowerShell native exit codes, and instruct users to review changes through `/hooks`.
+2
View File
@@ -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. 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. 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. 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.
+56 -2
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
import os import os
import tempfile import tempfile
from typing import List, Optional, Sequence from typing import List, Optional, Sequence, cast
from uuid import uuid4 from uuid import uuid4
from ._managed_files import ( from ._managed_files import (
@@ -30,6 +30,14 @@ class FileChange:
maximum_bytes: int maximum_bytes: int
@dataclass(frozen=True)
class FilePrecondition:
file_path: str
label: str
original: Optional[FileSnapshot]
maximum_bytes: int = MAX_SOURCE_BYTES
@dataclass @dataclass
class _StagedChange: class _StagedChange:
change: FileChange change: FileChange
@@ -39,6 +47,22 @@ class _StagedChange:
committed: bool = False 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( def read_optional(
file_path: str, file_path: str,
label: str, label: str,
@@ -55,6 +79,7 @@ def source_change(
next_source: Optional[str], next_source: Optional[str],
mode: int, mode: int,
maximum_bytes: int = MAX_SOURCE_BYTES, maximum_bytes: int = MAX_SOURCE_BYTES,
expected_snapshot: object = _EXPECTED_SNAPSHOT_UNSET,
) -> Optional[FileChange]: ) -> Optional[FileChange]:
if original == next_source: if original == next_source:
return None return None
@@ -69,6 +94,10 @@ def source_change(
current = None if snapshot is None else snapshot.contents current = None if snapshot is None else snapshot.contents
if current != original: if current != original:
raise OSError(f"{label} changed before staging: {file_path}") 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( return FileChange(
file_path, file_path,
label, label,
@@ -170,6 +199,23 @@ def _assert_unchanged(change: FileChange) -> None:
raise OSError(f"{change.label} changed during installation: {change.file_path}") 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: def _stage(change: FileChange) -> _StagedChange:
_assert_unchanged(change) _assert_unchanged(change)
directory = os.path.dirname(change.file_path) directory = os.path.dirname(change.file_path)
@@ -302,7 +348,11 @@ def _cleanup(staged: _StagedChange) -> None:
_remove_optional(staged.rollback_path) _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] filtered = [change for change in changes if change.original != change.next_source]
if not filtered: if not filtered:
return return
@@ -311,10 +361,14 @@ def write_changes(changes: Sequence[FileChange], label: str) -> None:
raise ValueError(f"{label} contains duplicate file targets") raise ValueError(f"{label} contains duplicate file targets")
staged: List[_StagedChange] = [] staged: List[_StagedChange] = []
try: try:
_assert_preconditions(preconditions, label)
for change in filtered: for change in filtered:
staged.append(_stage(change)) staged.append(_stage(change))
_assert_preconditions(preconditions, label)
for item in staged: for item in staged:
_assert_preconditions(preconditions, label)
_commit(item) _commit(item)
_assert_preconditions(preconditions, label)
except Exception as error: except Exception as error:
recovery_errors = [] recovery_errors = []
for item in reversed(staged): for item in reversed(staged):
+90 -149
View File
@@ -1,120 +1,37 @@
"""GitHub Copilot CLI user-hook integration.""" """GitHub Copilot CLI native user-hook integration."""
from __future__ import annotations from __future__ import annotations
import json
import os 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 .base import AgentPlugin, InstallConfig, PluginStatus
from .copilot_contract import ( from .copilot_contract import (
AGENT_KEY, AGENT_KEY,
AUDIT_SCRIPT,
CopilotHooks,
CopilotSources, CopilotSources,
RenderedDocument,
RuntimeContract, RuntimeContract,
build_handler, build_handler,
remove_managed_hooks, remove_managed_hooks,
render_document, render_document,
runtime_contracts, runtime_contracts,
) )
from .copilot_io import read_sources, rendered_change, runtime_files_exist from .copilot_installation import (
from .hook_template import generate_hook_script 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 _render_installation(
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(
sources: CopilotSources, sources: CopilotSources,
user_hooks: CopilotHooks, guard_path: str,
agent_id: str = "", audit_path: str,
) -> List[FileChange]: ) -> List[RenderedDocument]:
rendered = [render_document(sources.user, user_hooks)] user_hooks = remove_managed_hooks(sources.user.hooks)
if sources.legacy is not None:
rendered.append(render_document(
sources.legacy,
remove_managed_hooks(
sources.legacy.hooks,
ELYDORA_DIR,
agent_id,
),
))
return [
change
for item in rendered
for change in [rendered_change(item)]
if change is not None
]
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(
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
if sources.legacy is not None:
return sources.legacy.file_path
return sources.user.file_path
class CopilotPlugin(AgentPlugin):
"""Install Elydora into GitHub Copilot CLI's global user hooks."""
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["preToolUse"] = [
*user_hooks.get("preToolUse", []), *user_hooks.get("preToolUse", []),
build_handler(guard_path), build_handler(guard_path),
@@ -123,66 +40,90 @@ class CopilotPlugin(AgentPlugin):
*user_hooks.get("postToolUse", []), *user_hooks.get("postToolUse", []),
build_handler(audit_path), build_handler(audit_path),
] ]
user_hooks["postToolUseFailure"] = [
audit_script = generate_hook_script( *user_hooks.get("postToolUseFailure", []),
org_id=config.get("org_id", ""), build_handler(audit_path),
agent_id=agent_id,
kid=config.get("kid", ""),
base_url=config.get("base_url", "https://api.elydora.com"),
)
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] rendered = [render_document(sources.user, user_hooks)]
changes.extend(_rendered_changes(sources, user_hooks)) if sources.legacy is not None:
write_changes(changes, "Install GitHub Copilot hooks") rendered.append(render_document(
sources.legacy,
remove_managed_hooks(sources.legacy.hooks),
))
return rendered
def _render_uninstall(
sources: CopilotSources,
agent_id: str,
) -> List[RenderedDocument]:
rendered = [render_document(
sources.user,
remove_managed_hooks(sources.user.hooks, agent_id),
)]
if sources.legacy is not None:
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:
sources = read_sources()
paths = preflight_copilot_installation(config, sources)
rendered = _render_installation(
sources,
paths.guard_path,
paths.audit_path,
)
prepared = prepare_copilot_installation(config, sources, rendered)
commit_copilot_installation(prepared)
print(f"GitHub Copilot CLI hooks: {sources.user.file_path}") 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: def uninstall(self, agent_id: str = "") -> None:
sources = read_sources(_home_dir()) sources = read_sources()
user_hooks = remove_managed_hooks( rendered = _render_uninstall(sources, agent_id)
sources.user.hooks, commit_copilot_uninstall(prepare_copilot_uninstall(rendered))
ELYDORA_DIR,
agent_id,
)
changes = _rendered_changes(sources, user_hooks, agent_id)
write_changes(changes, "Uninstall GitHub Copilot hooks")
def status(self) -> PluginStatus: def status(self) -> PluginStatus:
sources = read_sources(_home_dir()) sources = read_sources()
contracts = _contracts(sources) contracts = _merged_contracts(sources)
if not contracts: configured = sources.disabled_by is None and bool(contracts)
return PluginStatus( installed = configured and runtime_files_exist(contracts)
installed=False,
agent=AGENT_KEY,
details="Not installed",
)
installed = runtime_files_exist(contracts)
config_path = _configured_path(sources, contracts)
details = ( details = (
f"Config: {config_path}" f"Config: {sources.user.file_path}"
if installed if installed
else f"Configured at {config_path}; runtime scripts missing" else (
f"Configured at {sources.user.file_path}; "
"managed contract incomplete"
)
) )
return PluginStatus( return PluginStatus(
installed=installed, installed=installed,
agent=AGENT_KEY, agent=AGENT_KEY,
details=details, details=details if contracts else "Not installed",
) )
+248 -103
View File
@@ -1,4 +1,4 @@
"""GitHub Copilot hook contract, ownership, and rendering.""" """GitHub Copilot CLI hook ownership and rendering."""
from __future__ import annotations from __future__ import annotations
@@ -8,7 +8,11 @@ import os
import re import re
import shlex import shlex
import sys 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" AGENT_KEY = "copilot"
@@ -18,7 +22,11 @@ CONFIG_FILE = "elydora-audit.json"
HOOK_TIMEOUT_SECONDS = 10 HOOK_TIMEOUT_SECONDS = 10
JsonObject = Dict[str, Any] 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) @dataclass(frozen=True)
@@ -27,13 +35,24 @@ class CopilotDocument:
file_path: str file_path: str
root: JsonObject root: JsonObject
hooks: CopilotHooks hooks: CopilotHooks
hooks_disabled: bool
raw: Optional[str] = None raw: Optional[str] = None
snapshot: Optional[FileSnapshot] = None
@dataclass(frozen=True)
class SourcePrecondition:
file_path: str
label: str
snapshot: Optional[FileSnapshot]
@dataclass(frozen=True) @dataclass(frozen=True)
class CopilotSources: class CopilotSources:
user: CopilotDocument user: CopilotDocument
legacy: Optional[CopilotDocument] legacy: Optional[CopilotDocument]
disabled_by: Optional[str]
settings_preconditions: Tuple[SourcePrecondition, ...]
@dataclass(frozen=True) @dataclass(frozen=True)
@@ -50,16 +69,36 @@ class RuntimeContract:
audit_path: str 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( return os.path.normcase(os.path.abspath(left)) == os.path.normcase(
os.path.abspath(right) 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) return os.path.normcase(left) == os.path.normcase(right)
def _quote_posix(value: str) -> str:
return "'" + value.replace("'", "'\"'\"'") + "'"
def _quote_powershell(value: str) -> str: def _quote_powershell(value: str) -> str:
return "'" + value.replace("'", "''") + "'" return "'" + value.replace("'", "''") + "'"
@@ -67,7 +106,7 @@ def _quote_powershell(value: str) -> str:
def build_handler(script_path: str) -> JsonObject: def build_handler(script_path: str) -> JsonObject:
return { return {
"type": "command", "type": "command",
"bash": shlex.join([sys.executable, script_path]), "bash": f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}",
"powershell": ( "powershell": (
f"& {_quote_powershell(sys.executable)} " f"& {_quote_powershell(sys.executable)} "
f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" 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]: def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]:
if set(handler) != {"type", "bash", "powershell", "timeoutSec"}: if start >= len(command) or command[start] != "'":
return None 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 return None
if handler.get("timeoutSec") != HOOK_TIMEOUT_SECONDS:
def _parse_generated_bash(command: Any) -> Optional[Tuple[str, str]]:
if not isinstance(command, str):
return None return None
bash = handler.get("bash") executable = _read_posix_argument(command, 0)
if not isinstance(bash, str): if executable is None or command[executable.next_index:executable.next_index + 1] != " ":
return None return None
script = _read_posix_argument(command, executable.next_index + 1)
if script is None or script.next_index != len(command):
return None
return executable.value, script.value
def _read_powershell_argument(
command: str, start: int
) -> Optional[_ParsedArgument]:
if start >= len(command) or command[start] != "'":
return None
value = ""
index = start + 1
while index < len(command):
if command[index] != "'":
value += command[index]
index += 1
continue
if index + 1 < len(command) and command[index + 1] == "'":
value += "'"
index += 2
continue
return _ParsedArgument(value, index + 1)
return None
def _parse_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: try:
arguments = shlex.split(bash) arguments = shlex.split(bash_command)
except ValueError: except ValueError:
return None 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 return None
script_path = arguments[1] return 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
def _legacy_script_path(handler: JsonObject) -> Optional[str]: 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 return None
if handler.get("type") != "command" or handler.get("timeoutSec") != 5: command = str(handler["bash"])
return None match = re.fullmatch(r'"[^"\r\n]+"\s+(.+)', command)
bash = handler.get("bash") return match.group(1) if match else command
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
def _managed_agent_id( def _managed_entry(
handler: JsonObject, handler: JsonObject, script_name: str
script_name: str, ) -> Optional[_ManagedEntry]:
runtime_root: str, script_path = (
) -> Optional[str]: _current_script_path(handler)
script_path = _generated_script_path(handler) or _legacy_script_path(handler) or _prior_script_path(handler)
if not script_path or os.path.basename(script_path) != script_name: 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 return None
agent_directory = os.path.dirname(script_path) 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 return None
agent_id = os.path.basename(agent_directory) 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: def parse_document(
if value is None: file_path: str,
raise ValueError(f'{label} field "hooks" must be an object') snapshot: FileSnapshot,
if not isinstance(value, dict): label: str,
raise ValueError(f'{label} field "hooks" must be an object') ) -> CopilotDocument:
hooks: CopilotHooks = {} document_label = f"{label} at {file_path}"
for event, handlers in value.items(): root = parse_json_object(snapshot.contents, document_label)
if not isinstance(event, str): version = root.get("version")
raise ValueError(f"{label} hook event names must be strings") if isinstance(version, bool) or version != 1:
if not isinstance(handlers, list): raise ValueError(f"{document_label} must declare version 1")
raise ValueError(f'{label} field "hooks.{event}" must be an array') if "disableAllHooks" in root and not isinstance(
if not all(isinstance(handler, dict) for handler in handlers): root["disableAllHooks"], bool
):
raise ValueError( raise ValueError(
f'{label} field "hooks.{event}" must contain objects' 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,
) )
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 create_document(file_path: str) -> CopilotDocument: def create_document(file_path: str) -> CopilotDocument:
return CopilotDocument(False, file_path, {}, {}) return CopilotDocument(False, file_path, {}, {}, False)
def remove_managed_hooks( def remove_managed_hooks(
hooks: CopilotHooks, hooks: CopilotHooks, agent_id: str = ""
runtime_root: str,
agent_id: str = "",
) -> CopilotHooks: ) -> CopilotHooks:
result = dict(hooks) result = {event: list(handlers) for event, handlers in hooks.items()}
for event, script_name in ( for event, script_name in MANAGED_EVENTS:
("preToolUse", GUARD_SCRIPT),
("postToolUse", AUDIT_SCRIPT),
):
handlers = [] handlers = []
for handler in result.get(event, []): for handler in result.get(event, []):
managed_id = _managed_agent_id(handler, script_name, runtime_root) managed = _managed_entry(handler, script_name)
remove = managed_id is not None and ( remove = managed is not None and (
not agent_id or _same_agent_id(managed_id, agent_id) not agent_id or same_agent_id(managed.agent_id, agent_id)
) )
if not remove: if not remove:
handlers.append(handler) handlers.append(handler)
@@ -190,7 +325,7 @@ def remove_managed_hooks(
def _empty_owned_document(root: JsonObject, hooks: CopilotHooks) -> bool: 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( def render_document(
@@ -206,37 +341,47 @@ def render_document(
root["hooks"] = hooks root["hooks"] = hooks
else: else:
root.pop("hooks", None) root.pop("hooks", None)
next_source = json.dumps(root, indent=2) + "\n" next_source = json.dumps(root, indent=2, ensure_ascii=False) + "\n"
return RenderedDocument(document, next_source != document.raw, next_source) 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( def _managed_entries(
handlers: List[JsonObject], handlers: List[JsonObject], script_name: str
script_name: str, ) -> Dict[str, List[_ManagedEntry]]:
runtime_root: str, result: Dict[str, List[_ManagedEntry]] = {}
) -> Set[str]:
result: Set[str] = set()
for handler in handlers: for handler in handlers:
agent_id = _managed_agent_id(handler, script_name, runtime_root) entry = _managed_entry(handler, script_name)
if agent_id: if entry is not None:
result.add(agent_id) result.setdefault(os.path.normcase(entry.agent_id), []).append(entry)
return result return result
def runtime_contracts( def runtime_contracts(hooks: CopilotHooks) -> List[RuntimeContract]:
hooks: CopilotHooks, guards = _managed_entries(hooks.get("preToolUse", []), GUARD_SCRIPT)
runtime_root: str, successes = _managed_entries(hooks.get("postToolUse", []), AUDIT_SCRIPT)
) -> List[RuntimeContract]: failures = _managed_entries(
guards = _managed_ids(hooks.get("preToolUse", []), GUARD_SCRIPT, runtime_root) hooks.get("postToolUseFailure", []), AUDIT_SCRIPT
audits = _managed_ids(hooks.get("postToolUse", []), AUDIT_SCRIPT, runtime_root) )
contracts: List[RuntimeContract] = [] contracts = []
for agent_id in sorted(guards): for key, guard in guards.items():
if not any(_same_agent_id(agent_id, audit_id) for audit_id in audits): 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 continue
agent_directory = os.path.join(runtime_root, agent_id)
contracts.append(RuntimeContract( contracts.append(RuntimeContract(
agent_id, guard[0].agent_id,
os.path.join(agent_directory, GUARD_SCRIPT), guard[0].script_path,
os.path.join(agent_directory, AUDIT_SCRIPT), success[0].script_path,
)) ))
return contracts return contracts
+277
View File
@@ -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")
+363 -58
View File
@@ -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 from __future__ import annotations
import json import base64
from dataclasses import dataclass
import os import os
from typing import Any, Dict, List, Optional from typing import Any, List, Optional, Tuple
import urllib.parse
from ._transaction import ( from ._jsonc import parse_jsonc
FileChange, from ._managed_files import (
read_optional, FileSnapshot,
regular_file_exists, MAX_CONFIG_BYTES,
source_change, 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 ( from .copilot_contract import (
AGENT_KEY, AGENT_KEY,
AUDIT_SCRIPT,
CONFIG_FILE, CONFIG_FILE,
GUARD_SCRIPT,
CopilotDocument, CopilotDocument,
CopilotSources, CopilotSources,
RenderedDocument,
RuntimeContract, RuntimeContract,
SourcePrecondition,
create_document, create_document,
parse_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]: @dataclass(frozen=True)
override = os.environ.get("COPILOT_HOME", "").strip() class CopilotPaths:
copilot_home = override or os.path.join(home_dir, ".copilot") copilot_home: str
user_path = os.path.join(copilot_home, "hooks", CONFIG_FILE) user_hooks_directory: str
legacy_path = os.path.join(os.getcwd(), ".github", "hooks", "hooks.json") user_hook_path: str
return user_path, legacy_path legacy_hook_path: str
settings_layers: Tuple[SettingsLayer, ...]
inspected_directories: Tuple[Tuple[str, str], ...]
def _read_document(file_path: str, label: str) -> Optional[CopilotDocument]: @dataclass(frozen=True)
raw = read_optional(file_path, label) class _ParsedSettingsLayer:
return None if raw is None else parse_document(file_path, raw, label) layer: SettingsLayer
disable_all_hooks: Optional[bool]
snapshot: Optional[FileSnapshot]
def read_sources(home_dir: str) -> CopilotSources: def resolve_copilot_paths(home_dir: Optional[str] = None) -> CopilotPaths:
user_path, legacy_path = config_paths(home_dir) home = home_dir if home_dir is not None else os.path.expanduser("~")
user = _read_document(user_path, "GitHub Copilot user hooks") override = os.environ.get("COPILOT_HOME")
legacy = _read_document(legacy_path, "GitHub Copilot legacy project hooks") copilot_home = override if override else os.path.join(home, ".copilot")
return CopilotSources(user or create_document(user_path), legacy) project = os.getcwd()
github = os.path.join(project, ".github")
github_copilot = os.path.join(github, "copilot")
def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: github_hooks = os.path.join(github, "hooks")
if not rendered.changed: claude = os.path.join(project, ".claude")
return None user_hooks = os.path.join(copilot_home, "hooks")
return source_change( return CopilotPaths(
rendered.document.file_path, copilot_home=copilot_home,
"GitHub Copilot hook source", user_hooks_directory=user_hooks,
rendered.document.raw, user_hook_path=os.path.join(user_hooks, CONFIG_FILE),
rendered.next_source, legacy_hook_path=os.path.join(github_hooks, "hooks.json"),
0o600, 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]: def _inspect_directories(locations: Tuple[Tuple[str, str], ...]) -> None:
raw = read_optional(file_path, "Elydora runtime config") for directory, label in locations:
if raw is None: physical_directory_exists(directory, label)
return None
try:
value = json.loads(raw) def _read_hook_document(
except json.JSONDecodeError as error: file_path: str, label: str
raise ValueError( ) -> Optional[CopilotDocument]:
f"Failed to parse Elydora runtime config at {file_path}: {error}" snapshot = read_physical_file(file_path, label, MAX_SOURCE_BYTES)
) from error 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): 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( 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 return value
def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: def _validate_runtime_config(
for contract in contracts: config: JsonObject,
agent_directory = os.path.dirname(contract.guard_path) 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_path = os.path.join(agent_directory, "config.json")
config = _read_runtime_config(config_path) 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 ( if (
config is None not isinstance(configured_id, str)
or not same_agent_id(configured_id, agent_id)
or config.get("agent_name") != AGENT_KEY or config.get("agent_name") != AGENT_KEY
or config.get("agent_id") != contract.agent_id
): ):
continue raise ValueError(
if regular_file_exists( "Elydora runtime config identity does not match Copilot agent "
contract.guard_path, "Elydora guard runtime" f"{agent_id}: {config_path}"
) and regular_file_exists( )
contract.audit_path, "Elydora audit runtime"
):
return True 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 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:
return any(_runtime_contract_exists(contract) for contract in contracts)
+276
View File
@@ -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}"
)
+1 -1
View File
@@ -28,7 +28,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = {
}, },
"copilot": { "copilot": {
"name": "GitHub Copilot CLI", "name": "GitHub Copilot CLI",
"hook_event": "preToolUse/postToolUse", "hook_event": "preToolUse/postToolUse/postToolUseFailure",
"config_path": "~/.copilot/hooks/elydora-audit.json", "config_path": "~/.copilot/hooks/elydora-audit.json",
}, },
"cursor": { "cursor": {
+216
View File
@@ -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
+3 -2
View File
@@ -390,9 +390,9 @@ def test_legacy_plugins_persist_one_owner_only_private_key(
runtime_root = case_root / ".elydora" runtime_root = case_root / ".elydora"
guard_path = runtime_root / "agent-1" / "guard.py" guard_path = runtime_root / "agent-1" / "guard.py"
guard_path.parent.mkdir(parents=True) 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") 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("HOME", str(case_root))
monkeypatch.setenv("USERPROFILE", str(case_root)) monkeypatch.setenv("USERPROFILE", str(case_root))
else: else:
@@ -410,6 +410,7 @@ def test_legacy_plugins_persist_one_owner_only_private_key(
monkeypatch.setattr(module, "PLUGIN_DIR", str(case_root / "plugins")) monkeypatch.setattr(module, "PLUGIN_DIR", str(case_root / "plugins"))
elif module is copilot: elif module is copilot:
case_root.mkdir(parents=True, exist_ok=True) case_root.mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("COPILOT_HOME", str(case_root / ".copilot"))
monkeypatch.chdir(case_root) monkeypatch.chdir(case_root)
plugin_type().install( plugin_type().install(
+100
View File
@@ -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("& ")
+250 -268
View File
@@ -1,183 +1,39 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass
import json import json
import os import os
from pathlib import Path from pathlib import Path
import subprocess
import sys
from typing import Any
import pytest import pytest
from elydora import cli from elydora import cli
from elydora.plugins import _transaction, copilot from elydora.plugins import copilot
from elydora.plugins.base import InstallConfig
from elydora.plugins.registry import SUPPORTED_AGENTS from elydora.plugins.registry import SUPPORTED_AGENTS
from copilot_support import (
AGENT_ID = "agent-1" AGENT_ID,
MISSING = object() 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) def test_copilot_is_registered_with_complete_native_contract() -> None:
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:
assert SUPPORTED_AGENTS["copilot"] == { assert SUPPORTED_AGENTS["copilot"] == {
"name": "GitHub Copilot CLI", "name": "GitHub Copilot CLI",
"hook_event": "preToolUse/postToolUse", "hook_event": "preToolUse/postToolUse/postToolUseFailure",
"config_path": "~/.copilot/hooks/elydora-audit.json", "config_path": "~/.copilot/hooks/elydora-audit.json",
} }
assert cli.PLUGIN_MAP["copilot"] is copilot.CopilotPlugin 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, monkeypatch: pytest.MonkeyPatch,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
@@ -188,15 +44,21 @@ def test_install_preserves_user_hooks_migrates_legacy_and_is_idempotent(
"version": 1, "version": 1,
"disableAllHooks": False, "disableAllHooks": False,
"hooks": { "hooks": {
"sessionStart": [{"type": "command", "command": "user-session"}], "sessionStart": [
"preToolUse": [{"type": "command", "command": "user-pre"}], {"type": "command", "command": "user-session"}
],
"preToolUse": [
{"type": "command", "command": "user-pre"}
],
}, },
}, },
) )
write_json_or_text( write_json_or_text(
fixture.legacy_path, fixture.legacy_path,
legacy_managed_config(fixture, { 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"]["preToolUse"]) == 2
assert len(settings["hooks"]["postToolUse"]) == 1 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, "preToolUse", "guard.py"))
assert_native_handler(managed_handler(settings, "postToolUse", "hook.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")) legacy = json.loads(fixture.legacy_path.read_text(encoding="utf-8"))
assert legacy == { assert legacy == {
"version": 1, "version": 1,
"hooks": { "hooks": {
"notification": [{"type": "command", "command": "user-notification"}], "notification": [
{"type": "command", "command": "user-notification"}
],
}, },
} }
assert fixture.runtime_config_path.is_file() runtime = json.loads(
assert fixture.private_key_path.read_text(encoding="utf-8") == "test-key" 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.hook_path.is_file()
assert fixture.plugin.status()["installed"] is True
def test_migration_removes_legacy_file_owned_entirely_by_elydora( 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")) 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, monkeypatch: pytest.MonkeyPatch,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path) fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config) fixture.plugin.install(fixture.config)
capture_path = tmp_path / "captured-event.json" 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( fixture.hook_path.write_text(
"from pathlib import Path\nimport sys\n" "from pathlib import Path\nimport sys\n"
f"Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n", f"Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n",
encoding="utf-8", encoding="utf-8",
) )
settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
pre_payload = json.dumps({ base = {
"sessionId": "session-1", "sessionId": "session-1",
"timestamp": 1, "timestamp": 1,
"cwd": str(fixture.project_dir), "cwd": str(fixture.project_dir),
"toolName": "powershell", "toolName": "powershell",
"toolArgs": {"command": "Get-ChildItem"}, "toolArgs": {"command": "Get-ChildItem"},
}, separators=(",", ":")) }
pre_payload = json.dumps(base, separators=(",", ":"))
guard = managed_handler(settings, "preToolUse", "guard.py") 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 guard_result.returncode == 2, guard_result.stderr
assert "Agent is frozen by Elydora" in guard_result.stderr assert "Agent is frozen by Elydora" in guard_result.stderr
post_payload = json.dumps({ success_payload = json.dumps(
"sessionId": "session-1", {**base, "toolResult": {"resultType": "success", "textResultForLlm": "ok"}},
"timestamp": 2, separators=(",", ":"),
"cwd": str(fixture.project_dir), )
"toolName": "powershell", success = managed_handler(settings, "postToolUse", "hook.py")
"toolArgs": {"command": "Get-ChildItem"}, success_result = run_hook(success, fixture, success_payload)
"toolResult": {"output": "ok"}, assert success_result.returncode == 0, success_result.stderr
}, separators=(",", ":")) assert capture_path.read_text(encoding="utf-8") == success_payload
audit = managed_handler(settings, "postToolUse", "hook.py")
audit_result = run_hook(audit, fixture.home_dir, post_payload) failure_payload = json.dumps(
assert audit_result.returncode == 0, audit_result.stderr {**base, "error": "command failed"},
assert capture_path.read_text(encoding="utf-8") == post_payload 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( def test_empty_home_override_uses_official_default(
@@ -287,32 +175,192 @@ def test_empty_home_override_uses_official_default(
) -> None: ) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path) fixture = prepare_fixture(monkeypatch, tmp_path)
monkeypatch.setenv("COPILOT_HOME", "") monkeypatch.setenv("COPILOT_HOME", "")
fixture.plugin.install(fixture.config) fixture.plugin.install(fixture.config)
default_path = fixture.home_dir / ".copilot" / "hooks" / "elydora-audit.json" default_path = fixture.home_dir / ".copilot" / "hooks" / "elydora-audit.json"
settings = json.loads(default_path.read_text(encoding="utf-8")) settings = json.loads(default_path.read_text(encoding="utf-8"))
assert_native_handler(managed_handler(settings, "preToolUse", "guard.py")) assert_native_handler(managed_handler(settings, "preToolUse", "guard.py"))
assert str(default_path) in fixture.plugin.status()["details"] 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, monkeypatch: pytest.MonkeyPatch,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path) fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config) fixture.plugin.install(fixture.config)
assert fixture.plugin.status()["installed"] is True assert fixture.plugin.status()["installed"] is True
settings = json.loads(fixture.config_path.read_text(encoding="utf-8")) 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) write_json_or_text(fixture.config_path, settings)
assert fixture.plugin.status()["installed"] is False assert fixture.plugin.status()["installed"] is False
fixture.plugin.install(fixture.config) 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") 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() 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": "(?<tool>shell)",
}],
},
},
)
fixture.plugin.install(fixture.config)
settings = json.loads(fixture.config_path.read_text(encoding="utf-8"))
assert settings["hooks"]["preToolUse"][0]["matcher"] == (
"(?<tool>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, monkeypatch: pytest.MonkeyPatch,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> 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 = json.loads(fixture.config_path.read_text(encoding="utf-8"))
settings["hooks"]["preToolUse"].append({ settings["hooks"]["preToolUse"].append({
"type": "command", "type": "command",
"bash": f"{sys.executable} {fixture.agent_dir.parent / 'agent-10' / 'guard.py'}", "bash": "user-pre",
"powershell": "user-decoy", "powershell": "user-pre",
"timeoutSec": 10, "timeoutSec": 10,
}) })
write_json_or_text(fixture.config_path, settings) 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"] == [ assert remaining["hooks"]["notification"] == [
{"type": "command", "command": "keep"} {"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 "postToolUse" not in remaining["hooks"]
assert len(remaining["hooks"]["preToolUse"]) == 1 assert "postToolUseFailure" not in remaining["hooks"]
assert "agent-10" in remaining["hooks"]["preToolUse"][0]["bash"]
def test_uninstall_leaves_absent_sources_absent( 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 assert fixture.legacy_path.exists() is False
@pytest.mark.parametrize( def test_preflight_validates_credentials_and_runtime_path_without_writes(
"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(
monkeypatch: pytest.MonkeyPatch, monkeypatch: pytest.MonkeyPatch,
tmp_path: Path, tmp_path: Path,
) -> None: ) -> None:
fixture = prepare_fixture(monkeypatch, tmp_path) fixture = prepare_fixture(monkeypatch, tmp_path)
fixture.plugin.install(fixture.config) invalid = dict(fixture.config)
for directory in (fixture.agent_dir, fixture.config_path.parent): invalid["private_key"] = "invalid"
assert all( with pytest.raises(ValueError, match="private_key"):
path.suffix not in {".tmp", ".rollback"} fixture.plugin.preflight_install(invalid) # type: ignore[arg-type]
for path in directory.iterdir() assert fixture.config_path.exists() is False
) assert_runtime_absent(fixture)
+238
View File
@@ -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()
)