From 407e4628a0e4793225e83c2694caa010c5af29b3 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:58:10 -0400 Subject: [PATCH] chore(python): audit providers kimi grok qwen letta Move the four providers onto _runtime and _shell_command; expected runtime scripts take the runtime config; add unique_preconditions. Claude-Session: https://claude.ai/code/session_01Agoc9bRVE9KL7mmWtey9Vq --- elydora/plugins/_runtime.py | 41 +++++- elydora/plugins/cline_io.py | 10 +- elydora/plugins/copilot_io.py | 6 +- elydora/plugins/droid_io.py | 9 +- elydora/plugins/grok_command.py | 188 +++-------------------- elydora/plugins/grok_contract.py | 7 +- elydora/plugins/grok_installation.py | 163 ++++---------------- elydora/plugins/grok_io.py | 189 ++---------------------- elydora/plugins/kimi_command.py | 188 +++-------------------- elydora/plugins/kimi_contract.py | 7 +- elydora/plugins/kimi_installation.py | 168 ++++----------------- elydora/plugins/kimi_io.py | 194 +----------------------- elydora/plugins/letta.py | 4 +- elydora/plugins/letta_command.py | 132 +++-------------- elydora/plugins/letta_contract.py | 15 +- elydora/plugins/letta_installation.py | 162 ++++---------------- elydora/plugins/letta_io.py | 205 ++------------------------ elydora/plugins/letta_sources.py | 17 +-- elydora/plugins/qwen.py | 4 +- elydora/plugins/qwen_command.py | 120 ++------------- elydora/plugins/qwen_contract.py | 11 +- elydora/plugins/qwen_installation.py | 158 ++++---------------- elydora/plugins/qwen_io.py | 204 ++----------------------- elydora/plugins/qwen_sources.py | 15 +- 24 files changed, 297 insertions(+), 1920 deletions(-) diff --git a/elydora/plugins/_runtime.py b/elydora/plugins/_runtime.py index 74afd19..6029ee1 100644 --- a/elydora/plugins/_runtime.py +++ b/elydora/plugins/_runtime.py @@ -6,7 +6,7 @@ import base64 from dataclasses import dataclass import json import os -from typing import Any, List, Optional, Protocol, Tuple +from typing import Any, Callable, Dict, List, Optional, Protocol, Sequence, Tuple import urllib.parse from elydora._runtime_paths import resolve_agent_directory, runtime_root @@ -20,7 +20,7 @@ from ._managed_files import ( read_physical_file, ) from ._strict_json import JsonObject, parse_json_object -from ._transaction import FileChange, file_change, source_change +from ._transaction import FileChange, FilePrecondition, file_change, source_change from .base import InstallConfig from .guard_template import generate_guard_script from .hook_template import generate_hook_script @@ -252,16 +252,31 @@ def validate_runtime_tree( ) -def expected_runtime_scripts(agent_key: str, agent_id: str) -> Tuple[str, str]: +ExpectedScripts = Callable[[str, JsonObject], Tuple[str, str]] + + +def expected_runtime_scripts( + agent_key: str, + agent_id: str, + config: Optional[JsonObject] = None, + **hook_options: Any, +) -> Tuple[str, str]: + """Guard and audit sources; identity fields come from config when given.""" + org_id, kid, base_url = ( + ("", "", "") + if config is None + else (str(config["org_id"]), str(config["kid"]), str(config["base_url"])) + ) return ( generate_guard_script(agent_key, agent_id), generate_hook_script( - org_id="", + org_id=org_id, agent_id=agent_id, - kid="", - base_url="", + kid=kid, + base_url=base_url, native_payload=True, agent_name=agent_key, + **hook_options, ), ) @@ -270,7 +285,7 @@ def runtime_contract_exists( contract: RuntimeContractLike, agent_key: str, product_label: str, - expected_scripts: Optional[Tuple[str, str]] = None, + expected_scripts: Optional[ExpectedScripts] = None, ) -> bool: """Strict status check: valid config and key plus runtime scripts.""" root = runtime_root() @@ -296,7 +311,8 @@ def runtime_contract_exists( validate_private_key(key.contents, "Elydora private key") 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] + expected_guard, expected_audit = expected_scripts(contract.agent_id, config) + return guard.contents == expected_guard and audit.contents == expected_audit def runtime_present(contract: RuntimeContractLike, agent_key: str) -> bool: @@ -361,3 +377,12 @@ def rendered_source_change( next_source: Optional[str], ) -> Optional[FileChange]: return source_change(file_path, label, original, next_source, 0o600, MAX_SOURCE_BYTES) + + +def unique_preconditions( + values: Sequence[FilePrecondition], +) -> Tuple[FilePrecondition, ...]: + result: Dict[str, FilePrecondition] = {} + for value in values: + result.setdefault(os.path.normcase(os.path.abspath(value.file_path)), value) + return tuple(result.values()) diff --git a/elydora/plugins/cline_io.py b/elydora/plugins/cline_io.py index bf01f06..7b4ceef 100644 --- a/elydora/plugins/cline_io.py +++ b/elydora/plugins/cline_io.py @@ -3,9 +3,11 @@ from __future__ import annotations import os +from typing import Tuple from ._managed_files import physical_directory_exists, read_physical_file from ._runtime import expected_runtime_scripts, runtime_contract_exists +from ._strict_json import JsonObject from .cline_contract import AGENT_KEY, HookFile, RuntimeContract, parse_metadata @@ -35,7 +37,9 @@ def require_available_hook_file(file: HookFile) -> None: ) +def _expected_scripts(agent_id: str, _config: JsonObject) -> Tuple[str, str]: + return expected_runtime_scripts(AGENT_KEY, agent_id) + + def runtime_files_exist(contract: RuntimeContract) -> bool: - return runtime_contract_exists( - contract, AGENT_KEY, "Cline", expected_runtime_scripts(AGENT_KEY, contract.agent_id) - ) + return runtime_contract_exists(contract, AGENT_KEY, "Cline", _expected_scripts) diff --git a/elydora/plugins/copilot_io.py b/elydora/plugins/copilot_io.py index d0ee093..418c67f 100644 --- a/elydora/plugins/copilot_io.py +++ b/elydora/plugins/copilot_io.py @@ -205,13 +205,17 @@ def require_hooks_enabled(sources: CopilotSources) -> None: ) +def _expected_scripts(agent_id: str, _config: JsonObject) -> Tuple[str, str]: + return expected_runtime_scripts(AGENT_KEY, agent_id) + + def runtime_files_exist(contracts: List[RuntimeContract]) -> bool: return any( runtime_contract_exists( contract, AGENT_KEY, "Copilot", - expected_runtime_scripts(AGENT_KEY, contract.agent_id), + _expected_scripts, ) for contract in contracts ) diff --git a/elydora/plugins/droid_io.py b/elydora/plugins/droid_io.py index e7a973a..6a1e047 100644 --- a/elydora/plugins/droid_io.py +++ b/elydora/plugins/droid_io.py @@ -4,10 +4,11 @@ from __future__ import annotations from dataclasses import dataclass import os -from typing import List, Optional +from typing import List, Optional, Tuple from ._managed_files import MAX_SOURCE_BYTES, physical_directory_exists, read_physical_file from ._runtime import expected_runtime_scripts, runtime_contract_exists +from ._strict_json import JsonObject from .droid_config import ( DroidDocument, DroidSources, @@ -109,13 +110,17 @@ def require_hooks_enabled(sources: DroidSources) -> None: ) +def _expected_scripts(agent_id: str, _config: JsonObject) -> Tuple[str, str]: + return expected_runtime_scripts(AGENT_KEY, agent_id) + + 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), + _expected_scripts, ) for contract in contracts ) diff --git a/elydora/plugins/grok_command.py b/elydora/plugins/grok_command.py index 66d3fae..f81e38d 100644 --- a/elydora/plugins/grok_command.py +++ b/elydora/plugins/grok_command.py @@ -2,16 +2,23 @@ from __future__ import annotations -import base64 from dataclasses import dataclass -import ntpath import os import re import shlex -import subprocess # nosec B404 - used only for legacy command rendering +import subprocess import sys from typing import Optional, Tuple +from ._runtime import managed_script_reference +from ._shell_command import ( + encoded_windows_command, + is_python_executable, + parse_encoded_windows_command, + parse_posix_command, + posix_source, +) + @dataclass(frozen=True) class GrokRuntimeReference: @@ -19,154 +26,14 @@ class GrokRuntimeReference: script_path: str -@dataclass(frozen=True) -class _ParsedArgument: - value: str - next_index: int - - -def runtime_root() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - -def same_grok_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_grok_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - -def _quote_posix(value: str) -> str: - return "'" + value.replace("'", "'\"'\"'") + "'" - - -def _quote_powershell(value: str) -> str: - return "'" + value.replace("'", "''") + "'" - - -def _windows_powershell_path() -> str: - configured = os.environ.get("SystemRoot") if os.name == "nt" else None - system_root = ( - configured - if configured - and ntpath.isabs(configured) - and re.search(r'["%\r\n]', configured) is None - else r"C:\Windows" - ) - return ntpath.join( - system_root, - "System32", - "WindowsPowerShell", - "v1.0", - "powershell.exe", - ) - - -def _windows_command(script_path: str) -> str: - source = ( - f"& {_quote_powershell(sys.executable)} " - f"{_quote_powershell(script_path)}; exit $LASTEXITCODE" - ) - encoded = base64.b64encode(source.encode("utf-16le")).decode("ascii") - return ( - f'"{_windows_powershell_path()}" -NoLogo -NoProfile ' - f"-NonInteractive -EncodedCommand {encoded}" - ) - - def build_grok_command(script_path: str) -> str: if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): raise ValueError( "Grok hook commands require absolute executable and script paths" ) if os.name == "nt": - return _windows_command(script_path) - return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" - - -def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - apostrophe = "'\"'\"'" - value = "" - index = start + 1 - while index < len(command): - if command.startswith(apostrophe, index): - value += "'" - index += len(apostrophe) - continue - if command[index] == "'": - return _ParsedArgument(value, index + 1) - value += command[index] - index += 1 - return None - - -def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: - executable = _read_posix_argument(command, 0) - if executable is None or command[executable.next_index : executable.next_index + 1] != " ": - return None - script = _read_posix_argument(command, executable.next_index + 1) - if script is None or script.next_index != len(command): - return None - return executable.value, script.value - - -def _read_powershell_argument( - command: str, start: int -) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - value = "" - index = start + 1 - while index < len(command): - if command[index] != "'": - value += command[index] - index += 1 - continue - if index + 1 < len(command) and command[index + 1] == "'": - value += "'" - index += 2 - continue - return _ParsedArgument(value, index + 1) - return None - - -def _parse_powershell_source(source: str) -> Optional[Tuple[str, str]]: - if not source.startswith("& "): - return None - executable = _read_powershell_argument(source, 2) - if executable is None or source[executable.next_index : executable.next_index + 1] != " ": - return None - script = _read_powershell_argument(source, executable.next_index + 1) - if script is None or source[script.next_index :] != "; exit $LASTEXITCODE": - return None - return executable.value, script.value - - -def _parse_windows_command(command: str) -> Optional[Tuple[str, str]]: - match = re.fullmatch( - r'"([^"\r\n]+)" -NoLogo -NoProfile -NonInteractive ' - r"-EncodedCommand ([A-Za-z0-9+/]+={0,2})", - command, - ) - if ( - match is None - or not ntpath.isabs(match.group(1)) - or ntpath.basename(match.group(1)).lower() != "powershell.exe" - ): - return None - try: - raw = base64.b64decode(match.group(2), validate=True) - if base64.b64encode(raw).decode("ascii") != match.group(2): - return None - source = raw.decode("utf-16le") - except (UnicodeDecodeError, ValueError): - return None - return _parse_powershell_source(source) + return encoded_windows_command(script_path) + return posix_source(script_path) def _parse_legacy_posix_command(command: str) -> Optional[Tuple[str, str]]: @@ -193,40 +60,19 @@ def _parse_legacy_windows_command(command: str) -> 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.IGNORECASE, - ) - is not None - ) - - def grok_runtime_reference( command: str, script_name: str ) -> Optional[GrokRuntimeReference]: parsed = ( - _parse_posix_command(command) - or _parse_windows_command(command) + parse_posix_command(command) + or parse_encoded_windows_command(command) or _parse_legacy_posix_command(command) or _parse_legacy_windows_command(command) ) if parsed is None: return None executable, script_path = parsed - if ( - not os.path.isabs(executable) - or not os.path.isabs(script_path) - or not _is_python_executable(executable) - or os.path.basename(script_path) != script_name - ): + if not os.path.isabs(executable) or not is_python_executable(executable): return None - agent_directory = os.path.dirname(script_path) - if not same_grok_path(os.path.dirname(agent_directory), runtime_root()): - return None - agent_id = os.path.basename(agent_directory) - if agent_id in {"", ".", ".."}: - return None - return GrokRuntimeReference(agent_id, script_path) + reference = managed_script_reference(script_path, script_name) + return None if reference is None else GrokRuntimeReference(*reference) diff --git a/elydora/plugins/grok_contract.py b/elydora/plugins/grok_contract.py index 4076dcf..836da08 100644 --- a/elydora/plugins/grok_contract.py +++ b/elydora/plugins/grok_contract.py @@ -7,13 +7,12 @@ import json import os from typing import Any, Dict, List, Optional, Tuple +from ._runtime import same_agent_id, same_path from ._strict_json import JsonObject, parse_json_object from .grok_command import ( GrokRuntimeReference, build_grok_command, grok_runtime_reference, - same_grok_agent_id, - same_grok_path, ) @@ -183,7 +182,7 @@ def _remove_from_groups( for handler in group["hooks"]: reference = _managed_reference(handler, script_name) remove = reference is not None and ( - not agent_id or same_grok_agent_id(reference.agent_id, agent_id) + not agent_id or same_agent_id(reference.agent_id, agent_id) ) if not remove: kept.append(handler) @@ -286,7 +285,7 @@ def grok_runtime_contracts(hooks: GrokHooks) -> List[GrokRuntimeContract]: failure = failures.get(key, []) if len(guard) != 1 or len(success) != 1 or len(failure) != 1: continue - if not same_grok_path(success[0].script_path, failure[0].script_path): + if not same_path(success[0].script_path, failure[0].script_path): continue contracts.append(GrokRuntimeContract( agent_id=guard[0].agent_id, diff --git a/elydora/plugins/grok_installation.py b/elydora/plugins/grok_installation.py index 1c429fe..a9315ea 100644 --- a/elydora/plugins/grok_installation.py +++ b/elydora/plugins/grok_installation.py @@ -2,169 +2,56 @@ 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, source_change, write_changes -from .base import InstallConfig -from .grok_command import runtime_root, same_grok_path -from .grok_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - GrokDocument, - RenderedGrokDocument, -) -from .grok_io import ( - validate_api_origin, - validate_private_key, +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 .grok_contract import AGENT_KEY, GrokDocument, RenderedGrokDocument +from .grok_io import rendered_change from .guard_template import generate_guard_script from .hook_template import generate_hook_script - -@dataclass(frozen=True) -class GrokRuntimePaths: - agent_id: str - agent_directory: str - guard_path: str - audit_path: str - - -def _runtime_config(config: InstallConfig, agent_id: str) -> dict: - value = { - "org_id": config.get("org_id", ""), - "agent_id": agent_id, - "kid": config.get("kid", ""), - "base_url": config.get("base_url", "https://api.elydora.com"), - "agent_name": AGENT_KEY, - } - token = config.get("token") - if token: - value["token"] = token - return value - - -def _json_source(value: dict) -> str: - return json.dumps(value, indent=2) + "\n" - - -def _agent_paths(config: InstallConfig) -> GrokRuntimePaths: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - agent_directory = resolve_agent_directory(runtime_root(), agent_id) - guard_path = os.path.join(agent_directory, GUARD_SCRIPT) - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - if not same_grok_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return GrokRuntimePaths(agent_id, agent_directory, guard_path, audit_path) - - -def _validate_install_config(config: InstallConfig) -> None: - for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): - value = config.get(field) - if not isinstance(value, str) or not value: - raise ValueError(f"{field} is required") - if config.get("agent_name") != AGENT_KEY: - raise ValueError(f"Grok installation requires agent_name {AGENT_KEY}") - validate_private_key(config["private_key"]) - validate_api_origin(config["base_url"]) - if "token" in config: - token = config["token"] - if not isinstance(token, str) or not token: - raise ValueError("token must be a non-empty string when provided") - _agent_paths(config) +PRODUCT = "Grok" def preflight_grok_installation( config: InstallConfig, document: GrokDocument -) -> GrokRuntimePaths: +) -> RuntimePaths: if not document.config_path: raise ValueError("Grok installation requires a hook configuration path") - _validate_install_config(config) - paths = _agent_paths(config) - validate_runtime_tree(paths.agent_directory, paths.agent_id) + 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 rendered_change( - rendered: RenderedGrokDocument, -) -> Optional[FileChange]: - if not rendered.changed: - return None - return source_change( - rendered.document.config_path, - "Grok user hooks", - rendered.document.raw, - rendered.next_source, - 0o600, - MAX_SOURCE_BYTES, - ) - - -def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - def prepare_grok_installation( config: InstallConfig, - paths: GrokRuntimePaths, + paths: RuntimePaths, rendered: RenderedGrokDocument, ) -> List[FileChange]: - guard_script = generate_guard_script( - AGENT_KEY, paths.agent_id, deny_protocol="grok" - ) + guard_script = generate_guard_script(AGENT_KEY, paths.agent_id, deny_protocol="grok") audit_script = generate_hook_script( org_id=config.get("org_id", ""), agent_id=paths.agent_id, kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), + 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_grok_installation(changes: List[FileChange]) -> None: @@ -172,7 +59,7 @@ def commit_grok_installation(changes: List[FileChange]) -> None: def prepare_grok_uninstall(rendered: RenderedGrokDocument) -> List[FileChange]: - return _present([rendered_change(rendered)]) + return present([rendered_change(rendered)]) def commit_grok_uninstall(changes: List[FileChange]) -> None: diff --git a/elydora/plugins/grok_io.py b/elydora/plugins/grok_io.py index ed188cd..9020a61 100644 --- a/elydora/plugins/grok_io.py +++ b/elydora/plugins/grok_io.py @@ -2,24 +2,17 @@ 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 .grok_command import runtime_root, same_grok_agent_id, same_grok_path +from ._managed_files import physical_directory_exists, read_physical_file +from ._runtime import rendered_source_change, runtime_contract_exists +from ._transaction import FileChange from .grok_contract import ( AGENT_KEY, GrokDocument, GrokRuntimeContract, + RenderedGrokDocument, create_grok_document, parse_grok_document, ) @@ -41,9 +34,7 @@ def grok_config_path() -> str: def read_grok_document() -> GrokDocument: config_path = grok_config_path() hooks_directory = os.path.dirname(config_path) - physical_directory_exists( - os.path.dirname(hooks_directory), "Grok home directory" - ) + physical_directory_exists(os.path.dirname(hooks_directory), "Grok home directory") physical_directory_exists(hooks_directory, "Grok hooks directory") snapshot = read_physical_file(config_path, "Grok user hooks") if snapshot is None: @@ -51,168 +42,16 @@ def read_grok_document() -> GrokDocument: return parse_grok_document(config_path, snapshot.contents) -def validate_api_origin(value: str, label: str = "base_url") -> None: - try: - parsed = urllib.parse.urlsplit(value) - hostname = parsed.hostname - parsed.port - except ValueError as error: - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error - invalid_character = "\\" in value or any( - character.isspace() or ord(character) < 32 for character in value - ) - if ( - parsed.scheme not in ("http", "https") - or not parsed.netloc - or hostname is None - or invalid_character - ): - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") - if ( - parsed.username is not None - or parsed.password is not None - or parsed.query - or parsed.fragment - ): - raise ValueError( - f"{label} must exclude credentials, query parameters, and fragments" - ) - - -def validate_private_key(value: str, label: str = "private_key") -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), - validate=True, - ) - canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - f"{label} must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or canonical != value: - raise ValueError(f"{label} must be a canonical 32-byte base64url value") - - -def _require_non_empty_string(value: Any, field: str, config_path: str) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError( - f"Elydora runtime config {field} is invalid: {config_path}" - ) - return value - - -def _validate_runtime_config( - config: JsonObject, expected_agent_id: str, config_path: str -) -> None: - supported = {"org_id", "agent_id", "kid", "base_url", "token", "agent_name"} - extra = next((key for key in config if key not in supported), None) - if extra is not None: - raise ValueError( - f'Elydora runtime config has unsupported field "{extra}": {config_path}' - ) - _require_non_empty_string(config.get("org_id"), "org_id", config_path) - _require_non_empty_string(config.get("kid"), "kid", config_path) - agent_id = _require_non_empty_string( - config.get("agent_id"), "agent_id", config_path - ) - if ( - not same_grok_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - f"Elydora runtime identity does not match Grok hooks: {config_path}" - ) - if "token" in config: - _require_non_empty_string(config.get("token"), "token", config_path) - base_url = _require_non_empty_string( - config.get("base_url"), "base_url", config_path - ) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, "Elydora runtime config", MAX_CONFIG_BYTES - ) - if snapshot is None: +def rendered_change(rendered: RenderedGrokDocument) -> Optional[FileChange]: + if not rendered.changed: return None - return parse_json_object( - snapshot.contents, f"Elydora runtime config at {file_path}" + return rendered_source_change( + rendered.document.config_path, + "Grok user hooks", + rendered.document.raw, + rendered.next_source, ) -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = runtime_root() - if not physical_directory_exists(root, "Elydora runtime directory"): - return - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return - config_path = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path) - artifacts = ( - ("private.key", "Elydora private key"), - ("guard.py", "Elydora guard runtime"), - ("hook.py", "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ) - artifact_states = [ - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ] - artifact_exists = any(artifact_states) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not same_grok_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - f"Elydora runtime config identity does not match Grok agent {agent_id}: " - f"{config_path}" - ) - - -def _runtime_contract_exists(contract: GrokRuntimeContract) -> bool: - root = runtime_root() - agent_directory = os.path.dirname(contract.guard_path) - if ( - not same_grok_path(os.path.dirname(agent_directory), root) - or not same_grok_path( - contract.audit_path, os.path.join(agent_directory, "hook.py") - ) - ): - return False - if not physical_directory_exists(root, "Elydora runtime directory"): - return False - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return False - config_path = os.path.join(agent_directory, "config.json") - key_path = os.path.join(agent_directory, "private.key") - config = _read_runtime_config(config_path) - key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) - guard = read_physical_file(contract.guard_path, "Elydora guard runtime") - audit = read_physical_file(contract.audit_path, "Elydora audit runtime") - if config is None or key is None or guard is None or audit is None: - return False - _validate_runtime_config(config, contract.agent_id, config_path) - validate_private_key(key.contents, "Elydora private key") - return bool(guard.contents) and bool(audit.contents) - - def grok_runtime_files_exist(contracts: List[GrokRuntimeContract]) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) + return any(runtime_contract_exists(contract, AGENT_KEY, "Grok") for contract in contracts) diff --git a/elydora/plugins/kimi_command.py b/elydora/plugins/kimi_command.py index 9a66e92..7b62d58 100644 --- a/elydora/plugins/kimi_command.py +++ b/elydora/plugins/kimi_command.py @@ -2,16 +2,23 @@ from __future__ import annotations -import base64 from dataclasses import dataclass -import ntpath import os import re import shlex -import subprocess # nosec B404 - used only for legacy command rendering +import subprocess import sys from typing import Optional, Tuple +from ._runtime import managed_script_reference +from ._shell_command import ( + encoded_windows_command, + is_python_executable, + parse_encoded_windows_command, + parse_posix_command, + posix_source, +) + @dataclass(frozen=True) class KimiRuntimeReference: @@ -19,154 +26,14 @@ class KimiRuntimeReference: script_path: str -@dataclass(frozen=True) -class _ParsedArgument: - value: str - next_index: int - - -def runtime_root() -> str: - return os.path.join(os.path.expanduser("~"), ".elydora") - - -def same_kimi_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_kimi_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_kimi_command(script_path: str) -> str: if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): raise ValueError( "Kimi hook commands require absolute executable and script paths" ) if os.name == "nt": - return _windows_command(script_path) - return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" - - -def _read_posix_argument(command: str, start: int) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - apostrophe = "'\"'\"'" - value = "" - index = start + 1 - while index < len(command): - if command.startswith(apostrophe, index): - value += "'" - index += len(apostrophe) - continue - if command[index] == "'": - return _ParsedArgument(value, index + 1) - value += command[index] - index += 1 - return None - - -def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: - executable = _read_posix_argument(command, 0) - if executable is None or command[executable.next_index : executable.next_index + 1] != " ": - return None - script = _read_posix_argument(command, executable.next_index + 1) - if script is None or script.next_index != len(command): - return None - return executable.value, script.value - - -def _read_powershell_argument( - command: str, start: int -) -> Optional[_ParsedArgument]: - if start >= len(command) or command[start] != "'": - return None - value = "" - index = start + 1 - while index < len(command): - if command[index] != "'": - value += command[index] - index += 1 - continue - if index + 1 < len(command) and command[index + 1] == "'": - value += "'" - index += 2 - continue - return _ParsedArgument(value, index + 1) - return None - - -def _parse_powershell_source(source: str) -> Optional[Tuple[str, str]]: - if not source.startswith("& "): - return None - executable = _read_powershell_argument(source, 2) - if executable is None or source[executable.next_index : executable.next_index + 1] != " ": - return None - script = _read_powershell_argument(source, executable.next_index + 1) - if script is None or source[script.next_index :] != "; exit $LASTEXITCODE": - return None - return executable.value, script.value - - -def _parse_windows_command(command: str) -> Optional[Tuple[str, str]]: - match = re.fullmatch( - r'"([^"\r\n]+)" -NoLogo -NoProfile -NonInteractive ' - r"-EncodedCommand ([A-Za-z0-9+/]+={0,2})", - command, - ) - if ( - match is None - or not ntpath.isabs(match.group(1)) - or ntpath.basename(match.group(1)).lower() != "powershell.exe" - ): - return None - try: - raw = base64.b64decode(match.group(2), validate=True) - if base64.b64encode(raw).decode("ascii") != match.group(2): - return None - source = raw.decode("utf-16le") - except (UnicodeDecodeError, ValueError): - return None - return _parse_powershell_source(source) + return encoded_windows_command(script_path) + return posix_source(script_path) def _parse_legacy_posix_command(command: str) -> Optional[Tuple[str, str]]: @@ -193,40 +60,19 @@ def _parse_legacy_windows_command(command: str) -> 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.IGNORECASE, - ) - is not None - ) - - def kimi_runtime_reference( command: str, script_name: str ) -> Optional[KimiRuntimeReference]: parsed = ( - _parse_posix_command(command) - or _parse_windows_command(command) + parse_posix_command(command) + or parse_encoded_windows_command(command) or _parse_legacy_posix_command(command) or _parse_legacy_windows_command(command) ) if parsed is None: return None executable, script_path = parsed - if ( - not os.path.isabs(executable) - or not os.path.isabs(script_path) - or not _is_python_executable(executable) - or os.path.basename(script_path) != script_name - ): + if not os.path.isabs(executable) or not is_python_executable(executable): return None - agent_directory = os.path.dirname(script_path) - if not same_kimi_path(os.path.dirname(agent_directory), runtime_root()): - return None - agent_id = os.path.basename(agent_directory) - if agent_id in {"", ".", ".."}: - return None - return KimiRuntimeReference(agent_id, script_path) + reference = managed_script_reference(script_path, script_name) + return None if reference is None else KimiRuntimeReference(*reference) diff --git a/elydora/plugins/kimi_contract.py b/elydora/plugins/kimi_contract.py index 2280a35..406edff 100644 --- a/elydora/plugins/kimi_contract.py +++ b/elydora/plugins/kimi_contract.py @@ -10,12 +10,11 @@ from typing import Any, Dict, FrozenSet, List, Optional, Tuple from tomlkit import aot, document, dumps, inline_table, parse, table from tomlkit.items import AoT, Array +from ._runtime import same_agent_id, same_path from .kimi_command import ( KimiRuntimeReference, build_kimi_command, kimi_runtime_reference, - same_kimi_agent_id, - same_kimi_path, ) @@ -216,7 +215,7 @@ def remove_managed_kimi_hooks( else None ) remove = reference is not None and ( - not agent_id or same_kimi_agent_id(reference.agent_id, agent_id) + not agent_id or same_agent_id(reference.agent_id, agent_id) ) if not remove: result.append(hook) @@ -318,7 +317,7 @@ def kimi_runtime_contracts( failure = failures.get(key, []) if len(guard) != 1 or len(success) != 1 or len(failure) != 1: continue - if not same_kimi_path(success[0].script_path, failure[0].script_path): + if not same_path(success[0].script_path, failure[0].script_path): continue contracts.append( KimiRuntimeContract( diff --git a/elydora/plugins/kimi_installation.py b/elydora/plugins/kimi_installation.py index 07d8d80..87c6a51 100644 --- a/elydora/plugins/kimi_installation.py +++ b/elydora/plugins/kimi_installation.py @@ -2,121 +2,39 @@ from __future__ import annotations -from dataclasses import dataclass -import json -import os from typing import List, Optional -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import FileChange, file_change, source_change, write_changes +from ._managed_files import 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, source_change, write_changes from .base import InstallConfig from .guard_template import generate_guard_script from .hook_template import generate_hook_script -from .kimi_command import runtime_root, same_kimi_path -from .kimi_contract import ( - AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - KimiDocument, - RenderedKimiDocument, -) -from .kimi_io import ( - validate_api_origin, - validate_private_key, - validate_runtime_tree, -) +from .kimi_contract import AGENT_KEY, KimiDocument, RenderedKimiDocument - -@dataclass(frozen=True) -class KimiRuntimePaths: - 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) -> KimiRuntimePaths: - 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_kimi_path(configured_guard, guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return KimiRuntimePaths(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"Kimi 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 = "Kimi" def preflight_kimi_installation( config: InstallConfig, documents: List[KimiDocument] -) -> KimiRuntimePaths: +) -> RuntimePaths: if not documents: raise ValueError("Kimi installation requires at least one hook contract") - _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 rendered_change( - rendered: RenderedKimiDocument, -) -> Optional[FileChange]: +def rendered_change(rendered: RenderedKimiDocument) -> Optional[FileChange]: if not rendered.changed: return None source = rendered.document @@ -130,66 +48,32 @@ def rendered_change( ) -def _present(changes: List[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - def prepare_kimi_installation( config: InstallConfig, - paths: KimiRuntimePaths, + paths: RuntimePaths, rendered: List[RenderedKimiDocument], ) -> List[FileChange]: - config_changes = _present([rendered_change(item) for item in rendered]) guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) audit_script = generate_hook_script( org_id=config.get("org_id", ""), agent_id=paths.agent_id, kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), + base_url=config.get("base_url", DEFAULT_BASE_URL), native_payload=True, agent_name=AGENT_KEY, ) - runtime_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, - ), - ] - ) - return [*runtime_changes, *config_changes] + return [ + *runtime_file_changes(config, paths, AGENT_KEY, guard_script, audit_script), + *present([rendered_change(item) for item in rendered]), + ] def commit_kimi_installation(changes: List[FileChange]) -> None: write_changes(changes, "Install Kimi hooks") -def prepare_kimi_uninstall( - rendered: List[RenderedKimiDocument], -) -> List[FileChange]: - return _present([rendered_change(item) for item in rendered]) +def prepare_kimi_uninstall(rendered: List[RenderedKimiDocument]) -> List[FileChange]: + return present([rendered_change(item) for item in rendered]) def commit_kimi_uninstall(changes: List[FileChange]) -> None: diff --git a/elydora/plugins/kimi_io.py b/elydora/plugins/kimi_io.py index 6a0f4b4..26aec01 100644 --- a/elydora/plugins/kimi_io.py +++ b/elydora/plugins/kimi_io.py @@ -2,24 +2,11 @@ from __future__ import annotations -import base64 import os -from typing import Any, List, Optional -import urllib.parse +from typing import List -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 .kimi_command import ( - runtime_root, - same_kimi_agent_id, - same_kimi_path, -) +from ._managed_files import physical_directory_exists, read_physical_file +from ._runtime import runtime_contract_exists, same_path from .kimi_contract import ( AGENT_KEY, LEGACY_EVENTS, @@ -76,7 +63,7 @@ def resolve_kimi_contracts() -> List[KimiContract]: legacy_home = os.path.join(home, ".kimi") stable = _stable_contract(os.path.join(stable_home, "config.toml")) legacy = _legacy_contract(os.path.join(legacy_home, "config.toml")) - if same_kimi_path(stable.config_path, legacy.config_path): + if same_path(stable.config_path, legacy.config_path): return [stable] stable_detected = explicit_home is not None or _entry_exists( @@ -101,176 +88,5 @@ def read_kimi_documents() -> List[KimiDocument]: return [_read_kimi_document(contract) for contract in resolve_kimi_contracts()] -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_kimi_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - f"Elydora runtime identity does not match Kimi hooks: {config_path}" - ) - if "token" in config: - _require_non_empty_string(config.get("token"), "token", config_path) - base_url = _require_non_empty_string( - config.get("base_url"), "base_url", config_path - ) - validate_api_origin(base_url, "Elydora runtime config base_url") - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, - "Elydora runtime config", - MAX_CONFIG_BYTES, - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, - f"Elydora runtime config at {file_path}", - ) - - -def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: - root = runtime_root() - if not physical_directory_exists(root, "Elydora runtime directory"): - return - if not physical_directory_exists( - agent_directory, "Elydora agent runtime directory" - ): - return - config_path = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path) - artifacts = [ - ("private.key", "Elydora private key"), - ("guard.py", "Elydora guard runtime"), - ("hook.py", "Elydora audit runtime"), - ("chain-state.json", "Elydora chain state"), - ("status-cache.json", "Elydora status cache"), - ("error.log", "Elydora error log"), - ] - artifact_states = [ - physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in artifacts - ] - artifact_exists = any(artifact_states) - if config is None: - if artifact_exists: - raise ValueError( - "Elydora runtime identity cannot be verified without config.json: " - f"{agent_directory}" - ) - return - configured_id = config.get("agent_id") - if ( - not isinstance(configured_id, str) - or not same_kimi_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - f"Elydora runtime config identity does not match Kimi agent {agent_id}: " - f"{config_path}" - ) - - -def _runtime_contract_exists(contract: KimiRuntimeContract) -> bool: - root = runtime_root() - agent_directory = os.path.dirname(contract.guard_path) - if ( - not same_kimi_path(os.path.dirname(agent_directory), root) - or not same_kimi_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 kimi_runtime_files_exist(contracts: List[KimiRuntimeContract]) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) + return any(runtime_contract_exists(contract, AGENT_KEY, "Kimi") for contract in contracts) diff --git a/elydora/plugins/letta.py b/elydora/plugins/letta.py index 58cc589..8fbceaa 100644 --- a/elydora/plugins/letta.py +++ b/elydora/plugins/letta.py @@ -4,6 +4,7 @@ from __future__ import annotations from typing import Dict +from ._runtime import RuntimePaths from .base import AgentPlugin, InstallConfig, PluginStatus from .letta_config import render_letta_document from .letta_contract import ( @@ -13,7 +14,6 @@ from .letta_contract import ( letta_runtime_contracts, ) from .letta_installation import ( - LettaRuntimePaths, commit_letta_installation, commit_letta_uninstall, preflight_letta_installation, @@ -24,7 +24,7 @@ from .letta_io import letta_runtime_files_exist from .letta_sources import read_letta_sources -def _installed_groups(paths: LettaRuntimePaths) -> Dict[str, JsonObject]: +def _installed_groups(paths: RuntimePaths) -> Dict[str, JsonObject]: return { "PreToolUse": build_letta_group(paths.guard_path), "PostToolUse": build_letta_group(paths.audit_path), diff --git a/elydora/plugins/letta_command.py b/elydora/plugins/letta_command.py index 4f557cd..50e9454 100644 --- a/elydora/plugins/letta_command.py +++ b/elydora/plugins/letta_command.py @@ -4,11 +4,17 @@ from __future__ import annotations from dataclasses import dataclass import os -import re import sys -from typing import Optional, Tuple +from typing import Optional -from elydora._runtime_paths import runtime_root +from ._runtime import managed_script_reference +from ._shell_command import ( + is_python_executable, + parse_posix_command, + parse_powershell_source, + posix_source, + powershell_source, +) @dataclass(frozen=True) @@ -18,103 +24,12 @@ class LettaRuntimeReference: executable_path: Optional[str] = None -def same_letta_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_letta_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_letta_command(script_path: str) -> str: if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): raise ValueError( "Letta Code hook commands require absolute executable and script paths" ) - invocation = ( - f"{_quote_powershell(sys.executable)} {_quote_powershell(script_path)}" - if os.name == "nt" - else f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" - ) - return f"& {invocation}; exit $LASTEXITCODE" if os.name == "nt" else invocation - - -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_command(command: str) -> Optional[Tuple[str, str]]: - if not command.startswith("& "): - return None - executable = _read_powershell_argument(command, 2) - if executable is None or executable[1] >= len(command): - return None - if command[executable[1]] != " ": - return None - script = _read_powershell_argument(command, executable[1] + 1) - if script is None or command[script[1] :] != "; exit $LASTEXITCODE": - return None - return executable[0], script[0] - - -def _is_python_executable(file_path: str) -> bool: - return bool(re.fullmatch(r"python(?:3(?:\.\d+)?)?(?:\.exe)?", os.path.basename(file_path), re.IGNORECASE)) + return powershell_source(script_path) if os.name == "nt" else posix_source(script_path) def _runtime_reference( @@ -122,30 +37,21 @@ def _runtime_reference( script_name: str, executable_path: Optional[str] = None, ) -> Optional[LettaRuntimeReference]: - if not os.path.isabs(script_path) or os.path.basename(script_path) != script_name: + reference = managed_script_reference(script_path, script_name) + if reference is None: return None - agent_directory = os.path.dirname(script_path) - if not same_letta_path(os.path.dirname(agent_directory), runtime_root()): - return None - agent_id = os.path.basename(agent_directory) - if agent_id in {"", ".", ".."}: - return None - return LettaRuntimeReference(agent_id, script_path, executable_path) + return LettaRuntimeReference(reference[0], reference[1], executable_path) def letta_runtime_reference( command: str, script_name: str ) -> Optional[LettaRuntimeReference]: parsed = ( - _parse_powershell_command(command) + parse_powershell_source(command) if os.name == "nt" - else _parse_posix_command(command) + else parse_posix_command(command) ) - if ( - parsed is None - or not os.path.isabs(parsed[0]) - or not _is_python_executable(parsed[0]) - ): + if parsed is None or not os.path.isabs(parsed[0]) or not is_python_executable(parsed[0]): return None return _runtime_reference(parsed[1], script_name, parsed[0]) @@ -160,11 +66,7 @@ def letta_legacy_guard_reference( return None executable = command[1:closing] script_path = command[closing + 2 :] - if ( - not os.path.isabs(executable) - or not _is_python_executable(executable) - or not script_path - ): + if not os.path.isabs(executable) or not is_python_executable(executable) or not script_path: return None return _runtime_reference(script_path, script_name, executable) diff --git a/elydora/plugins/letta_contract.py b/elydora/plugins/letta_contract.py index f9344db..57a3743 100644 --- a/elydora/plugins/letta_contract.py +++ b/elydora/plugins/letta_contract.py @@ -8,16 +8,13 @@ import os import sys from typing import Any, Dict, List, Optional, Tuple -from elydora._runtime_paths import runtime_root - +from ._runtime import same_agent_id, same_path from .letta_command import ( LettaRuntimeReference, build_letta_command, letta_legacy_audit_reference, letta_legacy_guard_reference, letta_runtime_reference, - same_letta_agent_id, - same_letta_path, ) @@ -63,10 +60,6 @@ class ManagedRemoval: remove_group: bool -def elydora_dir() -> str: - return runtime_root() - - def _validate_timeout(value: JsonObject, label: str) -> None: if "timeout" not in value: return @@ -222,7 +215,7 @@ def managed_letta_removals( is not None and ( agent_id is None - or same_letta_agent_id(reference.agent_id, agent_id) + or same_agent_id(reference.agent_id, agent_id) ) ) if indexes: @@ -247,7 +240,7 @@ def _references_for_event( if ( reference is None or reference.executable_path is None - or not same_letta_path(reference.executable_path, sys.executable) + or not same_path(reference.executable_path, sys.executable) ): continue key = os.path.normcase(reference.agent_id) @@ -267,7 +260,7 @@ def letta_runtime_contracts(hooks: LettaHooks) -> List[LettaRuntimeContract]: failure = failures.get(key, []) if len(guard) != 1 or len(post) != 1 or len(failure) != 1: continue - if not same_letta_path(post[0].script_path, failure[0].script_path): + if not same_path(post[0].script_path, failure[0].script_path): continue contracts.append(LettaRuntimeContract( guard[0].agent_id, diff --git a/elydora/plugins/letta_installation.py b/elydora/plugins/letta_installation.py index 47fa5bc..d7910eb 100644 --- a/elydora/plugins/letta_installation.py +++ b/elydora/plugins/letta_installation.py @@ -2,113 +2,41 @@ from __future__ import annotations -from dataclasses import dataclass -import json -import os -from typing import List, Optional, Sequence +from typing import List, Optional -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import ( - FileChange, - FilePrecondition, - file_change, - source_change, - write_changes, +from ._managed_files import MAX_SOURCE_BYTES +from ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + present, + 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 .guard_template import generate_guard_script from .hook_template import generate_hook_script -from .letta_command import same_letta_path from .letta_config import RenderedLettaDocument, letta_document_label -from .letta_contract import AGENT_KEY, AUDIT_SCRIPT, GUARD_SCRIPT, elydora_dir -from .letta_io import validate_api_origin, validate_private_key, validate_runtime_tree +from .letta_contract import AGENT_KEY from .letta_sources import LettaSources, require_letta_hooks_enabled - -@dataclass(frozen=True) -class LettaRuntimePaths: - 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) -> LettaRuntimePaths: - 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_letta_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return LettaRuntimePaths(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"Letta 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 = "Letta Code" def preflight_letta_installation( config: InstallConfig, sources: LettaSources -) -> LettaRuntimePaths: +) -> RuntimePaths: require_letta_hooks_enabled(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) + 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 _rendered_change( - rendered: RenderedLettaDocument, -) -> Optional[FileChange]: +def _rendered_change(rendered: RenderedLettaDocument) -> Optional[FileChange]: if not rendered.changed: return None return source_change( @@ -122,13 +50,9 @@ def _rendered_change( ) -def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - def prepare_letta_installation( config: InstallConfig, - paths: LettaRuntimePaths, + paths: RuntimePaths, rendered: RenderedLettaDocument, ) -> List[FileChange]: guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) @@ -136,50 +60,28 @@ def prepare_letta_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), fail_closed=True, 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 _read_only_preconditions( - sources: LettaSources, changed_path: Optional[str] + sources: LettaSources, changed_path: str ) -> List[FilePrecondition]: return [ condition for condition in sources.preconditions - if changed_path is None - or not same_letta_path(condition.file_path, changed_path) + if not same_path(condition.file_path, changed_path) ] -def commit_letta_installation( - changes: List[FileChange], sources: LettaSources -) -> None: +def commit_letta_installation(changes: List[FileChange], sources: LettaSources) -> None: write_changes( changes, "Install Letta Code hooks", @@ -187,15 +89,11 @@ def commit_letta_installation( ) -def prepare_letta_uninstall( - rendered: RenderedLettaDocument, -) -> Optional[FileChange]: +def prepare_letta_uninstall(rendered: RenderedLettaDocument) -> Optional[FileChange]: return _rendered_change(rendered) -def commit_letta_uninstall( - change: Optional[FileChange], sources: LettaSources -) -> None: +def commit_letta_uninstall(change: Optional[FileChange], sources: LettaSources) -> None: if change is None: return write_changes( diff --git a/elydora/plugins/letta_io.py b/elydora/plugins/letta_io.py index 5254fb8..ae782ed 100644 --- a/elydora/plugins/letta_io.py +++ b/elydora/plugins/letta_io.py @@ -2,200 +2,19 @@ from __future__ import annotations -import base64 -import os -from typing import Any, List, Optional -import urllib.parse +from typing import List, Tuple -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 .guard_template import generate_guard_script -from .hook_template import generate_hook_script -from .letta_command import same_letta_agent_id, same_letta_path -from .letta_contract import AGENT_KEY, LettaRuntimeContract, elydora_dir +from ._runtime import expected_runtime_scripts, runtime_contract_exists +from ._strict_json import JsonObject +from .letta_contract import AGENT_KEY, LettaRuntimeContract -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 _expected_scripts(agent_id: str, config: JsonObject) -> Tuple[str, str]: + return expected_runtime_scripts(AGENT_KEY, agent_id, config, fail_closed=True) + + +def letta_runtime_files_exist(contracts: List[LettaRuntimeContract]) -> bool: + return any( + runtime_contract_exists(contract, AGENT_KEY, "Letta Code", _expected_scripts) + 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_letta_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime identity does not match Letta 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 = 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_letta_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Letta Code agent " - f"{agent_id}: {config_path}" - ) - - -def _runtime_contract_exists(contract: LettaRuntimeContract) -> bool: - root = elydora_dir() - agent_directory = os.path.dirname(contract.guard_path) - if ( - not same_letta_path(os.path.dirname(agent_directory), root) - or not same_letta_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") - expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id) - expected_audit = generate_hook_script( - org_id=str(config["org_id"]), - agent_id=contract.agent_id, - kid=str(config["kid"]), - base_url=str(config["base_url"]), - fail_closed=True, - native_payload=True, - agent_name=AGENT_KEY, - ) - return guard.contents == expected_guard and audit.contents == expected_audit - - -def letta_runtime_files_exist( - contracts: List[LettaRuntimeContract], -) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/elydora/plugins/letta_sources.py b/elydora/plugins/letta_sources.py index b046c53..9ea6c51 100644 --- a/elydora/plugins/letta_sources.py +++ b/elydora/plugins/letta_sources.py @@ -4,15 +4,15 @@ from __future__ import annotations from dataclasses import dataclass import os -from typing import Dict, Optional, Tuple +from typing import Optional, Tuple from ._managed_files import ( MAX_SOURCE_BYTES, physical_directory_exists, read_physical_file, ) +from ._runtime import same_path, unique_preconditions from ._transaction import FilePrecondition -from .letta_command import same_letta_path from .letta_config import ( LettaDocument, create_letta_document, @@ -76,15 +76,6 @@ def _source_precondition(document: LettaDocument) -> FilePrecondition: ) -def _deduplicate_preconditions( - values: list[FilePrecondition], -) -> Tuple[FilePrecondition, ...]: - result: Dict[str, FilePrecondition] = {} - for value in values: - result.setdefault(_comparison_path(value.file_path), value) - return tuple(result.values()) - - def _effective_disable( global_settings: LettaDocument, project: LettaDocument, @@ -111,7 +102,7 @@ def read_letta_sources() -> LettaSources: physical_directory_exists( global_directory, "Letta Code global configuration directory" ) - if not same_letta_path(project_directory, global_directory): + if not same_path(project_directory, global_directory): physical_directory_exists( project_directory, "Letta Code project configuration directory" ) @@ -141,7 +132,7 @@ def read_letta_sources() -> LettaSources: _effective_disable( global_settings, project, project_local, project_active ), - _deduplicate_preconditions(preconditions), + unique_preconditions(preconditions), ) diff --git a/elydora/plugins/qwen.py b/elydora/plugins/qwen.py index 1ee145f..6d1b223 100644 --- a/elydora/plugins/qwen.py +++ b/elydora/plugins/qwen.py @@ -4,6 +4,7 @@ from __future__ import annotations from typing import Dict +from ._runtime import RuntimePaths from .base import AgentPlugin, InstallConfig, PluginStatus from .qwen_config import render_qwen_document from .qwen_contract import ( @@ -15,7 +16,6 @@ from .qwen_contract import ( qwen_runtime_contracts, ) from .qwen_installation import ( - QwenRuntimePaths, commit_qwen_installation, commit_qwen_uninstall, preflight_qwen_installation, @@ -26,7 +26,7 @@ from .qwen_io import qwen_runtime_files_exist from .qwen_sources import read_qwen_sources -def _installed_groups(paths: QwenRuntimePaths) -> Dict[str, JsonObject]: +def _installed_groups(paths: RuntimePaths) -> Dict[str, JsonObject]: return { "PreToolUse": build_qwen_group(paths.guard_path, GUARD_HOOK_NAME), "PostToolUse": build_qwen_group(paths.audit_path, AUDIT_HOOK_NAME), diff --git a/elydora/plugins/qwen_command.py b/elydora/plugins/qwen_command.py index 3462fc9..ae6005d 100644 --- a/elydora/plugins/qwen_command.py +++ b/elydora/plugins/qwen_command.py @@ -5,7 +5,15 @@ from __future__ import annotations from dataclasses import dataclass import os import sys -from typing import Optional, Tuple +from typing import Optional + +from ._runtime import managed_script_reference, same_path +from ._shell_command import ( + parse_posix_command, + parse_powershell_source, + posix_source, + powershell_source, +) @dataclass(frozen=True) @@ -14,121 +22,23 @@ class QwenRuntimeReference: script_path: str -def same_qwen_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_qwen_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_qwen_command(script_path: str) -> str: if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): raise ValueError( "Qwen Code hook commands require absolute executable and script paths" ) - invocation = ( - f"{_quote_powershell(sys.executable)} {_quote_powershell(script_path)}" - if os.name == "nt" - else f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" - ) - return f"& {invocation}; exit $LASTEXITCODE" if os.name == "nt" else invocation - - -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_command(command: str) -> Optional[Tuple[str, str]]: - if not command.startswith("& "): - return None - executable = _read_powershell_argument(command, 2) - if executable is None or executable[1] >= len(command): - return None - if command[executable[1]] != " ": - return None - script = _read_powershell_argument(command, executable[1] + 1) - if script is None or command[script[1] :] != "; exit $LASTEXITCODE": - return None - return executable[0], script[0] + return powershell_source(script_path) if os.name == "nt" else posix_source(script_path) def qwen_runtime_reference( command: str, script_name: str ) -> Optional[QwenRuntimeReference]: parsed = ( - _parse_powershell_command(command) + parse_powershell_source(command) if os.name == "nt" - else _parse_posix_command(command) + else parse_posix_command(command) ) - if ( - parsed is None - or not same_qwen_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_qwen_path(os.path.dirname(agent_directory), runtime_root): - return None - agent_id = os.path.basename(agent_directory) - if agent_id in {"", ".", ".."}: - return None - return QwenRuntimeReference(agent_id, parsed[1]) + reference = managed_script_reference(parsed[1], script_name) + return None if reference is None else QwenRuntimeReference(*reference) diff --git a/elydora/plugins/qwen_contract.py b/elydora/plugins/qwen_contract.py index 34e7857..1770065 100644 --- a/elydora/plugins/qwen_contract.py +++ b/elydora/plugins/qwen_contract.py @@ -10,12 +10,11 @@ import shutil import subprocess # nosec B404 from typing import Any, Dict, List, Optional, Sequence, Tuple +from ._runtime import same_agent_id, same_path from .qwen_command import ( QwenRuntimeReference, build_qwen_command, qwen_runtime_reference, - same_qwen_agent_id, - same_qwen_path, ) @@ -102,10 +101,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') @@ -352,7 +347,7 @@ def managed_qwen_removals( is not None and ( agent_id is None - or same_qwen_agent_id(reference.agent_id, agent_id) + or same_agent_id(reference.agent_id, agent_id) ) ) if indexes: @@ -398,7 +393,7 @@ def qwen_runtime_contracts(hooks: QwenHooks) -> List[QwenRuntimeContract]: failure = failures.get(key, []) if len(guard) != 1 or len(post) != 1 or len(failure) != 1: continue - if not same_qwen_path(post[0].script_path, failure[0].script_path): + if not same_path(post[0].script_path, failure[0].script_path): continue contracts.append(QwenRuntimeContract( guard[0].agent_id, diff --git a/elydora/plugins/qwen_installation.py b/elydora/plugins/qwen_installation.py index 759a436..07de3c8 100644 --- a/elydora/plugins/qwen_installation.py +++ b/elydora/plugins/qwen_installation.py @@ -2,107 +2,41 @@ from __future__ import annotations -from dataclasses import dataclass -import json -import os -from typing import List, Optional, Sequence +from typing import List, Optional -from elydora._runtime_paths import resolve_agent_directory - -from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES -from ._transaction import FileChange, FilePrecondition, file_change, source_change, write_changes +from ._managed_files import MAX_SOURCE_BYTES +from ._runtime import ( + DEFAULT_BASE_URL, + RuntimePaths, + present, + 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 .guard_template import generate_guard_script from .hook_template import generate_hook_script -from .qwen_command import same_qwen_path from .qwen_config import RenderedQwenDocument, qwen_document_label -from .qwen_contract import AGENT_KEY, AUDIT_SCRIPT, GUARD_SCRIPT, elydora_dir -from .qwen_io import validate_api_origin, validate_private_key, validate_runtime_tree +from .qwen_contract import AGENT_KEY from .qwen_sources import QwenSources, require_qwen_hooks_enabled - -@dataclass(frozen=True) -class QwenRuntimePaths: - 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) -> QwenRuntimePaths: - 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_qwen_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - return QwenRuntimePaths(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"Qwen 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 = "Qwen Code" def preflight_qwen_installation( config: InstallConfig, sources: QwenSources -) -> QwenRuntimePaths: +) -> RuntimePaths: require_qwen_hooks_enabled(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) + 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 _rendered_change( - rendered: RenderedQwenDocument, -) -> Optional[FileChange]: +def _rendered_change(rendered: RenderedQwenDocument) -> Optional[FileChange]: if not rendered.changed: return None return source_change( @@ -116,13 +50,9 @@ def _rendered_change( ) -def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - def prepare_qwen_installation( config: InstallConfig, - paths: QwenRuntimePaths, + paths: RuntimePaths, rendered: RenderedQwenDocument, ) -> List[FileChange]: guard_script = generate_guard_script(AGENT_KEY, paths.agent_id) @@ -130,49 +60,27 @@ def prepare_qwen_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 _read_only_preconditions( - sources: QwenSources, changed_path: Optional[str] + sources: QwenSources, changed_path: str ) -> List[FilePrecondition]: return [ condition for condition in sources.preconditions - if changed_path is None - or not same_qwen_path(condition.file_path, changed_path) + if not same_path(condition.file_path, changed_path) ] -def commit_qwen_installation( - changes: List[FileChange], sources: QwenSources -) -> None: +def commit_qwen_installation(changes: List[FileChange], sources: QwenSources) -> None: write_changes( changes, "Install Qwen Code hooks", @@ -180,15 +88,11 @@ def commit_qwen_installation( ) -def prepare_qwen_uninstall( - rendered: RenderedQwenDocument, -) -> Optional[FileChange]: +def prepare_qwen_uninstall(rendered: RenderedQwenDocument) -> Optional[FileChange]: return _rendered_change(rendered) -def commit_qwen_uninstall( - change: Optional[FileChange], sources: QwenSources -) -> None: +def commit_qwen_uninstall(change: Optional[FileChange], sources: QwenSources) -> None: if change is None: return write_changes( diff --git a/elydora/plugins/qwen_io.py b/elydora/plugins/qwen_io.py index b373f10..791a517 100644 --- a/elydora/plugins/qwen_io.py +++ b/elydora/plugins/qwen_io.py @@ -2,199 +2,19 @@ from __future__ import annotations -import base64 -import os -from typing import Any, List, Optional -import urllib.parse +from typing import List, Tuple -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 .guard_template import generate_guard_script -from .hook_template import generate_hook_script -from .qwen_command import same_qwen_agent_id, same_qwen_path -from .qwen_contract import AGENT_KEY, QwenRuntimeContract, elydora_dir +from ._runtime import expected_runtime_scripts, runtime_contract_exists +from ._strict_json import JsonObject +from .qwen_contract import AGENT_KEY, QwenRuntimeContract -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 _expected_scripts(agent_id: str, config: JsonObject) -> Tuple[str, str]: + return expected_runtime_scripts(AGENT_KEY, agent_id, config) + + +def qwen_runtime_files_exist(contracts: List[QwenRuntimeContract]) -> bool: + return any( + runtime_contract_exists(contract, AGENT_KEY, "Qwen Code", _expected_scripts) + 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_qwen_agent_id(agent_id, expected_agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime identity does not match Qwen 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 = 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_qwen_agent_id(configured_id, agent_id) - or config.get("agent_name") != AGENT_KEY - ): - raise ValueError( - "Elydora runtime config identity does not match Qwen Code agent " - f"{agent_id}: {config_path}" - ) - - -def _runtime_contract_exists(contract: QwenRuntimeContract) -> bool: - root = elydora_dir() - agent_directory = os.path.dirname(contract.guard_path) - if ( - not same_qwen_path(os.path.dirname(agent_directory), root) - or not same_qwen_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") - expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id) - expected_audit = generate_hook_script( - org_id=str(config["org_id"]), - agent_id=contract.agent_id, - kid=str(config["kid"]), - base_url=str(config["base_url"]), - native_payload=True, - agent_name=AGENT_KEY, - ) - return guard.contents == expected_guard and audit.contents == expected_audit - - -def qwen_runtime_files_exist( - contracts: List[QwenRuntimeContract], -) -> bool: - return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/elydora/plugins/qwen_sources.py b/elydora/plugins/qwen_sources.py index 86ee32d..b38849e 100644 --- a/elydora/plugins/qwen_sources.py +++ b/elydora/plugins/qwen_sources.py @@ -11,8 +11,8 @@ from typing import Dict, List, Optional, Set, Tuple from ._dotenv import parse_dotenv from ._managed_files import MAX_SOURCE_BYTES, FileSnapshot, read_physical_file +from ._runtime import same_path, unique_preconditions from ._transaction import FilePrecondition -from .qwen_command import same_qwen_path from .qwen_config import ( QwenDocument, create_qwen_document, @@ -127,7 +127,7 @@ def _resolve_qwen_routing() -> _RoutingResult: discovered_home = values.get("QWEN_HOME") if discovered_home and discovered_home != initial_qwen_home: discovered_directory = _resolve_config_path(discovered_home) - if not same_qwen_path(discovered_directory, initial_directory): + if not same_path(discovered_directory, initial_directory): read_candidate(os.path.join(discovered_directory, ".env")) resolved_home = values.get("QWEN_HOME") return _RoutingResult( @@ -288,15 +288,6 @@ def _source_precondition(document: QwenDocument) -> FilePrecondition: ) -def _deduplicate_preconditions( - values: List[FilePrecondition], -) -> Tuple[FilePrecondition, ...]: - result: Dict[str, FilePrecondition] = {} - for value in values: - result.setdefault(_comparison_path(value.file_path), value) - return tuple(result.values()) - - def read_qwen_sources() -> QwenSources: routing = _resolve_qwen_routing() system_path = _system_settings_path() @@ -355,7 +346,7 @@ def read_qwen_sources() -> QwenSources: system, workspace_active and workspace_trusted, ), - preconditions=_deduplicate_preconditions(preconditions), + preconditions=unique_preconditions(preconditions), )