From 37e8289510cb4ade7bef60d62ab4776527e3e01f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Wed, 2 Sep 2026 23:45:01 -0400 Subject: [PATCH] chore(python): audit providers copilot droid augment cline Move the four providers onto _runtime and _shell_command; add expected_runtime_scripts and extra artifact checks to _runtime. Claude-Session: https://claude.ai/code/session_01Agoc9bRVE9KL7mmWtey9Vq --- elydora/plugins/_runtime.py | 35 +++- elydora/plugins/augment_contract.py | 25 +-- elydora/plugins/augment_installation.py | 167 +++++------------ elydora/plugins/augment_io.py | 227 ++--------------------- elydora/plugins/cline_contract.py | 22 +-- elydora/plugins/cline_installation.py | 177 ++++-------------- elydora/plugins/cline_io.py | 227 +---------------------- elydora/plugins/copilot_contract.py | 128 +++---------- elydora/plugins/copilot_installation.py | 198 ++++---------------- elydora/plugins/copilot_io.py | 206 ++------------------- elydora/plugins/droid_config.py | 2 +- elydora/plugins/droid_contract.py | 133 ++------------ elydora/plugins/droid_installation.py | 230 +++++------------------- elydora/plugins/droid_io.py | 225 ++--------------------- 14 files changed, 291 insertions(+), 1711 deletions(-) diff --git a/elydora/plugins/_runtime.py b/elydora/plugins/_runtime.py index 4dba90e..74afd19 100644 --- a/elydora/plugins/_runtime.py +++ b/elydora/plugins/_runtime.py @@ -22,6 +22,8 @@ from ._managed_files import ( from ._strict_json import JsonObject, parse_json_object from ._transaction import FileChange, file_change, source_change from .base import InstallConfig +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script GUARD_SCRIPT = "guard.py" @@ -215,7 +217,11 @@ def validate_install_config( def validate_runtime_tree( - agent_directory: str, agent_id: str, agent_key: str, product_label: str + agent_directory: str, + agent_id: str, + agent_key: str, + product_label: str, + extra_artifacts: Tuple[Tuple[str, str], ...] = (), ) -> None: """Reject an existing agent directory that belongs to another identity.""" if not physical_directory_exists(runtime_root(), "Elydora runtime directory"): @@ -226,7 +232,7 @@ def validate_runtime_tree( config = read_runtime_config(config_path) artifact_states = [ physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in RUNTIME_ARTIFACTS + for name, label in (*RUNTIME_ARTIFACTS, *extra_artifacts) ] artifact_exists = any(artifact_states) if config is None: @@ -246,10 +252,27 @@ def validate_runtime_tree( ) +def expected_runtime_scripts(agent_key: str, agent_id: str) -> Tuple[str, str]: + return ( + generate_guard_script(agent_key, agent_id), + generate_hook_script( + org_id="", + agent_id=agent_id, + kid="", + base_url="", + native_payload=True, + agent_name=agent_key, + ), + ) + + def runtime_contract_exists( - contract: RuntimeContractLike, agent_key: str, product_label: str + contract: RuntimeContractLike, + agent_key: str, + product_label: str, + expected_scripts: Optional[Tuple[str, str]] = None, ) -> bool: - """Strict status check: valid config and key plus non-empty runtime scripts.""" + """Strict status check: valid config and key plus 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( @@ -271,7 +294,9 @@ def runtime_contract_exists( 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) + if expected_scripts is None: + return bool(guard.contents) and bool(audit.contents) + return guard.contents == expected_scripts[0] and audit.contents == expected_scripts[1] def runtime_present(contract: RuntimeContractLike, agent_key: str) -> bool: diff --git a/elydora/plugins/augment_contract.py b/elydora/plugins/augment_contract.py index a40a621..578f5e6 100644 --- a/elydora/plugins/augment_contract.py +++ b/elydora/plugins/augment_contract.py @@ -9,6 +9,9 @@ import os import sys from typing import Any, Dict, List, Optional, Set, Tuple +from elydora._runtime_paths import runtime_root + +from ._runtime import same_agent_id, same_path from ._strict_json import parse_json_object @@ -69,10 +72,6 @@ def home_dir() -> str: return os.path.expanduser("~") -def elydora_dir() -> str: - return os.path.join(home_dir(), ".elydora") - - def resolve_config_path() -> str: return os.path.join(home_dir(), ".augment", "settings.json") @@ -162,16 +161,6 @@ def _parse_wrapper_command(command: str) -> Optional[str]: return parser(command) -def _same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def _same_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - def _managed_agent_id(handler: JsonObject, wrapper_name: str) -> Optional[str]: if ( handler.get("type") != "command" @@ -188,7 +177,7 @@ def _managed_agent_id(handler: JsonObject, wrapper_name: str) -> Optional[str]: ): return None agent_directory = os.path.dirname(wrapper_path) - if not _same_path(os.path.dirname(agent_directory), elydora_dir()): + 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 @@ -324,7 +313,7 @@ def _remove_managed( for handler in group["hooks"]: managed_id = _managed_agent_id(handler, wrapper_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 remove: changed = True @@ -376,12 +365,12 @@ def runtime_contracts(hooks: AugmentHooks) -> List[RuntimeContract]: contracts: List[RuntimeContract] = [] for guard_id in sorted(guards, key=os.path.normcase): audit_id = next( - (candidate for candidate in audits if _same_agent_id(candidate, guard_id)), + (candidate for candidate in audits if same_agent_id(candidate, guard_id)), None, ) if audit_id is None: continue - agent_directory = os.path.join(elydora_dir(), guard_id) + agent_directory = os.path.join(runtime_root(), guard_id) contracts.append( RuntimeContract( agent_id=guard_id, diff --git a/elydora/plugins/augment_installation.py b/elydora/plugins/augment_installation.py index 86c6c10..79f9b46 100644 --- a/elydora/plugins/augment_installation.py +++ b/elydora/plugins/augment_installation.py @@ -3,107 +3,61 @@ 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 ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + present, + resolve_runtime_paths, + runtime_file_changes, + validate_install_config, + validate_runtime_tree, +) from ._transaction import FileChange, file_change, write_changes from .augment_contract import ( AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, + AUDIT_WRAPPER, + GUARD_WRAPPER, AugmentDocument, RenderedAugmentDocument, build_wrapper, - elydora_dir, wrapper_paths, ) -from .augment_io import ( - rendered_change, - validate_api_origin, - validate_private_key, - validate_runtime_tree, -) +from .augment_io import rendered_change from .base import InstallConfig from .guard_template import generate_guard_script from .hook_template import generate_hook_script +WRAPPER_ARTIFACTS = ( + (GUARD_WRAPPER, "Auggie guard wrapper"), + (AUDIT_WRAPPER, "Auggie audit wrapper"), +) + @dataclass(frozen=True) -class AugmentRuntimePaths: - agent_id: str - agent_directory: str - guard_path: str - audit_path: str +class AugmentRuntimePaths(RuntimePaths): guard_wrapper_path: str audit_wrapper_path: 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 _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) -> AugmentRuntimePaths: - 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_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - f"Elydora guard runtime must use the managed agent directory: {guard_path}" - ) - wrappers = wrapper_paths(agent_directory) + paths = resolve_runtime_paths(config) + wrappers = wrapper_paths(paths.agent_directory) return AugmentRuntimePaths( - agent_id, - agent_directory, - guard_path, - audit_path, + paths.agent_id, + paths.agent_directory, + paths.guard_path, + paths.audit_path, wrappers.guard_path, wrappers.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"Augment Code CLI installation requires agent_name {AGENT_KEY}" - ) - validate_private_key(config["private_key"]) - validate_api_origin(config["base_url"]) - if "token" in config: - token = config["token"] - if not isinstance(token, str) or not token: - raise ValueError("token must be a non-empty string when provided") - _agent_paths(config) +def _validate_runtime_tree(paths: AugmentRuntimePaths) -> None: + validate_runtime_tree( + paths.agent_directory, paths.agent_id, AGENT_KEY, "Auggie", WRAPPER_ARTIFACTS + ) def preflight_augment_installation( @@ -111,26 +65,12 @@ def preflight_augment_installation( ) -> AugmentRuntimePaths: if not document.config_path: raise ValueError("Augment Code CLI installation requires a settings path") - _validate_install_config(config) + validate_install_config(config, AGENT_KEY, "Augment Code CLI") paths = _agent_paths(config) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + _validate_runtime_tree(paths) 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_augment_installation( config: InstallConfig, paths: AugmentRuntimePaths, @@ -138,54 +78,35 @@ def prepare_augment_installation( ) -> List[FileChange]: if not rendered.changed and rendered.document.raw is None: raise ValueError("Auggie hook installation did not produce a settings document") - validate_runtime_tree(paths.agent_directory, paths.agent_id) + _validate_runtime_tree(paths) guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) audit_script = generate_hook_script( org_id=config.get("org_id", ""), agent_id=paths.agent_id, kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), + base_url=config.get("base_url", DEFAULT_BASE_URL), native_payload=True, agent_name=AGENT_KEY, ) - changes = _present( - [ - _runtime_change( - paths.guard_path, "Elydora guard runtime", guard_script, 0o700 - ), - _runtime_change( - os.path.join(paths.agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, paths.agent_id)), - 0o600, - MAX_CONFIG_BYTES, - ), - _runtime_change( - os.path.join(paths.agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - MAX_SECRET_BYTES, - ), - _runtime_change( - paths.audit_path, "Elydora audit runtime", audit_script, 0o700 - ), - _runtime_change( + changes = [ + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *present([ + file_change( paths.guard_wrapper_path, "Auggie guard wrapper", build_wrapper(paths.guard_path), 0o700, ), - _runtime_change( + file_change( paths.audit_wrapper_path, "Auggie audit wrapper", build_wrapper(paths.audit_path), 0o700, ), rendered_change(rendered), - ] - ) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + ]), + ] + _validate_runtime_tree(paths) return changes @@ -193,10 +114,8 @@ def commit_augment_installation(changes: List[FileChange]) -> None: write_changes(changes, "Install Augment Code CLI hooks") -def prepare_augment_uninstall( - rendered: RenderedAugmentDocument, -) -> List[FileChange]: - return _present([rendered_change(rendered)]) +def prepare_augment_uninstall(rendered: RenderedAugmentDocument) -> List[FileChange]: + return present([rendered_change(rendered)]) def commit_augment_uninstall(changes: List[FileChange]) -> None: diff --git a/elydora/plugins/augment_io.py b/elydora/plugins/augment_io.py index 706946d..60d2b6c 100644 --- a/elydora/plugins/augment_io.py +++ b/elydora/plugins/augment_io.py @@ -2,61 +2,36 @@ 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, - physical_directory_exists, - physical_file_exists, - read_physical_file, -) -from ._strict_json import JsonObject, parse_json_object +from ._managed_files import MAX_CONFIG_BYTES, physical_directory_exists, read_physical_file +from ._runtime import runtime_contract_exists, same_path from ._transaction import FileChange, source_change, write_changes from .augment_contract import ( AGENT_KEY, - AUDIT_SCRIPT, AUDIT_WRAPPER, - GUARD_SCRIPT, GUARD_WRAPPER, AugmentDocument, RenderedAugmentDocument, RuntimeContract, build_wrapper, create_augment_document, - elydora_dir, parse_augment_document, resolve_config_path, ) -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 read_augment_document() -> AugmentDocument: config_path = resolve_config_path() - physical_directory_exists( - os.path.dirname(config_path), "Auggie configuration directory" - ) + physical_directory_exists(os.path.dirname(config_path), "Auggie configuration directory") snapshot = read_physical_file(config_path, "Auggie user settings", MAX_CONFIG_BYTES) if snapshot is None: return create_augment_document(config_path) return parse_augment_document(config_path, snapshot.contents) -def rendered_change( - rendered: RenderedAugmentDocument, -) -> Optional[FileChange]: +def rendered_change(rendered: RenderedAugmentDocument) -> Optional[FileChange]: if not rendered.changed: return None return source_change( @@ -75,192 +50,26 @@ def write_augment_document(rendered: RenderedAugmentDocument) -> None: write_changes([change], "Uninstall Augment Code CLI hooks") -def validate_api_origin(value: str, label: str = "base_url") -> None: - try: - parsed = urllib.parse.urlsplit(value) - hostname = parsed.hostname - parsed.port - except ValueError as error: - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error - invalid_character = "\\" in value or any( - character.isspace() or ord(character) < 32 for character in value - ) - if ( - parsed.scheme not in ("http", "https") - or not parsed.netloc - or hostname is None - or invalid_character - ): - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") - if ( - parsed.username is not None - or parsed.password is not None - or parsed.query - or parsed.fragment - ): - raise ValueError( - f"{label} must exclude credentials, query parameters, and fragments" - ) - - -def validate_private_key(value: str, label: str = "private_key") -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), validate=True - ) - canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - f"{label} must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or canonical != value: - raise ValueError(f"{label} must be a canonical 32-byte base64url value") - - -def _require_non_empty_string(value: Any, field: str, config_path: str) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError(f"Elydora runtime config {field} is invalid: {config_path}") - return value - - -def _validate_runtime_config( - config: JsonObject, expected_agent_id: str, config_path: str -) -> None: - supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} - extra = next((key for key in config if key not in supported), None) - if extra is not None: - raise ValueError( - f'Elydora runtime config has unsupported field "{extra}": {config_path}' - ) - _require_non_empty_string(config.get("org_id"), "org_id", config_path) - _require_non_empty_string(config.get("kid"), "kid", config_path) - agent_id = _require_non_empty_string( - config.get("agent_id"), "agent_id", config_path - ) - if ( - not _same_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - f"Elydora runtime identity does not match Auggie hooks: {config_path}" - ) - if "token" in config: - _require_non_empty_string(config.get("token"), "token", config_path) - base_url = _require_non_empty_string( - config.get("base_url"), "base_url", config_path - ) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file(file_path, "Elydora runtime config", MAX_CONFIG_BYTES) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, f"Elydora runtime config at {file_path}" - ) - - -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = 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_SCRIPT, "Elydora guard runtime"), - (AUDIT_SCRIPT, "Elydora audit runtime"), - (GUARD_WRAPPER, "Auggie guard wrapper"), - (AUDIT_WRAPPER, "Auggie audit wrapper"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ) - artifact_exists = any( - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not _same_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Auggie agent " - f"{agent_id}: {config_path}" - ) - - -def _runtime_contract_exists(contract: RuntimeContract) -> bool: - root = elydora_dir() +def _wrappers_exist(contract: RuntimeContract) -> bool: 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) - ) - or not _same_path( - contract.guard_wrapper_path, - os.path.join(agent_directory, GUARD_WRAPPER), - ) - or not _same_path( - contract.audit_wrapper_path, - os.path.join(agent_directory, AUDIT_WRAPPER), - ) + if not same_path( + contract.guard_wrapper_path, os.path.join(agent_directory, GUARD_WRAPPER) + ) or not same_path( + contract.audit_wrapper_path, os.path.join(agent_directory, AUDIT_WRAPPER) ): 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") - guard_wrapper = read_physical_file( - contract.guard_wrapper_path, "Auggie guard wrapper" - ) - audit_wrapper = read_physical_file( - contract.audit_wrapper_path, "Auggie audit wrapper" - ) - if any( - item is None - for item in (config, key, guard, audit, guard_wrapper, audit_wrapper) - ): - return False - assert config is not None - assert key is not None - assert guard is not None - assert audit is not None - assert guard_wrapper is not None - assert audit_wrapper is not None - _validate_runtime_config(config, contract.agent_id, config_path) - validate_private_key(key.contents, "Elydora private key") + guard_wrapper = read_physical_file(contract.guard_wrapper_path, "Auggie guard wrapper") + audit_wrapper = read_physical_file(contract.audit_wrapper_path, "Auggie audit wrapper") return ( - bool(guard.contents) - and bool(audit.contents) + guard_wrapper is not None + and audit_wrapper is not None and guard_wrapper.contents == build_wrapper(contract.guard_path) and audit_wrapper.contents == build_wrapper(contract.audit_path) ) def augment_runtime_files_exist(contracts: List[RuntimeContract]) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) + return any( + runtime_contract_exists(contract, AGENT_KEY, "Auggie") and _wrappers_exist(contract) + for contract in contracts + ) diff --git a/elydora/plugins/cline_contract.py b/elydora/plugins/cline_contract.py index 44e5e7a..ed23cde 100644 --- a/elydora/plugins/cline_contract.py +++ b/elydora/plugins/cline_contract.py @@ -8,7 +8,9 @@ import json import os from typing import Any, Dict, Optional -from elydora._runtime_paths import resolve_agent_directory +from elydora._runtime_paths import resolve_agent_directory, runtime_root + +from ._runtime import same_agent_id, same_path AGENT_KEY = "cline" @@ -58,20 +60,6 @@ def home_dir() -> str: return os.path.expanduser("~") -def elydora_dir() -> str: - return os.path.join(home_dir(), ".elydora") - - -def same_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - -def _same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - def resolve_hooks_directory() -> str: configured = os.environ.get("CLINE_DIR", "").strip() cline_directory = ( @@ -267,10 +255,10 @@ def runtime_contract( raise ValueError("Elydora Cline hook files contain mismatched event metadata") if not same_agent_id(guard.agent_id, audit.agent_id): raise ValueError("Elydora Cline hook files reference different agents") - agent_directory = resolve_agent_directory(elydora_dir(), guard.agent_id) + agent_directory = resolve_agent_directory(runtime_root(), guard.agent_id) expected_guard = os.path.join(agent_directory, GUARD_SCRIPT) expected_audit = os.path.join(agent_directory, AUDIT_SCRIPT) - if not _same_path(guard.runtime_path, expected_guard) or not _same_path( + if not same_path(guard.runtime_path, expected_guard) or not same_path( audit.runtime_path, expected_audit ): raise ValueError( diff --git a/elydora/plugins/cline_installation.py b/elydora/plugins/cline_installation.py index 9fec152..721b7ab 100644 --- a/elydora/plugins/cline_installation.py +++ b/elydora/plugins/cline_installation.py @@ -2,107 +2,42 @@ from __future__ import annotations -from dataclasses import dataclass -import json import os import sys from typing import List, Optional, Sequence -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import FileChange, file_change, source_change, write_changes +from ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + present, + resolve_runtime_paths, + runtime_file_changes, + same_agent_id, + same_path, + validate_install_config, + validate_runtime_tree, +) +from ._transaction import FileChange, source_change, write_changes from .base import InstallConfig from .cline_contract import ( AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, HookFile, assert_wrapper_integrity, build_metadata, build_wrapper, - elydora_dir, resolve_hook_files, runtime_contract, - same_agent_id, -) -from .cline_io import ( - validate_api_origin, - validate_hook_tree, - validate_private_key, - validate_runtime_tree, ) +from .cline_io import validate_hook_tree from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -@dataclass(frozen=True) -class ClineRuntimePaths: - agent_id: str - agent_directory: str - guard_path: str - audit_path: 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 _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) -> ClineRuntimePaths: - 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_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - f"Elydora guard runtime must use the managed agent directory: {guard_path}" - ) - return ClineRuntimePaths(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"Cline 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") - if not os.path.isabs(sys.executable): - raise RuntimeError("Cline requires an absolute Python executable path") - _agent_paths(config) +PRODUCT = "Cline" def _validate_hook_files(guard_file: HookFile, audit_file: HookFile) -> None: expected = resolve_hook_files() - if not _same_path(guard_file.file_path, expected.guard_path) or not _same_path( + if not same_path(guard_file.file_path, expected.guard_path) or not same_path( audit_file.file_path, expected.audit_path ): raise ValueError("Cline installation received unexpected hook paths") @@ -113,41 +48,23 @@ def preflight_cline_installation( config: InstallConfig, guard_file: HookFile, audit_file: HookFile, -) -> ClineRuntimePaths: - _validate_install_config(config) +) -> RuntimePaths: + validate_install_config(config, AGENT_KEY, PRODUCT) + if not os.path.isabs(sys.executable): + raise RuntimeError("Cline requires an absolute Python executable path") _validate_hook_files(guard_file, audit_file) - paths = _agent_paths(config) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + 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 _hook_change( file: HookFile, label: str, next_source: Optional[str] ) -> Optional[FileChange]: original = file.source if file.exists else None if file.exists and original is None: raise RuntimeError(f"{label} snapshot is missing source") - return source_change( - file.file_path, - label, - original, - next_source, - 0o700, - ) - - -def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] + return source_change(file.file_path, label, original, next_source, 0o700) def prepare_cline_installation( @@ -169,37 +86,18 @@ def prepare_cline_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, ) - changes = _present( - ( - _runtime_change( - paths.guard_path, "Elydora guard runtime", guard_script, 0o700 - ), - _runtime_change( - os.path.join(paths.agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, paths.agent_id)), - 0o600, - MAX_CONFIG_BYTES, - ), - _runtime_change( - os.path.join(paths.agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - MAX_SECRET_BYTES, - ), - _runtime_change( - paths.audit_path, "Elydora audit runtime", audit_script, 0o700 - ), + changes = [ + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *present([ _hook_change(guard_file, "Cline PreToolUse hook", guard_wrapper), _hook_change(audit_file, "Cline PostToolUse hook", audit_wrapper), - ) - ) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + ]), + ] + validate_runtime_tree(paths.agent_directory, paths.agent_id, AGENT_KEY, PRODUCT) validate_hook_tree(os.path.dirname(guard_file.file_path)) return changes @@ -211,21 +109,14 @@ def commit_cline_installation(changes: Sequence[FileChange]) -> None: def prepare_cline_uninstall( files: Sequence[HookFile], agent_id: str = "" ) -> List[FileChange]: - owned = [ - file - for file in files - if file.metadata is not None - and (not agent_id or same_agent_id(file.metadata.agent_id, agent_id)) - ] changes: List[Optional[FileChange]] = [] - for file in owned: + for file in files: metadata = file.metadata - assert metadata is not None + if metadata is None or (agent_id and not same_agent_id(metadata.agent_id, agent_id)): + continue assert_wrapper_integrity(file) - changes.append( - _hook_change(file, f"Cline {metadata.kind} hook", None) - ) - return _present(changes) + changes.append(_hook_change(file, f"Cline {metadata.kind} hook", None)) + return present(changes) def commit_cline_uninstall(changes: Sequence[FileChange]) -> None: diff --git a/elydora/plugins/cline_io.py b/elydora/plugins/cline_io.py index a3b5a87..bf01f06 100644 --- a/elydora/plugins/cline_io.py +++ b/elydora/plugins/cline_io.py @@ -2,43 +2,15 @@ from __future__ import annotations -import base64 import os -from typing import Any, Optional -import urllib.parse -from ._managed_files import ( - MAX_CONFIG_BYTES, - MAX_SECRET_BYTES, - physical_directory_exists, - physical_file_exists, - read_physical_file, -) -from ._strict_json import JsonObject, parse_json_object -from .cline_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - HookFile, - RuntimeContract, - elydora_dir, - parse_metadata, - same_agent_id, -) -from .guard_template import generate_guard_script -from .hook_template import generate_hook_script - - -def _same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) +from ._managed_files import physical_directory_exists, read_physical_file +from ._runtime import expected_runtime_scripts, runtime_contract_exists +from .cline_contract import AGENT_KEY, HookFile, RuntimeContract, parse_metadata def validate_hook_tree(hooks_directory: str) -> None: - physical_directory_exists( - os.path.dirname(hooks_directory), "Cline configuration directory" - ) + physical_directory_exists(os.path.dirname(hooks_directory), "Cline configuration directory") physical_directory_exists(hooks_directory, "Cline hooks directory") @@ -63,194 +35,7 @@ def require_available_hook_file(file: HookFile) -> None: ) -def validate_api_origin(value: str, label: str = "base_url") -> None: - try: - parsed = urllib.parse.urlsplit(value) - hostname = parsed.hostname - parsed.port - except ValueError as error: - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error - invalid_character = "\\" in value or any( - character.isspace() or ord(character) < 32 for character in value - ) - if ( - parsed.scheme not in ("http", "https") - or not parsed.netloc - or hostname is None - or invalid_character - ): - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") - if ( - parsed.username is not None - or parsed.password is not None - or parsed.query - or parsed.fragment - ): - raise ValueError( - f"{label} must exclude credentials, query parameters, and fragments" - ) - - -def validate_private_key(value: str, label: str = "private_key") -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), validate=True - ) - canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - f"{label} must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or canonical != value: - raise ValueError(f"{label} must be a canonical 32-byte base64url value") - - -def _require_non_empty_string(value: Any, field: str, config_path: str) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError(f"Elydora runtime config {field} is invalid: {config_path}") - return value - - -def _validate_runtime_config( - config: JsonObject, expected_agent_id: str, config_path: str -) -> None: - supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} - extra = next((key for key in config if key not in supported), None) - if extra is not None: - raise ValueError( - f'Elydora runtime config has unsupported field "{extra}": {config_path}' - ) - _require_non_empty_string(config.get("org_id"), "org_id", config_path) - _require_non_empty_string(config.get("kid"), "kid", config_path) - agent_id = _require_non_empty_string( - config.get("agent_id"), "agent_id", config_path - ) - if ( - not same_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - f"Elydora runtime identity does not match Cline hooks: {config_path}" - ) - if "token" in config: - _require_non_empty_string(config.get("token"), "token", config_path) - base_url = _require_non_empty_string( - config.get("base_url"), "base_url", config_path - ) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, "Elydora runtime config", MAX_CONFIG_BYTES - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, f"Elydora runtime config at {file_path}" - ) - - -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = 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_SCRIPT, "Elydora guard runtime"), - (AUDIT_SCRIPT, "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ) - artifact_exists = any( - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not same_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Cline agent " - f"{agent_id}: {config_path}" - ) - - -def _valid_contract_paths(contract: RuntimeContract) -> bool: - return ( - _same_path(os.path.dirname(contract.agent_directory), elydora_dir()) - and _same_path( - contract.guard_path, - os.path.join(contract.agent_directory, GUARD_SCRIPT), - ) - and _same_path( - contract.audit_path, - os.path.join(contract.agent_directory, AUDIT_SCRIPT), - ) - ) - - def runtime_files_exist(contract: RuntimeContract) -> bool: - if not _valid_contract_paths(contract): - return False - root = elydora_dir() - if not physical_directory_exists(root, "Elydora runtime directory"): - return False - if not physical_directory_exists( - contract.agent_directory, "Elydora agent runtime directory" - ): - return False - config_path = os.path.join(contract.agent_directory, "config.json") - key_path = os.path.join(contract.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 any(item is None for item in (config, key, guard, audit)): - return False - assert config is not None - assert key is not None - assert guard is not None - assert audit is not None - _validate_runtime_config(config, contract.agent_id, config_path) - validate_private_key(key.contents, "Elydora private key") - expected_audit = generate_hook_script( - org_id="", - agent_id=contract.agent_id, - kid="", - base_url="", - native_payload=True, - agent_name=AGENT_KEY, + return runtime_contract_exists( + contract, AGENT_KEY, "Cline", expected_runtime_scripts(AGENT_KEY, contract.agent_id) ) - return ( - guard.contents == generate_guard_script(AGENT_KEY, contract.agent_id) - and audit.contents == expected_audit - ) - - -__all__ = [ - "read_hook_file", - "require_available_hook_file", - "runtime_files_exist", - "validate_api_origin", - "validate_hook_tree", - "validate_private_key", - "validate_runtime_tree", -] diff --git a/elydora/plugins/copilot_contract.py b/elydora/plugins/copilot_contract.py index b9b1fd2..20be48a 100644 --- a/elydora/plugins/copilot_contract.py +++ b/elydora/plugins/copilot_contract.py @@ -7,10 +7,19 @@ import json import os import re import shlex -import sys from typing import Any, Dict, List, Optional, Tuple +from elydora._runtime_paths import runtime_root + from ._managed_files import FileSnapshot +from ._runtime import same_agent_id, same_path +from ._shell_command import ( + is_python_executable, + parse_posix_command, + parse_powershell_source, + posix_source, + powershell_source, +) from ._strict_json import parse_json_object from .copilot_schema import CopilotHooks, validate_hooks @@ -69,124 +78,35 @@ class RuntimeContract: audit_path: str -@dataclass(frozen=True) -class _ParsedArgument: - value: str - next_index: int - - @dataclass(frozen=True) class _ManagedEntry: agent_id: str script_path: str -def runtime_root() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - -def same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_agent_id(left: str, right: str) -> bool: - 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: return { "type": "command", - "bash": f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}", - "powershell": ( - f"& {_quote_powershell(sys.executable)} " - f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" - ), + "bash": posix_source(script_path), + "powershell": powershell_source(script_path), "timeoutSec": HOOK_TIMEOUT_SECONDS, } -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_generated_bash(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_generated_powershell(command: Any) -> Optional[Tuple[str, str]]: - if not isinstance(command, str) or not command.startswith("& "): - return None - executable = _read_powershell_argument(command, 2) - if executable is None or command[executable.next_index:executable.next_index + 1] != " ": - return None - script = _read_powershell_argument(command, executable.next_index + 1) - if script is None or command[script.next_index:] != "; exit $LASTEXITCODE": - return None - return executable.value, script.value - - def _exact_handler_keys(handler: JsonObject) -> bool: return set(handler) == {"type", "bash", "powershell", "timeoutSec"} -def _is_python_executable(file_path: str) -> bool: - return os.path.isabs(file_path) and re.fullmatch( - r"(?:python|pypy)(?:[0-9]+(?:\.[0-9]+)*)?(?:\.exe)?", - os.path.basename(file_path), - re.IGNORECASE, - ) is not None +def _parse_generated_bash(command: Any) -> Optional[Tuple[str, str]]: + return parse_posix_command(command) if isinstance(command, str) else None + + +def _parse_generated_powershell(command: Any) -> Optional[Tuple[str, str]]: + return parse_powershell_source(command) if isinstance(command, str) else None + + +def _absolute_python(file_path: str) -> bool: + return os.path.isabs(file_path) and is_python_executable(file_path) def _current_script_path(handler: JsonObject) -> Optional[str]: @@ -201,7 +121,7 @@ def _current_script_path(handler: JsonObject) -> Optional[str]: if ( bash is None or powershell is None - or not _is_python_executable(bash[0]) + or not _absolute_python(bash[0]) or not same_path(bash[0], powershell[0]) or not same_path(bash[1], powershell[1]) ): @@ -227,7 +147,7 @@ def _prior_script_path(handler: JsonObject) -> Optional[str]: len(arguments) != 2 or shlex.join(arguments) != bash_command or powershell is None - or not _is_python_executable(arguments[0]) + or not _absolute_python(arguments[0]) or not same_path(arguments[0], powershell[0]) or not same_path(arguments[1], powershell[1]) ): diff --git a/elydora/plugins/copilot_installation.py b/elydora/plugins/copilot_installation.py index 0a63e41..191f686 100644 --- a/elydora/plugins/copilot_installation.py +++ b/elydora/plugins/copilot_installation.py @@ -3,48 +3,29 @@ from __future__ import annotations from dataclasses import dataclass -import json import os import sys -from typing import List, Optional, Sequence, Tuple +from typing import Optional, Sequence, Tuple -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import ( - FileChange, - FilePrecondition, - file_change, - source_change, - write_changes, -) -from .base import InstallConfig -from .copilot_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - CopilotSources, - RenderedDocument, - runtime_root, +from ._managed_files import MAX_SOURCE_BYTES +from ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + resolve_runtime_paths, + runtime_file_changes, same_path, -) -from .copilot_io import ( - require_hooks_enabled, - validate_api_origin, - validate_private_key, + validate_install_config, validate_runtime_tree, ) +from ._transaction import FileChange, FilePrecondition, source_change, write_changes +from .base import InstallConfig +from .copilot_contract import AGENT_KEY, CopilotSources, RenderedDocument +from .copilot_io import require_hooks_enabled from .copilot_schema import validate_javascript_regexes from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -@dataclass(frozen=True) -class CopilotRuntimePaths: - agent_id: str - agent_directory: str - guard_path: str - audit_path: str +PRODUCT = "GitHub Copilot CLI" @dataclass(frozen=True) @@ -53,95 +34,25 @@ class PreparedCopilotInstallation: preconditions: Tuple[FilePrecondition, ...] -def _runtime_config(config: InstallConfig, agent_id: str) -> dict: - value = { - "org_id": config.get("org_id", ""), - "agent_id": agent_id, - "kid": config.get("kid", ""), - "base_url": config.get("base_url", "https://api.elydora.com"), - "agent_name": AGENT_KEY, - } - token = config.get("token") - if token: - value["token"] = token - return value - - -def _json_source(value: dict) -> str: - return json.dumps(value, indent=2) + "\n" - - -def _agent_paths(config: InstallConfig) -> CopilotRuntimePaths: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - agent_directory = resolve_agent_directory(runtime_root(), agent_id) - guard_path = os.path.join(agent_directory, GUARD_SCRIPT) - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - if not same_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return CopilotRuntimePaths( - agent_id, - agent_directory, - guard_path, - audit_path, - ) - - -def _validate_install_config(config: InstallConfig) -> None: - for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): - value = config.get(field) - if not isinstance(value, str) or not value: - raise ValueError(f"{field} is required") - if config.get("agent_name") != AGENT_KEY: - raise ValueError( - f"GitHub Copilot CLI installation requires agent_name {AGENT_KEY}" - ) - if not os.path.isabs(sys.executable): - raise RuntimeError( - "GitHub Copilot CLI requires an absolute Python executable path" - ) - validate_private_key(config["private_key"]) - validate_api_origin(config["base_url"]) - if "token" in config: - token = config["token"] - if not isinstance(token, str) or not token: - raise ValueError("token must be a non-empty string when provided") - _agent_paths(config) - - def preflight_copilot_installation( config: InstallConfig, sources: CopilotSources, -) -> CopilotRuntimePaths: +) -> RuntimePaths: if not sources.user.file_path: - raise ValueError( - "GitHub Copilot CLI installation requires a user hook path" - ) + raise ValueError("GitHub Copilot CLI installation requires a user hook path") require_hooks_enabled(sources) hook_sources = [sources.user.hooks] if sources.legacy is not None: hook_sources.append(sources.legacy.hooks) validate_javascript_regexes(hook_sources) - _validate_install_config(config) - paths = _agent_paths(config) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + validate_install_config(config, AGENT_KEY, PRODUCT) + if not os.path.isabs(sys.executable): + raise RuntimeError("GitHub Copilot CLI requires an absolute Python executable path") + paths = resolve_runtime_paths(config) + validate_runtime_tree(paths.agent_directory, paths.agent_id, AGENT_KEY, "Copilot") return paths -def _runtime_change( - file_path: str, - label: str, - next_source: str, - mode: int, - maximum_bytes: int = MAX_SOURCE_BYTES, -) -> Optional[FileChange]: - return file_change(file_path, label, next_source, mode, maximum_bytes) - - def _document_change(rendered: RenderedDocument) -> Optional[FileChange]: if not rendered.changed: return None @@ -156,12 +67,6 @@ def _document_change(rendered: RenderedDocument) -> Optional[FileChange]: ) -def _present( - changes: Sequence[Optional[FileChange]], -) -> Tuple[FileChange, ...]: - return tuple(change for change in changes if change is not None) - - def _document_preconditions( rendered: Sequence[RenderedDocument], ) -> Tuple[FilePrecondition, ...]: @@ -177,16 +82,9 @@ def _document_preconditions( ) -def _settings_preconditions( - sources: CopilotSources, -) -> Tuple[FilePrecondition, ...]: +def _settings_preconditions(sources: CopilotSources) -> Tuple[FilePrecondition, ...]: return tuple( - FilePrecondition( - item.file_path, - item.label, - item.snapshot, - MAX_SOURCE_BYTES, - ) + FilePrecondition(item.file_path, item.label, item.snapshot, MAX_SOURCE_BYTES) for item in sources.settings_preconditions ) @@ -206,9 +104,7 @@ def _validate_rendered_documents( if same_path(item.document.file_path, document.file_path) ] if len(matches) != 1: - raise ValueError( - "GitHub Copilot rendered source set contains unexpected paths" - ) + raise ValueError("GitHub Copilot rendered source set contains unexpected paths") def prepare_copilot_installation( @@ -223,33 +119,18 @@ def prepare_copilot_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, ) - changes = _present(( - _runtime_change( - paths.guard_path, "Elydora guard runtime", guard_script, 0o700 - ), - _runtime_change( - os.path.join(paths.agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, paths.agent_id)), - 0o600, - MAX_CONFIG_BYTES, - ), - _runtime_change( - os.path.join(paths.agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - MAX_SECRET_BYTES, - ), - _runtime_change( - paths.audit_path, "Elydora audit runtime", audit_script, 0o700 - ), - *(_document_change(item) for item in rendered), - )) + document_changes = [ + change for change in (_document_change(item) for item in rendered) + if change is not None + ] + changes = ( + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *document_changes, + ) preconditions = ( *_settings_preconditions(sources), *_document_preconditions(rendered), @@ -257,20 +138,17 @@ def prepare_copilot_installation( return PreparedCopilotInstallation(changes, preconditions) -def commit_copilot_installation( - prepared: PreparedCopilotInstallation, -) -> None: - write_changes( - prepared.changes, - "Install GitHub Copilot hooks", - prepared.preconditions, - ) +def commit_copilot_installation(prepared: PreparedCopilotInstallation) -> None: + write_changes(prepared.changes, "Install GitHub Copilot hooks", prepared.preconditions) def prepare_copilot_uninstall( rendered: Sequence[RenderedDocument], ) -> Tuple[FileChange, ...]: - return _present(tuple(_document_change(item) for item in rendered)) + return tuple( + change for change in (_document_change(item) for item in rendered) + if change is not None + ) def commit_copilot_uninstall(changes: Sequence[FileChange]) -> None: diff --git a/elydora/plugins/copilot_io.py b/elydora/plugins/copilot_io.py index ae88f9e..d0ee093 100644 --- a/elydora/plugins/copilot_io.py +++ b/elydora/plugins/copilot_io.py @@ -2,40 +2,29 @@ from __future__ import annotations -import base64 from dataclasses import dataclass import os -from typing import Any, List, Optional, Tuple -import urllib.parse +from typing import List, Optional, Tuple from ._jsonc import parse_jsonc from ._managed_files import ( FileSnapshot, - MAX_CONFIG_BYTES, - MAX_SECRET_BYTES, MAX_SOURCE_BYTES, physical_directory_exists, - physical_file_exists, read_physical_file, ) +from ._runtime import expected_runtime_scripts, runtime_contract_exists from ._strict_json import JsonObject, parse_json_object from .copilot_contract import ( AGENT_KEY, - AUDIT_SCRIPT, CONFIG_FILE, - GUARD_SCRIPT, CopilotDocument, CopilotSources, RuntimeContract, SourcePrecondition, create_document, parse_document, - runtime_root, - same_agent_id, - same_path, ) -from .guard_template import generate_guard_script -from .hook_template import generate_hook_script @dataclass(frozen=True) @@ -216,186 +205,13 @@ def require_hooks_enabled(sources: CopilotSources) -> None: ) -def validate_api_origin(value: str, label: str = "base_url") -> None: - try: - parsed = urllib.parse.urlsplit(value) - hostname = parsed.hostname - parsed.port - except ValueError as error: - raise ValueError( - f"{label} must be an absolute HTTP or HTTPS URL" - ) from error - invalid_character = "\\" in value or any( - character.isspace() or ord(character) < 32 for character in value - ) - if ( - parsed.scheme not in ("http", "https") - or not parsed.netloc - or hostname is None - or invalid_character - ): - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") - if ( - parsed.username is not None - or parsed.password is not None - or parsed.query - or parsed.fragment - ): - raise ValueError( - f"{label} must exclude credentials, query parameters, and fragments" - ) - - -def validate_private_key(value: str, label: str = "private_key") -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), - validate=True, - ) - canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - f"{label} must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or canonical != value: - raise ValueError(f"{label} must be a canonical 32-byte base64url value") - - -def _require_string(value: Any, field: str, config_path: str) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError( - f"Elydora runtime config {field} is invalid: {config_path}" - ) - return value - - -def _validate_runtime_config( - config: JsonObject, - contract: RuntimeContract, - config_path: str, -) -> None: - supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} - extra = next((key for key in config if key not in supported), None) - if extra is not None: - raise ValueError( - f'Elydora runtime config has unsupported field "{extra}": {config_path}' - ) - _require_string(config.get("org_id"), "org_id", config_path) - _require_string(config.get("kid"), "kid", config_path) - agent_id = _require_string(config.get("agent_id"), "agent_id", config_path) - if ( - not same_agent_id(agent_id, contract.agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - f"Elydora runtime identity does not match Copilot hooks: {config_path}" - ) - if "token" in config: - _require_string(config.get("token"), "token", config_path) - base_url = _require_string(config.get("base_url"), "base_url", config_path) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, "Elydora runtime config", MAX_CONFIG_BYTES - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, f"Elydora runtime config at {file_path}" - ) - - -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = runtime_root() - if not physical_directory_exists(root, "Elydora runtime directory"): - return - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return - config_path = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path) - artifacts = ( - ("private.key", "Elydora private key"), - (GUARD_SCRIPT, "Elydora guard runtime"), - (AUDIT_SCRIPT, "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ) - artifact_exists = any( - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not same_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Copilot agent " - f"{agent_id}: {config_path}" - ) - - -def _valid_contract_paths(contract: RuntimeContract) -> bool: - agent_directory = os.path.dirname(contract.guard_path) - return ( - same_path(os.path.dirname(agent_directory), runtime_root()) - and same_path( - contract.guard_path, - os.path.join(agent_directory, GUARD_SCRIPT), - ) - and same_path( - contract.audit_path, - os.path.join(agent_directory, AUDIT_SCRIPT), - ) - ) - - -def _runtime_contract_exists(contract: RuntimeContract) -> bool: - if not _valid_contract_paths(contract): - return False - root = runtime_root() - agent_directory = os.path.dirname(contract.guard_path) - if not physical_directory_exists(root, "Elydora runtime directory"): - return False - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return False - config_path = os.path.join(agent_directory, "config.json") - key_path = os.path.join(agent_directory, "private.key") - config = _read_runtime_config(config_path) - key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) - guard = read_physical_file(contract.guard_path, "Elydora guard runtime") - audit = read_physical_file(contract.audit_path, "Elydora audit runtime") - if config is None or key is None or guard is None or audit is None: - return False - _validate_runtime_config(config, contract, config_path) - validate_private_key(key.contents, "Elydora private key") - expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id) - expected_audit = generate_hook_script( - org_id="", - agent_id=contract.agent_id, - kid="", - base_url="", - native_payload=True, - agent_name=AGENT_KEY, - ) - return guard.contents == expected_guard and audit.contents == expected_audit - - def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) + return any( + runtime_contract_exists( + contract, + AGENT_KEY, + "Copilot", + expected_runtime_scripts(AGENT_KEY, contract.agent_id), + ) + for contract in contracts + ) diff --git a/elydora/plugins/droid_config.py b/elydora/plugins/droid_config.py index 0cc4130..ab094a9 100644 --- a/elydora/plugins/droid_config.py +++ b/elydora/plugins/droid_config.py @@ -8,13 +8,13 @@ from typing import Dict, List, Optional, Tuple from ._jsonc import JsonPathPart, JsoncEditor from ._managed_files import FileSnapshot +from ._runtime import same_path from .droid_contract import ( TOOL_EVENTS, DroidHookMap, JsonObject, managed_removals, read_hook_map, - same_path, ) from .droid_policy import DroidPolicyState diff --git a/elydora/plugins/droid_contract.py b/elydora/plugins/droid_contract.py index 32716cd..b396055 100644 --- a/elydora/plugins/droid_contract.py +++ b/elydora/plugins/droid_contract.py @@ -12,6 +12,16 @@ import subprocess # nosec B404 import sys from typing import Any, Dict, List, Optional, Sequence, Set, Tuple +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, +) + AGENT_KEY = "droid" GUARD_SCRIPT = "guard.py" @@ -22,7 +32,6 @@ TOOL_EVENTS = ("PreToolUse", "PostToolUse") _HANDLER_KEYS = {"command", "timeout", "type"} _GROUP_KEYS = {"hooks", "matcher"} _REGEX_TIMEOUT_SECONDS = 10 -_WINDOWS_EXIT_SUFFIX = "; exit $LASTEXITCODE" _REGEX_VALIDATOR = """import fs from 'node:fs'; const entries = JSON.parse(fs.readFileSync(0, 'utf8')); for (const entry of entries) { @@ -55,39 +64,14 @@ class ManagedRemoval: remove_group: bool -def elydora_dir() -> 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_power_shell(value: str) -> str: - return "'" + value.replace("'", "''") + "'" - - -def _quote_posix(value: str) -> str: - return "'" + value.replace("'", "'\"'\"'") + "'" - - def build_command(script_path: str) -> str: if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): raise ValueError( "Factory Droid hook commands require absolute executable and script paths" ) if os.name == "nt": - return ( - f"& {_quote_power_shell(sys.executable)} " - f"{_quote_power_shell(script_path)}{_WINDOWS_EXIT_SUFFIX}" - ) - return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" + return powershell_source(script_path) + return posix_source(script_path) def build_group(script_path: str) -> JsonObject: @@ -214,83 +198,6 @@ def validate_javascript_regexes(sources: Sequence[DroidHookMap]) -> None: ) -def _read_power_shell_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 - - -_POSIX_APOSTROPHE = "'\"'\"'" - - -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_two_arguments( - command: str, - start: int, -) -> Optional[Tuple[str, str]]: - executable = _read_posix_argument(command, start) - 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 - if not executable[0] or not script[0]: - return None - return executable[0], script[0] - - -def _parse_power_shell_command(command: str) -> Optional[Tuple[str, str]]: - if not command.startswith("& "): - return None - executable = _read_power_shell_argument(command, 2) - if executable is None or executable[1] >= len(command): - return None - if command[executable[1]] != " ": - return None - script = _read_power_shell_argument(command, executable[1] + 1) - if ( - script is None - or command[script[1]:] != _WINDOWS_EXIT_SUFFIX - or not executable[0] - or not script[0] - ): - return None - return executable[0], script[0] - - def _parse_legacy_windows_command(command: str) -> Optional[Tuple[str, str]]: match = re.fullmatch(r'"([^"\r\n]+)" "([^"\r\n]+)"', command) return (match.group(1), match.group(2)) if match else None @@ -301,8 +208,8 @@ def _parse_generated_command( include_legacy: bool, ) -> Optional[Tuple[str, str]]: if os.name != "nt": - return _parse_two_arguments(command, 0) - current = _parse_power_shell_command(command) + return parse_posix_command(command) + current = parse_powershell_source(command) return current or ( _parse_legacy_windows_command(command) if include_legacy else None ) @@ -326,14 +233,8 @@ def managed_agent_id( parsed = _parse_generated_command(command, include_legacy) if parsed is None or not same_path(parsed[0], sys.executable): return None - script_path = parsed[1] - if 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), elydora_dir()): - return None - agent_id = os.path.basename(agent_directory) - return agent_id if agent_id not in {"", ".", ".."} else None + reference = managed_script_reference(parsed[1], script_name) + return None if reference is None else reference[0] def _exact_owned_group(group: JsonObject, indexes: Tuple[int, ...]) -> bool: @@ -397,7 +298,7 @@ def runtime_contracts(hooks: DroidHookMap) -> List[RuntimeContract]: for agent_id in sorted(guards): if not any(same_agent_id(agent_id, audit_id) for audit_id in audits): continue - root = os.path.join(elydora_dir(), agent_id) + root = os.path.join(runtime_root(), agent_id) contracts.append(RuntimeContract( agent_id, os.path.join(root, GUARD_SCRIPT), diff --git a/elydora/plugins/droid_installation.py b/elydora/plugins/droid_installation.py index a5ac28d..487a421 100644 --- a/elydora/plugins/droid_installation.py +++ b/elydora/plugins/droid_installation.py @@ -3,21 +3,21 @@ from __future__ import annotations from dataclasses import dataclass -import json import os import sys from typing import Optional, Sequence, Tuple -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import ( - FileChange, - FilePrecondition, - file_change, - source_change, - write_changes, +from ._managed_files import MAX_SOURCE_BYTES +from ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + resolve_runtime_paths, + runtime_file_changes, + same_path, + validate_install_config, + validate_runtime_tree, ) +from ._transaction import FileChange, FilePrecondition, source_change, write_changes from .base import InstallConfig from .droid_config import ( DroidDocument, @@ -26,30 +26,12 @@ from .droid_config import ( installation_documents, source_documents, ) -from .droid_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - elydora_dir, - same_path, - validate_javascript_regexes, -) -from .droid_io import ( - require_hooks_enabled, - validate_api_origin, - validate_private_key, - validate_runtime_tree, -) +from .droid_contract import AGENT_KEY, validate_javascript_regexes +from .droid_io import require_hooks_enabled from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -@dataclass(frozen=True) -class DroidRuntimePaths: - agent_id: str - agent_directory: str - guard_path: str - audit_path: str +PRODUCT = "Factory Droid" @dataclass(frozen=True) @@ -74,90 +56,20 @@ def _source_label(document: DroidDocument) -> str: return "Factory Droid user hooks" -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) -> DroidRuntimePaths: - 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_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return DroidRuntimePaths( - 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"Factory Droid installation requires agent_name {AGENT_KEY}" - ) - if not os.path.isabs(sys.executable): - raise RuntimeError( - "Factory Droid requires an absolute Python executable path" - ) - validate_private_key(config["private_key"]) - validate_api_origin(config["base_url"]) - if "token" in config: - token = config["token"] - if not isinstance(token, str) or not token: - raise ValueError("token must be a non-empty string when provided") - _agent_paths(config) - - def preflight_droid_installation( config: InstallConfig, sources: DroidSources, -) -> DroidRuntimePaths: +) -> RuntimePaths: require_hooks_enabled(sources) - validate_javascript_regexes([ - document.hooks for document in source_documents(sources) - ]) - _validate_install_config(config) - paths = _agent_paths(config) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + validate_javascript_regexes([document.hooks for document in source_documents(sources)]) + validate_install_config(config, AGENT_KEY, PRODUCT) + if not os.path.isabs(sys.executable): + raise RuntimeError("Factory Droid requires an absolute Python executable path") + 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 _document_change(rendered: RenderedDocument) -> Optional[FileChange]: if not rendered.changed: return None @@ -173,10 +85,11 @@ def _document_change(rendered: RenderedDocument) -> Optional[FileChange]: ) -def _present( - changes: Sequence[Optional[FileChange]], -) -> Tuple[FileChange, ...]: - return tuple(change for change in changes if change is not None) +def _document_changes(rendered: Sequence[RenderedDocument]) -> Tuple[FileChange, ...]: + return tuple( + change for change in (_document_change(item) for item in rendered) + if change is not None + ) def _validate_rendered_documents( @@ -188,18 +101,15 @@ def _validate_rendered_documents( raise ValueError("Factory Droid rendered source set is incomplete") for document in expected: matches = [ - item - for item in rendered + item for item in rendered if same_path(item.document.file_path, document.file_path) ] if len(matches) != 1: - raise ValueError( - "Factory Droid rendered source set contains unexpected paths" - ) + raise ValueError("Factory Droid rendered source set contains unexpected paths") -def _source_preconditions( - sources: DroidSources, +def _unchanged_preconditions( + documents: Sequence[DroidDocument], changed_paths: Sequence[str], ) -> Tuple[FilePrecondition, ...]: return tuple( @@ -209,11 +119,8 @@ def _source_preconditions( document.snapshot, MAX_SOURCE_BYTES, ) - for document in source_documents(sources) - if not any( - same_path(document.file_path, changed) - for changed in changed_paths - ) + for document in documents + if not any(same_path(document.file_path, changed) for changed in changed_paths) ) @@ -229,84 +136,35 @@ def prepare_droid_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, ) - document_changes = _present(tuple( - _document_change(item) for item in rendered - )) - changes = _present(( - _runtime_change( - paths.guard_path, - "Elydora guard runtime", - guard_script, - 0o700, - ), - _runtime_change( - os.path.join(paths.agent_directory, "config.json"), - "Elydora runtime config", - _json_source(_runtime_config(config, paths.agent_id)), - 0o600, - MAX_CONFIG_BYTES, - ), - _runtime_change( - os.path.join(paths.agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - MAX_SECRET_BYTES, - ), - _runtime_change( - paths.audit_path, - "Elydora audit runtime", - audit_script, - 0o700, - ), + document_changes = _document_changes(rendered) + changes = ( + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), *document_changes, - )) + ) changed_paths = [change.file_path for change in document_changes] preconditions = ( - *_source_preconditions(sources, changed_paths), + *_unchanged_preconditions(source_documents(sources), changed_paths), *sources.policy.preconditions, ) - return PreparedDroidInstallation(changes, tuple(preconditions)) + return PreparedDroidInstallation(changes, preconditions) -def commit_droid_installation( - prepared: PreparedDroidInstallation, -) -> None: - write_changes( - prepared.changes, - "Install Factory Droid hooks", - prepared.preconditions, - ) +def commit_droid_installation(prepared: PreparedDroidInstallation) -> None: + write_changes(prepared.changes, "Install Factory Droid hooks", prepared.preconditions) def prepare_droid_uninstall( rendered: Sequence[RenderedDocument], ) -> PreparedDroidUninstall: - changes = _present(tuple(_document_change(item) for item in rendered)) + changes = _document_changes(rendered) changed_paths = [change.file_path for change in changes] - preconditions = tuple( - FilePrecondition( - item.document.file_path, - _source_label(item.document), - item.document.snapshot, - MAX_SOURCE_BYTES, - ) - for item in rendered - if not any( - same_path(item.document.file_path, changed) - for changed in changed_paths - ) - ) - return PreparedDroidUninstall(changes, preconditions) + documents = [item.document for item in rendered] + return PreparedDroidUninstall(changes, _unchanged_preconditions(documents, changed_paths)) def commit_droid_uninstall(prepared: PreparedDroidUninstall) -> None: - write_changes( - prepared.changes, - "Uninstall Factory Droid hooks", - prepared.preconditions, - ) + write_changes(prepared.changes, "Uninstall Factory Droid hooks", prepared.preconditions) diff --git a/elydora/plugins/droid_io.py b/elydora/plugins/droid_io.py index 31f2bd1..e7a973a 100644 --- a/elydora/plugins/droid_io.py +++ b/elydora/plugins/droid_io.py @@ -2,21 +2,12 @@ from __future__ import annotations -import base64 from dataclasses import dataclass import os -from typing import Any, 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 ._managed_files import MAX_SOURCE_BYTES, physical_directory_exists, read_physical_file +from ._runtime import expected_runtime_scripts, runtime_contract_exists from .droid_config import ( DroidDocument, DroidSources, @@ -27,18 +18,8 @@ from .droid_config import ( hook_block, parse_document, ) -from .droid_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - RuntimeContract, - elydora_dir, - same_agent_id, - same_path, -) +from .droid_contract import AGENT_KEY, RuntimeContract from .droid_policy import read_droid_policy -from .guard_template import generate_guard_script -from .hook_template import generate_hook_script @dataclass(frozen=True) @@ -128,196 +109,16 @@ def require_hooks_enabled(sources: DroidSources) -> None: ) -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 runtime_files_exist(contracts: List[RuntimeContract]) -> bool: + return any( + runtime_contract_exists( + contract, + AGENT_KEY, + "Factory Droid", + expected_runtime_scripts(AGENT_KEY, contract.agent_id), + ) + 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_string(value: Any, field: str, config_path: str) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError( - f"Elydora runtime config {field} is invalid: {config_path}" - ) - return value - - -def _validate_runtime_config( - config: JsonObject, - contract: RuntimeContract, - config_path: str, -) -> None: - supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} - extra = next((key for key in config if key not in supported), None) - if extra is not None: - raise ValueError( - f'Elydora runtime config has unsupported field "{extra}": ' - f"{config_path}" - ) - _require_string(config.get("org_id"), "org_id", config_path) - _require_string(config.get("kid"), "kid", config_path) - agent_id = _require_string(config.get("agent_id"), "agent_id", config_path) - if ( - not same_agent_id(agent_id, contract.agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime identity does not match Factory Droid hooks: " - f"{config_path}" - ) - if "token" in config: - _require_string(config.get("token"), "token", config_path) - base_url = _require_string(config.get("base_url"), "base_url", config_path) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, - "Elydora runtime config", - MAX_CONFIG_BYTES, - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, - f"Elydora runtime config at {file_path}", - ) - - -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = 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_SCRIPT, "Elydora guard runtime"), - (AUDIT_SCRIPT, "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ) - artifact_exists = any( - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not same_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Factory Droid agent " - f"{agent_id}: {config_path}" - ) - - -def _valid_contract_paths(contract: RuntimeContract) -> bool: - agent_directory = os.path.dirname(contract.guard_path) - return ( - same_path(os.path.dirname(agent_directory), elydora_dir()) - and same_path( - contract.guard_path, - os.path.join(agent_directory, GUARD_SCRIPT), - ) - and same_path( - contract.audit_path, - os.path.join(agent_directory, AUDIT_SCRIPT), - ) - ) - - -def _runtime_contract_exists(contract: RuntimeContract) -> bool: - if not _valid_contract_paths(contract): - return False - root = elydora_dir() - agent_directory = os.path.dirname(contract.guard_path) - if not physical_directory_exists(root, "Elydora runtime directory"): - return False - if not physical_directory_exists( - agent_directory, - "Elydora agent runtime directory", - ): - return False - config_path = os.path.join(agent_directory, "config.json") - key_path = os.path.join(agent_directory, "private.key") - config = _read_runtime_config(config_path) - key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) - guard = read_physical_file(contract.guard_path, "Elydora guard runtime") - audit = read_physical_file(contract.audit_path, "Elydora audit runtime") - if config is None or key is None or guard is None or audit is None: - return False - _validate_runtime_config(config, contract, config_path) - validate_private_key(key.contents, "Elydora private key") - expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id) - expected_audit = generate_hook_script( - org_id="", - agent_id=contract.agent_id, - kid="", - base_url="", - native_payload=True, - agent_name=AGENT_KEY, - ) - return guard.contents == expected_guard and audit.contents == expected_audit - - -def runtime_files_exist(contracts: list[RuntimeContract]) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) def display_config_path(sources: DroidSources) -> str: