diff --git a/sdks/python/README.md b/sdks/python/README.md index 3540502..4307e4b 100644 --- a/sdks/python/README.md +++ b/sdks/python/README.md @@ -98,7 +98,7 @@ Auggie installation commits `~/.augment/settings.json`, both platform wrappers, Factory Droid 0.175.0 installation resolves one active user hook source in this order: `~/.factory/hooks.json`, the legacy `~/.factory/hooks/hooks.json`, then the `hooks` field from user `settings.local.json` and `settings.json`. Current hook files use the documented top-level `hooks` container; legacy direct event maps remain readable until Factory migrates them. Elydora writes exact `PreToolUse` and `PostToolUse` groups to that source, removes exact Elydora ownership from inactive user sources, and keeps project, folder, plugin, and organization hook definitions read-only. System, project, folder, and user `hooksDisabled` values follow Factory's extension-only precedence; system-managed `allowManagedHooksOnly` blocks a user-hook installation. Hook sources, generated runtimes, runtime config, and private key commit through one rollback-capable transaction. Managed commands preserve the complete native snake_case payload, propagate frozen or revoked state through exit code `2`, and use Factory's native PowerShell call operator with explicit `$LASTEXITCODE` propagation on Windows. Run `/hooks` after installation to refresh Droid's session snapshot and verify remote organization policy. -Qwen Code installation resolves its user settings through explicit `QWEN_HOME`, `~/.qwen/.env`, then `~/.env`. Elydora preserves comments and unrelated settings, leaves workspace settings unchanged, and uses Qwen's explicit Bash or PowerShell command contract with millisecond timeouts. Run `/hooks` after installation to review the changes. +Qwen Code 0.20.0 installation writes exact, named `PreToolUse`, `PostToolUse`, and `PostToolUseFailure` command hooks to `$QWEN_HOME/settings.json` (default `~/.qwen/settings.json`). Explicit environment values lead Qwen's home bootstrap, followed by `/.env` or `~/.qwen/.env`, then `~/.env`; a discovered home contributes its own `.env`. System defaults, trusted workspace settings, system overrides, and trust rules remain read-only and participate in effective `disableAllHooks` evaluation. User settings, generated runtimes, runtime config, and private key commit through one rollback-capable transaction guarded by snapshots of every consulted source. Managed handlers use ten-second millisecond timeouts, preserve Qwen's complete native snake_case payload for success and failure events, and propagate frozen or revoked state through exit code `2`. Run `/hooks` after installation and restart active Qwen Code sessions. Kiro CLI installation covers both runtime contracts. Kiro CLI v2 uses the generated custom agent through `kiro-cli --agent elydora-audit`. Kiro CLI v3 loads the global standalone hooks when started with `kiro-cli --v3`. diff --git a/sdks/python/elydora/plugins/qwen.py b/sdks/python/elydora/plugins/qwen.py index f5ef9d7..1ee145f 100644 --- a/sdks/python/elydora/plugins/qwen.py +++ b/sdks/python/elydora/plugins/qwen.py @@ -1,161 +1,86 @@ -"""Qwen Code native global user-hook integration.""" +"""Qwen Code 0.20 native global user-hook integration.""" from __future__ import annotations -import json -import os -from typing import List, Optional +from typing import Dict -from ._transaction import FileChange, file_change, write_changes from .base import AgentPlugin, InstallConfig, PluginStatus -from .hook_template import generate_hook_script -from .qwen_config import render_document +from .qwen_config import render_qwen_document from .qwen_contract import ( AGENT_KEY, - AUDIT_SCRIPT, - GUARD_SCRIPT, - build_group, - elydora_dir, - runtime_contracts, - validate_javascript_regexes, + AUDIT_HOOK_NAME, + GUARD_HOOK_NAME, + JsonObject, + build_qwen_group, + qwen_runtime_contracts, ) -from .qwen_io import ( - read_document, - rendered_change, - require_runtime, - runtime_files_exist, +from .qwen_installation import ( + QwenRuntimePaths, + commit_qwen_installation, + commit_qwen_uninstall, + preflight_qwen_installation, + prepare_qwen_installation, + prepare_qwen_uninstall, ) +from .qwen_io import qwen_runtime_files_exist +from .qwen_sources import read_qwen_sources -def _agent_directory(agent_id: str) -> str: - if ( - not agent_id - or agent_id in {".", ".."} - or os.path.basename(agent_id) != agent_id - or os.path.isabs(agent_id) - ): - raise ValueError("agent_id must be a single non-empty path segment") - return os.path.join(elydora_dir(), agent_id) - - -def _same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def _append_change(changes: List[FileChange], change: Optional[FileChange]) -> None: - if change is not None: - changes.append(change) +def _installed_groups(paths: QwenRuntimePaths) -> Dict[str, JsonObject]: + return { + "PreToolUse": build_qwen_group(paths.guard_path, GUARD_HOOK_NAME), + "PostToolUse": build_qwen_group(paths.audit_path, AUDIT_HOOK_NAME), + "PostToolUseFailure": build_qwen_group( + paths.audit_path, AUDIT_HOOK_NAME + ), + } class QwenPlugin(AgentPlugin): - """Install Elydora into Qwen Code's global user settings.""" + """Install Elydora into Qwen Code's native user settings.""" + + manages_guard_runtime = True + + def preflight_install(self, config: InstallConfig) -> None: + sources = read_qwen_sources() + preflight_qwen_installation(config, sources) def install(self, config: InstallConfig) -> None: - agent_id = config.get("agent_id", "") - agent_directory = _agent_directory(agent_id) - document = read_document() - validate_javascript_regexes([document.hooks]) - - guard_path = config.get("guard_script_path", "") - expected_guard_path = os.path.join(agent_directory, GUARD_SCRIPT) - if not _same_path(guard_path, expected_guard_path): - raise ValueError( - "Elydora guard runtime must use the managed agent directory: " - f"{expected_guard_path}" - ) - require_runtime(guard_path, "Elydora guard runtime") - audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) - rendered = render_document( - document, + sources = read_qwen_sources() + paths = preflight_qwen_installation(config, sources) + rendered = render_qwen_document( + sources.user, None, - { - "PreToolUse": build_group(guard_path), - "PostToolUse": build_group(audit_path), - }, + _installed_groups(paths), ) - - runtime_config = { - "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: List[FileChange] = [] - _append_change( - changes, - file_change( - os.path.join(agent_directory, "config.json"), - "Elydora runtime config", - json.dumps(runtime_config, indent=2) + "\n", - 0o600, - ), - ) - _append_change( - changes, - file_change( - os.path.join(agent_directory, "private.key"), - "Elydora private key", - config.get("private_key", ""), - 0o600, - ), - ) - _append_change( - changes, - file_change( - audit_path, - "Elydora audit runtime", - audit_script, - 0o700, - ), - ) - _append_change(changes, rendered_change(rendered)) - write_changes(changes, "Write Qwen Code installation") - - print(f"Qwen Code: user hooks installed at {document.file_path}") - print("Qwen Code: run /hooks to review the Elydora hook changes.") + changes = prepare_qwen_installation(config, paths, rendered) + commit_qwen_installation(changes, sources) + print(f"Qwen Code hooks installed at {sources.user.file_path}") + print("Qwen Code verification: run /hooks and restart active sessions.") def uninstall(self, agent_id: str = "") -> None: - document = read_document() - if not document.exists: - return - rendered = render_document(document, agent_id or None, {}) - change = rendered_change(rendered) - write_changes( - [change] if change is not None else [], - "Write Qwen Code settings", + sources = read_qwen_sources() + rendered = render_qwen_document( + sources.user, agent_id or None, {} ) + commit_qwen_uninstall(prepare_qwen_uninstall(rendered), sources) def status(self) -> PluginStatus: - document = read_document() - contracts = runtime_contracts(document.hooks) - if document.hooks_disabled: - return PluginStatus( - installed=False, - agent=AGENT_KEY, - details=f"Configured hooks are disabled: {document.file_path}", + sources = read_qwen_sources() + contracts = qwen_runtime_contracts(sources.user.hooks) + configured = not sources.disable_control.disabled and bool(contracts) + installed = configured and qwen_runtime_files_exist(contracts) + if installed: + details = f"Config: {sources.user.file_path}" + elif contracts and sources.disable_control.disabled: + details = f"Configured hooks are disabled: {sources.user.file_path}" + elif contracts: + details = ( + f"Configured at {sources.user.file_path}; " + "managed contract incomplete" ) - if not contracts: - return PluginStatus( - installed=False, - agent=AGENT_KEY, - details="Not installed", - ) - installed = runtime_files_exist(contracts) - details = ( - f"Config: {document.file_path}" - if installed - else f"Configured at {document.file_path}; runtime scripts missing" - ) + else: + details = "Not installed" return PluginStatus( installed=installed, agent=AGENT_KEY, diff --git a/sdks/python/elydora/plugins/qwen_command.py b/sdks/python/elydora/plugins/qwen_command.py new file mode 100644 index 0000000..3462fc9 --- /dev/null +++ b/sdks/python/elydora/plugins/qwen_command.py @@ -0,0 +1,134 @@ +"""Qwen Code command construction and exact runtime ownership.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import sys +from typing import Optional, Tuple + + +@dataclass(frozen=True) +class QwenRuntimeReference: + agent_id: str + script_path: str + + +def same_qwen_path(left: str, right: str) -> bool: + return os.path.normcase(os.path.abspath(left)) == os.path.normcase( + os.path.abspath(right) + ) + + +def same_qwen_agent_id(left: str, right: str) -> bool: + return os.path.normcase(left) == os.path.normcase(right) + + +def _quote_posix(value: str) -> str: + return "'" + value.replace("'", "'\"'\"'") + "'" + + +def _quote_powershell(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def build_qwen_command(script_path: str) -> str: + if not os.path.isabs(sys.executable) or not os.path.isabs(script_path): + raise ValueError( + "Qwen Code hook commands require absolute executable and script paths" + ) + invocation = ( + f"{_quote_powershell(sys.executable)} {_quote_powershell(script_path)}" + if os.name == "nt" + else f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" + ) + return f"& {invocation}; exit $LASTEXITCODE" if os.name == "nt" else invocation + + +def _read_posix_argument(command: str, start: int) -> Optional[Tuple[str, int]]: + if start >= len(command) or command[start] != "'": + return None + apostrophe = "'\"'\"'" + value = "" + index = start + 1 + while index < len(command): + if command.startswith(apostrophe, index): + value += "'" + index += len(apostrophe) + continue + if command[index] == "'": + return value, index + 1 + value += command[index] + index += 1 + return None + + +def _parse_posix_command(command: str) -> Optional[Tuple[str, str]]: + executable = _read_posix_argument(command, 0) + if executable is None or executable[1] >= len(command): + return None + if command[executable[1]] != " ": + return None + script = _read_posix_argument(command, executable[1] + 1) + if script is None or script[1] != len(command): + return None + return executable[0], script[0] + + +def _read_powershell_argument( + command: str, start: int +) -> Optional[Tuple[str, int]]: + if start >= len(command) or command[start] != "'": + return None + value = "" + index = start + 1 + while index < len(command): + if command[index] != "'": + value += command[index] + index += 1 + continue + if index + 1 < len(command) and command[index + 1] == "'": + value += "'" + index += 2 + continue + return value, index + 1 + return None + + +def _parse_powershell_command(command: str) -> Optional[Tuple[str, str]]: + if not command.startswith("& "): + return None + executable = _read_powershell_argument(command, 2) + if executable is None or executable[1] >= len(command): + return None + if command[executable[1]] != " ": + return None + script = _read_powershell_argument(command, executable[1] + 1) + if script is None or command[script[1] :] != "; exit $LASTEXITCODE": + return None + return executable[0], script[0] + + +def qwen_runtime_reference( + command: str, script_name: str +) -> Optional[QwenRuntimeReference]: + parsed = ( + _parse_powershell_command(command) + if os.name == "nt" + else _parse_posix_command(command) + ) + if ( + parsed is None + or not same_qwen_path(parsed[0], sys.executable) + or not os.path.isabs(parsed[1]) + or os.path.basename(parsed[1]) != script_name + ): + return None + agent_directory = os.path.dirname(parsed[1]) + runtime_root = os.path.join(os.path.expanduser("~"), ".elydora") + if not same_qwen_path(os.path.dirname(agent_directory), runtime_root): + return None + agent_id = os.path.basename(agent_directory) + if agent_id in {"", ".", ".."}: + return None + return QwenRuntimeReference(agent_id, parsed[1]) diff --git a/sdks/python/elydora/plugins/qwen_config.py b/sdks/python/elydora/plugins/qwen_config.py index 6529eb0..7d5602a 100644 --- a/sdks/python/elydora/plugins/qwen_config.py +++ b/sdks/python/elydora/plugins/qwen_config.py @@ -1,4 +1,4 @@ -"""Qwen Code settings parsing and source-preserving rendering.""" +"""Qwen Code JSONC parsing and source-preserving hook edits.""" from __future__ import annotations @@ -6,12 +6,13 @@ from dataclasses import dataclass from typing import Dict, Optional from ._jsonc import JsoncEditor +from ._managed_files import FileSnapshot from .qwen_contract import ( - TOOL_EVENTS, + MANAGED_EVENTS, JsonObject, - QwenHookSettings, - managed_removals, - read_hook_settings, + QwenHooks, + managed_qwen_removals, + read_qwen_hooks, ) @@ -20,83 +21,121 @@ OWNED_FILE_MARKER = "// Managed by Elydora" @dataclass(frozen=True) class QwenDocument: + kind: str file_path: str exists: bool raw: str + snapshot: Optional[FileSnapshot] root: JsonObject - hooks: QwenHookSettings + hooks: QwenHooks has_hooks_container: bool - hooks_disabled: bool + disable_all_hooks: Optional[bool] + folder_trust_enabled: Optional[bool] owned_file: bool @dataclass(frozen=True) -class RenderedDocument: +class RenderedQwenDocument: document: QwenDocument changed: bool next_source: Optional[str] -def _label(file_path: str) -> str: - return f"Qwen Code settings at {file_path}" +def qwen_source_label(kind: str) -> str: + labels = { + "system-defaults": "Qwen Code system defaults", + "user": "Qwen Code user settings", + "workspace": "Qwen Code workspace settings", + "system": "Qwen Code system override settings", + } + try: + return labels[kind] + except KeyError as error: + raise ValueError(f"Unsupported Qwen Code document kind: {kind}") from error -def parse_document( +def qwen_document_label(document: QwenDocument) -> str: + return qwen_source_label(document.kind) + + +def parse_qwen_jsonc_object(raw: str, label: str) -> JsonObject: + editor = JsoncEditor(raw, label, allow_trailing_commas=False) + if not isinstance(editor.value, dict): + raise ValueError(f"{label} must contain a JSON object") + return editor.value + + +def _folder_trust_enabled( + root: JsonObject, label: str +) -> Optional[bool]: + if "security" not in root: + return None + security = root["security"] + if not isinstance(security, dict): + raise ValueError(f'{label} field "security" must be an object') + if "folderTrust" not in security: + return None + folder_trust = security["folderTrust"] + if not isinstance(folder_trust, dict): + raise ValueError( + f'{label} field "security.folderTrust" must be an object' + ) + enabled = folder_trust.get("enabled") + if enabled is not None and not isinstance(enabled, bool): + raise ValueError( + f'{label} field "security.folderTrust.enabled" must be a boolean' + ) + return enabled + + +def parse_qwen_document( *, + kind: str, exists: bool, file_path: str, raw: str, + snapshot: Optional[FileSnapshot] = None, ) -> QwenDocument: - label = _label(file_path) - editor = JsoncEditor( - raw, - label, - allow_trailing_commas=False, - ) - if not isinstance(editor.value, dict): - raise ValueError(f"{label} must contain a JSON object") - root = editor.value + label = f"{qwen_source_label(kind)} at {file_path}" + root = parse_qwen_jsonc_object(raw, label) disabled = root.get("disableAllHooks") if "disableAllHooks" in root and not isinstance(disabled, bool): raise ValueError(f'{label} field "disableAllHooks" must be a boolean') - has_container = "hooks" in root - hooks = ( - read_hook_settings(root["hooks"], f'{label} field "hooks"') - if has_container - else {} - ) + has_hooks = "hooks" in root return QwenDocument( - file_path, - exists, - raw, - root, - hooks, - has_container, - disabled is True, - raw.startswith(OWNED_FILE_MARKER), + kind=kind, + file_path=file_path, + exists=exists, + raw=raw, + snapshot=snapshot, + root=root, + hooks=read_qwen_hooks(root["hooks"]) if has_hooks else {}, + has_hooks_container=has_hooks, + disable_all_hooks=disabled if isinstance(disabled, bool) else None, + folder_trust_enabled=_folder_trust_enabled(root, label), + owned_file=kind == "user" and raw.startswith(OWNED_FILE_MARKER), ) -def create_owned_document(file_path: str) -> QwenDocument: - return parse_document( +def create_qwen_document(kind: str, file_path: str) -> QwenDocument: + raw = f"{OWNED_FILE_MARKER}\n{{}}\n" if kind == "user" else "{}\n" + return parse_qwen_document( + kind=kind, exists=False, file_path=file_path, - raw=f"{OWNED_FILE_MARKER}\n{{}}\n", + raw=raw, ) -def _settings_from_editor( - editor: JsoncEditor, - document: QwenDocument, -) -> QwenHookSettings: - root = editor.value - if not isinstance(root, dict): - raise ValueError(f"{_label(document.file_path)} must contain a JSON object") - if "hooks" not in root: - return {} - return read_hook_settings( - root["hooks"], - f'{_label(document.file_path)} field "hooks"', +def _current_document( + document: QwenDocument, editor: JsoncEditor +) -> QwenDocument: + return parse_qwen_document( + kind=document.kind, + exists=document.exists, + file_path=document.file_path, + raw=editor.raw, + snapshot=document.snapshot, ) @@ -105,8 +144,8 @@ def _remove_managed( document: QwenDocument, agent_id: Optional[str], ) -> None: - removals = managed_removals(document.hooks, agent_id) - for event in TOOL_EVENTS: + removals = managed_qwen_removals(document.hooks, agent_id) + for event in MANAGED_EVENTS: event_removals = sorted( (item for item in removals if item.event == event), key=lambda item: item.group_index, @@ -120,14 +159,12 @@ def _remove_managed( for handler_index in sorted(removal.handler_indexes, reverse=True): editor.delete((*group_path, "hooks", handler_index)) if event_removals: - current = _settings_from_editor(editor, document) - if event in current and not current[event]: + current = _current_document(document, editor) + if event in current.hooks and not current.hooks[event]: editor.delete(("hooks", event)) - root = editor.value - if isinstance(root, dict) and "hooks" in root: - current = _settings_from_editor(editor, document) - if not current: - editor.delete(("hooks",)) + current = _current_document(document, editor) + if current.has_hooks_container and not current.hooks: + editor.delete(("hooks",)) def _append_group( @@ -136,42 +173,57 @@ def _append_group( event: str, group: JsonObject, ) -> None: - root = editor.value - if not isinstance(root, dict): - raise ValueError(f"{_label(document.file_path)} must contain a JSON object") - if "hooks" not in root: + current = _current_document(document, editor) + if not current.has_hooks_container: editor.add_property((), "hooks", {event: [group]}) - return - current = _settings_from_editor(editor, document) - if event in current: + elif event in current.hooks: editor.append(("hooks", event), group) else: editor.add_property(("hooks",), event, [group]) -def render_document( +def _already_installed( + document: QwenDocument, additions: Dict[str, JsonObject] +) -> bool: + if set(additions) != set(MANAGED_EVENTS): + return False + removals = managed_qwen_removals(document.hooks) + for event in MANAGED_EVENTS: + expected = additions[event] + event_removals = [item for item in removals if item.event == event] + exact_groups = [ + group for group in document.hooks.get(event, []) if group == expected + ] + if ( + len(event_removals) != 1 + or not event_removals[0].remove_group + or len(exact_groups) != 1 + ): + return False + return True + + +def render_qwen_document( document: QwenDocument, agent_id: Optional[str], additions: Dict[str, JsonObject], -) -> RenderedDocument: +) -> RenderedQwenDocument: + if agent_id is None and _already_installed(document, additions): + return RenderedQwenDocument(document, False, document.raw) editor = JsoncEditor( document.raw, - _label(document.file_path), + f"{qwen_document_label(document)} at {document.file_path}", allow_trailing_commas=False, ) _remove_managed(editor, document, agent_id) - for event in TOOL_EVENTS: + for event in MANAGED_EVENTS: group = additions.get(event) if group is not None: _append_group(editor, document, event, group) - next_document = parse_document( - exists=document.exists, - file_path=document.file_path, - raw=editor.raw, - ) - if not additions and document.owned_file and not next_document.root: - return RenderedDocument(document, True, None) - return RenderedDocument( + current = _current_document(document, editor) + if not additions and document.owned_file and not current.root: + return RenderedQwenDocument(document, True, None) + return RenderedQwenDocument( document, editor.raw != document.raw, editor.raw, diff --git a/sdks/python/elydora/plugins/qwen_contract.py b/sdks/python/elydora/plugins/qwen_contract.py index ddfe4e6..34e7857 100644 --- a/sdks/python/elydora/plugins/qwen_contract.py +++ b/sdks/python/elydora/plugins/qwen_contract.py @@ -1,4 +1,4 @@ -"""Qwen Code hook schema, commands, and exact ownership rules.""" +"""Qwen Code 0.20 hook schema and exact ownership rules.""" from __future__ import annotations @@ -8,17 +8,26 @@ import math import os import shutil import subprocess # nosec B404 -import sys -from typing import Any, Dict, List, Optional, Sequence, Set, Tuple +from typing import Any, Dict, List, Optional, Sequence, Tuple + +from .qwen_command import ( + QwenRuntimeReference, + build_qwen_command, + qwen_runtime_reference, + same_qwen_agent_id, + same_qwen_path, +) AGENT_KEY = "qwen" GUARD_SCRIPT = "guard.py" AUDIT_SCRIPT = "hook.py" +GUARD_HOOK_NAME = "elydora-guard" +AUDIT_HOOK_NAME = "elydora-audit" HOOK_TIMEOUT_MILLISECONDS = 10_000 -TOOL_EVENTS = ("PreToolUse", "PostToolUse") +MANAGED_EVENTS = ("PreToolUse", "PostToolUse", "PostToolUseFailure") -_EVENT_NAMES = { +KNOWN_EVENTS = { "PreToolUse", "PostToolUse", "PostToolUseFailure", @@ -41,9 +50,25 @@ _EVENT_NAMES = { "TodoCompleted", "InstructionsLoaded", } -_CONFIG_FIELDS = {"enabled", "disabled", "notifications"} -_HANDLER_KEYS = {"command", "shell", "timeout", "type"} -_GROUP_KEYS = {"hooks", "matcher"} + +REGEX_MATCHER_EVENTS = { + "PreToolUse", + "PostToolUse", + "PostToolUseFailure", + "PermissionRequest", + "PermissionDenied", + "SubagentStart", + "SubagentStop", + "PreCompact", + "PostCompact", + "SessionStart", + "SessionEnd", + "StopFailure", + "Notification", + "InstructionsLoaded", + "UserPromptExpansion", +} + _REGEX_TIMEOUT_SECONDS = 10 _REGEX_VALIDATOR = """import fs from 'node:fs'; const entries = JSON.parse(fs.readFileSync(0, 'utf8')); @@ -59,11 +84,11 @@ for (const entry of entries) { """ JsonObject = Dict[str, Any] -QwenHookSettings = Dict[str, Any] +QwenHooks = Dict[str, Any] @dataclass(frozen=True) -class RuntimeContract: +class QwenRuntimeContract: agent_id: str guard_path: str audit_path: str @@ -81,67 +106,37 @@ def elydora_dir() -> str: return os.path.join(os.path.expanduser("~"), ".elydora") -def _quote_posix(value: str) -> str: - return "'" + value.replace("'", "'\"'\"'") + "'" +def _optional_string(value: JsonObject, field: str, label: str) -> None: + if field in value and not isinstance(value[field], str): + raise ValueError(f'{label} field "{field}" must be a string') -def _quote_powershell(value: str) -> str: - return "'" + value.replace("'", "''") + "'" +def _optional_boolean(value: JsonObject, field: str, label: str) -> None: + if field in value and not isinstance(value[field], bool): + raise ValueError(f'{label} field "{field}" must be a boolean') -def build_command(script_path: str) -> str: - if os.name == "nt": - invocation = ( - f"{_quote_powershell(sys.executable)} {_quote_powershell(script_path)}" - ) - return f"& {invocation}; exit $LASTEXITCODE" - return f"{_quote_posix(sys.executable)} {_quote_posix(script_path)}" - - -def build_group(script_path: str) -> JsonObject: - return { - "matcher": "*", - "hooks": [ - { - "type": "command", - "command": build_command(script_path), - "shell": "powershell" if os.name == "nt" else "bash", - "timeout": HOOK_TIMEOUT_MILLISECONDS, - } - ], - } - - -def _optional_string(value: JsonObject, key: str, label: str) -> None: - if key in value and not isinstance(value[key], str): - raise ValueError(f'{label} field "{key}" must be a string') - - -def _optional_boolean(value: JsonObject, key: str, label: str) -> None: - if key in value and not isinstance(value[key], bool): - raise ValueError(f'{label} field "{key}" must be a boolean') - - -def _optional_string_map(value: JsonObject, key: str, label: str) -> None: - if key not in value: +def _optional_string_map(value: JsonObject, field: str, label: str) -> None: + if field not in value: return - item = value[key] + item = value[field] if not isinstance(item, dict) or any( not isinstance(entry, str) for entry in item.values() ): - raise ValueError(f'{label} field "{key}" must contain string values') + raise ValueError(f'{label} field "{field}" must map names to strings') def _validate_timeout(value: JsonObject, label: str) -> None: if "timeout" not in value: return timeout = value["timeout"] - if ( - isinstance(timeout, bool) - or not isinstance(timeout, (int, float)) - or not math.isfinite(timeout) - or timeout < 0 - ): + try: + finite = math.isfinite(timeout) + except (TypeError, OverflowError): + finite = False + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or not finite: + raise ValueError(f"{label} timeout must be a non-negative finite number") + if timeout < 0: raise ValueError(f"{label} timeout must be a non-negative finite number") @@ -151,20 +146,20 @@ def _validate_handler(value: Any, label: str) -> JsonObject: handler = dict(value) kind = handler.get("type") if kind not in {"command", "http", "prompt"}: - raise ValueError(f'{label} type must be "command", "http", or "prompt"') + raise ValueError(f'{label} has unsupported type "{kind}"') _validate_timeout(handler, label) - for key in ("name", "description", "statusMessage", "source"): - _optional_string(handler, key, label) + for field in ("name", "description", "statusMessage", "source"): + _optional_string(handler, field, label) if kind == "command": if not isinstance(handler.get("command"), str) or not handler["command"]: - raise ValueError(f"{label} command must be a non-empty string") + raise ValueError(f"{label} requires a non-empty command") _optional_string_map(handler, "env", label) _optional_boolean(handler, "async", label) if "shell" in handler and handler["shell"] not in {"bash", "powershell"}: raise ValueError(f'{label} shell must be "bash" or "powershell"') elif kind == "http": if not isinstance(handler.get("url"), str) or not handler["url"]: - raise ValueError(f"{label} url must be a non-empty string") + raise ValueError(f"{label} requires a non-empty url") _optional_string_map(handler, "headers", label) _optional_boolean(handler, "once", label) _optional_string(handler, "if", label) @@ -176,12 +171,13 @@ def _validate_handler(value: Any, label: str) -> JsonObject: raise ValueError(f"{label} allowedEnvVars must be an array of strings") else: if not isinstance(handler.get("prompt"), str) or not handler["prompt"]: - raise ValueError(f"{label} prompt must be a non-empty string") + raise ValueError(f"{label} requires a non-empty prompt") _optional_string(handler, "model", label) return handler -def _validate_group(value: Any, label: str) -> JsonObject: +def _validate_group(value: Any, event: str, index: int) -> JsonObject: + label = f"Qwen Code settings group hooks.{event}[{index}]" if not isinstance(value, dict): raise ValueError(f"{label} must be an object") group = dict(value) @@ -193,49 +189,47 @@ def _validate_group(value: Any, label: str) -> JsonObject: if not isinstance(handlers, list): raise ValueError(f"{label} must contain a hooks array") group["hooks"] = [ - _validate_handler(handler, f"{label}.hooks[{index}]") - for index, handler in enumerate(handlers) + _validate_handler(handler, f"{label}.hooks[{handler_index}]") + for handler_index, handler in enumerate(handlers) ] return group -def read_hook_settings(value: Any, label: str) -> QwenHookSettings: +def read_qwen_hooks(value: Any) -> QwenHooks: if not isinstance(value, dict): - raise ValueError(f"{label} must contain a JSON object") - settings = dict(value) + raise ValueError('Qwen Code settings field "hooks" must be an object') + hooks: QwenHooks = {} for event, groups in value.items(): - if event in _CONFIG_FIELDS: + if event not in KNOWN_EVENTS: + hooks[event] = groups continue - if event not in _EVENT_NAMES: - raise ValueError(f'{label} contains unsupported field "{event}"') if not isinstance(groups, list): - raise ValueError(f'{label} field "{event}" must be an array') - settings[event] = [ - _validate_group(group, f'{label} field "{event}"[{index}]') + raise ValueError( + f'Qwen Code settings field "hooks.{event}" must be an array' + ) + hooks[event] = [ + _validate_group(group, event, index) for index, group in enumerate(groups) ] - return settings + return hooks -def _regex_entries(settings: QwenHookSettings) -> List[JsonObject]: - entries = [] - for event, groups in settings.items(): - if event in _CONFIG_FIELDS: - continue - for index, group in enumerate(groups): - matcher = group.get("matcher") - if isinstance(matcher, str) and matcher.strip() not in {"", "*"}: - entries.append( - { - "label": f'Qwen Code hooks field "{event}"[{index}] matcher', +def _matcher_entries(sources: Sequence[QwenHooks]) -> List[JsonObject]: + entries: List[JsonObject] = [] + for hooks in sources: + for event in REGEX_MATCHER_EVENTS: + for index, group in enumerate(hooks.get(event, [])): + matcher = group.get("matcher") + if isinstance(matcher, str) and matcher.strip() not in {"", "*"}: + entries.append({ + "label": f"Qwen Code settings group hooks.{event}[{index}]", "pattern": matcher, - } - ) + }) return entries -def validate_javascript_regexes(settings: Sequence[QwenHookSettings]) -> None: - entries = [entry for source in settings for entry in _regex_entries(source)] +def validate_javascript_matchers(sources: Sequence[QwenHooks]) -> None: + entries = _matcher_entries(sources) if not entries: return node_path = shutil.which("node") @@ -267,143 +261,148 @@ def validate_javascript_regexes(settings: Sequence[QwenHookSettings]) -> None: or f"Node.js exited with code {result.returncode}" ) raise ValueError( - f"Qwen Code matcher must be a valid JavaScript regular expression: {message}" + "Qwen Code matcher must be a valid JavaScript regular expression: " + + message ) -def _read_quoted_argument(command: str, start: int) -> Optional[Tuple[str, int]]: - if start >= len(command) or command[start] != "'": - return None - value = "" - index = start + 1 - while index < len(command): - if os.name == "nt" and command[index : index + 2] == "''": - value += "'" - index += 2 - continue - if os.name != "nt" and command[index : index + 5] == "'\"'\"'": - value += "'" - index += 5 - continue - if command[index] == "'": - return value, index + 1 - value += command[index] - index += 1 - return None - - -def _parse_generated_command(command: str) -> Optional[Tuple[str, str]]: - if os.name == "nt" and not command.startswith("& "): - return None - start = 2 if os.name == "nt" else 0 - executable = _read_quoted_argument(command, start) - if executable is None or executable[1] >= len(command): - return None - if command[executable[1]] != " ": - return None - script = _read_quoted_argument(command, executable[1] + 1) - if script is None: - return None - suffix = "; exit $LASTEXITCODE" if os.name == "nt" else "" - if command[script[1] :] != suffix or not executable[0] or not script[0]: - return None - return executable[0], script[0] - - -def _same_path(left: str, right: str) -> bool: - return os.path.normcase(os.path.abspath(left)) == os.path.normcase( - os.path.abspath(right) - ) - - -def same_agent_id(left: str, right: str) -> bool: - return os.path.normcase(left) == os.path.normcase(right) - - -def managed_agent_id(handler: JsonObject, script_name: str) -> Optional[str]: - if set(handler) != _HANDLER_KEYS: - return None - expected_shell = "powershell" if os.name == "nt" else "bash" - if ( - handler.get("type") != "command" - or handler.get("timeout") != HOOK_TIMEOUT_MILLISECONDS - or handler.get("shell") != expected_shell - or not isinstance(handler.get("command"), str) - ): - return None - parsed = _parse_generated_command(handler["command"]) - if parsed is None or not _same_path(parsed[0], sys.executable): - return None - script_path = parsed[1] - if os.path.basename(script_path) != script_name: - return None - agent_directory = os.path.dirname(script_path) - if not _same_path(os.path.dirname(agent_directory), elydora_dir()): - return None - agent_id = os.path.basename(agent_directory) - return agent_id if agent_id not in {"", ".", ".."} else None - - -def _exact_owned_group(group: JsonObject, indexes: Tuple[int, ...]) -> bool: - return ( - set(group) == _GROUP_KEYS - and group.get("matcher") == "*" - and bool(indexes) - and len(indexes) == len(group["hooks"]) - ) - - -def managed_removals( - settings: QwenHookSettings, - agent_id: Optional[str] = None, -) -> List[ManagedRemoval]: - removals = [] - for event, script_name in ( - ("PreToolUse", GUARD_SCRIPT), - ("PostToolUse", AUDIT_SCRIPT), - ): - for group_index, group in enumerate(settings.get(event, [])): - indexes = tuple( - index - for index, handler in enumerate(group["hooks"]) - if ( - (managed_id := managed_agent_id(handler, script_name)) is not None - and (agent_id is None or same_agent_id(managed_id, agent_id)) - ) - ) - if indexes: - removals.append( - ManagedRemoval( - event, - group_index, - indexes, - _exact_owned_group(group, indexes), - ) - ) - return removals - - -def _managed_ids(groups: List[JsonObject], script_name: str) -> Set[str]: +def build_qwen_group(script_path: str, name: str) -> JsonObject: return { - agent_id - for group in groups - for handler in group["hooks"] - if (agent_id := managed_agent_id(handler, script_name)) is not None + "hooks": [{ + "type": "command", + "name": name, + "command": build_qwen_command(script_path), + "shell": "powershell" if os.name == "nt" else "bash", + "timeout": HOOK_TIMEOUT_MILLISECONDS, + }] } -def runtime_contracts(settings: QwenHookSettings) -> List[RuntimeContract]: - guards = _managed_ids(settings.get("PreToolUse", []), GUARD_SCRIPT) - audits = _managed_ids(settings.get("PostToolUse", []), AUDIT_SCRIPT) - contracts = [] - for agent_id in sorted(guards, key=os.path.normcase): - if any(same_agent_id(agent_id, audit_id) for audit_id in audits): - root = os.path.join(elydora_dir(), agent_id) - contracts.append( - RuntimeContract( - agent_id, - os.path.join(root, GUARD_SCRIPT), - os.path.join(root, AUDIT_SCRIPT), +def _exact_current_group(group: JsonObject) -> bool: + return set(group) == {"hooks"} + + +def _exact_legacy_group(group: JsonObject) -> bool: + return set(group) == {"hooks", "matcher"} and group.get("matcher") == "*" + + +def _current_reference( + handler: JsonObject, script_name: str, hook_name: str +) -> Optional[QwenRuntimeReference]: + if ( + set(handler) == {"command", "name", "shell", "timeout", "type"} + and handler.get("type") == "command" + and handler.get("name") == hook_name + and handler.get("shell") == ("powershell" if os.name == "nt" else "bash") + and handler.get("timeout") == HOOK_TIMEOUT_MILLISECONDS + and isinstance(handler.get("command"), str) + ): + return qwen_runtime_reference(handler["command"], script_name) + return None + + +def _legacy_reference( + handler: JsonObject, script_name: str +) -> Optional[QwenRuntimeReference]: + if ( + set(handler) == {"command", "shell", "timeout", "type"} + and handler.get("type") == "command" + and handler.get("shell") == ("powershell" if os.name == "nt" else "bash") + and handler.get("timeout") == HOOK_TIMEOUT_MILLISECONDS + and isinstance(handler.get("command"), str) + ): + return qwen_runtime_reference(handler["command"], script_name) + return None + + +def _managed_reference( + handler: JsonObject, + script_name: str, + hook_name: str, + include_legacy: bool, +) -> Optional[QwenRuntimeReference]: + current = _current_reference(handler, script_name, hook_name) + if current is not None or not include_legacy: + return current + return _legacy_reference(handler, script_name) + + +_EVENT_CONTRACTS = ( + ("PreToolUse", GUARD_SCRIPT, GUARD_HOOK_NAME), + ("PostToolUse", AUDIT_SCRIPT, AUDIT_HOOK_NAME), + ("PostToolUseFailure", AUDIT_SCRIPT, AUDIT_HOOK_NAME), +) + + +def managed_qwen_removals( + hooks: QwenHooks, agent_id: Optional[str] = None +) -> List[ManagedRemoval]: + removals: List[ManagedRemoval] = [] + for event, script_name, hook_name in _EVENT_CONTRACTS: + for group_index, group in enumerate(hooks.get(event, [])): + handlers = group["hooks"] + indexes = tuple( + index + for index, handler in enumerate(handlers) + if ( + reference := _managed_reference( + handler, script_name, hook_name, True + ) + ) + is not None + and ( + agent_id is None + or same_qwen_agent_id(reference.agent_id, agent_id) ) ) + if indexes: + removals.append(ManagedRemoval( + event, + group_index, + indexes, + (_exact_current_group(group) or _exact_legacy_group(group)) + and len(indexes) == len(handlers), + )) + return removals + + +def _references_for_event( + groups: List[JsonObject], script_name: str, hook_name: str +) -> Dict[str, List[QwenRuntimeReference]]: + references: Dict[str, List[QwenRuntimeReference]] = {} + for group in groups: + if not _exact_current_group(group): + continue + for handler in group["hooks"]: + reference = _current_reference(handler, script_name, hook_name) + if reference is None: + continue + key = os.path.normcase(reference.agent_id) + references.setdefault(key, []).append(reference) + return references + + +def qwen_runtime_contracts(hooks: QwenHooks) -> List[QwenRuntimeContract]: + guards = _references_for_event( + hooks.get("PreToolUse", []), GUARD_SCRIPT, GUARD_HOOK_NAME + ) + posts = _references_for_event( + hooks.get("PostToolUse", []), AUDIT_SCRIPT, AUDIT_HOOK_NAME + ) + failures = _references_for_event( + hooks.get("PostToolUseFailure", []), AUDIT_SCRIPT, AUDIT_HOOK_NAME + ) + contracts: List[QwenRuntimeContract] = [] + for key, guard in guards.items(): + post = posts.get(key, []) + failure = failures.get(key, []) + if len(guard) != 1 or len(post) != 1 or len(failure) != 1: + continue + if not same_qwen_path(post[0].script_path, failure[0].script_path): + continue + contracts.append(QwenRuntimeContract( + guard[0].agent_id, + guard[0].script_path, + post[0].script_path, + )) return contracts diff --git a/sdks/python/elydora/plugins/qwen_installation.py b/sdks/python/elydora/plugins/qwen_installation.py new file mode 100644 index 0000000..759a436 --- /dev/null +++ b/sdks/python/elydora/plugins/qwen_installation.py @@ -0,0 +1,198 @@ +"""Transactional Qwen Code hook and runtime installation.""" + +from __future__ import annotations + +from dataclasses import dataclass +import json +import os +from typing import List, Optional, Sequence + +from elydora._runtime_paths import resolve_agent_directory + +from ._managed_files import MAX_CONFIG_BYTES, MAX_SECRET_BYTES, MAX_SOURCE_BYTES +from ._transaction import FileChange, FilePrecondition, file_change, source_change, write_changes +from .base import InstallConfig +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script +from .qwen_command import same_qwen_path +from .qwen_config import RenderedQwenDocument, qwen_document_label +from .qwen_contract import AGENT_KEY, AUDIT_SCRIPT, GUARD_SCRIPT, elydora_dir +from .qwen_io import validate_api_origin, validate_private_key, validate_runtime_tree +from .qwen_sources import QwenSources, require_qwen_hooks_enabled + + +@dataclass(frozen=True) +class QwenRuntimePaths: + agent_id: str + agent_directory: str + guard_path: str + audit_path: str + + +def _runtime_config(config: InstallConfig, agent_id: str) -> dict: + value = { + "org_id": config.get("org_id", ""), + "agent_id": agent_id, + "kid": config.get("kid", ""), + "base_url": config.get("base_url", "https://api.elydora.com"), + "agent_name": AGENT_KEY, + } + token = config.get("token") + if token: + value["token"] = token + return value + + +def _json_source(value: dict) -> str: + return json.dumps(value, indent=2) + "\n" + + +def _agent_paths(config: InstallConfig) -> QwenRuntimePaths: + agent_id = config.get("agent_id", "") + if not agent_id: + raise ValueError("agent_id is required") + agent_directory = resolve_agent_directory(elydora_dir(), agent_id) + guard_path = os.path.join(agent_directory, GUARD_SCRIPT) + audit_path = os.path.join(agent_directory, AUDIT_SCRIPT) + if not same_qwen_path(config.get("guard_script_path", ""), guard_path): + raise ValueError( + "Elydora guard runtime must use the managed agent directory: " + f"{guard_path}" + ) + return QwenRuntimePaths(agent_id, agent_directory, guard_path, audit_path) + + +def _validate_install_config(config: InstallConfig) -> None: + for field in ("org_id", "agent_id", "kid", "private_key", "base_url"): + value = config.get(field) + if not isinstance(value, str) or not value: + raise ValueError(f"{field} is required") + if config.get("agent_name") != AGENT_KEY: + raise ValueError( + f"Qwen Code installation requires agent_name {AGENT_KEY}" + ) + validate_private_key(config["private_key"]) + validate_api_origin(config["base_url"]) + if "token" in config: + token = config["token"] + if not isinstance(token, str) or not token: + raise ValueError("token must be a non-empty string when provided") + _agent_paths(config) + + +def preflight_qwen_installation( + config: InstallConfig, sources: QwenSources +) -> QwenRuntimePaths: + require_qwen_hooks_enabled(sources) + _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: RenderedQwenDocument, +) -> Optional[FileChange]: + if not rendered.changed: + return None + return source_change( + rendered.document.file_path, + qwen_document_label(rendered.document), + rendered.document.raw if rendered.document.exists else None, + rendered.next_source, + 0o600, + MAX_SOURCE_BYTES, + expected_snapshot=rendered.document.snapshot, + ) + + +def _present(changes: Sequence[Optional[FileChange]]) -> List[FileChange]: + return [change for change in changes if change is not None] + + +def prepare_qwen_installation( + config: InstallConfig, + paths: QwenRuntimePaths, + rendered: RenderedQwenDocument, +) -> List[FileChange]: + 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, + ) + return _present([ + _runtime_change( + paths.guard_path, "Elydora guard runtime", guard_script, 0o700 + ), + _runtime_change( + os.path.join(paths.agent_directory, "config.json"), + "Elydora runtime config", + _json_source(_runtime_config(config, paths.agent_id)), + 0o600, + MAX_CONFIG_BYTES, + ), + _runtime_change( + os.path.join(paths.agent_directory, "private.key"), + "Elydora private key", + config.get("private_key", ""), + 0o600, + MAX_SECRET_BYTES, + ), + _runtime_change( + paths.audit_path, "Elydora audit runtime", audit_script, 0o700 + ), + _rendered_change(rendered), + ]) + + +def _read_only_preconditions( + sources: QwenSources, changed_path: Optional[str] +) -> List[FilePrecondition]: + return [ + condition + for condition in sources.preconditions + if changed_path is None + or not same_qwen_path(condition.file_path, changed_path) + ] + + +def commit_qwen_installation( + changes: List[FileChange], sources: QwenSources +) -> None: + write_changes( + changes, + "Install Qwen Code hooks", + _read_only_preconditions(sources, sources.user.file_path), + ) + + +def prepare_qwen_uninstall( + rendered: RenderedQwenDocument, +) -> Optional[FileChange]: + return _rendered_change(rendered) + + +def commit_qwen_uninstall( + change: Optional[FileChange], sources: QwenSources +) -> None: + if change is None: + return + write_changes( + [change], + "Uninstall Qwen Code hooks", + _read_only_preconditions(sources, sources.user.file_path), + ) diff --git a/sdks/python/elydora/plugins/qwen_io.py b/sdks/python/elydora/plugins/qwen_io.py index 32d10f7..b373f10 100644 --- a/sdks/python/elydora/plugins/qwen_io.py +++ b/sdks/python/elydora/plugins/qwen_io.py @@ -1,139 +1,200 @@ -"""Qwen Code home discovery, settings I/O, and runtime inspection.""" +"""Strict Qwen Code managed runtime inspection.""" from __future__ import annotations -import json +import base64 import os -import re -import tempfile -from typing import Optional +from typing import Any, List, Optional +import urllib.parse -from ._dotenv import parse_dotenv -from ._transaction import ( - FileChange, - read_optional, - regular_file_exists, - require_runtime, - source_change, +from ._managed_files import ( + MAX_CONFIG_BYTES, + MAX_SECRET_BYTES, + physical_directory_exists, + physical_file_exists, + read_physical_file, ) -from .qwen_config import ( - QwenDocument, - RenderedDocument, - create_owned_document, - parse_document, -) -from .qwen_contract import AGENT_KEY, JsonObject, RuntimeContract, same_agent_id +from ._strict_json import JsonObject, parse_json_object +from .guard_template import generate_guard_script +from .hook_template import generate_hook_script +from .qwen_command import same_qwen_agent_id, same_qwen_path +from .qwen_contract import AGENT_KEY, QwenRuntimeContract, elydora_dir -def _default_qwen_home() -> str: - home_dir = os.path.expanduser("~") - if not home_dir or home_dir == "~": - home_dir = tempfile.gettempdir() - return os.path.join(home_dir, ".qwen") - - -def _resolve_storage_path(value: str) -> str: - resolved = value - if value == "~": - resolved = os.path.expanduser("~") - elif value.startswith(("~/", "~\\")): - segments = [item for item in re.split(r"[/\\]+", value[2:]) if item] - resolved = os.path.join(os.path.expanduser("~"), *segments) - return resolved if os.path.isabs(resolved) else os.path.abspath(resolved) - - -def _qwen_home_from_env_file(file_path: str) -> Optional[str]: - raw = read_optional(file_path, "Qwen home environment") - if raw is None: - return None - value = parse_dotenv(raw).get("QWEN_HOME") - return value or None - - -def resolve_qwen_home() -> str: - initial_value = os.environ.get("QWEN_HOME") - initial_home = ( - _resolve_storage_path(initial_value) if initial_value else _default_qwen_home() - ) - if "QWEN_HOME" in os.environ: - return initial_home - candidates = ( - os.path.join(initial_home, ".env"), - os.path.join(os.path.dirname(initial_home), ".env"), - ) - for candidate in candidates: - discovered = _qwen_home_from_env_file(candidate) - if discovered: - return _resolve_storage_path(discovered) - return initial_home - - -def read_document() -> QwenDocument: - config_path = os.path.join(resolve_qwen_home(), "settings.json") - raw = read_optional(config_path, "Qwen Code settings") - if raw is None: - return create_owned_document(config_path) - return parse_document(exists=True, file_path=config_path, raw=raw) - - -def rendered_change(rendered: RenderedDocument) -> Optional[FileChange]: - if not rendered.changed: - return None - return source_change( - rendered.document.file_path, - "Qwen Code settings", - rendered.document.raw if rendered.document.exists else None, - rendered.next_source, - 0o600, - ) - - -def _read_runtime_config(file_path: str) -> Optional[JsonObject]: - raw = read_optional(file_path, "Elydora runtime config") - if raw is None: - return None +def validate_api_origin(value: str, label: str = "base_url") -> None: try: - value = json.loads(raw) - except json.JSONDecodeError as error: + 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"Failed to parse Elydora runtime config at {file_path}: {error}" + 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 not isinstance(value, dict): + 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 at {file_path} must contain a JSON object" + f"Elydora runtime config {field} is invalid: {config_path}" ) return value -def runtime_files_exist(contracts: list[RuntimeContract]) -> bool: - for contract in contracts: - agent_directory = os.path.dirname(contract.guard_path) - runtime_config = _read_runtime_config( - os.path.join(agent_directory, "config.json") +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}' ) - config_agent_id = runtime_config.get("agent_id") if runtime_config else None - if ( - runtime_config is None - or runtime_config.get("agent_name") != AGENT_KEY - or not isinstance(config_agent_id, str) - or not same_agent_id(config_agent_id, contract.agent_id) - ): - continue - if regular_file_exists( - contract.guard_path, - "Elydora guard runtime", - ) and regular_file_exists( - contract.audit_path, - "Elydora audit runtime", - ): - return True - return False + _require_non_empty_string(config.get("org_id"), "org_id", config_path) + _require_non_empty_string(config.get("kid"), "kid", config_path) + agent_id = _require_non_empty_string( + config.get("agent_id"), "agent_id", config_path + ) + if ( + not same_qwen_agent_id(agent_id, expected_agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime identity does not match Qwen Code hooks: " + f"{config_path}" + ) + if "token" in config: + _require_non_empty_string(config.get("token"), "token", config_path) + base_url = _require_non_empty_string( + config.get("base_url"), "base_url", config_path + ) + validate_api_origin(base_url, "Elydora runtime config base_url") -__all__ = [ - "read_document", - "rendered_change", - "require_runtime", - "resolve_qwen_home", - "runtime_files_exist", -] +def _read_runtime_config(file_path: str) -> Optional[JsonObject]: + snapshot = read_physical_file( + file_path, "Elydora runtime config", MAX_CONFIG_BYTES + ) + if snapshot is None: + return None + return parse_json_object( + snapshot.contents, f"Elydora runtime config at {file_path}" + ) + + +def validate_runtime_tree(agent_directory: str, agent_id: str) -> None: + root = elydora_dir() + if not physical_directory_exists(root, "Elydora runtime directory"): + return + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return + config_path = os.path.join(agent_directory, "config.json") + config = _read_runtime_config(config_path) + artifacts = ( + ("private.key", "Elydora private key"), + ("guard.py", "Elydora guard runtime"), + ("hook.py", "Elydora audit runtime"), + ("chain-state.json", "Elydora chain state"), + ("status-cache.json", "Elydora status cache"), + ("error.log", "Elydora error log"), + ) + artifact_exists = any( + physical_file_exists(os.path.join(agent_directory, name), label) + for name, label in artifacts + ) + if config is None: + if artifact_exists: + raise ValueError( + "Elydora runtime identity cannot be verified without config.json: " + f"{agent_directory}" + ) + return + configured_id = config.get("agent_id") + if ( + not isinstance(configured_id, str) + or not same_qwen_agent_id(configured_id, agent_id) + or config.get("agent_name") != AGENT_KEY + ): + raise ValueError( + "Elydora runtime config identity does not match Qwen Code agent " + f"{agent_id}: {config_path}" + ) + + +def _runtime_contract_exists(contract: QwenRuntimeContract) -> bool: + root = elydora_dir() + agent_directory = os.path.dirname(contract.guard_path) + if ( + not same_qwen_path(os.path.dirname(agent_directory), root) + or not same_qwen_path( + contract.audit_path, os.path.join(agent_directory, "hook.py") + ) + ): + return False + if not physical_directory_exists(root, "Elydora runtime directory"): + return False + if not physical_directory_exists( + agent_directory, "Elydora agent runtime directory" + ): + return False + config_path = os.path.join(agent_directory, "config.json") + key_path = os.path.join(agent_directory, "private.key") + config = _read_runtime_config(config_path) + key = read_physical_file(key_path, "Elydora private key", MAX_SECRET_BYTES) + guard = read_physical_file(contract.guard_path, "Elydora guard runtime") + audit = read_physical_file(contract.audit_path, "Elydora audit runtime") + if config is None or key is None or guard is None or audit is None: + return False + _validate_runtime_config(config, contract.agent_id, config_path) + validate_private_key(key.contents, "Elydora private key") + expected_guard = generate_guard_script(AGENT_KEY, contract.agent_id) + expected_audit = generate_hook_script( + org_id=str(config["org_id"]), + agent_id=contract.agent_id, + kid=str(config["kid"]), + base_url=str(config["base_url"]), + native_payload=True, + agent_name=AGENT_KEY, + ) + return guard.contents == expected_guard and audit.contents == expected_audit + + +def qwen_runtime_files_exist( + contracts: List[QwenRuntimeContract], +) -> bool: + return any(_runtime_contract_exists(contract) for contract in contracts) diff --git a/sdks/python/elydora/plugins/qwen_sources.py b/sdks/python/elydora/plugins/qwen_sources.py new file mode 100644 index 0000000..86ee32d --- /dev/null +++ b/sdks/python/elydora/plugins/qwen_sources.py @@ -0,0 +1,373 @@ +"""Qwen Code 0.20 settings discovery, trust, and effective controls.""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import re +import sys +import tempfile +from typing import Dict, List, Optional, Set, Tuple + +from ._dotenv import parse_dotenv +from ._managed_files import MAX_SOURCE_BYTES, FileSnapshot, read_physical_file +from ._transaction import FilePrecondition +from .qwen_command import same_qwen_path +from .qwen_config import ( + QwenDocument, + create_qwen_document, + parse_qwen_document, + parse_qwen_jsonc_object, + qwen_document_label, + qwen_source_label, +) +from .qwen_contract import validate_javascript_matchers + + +_HOME_ENV_KEYS = ("QWEN_HOME", "QWEN_RUNTIME_DIR") + + +@dataclass(frozen=True) +class QwenDisableControl: + disabled: bool + source: Optional[QwenDocument] + + +@dataclass(frozen=True) +class QwenSources: + qwen_home: str + system_defaults: QwenDocument + user: QwenDocument + workspace: QwenDocument + system: QwenDocument + workspace_active: bool + workspace_trusted: bool + disable_control: QwenDisableControl + preconditions: Tuple[FilePrecondition, ...] + + +@dataclass(frozen=True) +class _RoutingResult: + qwen_home: str + preconditions: Tuple[FilePrecondition, ...] + + +def _default_qwen_home() -> str: + home = os.path.expanduser("~") + if not home or home == "~": + home = tempfile.gettempdir() + return os.path.join(home, ".qwen") + + +def _resolve_config_path(value: str) -> str: + resolved = value + if value == "~": + resolved = os.path.expanduser("~") + elif value.startswith(("~/", "~\\")): + segments = [item for item in re.split(r"[/\\]+", value[2:]) if item] + resolved = os.path.join(os.path.expanduser("~"), *segments) + return resolved if os.path.isabs(resolved) else os.path.abspath(resolved) + + +def _environment_value(key: str) -> Tuple[bool, Optional[str]]: + return key in os.environ, os.environ.get(key) + + +def _resolve_qwen_routing() -> _RoutingResult: + values: Dict[str, Optional[str]] = {} + owned: Set[str] = set() + for key in _HOME_ENV_KEYS: + present, value = _environment_value(key) + values[key] = value + if present: + owned.add(key) + if all(values.get(key) for key in _HOME_ENV_KEYS): + return _RoutingResult( + _resolve_config_path(values["QWEN_HOME"] or ""), () + ) + + initial_qwen_home = values.get("QWEN_HOME") + initial_directory = ( + _resolve_config_path(initial_qwen_home) + if initial_qwen_home + else _default_qwen_home() + ) + candidates = [os.path.join(initial_directory, ".env")] + if not initial_qwen_home: + candidates.append(os.path.join(os.path.dirname(initial_directory), ".env")) + preconditions: List[FilePrecondition] = [] + visited: Set[str] = set() + + def read_candidate(file_path: str) -> None: + resolved = os.path.abspath(file_path) + key = os.path.normcase(resolved) + if key in visited: + return + visited.add(key) + snapshot = read_physical_file( + resolved, "Qwen Code home environment", MAX_SOURCE_BYTES + ) + preconditions.append(FilePrecondition( + resolved, + "Qwen Code home environment", + snapshot, + MAX_SOURCE_BYTES, + )) + if snapshot is None: + return + parsed = parse_dotenv(snapshot.contents) + for env_key in _HOME_ENV_KEYS: + value = parsed.get(env_key) + if value and env_key not in owned: + values[env_key] = value + owned.add(env_key) + + for candidate in candidates: + read_candidate(candidate) + discovered_home = values.get("QWEN_HOME") + if discovered_home and discovered_home != initial_qwen_home: + discovered_directory = _resolve_config_path(discovered_home) + if not same_qwen_path(discovered_directory, initial_directory): + read_candidate(os.path.join(discovered_directory, ".env")) + resolved_home = values.get("QWEN_HOME") + return _RoutingResult( + _resolve_config_path(resolved_home) if resolved_home else _default_qwen_home(), + tuple(preconditions), + ) + + +def resolve_qwen_home() -> str: + return _resolve_qwen_routing().qwen_home + + +def _system_settings_path() -> str: + configured = os.environ.get("QWEN_CODE_SYSTEM_SETTINGS_PATH") + if configured: + return os.path.abspath(configured) + if sys.platform == "darwin": + return "/Library/Application Support/QwenCode/settings.json" + if os.name == "nt": + return r"C:\ProgramData\qwen-code\settings.json" + return "/etc/qwen-code/settings.json" + + +def _system_defaults_path(system_path: str) -> str: + configured = os.environ.get("QWEN_CODE_SYSTEM_DEFAULTS_PATH") + return ( + os.path.abspath(configured) + if configured + else os.path.join(os.path.dirname(system_path), "system-defaults.json") + ) + + +def _read_document(kind: str, file_path: str) -> QwenDocument: + snapshot = read_physical_file( + file_path, qwen_source_label(kind), MAX_SOURCE_BYTES + ) + if snapshot is None: + return create_qwen_document(kind, file_path) + return parse_qwen_document( + kind=kind, + exists=True, + file_path=file_path, + raw=snapshot.contents, + snapshot=snapshot, + ) + + +def _canonical_path(file_path: str) -> str: + try: + return os.path.realpath(file_path) + except OSError: + return os.path.abspath(file_path) + + +def _comparison_path(file_path: str) -> str: + return os.path.normcase(os.path.abspath(file_path)) + + +def _is_within(child: str, parent: str) -> bool: + try: + relative = os.path.relpath(_comparison_path(child), _comparison_path(parent)) + except ValueError: + return False + return relative == "." or ( + relative != ".." + and not relative.startswith(".." + os.sep) + and not os.path.isabs(relative) + ) + + +def _workspace_trust( + system: QwenDocument, + user: QwenDocument, + qwen_home: str, + workspace_path: str, +) -> Tuple[bool, Optional[FilePrecondition]]: + enabled = ( + user.folder_trust_enabled + if user.folder_trust_enabled is not None + else system.folder_trust_enabled + ) + if enabled is not True: + return True, None + configured = os.environ.get("QWEN_CODE_TRUSTED_FOLDERS_PATH") + file_path = ( + os.path.abspath(configured) + if configured + else os.path.join(qwen_home, "trustedFolders.json") + ) + snapshot = read_physical_file( + file_path, "Qwen Code trusted folders", MAX_SOURCE_BYTES + ) + precondition = FilePrecondition( + file_path, + "Qwen Code trusted folders", + snapshot, + MAX_SOURCE_BYTES, + ) + if snapshot is None: + return True, precondition + rules = parse_qwen_jsonc_object( + snapshot.contents, f"Qwen Code trusted folders at {file_path}" + ) + allowed_levels = {"TRUST_FOLDER", "TRUST_PARENT", "DO_NOT_TRUST"} + for rule_path, level in rules.items(): + if level not in allowed_levels: + raise ValueError( + "Qwen Code trusted folders has invalid trust level for " + f'"{rule_path}"' + ) + workspace = _canonical_path(workspace_path) + for rule_path, level in rules.items(): + canonical_rule = _canonical_path(rule_path) + trust_root = ( + os.path.dirname(canonical_rule) + if level == "TRUST_PARENT" + else canonical_rule + ) + if level in {"TRUST_FOLDER", "TRUST_PARENT"} and _is_within( + workspace, trust_root + ): + return True, precondition + for rule_path, level in rules.items(): + if level == "DO_NOT_TRUST" and _comparison_path( + workspace + ) == _comparison_path(_canonical_path(rule_path)): + return False, precondition + return True, precondition + + +def _effective_disable( + system_defaults: QwenDocument, + user: QwenDocument, + workspace: QwenDocument, + system: QwenDocument, + use_workspace: bool, +) -> QwenDisableControl: + disabled = False + source: Optional[QwenDocument] = None + documents = [system_defaults, user] + if use_workspace: + documents.append(workspace) + documents.append(system) + for document in documents: + if document.disable_all_hooks is None: + continue + disabled = document.disable_all_hooks + source = document + return QwenDisableControl(disabled, source) + + +def _source_precondition(document: QwenDocument) -> FilePrecondition: + return FilePrecondition( + document.file_path, + qwen_document_label(document), + document.snapshot, + MAX_SOURCE_BYTES, + ) + + +def _deduplicate_preconditions( + values: List[FilePrecondition], +) -> Tuple[FilePrecondition, ...]: + result: Dict[str, FilePrecondition] = {} + for value in values: + result.setdefault(_comparison_path(value.file_path), value) + return tuple(result.values()) + + +def read_qwen_sources() -> QwenSources: + routing = _resolve_qwen_routing() + system_path = _system_settings_path() + workspace_path = os.path.join(os.getcwd(), ".qwen", "settings.json") + system = _read_document("system", system_path) + system_defaults = _read_document( + "system-defaults", _system_defaults_path(system_path) + ) + user = _read_document( + "user", os.path.join(routing.qwen_home, "settings.json") + ) + canonical_workspace = _canonical_path(os.getcwd()) + canonical_home = _canonical_path(os.path.expanduser("~")) + workspace_active = _comparison_path(canonical_workspace) != _comparison_path( + canonical_home + ) + workspace = ( + _read_document("workspace", workspace_path) + if workspace_active + else create_qwen_document("workspace", workspace_path) + ) + if workspace_active: + workspace_trusted, trust_precondition = _workspace_trust( + system, user, routing.qwen_home, canonical_workspace + ) + else: + workspace_trusted, trust_precondition = False, None + validate_javascript_matchers([ + system_defaults.hooks, + user.hooks, + workspace.hooks, + system.hooks, + ]) + preconditions = list(routing.preconditions) + preconditions.extend([ + _source_precondition(system_defaults), + _source_precondition(user), + ]) + if workspace_active: + preconditions.append(_source_precondition(workspace)) + preconditions.append(_source_precondition(system)) + if trust_precondition is not None: + preconditions.append(trust_precondition) + return QwenSources( + qwen_home=routing.qwen_home, + system_defaults=system_defaults, + user=user, + workspace=workspace, + system=system, + workspace_active=workspace_active, + workspace_trusted=workspace_trusted, + disable_control=_effective_disable( + system_defaults, + user, + workspace, + system, + workspace_active and workspace_trusted, + ), + preconditions=_deduplicate_preconditions(preconditions), + ) + + +def require_qwen_hooks_enabled(sources: QwenSources) -> None: + if not sources.disable_control.disabled: + return + source = sources.disable_control.source + location = ( + f"{qwen_document_label(source)} at {source.file_path}" + if source is not None + else "effective settings" + ) + raise ValueError( + f"Qwen Code hooks are disabled by disableAllHooks in {location}" + ) diff --git a/sdks/python/elydora/plugins/registry.py b/sdks/python/elydora/plugins/registry.py index 33aa597..d902d4d 100644 --- a/sdks/python/elydora/plugins/registry.py +++ b/sdks/python/elydora/plugins/registry.py @@ -78,7 +78,7 @@ SUPPORTED_AGENTS: Dict[str, Dict[str, str]] = { }, "qwen": { "name": "Qwen Code", - "hook_event": "PreToolUse/PostToolUse", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", "config_path": "~/.qwen/settings.json", }, } diff --git a/sdks/python/tests/qwen_support.py b/sdks/python/tests/qwen_support.py index 680bec1..a346acf 100644 --- a/sdks/python/tests/qwen_support.py +++ b/sdks/python/tests/qwen_support.py @@ -1,134 +1,309 @@ from __future__ import annotations -import base64 +from dataclasses import dataclass +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json +import ntpath import os from pathlib import Path import subprocess # nosec B404 import sys -from typing import Any, Dict, Optional +from threading import Thread +from typing import Any, Mapping, Optional import pytest -from elydora.plugins import qwen from elydora.plugins._jsonc import parse_jsonc -from elydora.plugins.base import InstallConfig +from elydora.plugins.qwen import QwenPlugin +from elydora.plugins.qwen_command import build_qwen_command +from elydora.utils import base64url_encode AGENT_ID = "agent-1" MISSING = object() -JsonObject = Dict[str, Any] - - -def generated_command(script_path: Path) -> str: - if os.name == "nt": - executable = "'" + sys.executable.replace("'", "''") + "'" - script = "'" + str(script_path).replace("'", "''") + "'" - return f"& {executable} {script}; exit $LASTEXITCODE" - executable = "'" + sys.executable.replace("'", "'\"'\"'") + "'" - script = "'" + str(script_path).replace("'", "'\"'\"'") + "'" - return f"{executable} {script}" - - -def parse_settings(path: Path) -> JsonObject: - value = parse_jsonc( - path.read_text(encoding="utf-8"), - str(path), - allow_trailing_commas=False, - ) - assert isinstance(value, dict) - return value - - -def managed_handler( - settings: JsonObject, - event: str, - script_path: Path, -) -> Optional[JsonObject]: - command = generated_command(script_path) - for group in settings.get("hooks", {}).get(event, []): - for handler in group["hooks"]: - if handler.get("command") == command: - return handler - return None - - -def run_handler( - handler: JsonObject, - payload: str, -) -> subprocess.CompletedProcess[str]: - if handler["shell"] == "powershell": - command = [ - "powershell", - "-NoProfile", - "-NonInteractive", - "-Command", - handler["command"], - ] - else: - command = ["bash", "-c", handler["command"]] - return subprocess.run( # nosec B603 - command, - input=payload, - text=True, - capture_output=True, - check=False, - timeout=10, - ) +VALID_PRIVATE_KEY = base64url_encode(bytes(range(32))) +GUARD_NAME = "elydora-guard" +AUDIT_NAME = "elydora-audit" +@dataclass class QwenFixture: - def __init__( - self, - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - *, - existing_settings: Any = MISSING, - create_guard: bool = True, - ) -> None: - self.home_dir = tmp_path / "home with spaces and 'quote" - self.workspace_dir = self.home_dir / "workspace" - self.qwen_dir = self.home_dir / ".qwen" - self.config_path = self.qwen_dir / "settings.json" - self.agent_dir = self.home_dir / ".elydora" / AGENT_ID - self.guard_path = self.agent_dir / "guard.py" - self.audit_path = self.agent_dir / "hook.py" - self.runtime_config = self.agent_dir / "config.json" - self.private_key_path = self.agent_dir / "private.key" - self.workspace_dir.mkdir(parents=True) - self.agent_dir.mkdir(parents=True) - if create_guard: - self.guard_path.write_text( - "import sys\n" - "sys.stdin.buffer.read()\n" - "sys.stderr.write('Agent is frozen by Elydora.\\n')\n" - "raise SystemExit(2)\n", - encoding="utf-8", - ) - if existing_settings is not MISSING: - self._write(self.config_path, existing_settings) - monkeypatch.setenv("HOME", str(self.home_dir)) - monkeypatch.setenv("USERPROFILE", str(self.home_dir)) - monkeypatch.delenv("QWEN_HOME", raising=False) - monkeypatch.chdir(self.workspace_dir) - private_key = base64.urlsafe_b64encode(bytes([1]) * 32).rstrip(b"=").decode() - self.config: InstallConfig = { - "agent_id": AGENT_ID, - "agent_name": "qwen", - "org_id": "org-1", - "private_key": private_key, - "kid": "kid-1", - "base_url": "https://api.elydora.test", - "guard_script_path": str(self.guard_path), - } - self.plugin = qwen.QwenPlugin() - - @staticmethod - def _write(path: Path, value: Any) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - raw = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n" - with open(path, "w", encoding="utf-8", newline="") as file: - file.write(raw) + root_dir: Path + home_dir: Path + project_dir: Path + config_path: Path + system_defaults_path: Path + system_path: Path + trusted_folders_path: Path + agent_dir: Path + guard_path: Path + audit_path: Path + runtime_config_path: Path + private_key_path: Path + plugin: QwenPlugin + config: dict[str, str] def install(self) -> None: self.plugin.install(self.config) + + def source(self, file_path: Optional[Path] = None) -> str: + path = file_path or self.config_path + with path.open("r", encoding="utf-8", newline="") as file: + return file.read() + + def settings(self, file_path: Optional[Path] = None) -> dict[str, Any]: + path = file_path or self.config_path + value = parse_jsonc( + self.source(path), + f"Qwen Code test settings at {path}", + allow_trailing_commas=False, + ) + assert isinstance(value, dict) + return value + + +class ElydoraApiHandler(BaseHTTPRequestHandler): + agent_status: Any = "active" + operation_status = 201 + requests: list[dict[str, Any]] = [] + + @classmethod + def reset(cls, status: Any, operation_status: int) -> None: + cls.agent_status = status + cls.operation_status = operation_status + 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), + }) + self._respond( + type(self).operation_status, + {"operation": {"accepted": True}}, + ) + + def log_message(self, _format: str, *_args: object) -> None: + return + + +def start_api_server( + *, status: Any = "active", operation_status: int = 201 +) -> ThreadingHTTPServer: + ElydoraApiHandler.reset(status, operation_status) + server = ThreadingHTTPServer(("127.0.0.1", 0), ElydoraApiHandler) + 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", newline="") + os.chmod(path, mode) + + +def write_json(path: Path, value: object, mode: int = 0o600) -> None: + source = value if isinstance(value, str) else json.dumps(value, indent=2) + "\n" + write_text(path, source, mode) + + +def prepare_fixture( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + existing_settings: object = MISSING, + base_url: str = "http://127.0.0.1:9", +) -> QwenFixture: + home_dir = tmp_path / "home with spaces and 'quote $QWEN_CWD %QWEN_CWD%" + project_dir = tmp_path / "project with spaces" + project_dir.mkdir(parents=True) + monkeypatch.setenv("HOME", str(home_dir)) + monkeypatch.setenv("USERPROFILE", str(home_dir)) + monkeypatch.delenv("QWEN_HOME", raising=False) + monkeypatch.delenv("QWEN_RUNTIME_DIR", raising=False) + monkeypatch.chdir(project_dir) + config_path = home_dir / ".qwen" / "settings.json" + system_defaults_path = tmp_path / "system-defaults.json" + system_path = tmp_path / "system-settings.json" + trusted_folders_path = tmp_path / "trusted-folders.json" + monkeypatch.setenv( + "QWEN_CODE_SYSTEM_DEFAULTS_PATH", str(system_defaults_path) + ) + monkeypatch.setenv("QWEN_CODE_SYSTEM_SETTINGS_PATH", str(system_path)) + monkeypatch.setenv( + "QWEN_CODE_TRUSTED_FOLDERS_PATH", str(trusted_folders_path) + ) + agent_dir = home_dir / ".elydora" / AGENT_ID + guard_path = agent_dir / "guard.py" + audit_path = agent_dir / "hook.py" + runtime_config_path = agent_dir / "config.json" + private_key_path = agent_dir / "private.key" + if existing_settings is not MISSING: + write_json(config_path, existing_settings) + return QwenFixture( + root_dir=tmp_path, + home_dir=home_dir, + project_dir=project_dir, + config_path=config_path, + system_defaults_path=system_defaults_path, + system_path=system_path, + trusted_folders_path=trusted_folders_path, + agent_dir=agent_dir, + guard_path=guard_path, + audit_path=audit_path, + runtime_config_path=runtime_config_path, + private_key_path=private_key_path, + plugin=QwenPlugin(), + config={ + "org_id": "org-1", + "agent_id": AGENT_ID, + "agent_name": "qwen", + "private_key": VALID_PRIVATE_KEY, + "kid": "kid-1", + "token": "token-1", + "base_url": base_url, + "guard_script_path": str(guard_path), + }, + ) + + +def managed_handler( + settings: Mapping[str, Any], event: str, name: str +) -> dict[str, Any]: + for group in settings.get("hooks", {}).get(event, []): + if set(group) != {"hooks"}: + continue + for handler in group["hooks"]: + if ( + set(handler) + == {"type", "name", "command", "shell", "timeout"} + and handler.get("type") == "command" + and handler.get("name") == name + and handler.get("shell") + == ("powershell" if os.name == "nt" else "bash") + and handler.get("timeout") == 10_000 + ): + return handler + raise AssertionError(f"managed handler for {event!r} not found") + + +def assert_managed_triple(settings: Mapping[str, Any]) -> dict[str, Any]: + guard = managed_handler(settings, "PreToolUse", GUARD_NAME) + audit = managed_handler(settings, "PostToolUse", AUDIT_NAME) + failure = managed_handler(settings, "PostToolUseFailure", AUDIT_NAME) + assert audit["command"] == failure["command"] + return {"guard": guard, "audit": audit, "failure": failure} + + +def legacy_group(script_path: Path) -> dict[str, Any]: + return { + "matcher": "*", + "hooks": [{ + "type": "command", + "command": build_qwen_command(str(script_path)), + "shell": "powershell" if os.name == "nt" else "bash", + "timeout": 10_000, + }], + } + + +def run_handler( + handler: Mapping[str, Any], payload: bytes, fixture: QwenFixture +) -> subprocess.CompletedProcess[bytes]: + environment = { + **os.environ, + "HOME": str(fixture.home_dir), + "USERPROFILE": str(fixture.home_dir), + } + if os.name == "nt": + powershell = ntpath.join( + os.environ.get("SystemRoot", r"C:\Windows"), + "System32", + "WindowsPowerShell", + "v1.0", + "powershell.exe", + ) + command = [ + powershell, + "-NoLogo", + "-NoProfile", + "-NonInteractive", + "-Command", + str(handler["command"]), + ] + else: + command = ["/bin/bash", "-c", str(handler["command"])] + return subprocess.run( # nosec B603 + command, + input=payload, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + timeout=15, + ) + + +def official_input(fixture: QwenFixture, event: str) -> dict[str, Any]: + payload: dict[str, Any] = { + "session_id": "session-1", + "transcript_path": str(fixture.project_dir / "transcript.jsonl"), + "cwd": str(fixture.project_dir), + "hook_event_name": event, + "timestamp": "2026-07-19T00:00:00.000Z", + "permission_mode": "default", + "tool_name": "run_shell_command", + "tool_input": {"command": "echo test"}, + "tool_use_id": "toolu_1", + "tool_call_id": "call_1", + } + if event == "PostToolUse": + payload["tool_response"] = {"output": "test", "error": None} + if event == "PostToolUseFailure": + payload.update({"error": "command failed", "is_interrupt": False}) + return payload + + +def run_cli( + fixture: QwenFixture, arguments: list[str] +) -> subprocess.CompletedProcess[str]: + environment = dict(os.environ) + repository = str(Path(__file__).resolve().parents[1]) + existing = environment.get("PYTHONPATH") + environment["PYTHONPATH"] = ( + repository + os.pathsep + existing if existing else repository + ) + return subprocess.run( # nosec B603 + [sys.executable, "-m", "elydora.cli", *arguments], + text=True, + capture_output=True, + check=False, + cwd=fixture.project_dir, + env=environment, + timeout=30, + ) diff --git a/sdks/python/tests/test_qwen_plugin.py b/sdks/python/tests/test_qwen_plugin.py index f3cb2f1..2a2c045 100644 --- a/sdks/python/tests/test_qwen_plugin.py +++ b/sdks/python/tests/test_qwen_plugin.py @@ -9,30 +9,41 @@ import pytest from elydora import cli from elydora.plugins import _transaction, qwen +from elydora.plugins.qwen_config import render_qwen_document +from elydora.plugins.qwen_contract import ( + AUDIT_HOOK_NAME, + GUARD_HOOK_NAME, + build_qwen_group, +) +from elydora.plugins.qwen_installation import ( + commit_qwen_installation, + preflight_qwen_installation, + prepare_qwen_installation, +) +from elydora.plugins.qwen_sources import read_qwen_sources from elydora.plugins.registry import SUPPORTED_AGENTS from qwen_support import ( AGENT_ID, - QwenFixture, - generated_command, - managed_handler, - parse_settings, - run_handler, + VALID_PRIVATE_KEY, + assert_managed_triple, + legacy_group, + prepare_fixture, + run_cli, + write_json, + write_text, ) -EXPECTED_SHELL = "powershell" if os.name == "nt" else "bash" - - def test_qwen_is_registered_in_the_sdk_and_cli() -> None: assert SUPPORTED_AGENTS["qwen"] == { "name": "Qwen Code", - "hook_event": "PreToolUse/PostToolUse", + "hook_event": "PreToolUse/PostToolUse/PostToolUseFailure", "config_path": "~/.qwen/settings.json", } assert cli.PLUGIN_MAP["qwen"] is qwen.QwenPlugin -def test_install_preserves_jsonc_and_is_idempotent( +def test_install_preserves_all_sources_and_is_idempotent( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], @@ -42,208 +53,132 @@ def test_install_preserves_jsonc_and_is_idempotent( " // Keep this user preference.\r\n" ' "theme": "GitHub",\r\n' ' "hooks": {\r\n' - ' "SessionStart": [{ "hooks": [{ "type": "command", "command": "session-hook" }] }],\r\n' - ' "PreToolUse": [{ "matcher": "read_file", "hooks": [{ "type": "command", "command": "user-hook" }] }]\r\n' + ' "FutureEvent": [null],\r\n' + ' "PreToolUse": [{ "matcher": "read_file", "hooks": ' + '[{ "type": "command", "command": "user-hook" }] }]\r\n' " }\r\n" "}\r\n" ) - fixture = QwenFixture(monkeypatch, tmp_path, existing_settings=source) - workspace_settings = fixture.workspace_dir / ".qwen" / "settings.json" - workspace_settings.parent.mkdir(parents=True) - workspace_settings.write_text('{ "owner": "workspace" }\n', encoding="utf-8") - fixture.install() - fixture.install() - output = capsys.readouterr().out - with open(fixture.config_path, "r", encoding="utf-8", newline="") as file: - raw = file.read() - settings = parse_settings(fixture.config_path) - assert "Keep this user preference" in raw - assert "\r\n" in raw - assert settings["theme"] == "GitHub" - assert settings["hooks"]["SessionStart"][0]["hooks"][0]["command"] == "session-hook" - assert len(settings["hooks"]["PreToolUse"]) == 2 - assert len(settings["hooks"]["PostToolUse"]) == 1 - for event, script_path in ( - ("PreToolUse", fixture.guard_path), - ("PostToolUse", fixture.audit_path), - ): - handler = managed_handler(settings, event, script_path) - assert handler is not None - assert sorted(handler) == ["command", "shell", "timeout", "type"] - assert handler["shell"] == EXPECTED_SHELL - assert handler["timeout"] == 10_000 - assert parse_settings(workspace_settings) == {"owner": "workspace"} - assert ( - json.loads(fixture.runtime_config.read_text(encoding="utf-8"))["agent_name"] - == "qwen" + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=source ) - assert fixture.private_key_path.is_file() - assert fixture.audit_path.is_file() - assert "run /hooks" in output + workspace_path = fixture.project_dir / ".qwen" / "settings.json" + write_json(workspace_path, {"owner": "workspace"}) + write_json(fixture.system_defaults_path, {"owner": "defaults"}) + write_json(fixture.system_path, {"owner": "system"}) + + fixture.install() + first_source = fixture.source() + first = fixture.settings() + assert "Keep this user preference" in first_source + assert "\r\n" in first_source + assert first["theme"] == "GitHub" + assert first["hooks"]["FutureEvent"] == [None] + assert first["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "user-hook" + assert_managed_triple(first) + for file_path in ( + fixture.guard_path, + fixture.audit_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert file_path.is_file() + + fixture.install() + assert fixture.source() == first_source + assert json.loads(workspace_path.read_text(encoding="utf-8")) == { + "owner": "workspace" + } + assert json.loads(fixture.system_defaults_path.read_text()) == { + "owner": "defaults" + } + assert json.loads(fixture.system_path.read_text()) == {"owner": "system"} + assert "run /hooks" in capsys.readouterr().out -def test_qwen_home_uses_official_user_env_precedence( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, +def test_qwen_home_uses_complete_official_bootstrap_precedence( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = QwenFixture(monkeypatch, tmp_path) + fixture = prepare_fixture(monkeypatch, tmp_path) first_home = tmp_path / "first # qwen home" second_home = tmp_path / "second qwen home" - fixture.qwen_dir.mkdir(parents=True) - (fixture.qwen_dir / ".env").write_text( + write_text( + fixture.home_dir / ".qwen" / ".env", f'export QWEN_HOME = "{first_home}" # selected by Qwen\n', - encoding="utf-8", - ) - (fixture.home_dir / ".env").write_text( - f"QWEN_HOME={second_home}\n", encoding="utf-8" ) + write_text(fixture.home_dir / ".env", f"QWEN_HOME={second_home}\n") + write_text(first_home / ".env", "QWEN_RUNTIME_DIR=runtime\n") + fixture.install() selected = first_home / "settings.json" - assert managed_handler(parse_settings(selected), "PreToolUse", fixture.guard_path) + assert_managed_triple(fixture.settings(selected)) assert not (second_home / "settings.json").exists() assert not fixture.config_path.exists() assert str(selected) in fixture.plugin.status()["details"] @pytest.mark.parametrize( - ("value", "relative"), + ("value", "expected"), [ - ("relative-qwen", "relative-qwen"), - ("~/custom-qwen", "custom-qwen"), - ("", ".qwen"), + ("relative-qwen", "project"), + ("~/custom-qwen", "home"), + ("", "default"), ], ) -def test_explicit_qwen_home_supports_relative_tilde_and_empty_values( +def test_explicit_qwen_home_preserves_process_ownership_semantics( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, value: str, - relative: str, + expected: str, ) -> None: - fixture = QwenFixture(monkeypatch, tmp_path / (relative.replace(".", "default"))) - fixture.qwen_dir.mkdir(parents=True) - ignored = tmp_path / "ignored-qwen-home" - (fixture.qwen_dir / ".env").write_text(f"QWEN_HOME={ignored}\n", encoding="utf-8") + fixture = prepare_fixture(monkeypatch, tmp_path) + ignored = tmp_path / "ignored-home" + write_text( + fixture.home_dir / ".qwen" / ".env", f"QWEN_HOME={ignored}\n" + ) monkeypatch.setenv("QWEN_HOME", value) fixture.install() - if value.startswith("relative"): - selected = fixture.workspace_dir / relative / "settings.json" - else: - selected = fixture.home_dir / relative / "settings.json" - assert "hooks" in parse_settings(selected) + paths = { + "project": fixture.project_dir / "relative-qwen" / "settings.json", + "home": fixture.home_dir / "custom-qwen" / "settings.json", + "default": fixture.config_path, + } + assert_managed_triple(fixture.settings(paths[expected])) assert not (ignored / "settings.json").exists() -def test_commands_block_and_forward_official_input_byte_for_byte( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, +def test_qwen_validates_current_schemas_and_preserves_future_events( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = QwenFixture(monkeypatch, tmp_path) + fixture = prepare_fixture(monkeypatch, tmp_path, existing_settings={ + "hooks": { + "FutureEvent": [None], + "MessageDisplay": [{ + "matcher": "[", + "hooks": [{ + "type": "http", + "url": "http://127.0.0.1:8080/hook", + "headers": {"Authorization": "Bearer ${TOKEN}"}, + "allowedEnvVars": ["TOKEN"], + "timeout": 10, + "once": True, + }], + }], + "Stop": [{ + "hooks": [{ + "type": "prompt", + "prompt": "Evaluate $ARGUMENTS", + "model": "fast", + "timeout": 30, + }], + }], + }, + }) fixture.install() - capture_path = tmp_path / "captured-event.json" - fixture.audit_path.write_text( - "import pathlib, sys\n" - f"pathlib.Path({str(capture_path)!r}).write_bytes(sys.stdin.buffer.read())\n", - encoding="utf-8", - ) - settings = parse_settings(fixture.config_path) - guard = managed_handler(settings, "PreToolUse", fixture.guard_path) - audit = managed_handler(settings, "PostToolUse", fixture.audit_path) - assert guard is not None and audit is not None - payload = json.dumps( - { - "session_id": "session-1", - "transcript_path": "transcript.jsonl", - "cwd": str(fixture.workspace_dir), - "hook_event_name": "PreToolUse", - "timestamp": "2026-07-19T00:00:00.000Z", - "tool_name": "run_shell_command", - "tool_input": {"command": "echo test"}, - } - ) - guard_result = run_handler(guard, payload) - assert guard_result.returncode == 2 - assert "Agent is frozen by Elydora" in guard_result.stderr - post_payload = payload.replace("PreToolUse", "PostToolUse") - audit_result = run_handler(audit, post_payload) - assert audit_result.returncode == 0, audit_result.stderr - assert capture_path.read_text(encoding="utf-8") == post_payload - - -def test_status_requires_enabled_pair_and_complete_runtime( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = QwenFixture(monkeypatch, tmp_path) - fixture.install() - assert fixture.plugin.status()["installed"] is True - settings = parse_settings(fixture.config_path) - settings["disableAllHooks"] = True - fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") - status = fixture.plugin.status() - assert status["installed"] is False - assert "disabled" in status["details"].lower() - settings["disableAllHooks"] = False - fixture.config_path.write_text(json.dumps(settings), encoding="utf-8") - fixture.guard_path.unlink() - status = fixture.plugin.status() - assert status["installed"] is False - assert "runtime" in status["details"].lower() - - -def test_uninstall_preserves_external_mutations_and_exact_ownership( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = QwenFixture( - monkeypatch, - tmp_path, - existing_settings='{"$version":4,"owner":"user"}', - ) - fixture.install() - settings = parse_settings(fixture.config_path) - managed_group = settings["hooks"]["PreToolUse"][-1] - managed_group["hooks"].append({"type": "command", "command": "user-command"}) - lookalike = fixture.home_dir / ".elydora" / "agent-10" / "guard.py" - settings["hooks"]["PreToolUse"].append( - { - "matcher": "*", - "hooks": [ - { - "type": "command", - "command": generated_command(lookalike), - "shell": EXPECTED_SHELL, - "timeout": 10_000, - } - ], - } - ) - fixture.config_path.write_text( - json.dumps(settings, indent=2) + "\n", encoding="utf-8" - ) - before = fixture.config_path.read_text(encoding="utf-8") - fixture.plugin.uninstall("other-agent") - assert fixture.config_path.read_text(encoding="utf-8") == before - uninstall_id = "AGENT-1" if os.name == "nt" else AGENT_ID - fixture.plugin.uninstall(uninstall_id) - remaining = parse_settings(fixture.config_path) - assert remaining["$version"] == 4 - assert remaining["owner"] == "user" - assert remaining["hooks"]["PreToolUse"][0]["hooks"][0]["command"] == "user-command" - assert "agent-10" in json.dumps(remaining) - assert "PostToolUse" not in remaining.get("hooks", {}) - - -def test_uninstall_deletes_an_empty_elydora_owned_settings_file( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = QwenFixture(monkeypatch, tmp_path) - fixture.install() - assert fixture.config_path.read_text(encoding="utf-8").startswith( - "// Managed by Elydora" - ) - fixture.plugin.uninstall(AGENT_ID) - assert not fixture.config_path.exists() + settings = fixture.settings() + assert settings["hooks"]["FutureEvent"] == [None] + assert settings["hooks"]["MessageDisplay"][0]["matcher"] == "[" + assert settings["hooks"]["Stop"][0]["hooks"][0]["type"] == "prompt" @pytest.mark.parametrize( @@ -254,81 +189,132 @@ def test_uninstall_deletes_an_empty_elydora_owned_settings_file( '{ "owner": true, }', '{ "hooks": {}, "hooks": {} }', '{ "disableAllHooks": "yes" }', + '{ "hooks": null }', '{ "hooks": [] }', - '{ "hooks": { "UnknownEvent": [] } }', '{ "hooks": { "PreToolUse": null } }', '{ "hooks": { "PreToolUse": [null] } }', + '{ "hooks": { "PreToolUse": [{ "matcher": 1, "hooks": [] }] } }', '{ "hooks": { "PreToolUse": [{ "matcher": "[", "hooks": [] }] } }', - '{ "hooks": { "PreToolUse": [{ "sequential": "yes", "hooks": [] }] } }', - '{ "hooks": { "PreToolUse": [{ "hooks": null }] } }', + '{ "hooks": { "Notification": [{ "matcher": "[", "hooks": [] }] } }', + '{ "hooks": { "PreToolUse": [{ "sequential": 1, "hooks": [] }] } }', + '{ "hooks": { "PreToolUse": [{}] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [null] }] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "function" }] }] } }', '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command" }] }] } }', - '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "http" }] }] }', - '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "function", "command": "x" }] }] } }', - '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command", "command": "x", "timeout": "ten" }] }] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "http" }] }] } }', + '{ "hooks": { "Stop": [{ "hooks": [{ "type": "prompt" }] }] } }', + '{ "hooks": { "PreToolUse": [{ "hooks": [{ "type": "command", "command": "x", "timeout": -1 }] }] } }', + '{ "security": { "folderTrust": { "enabled": "yes" } } }', ], ) -def test_install_rejects_malformed_settings_before_every_write( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - source: str, +def test_install_rejects_malformed_sources_before_every_write( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path, source: str ) -> None: - fixture = QwenFixture(monkeypatch, tmp_path, existing_settings=source) - with pytest.raises((ValueError, RuntimeError), match="Qwen"): + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings=source + ) + with pytest.raises((OSError, ValueError), match="Qwen"): fixture.install() - assert fixture.config_path.read_text(encoding="utf-8") == source - assert not fixture.audit_path.exists() - assert not fixture.runtime_config.exists() - assert not fixture.private_key_path.exists() + assert fixture.source() == source + assert not fixture.agent_dir.exists() -def test_install_fails_on_unreadable_env_and_missing_guard( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, +def test_install_surfaces_malformed_read_only_sources_and_routing( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - env_fixture = QwenFixture(monkeypatch, tmp_path / "env") - (env_fixture.qwen_dir / ".env").mkdir(parents=True) - with pytest.raises(OSError, match="Qwen home environment"): - env_fixture.install() - assert not env_fixture.config_path.exists() - assert not env_fixture.runtime_config.exists() + system = prepare_fixture(monkeypatch, tmp_path / "system") + write_text(system.system_path, "{ malformed") + with pytest.raises(ValueError, match="system override"): + system.install() + assert not system.config_path.exists() + assert not system.agent_dir.exists() - runtime_fixture = QwenFixture( + routing = prepare_fixture(monkeypatch, tmp_path / "routing") + (routing.home_dir / ".qwen" / ".env").mkdir(parents=True) + with pytest.raises(OSError, match="Qwen Code home environment"): + routing.install() + assert not routing.config_path.exists() + assert not routing.agent_dir.exists() + + +def test_disable_precedence_respects_workspace_trust( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user = prepare_fixture( monkeypatch, - tmp_path / "runtime", - create_guard=False, + tmp_path / "user", + existing_settings={"disableAllHooks": False}, ) - with pytest.raises(FileNotFoundError, match="guard runtime is missing"): - runtime_fixture.install() - assert not runtime_fixture.config_path.exists() - assert not runtime_fixture.runtime_config.exists() + write_json(user.system_defaults_path, {"disableAllHooks": True}) + user.install() + system = prepare_fixture(monkeypatch, tmp_path / "system") + write_json(system.system_path, {"disableAllHooks": True}) + with pytest.raises(ValueError, match="system override"): + system.install() -def test_status_surfaces_malformed_runtime_metadata_and_cleans_staging( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = QwenFixture(monkeypatch, tmp_path) - fixture.install() - assert [ - path - for path in fixture.home_dir.rglob("*") - if path.suffix in {".tmp", ".rollback"} - ] == [] - fixture.runtime_config.write_text("{ malformed", encoding="utf-8") - with pytest.raises(ValueError, match="parse Elydora runtime config"): - fixture.plugin.status() - - -def test_transaction_rolls_back_runtime_and_settings( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, -) -> None: - fixture = QwenFixture( + untrusted = prepare_fixture( monkeypatch, - tmp_path, - existing_settings={"owner": "user"}, + tmp_path / "untrusted", + existing_settings={"security": {"folderTrust": {"enabled": True}}}, ) - original = fixture.config_path.read_text(encoding="utf-8") + write_json( + untrusted.project_dir / ".qwen" / "settings.json", + {"disableAllHooks": True}, + ) + write_json( + untrusted.trusted_folders_path, + {str(untrusted.project_dir): "DO_NOT_TRUST"}, + ) + untrusted.install() + + trusted = prepare_fixture( + monkeypatch, + tmp_path / "trusted", + existing_settings={"security": {"folderTrust": {"enabled": True}}}, + ) + write_json( + trusted.project_dir / ".qwen" / "settings.json", + {"disableAllHooks": True}, + ) + write_json( + trusted.trusted_folders_path, + {str(trusted.project_dir): "TRUST_FOLDER"}, + ) + with pytest.raises(ValueError, match="workspace settings"): + trusted.install() + + +def test_transaction_aborts_when_read_only_source_changes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + write_json(fixture.system_path, {"owner": "before"}) + sources = read_qwen_sources() + paths = preflight_qwen_installation(fixture.config, sources) + rendered = render_qwen_document(sources.user, None, { + "PreToolUse": build_qwen_group(paths.guard_path, GUARD_HOOK_NAME), + "PostToolUse": build_qwen_group(paths.audit_path, AUDIT_HOOK_NAME), + "PostToolUseFailure": build_qwen_group( + paths.audit_path, AUDIT_HOOK_NAME + ), + }) + changes = prepare_qwen_installation(fixture.config, paths, rendered) + write_json(fixture.system_path, {"owner": "after"}) + with pytest.raises(OSError, match="system override settings changed"): + commit_qwen_installation(changes, sources) + assert json.loads(fixture.system_path.read_text()) == {"owner": "after"} + assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() + + +def test_transaction_rolls_back_every_runtime_and_settings_change( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture( + monkeypatch, tmp_path, existing_settings={"owner": "user"} + ) + original = fixture.source() real_replace = _transaction.os.replace failed = False @@ -340,13 +326,17 @@ def test_transaction_rolls_back_runtime_and_settings( real_replace(source, destination) monkeypatch.setattr(_transaction.os, "replace", fail_settings_commit) - with pytest.raises(OSError, match="Write Qwen Code installation"): + with pytest.raises(OSError, match="Install Qwen Code hooks"): fixture.install() - assert failed is True - assert fixture.config_path.read_text(encoding="utf-8") == original - assert not fixture.audit_path.exists() - assert not fixture.runtime_config.exists() - assert not fixture.private_key_path.exists() + assert failed + assert fixture.source() == original + for file_path in ( + fixture.guard_path, + fixture.audit_path, + fixture.runtime_config_path, + fixture.private_key_path, + ): + assert not file_path.exists() assert [ path for path in fixture.home_dir.rglob("*") @@ -354,12 +344,135 @@ def test_transaction_rolls_back_runtime_and_settings( ] == [] -def test_install_rejects_invalid_agent_id_before_writes( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, +def test_install_migrates_legacy_and_uninstall_preserves_lookalikes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: - fixture = QwenFixture(monkeypatch, tmp_path) - fixture.config["agent_id"] = "../escape" - with pytest.raises(ValueError, match="single non-empty path segment"): - fixture.install() + fixture = prepare_fixture(monkeypatch, tmp_path) + lookalike = legacy_group(fixture.guard_path) + lookalike["hooks"][0]["timeout"] = 9_000 + write_json(fixture.config_path, { + "owner": "user", + "hooks": { + "PreToolUse": [legacy_group(fixture.guard_path), lookalike], + "PostToolUse": [legacy_group(fixture.audit_path)], + }, + }) + fixture.install() + settings = fixture.settings() + assert_managed_triple(settings) + managed = settings["hooks"]["PreToolUse"][-1] + managed["userField"] = "preserve-group" + managed["hooks"].append({"type": "command", "command": "user-command"}) + write_json(fixture.config_path, settings) + + fixture.plugin.uninstall(AGENT_ID) + remaining = fixture.settings() + assert remaining["owner"] == "user" + assert any( + group["hooks"][0].get("timeout") == 9_000 + for group in remaining["hooks"]["PreToolUse"] + ) + assert any( + handler.get("command") == "user-command" + for group in remaining["hooks"]["PreToolUse"] + for handler in group["hooks"] + ) + assert "PostToolUse" not in remaining["hooks"] + assert "PostToolUseFailure" not in remaining["hooks"] + + +def test_uninstall_removes_owned_file_and_preserves_user_settings( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + user = prepare_fixture( + monkeypatch, + tmp_path / "user", + existing_settings={"theme": "GitHub", "hooks": {"Notification": []}}, + ) + user.install() + user.plugin.uninstall(AGENT_ID) + assert user.settings() == { + "theme": "GitHub", + "hooks": {"Notification": []}, + } + + owned = prepare_fixture(monkeypatch, tmp_path / "owned") + owned.install() + assert owned.source().startswith("// Managed by Elydora") + owned.plugin.uninstall(AGENT_ID) + assert not owned.config_path.exists() + + +def test_status_requires_exact_contract_and_runtime_identity( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + assert fixture.plugin.status()["installed"] is True + settings = fixture.settings() + settings["hooks"]["PostToolUseFailure"].append( + settings["hooks"]["PostToolUseFailure"][-1] + ) + write_json(fixture.config_path, settings) + assert fixture.plugin.status()["installed"] is False + + fixture.install() + write_text(fixture.private_key_path, "invalid") + with pytest.raises(ValueError, match="private key"): + fixture.plugin.status() + fixture.install() + write_text(fixture.guard_path, "tampered") + assert fixture.plugin.status()["installed"] is False + fixture.install() + runtime = json.loads(fixture.runtime_config_path.read_text()) + runtime["agent_name"] = "other" + write_json(fixture.runtime_config_path, runtime) + with pytest.raises(ValueError, match="runtime identity"): + fixture.plugin.status() + + +def test_preflight_rejects_unverifiable_runtime_and_invalid_agent_id( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + identity = prepare_fixture(monkeypatch, tmp_path / "identity") + write_text(identity.audit_path, "unverified") + with pytest.raises(ValueError, match="without config.json"): + identity.install() + assert not identity.config_path.exists() + + invalid = prepare_fixture(monkeypatch, tmp_path / "invalid") + invalid.config["agent_id"] = "../escape" + with pytest.raises(ValueError, match="Invalid agent ID"): + invalid.install() + assert not invalid.config_path.exists() + + +def test_qwen_cli_install_status_and_uninstall_end_to_end( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + key_file = tmp_path / "install-private.key" + token_file = tmp_path / "install-token.txt" + write_text(key_file, VALID_PRIVATE_KEY + "\n") + write_text(token_file, "token-1\n") + install = run_cli(fixture, [ + "install", + "--agent", "qwen", + "--org_id", "org-1", + "--agent_id", AGENT_ID, + "--kid", "kid-1", + "--private_key_file", str(key_file), + "--token_file", str(token_file), + "--base_url", "http://127.0.0.1:9", + ]) + assert install.returncode == 0, install.stderr + assert_managed_triple(fixture.settings()) + status = run_cli(fixture, ["status"]) + assert status.returncode == 0, status.stderr + assert "Qwen Code" in status.stdout and "[installed]" in status.stdout + uninstall = run_cli(fixture, [ + "uninstall", "--agent", "qwen", "--agent_id", AGENT_ID + ]) + assert uninstall.returncode == 0, uninstall.stderr assert not fixture.config_path.exists() + assert not fixture.agent_dir.exists() diff --git a/sdks/python/tests/test_qwen_runtime.py b/sdks/python/tests/test_qwen_runtime.py new file mode 100644 index 0000000..3e3680f --- /dev/null +++ b/sdks/python/tests/test_qwen_runtime.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import json +import os +from pathlib import Path +import stat + +import pytest + +from qwen_support import ( + ElydoraApiHandler, + assert_managed_triple, + official_input, + prepare_fixture, + run_handler, + server_base_url, + start_api_server, +) + + +def _close_server(server: object) -> None: + server.shutdown() # type: ignore[attr-defined] + server.server_close() # type: ignore[attr-defined] + + +def test_qwen_guard_accepts_active_agents_with_native_exit_semantics( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + server = start_api_server(status="active") + fixture = prepare_fixture( + monkeypatch, tmp_path, base_url=server_base_url(server) + ) + try: + fixture.install() + guard = assert_managed_triple(fixture.settings())["guard"] + payload = json.dumps( + official_input(fixture, "PreToolUse") + ).encode() + result = run_handler(guard, payload, fixture) + assert result.returncode == 0, result.stderr.decode() + assert result.stdout == b"" + assert len([ + request + for request in ElydoraApiHandler.requests + if request["method"] == "GET" + ]) == 1 + finally: + _close_server(server) + + +@pytest.mark.parametrize("status", ["frozen", "revoked"]) +def test_qwen_guard_propagates_blocking_states_through_exit_code_two( + 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) + ) + try: + fixture.install() + guard = assert_managed_triple(fixture.settings())["guard"] + result = run_handler( + guard, + json.dumps(official_input(fixture, "PreToolUse")).encode(), + fixture, + ) + assert result.returncode == 2 + assert result.stdout == b"" + assert status.encode() in result.stderr.lower() + finally: + _close_server(server) + + +def test_qwen_audit_forwards_complete_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) + ) + try: + fixture.install() + handlers = assert_managed_triple(fixture.settings()) + expected = [] + for handler_name, event in ( + ("audit", "PostToolUse"), + ("failure", "PostToolUseFailure"), + ): + payload = official_input(fixture, event) + payload["future_provider_field"] = {"preserved": True} + expected.append(payload) + result = run_handler( + handlers[handler_name], json.dumps(payload).encode(), fixture + ) + assert result.returncode == 0, result.stderr.decode() + assert result.stdout == b"" + posts = [ + request + for request in ElydoraApiHandler.requests + if request["method"] == "POST" + ] + assert len(posts) == 2 + assert posts[0]["json"]["payload"] == expected[0] + assert posts[1]["json"]["payload"] == expected[1] + finally: + _close_server(server) + + +def test_qwen_audit_keeps_delivery_failures_observable_and_fail_open( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + audit = assert_managed_triple(fixture.settings())["audit"] + result = run_handler( + audit, + json.dumps(official_input(fixture, "PostToolUse")).encode(), + fixture, + ) + assert result.returncode == 0, result.stderr.decode() + assert result.stdout == b"" + error_log = (fixture.agent_dir / "error.log").read_text(encoding="utf-8").lower() + assert "elydora-hook" in error_log and "urlerror" in error_log + + +@pytest.mark.skipif( + os.name == "nt", reason="POSIX mode bits are not authoritative on Windows" +) +def test_qwen_runtime_artifacts_use_private_modes( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + fixture = prepare_fixture(monkeypatch, tmp_path) + fixture.install() + for file_path, expected in ( + (fixture.runtime_config_path, 0o600), + (fixture.private_key_path, 0o600), + (fixture.guard_path, 0o700), + (fixture.audit_path, 0o700), + (fixture.config_path, 0o600), + ): + assert stat.S_IMODE(file_path.stat().st_mode) == expected