diff --git a/elydora/plugins/_file_io.py b/elydora/plugins/_file_io.py index 13dd461..29fea6a 100644 --- a/elydora/plugins/_file_io.py +++ b/elydora/plugins/_file_io.py @@ -4,14 +4,15 @@ from __future__ import annotations import json import os +import stat import tempfile -from typing import Any, Dict +from typing import Any, Dict, Optional JsonObject = Dict[str, Any] -def _cleanup_failed_write(path: str, label: str, cause: Exception) -> None: +def _cleanup_failed_write(path: str, label: str, cause: BaseException) -> None: if not path: return try: @@ -47,7 +48,7 @@ def write_text_atomic(path: str, content: str, mode: int, label: str) -> None: os.fsync(file.fileno()) os.chmod(temporary_path, mode) os.replace(temporary_path, path) - except Exception as error: + except BaseException as error: if descriptor >= 0: try: os.close(descriptor) @@ -58,8 +59,46 @@ def write_text_atomic(path: str, content: str, mode: int, label: str) -> None: f"close failed: {close_error}" ) from error _cleanup_failed_write(temporary_path, label, error) - raise OSError(f"Write {label} at {path}: {error}") from error + if isinstance(error, Exception): + raise OSError(f"Write {label} at {path}: {error}") from error + raise def write_json_atomic(path: str, value: JsonObject, mode: int, label: str) -> None: write_text_atomic(path, json.dumps(value, indent=2) + "\n", mode, label) + + +def read_json(path: str, label: str) -> Optional[JsonObject]: + try: + with open(path, "r", encoding="utf-8") as file: + raw = file.read() + except FileNotFoundError: + return None + except OSError as error: + raise OSError(f"Read {label} at {path}: {error}") from error + try: + value = json.loads(raw) + except json.JSONDecodeError as error: + raise ValueError(f"Failed to parse {label} at {path}: {error}") from error + if not isinstance(value, dict): + raise ValueError(f"{label} at {path} must contain a JSON object") + return value + + +def remove_file(path: str, label: str) -> None: + try: + os.remove(path) + except FileNotFoundError: + return + except OSError as error: + raise OSError(f"Remove {label} at {path}: {error}") from error + + +def regular_file_exists(path: str, label: str) -> bool: + try: + metadata = os.stat(path) + except FileNotFoundError: + return False + except OSError as error: + raise OSError(f"Read {label} at {path}: {error}") from error + return stat.S_ISREG(metadata.st_mode) diff --git a/elydora/plugins/_runtime.py b/elydora/plugins/_runtime.py index 6029ee1..55035ec 100644 --- a/elydora/plugins/_runtime.py +++ b/elydora/plugins/_runtime.py @@ -169,13 +169,16 @@ def validate_runtime_config( validate_api_origin(base_url, "Elydora runtime config base_url") -def runtime_config_source(config: InstallConfig, agent_id: str, agent_key: str) -> str: +def runtime_config_source( + config: InstallConfig, agent_id: str, agent_key: str, **extra: str +) -> str: value = { "org_id": config.get("org_id", ""), "agent_id": agent_id, "kid": config.get("kid", ""), "base_url": config.get("base_url", DEFAULT_BASE_URL), "agent_name": agent_key, + **extra, } token = config.get("token") if token: diff --git a/elydora/plugins/_shell_command.py b/elydora/plugins/_shell_command.py index f0e6db1..fee3c94 100644 --- a/elydora/plugins/_shell_command.py +++ b/elydora/plugins/_shell_command.py @@ -57,10 +57,11 @@ def encode_powershell_source(source: str) -> str: return base64.b64encode(source.encode("utf-16le")).decode("ascii") -def encoded_windows_command(script_path: str) -> str: +def encoded_windows_command(script_path: str, preamble: str = "") -> str: + encoded = encode_powershell_source(preamble + powershell_source(script_path)) return ( f'"{windows_powershell_path()}" -NoLogo -NoProfile -NonInteractive ' - f"-EncodedCommand {encode_powershell_source(powershell_source(script_path))}" + f"-EncodedCommand {encoded}" ) @@ -141,11 +142,19 @@ def decode_powershell_source(encoded: str) -> Optional[str]: return None +def same_windows_path(left: str, right: str) -> bool: + return ntpath.normcase(ntpath.abspath(left)) == ntpath.normcase(ntpath.abspath(right)) + + def parse_encoded_windows_command( - command: str, prefix: str = "& " + command: str, + prefix: str = "& ", + powershell_path: Optional[str] = None, ) -> Optional[CommandParts]: match = _ENCODED_COMMAND.fullmatch(command) if match is None or not is_powershell_executable(match.group(1)): return None + if powershell_path is not None and not same_windows_path(match.group(1), powershell_path): + return None source = decode_powershell_source(match.group(2)) return None if source is None else parse_powershell_source(source, prefix) diff --git a/elydora/plugins/kirocli.py b/elydora/plugins/kirocli.py index 50bad97..c09f02f 100644 --- a/elydora/plugins/kirocli.py +++ b/elydora/plugins/kirocli.py @@ -2,15 +2,13 @@ from __future__ import annotations -import json import os import shlex -import stat import subprocess import sys -import tempfile from typing import Any, Dict, List, Optional, Tuple +from ._file_io import read_json, regular_file_exists, remove_file, write_json_atomic, write_text_atomic from .base import AgentPlugin, InstallConfig, PluginStatus from .hook_template import generate_hook_script @@ -42,11 +40,11 @@ class KiroCliPlugin(AgentPlugin): raise ValueError("agent_id is required") if not guard_path: raise ValueError("guard_script_path is required") - if not _regular_file_exists(guard_path, "Elydora guard runtime"): + if not regular_file_exists(guard_path, "Elydora guard runtime"): raise FileNotFoundError(f"Elydora guard runtime is missing: {guard_path}") - v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") or {} - v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") or {} + v2_settings = read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") or {} + v3_settings = read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") or {} v2_hooks = _hooks_object(v2_settings, "Kiro CLI v2 agent config") current_v3_hooks = _v3_hooks(v3_settings) @@ -108,22 +106,28 @@ class KiroCliPlugin(AgentPlugin): kid=config.get("kid", ""), base_url=config.get("base_url", "https://api.elydora.com"), ) - _write_json_atomic(os.path.join(agent_dir, "config.json"), runtime_config, 0o600) - _write_text_atomic( + write_json_atomic( + os.path.join(agent_dir, "config.json"), + runtime_config, + 0o600, + "Elydora runtime config", + ) + write_text_atomic( os.path.join(agent_dir, "private.key"), config.get("private_key", ""), 0o600, + "Elydora private key", ) - _write_text_atomic(hook_path, hook_script, 0o700) - _write_json_atomic(V2_AGENT_PATH, next_v2_settings, 0o600) - _write_json_atomic(V3_HOOKS_PATH, next_v3_settings, 0o600) + write_text_atomic(hook_path, hook_script, 0o700, "Elydora audit runtime") + write_json_atomic(V2_AGENT_PATH, next_v2_settings, 0o600, "Kiro CLI v2 agent config") + write_json_atomic(V3_HOOKS_PATH, next_v3_settings, 0o600, "Kiro CLI v3 hooks config") print('Kiro CLI v2: start with "kiro-cli --agent elydora-audit".') print('Kiro CLI v3: start with "kiro-cli --v3"; global hooks load automatically.') def uninstall(self, agent_id: str = "") -> None: - v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") - v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") + v2_settings = read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") + v3_settings = read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") next_v2: Optional[Tuple[JsonObject, bool]] = None if v2_settings is not None: @@ -153,22 +157,22 @@ class KiroCliPlugin(AgentPlugin): if next_v2 is not None: settings, remove = next_v2 if remove: - _remove_file(V2_AGENT_PATH, "Kiro CLI v2 agent config") + remove_file(V2_AGENT_PATH, "Kiro CLI v2 agent config") else: - _write_json_atomic(V2_AGENT_PATH, settings, 0o600) + write_json_atomic(V2_AGENT_PATH, settings, 0o600, "Kiro CLI v2 agent config") if next_v3 is not None: settings, remove = next_v3 if remove: - _remove_file(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") + remove_file(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") else: - _write_json_atomic(V3_HOOKS_PATH, settings, 0o600) + write_json_atomic(V3_HOOKS_PATH, settings, 0o600, "Kiro CLI v3 hooks config") print("Elydora hooks uninstalled from Kiro CLI.") def status(self) -> PluginStatus: - v2_settings = _read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") - v3_settings = _read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") + v2_settings = read_json(V2_AGENT_PATH, "Kiro CLI v2 agent config") + v3_settings = read_json(V3_HOOKS_PATH, "Kiro CLI v3 hooks config") contracts: List[HookContract] = [] if v2_settings is not None: @@ -204,63 +208,6 @@ class KiroCliPlugin(AgentPlugin): return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) -def _read_json(path: str, label: str) -> Optional[JsonObject]: - try: - with open(path, "r", encoding="utf-8") as file: - raw = file.read() - except FileNotFoundError: - return None - except OSError as error: - raise OSError(f"Read {label} at {path}: {error}") from error - - try: - value = json.loads(raw) - except json.JSONDecodeError as error: - raise ValueError(f"Failed to parse {label} at {path}: {error}") from error - if not isinstance(value, dict): - raise ValueError(f"{label} at {path} must contain a JSON object") - return value - - -def _write_text_atomic(path: str, content: str, mode: int) -> None: - directory = os.path.dirname(path) or "." - os.makedirs(directory, exist_ok=True) - descriptor, temporary_path = tempfile.mkstemp( - dir=directory, - prefix=f".{os.path.basename(path)}.", - suffix=".tmp", - text=True, - ) - try: - with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as file: - file.write(content) - os.chmod(temporary_path, mode) - os.replace(temporary_path, path) - except BaseException as write_error: - try: - os.remove(temporary_path) - except FileNotFoundError: - pass - except OSError as cleanup_error: - raise RuntimeError( - f"Write {path} failed and temporary file cleanup also failed: {cleanup_error}" - ) from write_error - raise - - -def _write_json_atomic(path: str, value: JsonObject, mode: int) -> None: - _write_text_atomic(path, json.dumps(value, indent=2) + "\n", mode) - - -def _remove_file(path: str, label: str) -> None: - try: - os.remove(path) - except FileNotFoundError: - return - except OSError as error: - raise OSError(f"Remove {label} at {path}: {error}") from error - - def _build_command(script_path: str) -> str: arguments = [sys.executable, script_path] return subprocess.list2cmdline(arguments) if os.name == "nt" else shlex.join(arguments) @@ -385,16 +332,6 @@ def _find_v3_command( return None -def _regular_file_exists(path: str, label: str) -> bool: - try: - metadata = os.stat(path) - except FileNotFoundError: - return False - except OSError as error: - raise OSError(f"Read {label} at {path}: {error}") from error - return stat.S_ISREG(metadata.st_mode) - - def _runtime_scripts_exist(contracts: List[HookContract]) -> bool: try: with os.scandir(ELYDORA_DIR) as iterator: @@ -421,10 +358,10 @@ def _runtime_scripts_exist(contracts: List[HookContract]) -> bool: continue config_path = os.path.join(entry.path, "config.json") - config = _read_json(config_path, "Elydora runtime config") + config = read_json(config_path, "Elydora runtime config") if config is None or config.get("agent_name") != AGENT_KEY: continue - return _regular_file_exists( + return regular_file_exists( guard_path, "Elydora guard runtime" - ) and _regular_file_exists(hook_path, "Elydora audit runtime") + ) and regular_file_exists(hook_path, "Elydora audit runtime") return False diff --git a/elydora/plugins/kiroide.py b/elydora/plugins/kiroide.py index 03bdf9b..cefda52 100644 --- a/elydora/plugins/kiroide.py +++ b/elydora/plugins/kiroide.py @@ -2,6 +2,7 @@ from __future__ import annotations +from ._runtime import same_agent_id from .base import AgentPlugin, InstallConfig, PluginStatus from .kiroide_contract import ( AGENT_KEY, @@ -21,7 +22,6 @@ from .kiroide_installation import ( prepare_kiroide_installation, prepare_kiroide_uninstall, ) -from .kiroide_command import same_kiroide_agent_id from .kiroide_io import ( kiroide_runtime_files_exist, legacy_kiroide_contract_matches_agent, @@ -78,13 +78,13 @@ class KiroIdePlugin(AgentPlugin): agent_ids: list[str] = [] for discovered_agent_id in discovered_agent_ids: if not any( - same_kiroide_agent_id(discovered_agent_id, existing_agent_id) + same_agent_id(discovered_agent_id, existing_agent_id) for existing_agent_id in agent_ids ): agent_ids.append(discovered_agent_id) for managed_agent_id in agent_ids: owns_workspace_runtime = any( - same_kiroide_agent_id(contract.agent_id, managed_agent_id) + same_agent_id(contract.agent_id, managed_agent_id) for contract in contracts ) require_kiroide_workspace_owner( diff --git a/elydora/plugins/kiroide_command.py b/elydora/plugins/kiroide_command.py index fbc85cb..ccd4e23 100644 --- a/elydora/plugins/kiroide_command.py +++ b/elydora/plugins/kiroide_command.py @@ -2,15 +2,23 @@ from __future__ import annotations -import base64 from dataclasses import dataclass -import ntpath import os import re import sys from typing import Optional, Tuple -from elydora._runtime_paths import runtime_root +from ._runtime import managed_script_reference, same_path +from ._shell_command import ( + encoded_windows_command, + is_python_executable, + parse_encoded_windows_command, + parse_posix_command, + posix_source, + windows_powershell_path, +) + +_PREAMBLE = "$ErrorActionPreference = 'Stop'; $global:LASTEXITCODE = 1; " @dataclass(frozen=True) @@ -19,71 +27,10 @@ class KiroIdeRuntimeReference: script_path: str -@dataclass(frozen=True) -class _ParsedArgument: - value: str - next_index: int - - -def same_kiroide_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_kiroide_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 _same_windows_path(left: str, right: str) -> bool: - return ntpath.normcase(ntpath.abspath(left)) == ntpath.normcase( - ntpath.abspath(right) - ) - - -def _windows_command(script_path: str) -> str: - source = ( - "$ErrorActionPreference = 'Stop'; $global:LASTEXITCODE = 1; " - 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_kiroide_command(script_path: str) -> str: if ( not os.path.isabs(sys.executable) - or not _is_python_executable(sys.executable) + or not is_python_executable(sys.executable) or not os.path.isabs(script_path) ): raise ValueError( @@ -91,136 +38,25 @@ def build_kiroide_command(script_path: str) -> str: "and script path" ) 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]]: - prefix = "$ErrorActionPreference = 'Stop'; $global:LASTEXITCODE = 1; & " - if not source.startswith(prefix): - return None - executable = _read_powershell_argument(source, len(prefix)) - 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 not _same_windows_path(match.group(1), _windows_powershell_path()) - ): - return None - try: - raw = base64.b64decode(match.group(2), validate=True) - if base64.b64encode(raw).decode("ascii") != match.group(2): - return None - source = raw.decode("utf-16le") - except (UnicodeDecodeError, ValueError): - return None - return _parse_powershell_source(source) - - -def _is_python_executable(file_path: str) -> bool: - return ( - re.fullmatch( - r"(?:python(?:[0-9]+(?:\.[0-9]+)*)?[td]?|" - r"pypy(?:[0-9]+(?:\.[0-9]+)*)?)(?:\.exe)?", - os.path.basename(file_path), - re.IGNORECASE, - ) - is not None - ) + return encoded_windows_command(script_path, _PREAMBLE) + return posix_source(script_path) def _runtime_reference( executable: str, script_path: str, script_name: str ) -> Optional[KiroIdeRuntimeReference]: - if ( - not os.path.isabs(executable) - or not _is_python_executable(executable) - or not os.path.isabs(script_path) - 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_kiroide_path(os.path.dirname(agent_directory), runtime_root()): - return None - agent_id = os.path.basename(agent_directory) - if agent_id in {"", ".", ".."}: - return None - return KiroIdeRuntimeReference(agent_id, script_path) + reference = managed_script_reference(script_path, script_name) + return None if reference is None else KiroIdeRuntimeReference(*reference) def _parse_kiroide_command(command: str) -> Optional[Tuple[str, str]]: - return ( - _parse_windows_command(command) - if os.name == "nt" - else _parse_posix_command(command) - ) + if os.name == "nt": + return parse_encoded_windows_command( + command, _PREAMBLE + "& ", windows_powershell_path() + ) + return parse_posix_command(command) def owned_kiroide_runtime_reference( @@ -236,9 +72,7 @@ def kiroide_runtime_reference( command: str, script_name: str ) -> Optional[KiroIdeRuntimeReference]: parsed = _parse_kiroide_command(command) - if parsed is None: - return None - if not same_kiroide_path(parsed[0], sys.executable): + if parsed is None or not same_path(parsed[0], sys.executable): return None return _runtime_reference(parsed[0], parsed[1], script_name) diff --git a/elydora/plugins/kiroide_contract.py b/elydora/plugins/kiroide_contract.py index 914fc13..977b454 100644 --- a/elydora/plugins/kiroide_contract.py +++ b/elydora/plugins/kiroide_contract.py @@ -8,6 +8,7 @@ import os from typing import Any, Callable, Dict, List, Optional, Tuple from ._managed_files import FileSnapshot +from ._runtime import same_agent_id, same_path from ._strict_json import JsonObject, parse_json_object from .kiroide_command import ( KiroIdeRuntimeReference, @@ -16,8 +17,6 @@ from .kiroide_command import ( legacy_audit_reference, legacy_guard_reference, owned_kiroide_runtime_reference, - same_kiroide_agent_id, - same_kiroide_path, ) @@ -240,7 +239,7 @@ def _managed_pair( if ( guard is None or audit is None - or not same_kiroide_agent_id(guard.agent_id, audit.agent_id) + or not same_agent_id(guard.agent_id, audit.agent_id) ): return None return (guards[0][0], guard), (audits[0][0], audit) @@ -289,7 +288,7 @@ def without_managed_kiroide_hooks( for index, hook in enumerate(hooks): reference = owned.get(index) remove = reference is not None and ( - not agent_id or same_kiroide_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) @@ -382,8 +381,8 @@ def legacy_kiroide_runtime_contract( if ( guard is None or audit is None - or not same_kiroide_agent_id(guard.agent_id, audit.agent_id) - or not same_kiroide_path( + or not same_agent_id(guard.agent_id, audit.agent_id) + or not same_path( os.path.dirname(guard.script_path), os.path.dirname(audit.script_path) ) ): diff --git a/elydora/plugins/kiroide_installation.py b/elydora/plugins/kiroide_installation.py index d2aeec9..f9dc2bc 100644 --- a/elydora/plugins/kiroide_installation.py +++ b/elydora/plugins/kiroide_installation.py @@ -3,12 +3,9 @@ from __future__ import annotations from dataclasses import dataclass -import json import os from typing import List, Optional, Sequence, Tuple -from elydora._runtime_paths import resolve_agent_directory, runtime_root - from ._managed_files import ( MAX_CONFIG_BYTES, MAX_SECRET_BYTES, @@ -16,6 +13,14 @@ from ._managed_files import ( read_physical_file, ) from ._opaque_removal import OpaqueRemovalChange +from ._runtime import ( + DEFAULT_BASE_URL, + present, + resolve_runtime_paths, + runtime_config_source, + same_agent_id, + validate_install_config, +) from ._transaction import ( DirectoryPrecondition, FileChange, @@ -34,11 +39,8 @@ from ._runtime_removal import ( from .base import InstallConfig from .guard_template import generate_guard_script from .hook_template import generate_hook_script -from .kiroide_command import same_kiroide_agent_id, same_kiroide_path from .kiroide_contract import ( AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, RenderedKiroIdeDocument, kiroide_runtime_contracts, ) @@ -54,11 +56,13 @@ from .kiroide_io import ( LegacyKiroIdeDocument, legacy_kiroide_contract_matches_agent, require_physical_legacy_directory, - validate_api_origin, validate_kiroide_runtime_config, - validate_private_key, validate_runtime_tree, ) + +PRODUCT = "Kiro IDE" + + @dataclass(frozen=True) class KiroIdeRuntimePaths: agent_id: str @@ -74,6 +78,8 @@ class KiroIdeRuntimePaths: class PreparedKiroIdeInstallation: changes: List[FileChange] runtime_preconditions: List[FilePrecondition] + + @dataclass(frozen=True) class PreparedKiroIdeUninstall: changes: List[FileChange] @@ -92,79 +98,30 @@ def _require_runtime_state( ) -def _runtime_config( - config: InstallConfig, agent_id: str, workspace_root: 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, - "workspace_root": workspace_root, - } - 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) -> KiroIdeRuntimePaths: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - root = runtime_root() + paths = resolve_runtime_paths(config) + root = os.path.dirname(paths.agent_directory) runtime_parent = os.path.dirname(root) - agent_directory = resolve_agent_directory(root, agent_id) - guard_path = os.path.join(agent_directory, GUARD_SCRIPT) - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - if not same_kiroide_path(config.get("guard_script_path", ""), guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{guard_path}" - ) - directories = inspect_runtime_directories( - runtime_parent, root, agent_directory - ) return KiroIdeRuntimePaths( - agent_id, + paths.agent_id, runtime_parent, root, - agent_directory, - guard_path, - audit_path, - directories, + paths.agent_directory, + paths.guard_path, + paths.audit_path, + inspect_runtime_directories(runtime_parent, root, paths.agent_directory), ) -def _validate_install_config(config: InstallConfig) -> KiroIdeRuntimePaths: - 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"Kiro IDE 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") - return _agent_paths(config) - - def preflight_kiroide_installation( config: InstallConfig, sources: KiroIdeSources ) -> KiroIdeRuntimePaths: if not sources.document.file_path: raise ValueError("Kiro IDE installation requires a workspace hook path") - paths = _validate_install_config(config) + validate_install_config(config, AGENT_KEY, PRODUCT) + paths = _agent_paths(config) owns_workspace_runtime = any( - same_kiroide_agent_id(contract.agent_id, paths.agent_id) + same_agent_id(contract.agent_id, paths.agent_id) for contract in kiroide_runtime_contracts(sources.document.hooks) ) owns_global_legacy_runtime = legacy_kiroide_contract_matches_agent( @@ -229,7 +186,7 @@ def _removes_legacy( ) -> bool: return legacy.contract is not None and ( not agent_id - or same_kiroide_agent_id(legacy.contract.agent_id, agent_id) + or same_agent_id(legacy.contract.agent_id, agent_id) ) @@ -252,10 +209,6 @@ def _legacy_change( ) -def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]: - return [change for change in changes if change is not None] - - def prepare_kiroide_installation( config: InstallConfig, paths: KiroIdeRuntimePaths, @@ -269,7 +222,7 @@ def prepare_kiroide_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, ) @@ -280,12 +233,11 @@ def prepare_kiroide_installation( _runtime_change( os.path.join(paths.agent_directory, "config.json"), "Elydora runtime config", - _json_source( - _runtime_config( - config, - paths.agent_id, - sources.paths.workspace_root, - ) + runtime_config_source( + config, + paths.agent_id, + AGENT_KEY, + workspace_root=sources.paths.workspace_root, ), 0o600, MAX_CONFIG_BYTES, @@ -303,7 +255,7 @@ def prepare_kiroide_installation( paths.audit_path, "Elydora audit runtime", audit_script, 0o700 ), ] - changes = _present( + changes = present( [ *(change for change, _condition in runtime), _rendered_change(rendered), @@ -420,7 +372,7 @@ def prepare_kiroide_uninstall( candidates = runtime_agent_ids or ((agent_id,) if agent_id else ()) for runtime_agent_id in candidates: if any( - same_kiroide_agent_id(runtime_agent_id, selected_agent_id) + same_agent_id(runtime_agent_id, selected_agent_id) for selected_agent_id in selected_agent_ids ): continue @@ -436,7 +388,7 @@ def prepare_kiroide_uninstall( config_path, sources.paths.workspace_root, allow_missing_workspace_root=any( - same_kiroide_agent_id( + same_agent_id( contract.agent_id, runtime_agent_id ) for contract in kiroide_runtime_contracts( @@ -454,7 +406,7 @@ def prepare_kiroide_uninstall( ) if runtime_removal is not None: runtime_removals.append(runtime_removal) - provider_changes = _present( + provider_changes = present( [_rendered_change(rendered), _legacy_change(sources.legacy, agent_id)] ) return PreparedKiroIdeUninstall( diff --git a/elydora/plugins/kiroide_io.py b/elydora/plugins/kiroide_io.py index c8d97cf..116389b 100644 --- a/elydora/plugins/kiroide_io.py +++ b/elydora/plugins/kiroide_io.py @@ -2,11 +2,9 @@ from __future__ import annotations -import base64 from dataclasses import dataclass import os -from typing import Any, List, Optional -import urllib.parse +from typing import List, Optional from elydora._runtime_paths import resolve_agent_directory, runtime_root @@ -20,10 +18,17 @@ from ._managed_files import ( read_physical_directory, read_physical_file, ) +from ._runtime import ( + RUNTIME_ARTIFACTS, + RUNTIME_CONFIG_FIELDS, + expected_runtime_scripts, + require_non_empty_string, + same_agent_id, + same_path, + validate_api_origin, + validate_private_key, +) from ._strict_json import JsonObject, parse_json_object -from .guard_template import generate_guard_script -from .hook_template import generate_hook_script -from .kiroide_command import same_kiroide_agent_id, same_kiroide_path from .kiroide_contract import ( AGENT_KEY, CONFIG_FILE, @@ -187,67 +192,7 @@ def require_physical_legacy_directory(legacy: LegacyKiroIdeDocument) -> None: ) -def validate_api_origin(value: str, label: str = "base_url") -> None: - try: - parsed = urllib.parse.urlsplit(value) - hostname = parsed.hostname - parsed.port - except ValueError as error: - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") from error - invalid_character = "\\" in value or any( - character.isspace() or ord(character) < 32 for character in value - ) - if ( - parsed.scheme not in ("http", "https") - or not parsed.netloc - or hostname is None - or invalid_character - ): - raise ValueError(f"{label} must be an absolute HTTP or HTTPS URL") - if ( - parsed.username is not None - or parsed.password is not None - or parsed.query - or parsed.fragment - ): - raise ValueError( - f"{label} must exclude credentials, query parameters, and fragments" - ) - - -def validate_private_key(value: str, label: str = "private_key") -> None: - try: - padded = value + "=" * ((4 - len(value) % 4) % 4) - seed = base64.b64decode( - padded.replace("-", "+").replace("_", "/"), validate=True - ) - canonical = base64.urlsafe_b64encode(seed).rstrip(b"=").decode("ascii") - except (ValueError, UnicodeEncodeError) as error: - raise ValueError( - f"{label} must be a canonical 32-byte base64url value" - ) from error - if len(seed) != 32 or canonical != value: - raise ValueError(f"{label} must be a canonical 32-byte base64url value") - - -def _require_non_empty_string(value: Any, field: str, config_path: str) -> str: - if not isinstance(value, str) or not value.strip(): - raise ValueError(f"Elydora runtime config {field} is invalid: {config_path}") - return value - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - snapshot = read_physical_file( - file_path, "Elydora runtime config", MAX_CONFIG_BYTES - ) - if snapshot is None: - return None - return parse_json_object( - snapshot.contents, f"Elydora runtime config at {file_path}" - ) - - -def _validate_runtime_config( +def validate_kiroide_runtime_config( config: JsonObject, expected_agent_id: str, config_path: str, @@ -255,27 +200,19 @@ def _validate_runtime_config( allow_missing_workspace_root: bool = False, allow_legacy_ownerless_config: bool = False, ) -> None: - supported = { - "org_id", - "agent_id", - "kid", - "base_url", - "token", - "agent_name", - "workspace_root", - } + supported = RUNTIME_CONFIG_FIELDS | {"workspace_root"} 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( + 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_kiroide_agent_id(agent_id, expected_agent_id) + not same_agent_id(agent_id, expected_agent_id) or config.get("agent_name") != AGENT_KEY ): raise ValueError( @@ -289,8 +226,8 @@ def _validate_runtime_config( if "token" in config and not ( legacy_ownerless_config and config.get("token") == "" ): - _require_non_empty_string(config.get("token"), "token", config_path) - base_url = _require_non_empty_string( + 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") @@ -298,39 +235,20 @@ def _validate_runtime_config( allow_missing_workspace_root or legacy_ownerless_config ): return - workspace_root = _require_non_empty_string( + workspace_root = require_non_empty_string( configured_workspace, "workspace_root", config_path ) if not os.path.isabs(workspace_root): raise ValueError( f"Elydora runtime config workspace_root must be absolute: {config_path}" ) - if not same_kiroide_path(workspace_root, expected_workspace_root): + if not same_path(workspace_root, expected_workspace_root): raise ValueError( "Elydora Kiro IDE agent is bound to another workspace: " f"{workspace_root}" ) -def validate_kiroide_runtime_config( - config: JsonObject, - expected_agent_id: str, - config_path: str, - expected_workspace_root: str, - allow_missing_workspace_root: bool = False, - allow_legacy_ownerless_config: bool = False, -) -> None: - """Validate a runtime config already read through a pinned directory.""" - _validate_runtime_config( - config, - expected_agent_id, - config_path, - expected_workspace_root, - allow_missing_workspace_root, - allow_legacy_ownerless_config, - ) - - def validate_runtime_tree( agent_directory: str, agent_id: str, @@ -346,27 +264,21 @@ def validate_runtime_tree( ): return config_path = os.path.join(agent_directory, "config.json") - config = _read_runtime_config(config_path) - artifact_exists = any( + config_snapshot = read_physical_file(config_path, "Elydora runtime config", MAX_CONFIG_BYTES) + artifact_states = [ physical_file_exists(os.path.join(agent_directory, name), label) - for name, label in ( - ("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"), - ) - ) - if config is None: + for name, label in RUNTIME_ARTIFACTS + ] + artifact_exists = any(artifact_states) + if config_snapshot is None: if artifact_exists: raise ValueError( "Elydora runtime identity cannot be verified without config.json: " f"{agent_directory}" ) return - _validate_runtime_config( - config, + validate_kiroide_runtime_config( + parse_json_object(config_snapshot.contents, f"Elydora runtime config at {config_path}"), agent_id, config_path, workspace_root, @@ -381,8 +293,8 @@ def _runtime_contract_exists( root = runtime_root() agent_directory = os.path.dirname(contract.guard_path) if ( - not same_kiroide_path(os.path.dirname(agent_directory), root) - or not same_kiroide_path( + not same_path(os.path.dirname(agent_directory), root) + or not same_path( contract.audit_path, os.path.join(agent_directory, "hook.py") ) ): @@ -414,7 +326,7 @@ def _runtime_contract_exists( 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( + validate_kiroide_runtime_config( config, contract.agent_id, config_path, @@ -436,15 +348,7 @@ def _runtime_contract_exists( f"its owner: {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, - ) + expected_guard, expected_audit = expected_runtime_scripts(AGENT_KEY, contract.agent_id, config) return guard.contents == expected_guard and audit.contents == expected_audit @@ -478,15 +382,15 @@ def legacy_kiroide_contract_matches_agent( legacy: LegacyKiroIdeDocument, agent_id: str ) -> bool: contract = legacy.contract - if contract is None or not same_kiroide_agent_id( + if contract is None or not same_agent_id( contract.agent_id, agent_id ): return False agent_directory = resolve_agent_directory(runtime_root(), agent_id) - return same_kiroide_path( + return same_path( contract.guard_path, os.path.join(agent_directory, "guard.py"), - ) and same_kiroide_path( + ) and same_path( contract.audit_path, os.path.join(agent_directory, "hook.py"), ) diff --git a/tests/test_kiroide_plugin.py b/tests/test_kiroide_plugin.py index 59a6f97..ee32f9e 100644 --- a/tests/test_kiroide_plugin.py +++ b/tests/test_kiroide_plugin.py @@ -18,6 +18,7 @@ import pytest from elydora import cli from elydora.plugins import kiroide +from elydora.plugins._shell_command import is_python_executable, windows_powershell_path from elydora.plugins import ( _transaction, kiroide_command, @@ -1377,7 +1378,7 @@ def test_kiroide_current_commands_require_the_system_powershell_launcher( ) -> None: fixture = _prepare_fixture(monkeypatch, tmp_path) command = kiroide_command.build_kiroide_command(str(fixture.guard_path)) - expected = f'"{kiroide_command._windows_powershell_path()}"' + expected = f'"{windows_powershell_path()}"' altered = command.replace(expected, r'"C:\Other\powershell.exe"', 1) assert kiroide_command.kiroide_runtime_reference(altered, "guard.py") is None @@ -1415,7 +1416,7 @@ def test_kiroide_windows_launcher_fails_when_python_is_missing( def test_kiroide_ownership_accepts_supported_python_runtime_names( executable_name: str, ) -> None: - assert kiroide_command._is_python_executable(executable_name) + assert is_python_executable(executable_name) def test_kiroide_builder_rejects_unrecoverable_executable_names(