From fd62dca155250f1dad970de7cf59776bb390e98f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?= Date: Mon, 20 Jul 2026 04:31:42 -0400 Subject: [PATCH] fix(python): align Kimi hooks with official contracts --- AGENTS.md | 2 +- sdks/python/README.md | 2 +- sdks/python/elydora/plugins/kimi.py | 509 +++-------------- sdks/python/elydora/plugins/kimi_command.py | 232 ++++++++ sdks/python/elydora/plugins/kimi_contract.py | 331 +++++++++++ .../elydora/plugins/kimi_installation.py | 195 +++++++ sdks/python/elydora/plugins/kimi_io.py | 276 ++++++++++ sdks/python/elydora/plugins/registry.py | 2 +- sdks/python/tests/kimi_support.py | 265 +++++++++ sdks/python/tests/test_kimi_plugin.py | 517 ++++++++---------- sdks/python/tests/test_kimi_runtime.py | 161 ++++++ sdks/python/tests/test_kimi_security.py | 197 +++++++ sdks/python/tests/test_kimi_transaction.py | 161 ++++++ 13 files changed, 2105 insertions(+), 745 deletions(-) create mode 100644 sdks/python/elydora/plugins/kimi_command.py create mode 100644 sdks/python/elydora/plugins/kimi_contract.py create mode 100644 sdks/python/elydora/plugins/kimi_installation.py create mode 100644 sdks/python/elydora/plugins/kimi_io.py create mode 100644 sdks/python/tests/kimi_support.py create mode 100644 sdks/python/tests/test_kimi_runtime.py create mode 100644 sdks/python/tests/test_kimi_security.py create mode 100644 sdks/python/tests/test_kimi_transaction.py diff --git a/AGENTS.md b/AGENTS.md index b1c2d83..445e0c2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -24,7 +24,7 @@ This repository publishes the open-source server, console, integration catalog, - Report installation as healthy only when a complete contract references all required runtime files. - Keep stable, legacy, and early-access contracts explicit, including their activation commands. - Resolve stable Kimi hooks through `$KIMI_CODE_HOME/config.toml` with `~/.kimi-code/config.toml` as the default, and treat an empty override as the default. Activate `~/.kimi/config.toml` only when the legacy home exists. Register exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` rules with ten-second commands; stable Kimi supports its current sixteen-event schema and legacy kimi-cli supports its thirteen-event schema. Preserve the native snake_case payload, propagate exit code `2`, and keep audit delivery fail-open with owner-only error logs. Use encoded PowerShell commands on Windows and exact two-argument shell commands on Unix. Commit every detected config, generated runtime, runtime config, and private key through one rollback-capable transaction. -- Preserve Kimi TOML comments and unrelated formatting through range-based edits, then parse every selected document before writing any file. +- Preserve Kimi TOML comments, array style, and unrelated formatting through syntax-preserving edits, then parse every selected document before writing any file. - Write Grok Build integrations to a native user hook file under non-empty `GROK_HOME` or `~/.grok`; keep Claude Code and Cursor compatibility sources read-only. - Write Auggie hooks only to `~/.augment/settings.json`; keep system and workspace settings read-only. Generate `.cmd` wrappers on Windows and `.sh` wrappers on Unix because Auggie dispatches supported script paths, and express hook timeouts in milliseconds. - Validate Auggie matcher syntax during installation with Node.js `new RegExp`; keep status and uninstall independent from the JavaScript validator so recovery remains available offline. diff --git a/sdks/python/README.md b/sdks/python/README.md index 8a39fc4..2865331 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -84,7 +84,7 @@ Cursor installation writes native global `preToolUse`, `postToolUse`, and `postT Cline installation writes `PreToolUse.mjs` and `PostToolUse.mjs` to `$CLINE_DIR/hooks` (default `~/.cline/hooks`). Elydora leaves the Documents and workspace hook roots unchanged. The wrappers preserve Cline's official payload and translate a frozen guard into JSON stdout cancellation control. -Kimi installation writes the strict hook contract to each detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the migrating Python CLI's `~/.kimi/config.toml`. A fresh installation targets current Kimi Code, avoiding cross-runtime migration markers. Both runtimes load the hooks globally; run `/hooks` to inspect them. +Kimi installation writes exact `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` rules to every detected runtime: Kimi Code's `$KIMI_CODE_HOME/config.toml` (default `~/.kimi-code/config.toml`) and the legacy Python CLI's `~/.kimi/config.toml`. A fresh installation targets stable Kimi Code; an existing legacy home activates the migration contract. Empty `KIMI_CODE_HOME` values use the default. The selected TOML files, generated runtimes, runtime config, and private key commit as one rollback-capable transaction. Managed commands preserve Kimi's native snake_case event JSON, propagate freeze and revocation through exit code `2`, and use encoded PowerShell on Windows. Run `/hooks` to inspect the global rules. Grok Build installation writes native global hooks to `$GROK_HOME/hooks/elydora-audit.json` (default `~/.grok/hooks/elydora-audit.json`). Project hooks still follow Grok's `/hooks-trust` workflow; Elydora leaves project, Claude Code, and Cursor compatibility files unchanged. diff --git a/sdks/python/elydora/plugins/kimi.py b/sdks/python/elydora/plugins/kimi.py index 0b48137..84bf826 100644 --- a/sdks/python/elydora/plugins/kimi.py +++ b/sdks/python/elydora/plugins/kimi.py @@ -2,472 +2,97 @@ from __future__ import annotations -import copy -from dataclasses import dataclass -import os -import shlex -import shutil -# subprocess provides argument quoting only; this module starts no process. -import subprocess # nosec B404 -import sys -from typing import Any, Dict, FrozenSet, List, Optional - -from tomlkit import aot, document, dumps, inline_table, parse, table -from tomlkit.items import AoT, Array +from typing import List from .base import AgentPlugin, InstallConfig, PluginStatus -from ._file_io import ( - read_json as _read_json, - regular_file_exists as _regular_file_exists, - remove_file as _remove_file, - require_runtime as _require_runtime, - write_json_atomic as _write_json_atomic, - write_text_atomic as _write_text_atomic, +from .kimi_contract import ( + AGENT_KEY, + KimiDocument, + TomlObject, + build_kimi_hook, + kimi_runtime_contracts, + remove_managed_kimi_hooks, + render_kimi_document, ) -from .hook_template import generate_hook_script +from .kimi_installation import ( + commit_kimi_installation, + commit_kimi_uninstall, + preflight_kimi_installation, + prepare_kimi_installation, + prepare_kimi_uninstall, +) +from .kimi_io import kimi_runtime_files_exist, read_kimi_documents -AGENT_KEY = "kimi" -GUARD_SCRIPT = "guard.py" -AUDIT_SCRIPT = "hook.py" -HOOK_TIMEOUT_SECONDS = 10 -ELYDORA_DIR = os.path.join(os.path.expanduser("~"), ".elydora") - -SHARED_EVENTS: FrozenSet[str] = frozenset({ - "PreToolUse", - "PostToolUse", - "PostToolUseFailure", - "UserPromptSubmit", - "Stop", - "StopFailure", - "SessionStart", - "SessionEnd", - "SubagentStart", - "SubagentStop", - "PreCompact", - "PostCompact", - "Notification", -}) -MODERN_EVENTS = SHARED_EVENTS | { - "PermissionRequest", - "PermissionResult", - "Interrupt", -} -LEGACY_EVENTS = SHARED_EVENTS -SUPPORTED_FIELDS = frozenset({"event", "matcher", "command", "timeout"}) - -TomlObject = Dict[str, Any] - - -@dataclass(frozen=True) -class KimiContract: - runtime_name: str - label: str - config_path: str - events: FrozenSet[str] - - -@dataclass(frozen=True) -class KimiDocument: - contract: KimiContract - exists: bool - raw: str - value: Any - hooks: List[TomlObject] - - -@dataclass(frozen=True) -class RuntimeContract: - guard: str - audit: str - config_path: str - - -@dataclass(frozen=True) -class ConfigMutation: - kind: str - source: KimiDocument - raw: str = "" - - -def _home_dir() -> str: - return os.path.expanduser("~") - - -def _path_exists(path: str, label: str) -> bool: - try: - os.stat(path) - return True - except (FileNotFoundError, NotADirectoryError): - return False - except OSError as error: - raise OSError(f"Inspect {label} at {path}: {error}") from error - - -def _legacy_cli_on_path() -> bool: - try: - return shutil.which("kimi-cli") is not None - except OSError as error: - raise OSError(f"Inspect kimi-cli executable on PATH: {error}") from error - - -def _resolve_contracts() -> List[KimiContract]: - home = _home_dir() - explicit_home = os.environ.get("KIMI_CODE_HOME") or None - kimi_home = explicit_home or os.path.join(home, ".kimi-code") - modern = KimiContract( - runtime_name="Kimi Code", - label="Kimi Code hooks config", - config_path=os.path.join(kimi_home, "config.toml"), - events=MODERN_EVENTS, - ) - legacy = KimiContract( - runtime_name="kimi-cli", - label="kimi-cli legacy hooks config", - config_path=os.path.join(home, ".kimi", "config.toml"), - events=LEGACY_EVENTS, - ) - modern_detected = explicit_home is not None or _path_exists( - kimi_home, "Kimi Code home" - ) - legacy_detected = _path_exists( - legacy.config_path, legacy.label - ) or _legacy_cli_on_path() - if legacy_detected and not modern_detected: - return [legacy] - return [modern, legacy] if legacy_detected else [modern] - - -def _unwrap(value: Any) -> Any: - unwrap = getattr(value, "unwrap", None) - return unwrap() if callable(unwrap) else value - - -def _validate_hook(value: Any, contract: KimiContract, index: int) -> TomlObject: - raw = _unwrap(value) - if not isinstance(raw, dict): - raise ValueError(f"{contract.label} hook {index + 1} must be a table") - hook = dict(raw) - unsupported = next((key for key in hook if key not in SUPPORTED_FIELDS), None) - if unsupported is not None: - raise ValueError( - f'{contract.label} hook {index + 1} has unsupported field "{unsupported}"' - ) - event = hook.get("event") - if not isinstance(event, str) or event not in contract.events: - raise ValueError( - f'{contract.label} hook {index + 1} has unsupported event "{event}"' - ) - command = hook.get("command") - if not isinstance(command, str) or not command: - raise ValueError( - f"{contract.label} hook {index + 1} requires a non-empty command" - ) - matcher = hook.get("matcher") - if matcher is not None and not isinstance(matcher, str): - raise ValueError(f"{contract.label} hook {index + 1} matcher must be a string") - timeout = hook.get("timeout") - if timeout is not None and ( - isinstance(timeout, bool) - or not isinstance(timeout, int) - or timeout < 1 - or timeout > 600 - ): - raise ValueError( - f"{contract.label} hook {index + 1} timeout must be an integer from 1 to 600" - ) - return hook - - -def _read_hooks(value: Any, contract: KimiContract) -> List[TomlObject]: - if "hooks" not in value: - return [] - container = value["hooks"] - raw = _unwrap(container) - if not isinstance(raw, list): - raise ValueError(f'{contract.label} field "hooks" must be an array') - return [_validate_hook(item, contract, index) for index, item in enumerate(raw)] - - -def _read_config(contract: KimiContract) -> KimiDocument: - try: - with open( - contract.config_path, - "r", - encoding="utf-8", - newline="", - ) as file: - raw = file.read() - except FileNotFoundError: - value = document() - return KimiDocument(contract, False, "", value, []) - except OSError as error: - raise OSError( - f"Read {contract.label} at {contract.config_path}: {error}" - ) from error - - try: - value = parse(raw) - except Exception as error: - raise ValueError( - f"Failed to parse {contract.label} at {contract.config_path}: {error}" - ) from error - return KimiDocument(contract, True, raw, value, _read_hooks(value, contract)) - - -def _read_all_configs() -> List[KimiDocument]: - return [_read_config(contract) for contract in _resolve_contracts()] - - -def _build_command(script_path: str) -> str: - arguments = [sys.executable, script_path] - return ( - subprocess.list2cmdline(arguments) - if os.name == "nt" - else shlex.join(arguments) - ) - - -def _build_hook(event: str, script_path: str) -> TomlObject: - return { - "event": event, - "command": _build_command(script_path), - "timeout": HOOK_TIMEOUT_SECONDS, - } - - -def _render_hooks( +def _installed_hooks( source: KimiDocument, - keep_indices: List[int], - additions: List[TomlObject], -) -> str: - candidate = copy.deepcopy(source.value) - if "hooks" in candidate: - container: Any = candidate["hooks"] - if not isinstance(container, (AoT, Array)): - raise ValueError(f'{source.contract.label} field "hooks" must be an array') - keep = set(keep_indices) - for index in range(len(source.hooks) - 1, -1, -1): - if index not in keep: - del container[index] - else: - container = aot() - candidate.add("hooks", container) - - for hook in additions: - item = table() if isinstance(container, AoT) else inline_table() - for key, value in hook.items(): - item[key] = value - container.append(item) - if len(container) == 0: - del candidate["hooks"] - return dumps(candidate) - - -def _normalize(value: str) -> str: - return os.path.normcase(value) - - -def _command_references(command: str, path: str) -> bool: - return _normalize(path) in _normalize(command) - - -def _is_managed_hook( - hook: TomlObject, - event: str, - script_name: str, - agent_id: str = "", -) -> bool: - if hook.get("event") != event or not isinstance(hook.get("command"), str): - return False - command = str(hook["command"]) - if agent_id: - script_path = os.path.join(ELYDORA_DIR, agent_id, script_name) - return _command_references(command, script_path) - normalized = _normalize(command) - return ( - _normalize(f"{os.sep}.elydora{os.sep}") in normalized - and _normalize(f"{os.sep}{script_name}") in normalized - ) - - -def _kept_hook_indices(source: KimiDocument, agent_id: str = "") -> List[int]: + guard_path: str, + audit_path: str, +) -> List[TomlObject]: return [ - index - for index, hook in enumerate(source.hooks) - if not _is_managed_hook(hook, "PreToolUse", GUARD_SCRIPT, agent_id) - and not _is_managed_hook(hook, "PostToolUse", AUDIT_SCRIPT, agent_id) + *remove_managed_kimi_hooks(source.hooks), + build_kimi_hook("PreToolUse", guard_path), + build_kimi_hook("PostToolUse", audit_path), + build_kimi_hook("PostToolUseFailure", audit_path), ] -def _runtime_contract(source: KimiDocument) -> Optional[RuntimeContract]: - guard = next( - ( - hook - for hook in source.hooks - if _is_managed_hook(hook, "PreToolUse", GUARD_SCRIPT) - ), - None, - ) - audit = next( - ( - hook - for hook in source.hooks - if _is_managed_hook(hook, "PostToolUse", AUDIT_SCRIPT) - ), - None, - ) - if guard is None or audit is None: - return None - return RuntimeContract( - guard=str(guard["command"]), - audit=str(audit["command"]), - config_path=source.contract.config_path, - ) - - -def _runtime_scripts_exist(contracts: List[RuntimeContract]) -> bool: - try: - with os.scandir(ELYDORA_DIR) as iterator: - entries = sorted(iterator, key=lambda entry: entry.name) - except FileNotFoundError: - return False - except OSError as error: - raise OSError(f"Read Elydora runtime directory at {ELYDORA_DIR}: {error}") from error - - for entry in entries: - try: - if not entry.is_dir(follow_symlinks=False): - continue - except OSError as error: - raise OSError(f"Read Elydora runtime entry at {entry.path}: {error}") from error - guard_path = os.path.join(entry.path, GUARD_SCRIPT) - audit_path = os.path.join(entry.path, AUDIT_SCRIPT) - if not any( - _command_references(contract.guard, guard_path) - and _command_references(contract.audit, audit_path) - for contract in contracts - ): - continue - config_path = os.path.join(entry.path, "config.json") - runtime_config = _read_json(config_path, "Elydora runtime config") - if runtime_config is None: - continue - agent_name = runtime_config.get("agent_name") - if not isinstance(agent_name, str): - raise ValueError( - f'Elydora runtime config at {config_path} field "agent_name" ' - "must be a string" - ) - if agent_name != AGENT_KEY: - continue - return _regular_file_exists( - guard_path, "Elydora guard runtime" - ) and _regular_file_exists(audit_path, "Elydora audit runtime") - return False - - -def _apply_mutation(mutation: ConfigMutation) -> None: - path = mutation.source.contract.config_path - if mutation.kind == "write": - _write_text_atomic(path, mutation.raw, 0o600, mutation.source.contract.label) - elif mutation.kind == "remove": - _remove_file(path, mutation.source.contract.label) - - class KimiPlugin(AgentPlugin): """Install Elydora into detected Kimi user hook contracts.""" + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + documents = read_kimi_documents() + preflight_kimi_installation(config, documents) + def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - if not agent_id: - raise ValueError("agent_id is required") - sources = _read_all_configs() - guard_path = config.get("guard_script_path", "") - _require_runtime(guard_path, "Elydora guard runtime") - agent_dir = os.path.join(ELYDORA_DIR, agent_id) - audit_path = os.path.join(agent_dir, AUDIT_SCRIPT) - mutations = [ - ConfigMutation( - "write", + documents = read_kimi_documents() + paths = preflight_kimi_installation(config, documents) + rendered = [ + render_kimi_document( source, - _render_hooks( - source, - _kept_hook_indices(source), - [ - _build_hook("PreToolUse", guard_path), - _build_hook("PostToolUse", audit_path), - ], - ), + _installed_hooks(source, paths.guard_path, paths.audit_path), ) - for source in sources + for source in documents ] - runtime_config: TomlObject = { - "org_id": config.get("org_id", ""), - "agent_id": agent_id, - "kid": config.get("kid", ""), - "base_url": config.get("base_url", "https://api.elydora.com"), - "token": config.get("token", ""), - "agent_name": AGENT_KEY, - } - audit_script = generate_hook_script( - org_id=config.get("org_id", ""), - agent_id=agent_id, - kid=config.get("kid", ""), - base_url=config.get("base_url", "https://api.elydora.com"), + changes = prepare_kimi_installation(config, paths, rendered) + commit_kimi_installation(changes) + runtimes = " and ".join( + source.contract.runtime_name for source in documents ) - _write_json_atomic( - os.path.join(agent_dir, "config.json"), - runtime_config, - 0o600, - "Elydora runtime config", + print( + f"{runtimes}: global PreToolUse, PostToolUse, " + "and PostToolUseFailure hooks installed." ) - _write_text_atomic( - os.path.join(agent_dir, "private.key"), - config.get("private_key", ""), - 0o600, - "Elydora private key", - ) - _write_text_atomic(audit_path, audit_script, 0o700, "Elydora audit runtime") - for mutation in mutations: - _apply_mutation(mutation) - runtimes = " and ".join(source.contract.runtime_name for source in sources) - print(f"{runtimes}: global PreToolUse and PostToolUse hooks installed.") def uninstall(self, agent_id: str = "") -> None: - sources = _read_all_configs() - mutations: List[ConfigMutation] = [] - for source in sources: - keep_indices = _kept_hook_indices(source, agent_id) - if len(keep_indices) == len(source.hooks): - mutations.append(ConfigMutation("none", source)) - continue - raw = _render_hooks(source, keep_indices, []) - mutations.append( - ConfigMutation("write" if raw.strip() else "remove", source, raw) + documents = read_kimi_documents() + rendered = [ + render_kimi_document( + source, + remove_managed_kimi_hooks(source.hooks, agent_id), ) - for mutation in mutations: - _apply_mutation(mutation) + for source in documents + ] + commit_kimi_uninstall(prepare_kimi_uninstall(rendered)) def status(self) -> PluginStatus: - sources = _read_all_configs() - contracts = [ - contract - for source in sources - for contract in [_runtime_contract(source)] - if contract is not None - ] - if not contracts: - return PluginStatus( - installed=False, - agent=AGENT_KEY, - details="Not installed", - ) - installed = _runtime_scripts_exist(contracts) - paths = ", ".join(contract.config_path for contract in contracts) + documents = read_kimi_documents() + contracts = kimi_runtime_contracts(documents) + installed = bool(contracts) and kimi_runtime_files_exist(contracts) + config_path = ( + contracts[-1].config_path + if contracts + else documents[0].contract.config_path + ) details = ( - f"Config: {paths}" + f"Config: {config_path}" if installed - else f"Configured at {paths}; runtime scripts missing" + else f"Configured at {config_path}; managed contract incomplete" + ) + return PluginStatus( + installed=installed, + agent=AGENT_KEY, + details=details if contracts else "Not installed", ) - return PluginStatus(installed=installed, agent=AGENT_KEY, details=details) diff --git a/sdks/python/elydora/plugins/kimi_command.py b/sdks/python/elydora/plugins/kimi_command.py new file mode 100644 index 0000000..9a66e92 --- /dev/null +++ b/sdks/python/elydora/plugins/kimi_command.py @@ -0,0 +1,232 @@ +"""Kimi hook command rendering and exact runtime ownership.""" + +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 sys +from typing import Optional, Tuple + + +@dataclass(frozen=True) +class KimiRuntimeReference: + agent_id: str + 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) + + +def _parse_legacy_posix_command(command: str) -> Optional[Tuple[str, str]]: + try: + arguments = shlex.split(command, posix=True) + except ValueError: + return None + if len(arguments) != 2 or shlex.join(arguments) != command: + return None + return arguments[0], arguments[1] + + +def _parse_legacy_windows_command(command: str) -> Optional[Tuple[str, str]]: + match = re.fullmatch( + r'(?:(?:"([^"\r\n]+)")|([^\s"\r\n]+)) ' + r'(?:(?:"([^"\r\n]+)")|([^\s"\r\n]+))', + command, + ) + if match is None: + return None + arguments = [match.group(1) or match.group(2), match.group(3) or match.group(4)] + if subprocess.list2cmdline(arguments) != command: + return None + 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) + 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 + ): + 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) diff --git a/sdks/python/elydora/plugins/kimi_contract.py b/sdks/python/elydora/plugins/kimi_contract.py new file mode 100644 index 0000000..2280a35 --- /dev/null +++ b/sdks/python/elydora/plugins/kimi_contract.py @@ -0,0 +1,331 @@ +"""Kimi hook schemas, exact ownership, and TOML rendering.""" + +from __future__ import annotations + +import copy +from dataclasses import dataclass +import os +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 .kimi_command import ( + KimiRuntimeReference, + build_kimi_command, + kimi_runtime_reference, + same_kimi_agent_id, + same_kimi_path, +) + + +AGENT_KEY = "kimi" +GUARD_SCRIPT = "guard.py" +AUDIT_SCRIPT = "hook.py" +HOOK_TIMEOUT_SECONDS = 10 + +SHARED_EVENTS: FrozenSet[str] = frozenset( + { + "PreToolUse", + "PostToolUse", + "PostToolUseFailure", + "UserPromptSubmit", + "Stop", + "StopFailure", + "SessionStart", + "SessionEnd", + "SubagentStart", + "SubagentStop", + "PreCompact", + "PostCompact", + "Notification", + } +) +STABLE_EVENTS = SHARED_EVENTS | { + "PermissionRequest", + "PermissionResult", + "Interrupt", +} +LEGACY_EVENTS = SHARED_EVENTS +SUPPORTED_FIELDS = frozenset({"event", "matcher", "command", "timeout"}) + +TomlObject = Dict[str, Any] +ManagedEvent = str + + +@dataclass(frozen=True) +class KimiContract: + generation: str + runtime_name: str + label: str + directory_label: str + config_path: str + events: FrozenSet[str] + + +@dataclass(frozen=True) +class KimiDocument: + contract: KimiContract + exists: bool + value: Any + hooks: List[TomlObject] + uses_hook_tables: bool + raw: Optional[str] = None + + +@dataclass(frozen=True) +class RenderedKimiDocument: + document: KimiDocument + changed: bool + next_source: Optional[str] = None + + +@dataclass(frozen=True) +class KimiRuntimeContract: + agent_id: str + guard_path: str + audit_path: str + config_path: str + + +def _unwrap(value: Any) -> Any: + unwrap = getattr(value, "unwrap", None) + return unwrap() if callable(unwrap) else value + + +def validate_kimi_hook( + value: Any, contract: KimiContract, index: int +) -> TomlObject: + raw = _unwrap(value) + if not isinstance(raw, dict): + raise ValueError(f"{contract.label} hook {index + 1} must be a table") + hook = dict(raw) + unsupported = next( + (key for key in hook if key not in SUPPORTED_FIELDS), + None, + ) + if unsupported is not None: + raise ValueError( + f'{contract.label} hook {index + 1} has unsupported field "{unsupported}"' + ) + event = hook.get("event") + if not isinstance(event, str) or event not in contract.events: + raise ValueError( + f'{contract.label} hook {index + 1} has unsupported event "{event}"' + ) + command = hook.get("command") + if not isinstance(command, str) or not command: + raise ValueError( + f"{contract.label} hook {index + 1} requires a non-empty command" + ) + matcher = hook.get("matcher") + if matcher is not None and not isinstance(matcher, str): + raise ValueError( + f"{contract.label} hook {index + 1} matcher must be a string" + ) + timeout = hook.get("timeout") + if timeout is not None and ( + type(timeout) is not int or timeout < 1 or timeout > 600 + ): + raise ValueError( + f"{contract.label} hook {index + 1} timeout must be an integer from 1 to 600" + ) + return hook + + +def _read_hooks(value: Any, contract: KimiContract) -> List[TomlObject]: + if "hooks" not in value: + return [] + container = value["hooks"] + raw = _unwrap(container) + if not isinstance(raw, list): + raise ValueError(f'{contract.label} field "hooks" must be an array') + return [ + validate_kimi_hook(item, contract, index) + for index, item in enumerate(raw) + ] + + +def parse_kimi_document(contract: KimiContract, raw: str) -> KimiDocument: + try: + value = parse(raw) + except Exception as error: + raise ValueError( + f"Failed to parse {contract.label} at {contract.config_path}: {error}" + ) from error + hooks = _read_hooks(value, contract) + return KimiDocument( + contract=contract, + exists=True, + value=value, + hooks=hooks, + uses_hook_tables=isinstance(value.get("hooks"), AoT), + raw=raw, + ) + + +def create_kimi_document(contract: KimiContract) -> KimiDocument: + return KimiDocument(contract, False, document(), [], False) + + +def build_kimi_hook(event: ManagedEvent, script_path: str) -> TomlObject: + if event not in {"PreToolUse", "PostToolUse", "PostToolUseFailure"}: + raise ValueError(f"Unsupported managed Kimi event: {event}") + return { + "event": event, + "command": build_kimi_command(script_path), + "timeout": HOOK_TIMEOUT_SECONDS, + } + + +def _managed_reference( + hook: TomlObject, + event: ManagedEvent, + script_name: str, +) -> Optional[KimiRuntimeReference]: + if ( + set(hook) != {"event", "command", "timeout"} + or hook.get("event") != event + or type(hook.get("timeout")) is not int + or hook.get("timeout") != HOOK_TIMEOUT_SECONDS + or not isinstance(hook.get("command"), str) + ): + return None + return kimi_runtime_reference(str(hook["command"]), script_name) + + +def _managed_contract(event: Any) -> Optional[Tuple[ManagedEvent, str]]: + if event == "PreToolUse": + return "PreToolUse", GUARD_SCRIPT + if event == "PostToolUse": + return "PostToolUse", AUDIT_SCRIPT + if event == "PostToolUseFailure": + return "PostToolUseFailure", AUDIT_SCRIPT + return None + + +def remove_managed_kimi_hooks( + hooks: List[TomlObject], agent_id: str = "" +) -> List[TomlObject]: + result = [] + for hook in hooks: + expected = _managed_contract(hook.get("event")) + reference = ( + _managed_reference(hook, expected[0], expected[1]) + if expected is not None + else None + ) + remove = reference is not None and ( + not agent_id or same_kimi_agent_id(reference.agent_id, agent_id) + ) + if not remove: + result.append(hook) + return result + + +def _matched_indices( + original: List[TomlObject], desired: List[TomlObject] +) -> Tuple[set[int], set[int]]: + original_matches: set[int] = set() + desired_matches: set[int] = set() + start = 0 + for desired_index, desired_hook in enumerate(desired): + for original_index in range(start, len(original)): + if original[original_index] == desired_hook: + original_matches.add(original_index) + desired_matches.add(desired_index) + start = original_index + 1 + break + return original_matches, desired_matches + + +def _append_hook(container: Any, hook: TomlObject) -> None: + item = table() if isinstance(container, AoT) else inline_table() + for key, value in hook.items(): + item[key] = value + container.append(item) + + +def _render_hook_source( + source: KimiDocument, hooks: List[TomlObject] +) -> Optional[str]: + candidate = copy.deepcopy(source.value) + original_matches, desired_matches = _matched_indices(source.hooks, hooks) + if "hooks" in candidate: + container: Any = candidate["hooks"] + if not isinstance(container, (AoT, Array)): + raise ValueError(f'{source.contract.label} field "hooks" must be an array') + for index in range(len(source.hooks) - 1, -1, -1): + if index not in original_matches: + del container[index] + else: + container = aot() + candidate.add("hooks", container) + + for index, hook in enumerate(hooks): + if index not in desired_matches: + _append_hook(container, hook) + + if len(container) == 0 and source.uses_hook_tables: + del candidate["hooks"] + rendered = dumps(candidate) + if not rendered.strip(): + return None + parse_kimi_document(source.contract, rendered) + return rendered + + +def render_kimi_document( + source: KimiDocument, hooks: List[TomlObject] +) -> RenderedKimiDocument: + if not source.exists and not hooks: + return RenderedKimiDocument(source, False) + next_source = _render_hook_source(source, hooks) + return RenderedKimiDocument( + document=source, + changed=next_source != source.raw, + next_source=next_source, + ) + + +def _references_for_event( + hooks: List[TomlObject], event: ManagedEvent, script_name: str +) -> Dict[str, List[KimiRuntimeReference]]: + result: Dict[str, List[KimiRuntimeReference]] = {} + for hook in hooks: + reference = _managed_reference(hook, event, script_name) + if reference is None: + continue + key = os.path.normcase(reference.agent_id) + result.setdefault(key, []).append(reference) + return result + + +def kimi_runtime_contracts( + documents: List[KimiDocument], +) -> List[KimiRuntimeContract]: + contracts: List[KimiRuntimeContract] = [] + for source in documents: + guards = _references_for_event(source.hooks, "PreToolUse", GUARD_SCRIPT) + successes = _references_for_event( + source.hooks, "PostToolUse", AUDIT_SCRIPT + ) + failures = _references_for_event( + source.hooks, "PostToolUseFailure", AUDIT_SCRIPT + ) + for key, guard in guards.items(): + success = successes.get(key, []) + 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): + continue + contracts.append( + KimiRuntimeContract( + agent_id=guard[0].agent_id, + guard_path=guard[0].script_path, + audit_path=success[0].script_path, + config_path=source.contract.config_path, + ) + ) + return contracts diff --git a/sdks/python/elydora/plugins/kimi_installation.py b/sdks/python/elydora/plugins/kimi_installation.py new file mode 100644 index 0000000..6ae57df --- /dev/null +++ b/sdks/python/elydora/plugins/kimi_installation.py @@ -0,0 +1,195 @@ +"""Transactional Kimi hook and runtime installation.""" + +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 .base import InstallConfig +from .hook_template import generate_guard_script, 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, +) + + +@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) + + +def preflight_kimi_installation( + config: InstallConfig, documents: List[KimiDocument] +) -> KimiRuntimePaths: + 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) + 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]: + if not rendered.changed: + return None + source = rendered.document + return source_change( + source.contract.config_path, + source.contract.label, + source.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_kimi_installation( + config: InstallConfig, + paths: KimiRuntimePaths, + 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"), + 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] + + +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 commit_kimi_uninstall(changes: List[FileChange]) -> None: + write_changes(changes, "Uninstall Kimi hooks") diff --git a/sdks/python/elydora/plugins/kimi_io.py b/sdks/python/elydora/plugins/kimi_io.py new file mode 100644 index 0000000..6a0f4b4 --- /dev/null +++ b/sdks/python/elydora/plugins/kimi_io.py @@ -0,0 +1,276 @@ +"""Kimi configuration discovery and managed runtime I/O.""" + +from __future__ import annotations + +import base64 +import os +from typing import Any, List, Optional +import urllib.parse + +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, +) +from ._strict_json import JsonObject, parse_json_object +from .kimi_command import ( + runtime_root, + same_kimi_agent_id, + same_kimi_path, +) +from .kimi_contract import ( + AGENT_KEY, + LEGACY_EVENTS, + STABLE_EVENTS, + KimiContract, + KimiDocument, + KimiRuntimeContract, + create_kimi_document, + parse_kimi_document, +) + + +def _entry_exists(file_path: str, label: str) -> bool: + try: + os.lstat(file_path) + return True + except (FileNotFoundError, NotADirectoryError): + return False + except OSError as error: + raise OSError(f"Inspect {label} at {file_path}: {error}") from error + + +def _stable_contract(config_path: str) -> KimiContract: + return KimiContract( + generation="stable", + runtime_name="Kimi Code", + label="Kimi Code hooks config", + directory_label="Kimi Code home directory", + config_path=config_path, + events=frozenset(STABLE_EVENTS), + ) + + +def _legacy_contract(config_path: str) -> KimiContract: + return KimiContract( + generation="legacy", + runtime_name="kimi-cli", + label="kimi-cli legacy hooks config", + directory_label="kimi-cli legacy home directory", + config_path=config_path, + events=frozenset(LEGACY_EVENTS), + ) + + +def resolve_kimi_contracts() -> List[KimiContract]: + home = os.path.expanduser("~") + configured_home = os.environ.get("KIMI_CODE_HOME") + explicit_home = ( + None + if configured_home is None or configured_home == "" + else os.path.abspath(configured_home) + ) + stable_home = explicit_home or os.path.join(home, ".kimi-code") + 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): + return [stable] + + stable_detected = explicit_home is not None or _entry_exists( + stable_home, "Kimi Code home" + ) + legacy_detected = _entry_exists(legacy_home, "kimi-cli legacy home") + if legacy_detected and not stable_detected: + return [legacy] + return [stable, legacy] if legacy_detected else [stable] + + +def _read_kimi_document(contract: KimiContract) -> KimiDocument: + directory = os.path.dirname(contract.config_path) + physical_directory_exists(directory, contract.directory_label) + snapshot = read_physical_file(contract.config_path, contract.label) + if snapshot is None: + return create_kimi_document(contract) + return parse_kimi_document(contract, snapshot.contents) + + +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) diff --git a/sdks/python/elydora/plugins/registry.py b/sdks/python/elydora/plugins/registry.py index 4475cf7..de8e72f 100644 --- a/sdks/python/elydora/plugins/registry.py +++ b/sdks/python/elydora/plugins/registry.py @@ -63,7 +63,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "kimi": { "name": "Kimi Code", - "hook_event": "PreToolUse/PostToolUse", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", "config_path": "~/.kimi-code/config.toml", }, "letta": { diff --git a/sdks/python/tests/kimi_support.py b/sdks/python/tests/kimi_support.py new file mode 100644 index 0000000..54b9f92 --- /dev/null +++ b/sdks/python/tests/kimi_support.py @@ -0,0 +1,265 @@ +from __future__ import annotations + +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import json +import os +from pathlib import Path +import shlex +import subprocess # nosec B404 +import sys +from threading import Thread +from typing import Any, Mapping + +import pytest +from tomlkit import parse + +from elydora.plugins.kimi import KimiPlugin +from elydora.utils import base64url_encode + + +AGENT_ID = "agent-1" +MISSING = object() +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) + + +@dataclass +class KimiFixture: + root_dir: Path + home_dir: Path + project_dir: Path + kimi_home: Path + stable_path: Path + legacy_home: Path + legacy_path: Path + agent_dir: Path + guard_path: Path + hook_path: Path + runtime_config_path: Path + private_key_path: Path + plugin: KimiPlugin + config: dict[str, str] + + def install(self) -> None: + self.plugin.install(self.config) + + +class KimiApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + operation_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls) -> None: + cls.agent_status = "active" + cls.operation_status = 201 + cls.requests = [] + + def _respond(self, status: int, value: object) -> None: + body = json.dumps(value).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + type(self).requests.append( + { + "method": "GET", + "path": self.path, + "authorization": self.headers.get("Authorization"), + } + ) + self._respond(200, {"agent": {"status": type(self).agent_status}}) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length) + type(self).requests.append( + { + "method": "POST", + "path": self.path, + "authorization": self.headers.get("Authorization"), + "raw": raw, + "json": json.loads(raw), + } + ) + status = type(self).operation_status + value: object = ( + {"operation": {"accepted": True}} + if 200 <= status < 300 + else {"error": {"code": "UPSTREAM_FAILURE", "message": "failed"}} + ) + self._respond(status, value) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server( + *, status: str = "active", operation_status: int = 201 +) -> ThreadingHTTPServer: + KimiApiHandler.reset() + KimiApiHandler.agent_status = status + KimiApiHandler.operation_status = operation_status + server = ThreadingHTTPServer(("127.0.0.1", 0), KimiApiHandler) + Thread(target=server.serve_forever, daemon=True).start() + return server + + +def server_base_url(server: ThreadingHTTPServer) -> str: + return f"http://127.0.0.1:{server.server_port}" + + +def write_text(path: Path, source: str, mode: int = 0o600) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source, encoding="utf-8") + os.chmod(path, mode) + + +def write_optional(path: Path, value: object) -> None: + if value is MISSING: + return + write_text(path, str(value)) + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + stable_config: object = MISSING, + legacy_config: object = MISSING, + stable_detected: bool = True, + legacy_detected: bool = True, + explicit_kimi_home: bool = True, + base_url: str = "http://127.0.0.1:9", +) -> KimiFixture: + home_dir = tmp_path / "home with spaces and 'quote %ELYDORA_HOOK_PATH%" + project_dir = tmp_path / "project with spaces" + kimi_home = ( + home_dir / "custom kimi-code" + if explicit_kimi_home + else home_dir / ".kimi-code" + ) + stable_path = kimi_home / "config.toml" + legacy_home = home_dir / ".kimi" + legacy_path = legacy_home / "config.toml" + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / "guard.py" + hook_path = agent_dir / "hook.py" + project_dir.mkdir(parents=True) + if stable_detected and not explicit_kimi_home: + kimi_home.mkdir(parents=True) + if legacy_detected: + legacy_home.mkdir(parents=True) + write_optional(stable_path, stable_config) + write_optional(legacy_path, legacy_config) + + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.setenv("ELYDORA_HOOK_PATH", str(tmp_path / "expanded")) + if explicit_kimi_home: + monkeypatch.setenv("KIMI_CODE_HOME", str(kimi_home)) + else: + monkeypatch.delenv("KIMI_CODE_HOME", raising=False) + + config = { + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "kimi", + "private_key": VALID_PRIVATE_KEY, + "kid": "kid-1", + "token": "token-1", + "base_url": base_url, + "guard_script_path": str(guard_path), + } + return KimiFixture( + root_dir=tmp_path, + home_dir=home_dir, + project_dir=project_dir, + kimi_home=kimi_home, + stable_path=stable_path, + legacy_home=legacy_home, + legacy_path=legacy_path, + agent_dir=agent_dir, + guard_path=guard_path, + hook_path=hook_path, + runtime_config_path=agent_dir / "config.json", + private_key_path=agent_dir / "private.key", + plugin=KimiPlugin(), + config=config, + ) + + +def parsed_hooks(path: Path) -> list[dict[str, Any]]: + value = parse(path.read_text(encoding="utf-8"))["hooks"].unwrap() + assert isinstance(value, list) + return value + + +def managed_hook( + hooks: list[dict[str, Any]], event: str +) -> dict[str, Any]: + return next( + hook + for hook in reversed(hooks) + if hook.get("event") == event and hook.get("timeout") == 10 + ) + + +def assert_managed_hook(hook: Mapping[str, Any], event: str) -> None: + assert set(hook) == {"event", "command", "timeout"} + assert hook["event"] == event + assert hook["timeout"] == 10 + command = str(hook["command"]) + if os.name == "nt": + assert "powershell.exe" in command.lower() + assert " -EncodedCommand " in command + else: + assert command.startswith("'") + + +def assert_managed_triple(hooks: list[dict[str, Any]]) -> None: + for event in ("PreToolUse", "PostToolUse", "PostToolUseFailure"): + assert_managed_hook(managed_hook(hooks, event), event) + assert managed_hook(hooks, "PostToolUse")["command"] == managed_hook( + hooks, "PostToolUseFailure" + )["command"] + + +def run_hook( + command: str, + payload: str, + fixture: KimiFixture, + environment: Mapping[str, str] | None = None, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + shell=True, + input=payload, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env={ + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + **dict(environment or {}), + }, + text=True, + ) + + +def legacy_command(script_path: Path) -> str: + arguments = [sys.executable, str(script_path)] + return ( + subprocess.list2cmdline(arguments) + if os.name == "nt" + else shlex.join(arguments) + ) + + +def assert_no_transaction_files(root: Path) -> None: + assert not list(root.rglob("*.tmp")) + assert not list(root.rglob("*.rollback")) diff --git a/sdks/python/tests/test_kimi_plugin.py b/sdks/python/tests/test_kimi_plugin.py index 87793c9..10d07f7 100644 --- a/sdks/python/tests/test_kimi_plugin.py +++ b/sdks/python/tests/test_kimi_plugin.py @@ -1,423 +1,340 @@ from __future__ import annotations -from dataclasses import dataclass import json import os from pathlib import Path -import subprocess -from typing import Any import pytest -from tomlkit import parse +from tomlkit import dumps, parse from elydora import cli from elydora.plugins import kimi -from elydora.plugins.base import InstallConfig from elydora.plugins.registry import SUPPORTED_AGENTS +from kimi_support import ( + AGENT_ID, + MISSING, + assert_managed_triple, + assert_no_transaction_files, + legacy_command, + parsed_hooks, + prepare_fixture, +) -AGENT_ID = "agent-1" -MISSING = object() - - -@dataclass(frozen=True) -class KimiFixture: - plugin: kimi.KimiPlugin - config: InstallConfig - home_dir: Path - agent_dir: Path - modern_path: Path - legacy_path: Path - guard_path: Path - hook_path: Path - runtime_config_path: Path - private_key_path: Path - - -def write_optional(path: Path, value: object) -> None: - if value is MISSING: - return - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(str(value), encoding="utf-8") - - -def prepare_fixture( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - *, - modern_config: object = MISSING, - legacy_config: object = MISSING, - legacy_installed: bool = True, - explicit_kimi_home: bool = True, - create_guard: bool = True, -) -> KimiFixture: - home_dir = tmp_path / "home with spaces" - agent_dir = home_dir / ".elydora" / AGENT_ID - kimi_home = ( - home_dir / "custom kimi code" - if explicit_kimi_home - else home_dir / ".kimi-code" - ) - modern_path = kimi_home / "config.toml" - legacy_path = home_dir / ".kimi" / "config.toml" - guard_path = agent_dir / "guard.py" - hook_path = agent_dir / "hook.py" - agent_dir.mkdir(parents=True) - if create_guard: - guard_path.write_text( - "import sys\nsys.stdin.read()\n" - "sys.stderr.write('Agent is frozen by Elydora.')\n" - "raise SystemExit(2)\n", - encoding="utf-8", - ) - write_optional(modern_path, modern_config) - write_optional(legacy_path, legacy_config) - - monkeypatch.setattr(kimi, "ELYDORA_DIR", str(agent_dir.parent)) - monkeypatch.setattr(kimi, "_home_dir", lambda: str(home_dir)) - monkeypatch.setattr( - kimi.shutil, - "which", - lambda command: str(home_dir / "bin" / command) - if legacy_installed and command == "kimi-cli" - else None, - ) - if explicit_kimi_home: - monkeypatch.setenv("KIMI_CODE_HOME", str(kimi_home)) - else: - monkeypatch.delenv("KIMI_CODE_HOME", raising=False) - - config: InstallConfig = { - "agent_id": AGENT_ID, - "agent_name": "kimi", - "org_id": "org-1", - "private_key": "test-key", - "kid": "kid-1", - "base_url": "https://api.elydora.test", - "guard_script_path": str(guard_path), - } - return KimiFixture( - plugin=kimi.KimiPlugin(), - config=config, - home_dir=home_dir, - agent_dir=agent_dir, - modern_path=modern_path, - legacy_path=legacy_path, - guard_path=guard_path, - hook_path=hook_path, - runtime_config_path=agent_dir / "config.json", - private_key_path=agent_dir / "private.key", - ) - - -def parsed_hooks(path: Path) -> list[dict[str, Any]]: - document = parse(path.read_text(encoding="utf-8")) - value = document["hooks"].unwrap() - assert isinstance(value, list) - return value - - -def managed_hook( - hooks: list[dict[str, Any]], event: str, script_name: str -) -> dict[str, Any]: - return next( - hook - for hook in hooks - if hook.get("event") == event and script_name in str(hook.get("command")) - ) - - -def assert_strict_hook(hook: dict[str, Any]) -> None: - assert set(hook) == {"event", "command", "timeout"} - assert hook["timeout"] == 10 - - -def run_command( - command: str, home_dir: Path, payload: dict[str, Any] -) -> subprocess.CompletedProcess[str]: - return subprocess.run( - command, - shell=True, - capture_output=True, - check=False, - env={**os.environ, "HOME": str(home_dir), "USERPROFILE": str(home_dir)}, - input=json.dumps(payload), - text=True, - ) +STABLE_USER_CONFIG = ( + '# stable user config\ndefault_model = "kimi-code/k3"\n\n' + '[[hooks]]\nevent = "SessionStart"\ncommand = "existing-stable"\n' + "timeout = 30 # keep stable hook\n" +) +LEGACY_USER_CONFIG = ( + "# legacy user config\ntelemetry = false\n\n" + '[[hooks]]\nevent = "SessionEnd"\ncommand = "existing-legacy"\n' +) def test_kimi_is_registered_in_sdk_and_cli() -> None: assert SUPPORTED_AGENTS["kimi"] == { "name": "Kimi Code", - "hook_event": "PreToolUse/PostToolUse", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", "config_path": "~/.kimi-code/config.toml", } assert cli.PLUGIN_MAP["kimi"] is kimi.KimiPlugin -def test_install_preserves_both_configs_and_is_idempotent( +def test_install_writes_exact_triples_to_both_configs_and_is_idempotent( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: - modern = ( - '# modern user config\ndefault_model = "kimi-code/k3"\n\n' - '[[hooks]]\nevent = "SessionStart"\ncommand = "existing-modern"\n' - "timeout = 30 # keep modern hook\n" - ) - legacy = ( - "# legacy user config\ntelemetry = false\n\n" - '[[hooks]]\nevent = "SessionEnd"\ncommand = "existing-legacy"\n' - ) fixture = prepare_fixture( monkeypatch, tmp_path, - modern_config=modern, - legacy_config=legacy, + stable_config=STABLE_USER_CONFIG, + legacy_config=LEGACY_USER_CONFIG, ) - fixture.plugin.install(fixture.config) - fixture.plugin.install(fixture.config) + fixture.install() + first_sources = { + path: path.read_text(encoding="utf-8") + for path in (fixture.stable_path, fixture.legacy_path) + } + fixture.install() assert "Kimi Code and kimi-cli" in capsys.readouterr().out for path, comment, command in ( - (fixture.modern_path, "# modern user config", "existing-modern"), + (fixture.stable_path, "# stable user config", "existing-stable"), (fixture.legacy_path, "# legacy user config", "existing-legacy"), ): raw = path.read_text(encoding="utf-8") hooks = parsed_hooks(path) + assert raw == first_sources[path] assert comment in raw - assert command in raw - assert len(hooks) == 3 - assert_strict_hook(managed_hook(hooks, "PreToolUse", "guard.py")) - assert_strict_hook(managed_hook(hooks, "PostToolUse", "hook.py")) - assert (fixture.home_dir / ".kimi-code" / "config.toml").exists() is False + assert any(hook.get("command") == command for hook in hooks) + assert len(hooks) == 4 + assert_managed_triple(hooks) + assert not (fixture.home_dir / ".kimi-code" / "config.toml").exists() def test_install_preserves_inline_hook_array_style( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - modern = ( + source = ( '# inline user hook\nhooks = [{ event = "SessionStart", ' 'command = "existing-inline" }] # keep array\n' ) fixture = prepare_fixture( monkeypatch, tmp_path, - modern_config=modern, - legacy_installed=False, + stable_config=source, + legacy_detected=False, ) - fixture.plugin.install(fixture.config) - fixture.plugin.install(fixture.config) + fixture.install() + fixture.install() - raw = fixture.modern_path.read_text(encoding="utf-8") + raw = fixture.stable_path.read_text(encoding="utf-8") assert "# inline user hook" in raw assert "# keep array" in raw assert "hooks = [" in raw - assert len(parsed_hooks(fixture.modern_path)) == 3 + hooks = parsed_hooks(fixture.stable_path) + assert len(hooks) == 4 + assert_managed_triple(hooks) -def test_modern_install_avoids_false_legacy_migration_source( +def test_uninstall_preserves_user_owned_empty_inline_hook_array( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path, legacy_installed=False) - fixture.plugin.install(fixture.config) + source = "# user container\nhooks = [] # keep empty array\n" + fixture = prepare_fixture( + monkeypatch, + tmp_path, + stable_config=source, + legacy_detected=False, + ) + fixture.install() + fixture.plugin.uninstall(AGENT_ID) - assert len(parsed_hooks(fixture.modern_path)) == 2 - assert fixture.legacy_path.exists() is False + raw = fixture.stable_path.read_text(encoding="utf-8") + assert "# user container" in raw + assert "# keep empty array" in raw + assert parse(raw)["hooks"].unwrap() == [] -def test_empty_home_override_uses_official_default( +def test_fresh_stable_install_leaves_legacy_source_absent( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + fixture.install() + + hooks = parsed_hooks(fixture.stable_path) + assert len(hooks) == 3 + assert_managed_triple(hooks) + assert not fixture.legacy_path.exists() + + +def test_empty_home_override_uses_documented_default( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: fixture = prepare_fixture( monkeypatch, tmp_path, explicit_kimi_home=False, - legacy_installed=False, + legacy_detected=False, ) - fixture.plugin.install(fixture.config) + fixture.install() monkeypatch.setenv("KIMI_CODE_HOME", "") assert fixture.plugin.status()["installed"] is True + assert_managed_triple(parsed_hooks(fixture.stable_path)) -def test_legacy_install_avoids_premature_modern_migration_target( +def test_legacy_only_install_leaves_stable_target_absent( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path, explicit_kimi_home=False) - fixture.plugin.install(fixture.config) - - assert len(parsed_hooks(fixture.legacy_path)) == 2 - assert fixture.modern_path.exists() is False - monkeypatch.setattr(kimi.shutil, "which", lambda command: None) - assert fixture.plugin.status()["installed"] is True - - -def test_commands_block_and_forward_official_payload( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - capture_path = tmp_path / "captured-event.json" - fixture.hook_path.write_text( - "from pathlib import Path\nimport sys\n" - f"Path({str(capture_path)!r}).write_text(sys.stdin.read(), encoding='utf-8')\n", - encoding="utf-8", - ) - hooks = parsed_hooks(fixture.modern_path) - payload = { - "hook_event_name": "PreToolUse", - "session_id": "session-1", - "cwd": str(fixture.home_dir), - "tool_name": "Bash", - "tool_input": {"command": "echo test"}, - "tool_call_id": "call-1", - } - - guard = managed_hook(hooks, "PreToolUse", "guard.py") - guard_result = run_command(guard["command"], fixture.home_dir, payload) - assert guard_result.returncode == 2 - assert "Agent is frozen by Elydora" in guard_result.stderr - - payload["hook_event_name"] = "PostToolUse" - payload["tool_output"] = {"output": "test"} - audit = managed_hook(hooks, "PostToolUse", "hook.py") - audit_result = run_command(audit["command"], fixture.home_dir, payload) - assert audit_result.returncode == 0 - assert json.loads(capture_path.read_text(encoding="utf-8")) == payload - - -def test_status_accepts_either_contract_and_requires_both_scripts( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - fixture.modern_path.unlink() - assert fixture.plugin.status()["installed"] is True - - fixture.plugin.install(fixture.config) - fixture.legacy_path.unlink() - assert fixture.plugin.status()["installed"] is True - - fixture.guard_path.unlink() - assert fixture.plugin.status()["installed"] is False - - -def test_uninstall_preserves_user_hooks_in_both_contracts( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - user_hook = ( - "# user hook\n[[hooks]]\nevent = \"SessionStart\"\n" - 'command = "existing-command"\ntimeout = 30 # keep timeout\n' - ) fixture = prepare_fixture( monkeypatch, tmp_path, - modern_config=user_hook, - legacy_config=user_hook, + explicit_kimi_home=False, + stable_detected=False, ) - fixture.plugin.install(fixture.config) + fixture.install() - fixture.plugin.uninstall(AGENT_ID) - - for path in (fixture.modern_path, fixture.legacy_path): - raw = path.read_text(encoding="utf-8") - assert "# user hook" in raw - assert "# keep timeout" in raw - assert parsed_hooks(path) == [{ - "event": "SessionStart", - "command": "existing-command", - "timeout": 30, - }] + assert_managed_triple(parsed_hooks(fixture.legacy_path)) + assert not fixture.stable_path.exists() + assert fixture.plugin.status()["installed"] is True -def test_uninstall_removes_configs_created_by_elydora( +def test_install_parses_every_selected_config_before_runtime_writes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - - fixture.plugin.uninstall(AGENT_ID) - - assert fixture.modern_path.exists() is False - assert fixture.legacy_path.exists() is False - - -def test_install_parses_every_config_before_runtime_writes( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - modern = '# untouched modern\ndefault_model = "kimi-code/k3"\n' fixture = prepare_fixture( monkeypatch, tmp_path, - modern_config=modern, + stable_config=STABLE_USER_CONFIG, legacy_config="[malformed", ) with pytest.raises(ValueError, match="parse kimi-cli legacy hooks config"): - fixture.plugin.install(fixture.config) + fixture.install() - assert fixture.modern_path.read_text(encoding="utf-8") == modern + assert fixture.stable_path.read_text(encoding="utf-8") == STABLE_USER_CONFIG assert fixture.legacy_path.read_text(encoding="utf-8") == "[malformed" - assert fixture.hook_path.exists() is False - assert fixture.runtime_config_path.exists() is False - assert fixture.private_key_path.exists() is False + assert not fixture.agent_dir.exists() -def test_install_rejects_invalid_hook_shape_without_writes( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path +@pytest.mark.parametrize( + ("stable_config", "legacy_config", "pattern"), + [ + ( + '[[hooks]]\nevent = "PreToolUse"\ncommand = "existing"\n' + 'cwd = "/tmp"\n', + None, + 'unsupported field "cwd"', + ), + ( + None, + '[[hooks]]\nevent = "Interrupt"\ncommand = "existing"\n', + 'unsupported event "Interrupt"', + ), + ( + '[[hooks]]\nevent = "PreToolUse"\ncommand = "existing"\n' + "timeout = 601\n", + None, + "integer from 1 to 600", + ), + ], +) +def test_install_rejects_fields_and_events_outside_each_contract( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + stable_config: str | None, + legacy_config: str | None, + pattern: str, ) -> None: - modern = ( - '[[hooks]]\nevent = "PreToolUse"\ncommand = "existing-command"\n' - 'cwd = "/tmp"\n' - ) fixture = prepare_fixture( monkeypatch, tmp_path, - modern_config=modern, - legacy_installed=False, + stable_config=stable_config if stable_config is not None else MISSING, + legacy_config=legacy_config if legacy_config is not None else MISSING, ) - with pytest.raises(ValueError, match='unsupported field "cwd"'): - fixture.plugin.install(fixture.config) + with pytest.raises(ValueError, match=pattern): + fixture.install() - assert fixture.modern_path.read_text(encoding="utf-8") == modern - assert fixture.hook_path.exists() is False - assert fixture.runtime_config_path.exists() is False + assert not fixture.agent_dir.exists() -def test_install_rejects_missing_guard_before_creating_files( +def test_install_migrates_exact_legacy_commands_and_preserves_lookalikes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path, create_guard=False) + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + lookalike = f"{legacy_command(fixture.guard_path)} --inspect" + source = ( + '[[hooks]]\nevent = "PreToolUse"\n' + f"command = {json.dumps(legacy_command(fixture.guard_path))}\n" + "timeout = 10\n\n" + '[[hooks]]\nevent = "PostToolUse"\n' + f"command = {json.dumps(legacy_command(fixture.hook_path))}\n" + "timeout = 10\n\n" + '[[hooks]]\nevent = "PreToolUse"\n' + f"command = {json.dumps(lookalike)}\ntimeout = 10\n" + ) + fixture.stable_path.parent.mkdir(parents=True, exist_ok=True) + fixture.stable_path.write_text(source, encoding="utf-8") - with pytest.raises(FileNotFoundError, match="guard runtime is missing"): - fixture.plugin.install(fixture.config) + fixture.install() - assert fixture.modern_path.exists() is False - assert fixture.legacy_path.exists() is False - assert fixture.hook_path.exists() is False - assert fixture.runtime_config_path.exists() is False + hooks = parsed_hooks(fixture.stable_path) + assert len(hooks) == 4 + assert any(hook.get("command") == lookalike for hook in hooks) + assert_managed_triple(hooks) + + fixture.plugin.uninstall(AGENT_ID) + assert parsed_hooks(fixture.stable_path) == [ + {"event": "PreToolUse", "command": lookalike, "timeout": 10} + ] -def test_status_surfaces_malformed_runtime_metadata( +def test_uninstall_preserves_user_configs_and_removes_managed_configs( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) - fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") + fixture = prepare_fixture( + monkeypatch, + tmp_path, + stable_config=STABLE_USER_CONFIG, + legacy_config=LEGACY_USER_CONFIG, + ) + fixture.install() + fixture.plugin.uninstall(AGENT_ID) - with pytest.raises(ValueError, match="parse Elydora runtime config"): + assert "# keep stable hook" in fixture.stable_path.read_text(encoding="utf-8") + assert "# legacy user config" in fixture.legacy_path.read_text(encoding="utf-8") + assert [hook["command"] for hook in parsed_hooks(fixture.stable_path)] == [ + "existing-stable" + ] + assert [hook["command"] for hook in parsed_hooks(fixture.legacy_path)] == [ + "existing-legacy" + ] + + managed = prepare_fixture( + monkeypatch, + tmp_path / "managed", + ) + managed.install() + managed.plugin.uninstall(AGENT_ID) + assert not managed.stable_path.exists() + assert not managed.legacy_path.exists() + + +def test_status_requires_complete_triple_runtime_identity_and_private_key( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + fixture.install() + assert fixture.plugin.status()["installed"] is True + + document = parse(fixture.stable_path.read_text(encoding="utf-8")) + hooks = document["hooks"] + for index in range(len(hooks) - 1, -1, -1): + if hooks[index]["event"] == "PostToolUseFailure": + del hooks[index] + fixture.stable_path.write_text(dumps(document), encoding="utf-8") + assert fixture.plugin.status()["installed"] is False + + fixture.install() + fixture.private_key_path.write_text("invalid", encoding="utf-8") + with pytest.raises(ValueError, match="private key.*canonical 32-byte"): fixture.plugin.status() -def test_atomic_writes_leave_no_temporary_files( +@pytest.mark.parametrize( + "runtime_name", + ["guard.py", "hook.py", "config.json", "private.key"], +) +def test_status_requires_every_runtime_file( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + runtime_name: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + fixture.install() + (fixture.agent_dir / runtime_name).unlink() + + assert fixture.plugin.status()["installed"] is False + + +def test_installation_leaves_no_transaction_files( monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: fixture = prepare_fixture(monkeypatch, tmp_path) - fixture.plugin.install(fixture.config) + fixture.install() - for directory in (fixture.modern_path.parent, fixture.legacy_path.parent): - assert all(path.suffix != ".tmp" for path in directory.iterdir()) + assert_no_transaction_files(fixture.home_dir) + if os.name != "nt": + for path in ( + fixture.stable_path, + fixture.legacy_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert path.stat().st_mode & 0o777 == 0o600 diff --git a/sdks/python/tests/test_kimi_runtime.py b/sdks/python/tests/test_kimi_runtime.py new file mode 100644 index 0000000..0f16149 --- /dev/null +++ b/sdks/python/tests/test_kimi_runtime.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from kimi_support import ( + KimiApiHandler, + managed_hook, + parsed_hooks, + prepare_fixture, + run_hook, + server_base_url, + start_api_server, +) + + +PRE_PAYLOAD = { + "hook_event_name": "PreToolUse", + "session_id": "session-1", + "cwd": "C:/project", + "tool_name": "Bash", + "tool_input": {"command": "echo test"}, + "tool_call_id": "call-1", +} + + +def test_runtimes_enforce_active_state_and_preserve_success_and_failure_payloads( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server() + fixture = prepare_fixture( + monkeypatch, + tmp_path, + base_url=server_base_url(server), + legacy_detected=False, + ) + try: + fixture.install() + hooks = parsed_hooks(fixture.stable_path) + guard = managed_hook(hooks, "PreToolUse") + success_audit = managed_hook(hooks, "PostToolUse") + failure_audit = managed_hook(hooks, "PostToolUseFailure") + + guard_result = run_hook( + str(guard["command"]), + json.dumps(PRE_PAYLOAD), + fixture, + {"ELYDORA_HOOK_PATH": "injected-command-fragment"}, + ) + assert guard_result.returncode == 0, guard_result.stderr + + success_payload = { + **PRE_PAYLOAD, + "hook_event_name": "PostToolUse", + "tool_output": "test\n", + } + success = run_hook( + str(success_audit["command"]), + json.dumps(success_payload), + fixture, + ) + assert success.returncode == 0, success.stderr + + failure_payload = { + **PRE_PAYLOAD, + "hook_event_name": "PostToolUseFailure", + "error": { + "name": "ToolError", + "message": "command failed", + "code": "tool.failed", + }, + } + failure = run_hook( + str(failure_audit["command"]), + json.dumps(failure_payload), + fixture, + ) + assert failure.returncode == 0, failure.stderr + + operations = [ + request["json"] + for request in KimiApiHandler.requests + if request["method"] == "POST" + ] + assert len(operations) == 2 + assert operations[0]["payload"] == success_payload + assert operations[1]["payload"] == failure_payload + assert operations[0]["subject"]["session_id"] == "session-1" + assert operations[1]["action"]["tool"] == "Bash" + finally: + server.shutdown() + server.server_close() + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_guard_propagates_exit_code_two_for_blocked_agents( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + status: str, +) -> None: + server = start_api_server(status=status) + fixture = prepare_fixture( + monkeypatch, + tmp_path, + base_url=server_base_url(server), + legacy_detected=False, + ) + try: + fixture.install() + command = managed_hook( + parsed_hooks(fixture.stable_path), "PreToolUse" + )["command"] + result = run_hook(str(command), json.dumps(PRE_PAYLOAD), fixture) + assert result.returncode == 2 + assert status in result.stderr.lower() + finally: + server.shutdown() + server.server_close() + + +def test_audit_runtime_records_malformed_input_and_api_failures( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server(operation_status=503) + fixture = prepare_fixture( + monkeypatch, + tmp_path, + base_url=server_base_url(server), + legacy_detected=False, + ) + try: + fixture.install() + command = managed_hook( + parsed_hooks(fixture.stable_path), "PostToolUse" + )["command"] + + malformed = run_hook(str(command), "{ malformed", fixture) + assert malformed.returncode == 0 + assert malformed.stderr == "" + + upstream = run_hook( + str(command), + json.dumps( + { + **PRE_PAYLOAD, + "hook_event_name": "PostToolUse", + "tool_output": "test", + } + ), + fixture, + ) + assert upstream.returncode == 0 + assert upstream.stderr == "" + log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8") + assert "invalid JSON" in log + assert "HTTP 503" in log + finally: + server.shutdown() + server.server_close() diff --git a/sdks/python/tests/test_kimi_security.py b/sdks/python/tests/test_kimi_security.py new file mode 100644 index 0000000..983e114 --- /dev/null +++ b/sdks/python/tests/test_kimi_security.py @@ -0,0 +1,197 @@ +from __future__ import annotations + +import os +from pathlib import Path +import shutil +import subprocess # nosec B404 +import sys + +import pytest + +from kimi_support import ( + AGENT_ID, + VALID_PRIVATE_KEY, + prepare_fixture, + write_text, +) + + +def symlink_or_skip( + target: Path, link: Path, *, directory: bool = False +) -> None: + try: + link.symlink_to(target, target_is_directory=directory) + except OSError as error: + pytest.skip(f"Symbolic links unavailable: {error}") + + +def test_install_rejects_linked_config_before_creating_runtimes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + target = tmp_path / "config-target.toml" + source = "# protected target\ntelemetry = false\n" + write_text(target, source) + fixture.stable_path.parent.mkdir(parents=True) + symlink_or_skip(target, fixture.stable_path) + + with pytest.raises(OSError, match="not a physical file"): + fixture.install() + + assert target.read_text(encoding="utf-8") == source + assert not fixture.agent_dir.exists() + + +@pytest.mark.parametrize("kind", ["config", "runtime"]) +def test_install_rejects_linked_config_and_runtime_directories( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + target = tmp_path / f"{kind}-target" + target.mkdir() + link = fixture.kimi_home if kind == "config" else fixture.agent_dir.parent + if link.exists(): + shutil.rmtree(link) + link.parent.mkdir(parents=True, exist_ok=True) + symlink_or_skip(target, link, directory=True) + + with pytest.raises(OSError, match="physical directory"): + fixture.install() + + assert list(target.iterdir()) == [] + untouched = fixture.agent_dir if kind == "config" else fixture.stable_path + assert not untouched.exists() + + +def test_status_rejects_linked_private_key( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + fixture.install() + target = tmp_path / "private-key-target" + source = fixture.private_key_path.read_text(encoding="utf-8") + write_text(target, source) + fixture.private_key_path.unlink() + symlink_or_skip(target, fixture.private_key_path) + + with pytest.raises(OSError, match="not a physical file"): + fixture.plugin.status() + + assert target.read_text(encoding="utf-8") == source + + +@pytest.mark.parametrize("kind", ["orphaned", "mismatched"]) +def test_install_rejects_unverifiable_runtime_identity_before_config_writes( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + kind: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + fixture.agent_dir.mkdir(parents=True) + if kind == "orphaned": + write_text(fixture.guard_path, "orphaned guard\n", 0o700) + else: + write_text( + fixture.runtime_config_path, + '{"agent_name":"kimi","agent_id":"another-agent"}\n', + ) + + pattern = "identity cannot be verified" if kind == "orphaned" else "identity does not match" + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert not fixture.stable_path.exists() + + +@pytest.mark.parametrize( + ("field", "value", "pattern"), + [ + ("agent_name", "codex", "requires agent_name kimi"), + ("agent_id", "../escape", "Invalid agent ID"), + ("private_key", "invalid", "canonical 32-byte"), + ("token", "", "non-empty string"), + ( + "base_url", + "https://api.elydora.com/path?token=secret", + "query parameters", + ), + ("guard_script_path", "outside", "managed agent directory"), + ], +) +def test_install_validates_runtime_inputs_before_creating_files( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + field: str, + value: str, + pattern: str, +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + fixture.config[field] = ( + str(fixture.home_dir / "outside" / "guard.py") + if field == "guard_script_path" + else value + ) + + with pytest.raises(ValueError, match=pattern): + fixture.install() + + assert not fixture.stable_path.exists() + assert not fixture.agent_dir.exists() + + +def test_status_surfaces_malformed_runtime_metadata( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + fixture.install() + fixture.runtime_config_path.write_text("{ malformed", encoding="utf-8") + + with pytest.raises(ValueError, match="parse Elydora runtime config"): + fixture.plugin.status() + + +def test_cli_preflight_preserves_malformed_config_before_runtime_creation( + tmp_path: Path, +) -> None: + home_dir = tmp_path / "cli-home" + config_path = home_dir / ".kimi-code" / "config.toml" + key_path = tmp_path / "private.key" + write_text(config_path, "[malformed") + write_text(key_path, VALID_PRIVATE_KEY) + result = subprocess.run( + [ + sys.executable, + "-m", + "elydora.cli", + "install", + "--agent", + "kimi", + "--org_id", + "org-1", + "--agent_id", + AGENT_ID, + "--private_key_file", + str(key_path), + "--kid", + "kid-1", + "--base_url", + "https://api.elydora.com", + ], + capture_output=True, + check=False, + cwd=Path(__file__).parents[1], + env={ + **os.environ, + "HOME": str(home_dir), + "USERPROFILE": str(home_dir), + "KIMI_CODE_HOME": "", + }, + text=True, + ) + + assert result.returncode == 1 + assert "parse Kimi Code hooks config" in result.stderr + assert config_path.read_text(encoding="utf-8") == "[malformed" + assert not (home_dir / ".elydora").exists() diff --git a/sdks/python/tests/test_kimi_transaction.py b/sdks/python/tests/test_kimi_transaction.py new file mode 100644 index 0000000..09b4d12 --- /dev/null +++ b/sdks/python/tests/test_kimi_transaction.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from elydora.plugins import _transaction, kimi_installation +from elydora.plugins.kimi_contract import ( + build_kimi_hook, + remove_managed_kimi_hooks, + render_kimi_document, +) +from elydora.plugins.kimi_io import read_kimi_documents +from kimi_support import ( + AGENT_ID, + assert_no_transaction_files, + prepare_fixture, + write_text, +) + + +RUNTIME_NAMES = ("guard.py", "config.json", "private.key", "hook.py") + + +def snapshot_files(paths: list[Path]) -> dict[Path, str]: + return {path: path.read_text(encoding="utf-8") for path in paths} + + +def assert_snapshot(snapshot: dict[Path, str]) -> None: + for path, source in snapshot.items(): + assert path.read_text(encoding="utf-8") == source + + +def test_install_restores_both_configs_when_final_commit_fails( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + stable = '# stable owner config\ndefault_model = "kimi-code/k3"\n' + legacy = "# legacy owner config\ntelemetry = false\n" + fixture = prepare_fixture( + monkeypatch, + tmp_path, + stable_config=stable, + legacy_config=legacy, + ) + real_replace = _transaction.os.replace + failed = False + + def fail_legacy_once(source: str, destination: str) -> None: + nonlocal failed + if os.path.abspath(destination) == str(fixture.legacy_path) and not failed: + failed = True + raise OSError("injected legacy config failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_legacy_once) + + with pytest.raises(OSError, match="injected legacy config failure"): + fixture.install() + + assert fixture.stable_path.read_text(encoding="utf-8") == stable + assert fixture.legacy_path.read_text(encoding="utf-8") == legacy + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_install_rejects_concurrent_config_change_before_runtime_staging( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + original = '# owner config\ndefault_model = "kimi-code/k3"\n' + concurrent = "# concurrent owner change\ntelemetry = false\n" + fixture = prepare_fixture( + monkeypatch, + tmp_path, + stable_config=original, + legacy_detected=False, + ) + real_rendered_change = kimi_installation.rendered_change + changed = False + + def mutate_before_change(rendered: object): + nonlocal changed + if not changed: + changed = True + write_text(fixture.stable_path, concurrent) + return real_rendered_change(rendered) # type: ignore[arg-type] + + monkeypatch.setattr(kimi_installation, "rendered_change", mutate_before_change) + + with pytest.raises(OSError, match="changed before staging"): + fixture.install() + + assert fixture.stable_path.read_text(encoding="utf-8") == concurrent + assert not fixture.agent_dir.exists() + assert_no_transaction_files(fixture.home_dir) + + +def test_uninstall_restores_first_config_when_second_removal_fails( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + paths = [ + *(fixture.agent_dir / name for name in RUNTIME_NAMES), + fixture.stable_path, + fixture.legacy_path, + ] + before = snapshot_files(paths) + real_replace = _transaction.os.replace + removal_commits = 0 + + def fail_second_removal(source: str, destination: str) -> None: + nonlocal removal_commits + if destination.endswith(".rollback"): + removal_commits += 1 + if removal_commits == 2: + raise OSError("injected uninstall failure") + real_replace(source, destination) + + monkeypatch.setattr(_transaction.os, "replace", fail_second_removal) + + with pytest.raises(OSError, match="injected uninstall failure"): + fixture.plugin.uninstall(AGENT_ID) + + assert_snapshot(before) + assert_no_transaction_files(fixture.home_dir) + + +def test_prepared_install_detects_change_before_commit_and_rolls_back_staging( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path, legacy_detected=False) + documents = read_kimi_documents() + paths = kimi_installation.preflight_kimi_installation( + fixture.config, documents + ) + rendered = [ + render_kimi_document( + document, + [ + *remove_managed_kimi_hooks(document.hooks), + build_kimi_hook("PreToolUse", paths.guard_path), + build_kimi_hook("PostToolUse", paths.audit_path), + build_kimi_hook("PostToolUseFailure", paths.audit_path), + ], + ) + for document in documents + ] + changes = kimi_installation.prepare_kimi_installation( + fixture.config, paths, rendered + ) + write_text(fixture.stable_path, "# changed after preparation\n") + + with pytest.raises(OSError, match="changed during installation"): + kimi_installation.commit_kimi_installation(changes) + + for name in RUNTIME_NAMES: + assert not (fixture.agent_dir / name).exists() + assert fixture.stable_path.read_text(encoding="utf-8") == "# changed after preparation\n" + assert_no_transaction_files(fixture.home_dir)