diff --git a/elydora/plugins/_runtime.py b/elydora/plugins/_runtime.py new file mode 100644 index 0000000..4dba90e --- /dev/null +++ b/elydora/plugins/_runtime.py @@ -0,0 +1,338 @@ +"""Runtime directory identity, config validation, and status checks shared by providers.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +import json +import os +from typing import Any, List, Optional, Protocol, Tuple +import urllib.parse + +from elydora._runtime_paths import resolve_agent_directory, runtime_root + +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + MAX_SOURCE_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, +) +from ._strict_json import JsonObject, parse_json_object +from ._transaction import FileChange, file_change, source_change +from .base import InstallConfig + + +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +DEFAULT_BASE_URL = "https://api.elydora.com" +RUNTIME_CONFIG_FIELDS = frozenset( + {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} +) +RUNTIME_ARTIFACTS = ( + ("private.key", "Elydora private key"), + ("guard.py", "Elydora guard runtime"), + ("hook.py", "Elydora audit runtime"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), +) + + +@dataclass(frozen=True) +class RuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +class RuntimeContractLike(Protocol): + @property + def agent_id(self) -> str: ... + + @property + def guard_path(self) -> str: ... + + @property + def audit_path(self) -> str: ... + + +def same_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def same_agent_id(value: Any, expected: str) -> bool: + return isinstance(value, str) and os.path.normcase(value) == os.path.normcase( + expected + ) + + +def managed_script_reference( + script_path: str, script_name: str +) -> Optional[Tuple[str, str]]: + """Return (agent_id, script_path) for an absolute ~/.elydora//.""" + if not os.path.isabs(script_path) or os.path.basename(script_path) != script_name: + return None + agent_directory = os.path.dirname(script_path) + if not same_path(os.path.dirname(agent_directory), runtime_root()): + return None + agent_id = os.path.basename(agent_directory) + if agent_id in ("", ".", ".."): + return None + return agent_id, script_path + + +def validate_api_origin(value: str, label: str = "base_url") -> None: + try: + parsed = urllib.parse.urlsplit(value) + hostname = parsed.hostname + parsed.port + except ValueError as error: + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error + invalid_character = "\\" in value or any( + character.isspace() or ord(character) < 32 for character in value + ) + if ( + parsed.scheme not in ("http", "https") + or not parsed.netloc + or hostname is None + or invalid_character + ): + raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") + if ( + parsed.username is not None + or parsed.password is not None + or parsed.query + or parsed.fragment + ): + raise ValueError( + f"{label} must exclude credentials, query parameters, and fragments" + ) + + +def validate_private_key(value: str, label: str = "private_key") -> None: + try: + padded = value + "=" * ((4 - len(value) % 4) % 4) + seed = base64.b64decode( + padded.replace("-", "+").replace("_", "/"), validate=True + ) + canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") + except (ValueError, UnicodeEncodeError) as error: + raise ValueError( + f"{label} must be a canonical 32-byte base64url value" + ) from error + if len(seed) != 32 or canonical != value: + raise ValueError(f"{label} must be a canonical 32-byte base64url value") + + +def require_non_empty_string(value: Any, field: str, config_path: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"Elydora runtime config {field} is invalid: {config_path}") + return value + + +def 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_config( + config: JsonObject, + expected_agent_id: str, + config_path: str, + agent_key: str, + product_label: str, +) -> None: + extra = next((key for key in config if key not in RUNTIME_CONFIG_FIELDS), None) + if extra is not None: + raise ValueError( + f'Elydora runtime config has unsupported field "{extra}": {config_path}' + ) + require_non_empty_string(config.get("org_id"), "org_id", config_path) + require_non_empty_string(config.get("kid"), "kid", config_path) + agent_id = require_non_empty_string(config.get("agent_id"), "agent_id", config_path) + if not same_agent_id(agent_id, expected_agent_id) or config.get("agent_name") != agent_key: + raise ValueError( + f"Elydora runtime identity does not match {product_label} hooks: {config_path}" + ) + if "token" in config: + require_non_empty_string(config.get("token"), "token", config_path) + base_url = require_non_empty_string(config.get("base_url"), "base_url", config_path) + validate_api_origin(base_url, "Elydora runtime config base_url") + + +def runtime_config_source(config: InstallConfig, agent_id: str, agent_key: str) -> str: + value = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", DEFAULT_BASE_URL), + "agent_name": agent_key, + } + token = config.get("token") + if token: + value["token"] = token + return json.dumps(value, indent=2) + "\n" + + +def resolve_runtime_paths(config: InstallConfig) -> RuntimePaths: + 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) + if not same_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + f"Elydora guard runtime must use the managed agent directory: {guard_path}" + ) + return RuntimePaths( + agent_id, agent_directory, guard_path, os.path.join(agent_directory, AUDIT_SCRIPT) + ) + + +def validate_install_config( + config: InstallConfig, agent_key: str, product_label: str +) -> 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"{product_label} installation requires agent_name {agent_key}") + validate_private_key(config["private_key"]) + validate_api_origin(config["base_url"]) + if "token" in config: + token = config["token"] + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string when provided") + resolve_runtime_paths(config) + + +def validate_runtime_tree( + agent_directory: str, agent_id: str, agent_key: str, product_label: str +) -> None: + """Reject an existing agent directory that belongs to another identity.""" + if not physical_directory_exists(runtime_root(), "Elydora runtime directory"): + return + if not physical_directory_exists(agent_directory, "Elydora agent runtime directory"): + return + config_path = os.path.join(agent_directory, "config.json") + config = read_runtime_config(config_path) + artifact_states = [ + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in RUNTIME_ARTIFACTS + ] + artifact_exists = any(artifact_states) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + if ( + not same_agent_id(config.get("agent_id"), agent_id) + or config.get("agent_name") != agent_key + ): + raise ValueError( + f"Elydora runtime config identity does not match {product_label} agent " + f"{agent_id}: {config_path}" + ) + + +def runtime_contract_exists( + contract: RuntimeContractLike, agent_key: str, product_label: str +) -> bool: + """Strict status check: valid config and key plus non-empty runtime scripts.""" + root = runtime_root() + agent_directory = os.path.dirname(contract.guard_path) + if not same_path(os.path.dirname(agent_directory), root) or not same_path( + contract.audit_path, os.path.join(agent_directory, AUDIT_SCRIPT) + ): + return False + if not physical_directory_exists(root, "Elydora runtime directory"): + return False + if not physical_directory_exists(agent_directory, "Elydora agent runtime directory"): + return False + config_path = os.path.join(agent_directory, "config.json") + config = read_runtime_config(config_path) + key = read_physical_file( + os.path.join(agent_directory, "private.key"), "Elydora private key", MAX_SECRET_BYTES + ) + guard = read_physical_file(contract.guard_path, "Elydora guard runtime") + audit = read_physical_file(contract.audit_path, "Elydora audit runtime") + if config is None or key is None or guard is None or audit is None: + return False + validate_runtime_config(config, contract.agent_id, config_path, agent_key, product_label) + validate_private_key(key.contents, "Elydora private key") + return bool(guard.contents) and bool(audit.contents) + + +def runtime_present(contract: RuntimeContractLike, agent_key: str) -> bool: + """Presence-only status check; identity mismatch reports False.""" + agent_directory = os.path.dirname(contract.guard_path) + if not physical_directory_exists(agent_directory, "Elydora agent runtime directory"): + return False + config = read_runtime_config(os.path.join(agent_directory, "config.json")) + if ( + config is None + or config.get("agent_name") != agent_key + or not same_agent_id(config.get("agent_id"), contract.agent_id) + ): + return False + return all( + ( + physical_file_exists(contract.guard_path, "Elydora guard runtime"), + physical_file_exists(contract.audit_path, "Elydora audit runtime"), + physical_file_exists( + os.path.join(agent_directory, "private.key"), "Elydora private key" + ), + ) + ) + + +def present(changes: List[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def runtime_file_changes( + config: InstallConfig, + paths: RuntimePaths, + agent_key: str, + guard_script: str, + audit_script: str, +) -> List[FileChange]: + """Changes for the guard, config, key, and audit runtime files.""" + return present([ + file_change(paths.guard_path, "Elydora guard runtime", guard_script, 0o700), + file_change( + os.path.join(paths.agent_directory, "config.json"), + "Elydora runtime config", + runtime_config_source(config, paths.agent_id, agent_key), + 0o600, + MAX_CONFIG_BYTES, + ), + file_change( + os.path.join(paths.agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + file_change(paths.audit_path, "Elydora audit runtime", audit_script, 0o700), + ]) + + +def rendered_source_change( + file_path: str, + label: str, + original: Optional[str], + next_source: Optional[str], +) -> Optional[FileChange]: + return source_change(file_path, label, original, next_source, 0o600, MAX_SOURCE_BYTES) diff --git a/elydora/plugins/_shell_command.py b/elydora/plugins/_shell_command.py new file mode 100644 index 0000000..f0e6db1 --- /dev/null +++ b/elydora/plugins/_shell_command.py @@ -0,0 +1,151 @@ +"""Shell quoting and parsing for generated hook commands.""" + +from __future__ import annotations + +import base64 +import ntpath +import os +import re +import sys +from typing import Optional, Tuple + +CommandParts = Tuple[str, str] +POWERSHELL_EXIT_SUFFIX = "; exit $LASTEXITCODE" +_POSIX_APOSTROPHE = "'\"'\"'" +_ENCODED_COMMAND = re.compile( + r'"([^"\r\n]+)" -NoLogo -NoProfile -NonInteractive ' + r"-EncodedCommand ([A-Za-z0-9+/]+={0,2})" +) +_PYTHON_EXECUTABLE = re.compile( + r"(?:python(?:[0-9]+(?:\.[0-9]+)*)?[td]?|pypy(?:[0-9]+(?:\.[0-9]+)*)?)(?:\.exe)?", + re.IGNORECASE, +) + + +def quote_posix(value: str) -> str: + return "'" + value.replace("'", _POSIX_APOSTROPHE) + "'" + + +def quote_powershell(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def posix_source(script_path: str) -> str: + return f"{quote_posix(sys.executable)} {quote_posix(script_path)}" + + +def powershell_source(script_path: str) -> str: + return ( + f"& {quote_powershell(sys.executable)} {quote_powershell(script_path)}" + f"{POWERSHELL_EXIT_SUFFIX}" + ) + + +def windows_powershell_path() -> str: + configured = os.environ.get("SystemRoot") if os.name == "nt" else None + system_root = ( + configured + if configured + and ntpath.isabs(configured) + and re.search(r'["%\r\n]', configured) is None + else r"C:\Windows" + ) + return ntpath.join(system_root, "System32", "WindowsPowerShell", "v1.0", "powershell.exe") + + +def encode_powershell_source(source: str) -> str: + return base64.b64encode(source.encode("utf-16le")).decode("ascii") + + +def encoded_windows_command(script_path: str) -> str: + return ( + f'"{windows_powershell_path()}" -NoLogo -NoProfile -NonInteractive ' + f"-EncodedCommand {encode_powershell_source(powershell_source(script_path))}" + ) + + +def is_python_executable(file_path: str) -> bool: + return _PYTHON_EXECUTABLE.fullmatch(os.path.basename(file_path)) is not None + + +def is_powershell_executable(file_path: str) -> bool: + return ntpath.isabs(file_path) and ntpath.basename(file_path).lower() == "powershell.exe" + + +def read_posix_argument(command: str, start: int) -> Optional[Tuple[str, int]]: + if start >= len(command) or command[start] != "'": + return None + value = "" + index = start + 1 + while index < len(command): + if command.startswith(_POSIX_APOSTROPHE, index): + value += "'" + index += len(_POSIX_APOSTROPHE) + continue + if command[index] == "'": + return value, index + 1 + value += command[index] + index += 1 + return None + + +def parse_posix_command(command: str) -> Optional[CommandParts]: + executable = read_posix_argument(command, 0) + if executable is None or command[executable[1] : executable[1] + 1] != " ": + return None + script = read_posix_argument(command, executable[1] + 1) + if script is None or script[1] != len(command): + return None + return executable[0], script[0] + + +def read_powershell_argument(command: str, start: int) -> Optional[Tuple[str, int]]: + if start >= len(command) or command[start] != "'": + return None + value = "" + index = start + 1 + while index < len(command): + if command[index] != "'": + value += command[index] + index += 1 + continue + if index + 1 < len(command) and command[index + 1] == "'": + value += "'" + index += 2 + continue + return value, index + 1 + return None + + +def parse_powershell_source(source: str, prefix: str = "& ") -> Optional[CommandParts]: + if not source.startswith(prefix): + return None + executable = read_powershell_argument(source, len(prefix)) + if executable is None or source[executable[1] : executable[1] + 1] != " ": + return None + script = read_powershell_argument(source, executable[1] + 1) + if script is None or source[script[1] :] != POWERSHELL_EXIT_SUFFIX: + return None + return executable[0], script[0] + + +def decode_powershell_source(encoded: str) -> Optional[str]: + if re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", encoded) is None: + return None + try: + payload = base64.b64decode(encoded, validate=True) + if base64.b64encode(payload).decode("ascii") != encoded or len(payload) % 2: + return None + return payload.decode("utf-16le") + except (ValueError, UnicodeDecodeError): + return None + + +def parse_encoded_windows_command( + command: str, prefix: str = "& " +) -> Optional[CommandParts]: + match = _ENCODED_COMMAND.fullmatch(command) + if match is None or not is_powershell_executable(match.group(1)): + return None + source = decode_powershell_source(match.group(2)) + return None if source is None else parse_powershell_source(source, prefix) diff --git a/elydora/plugins/claudecode_contract.py b/elydora/plugins/claudecode_contract.py index f0f482b..cfca1ca 100644 --- a/elydora/plugins/claudecode_contract.py +++ b/elydora/plugins/claudecode_contract.py @@ -11,6 +11,7 @@ import sys from typing import Any, Dict, List, Optional, Tuple import urllib.parse +from ._runtime import managed_script_reference, same_agent_id, same_path from ._strict_json import JsonObject, parse_json_object @@ -82,20 +83,6 @@ class _RuntimeReference: script_path: str -def runtime_root() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - -def same_claude_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_claude_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - def _require_known_keys( value: JsonObject, allowed: frozenset[str], label: str ) -> None: @@ -292,15 +279,8 @@ def _exact_managed_group(group: JsonObject) -> bool: def _runtime_reference( script_path: str, script_name: str ) -> Optional[_RuntimeReference]: - if not os.path.isabs(script_path) or os.path.basename(script_path) != script_name: - return None - agent_directory = os.path.dirname(script_path) - if not same_claude_path(os.path.dirname(agent_directory), runtime_root()): - return None - agent_id = os.path.basename(agent_directory) - if agent_id in ("", ".", ".."): - return None - return _RuntimeReference(agent_id, script_path) + reference = managed_script_reference(script_path, script_name) + return None if reference is None else _RuntimeReference(*reference) def _legacy_reference( @@ -314,7 +294,7 @@ def _legacy_reference( if ( match is None or not os.path.isabs(match.group(1)) - or not same_claude_path(match.group(1), sys.executable) + or not same_path(match.group(1), sys.executable) ): return None return _runtime_reference(match.group(2), script_name) @@ -334,7 +314,7 @@ def _managed_reference( and handler.get("statusMessage") == status_message and isinstance(handler.get("command"), str) and os.path.isabs(str(handler["command"])) - and same_claude_path(str(handler["command"]), sys.executable) + and same_path(str(handler["command"]), sys.executable) and isinstance(handler.get("args"), list) and len(handler["args"]) == 1 and isinstance(handler["args"][0], str) @@ -365,7 +345,7 @@ def _remove_from_groups( ) owned = reference is not None and ( not agent_id - or same_claude_agent_id(reference.agent_id, agent_id) + or same_agent_id(reference.agent_id, agent_id) ) if owned: removed = True @@ -485,7 +465,7 @@ def claude_runtime_contracts( failure = failures.get(key, []) if len(guard) != 1 or len(success) != 1 or len(failure) != 1: continue - if not same_claude_path(success[0].script_path, failure[0].script_path): + if not same_path(success[0].script_path, failure[0].script_path): continue contracts.append(ClaudeRuntimeContract( guard[0].agent_id, diff --git a/elydora/plugins/claudecode_installation.py b/elydora/plugins/claudecode_installation.py index 49cc92a..74b7238 100644 --- a/elydora/plugins/claudecode_installation.py +++ b/elydora/plugins/claudecode_installation.py @@ -2,122 +2,41 @@ from __future__ import annotations -from dataclasses import dataclass -import json -import os -from typing import List, Optional +from typing import List -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import FileChange, file_change, write_changes -from .base import InstallConfig -from .claudecode_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - ClaudeDocument, - RenderedClaudeDocument, - runtime_root, - same_claude_path, -) -from .claudecode_io import ( - rendered_change, - validate_api_origin, - validate_private_key, +from ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + present, + resolve_runtime_paths, + runtime_file_changes, + validate_install_config, validate_runtime_tree, ) +from ._transaction import FileChange, write_changes +from .base import InstallConfig +from .claudecode_contract import AGENT_KEY, ClaudeDocument, RenderedClaudeDocument +from .claudecode_io import rendered_change from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -@dataclass(frozen=True) -class ClaudeRuntimePaths: - agent_id: str - agent_directory: str - guard_path: str - audit_path: str - - -def _runtime_config(config: InstallConfig, agent_id: str) -> dict: - value = { - "org_id": config.get("org_id", ""), - "agent_id": agent_id, - "kid": config.get("kid", ""), - "base_url": config.get("base_url", "https://api.elydora.com"), - "agent_name": AGENT_KEY, - } - token = config.get("token") - if token: - value["token"] = token - return value - - -def _json_source(value: dict) -> str: - return json.dumps(value, indent=2) + "\n" - - -def _agent_paths(config: InstallConfig) -> ClaudeRuntimePaths: - 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_claude_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return ClaudeRuntimePaths(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"Claude Code installation requires agent_name {AGENT_KEY}" - ) - validate_private_key(config["private_key"]) - validate_api_origin(config["base_url"]) - if "token" in config: - token = config["token"] - if not isinstance(token, str) or not token: - raise ValueError("token must be a non-empty string when provided") - _agent_paths(config) +PRODUCT = "Claude Code" def preflight_claude_installation( config: InstallConfig, document: ClaudeDocument -) -> ClaudeRuntimePaths: +) -> RuntimePaths: if not document.file_path: raise ValueError("Claude Code installation requires a settings path") - _validate_install_config(config) - paths = _agent_paths(config) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + validate_install_config(config, AGENT_KEY, PRODUCT) + paths = resolve_runtime_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id, AGENT_KEY, PRODUCT) return paths -def _runtime_change( - file_path: str, - label: str, - next_source: str, - mode: int, - maximum_bytes: int = MAX_SOURCE_BYTES, -) -> Optional[FileChange]: - return file_change(file_path, label, next_source, mode, maximum_bytes) - - -def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - def prepare_claude_installation( config: InstallConfig, - paths: ClaudeRuntimePaths, + paths: RuntimePaths, rendered: RenderedClaudeDocument, ) -> List[FileChange]: guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) @@ -125,43 +44,22 @@ def prepare_claude_installation( 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"), + base_url=config.get("base_url", DEFAULT_BASE_URL), native_payload=True, agent_name=AGENT_KEY, ) - return _present([ - _runtime_change( - paths.guard_path, "Elydora guard runtime", guard_script, 0o700 - ), - _runtime_change( - os.path.join(paths.agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, paths.agent_id)), - 0o600, - MAX_CONFIG_BYTES, - ), - _runtime_change( - os.path.join(paths.agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - MAX_SECRET_BYTES, - ), - _runtime_change( - paths.audit_path, "Elydora audit runtime", audit_script, 0o700 - ), - rendered_change(rendered), - ]) + return [ + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *present([rendered_change(rendered)]), + ] def commit_claude_installation(changes: List[FileChange]) -> None: write_changes(changes, "Install Claude Code hooks") -def prepare_claude_uninstall( - rendered: RenderedClaudeDocument, -) -> List[FileChange]: - return _present([rendered_change(rendered)]) +def prepare_claude_uninstall(rendered: RenderedClaudeDocument) -> List[FileChange]: + return present([rendered_change(rendered)]) def commit_claude_uninstall(changes: List[FileChange]) -> None: diff --git a/elydora/plugins/claudecode_io.py b/elydora/plugins/claudecode_io.py index ca0b95b..a1e26de 100644 --- a/elydora/plugins/claudecode_io.py +++ b/elydora/plugins/claudecode_io.py @@ -2,21 +2,12 @@ from __future__ import annotations -import base64 import os -from typing import Any, List, Optional -import urllib.parse +from typing import List, Optional -from ._managed_files import ( - MAX_CONFIG_BYTES, - MAX_SECRET_BYTES, - MAX_SOURCE_BYTES, - physical_directory_exists, - physical_file_exists, - read_physical_file, -) -from ._strict_json import JsonObject, parse_json_object -from ._transaction import FileChange, source_change, write_changes +from ._managed_files import MAX_SOURCE_BYTES, physical_directory_exists, read_physical_file +from ._runtime import rendered_source_change, runtime_contract_exists +from ._transaction import FileChange, write_changes from .claudecode_contract import ( AGENT_KEY, CONFIG_FILE, @@ -25,9 +16,6 @@ from .claudecode_contract import ( RenderedClaudeDocument, create_claude_document, parse_claude_document, - runtime_root, - same_claude_agent_id, - same_claude_path, ) @@ -43,29 +31,21 @@ def claude_settings_path() -> str: def read_claude_document() -> ClaudeDocument: file_path = claude_settings_path() - physical_directory_exists( - os.path.dirname(file_path), "Claude Code configuration directory" - ) - snapshot = read_physical_file( - file_path, "Claude Code user settings", MAX_SOURCE_BYTES - ) + physical_directory_exists(os.path.dirname(file_path), "Claude Code configuration directory") + snapshot = read_physical_file(file_path, "Claude Code user settings", MAX_SOURCE_BYTES) if snapshot is None: return create_claude_document(file_path) return parse_claude_document(file_path, snapshot.contents) -def rendered_change( - rendered: RenderedClaudeDocument, -) -> Optional[FileChange]: +def rendered_change(rendered: RenderedClaudeDocument) -> Optional[FileChange]: if not rendered.changed: return None - return source_change( + return rendered_source_change( rendered.document.file_path, "Claude Code user settings", rendered.document.raw, rendered.next_source, - 0o600, - MAX_SOURCE_BYTES, ) @@ -75,172 +55,7 @@ def write_claude_document(rendered: RenderedClaudeDocument) -> None: write_changes([change], "Uninstall Claude Code hooks") -def validate_api_origin(value: str, label: str = "base_url") -> None: - try: - parsed = urllib.parse.urlsplit(value) - hostname = parsed.hostname - parsed.port - except ValueError as error: - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error - invalid_character = "\\" in value or any( - character.isspace() or ord(character) < 32 for character in value +def claude_runtime_files_exist(contracts: List[ClaudeRuntimeContract]) -> bool: + return any( + runtime_contract_exists(contract, AGENT_KEY, "Claude Code") for contract in contracts ) - if ( - parsed.scheme not in ("http", "https") - or not parsed.netloc - or hostname is None - or invalid_character - ): - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") - if ( - parsed.username is not None - or parsed.password is not None - or parsed.query - or parsed.fragment - ): - raise ValueError( - f"{label} must exclude credentials, query parameters, and fragments" - ) - - -def validate_private_key(value: str, label: str = "private_key") -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), validate=True - ) - canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - f"{label} must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or canonical != value: - raise ValueError(f"{label} must be a canonical 32-byte base64url value") - - -def _require_non_empty_string( - value: Any, field: str, config_path: str -) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError( - f"Elydora runtime config {field} is invalid: {config_path}" - ) - return value - - -def _validate_runtime_config( - config: JsonObject, expected_agent_id: str, config_path: str -) -> None: - supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} - extra = next((key for key in config if key not in supported), None) - if extra is not None: - raise ValueError( - f'Elydora runtime config has unsupported field "{extra}": {config_path}' - ) - _require_non_empty_string(config.get("org_id"), "org_id", config_path) - _require_non_empty_string(config.get("kid"), "kid", config_path) - agent_id = _require_non_empty_string( - config.get("agent_id"), "agent_id", config_path - ) - if ( - not same_claude_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime identity does not match Claude Code hooks: " - f"{config_path}" - ) - if "token" in config: - _require_non_empty_string(config.get("token"), "token", config_path) - base_url = _require_non_empty_string( - config.get("base_url"), "base_url", config_path - ) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, "Elydora runtime config", MAX_CONFIG_BYTES - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, f"Elydora runtime config at {file_path}" - ) - - -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = runtime_root() - if not physical_directory_exists(root, "Elydora runtime directory"): - return - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return - config_path = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path) - artifacts = ( - ("private.key", "Elydora private key"), - ("guard.py", "Elydora guard runtime"), - ("hook.py", "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ) - artifact_states = [ - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ] - artifact_exists = any(artifact_states) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not same_claude_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Claude Code agent " - f"{agent_id}: {config_path}" - ) - - -def _runtime_contract_exists(contract: ClaudeRuntimeContract) -> bool: - root = runtime_root() - agent_directory = os.path.dirname(contract.guard_path) - if ( - not same_claude_path(os.path.dirname(agent_directory), root) - or not same_claude_path( - contract.audit_path, os.path.join(agent_directory, "hook.py") - ) - ): - return False - if not physical_directory_exists(root, "Elydora runtime directory"): - return False - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return False - config_path = os.path.join(agent_directory, "config.json") - key_path = os.path.join(agent_directory, "private.key") - config = _read_runtime_config(config_path) - key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) - guard = read_physical_file(contract.guard_path, "Elydora guard runtime") - audit = read_physical_file(contract.audit_path, "Elydora audit runtime") - if config is None or key is None or guard is None or audit is None: - return False - _validate_runtime_config(config, contract.agent_id, config_path) - validate_private_key(key.contents, "Elydora private key") - return bool(guard.contents) and bool(audit.contents) - - -def claude_runtime_files_exist( - contracts: List[ClaudeRuntimeContract], -) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/elydora/plugins/codex.py b/elydora/plugins/codex.py index 9fe7a97..b670c30 100644 --- a/elydora/plugins/codex.py +++ b/elydora/plugins/codex.py @@ -2,134 +2,35 @@ from __future__ import annotations -import base64 -import json -import os -import urllib.parse -from typing import List, Optional - -from elydora._runtime_paths import resolve_agent_directory -from elydora.utils import base64url_encode - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES -from ._transaction import FileChange +from ._runtime import ( + DEFAULT_BASE_URL, + present, + resolve_runtime_paths, + runtime_file_changes, + validate_install_config, + validate_runtime_tree, +) +from ._transaction import write_changes from .base import AgentPlugin, InstallConfig, PluginStatus from .codex_contract import ( AGENT_KEY, - AUDIT_SCRIPT, AUDIT_STATUS, - GUARD_SCRIPT, GUARD_STATUS, build_handler, remove_managed_hooks, render_document, runtime_contracts, - runtime_root, - same_path, ) from .codex_io import ( read_document, rendered_change, - runtime_change, runtime_files_exist, validate_hooks_directory, - validate_runtime_tree, - write_codex_changes, ) from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -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 _present(changes: List[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - -def _agent_paths(config: InstallConfig) -> tuple[str, str, str, str]: - 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) - configured_guard = config.get("guard_script_path", "") - if not same_path(configured_guard, guard_path): - raise ValueError( - f"Elydora guard runtime must use the managed agent directory: {guard_path}" - ) - return agent_id, agent_directory, guard_path, audit_path - - -def _validate_private_key(value: str) -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), - validate=True, - ) - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - "private_key must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or base64url_encode(seed) != value: - raise ValueError("private_key must be a canonical 32-byte base64url value") - - -def _validate_install_config(config: InstallConfig) -> None: - for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): - value = config.get(field) - if not isinstance(value, str) or not value: - raise ValueError(f"{field} is required") - if config.get("agent_name") != AGENT_KEY: - raise ValueError(f"Codex installation requires agent_name {AGENT_KEY}") - _validate_private_key(config["private_key"]) - - base_url = config["base_url"] - try: - parsed = urllib.parse.urlsplit(base_url) - hostname = parsed.hostname - parsed.port - has_credentials = parsed.username is not None or parsed.password is not None - except ValueError as error: - raise ValueError("base_url must be an absolute HTTP or HTTPS URL") from error - invalid_character = "\\" in base_url or any( - character.isspace() or ord(character) < 32 for character in base_url - ) - valid_origin = ( - parsed.scheme in ("http", "https") - and bool(parsed.netloc) - and hostname is not None - and not invalid_character - ) - if not valid_origin: - raise ValueError("base_url must be an absolute HTTP or HTTPS URL") - if has_credentials or parsed.query or parsed.fragment: - raise ValueError( - "base_url must exclude credentials, query parameters, and fragments" - ) - - if "token" in config: - token = config["token"] - if not isinstance(token, str) or not token: - raise ValueError("token must be a non-empty string when provided") - _agent_paths(config) +PRODUCT = "Codex" class CodexPlugin(AgentPlugin): @@ -138,74 +39,39 @@ class CodexPlugin(AgentPlugin): manages_guard_runtime = True def preflight_install(self, config: InstallConfig) -> None: - _validate_install_config(config) + validate_install_config(config, AGENT_KEY, PRODUCT) document = read_document() validate_hooks_directory(document.file_path) - agent_id, agent_directory, _, _ = _agent_paths(config) - validate_runtime_tree(runtime_root(), agent_directory, agent_id) + paths = resolve_runtime_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id, AGENT_KEY, PRODUCT) def install(self, config: InstallConfig) -> None: self.preflight_install(config) document = read_document() - agent_id, agent_directory, guard_path, audit_path = _agent_paths(config) - + paths = resolve_runtime_paths(config) hooks = remove_managed_hooks(document.hooks) hooks["PreToolUse"] = [ *hooks.get("PreToolUse", []), - { - "matcher": "*", - "hooks": [build_handler(guard_path, GUARD_STATUS)], - }, + {"matcher": "*", "hooks": [build_handler(paths.guard_path, GUARD_STATUS)]}, ] hooks["PostToolUse"] = [ *hooks.get("PostToolUse", []), - { - "matcher": "*", - "hooks": [build_handler(audit_path, AUDIT_STATUS)], - }, + {"matcher": "*", "hooks": [build_handler(paths.audit_path, AUDIT_STATUS)]}, ] - - guard_script = generate_guard_script(AGENT_KEY, agent_id) + guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) audit_script = generate_hook_script( org_id=config.get("org_id", ""), - agent_id=agent_id, + agent_id=paths.agent_id, kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), + base_url=config.get("base_url", DEFAULT_BASE_URL), native_payload=True, agent_name=AGENT_KEY, ) - changes = _present( - [ - runtime_change( - guard_path, - "Elydora guard runtime", - guard_script, - 0o700, - ), - runtime_change( - os.path.join(agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, agent_id)), - 0o600, - MAX_CONFIG_BYTES, - ), - runtime_change( - os.path.join(agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - MAX_SECRET_BYTES, - ), - runtime_change( - audit_path, - "Elydora audit runtime", - audit_script, - 0o700, - ), - rendered_change(render_document(document, hooks)), - ] - ) - write_codex_changes(changes, "Install Codex hooks") + changes = [ + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *present([rendered_change(render_document(document, hooks))]), + ] + write_changes(changes, "Install Codex hooks") print(f" Codex hooks: {document.file_path}") print(" Codex trust: run /hooks and approve both Elydora command hooks.") @@ -216,25 +82,17 @@ class CodexPlugin(AgentPlugin): if change is None: return validate_hooks_directory(document.file_path) - write_codex_changes([change], "Uninstall Codex hooks") + write_changes([change], "Uninstall Codex hooks") def status(self) -> PluginStatus: document = read_document() contracts = runtime_contracts(document.hooks) if not contracts: - return PluginStatus( - installed=False, - agent=AGENT_KEY, - details="Not installed", - ) + return PluginStatus(installed=False, agent=AGENT_KEY, details="Not installed") installed = runtime_files_exist(contracts) details = ( f"Config: {document.file_path}" if installed else f"Configured at {document.file_path}; runtime files missing" ) - return PluginStatus( - installed=installed, - agent=AGENT_KEY, - details=details, - ) + return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) diff --git a/elydora/plugins/codex_contract.py b/elydora/plugins/codex_contract.py index 1218090..e618280 100644 --- a/elydora/plugins/codex_contract.py +++ b/elydora/plugins/codex_contract.py @@ -2,17 +2,23 @@ from __future__ import annotations -import base64 from dataclasses import dataclass import json -import ntpath import os -import re import shlex -import subprocess # nosec B404 - used only for legacy argument rendering -import sys +import subprocess from typing import Any, Dict, List, Optional, Tuple +from elydora._runtime_paths import runtime_root + +from ._runtime import managed_script_reference, same_agent_id, same_path +from ._shell_command import ( + encoded_windows_command, + is_python_executable, + parse_encoded_windows_command, + parse_posix_command, + posix_source, +) from ._strict_json import JsonObject, parse_json_object @@ -51,164 +57,16 @@ class RuntimeContract: audit_path: str -@dataclass(frozen=True) -class _ParsedArgument: - value: str - next_index: int - - -def runtime_root() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - -def same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def _same_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - -def _quote_posix(value: str) -> str: - return "'" + value.replace("'", "'\"'\"'") + "'" - - -def _quote_powershell(value: str) -> str: - return "'" + value.replace("'", "''") + "'" - - -def _windows_powershell_path() -> str: - configured = os.environ.get("SystemRoot") if os.name == "nt" else None - system_root = ( - configured - if configured - and ntpath.isabs(configured) - and re.search(r'["%\r\n]', configured) is None - else r"C:\Windows" - ) - return ntpath.join( - system_root, - "System32", - "WindowsPowerShell", - "v1.0", - "powershell.exe", - ) - - -def _windows_command(script_path: str) -> str: - source = ( - f"& {_quote_powershell(sys.executable)} " - f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" - ) - encoded = base64.b64encode(source.encode("utf-16le")).decode("ascii") - return ( - f'"{_windows_powershell_path()}" -NoLogo -NoProfile ' - f"-NonInteractive -EncodedCommand {encoded}" - ) - - def build_handler(script_path: str, status_message: str) -> JsonObject: return { "type": "command", - "command": f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}", - "commandWindows": _windows_command(script_path), + "command": posix_source(script_path), + "commandWindows": encoded_windows_command(script_path), "timeout": HOOK_TIMEOUT_SECONDS, "statusMessage": status_message, } -def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - apostrophe = "'\"'\"'" - value = "" - index = start + 1 - while index < len(command): - if command.startswith(apostrophe, index): - value += "'" - index += len(apostrophe) - continue - if command[index] == "'": - return _ParsedArgument(value, index + 1) - value += command[index] - index += 1 - return None - - -def _parse_posix_command(command: Any) -> Optional[Tuple[str, str]]: - if not isinstance(command, str): - return None - executable = _read_posix_argument(command, 0) - if ( - executable is None - or command[executable.next_index : executable.next_index + 1] != " " - ): - return None - script = _read_posix_argument(command, executable.next_index + 1) - if script is None or script.next_index != len(command): - return None - return executable.value, script.value - - -def _read_powershell_argument(command: str, start: int) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - value = "" - index = start + 1 - while index < len(command): - if command[index] != "'": - value += command[index] - index += 1 - continue - if index + 1 < len(command) and command[index + 1] == "'": - value += "'" - index += 2 - continue - return _ParsedArgument(value, index + 1) - return None - - -def _parse_powershell_source(source: str) -> Optional[Tuple[str, str]]: - if not source.startswith("& "): - return None - executable = _read_powershell_argument(source, 2) - if ( - executable is None - or source[executable.next_index : executable.next_index + 1] != " " - ): - return None - script = _read_powershell_argument(source, executable.next_index + 1) - if script is None or source[script.next_index :] != "; exit $LASTEXITCODE": - return None - return executable.value, script.value - - -def _parse_windows_command(command: Any) -> Optional[Tuple[str, str]]: - if not isinstance(command, str): - return None - match = re.fullmatch( - r'"([^"\r\n]+)" -NoLogo -NoProfile -NonInteractive ' - r"-EncodedCommand ([A-Za-z0-9+/]+={0,2})", - command, - ) - if ( - match is None - or not ntpath.isabs(match.group(1)) - or ntpath.basename(match.group(1)).lower() != "powershell.exe" - ): - return None - try: - raw = base64.b64decode(match.group(2), validate=True) - if base64.b64encode(raw).decode("ascii") != match.group(2): - return None - source = raw.decode("utf-16le") - except (UnicodeDecodeError, ValueError): - return None - return _parse_powershell_source(source) - - def _parse_legacy_commands(handler: JsonObject) -> Optional[Tuple[str, str]]: command = handler.get("command") command_windows = handler.get("commandWindows") @@ -228,17 +86,6 @@ def _parse_legacy_commands(handler: JsonObject) -> Optional[Tuple[str, str]]: return arguments[0], arguments[1] -def _is_python_executable(file_path: str) -> bool: - return ( - re.fullmatch( - r"python(?:[0-9]+(?:\.[0-9]+)*)?(?:\.exe)?", - os.path.basename(file_path), - re.I, - ) - is not None - ) - - def _managed_script_path( handler: JsonObject, status_message: str, @@ -252,8 +99,14 @@ def _managed_script_path( or handler.get("statusMessage") != status_message ): return None - posix = _parse_posix_command(handler.get("command")) - windows = _parse_windows_command(handler.get("commandWindows")) + command = handler.get("command") + command_windows = handler.get("commandWindows") + posix = parse_posix_command(command) if isinstance(command, str) else None + windows = ( + parse_encoded_windows_command(command_windows) + if isinstance(command_windows, str) + else None + ) if posix is None or windows is None: legacy = _parse_legacy_commands(handler) posix = legacy @@ -263,8 +116,8 @@ def _managed_script_path( or windows is None or not os.path.isabs(posix[0]) or not os.path.isabs(posix[1]) - or not _is_python_executable(posix[0]) - or not _is_python_executable(windows[0]) + or not is_python_executable(posix[0]) + or not is_python_executable(windows[0]) or not same_path(posix[0], windows[0]) or not same_path(posix[1], windows[1]) ): @@ -278,13 +131,10 @@ def _managed_agent_id( status_message: str, ) -> Optional[str]: script_path = _managed_script_path(handler, status_message) - if script_path is None or os.path.basename(script_path) != script_name: + if script_path is None: return None - agent_directory = os.path.dirname(script_path) - if not same_path(os.path.dirname(agent_directory), runtime_root()): - return None - agent_id = os.path.basename(agent_directory) - return agent_id if agent_id not in {"", ".", ".."} else None + reference = managed_script_reference(script_path, script_name) + return None if reference is None else reference[0] def _read_hooks(value: Any, label: str) -> CodexHooks: @@ -339,7 +189,7 @@ def _remove_from_groups( for handler in group["hooks"]: managed_id = _managed_agent_id(handler, script_name, status_message) remove = managed_id is not None and ( - not agent_id or _same_agent_id(managed_id, agent_id) + not agent_id or same_agent_id(managed_id, agent_id) ) if not remove: handlers.append(handler) diff --git a/elydora/plugins/codex_io.py b/elydora/plugins/codex_io.py index 6d06e3a..3ad0ca7 100644 --- a/elydora/plugins/codex_io.py +++ b/elydora/plugins/codex_io.py @@ -4,17 +4,11 @@ from __future__ import annotations import os import stat -from typing import Any, List, Optional +from typing import List, Optional -from ._managed_files import ( - MAX_CONFIG_BYTES, - MAX_SOURCE_BYTES, - physical_directory_exists, - physical_file_exists, - read_physical_file, -) -from ._strict_json import JsonObject, parse_json_object -from ._transaction import FileChange, source_change, write_changes +from ._managed_files import MAX_SOURCE_BYTES, physical_directory_exists, read_physical_file +from ._runtime import rendered_source_change, runtime_present +from ._transaction import FileChange from .codex_contract import ( AGENT_KEY, CONFIG_FILE, @@ -26,12 +20,6 @@ from .codex_contract import ( ) -def _same_agent_id(value: Any, expected: str) -> bool: - return isinstance(value, str) and os.path.normcase(value) == os.path.normcase( - expected - ) - - def codex_home_path() -> str: configured = os.environ.get("CODEX_HOME") if configured is None or configured == "": @@ -64,130 +52,19 @@ def read_document() -> CodexDocument: def validate_hooks_directory(file_path: str) -> None: - directory = os.path.dirname(file_path) - physical_directory_exists(directory, "Codex hooks directory") + physical_directory_exists(os.path.dirname(file_path), "Codex hooks directory") def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: if not rendered.changed: return None - return source_change( + return rendered_source_change( rendered.document.file_path, "Codex user hooks", rendered.document.raw, rendered.next_source, - 0o600, - MAX_SOURCE_BYTES, ) -def runtime_change( - file_path: str, - label: str, - next_source: str, - mode: int, - maximum_bytes: int = MAX_SOURCE_BYTES, -) -> Optional[FileChange]: - snapshot = read_physical_file(file_path, label, maximum_bytes) - return source_change( - file_path, - label, - None if snapshot is None else snapshot.contents, - next_source, - mode, - maximum_bytes, - ) - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, - "Elydora runtime config", - MAX_CONFIG_BYTES, - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, - f"Elydora runtime config at {file_path}", - ) - - -def validate_runtime_identity(agent_directory: str, agent_id: str) -> bool: - config_path_value = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path_value) - artifacts = [ - ("private.key", "Elydora private key"), - ("guard.py", "Elydora guard runtime"), - ("hook.py", "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ] - artifact_states = [ - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ] - artifact_exists = any(artifact_states) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return False - if config.get("agent_name") != AGENT_KEY or not _same_agent_id( - config.get("agent_id"), - agent_id, - ): - raise ValueError( - f"Elydora runtime config identity does not match Codex agent {agent_id}: " - f"{config_path_value}" - ) - return True - - -def validate_runtime_tree( - runtime_root: str, agent_directory: str, agent_id: str -) -> None: - if not physical_directory_exists(runtime_root, "Elydora runtime directory"): - return - if not physical_directory_exists( - agent_directory, - "Elydora agent runtime directory", - ): - return - validate_runtime_identity(agent_directory, agent_id) - - def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: - for contract in contracts: - agent_directory = os.path.dirname(contract.guard_path) - if not physical_directory_exists( - agent_directory, - "Elydora agent runtime directory", - ): - continue - config = _read_runtime_config(os.path.join(agent_directory, "config.json")) - if ( - config is None - or config.get("agent_name") != AGENT_KEY - or not _same_agent_id(config.get("agent_id"), contract.agent_id) - ): - continue - if all( - ( - physical_file_exists(contract.guard_path, "Elydora guard runtime"), - physical_file_exists(contract.audit_path, "Elydora audit runtime"), - physical_file_exists( - os.path.join(agent_directory, "private.key"), - "Elydora private key", - ), - ) - ): - return True - return False - - -def write_codex_changes(changes: List[FileChange], label: str) -> None: - write_changes(changes, label) + return any(runtime_present(contract, AGENT_KEY) for contract in contracts) diff --git a/elydora/plugins/cursor.py b/elydora/plugins/cursor.py index fae03b2..4218024 100644 --- a/elydora/plugins/cursor.py +++ b/elydora/plugins/cursor.py @@ -2,79 +2,36 @@ from __future__ import annotations -import base64 -import json -import os import urllib.parse -from typing import List, Optional -from elydora._runtime_paths import resolve_agent_directory -from elydora.utils import base64url_encode - -from ._transaction import FileChange +from ._managed_files import physical_directory_exists +from ._runtime import ( + DEFAULT_BASE_URL, + present, + resolve_runtime_paths, + runtime_file_changes, + validate_private_key, + validate_runtime_tree, +) +from ._transaction import write_changes from .base import AgentPlugin, InstallConfig, PluginStatus from .cursor_contract import ( AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, build_handler, remove_managed_hooks, render_document, runtime_contracts, - runtime_root, - same_path, ) from .cursor_io import ( read_document, rendered_change, - physical_directory_exists, - physical_file_exists, - require_runtime_directory, - runtime_change, runtime_files_exist, validate_config_directory, - validate_runtime_identity, - write_cursor_changes, ) from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -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 _present(changes: List[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - -def _agent_paths(config: InstallConfig) -> tuple[str, str, str]: - 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) - configured_guard = config.get("guard_script_path", "") - if not same_path(configured_guard, guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return agent_id, agent_directory, guard_path +PRODUCT = "Cursor" def _validate_install_config(config: InstallConfig) -> None: @@ -84,27 +41,14 @@ def _validate_install_config(config: InstallConfig) -> None: raise ValueError(f"{field} is required") if config.get("agent_name") != AGENT_KEY: raise ValueError(f"Cursor installation requires agent_name {AGENT_KEY}") - private_key = config["private_key"] - try: - padded = private_key + "=" * ((4 - len(private_key) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), - validate=True, - ) - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - "private_key must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or base64url_encode(seed) != private_key: - raise ValueError("private_key must be a canonical 32-byte base64url value") - base_url = config.get("base_url", "https://api.elydora.com") + validate_private_key(config["private_key"]) + base_url = config.get("base_url", DEFAULT_BASE_URL) if not isinstance(base_url, str): raise ValueError("base_url must be a string") parsed = urllib.parse.urlsplit(base_url) if parsed.scheme not in ("http", "https") or not parsed.netloc: raise ValueError("base_url must be an absolute HTTP or HTTPS URL") - token = config.get("token", "") - if not isinstance(token, str): + if not isinstance(config.get("token", ""), str): raise ValueError("token must be a string") @@ -117,58 +61,32 @@ class CursorPlugin(AgentPlugin): _validate_install_config(config) read_document() validate_config_directory() - agent_id, agent_directory, guard_path = _agent_paths(config) - if not physical_directory_exists(runtime_root()): - return - if not physical_directory_exists(agent_directory): - return - identity_exists = validate_runtime_identity( - os.path.join(agent_directory, "config.json"), - agent_id, - ) - runtime_exists = False - for file_path, label in ( - (guard_path, "Elydora guard runtime"), - (os.path.join(agent_directory, AUDIT_SCRIPT), "Elydora audit runtime"), - (os.path.join(agent_directory, "private.key"), "Elydora private key"), - (os.path.join(agent_directory, "chain-state.json"), "Elydora chain state"), - (os.path.join(agent_directory, "status-cache.json"), "Elydora status cache"), - (os.path.join(agent_directory, "error.log"), "Elydora error log"), - ): - runtime_exists = physical_file_exists(file_path, label) or runtime_exists - if runtime_exists and not identity_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) + paths = resolve_runtime_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id, AGENT_KEY, PRODUCT) def install(self, config: InstallConfig) -> None: self.preflight_install(config) document = read_document() - agent_id, agent_directory, guard_path = _agent_paths(config) - require_runtime_directory(agent_directory) - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - - runtime_config_path = os.path.join(agent_directory, "config.json") - validate_runtime_identity(runtime_config_path, agent_id) + paths = resolve_runtime_paths(config) + if not physical_directory_exists( + paths.agent_directory, "Elydora agent runtime directory" + ): + raise FileNotFoundError( + f"Elydora agent runtime directory is missing: {paths.agent_directory}" + ) + validate_runtime_tree(paths.agent_directory, paths.agent_id, AGENT_KEY, PRODUCT) hooks = remove_managed_hooks(document.hooks) - hooks["preToolUse"] = [ - *hooks.get("preToolUse", []), - build_handler(guard_path), - ] - hooks["postToolUse"] = [ - *hooks.get("postToolUse", []), - build_handler(audit_path), - ] + hooks["preToolUse"] = [*hooks.get("preToolUse", []), build_handler(paths.guard_path)] + hooks["postToolUse"] = [*hooks.get("postToolUse", []), build_handler(paths.audit_path)] hooks["postToolUseFailure"] = [ *hooks.get("postToolUseFailure", []), - build_handler(audit_path), + build_handler(paths.audit_path), ] - hook_script = generate_hook_script( + audit_script = generate_hook_script( org_id=config.get("org_id", ""), - agent_id=agent_id, + agent_id=paths.agent_id, kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), + base_url=config.get("base_url", DEFAULT_BASE_URL), success_output="{}\n", fail_closed=True, native_payload=True, @@ -176,67 +94,34 @@ class CursorPlugin(AgentPlugin): ) guard_script = generate_guard_script( AGENT_KEY, - agent_id, + paths.agent_id, success_output='{"permission":"allow"}\n', fail_closed=True, deny_protocol="cursor", ) - changes = _present([ - runtime_change( - guard_path, - "Elydora guard runtime", - guard_script, - 0o700, - ), - runtime_change( - runtime_config_path, - "Elydora runtime config", - _json_source(_runtime_config(config, agent_id)), - 0o600, - ), - runtime_change( - os.path.join(agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - ), - runtime_change( - audit_path, - "Elydora audit runtime", - hook_script, - 0o700, - ), - rendered_change(render_document(document, hooks)), - ]) - write_cursor_changes(changes, "Install Cursor hooks") + changes = [ + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *present([rendered_change(render_document(document, hooks))]), + ] + write_changes(changes, "Install Cursor hooks") print(f" Cursor hooks: {document.file_path}") def uninstall(self, agent_id: str = "") -> None: document = read_document() hooks = remove_managed_hooks(document.hooks, agent_id) change = rendered_change(render_document(document, hooks)) - write_cursor_changes( - [] if change is None else [change], - "Uninstall Cursor hooks", - ) + if change is not None: + write_changes([change], "Uninstall Cursor hooks") def status(self) -> PluginStatus: document = read_document() contracts = runtime_contracts(document.hooks) if not contracts: - return PluginStatus( - installed=False, - agent=AGENT_KEY, - details="Not installed", - ) + return PluginStatus(installed=False, agent=AGENT_KEY, details="Not installed") installed = runtime_files_exist(contracts) details = ( f"Config: {document.file_path}" if installed else f"Configured at {document.file_path}; runtime files missing" ) - return PluginStatus( - installed=installed, - agent=AGENT_KEY, - details=details, - ) + return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) diff --git a/elydora/plugins/cursor_contract.py b/elydora/plugins/cursor_contract.py index 0f3557d..acb0d07 100644 --- a/elydora/plugins/cursor_contract.py +++ b/elydora/plugins/cursor_contract.py @@ -7,8 +7,17 @@ import json import os import re import sys -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, Dict, List, Optional +from elydora._runtime_paths import runtime_root + +from ._runtime import managed_script_reference, same_agent_id, same_path +from ._shell_command import ( + parse_posix_command, + parse_powershell_source, + posix_source, + powershell_source, +) from ._strict_json import JsonObject, parse_json_object @@ -43,42 +52,8 @@ class RuntimeContract: audit_path: str -@dataclass(frozen=True) -class _ParsedArgument: - value: str - next_index: int - - -def runtime_root() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - -def same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def _same_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - -def _quote_posix(value: str) -> str: - return "'" + value.replace("'", "'\"'\"'") + "'" - - -def _quote_powershell(value: str) -> str: - return "'" + value.replace("'", "''") + "'" - - def build_handler(script_path: str) -> JsonObject: - if os.name == "nt": - command = ( - f"& {_quote_powershell(sys.executable)} " - f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" - ) - else: - command = f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" + command = powershell_source(script_path) if os.name == "nt" else posix_source(script_path) return { "command": command, "timeout": HOOK_TIMEOUT_SECONDS, @@ -86,73 +61,6 @@ def build_handler(script_path: str) -> JsonObject: } -def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - apostrophe = "'\"'\"'" - value = "" - index = start + 1 - while index < len(command): - if command.startswith(apostrophe, index): - value += "'" - index += len(apostrophe) - continue - if command[index] == "'": - return _ParsedArgument(value, index + 1) - value += command[index] - index += 1 - return None - - -def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: - executable = _read_posix_argument(command, 0) - if ( - executable is None - or command[executable.next_index : executable.next_index + 1] != " " - ): - return None - script = _read_posix_argument(command, executable.next_index + 1) - if script is None or script.next_index != len(command): - return None - return executable.value, script.value - - -def _read_powershell_argument( - command: str, - start: int, -) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - value = "" - index = start + 1 - while index < len(command): - if command[index] != "'": - value += command[index] - index += 1 - continue - if index + 1 < len(command) and command[index + 1] == "'": - value += "'" - index += 2 - continue - return _ParsedArgument(value, index + 1) - return None - - -def _parse_powershell_command(command: str) -> Optional[Tuple[str, str]]: - if not command.startswith("& "): - return None - executable = _read_powershell_argument(command, 2) - if ( - executable is None - or command[executable.next_index : executable.next_index + 1] != " " - ): - 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 _legacy_script_path(command: str) -> Optional[str]: quoted = re.fullmatch(r'"([^"\r\n]+)" ([^\r\n]+)', command) if quoted and same_path(quoted.group(1), sys.executable): @@ -169,9 +77,9 @@ def _managed_script_path(handler: JsonObject) -> Optional[str]: and handler.get("failClosed") is True ): parsed = ( - _parse_powershell_command(handler["command"]) + parse_powershell_source(handler["command"]) if os.name == "nt" - else _parse_posix_command(handler["command"]) + else parse_posix_command(handler["command"]) ) if ( parsed is not None @@ -185,18 +93,12 @@ def _managed_script_path(handler: JsonObject) -> Optional[str]: return _legacy_script_path(handler["command"]) -def _managed_agent_id( - handler: JsonObject, - script_name: str, -) -> Optional[str]: +def _managed_agent_id(handler: JsonObject, script_name: str) -> Optional[str]: script_path = _managed_script_path(handler) - if script_path is None or os.path.basename(script_path) != script_name: + if script_path is None: return None - agent_directory = os.path.dirname(script_path) - if not same_path(os.path.dirname(agent_directory), runtime_root()): - return None - agent_id = os.path.basename(agent_directory) - return agent_id if agent_id not in {"", ".", ".."} else None + reference = managed_script_reference(script_path, script_name) + return None if reference is None else reference[0] def _read_hooks(value: Any, label: str) -> CursorHooks: @@ -256,7 +158,7 @@ def remove_managed_hooks( for handler in result.get(event, []): managed_id = _managed_agent_id(handler, script_name) remove = managed_id is not None and ( - not agent_id or _same_agent_id(managed_id, agent_id) + not agent_id or same_agent_id(managed_id, agent_id) ) if not remove: handlers.append(handler) diff --git a/elydora/plugins/cursor_io.py b/elydora/plugins/cursor_io.py index 78c82ac..8902d84 100644 --- a/elydora/plugins/cursor_io.py +++ b/elydora/plugins/cursor_io.py @@ -3,19 +3,18 @@ from __future__ import annotations import os -import stat -from typing import Any, List, Optional +from typing import List, Optional -from ._transaction import FileChange, source_change, write_changes +from ._managed_files import MAX_SOURCE_BYTES, physical_directory_exists, read_physical_file +from ._runtime import rendered_source_change, runtime_present +from ._transaction import FileChange from .cursor_contract import ( AGENT_KEY, CursorDocument, - JsonObject, RenderedDocument, RuntimeContract, create_document, parse_document, - parse_json_object, ) @@ -26,180 +25,28 @@ def config_path() -> str: return os.path.join(os.path.expanduser("~"), ".cursor", CONFIG_FILE) -def _read_optional_physical(file_path: str, label: str) -> Optional[str]: - try: - metadata = os.lstat(file_path) - except FileNotFoundError: - return None - except OSError as error: - raise OSError(f"Inspect {label} at {file_path}: {error}") from error - if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): - raise OSError(f"{label} path is not a physical file: {file_path}") - try: - with open(file_path, "r", encoding="utf-8", newline="") as file: - return file.read() - except OSError as error: - raise OSError(f"Read {label} at {file_path}: {error}") from error - - def read_document() -> CursorDocument: file_path = config_path() - raw = _read_optional_physical(file_path, "Cursor user hooks") - return create_document(file_path) if raw is None else parse_document(file_path, raw) + snapshot = read_physical_file(file_path, "Cursor user hooks", MAX_SOURCE_BYTES) + if snapshot is None: + return create_document(file_path) + return parse_document(file_path, snapshot.contents) def validate_config_directory() -> None: - directory = os.path.dirname(config_path()) - try: - metadata = os.lstat(directory) - except FileNotFoundError: - return - except OSError as error: - raise OSError(f"Inspect Cursor hooks directory at {directory}: {error}") from error - if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): - raise OSError(f"Cursor hooks directory is not a physical directory: {directory}") + physical_directory_exists(os.path.dirname(config_path()), "Cursor hooks directory") def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: if not rendered.changed: return None - return source_change( + return rendered_source_change( rendered.document.file_path, "Cursor user hooks", rendered.document.raw, rendered.next_source, - 0o600, - ) - - -def runtime_change( - file_path: str, - label: str, - next_source: str, - mode: int, -) -> Optional[FileChange]: - original = _read_optional_physical(file_path, label) - return source_change(file_path, label, original, next_source, mode) - - -def physical_file_exists(file_path: str, label: str) -> bool: - try: - metadata = os.lstat(file_path) - except FileNotFoundError: - return False - except OSError as error: - raise OSError(f"Inspect {label} at {file_path}: {error}") from error - if not stat.S_ISREG(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): - raise OSError(f"{label} path is not a physical file: {file_path}") - return True - - -def require_runtime(file_path: str, label: str) -> None: - if not file_path: - raise ValueError(f"{label} path is required") - if not physical_file_exists(file_path, label): - raise FileNotFoundError(f"{label} is missing: {file_path}") - - -def physical_directory_exists(directory: str) -> bool: - try: - metadata = os.lstat(directory) - except FileNotFoundError: - return False - except OSError as error: - raise OSError( - f"Inspect Elydora agent runtime directory at {directory}: {error}" - ) from error - if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): - raise OSError( - f"Elydora agent runtime path is not a physical directory: {directory}" - ) - return True - - -def require_runtime_directory(directory: str) -> None: - if not physical_directory_exists(directory): - raise FileNotFoundError( - f"Elydora agent runtime directory is missing: {directory}" - ) - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - raw = _read_optional_physical(file_path, "Elydora runtime config") - if raw is None: - return None - return parse_json_object(raw, f"Elydora runtime config at {file_path}") - - -def validate_runtime_identity( - file_path: str, - agent_id: str, -) -> bool: - config = _read_runtime_config(file_path) - if config is None: - return False - if config.get("agent_name") != AGENT_KEY or not _same_agent_id( - config.get("agent_id"), agent_id - ): - raise ValueError( - f"Elydora runtime config identity does not match Cursor agent {agent_id}: " - f"{file_path}" - ) - return True - - -def _same_agent_id(value: Any, expected: str) -> bool: - return isinstance(value, str) and os.path.normcase(value) == os.path.normcase( - expected ) def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: - for contract in contracts: - agent_directory = os.path.dirname(contract.guard_path) - if not physical_directory_exists(agent_directory): - continue - config_path_value = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path_value) - if ( - config is None - or config.get("agent_name") != AGENT_KEY - or not _same_agent_id(config.get("agent_id"), contract.agent_id) - ): - continue - if physical_file_exists( - contract.guard_path, - "Elydora guard runtime", - ) and physical_file_exists( - contract.audit_path, - "Elydora audit runtime", - ) and physical_file_exists( - os.path.join(agent_directory, "private.key"), - "Elydora private key", - ): - return True - return False - - -def _ensure_physical_directory(directory: str, label: str) -> None: - try: - os.makedirs(directory, mode=0o700, exist_ok=True) - metadata = os.lstat(directory) - except OSError as error: - raise OSError(f"Prepare {label} directory at {directory}: {error}") from error - if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode): - raise OSError(f"{label} directory is not a physical directory: {directory}") - - -def write_cursor_changes(changes: List[FileChange], label: str) -> None: - effective = [change for change in changes if change.original != change.next_source] - if not effective: - return - directories = { - os.path.normcase(os.path.abspath(os.path.dirname(change.file_path))): - os.path.dirname(change.file_path) - for change in effective - } - for directory in directories.values(): - _ensure_physical_directory(directory, label) - write_changes(effective, label) + return any(runtime_present(contract, AGENT_KEY) for contract in contracts) diff --git a/elydora/plugins/gemini_command.py b/elydora/plugins/gemini_command.py index c98097c..20dc044 100644 --- a/elydora/plugins/gemini_command.py +++ b/elydora/plugins/gemini_command.py @@ -2,14 +2,28 @@ from __future__ import annotations -import base64 from dataclasses import dataclass -import ntpath import os import re import sys from typing import Optional, Tuple +from ._runtime import managed_script_reference, same_path +from ._shell_command import ( + decode_powershell_source, + encode_powershell_source, + is_powershell_executable, + parse_posix_command, + parse_powershell_source, + posix_source, + powershell_source, + quote_powershell, + read_powershell_argument, + windows_powershell_path, +) + +_ENCODED_FLAGS = " -NoLogo -NoProfile -NonInteractive -EncodedCommand " + @dataclass(frozen=True) class GeminiRuntimeReference: @@ -17,163 +31,33 @@ class GeminiRuntimeReference: script_path: str -def same_gemini_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_gemini_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - -def _quote_posix(value: str) -> str: - return "'" + value.replace("'", "'\"'\"'") + "'" - - -def _quote_powershell(value: str) -> str: - return "'" + value.replace("'", "''") + "'" - - -def _windows_powershell_path() -> str: - configured = os.environ.get("SystemRoot") - system_root = ( - configured - if configured - and ntpath.isabs(configured) - and re.search(r'["%\r\n]', configured) is None - else r"C:\Windows" - ) - return ntpath.join( - system_root, - "System32", - "WindowsPowerShell", - "v1.0", - "powershell.exe", - ) - - -def _windows_command(script_path: str) -> str: - source = ( - f"& {_quote_powershell(sys.executable)} " - f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" - ) - encoded = base64.b64encode(source.encode("utf-16le")).decode("ascii") - return ( - f"& {_quote_powershell(_windows_powershell_path())} " - "-NoLogo -NoProfile -NonInteractive -EncodedCommand " - f"{encoded}" - ) - - def build_gemini_command(script_path: str) -> str: if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): raise ValueError( "Gemini CLI hook commands require absolute executable and script paths" ) if os.name == "nt": - return _windows_command(script_path) - return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" - - -def _read_posix_argument(command: str, start: int) -> Optional[Tuple[str, int]]: - if start >= len(command) or command[start] != "'": - return None - apostrophe = "'\"'\"'" - value = "" - index = start + 1 - while index < len(command): - if command.startswith(apostrophe, index): - value += "'" - index += len(apostrophe) - continue - if command[index] == "'": - return value, index + 1 - value += command[index] - index += 1 - return None - - -def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: - executable = _read_posix_argument(command, 0) - if executable is None or executable[1] >= len(command): - return None - if command[executable[1]] != " ": - return None - script = _read_posix_argument(command, executable[1] + 1) - if script is None or script[1] != len(command): - return None - return executable[0], script[0] - - -def _read_powershell_argument( - command: str, start: int -) -> Optional[Tuple[str, int]]: - if start >= len(command) or command[start] != "'": - return None - value = "" - index = start + 1 - while index < len(command): - if command[index] != "'": - value += command[index] - index += 1 - continue - if index + 1 < len(command) and command[index + 1] == "'": - value += "'" - index += 2 - continue - return value, index + 1 - return None - - -def _parse_powershell_source(source: str) -> Optional[Tuple[str, str]]: - if not source.startswith("& "): - return None - executable = _read_powershell_argument(source, 2) - if executable is None or executable[1] >= len(source): - return None - if source[executable[1]] != " ": - return None - script = _read_powershell_argument(source, executable[1] + 1) - if script is None or source[script[1] :] != "; exit $LASTEXITCODE": - return None - return executable[0], script[0] + encoded = encode_powershell_source(powershell_source(script_path)) + return f"& {quote_powershell(windows_powershell_path())}{_ENCODED_FLAGS}{encoded}" + return posix_source(script_path) def _parse_windows_command(command: str) -> Optional[Tuple[str, str]]: if not command.startswith("& "): return None - powershell = _read_powershell_argument(command, 2) - if ( - powershell is None - or not ntpath.isabs(powershell[0]) - or ntpath.basename(powershell[0]).lower() != "powershell.exe" - ): + powershell = read_powershell_argument(command, 2) + if powershell is None or not is_powershell_executable(powershell[0]): return None - prefix = " -NoLogo -NoProfile -NonInteractive -EncodedCommand " - if not command.startswith(prefix, powershell[1]): + if not command.startswith(_ENCODED_FLAGS, powershell[1]): return None - encoded = command[powershell[1] + len(prefix) :] - if re.fullmatch(r"[A-Za-z0-9+/]+={0,2}", encoded) is None: - return None - try: - payload = base64.b64decode(encoded, validate=True) - if base64.b64encode(payload).decode("ascii") != encoded: - return None - if len(payload) % 2: - return None - source = payload.decode("utf-16le") - except (ValueError, UnicodeDecodeError): - return None - return _parse_powershell_source(source) + source = decode_powershell_source(command[powershell[1] + len(_ENCODED_FLAGS) :]) + return None if source is None else parse_powershell_source(source) -def _parse_legacy_command( - command: str, script_name: str -) -> Optional[Tuple[str, str]]: +def _parse_legacy_command(command: str, script_name: str) -> Optional[Tuple[str, str]]: if script_name == "guard.py": match = re.fullmatch(r'"([^"\r\n]+)" ([^\r\n]+)', command) - if match is None or not same_gemini_path(match.group(1), sys.executable): + if match is None or not same_path(match.group(1), sys.executable): return None return match.group(1), match.group(2) if "\r" in command or "\n" in command: @@ -186,21 +70,10 @@ def gemini_runtime_reference( script_name: str, include_legacy: bool = False, ) -> Optional[GeminiRuntimeReference]: - parsed = _parse_posix_command(command) or _parse_windows_command(command) + parsed = parse_posix_command(command) or _parse_windows_command(command) if parsed is None and include_legacy: parsed = _parse_legacy_command(command, script_name) - if ( - parsed is None - or not same_gemini_path(parsed[0], sys.executable) - or not os.path.isabs(parsed[1]) - or os.path.basename(parsed[1]) != script_name - ): + if parsed is None or not same_path(parsed[0], sys.executable): return None - agent_directory = os.path.dirname(parsed[1]) - runtime_root = os.path.join(os.path.expanduser("~"), ".elydora") - if not same_gemini_path(os.path.dirname(agent_directory), runtime_root): - return None - agent_id = os.path.basename(agent_directory) - if agent_id in {"", ".", ".."}: - return None - return GeminiRuntimeReference(agent_id, parsed[1]) + reference = managed_script_reference(parsed[1], script_name) + return None if reference is None else GeminiRuntimeReference(*reference) diff --git a/elydora/plugins/gemini_contract.py b/elydora/plugins/gemini_contract.py index caac42a..e4b728d 100644 --- a/elydora/plugins/gemini_contract.py +++ b/elydora/plugins/gemini_contract.py @@ -7,11 +7,11 @@ import math import os from typing import Any, Dict, List, Optional, Tuple +from ._runtime import same_agent_id from .gemini_command import ( GeminiRuntimeReference, build_gemini_command, gemini_runtime_reference, - same_gemini_agent_id, ) @@ -63,10 +63,6 @@ class ManagedRemoval: remove_group: bool -def elydora_dir() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - def _optional_string(value: JsonObject, field: str, label: str) -> None: if field in value and not isinstance(value[field], str): raise ValueError(f'{label} field "{field}" must be a string') @@ -286,7 +282,7 @@ def managed_gemini_removals( is not None and ( agent_id is None - or same_gemini_agent_id(reference.agent_id, agent_id) + or same_agent_id(reference.agent_id, agent_id) ) ) if indexes: diff --git a/elydora/plugins/gemini_installation.py b/elydora/plugins/gemini_installation.py index f3a58e6..c62d38b 100644 --- a/elydora/plugins/gemini_installation.py +++ b/elydora/plugins/gemini_installation.py @@ -2,160 +2,58 @@ from __future__ import annotations -from dataclasses import dataclass -import json -import os -from typing import List, Optional +from typing import List -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import FileChange, file_change, write_changes -from .base import InstallConfig -from .gemini_command import same_gemini_path -from .gemini_config import GeminiDocument, RenderedGeminiDocument -from .gemini_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - elydora_dir, -) -from .gemini_io import ( - rendered_change, - validate_api_origin, - validate_private_key, +from ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + present, + resolve_runtime_paths, + runtime_file_changes, + validate_install_config, validate_runtime_tree, ) +from ._transaction import FileChange, write_changes +from .base import InstallConfig +from .gemini_config import GeminiDocument, RenderedGeminiDocument +from .gemini_contract import AGENT_KEY +from .gemini_io import rendered_change from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -@dataclass(frozen=True) -class GeminiRuntimePaths: - agent_id: str - agent_directory: str - guard_path: str - audit_path: str - - -def _runtime_config(config: InstallConfig, agent_id: str) -> dict: - value = { - "org_id": config.get("org_id", ""), - "agent_id": agent_id, - "kid": config.get("kid", ""), - "base_url": config.get("base_url", "https://api.elydora.com"), - "agent_name": AGENT_KEY, - } - token = config.get("token") - if token: - value["token"] = token - return value - - -def _json_source(value: dict) -> str: - return json.dumps(value, indent=2) + "\n" - - -def _agent_paths(config: InstallConfig) -> GeminiRuntimePaths: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - agent_directory = resolve_agent_directory(elydora_dir(), agent_id) - guard_path = os.path.join(agent_directory, GUARD_SCRIPT) - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - if not same_gemini_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return GeminiRuntimePaths(agent_id, agent_directory, guard_path, audit_path) - - -def _validate_install_config(config: InstallConfig) -> None: - for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): - value = config.get(field) - if not isinstance(value, str) or not value: - raise ValueError(f"{field} is required") - if config.get("agent_name") != AGENT_KEY: - raise ValueError( - f"Gemini CLI installation requires agent_name {AGENT_KEY}" - ) - validate_private_key(config["private_key"]) - validate_api_origin(config["base_url"]) - if "token" in config: - token = config["token"] - if not isinstance(token, str) or not token: - raise ValueError("token must be a non-empty string when provided") - _agent_paths(config) +PRODUCT = "Gemini CLI" def preflight_gemini_installation( config: InstallConfig, document: GeminiDocument -) -> GeminiRuntimePaths: +) -> RuntimePaths: if not document.file_path: raise ValueError("Gemini CLI installation requires a settings path") - _validate_install_config(config) - paths = _agent_paths(config) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + validate_install_config(config, AGENT_KEY, PRODUCT) + paths = resolve_runtime_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id, AGENT_KEY, PRODUCT) return paths -def _runtime_change( - file_path: str, - label: str, - next_source: str, - mode: int, - maximum_bytes: int = MAX_SOURCE_BYTES, -) -> Optional[FileChange]: - return file_change(file_path, label, next_source, mode, maximum_bytes) - - -def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - def prepare_gemini_installation( config: InstallConfig, - paths: GeminiRuntimePaths, + paths: RuntimePaths, rendered: RenderedGeminiDocument, ) -> List[FileChange]: - guard_script = generate_guard_script( - AGENT_KEY, - paths.agent_id, - success_output="{}\n", - ) + guard_script = generate_guard_script(AGENT_KEY, paths.agent_id, success_output="{}\n") audit_script = generate_hook_script( org_id=config.get("org_id", ""), agent_id=paths.agent_id, kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), + base_url=config.get("base_url", DEFAULT_BASE_URL), success_output="{}\n", native_payload=True, agent_name=AGENT_KEY, ) - return _present([ - _runtime_change( - paths.guard_path, "Elydora guard runtime", guard_script, 0o700 - ), - _runtime_change( - os.path.join(paths.agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, paths.agent_id)), - 0o600, - MAX_CONFIG_BYTES, - ), - _runtime_change( - os.path.join(paths.agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - MAX_SECRET_BYTES, - ), - _runtime_change( - paths.audit_path, "Elydora audit runtime", audit_script, 0o700 - ), - rendered_change(rendered), - ]) + return [ + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *present([rendered_change(rendered)]), + ] def commit_gemini_installation(changes: List[FileChange]) -> None: diff --git a/elydora/plugins/gemini_io.py b/elydora/plugins/gemini_io.py index 6ba5b51..68091ab 100644 --- a/elydora/plugins/gemini_io.py +++ b/elydora/plugins/gemini_io.py @@ -2,34 +2,19 @@ from __future__ import annotations -import base64 import os -from typing import Any, List, Optional -import urllib.parse +from typing import List, Optional -from ._managed_files import ( - MAX_CONFIG_BYTES, - MAX_SECRET_BYTES, - MAX_SOURCE_BYTES, - physical_directory_exists, - physical_file_exists, - read_physical_file, -) -from ._strict_json import JsonObject, parse_json_object -from ._transaction import FileChange, source_change, write_changes -from .gemini_command import same_gemini_agent_id, same_gemini_path +from ._managed_files import MAX_SOURCE_BYTES, physical_directory_exists, read_physical_file +from ._runtime import rendered_source_change, runtime_contract_exists +from ._transaction import FileChange, write_changes from .gemini_config import ( GeminiDocument, RenderedGeminiDocument, create_gemini_document, parse_gemini_document, ) -from .gemini_contract import ( - AGENT_KEY, - CONFIG_FILE, - GeminiRuntimeContract, - elydora_dir, -) +from .gemini_contract import AGENT_KEY, CONFIG_FILE, GeminiRuntimeContract def gemini_configuration_directory() -> str: @@ -44,33 +29,21 @@ def gemini_settings_path() -> str: def read_gemini_document() -> GeminiDocument: file_path = gemini_settings_path() - physical_directory_exists( - os.path.dirname(file_path), "Gemini CLI configuration directory" - ) - snapshot = read_physical_file( - file_path, "Gemini CLI user settings", MAX_SOURCE_BYTES - ) + physical_directory_exists(os.path.dirname(file_path), "Gemini CLI configuration directory") + snapshot = read_physical_file(file_path, "Gemini CLI user settings", MAX_SOURCE_BYTES) if snapshot is None: return create_gemini_document(file_path) - return parse_gemini_document( - exists=True, - file_path=file_path, - raw=snapshot.contents, - ) + return parse_gemini_document(exists=True, file_path=file_path, raw=snapshot.contents) -def rendered_change( - rendered: RenderedGeminiDocument, -) -> Optional[FileChange]: +def rendered_change(rendered: RenderedGeminiDocument) -> Optional[FileChange]: if not rendered.changed: return None - return source_change( + return rendered_source_change( rendered.document.file_path, "Gemini CLI user settings", rendered.document.raw if rendered.document.exists else None, rendered.next_source, - 0o600, - MAX_SOURCE_BYTES, ) @@ -80,171 +53,7 @@ def write_gemini_document(rendered: RenderedGeminiDocument) -> None: write_changes([change], "Uninstall Gemini CLI hooks") -def validate_api_origin(value: str, label: str = "base_url") -> None: - try: - parsed = urllib.parse.urlsplit(value) - hostname = parsed.hostname - parsed.port - except ValueError as error: - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error - invalid_character = "\\" in value or any( - character.isspace() or ord(character) < 32 for character in value +def gemini_runtime_files_exist(contracts: List[GeminiRuntimeContract]) -> bool: + return any( + runtime_contract_exists(contract, AGENT_KEY, "Gemini CLI") for contract in contracts ) - if ( - parsed.scheme not in ("http", "https") - or not parsed.netloc - or hostname is None - or invalid_character - ): - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") - if ( - parsed.username is not None - or parsed.password is not None - or parsed.query - or parsed.fragment - ): - raise ValueError( - f"{label} must exclude credentials, query parameters, and fragments" - ) - - -def validate_private_key(value: str, label: str = "private_key") -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), validate=True - ) - canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - f"{label} must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or canonical != value: - raise ValueError(f"{label} must be a canonical 32-byte base64url value") - - -def _require_non_empty_string( - value: Any, field: str, config_path: str -) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError( - f"Elydora runtime config {field} is invalid: {config_path}" - ) - return value - - -def _validate_runtime_config( - config: JsonObject, expected_agent_id: str, config_path: str -) -> None: - supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} - extra = next((key for key in config if key not in supported), None) - if extra is not None: - raise ValueError( - f'Elydora runtime config has unsupported field "{extra}": {config_path}' - ) - _require_non_empty_string(config.get("org_id"), "org_id", config_path) - _require_non_empty_string(config.get("kid"), "kid", config_path) - agent_id = _require_non_empty_string( - config.get("agent_id"), "agent_id", config_path - ) - if ( - not same_gemini_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime identity does not match Gemini CLI hooks: " - f"{config_path}" - ) - if "token" in config: - _require_non_empty_string(config.get("token"), "token", config_path) - base_url = _require_non_empty_string( - config.get("base_url"), "base_url", config_path - ) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, "Elydora runtime config", MAX_CONFIG_BYTES - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, f"Elydora runtime config at {file_path}" - ) - - -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = elydora_dir() - if not physical_directory_exists(root, "Elydora runtime directory"): - return - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return - config_path = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path) - artifacts = ( - ("private.key", "Elydora private key"), - ("guard.py", "Elydora guard runtime"), - ("hook.py", "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ) - artifact_exists = any( - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not same_gemini_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Gemini CLI agent " - f"{agent_id}: {config_path}" - ) - - -def _runtime_contract_exists(contract: GeminiRuntimeContract) -> bool: - root = elydora_dir() - agent_directory = os.path.dirname(contract.guard_path) - if ( - not same_gemini_path(os.path.dirname(agent_directory), root) - or not same_gemini_path( - contract.audit_path, os.path.join(agent_directory, "hook.py") - ) - ): - return False - if not physical_directory_exists(root, "Elydora runtime directory"): - return False - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return False - config_path = os.path.join(agent_directory, "config.json") - key_path = os.path.join(agent_directory, "private.key") - config = _read_runtime_config(config_path) - key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) - guard = read_physical_file(contract.guard_path, "Elydora guard runtime") - audit = read_physical_file(contract.audit_path, "Elydora audit runtime") - if config is None or key is None or guard is None or audit is None: - return False - _validate_runtime_config(config, contract.agent_id, config_path) - validate_private_key(key.contents, "Elydora private key") - return bool(guard.contents) and bool(audit.contents) - - -def gemini_runtime_files_exist( - contracts: List[GeminiRuntimeContract], -) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts)