From 82e50b0f1b825ef40b60fcae4e748bfeb42f1d5e Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Tue, 11 Aug 2026 00:27:26 +0900 Subject: [PATCH] refactor(plugins): consolidate validation boundaries --- docs/configuration.md | 49 ++----- nanobot/agent/plugins.py | 215 +++++++++++----------------- nanobot/agent/skills.py | 58 +++++--- nanobot/apps/cli/service.py | 37 ++--- nanobot/webui/mcp_presets_api.py | 18 +-- tests/agent/test_agent_plugins.py | 102 ++++++------- tests/cli_apps/test_service.py | 9 +- tests/webui/test_mcp_presets_api.py | 36 ++--- 8 files changed, 202 insertions(+), 322 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 8255baabb..d290f61ff 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -2308,10 +2308,9 @@ Disabled skills are excluded from the main agent's skill summary, from always-on ### Agent Plugins v1 -nanobot also loads locally installed [Agent Plugins](https://agent-plugins.org/) from -`/plugins//`. Package presence in this directory is the installation state; -enabling it is a separate trust decision. A supported package has a root `plugin.json` that -targets Agent Plugins v1 and may provide skills, MCP servers, or both: +nanobot discovers [Agent Plugins](https://agent-plugins.org/) in +`/plugins//`. A v1 package has `plugin.json` and may provide skills, MCP +servers, or both: ```text plugins/ @@ -2323,38 +2322,20 @@ plugins/ └── SKILL.md ``` -Plugin skills use the same progressive loading and `$skill-name` invocation as workspace -skills after the plugin is explicitly enabled. Disabling a plugin removes both its skills and -MCP servers from the agent. A workspace skill wins when it has the same name as an enabled -plugin skill; plugin skills win over built-in skills. Invalid manifests, invalid Agent Skills, -nested skill directories, and paths that resolve outside the plugin root are ignored. +Directory presence means installed; activation is an explicit trust decision in **Apps**. +Enabled skills use normal progressive loading and `$skill-name` invocation. Workspace skills +override plugin skills, which override built-ins. Enabled `stdio` servers from `mcp.json` receive +contained `PLUGIN_ROOT` and isolated `PLUGIN_DATA` paths; explicit `tools.mcpServers` entries win +name collisions. Invalid manifests, components, nested skills, and escaping paths are ignored. -Portable MCP servers declared in `mcp.json` appear in **Apps**, but are never started merely -because a package exists. Enabling a plugin there is the explicit trust decision that activates -its executable components. The host expands `PLUGIN_ROOT` and an isolated `PLUGIN_DATA`, checks -package paths before launch, and hot-reloads MCP connections. Explicit `tools.mcpServers` -configuration wins over a plugin server if their host names collide. The v1 host currently -supports plugin `stdio` servers; unsupported remote transports are skipped independently. +Enabled plugins run as the nanobot user; declared permissions are descriptive, not an OS sandbox. +The optional `extensions.dev.nanobot.installCommand` is a shell-free argv run once per version +before local enable. Remote setup requires `tools.webuiAllowRemotePackageInstall`. The optional +`extensions.dev.nanobot.logo` accepts a contained PNG, JPEG, or WebP up to 256 KiB. -Treat enabled plugins as local code running with the nanobot user's privileges. Manifest -permissions are descriptive; nanobot does not currently enforce them with an OS sandbox. - -Plugins may optionally declare a shell-free `extensions.dev.nanobot.installCommand` array. The -local WebUI runs it once per plugin version before first enable; remote WebUI clients cannot run -plugin setup unless remote package installation was explicitly allowed. Enabling an already set -up package does not install it and is allowed remotely. Agent Plugins v1 deliberately leaves -distribution and installation UX to each host. A future catalog can therefore acquire, verify, -and place a package atomically before handing it to this same runtime; users should still see one -Install action, not separate download and installation steps. - -The optional `extensions.dev.nanobot.logo` field points to a packaged PNG, JPEG, or WebP asset -such as `./assets/icon.png`. nanobot only reads contained raster files up to 256 KiB and embeds -them locally in the Apps catalog; invalid or missing assets fall back to the plugin initials. - -CLI Apps installed from the WebUI use the same package layout. nanobot installs the CLI through -its catalog adapter, then writes and enables a skills-only Agent Plugin under -`/plugins/`; updates refresh that package and uninstall removes it. The external -executable remains managed by the CLI Apps installer rather than by the Agent Plugins manifest. +WebUI-installed CLI Apps use the same package layout as skills-only plugins. Their external +executables remain managed by the CLI Apps installer; update refreshes the package and uninstall +removes it. Future catalogs can acquire and place packages before using this same activation path. ## Tool Hint Max Length diff --git a/nanobot/agent/plugins.py b/nanobot/agent/plugins.py index d88ea6e8b..56d574f90 100644 --- a/nanobot/agent/plugins.py +++ b/nanobot/agent/plugins.py @@ -9,12 +9,13 @@ import subprocess from dataclasses import dataclass from hashlib import sha256 from pathlib import Path -from typing import Any, cast +from typing import cast -import yaml from filelock import FileLock from loguru import logger +from pydantic import ValidationError +from nanobot.agent.skills import parse_skill_metadata, valid_skill_metadata from nanobot.config.loader import get_config_path from nanobot.config.schema import MCPServerConfig @@ -22,12 +23,9 @@ AGENT_PLUGIN_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/plugin.schema.jso AGENT_PLUGIN_MCP_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json" _PLUGIN_NAME = re.compile(r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$") -_SKILL_NAME = re.compile(r"^(?!.*--)[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$") -_SKILL_FRONTMATTER = re.compile(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", re.DOTALL) _MCP_SERVER_FIELDS = {"type", "command", "args", "env", "cwd"} _SETUP_ENV = {"HOME", "LANG", "LC_ALL", "LOGNAME", "PATH", "SHELL", "TMPDIR", "USER"} _SETUP_TIMEOUT_SECONDS = 600 -_LOGO_SUFFIXES = {".jpeg", ".jpg", ".png", ".webp"} _MAX_LOGO_BYTES = 256 * 1024 @@ -70,18 +68,11 @@ class AgentPluginState: def _discover_agent_plugins(workspace: Path) -> list[AgentPlugin]: """Return installed packages found under ``/plugins/*``.""" workspace = workspace.expanduser().resolve() - plugins_root = workspace / "plugins" - if not plugins_root.is_dir(): + root = _contained_directory(workspace / "plugins", workspace) + if root is None: return [] try: - root = plugins_root.resolve(strict=True) - except OSError: - return [] - if not root.is_relative_to(workspace): - logger.warning("Ignoring Agent Plugins directory outside the workspace") - return [] - try: - candidates = sorted(plugins_root.iterdir(), key=lambda path: path.name) + candidates = sorted(root.iterdir(), key=lambda path: path.name) except OSError as exc: logger.warning("Could not inspect Agent Plugins directory: {}", exc) return [] @@ -107,19 +98,9 @@ def enabled_agent_plugin_skills(workspace: Path) -> list[AgentPluginSkill]: def _load_manifest(plugin_root: Path) -> AgentPlugin | None: - manifest = _contained_file(plugin_root / "plugin.json", plugin_root) - if manifest is None: + payload = _read_object(plugin_root / "plugin.json", plugin_root) + if payload is None: return None - try: - value = cast(object, json.loads(manifest.read_text(encoding="utf-8"))) - except (OSError, UnicodeError, json.JSONDecodeError) as exc: - logger.warning("Ignoring invalid Agent Plugin manifest '{}': {}", manifest, exc) - return None - if not isinstance(value, dict): - logger.warning("Ignoring Agent Plugin manifest '{}': expected a JSON object", manifest) - return None - - payload = cast(dict[str, Any], value) if payload.get("$schema") != AGENT_PLUGIN_SCHEMA: return None name = payload.get("name") @@ -128,7 +109,7 @@ def _load_manifest(plugin_root: Path) -> AgentPlugin | None: or len(name) > 64 or _PLUGIN_NAME.fullmatch(name) is None ): - logger.warning("Ignoring Agent Plugin manifest '{}': invalid name", manifest) + logger.warning("Ignoring Agent Plugin manifest in '{}': invalid name", plugin_root) return None extension = payload.get("extensions") extension_payload = cast(dict[str, object], extension) if isinstance(extension, dict) else {} @@ -165,27 +146,24 @@ def agent_plugin_mcp_servers( for name, server in plugin_servers.items(): host_name = plugin.name if len(plugin_servers) == 1 else f"{plugin.name}-{name}" servers[host_name] = server - for name, server in (configured or {}).items(): - if name in servers: - logger.warning("Configured MCP server '{}' overrides an Agent Plugin server", name) - servers[name] = server - return servers + configured = configured or {} + if collisions := servers.keys() & configured.keys(): + logger.warning("Configured MCP servers override Agent Plugins: {}", ", ".join(sorted(collisions))) + return servers | configured def discover_agent_plugin_states(workspace: Path) -> list[AgentPluginState]: """Return component and lifecycle state for discovered plugins.""" - states: list[AgentPluginState] = [] - for plugin in _discover_agent_plugins(workspace): - states.append( - AgentPluginState( - plugin=plugin, - mcp_servers=tuple(sorted(_plugin_mcp_servers(workspace, plugin))), - enabled=_enabled(workspace, plugin.name), - setup_required=bool(plugin.install_command) - and _setup_version(workspace, plugin.name) != (plugin.version or "unknown"), - ) + return [ + AgentPluginState( + plugin=plugin, + mcp_servers=tuple(sorted(_plugin_mcp_servers(workspace, plugin))), + enabled=_enabled(workspace, plugin.name), + setup_required=bool(plugin.install_command) + and _setup_version(workspace, plugin.name) != (plugin.version or "unknown"), ) - return states + for plugin in _discover_agent_plugins(workspace) + ] def set_agent_plugin_enabled(workspace: Path, name: str, enabled: bool) -> AgentPlugin: @@ -212,9 +190,7 @@ def _string(value: object) -> str: def _string_tuple(value: object) -> tuple[str, ...]: - if not isinstance(value, list): - return () - items = cast(list[object], value) + items = cast(list[object], value) if isinstance(value, list) else [] return tuple(item.strip() for item in items if isinstance(item, str) and item.strip()) @@ -230,16 +206,20 @@ def _plugin_logo(value: object, plugin_root: Path) -> Path | None: logger.warning("Ignoring invalid Agent Plugin logo in '{}'", plugin_root) return None logo = _contained_file(plugin_root / value[2:], plugin_root) - if logo is None or logo.suffix.lower() not in _LOGO_SUFFIXES: - logger.warning("Ignoring invalid Agent Plugin logo in '{}'", plugin_root) - return None try: - if logo.stat().st_size > _MAX_LOGO_BYTES: - logger.warning("Ignoring oversized Agent Plugin logo in '{}'", plugin_root) - return None + data = logo.read_bytes() if logo is not None else b"" + suffix = logo.suffix.lower() if logo is not None else "" + valid = ( + suffix == ".png" and data.startswith(b"\x89PNG\r\n\x1a\n") + or suffix in {".jpg", ".jpeg"} and data.startswith(b"\xff\xd8\xff") + or suffix == ".webp" and data.startswith(b"RIFF") and data[8:12] == b"WEBP" + ) + if valid and len(data) <= _MAX_LOGO_BYTES: + return logo except OSError: - return None - return logo + pass + logger.warning("Ignoring invalid Agent Plugin logo in '{}'", plugin_root) + return None def _install_command(value: object, plugin_root: Path) -> tuple[str, ...]: @@ -263,17 +243,9 @@ def _install_command(value: object, plugin_root: Path) -> tuple[str, ...]: def _plugin_mcp_servers(workspace: Path, plugin: AgentPlugin) -> dict[str, MCPServerConfig]: - path = _contained_file(plugin.root / "mcp.json", plugin.root) - if path is None: + payload = _read_object(plugin.root / "mcp.json", plugin.root) + if payload is None: return {} - try: - value = cast(object, json.loads(path.read_text(encoding="utf-8"))) - except (OSError, UnicodeError, json.JSONDecodeError) as exc: - logger.warning("Ignoring invalid MCP component for Agent Plugin '{}': {}", plugin.name, exc) - return {} - if not isinstance(value, dict): - return {} - payload = cast(dict[str, Any], value) raw_servers = payload.get("mcpServers") if payload.get("$schema") != AGENT_PLUGIN_MCP_SCHEMA or not isinstance(raw_servers, dict): logger.warning("Ignoring invalid MCP component for Agent Plugin '{}'", plugin.name) @@ -296,41 +268,30 @@ def _plugin_mcp_servers(workspace: Path, plugin: AgentPlugin) -> dict[str, MCPSe def _plugin_mcp_server(raw: object, root: Path, data: Path) -> MCPServerConfig | None: if not isinstance(raw, dict): return None - payload = cast(dict[str, Any], raw) - if payload.get("type") != "stdio" or payload.keys() - _MCP_SERVER_FIELDS: + payload = cast(dict[str, object], raw) + if payload.keys() - _MCP_SERVER_FIELDS: return None - command = _stdio_command(payload.get("command"), root) - args = payload.get("args", []) - env = payload.get("env", {}) + try: + server = MCPServerConfig.model_validate(payload) + except ValidationError: + return None + command = _stdio_command(server.command, root) cwd = _stdio_cwd(payload.get("cwd"), root, data) - if ( - command is None - or not isinstance(args, list) - or not all(isinstance(item, str) for item in cast(list[object], args)) - or not isinstance(env, dict) - or cwd is None - ): + if server.type != "stdio" or command is None or cwd is None: return None - env_payload = cast(dict[object, object], env) - if any( - not isinstance(key, str) - or key in {"PLUGIN_ROOT", "PLUGIN_DATA"} - or not isinstance(value, str) - for key, value in env_payload.items() - ): + if {"PLUGIN_ROOT", "PLUGIN_DATA"} & server.env.keys(): return None - string_env = cast(dict[str, str], env) - replacements = {"${PLUGIN_ROOT}": str(root), "${PLUGIN_DATA}": str(data)} - return MCPServerConfig( - type="stdio", - command=command, - args=[_expand(item, replacements) for item in cast(list[str], args)], - env={ - **{key: _expand(value, replacements) for key, value in string_env.items()}, - "PLUGIN_ROOT": str(root), - "PLUGIN_DATA": str(data), - }, - cwd=str(cwd), + return server.model_copy( + update={ + "command": command, + "args": [_expand(item, root, data) for item in server.args], + "env": { + **{key: _expand(value, root, data) for key, value in server.env.items()}, + "PLUGIN_ROOT": str(root), + "PLUGIN_DATA": str(data), + }, + "cwd": str(cwd), + } ) @@ -365,10 +326,8 @@ def _stdio_cwd(value: object, root: Path, data: Path) -> Path | None: return None -def _expand(value: str, replacements: dict[str, str]) -> str: - for token, replacement in replacements.items(): - value = value.replace(token, replacement) - return value +def _expand(value: str, root: Path, data: Path) -> str: + return value.replace("${PLUGIN_ROOT}", str(root)).replace("${PLUGIN_DATA}", str(data)) def _plugin_data_dir(workspace: Path, name: str, *, create: bool) -> Path: @@ -408,12 +367,7 @@ def _setup_version(workspace: Path, name: str) -> str: def _write_state(path: Path, value: str) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - path.parent.chmod(0o700) - temporary = path.with_name(f".{path.name}.tmp") - temporary.write_text(value, encoding="utf-8") - temporary.chmod(0o600) - temporary.replace(path) + path.write_text(value, encoding="utf-8") path.chmod(0o600) @@ -441,12 +395,8 @@ def _run_install(plugin: AgentPlugin, data: Path) -> None: def _discover_plugin_skills(plugin_name: str, plugin_root: Path) -> list[AgentPluginSkill]: - skills_root = plugin_root / "skills" - if not skills_root.exists(): - return [] - resolved_skills_root = _contained_directory(skills_root, plugin_root) - if resolved_skills_root is None: - logger.warning("Ignoring invalid skills component in Agent Plugin '{}'", plugin_name) + skills_root = _contained_directory(plugin_root / "skills", plugin_root) + if skills_root is None: return [] try: @@ -457,7 +407,7 @@ def _discover_plugin_skills(plugin_name: str, plugin_root: Path) -> list[AgentPl skills: list[AgentPluginSkill] = [] for candidate in candidates: - skill_root = _contained_directory(candidate, resolved_skills_root) + skill_root = _contained_directory(candidate, skills_root) if skill_root is None: continue skill_file = _contained_file(skill_root / "SKILL.md", plugin_root) @@ -474,31 +424,14 @@ def _valid_skill(path: Path, directory_name: str, plugin_name: str) -> bool: content = path.read_text(encoding="utf-8") except (OSError, UnicodeError): return False - match = _SKILL_FRONTMATTER.match(content) - if match is None: + metadata = parse_skill_metadata(content) + if metadata is None: logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid frontmatter", plugin_name, directory_name) return False - try: - metadata = cast(object, yaml.safe_load(match.group(1))) - except yaml.YAMLError: - metadata = None - if not isinstance(metadata, dict): - logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid frontmatter", plugin_name, directory_name) - return False - payload = cast(dict[object, object], metadata) - name = payload.get("name") - description = payload.get("description") - valid = ( - name == directory_name - and isinstance(name, str) - and len(name) <= 64 - and _SKILL_NAME.fullmatch(name) is not None - and isinstance(description, str) - and 1 <= len(description.strip()) <= 1024 - ) - if not valid: + if not valid_skill_metadata(metadata, directory_name): logger.warning("Ignoring Agent Plugin '{}' skill '{}': invalid metadata", plugin_name, directory_name) - return valid + return False + return True def _contained_directory(path: Path, root: Path) -> Path | None: @@ -509,6 +442,18 @@ def _contained_directory(path: Path, root: Path) -> Path | None: return resolved if resolved.is_dir() and resolved.is_relative_to(root) else None +def _read_object(path: Path, root: Path) -> dict[str, object] | None: + contained = _contained_file(path, root) + if contained is None: + return None + try: + value = cast(object, json.loads(contained.read_text(encoding="utf-8"))) + except (OSError, UnicodeError, json.JSONDecodeError) as exc: + logger.warning("Ignoring invalid Agent Plugin component '{}': {}", contained, exc) + return None + return cast(dict[str, object], value) if isinstance(value, dict) else None + + def _contained_file(path: Path, root: Path) -> Path | None: try: resolved = path.resolve(strict=True) diff --git a/nanobot/agent/skills.py b/nanobot/agent/skills.py index 31ec29123..bf38e884e 100644 --- a/nanobot/agent/skills.py +++ b/nanobot/agent/skills.py @@ -17,9 +17,48 @@ _STRIP_SKILL_FRONTMATTER = re.compile( r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", re.DOTALL, ) +_SKILL_NAME = re.compile(r"^(?!.*--)[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$") +_SKILL_NAME_LINE = re.compile(r"^name\s*:.*$", re.MULTILINE) _SKILL_REFERENCE = re.compile(r"(? dict[str, object] | None: + """Parse a skill document's YAML frontmatter.""" + if not (match := _STRIP_SKILL_FRONTMATTER.match(content)): + return None + try: + parsed = yaml.safe_load(match.group(1)) + except yaml.YAMLError: + return None + if not isinstance(parsed, dict): + return None + return {str(key): value for key, value in cast(dict[object, object], parsed).items()} + + +def valid_skill_metadata(metadata: dict[str, object], name: str) -> bool: + """Return whether metadata satisfies the Agent Skills identity contract.""" + description = metadata.get("description") + return ( + metadata.get("name") == name + and len(name) <= 64 + and _SKILL_NAME.fullmatch(name) is not None + and isinstance(description, str) + and 1 <= len(description.strip()) <= 1024 + ) + + +def normalize_skill_document(content: str, name: str) -> str | None: + """Return a valid skill document with a canonical name.""" + match = _STRIP_SKILL_FRONTMATTER.match(content) + metadata = parse_skill_metadata(content) + if match is None or metadata is None or not valid_skill_metadata(metadata | {"name": name}, name): + return None + frontmatter, replaced = _SKILL_NAME_LINE.subn(f"name: {name}", match.group(1), count=1) + if not replaced: + frontmatter = f"name: {name}\n{frontmatter}" + return f"---\n{frontmatter.strip()}\n---\n\n{content[match.end():].lstrip()}" + + class SkillsLoader: """ Loader for agent skills. @@ -291,21 +330,4 @@ class SkillsLoader: Returns: Metadata dict or None. """ - content = self.load_skill(name) - if not content or not content.startswith("---"): - return None - match = _STRIP_SKILL_FRONTMATTER.match(content) - if not match: - return None - try: - parsed = yaml.safe_load(match.group(1)) - except yaml.YAMLError: - return None - if not isinstance(parsed, dict): - return None - # yaml.safe_load returns native types (int, bool, list, etc.); - # keep values as-is so downstream consumers get correct types. - metadata: dict[str, object] = {} - for key, value in cast(dict[object, object], parsed).items(): - metadata[str(key)] = value - return metadata + return parse_skill_metadata(self.load_skill(name) or "") diff --git a/nanobot/apps/cli/service.py b/nanobot/apps/cli/service.py index 73c8bbb08..6a594bcc8 100644 --- a/nanobot/apps/cli/service.py +++ b/nanobot/apps/cli/service.py @@ -18,9 +18,9 @@ from typing import Any, cast from urllib.parse import urlparse import httpx -import yaml from loguru import logger +from nanobot.agent.skills import normalize_skill_document from nanobot.apps.protocol import app_manifest, compact_dict from nanobot.config.paths import get_runtime_subdir from nanobot.security.workspace_policy import is_path_within @@ -43,8 +43,6 @@ _MAX_ARTIFACT_REPORT = 12 _SAFE_NAME_RE = re.compile(r"[^a-z0-9_-]+") _SAFE_NPM_DIR_RE = re.compile(r"^[a-z0-9._-]+$", re.IGNORECASE) _MENTION_RE = re.compile(r"(^|[\s([{])@([a-z0-9_-]+)\b", re.IGNORECASE) -_SKILL_FRONTMATTER_RE = re.compile(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", re.DOTALL) -_SKILL_NAME_LINE_RE = re.compile(r"^name\s*:.*$", re.MULTILINE) _SHELL_META_CHARS = ("|", "&&", "||", ";", "$(", "`", ">", "<") _ENDORSEMENT_WORD_RE = re.compile(r"\bofficial\s+", re.IGNORECASE) _ARTIFACT_EXTENSIONS = frozenset({ @@ -232,11 +230,11 @@ def _plugin_skill_relative_path(name: str) -> str: def cli_app_skill_relative_path(workspace: Path, name: str) -> str: """Return a CLI App's skill path, including the legacy location.""" - canonical = Path(_plugin_skill_relative_path(name)) - legacy = Path("skills") / _legacy_skill_name(name) / "SKILL.md" + canonical = _plugin_skill_relative_path(name) + legacy = f"skills/{_legacy_skill_name(name)}/SKILL.md" if not (workspace / canonical).is_file() and (workspace / legacy).is_file(): - return legacy.as_posix() - return canonical.as_posix() + return legacy + return canonical def _has_shell_meta(command: str) -> bool: @@ -1093,31 +1091,12 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in return "".join(lines[: index + 1]) + "\n" + note + "\n" + "".join(lines[index + 1 :]) return note + "\n" + content - def _normalise_skill(self, content: str, app: dict[str, Any]) -> str: - """Give a catalog skill the identity required by its plugin directory.""" - match = _SKILL_FRONTMATTER_RE.match(content) - if match is None: - return self._fallback_skill(app) - try: - metadata = _as_object_dict(cast(object, yaml.safe_load(match.group(1)))) - except yaml.YAMLError: - return self._fallback_skill(app) - description = metadata.get("description") if metadata is not None else None - if not isinstance(description, str) or not 1 <= len(description.strip()) <= 1024: - return self._fallback_skill(app) - - name = _safe_skill_name(str(app["name"])) - frontmatter, replaced = _SKILL_NAME_LINE_RE.subn(f"name: {name}", match.group(1), count=1) - if not replaced: - frontmatter = f"name: {name}\n{frontmatter}" - body = content[match.end():].lstrip() - return f"---\n{frontmatter.strip()}\n---\n\n{body}" - def install_skill(self, app: dict[str, Any]) -> Path: - path = self.workspace / _plugin_skill_relative_path(str(app["name"])) + name = str(app["name"]) + path = self.workspace / _plugin_skill_relative_path(name) path.parent.mkdir(parents=True, exist_ok=True) content = self._fetch_skill_content(app) or self._fallback_skill(app) - content = self._normalise_skill(content, app) + content = normalize_skill_document(content, _safe_skill_name(name)) or self._fallback_skill(app) content = self._with_nanobot_skill_note(content, app) path.write_text(content, encoding="utf-8") plugin_root = path.parents[2] diff --git a/nanobot/webui/mcp_presets_api.py b/nanobot/webui/mcp_presets_api.py index c96c46120..702de62ac 100644 --- a/nanobot/webui/mcp_presets_api.py +++ b/nanobot/webui/mcp_presets_api.py @@ -57,13 +57,6 @@ _MAX_TEST_TOOLS = 16 _DEFAULT_TEST_TIMEOUT = 20 _DEFAULT_CUSTOM_TIMEOUT = 30 _CUSTOM_ACTIONS = {"custom", "import", "import-cursor", "tools"} -_PLUGIN_LOGO_MIME_TYPES = { - ".jpeg": "image/jpeg", - ".jpg": "image/jpeg", - ".png": "image/png", - ".webp": "image/webp", -} - McpReload = Callable[[], Awaitable[dict[str, Any]]] @@ -859,16 +852,9 @@ def _plugin_logo_data_url(path: Path | None) -> str | None: data = path.read_bytes() except OSError: return None - suffix = path.suffix.lower() - valid = ( - suffix == ".png" and data.startswith(b"\x89PNG\r\n\x1a\n") - or suffix in {".jpg", ".jpeg"} and data.startswith(b"\xff\xd8\xff") - or suffix == ".webp" and data.startswith(b"RIFF") and data[8:12] == b"WEBP" - ) - if not valid: - return None encoded = base64.b64encode(data).decode("ascii") - return f"data:{_PLUGIN_LOGO_MIME_TYPES[suffix]};base64,{encoded}" + image_format = path.suffix.lower().lstrip(".").replace("jpg", "jpeg") + return f"data:image/{image_format};base64,{encoded}" def _agent_plugin_payload(state: AgentPluginState) -> dict[str, Any]: diff --git a/tests/agent/test_agent_plugins.py b/tests/agent/test_agent_plugins.py index 51370bf00..816b62f08 100644 --- a/tests/agent/test_agent_plugins.py +++ b/tests/agent/test_agent_plugins.py @@ -40,6 +40,10 @@ def _write_skill(root: Path, name: str, *, description: str = "Plugin skill.") - return skill +def _manifest(name: str, **fields: object) -> dict[str, object]: + return {"$schema": AGENT_PLUGIN_SCHEMA, "name": name, **fields} + + def _write_plugin( workspace: Path, directory: str, @@ -49,24 +53,25 @@ def _write_plugin( ) -> Path: root = workspace / "plugins" / directory root.mkdir(parents=True) - payload = manifest or { - "$schema": AGENT_PLUGIN_SCHEMA, - "name": name or directory, - } + payload = manifest or _manifest(name or directory) (root / "plugin.json").write_text(json.dumps(payload), encoding="utf-8") return root +def _write_mcp(root: Path, servers: dict[str, object], **fields: object) -> None: + payload = {"$schema": AGENT_PLUGIN_MCP_SCHEMA, "mcpServers": servers, **fields} + (root / "mcp.json").write_text(json.dumps(payload), encoding="utf-8") + + def _write_setup_plugin(workspace: Path) -> tuple[Path, Path]: plugin = _write_plugin( workspace, "desktop", - manifest={ - "$schema": AGENT_PLUGIN_SCHEMA, - "name": "desktop", - "version": "1.2.3", - "extensions": {"dev.nanobot": {"installCommand": ["./bin/install"]}}, - }, + manifest=_manifest( + "desktop", + version="1.2.3", + extensions={"dev.nanobot": {"installCommand": ["./bin/install"]}}, + ), ) executable = plugin / "bin" / "install" executable.parent.mkdir() @@ -150,14 +155,13 @@ def test_unknown_manifest_fields_and_non_object_extensions_are_ignored(tmp_path: plugin = _write_plugin( tmp_path, "demo", - manifest={ - "$schema": AGENT_PLUGIN_SCHEMA, - "name": "demo", - "futureField": True, - "author": None, - "keywords": None, - "extensions": "invalid but non-fatal", - }, + manifest=_manifest( + "demo", + futureField=True, + author=None, + keywords=None, + extensions="invalid but non-fatal", + ), ) _write_skill(plugin, "example") set_agent_plugin_enabled(tmp_path, "demo", True) @@ -169,11 +173,10 @@ def test_agent_plugin_discovers_contained_raster_logo(tmp_path: Path) -> None: plugin = _write_plugin( tmp_path, "demo", - manifest={ - "$schema": AGENT_PLUGIN_SCHEMA, - "name": "demo", - "extensions": {"dev.nanobot": {"logo": "./assets/icon.png"}}, - }, + manifest=_manifest( + "demo", + extensions={"dev.nanobot": {"logo": "./assets/icon.png"}}, + ), ) assets = plugin / "assets" assets.mkdir() @@ -187,11 +190,10 @@ def test_agent_plugin_logo_cannot_escape_package(tmp_path: Path) -> None: plugin = _write_plugin( tmp_path, "demo", - manifest={ - "$schema": AGENT_PLUGIN_SCHEMA, - "name": "demo", - "extensions": {"dev.nanobot": {"logo": "./assets/icon.png"}}, - }, + manifest=_manifest( + "demo", + extensions={"dev.nanobot": {"logo": "./assets/icon.png"}}, + ), ) assets = plugin / "assets" assets.mkdir() @@ -295,22 +297,17 @@ def test_plugin_mcp_requires_explicit_enable(tmp_path: Path) -> None: executable = plugin / "bin" / "server" executable.parent.mkdir() executable.write_text("#!/bin/sh\n", encoding="utf-8") - (plugin / "mcp.json").write_text( - json.dumps( - { - "$schema": AGENT_PLUGIN_MCP_SCHEMA, - "futureField": True, - "mcpServers": { - "desktop": { - "type": "stdio", - "command": "./bin/server", - "args": ["--data", "${PLUGIN_DATA}/state"], - "cwd": "${PLUGIN_ROOT}", - } - }, + _write_mcp( + plugin, + { + "desktop": { + "type": "stdio", + "command": "./bin/server", + "args": ["--data", "${PLUGIN_DATA}/state"], + "cwd": "${PLUGIN_ROOT}", } - ), - encoding="utf-8", + }, + futureField=True, ) assert agent_plugin_mcp_servers(tmp_path) == {} @@ -385,18 +382,13 @@ def test_invalid_plugin_mcp_entries_do_not_block_valid_servers(tmp_path: Path) - executable = plugin / "bin" / "server" executable.parent.mkdir() executable.write_text("#!/bin/sh\n", encoding="utf-8") - (plugin / "mcp.json").write_text( - json.dumps( - { - "$schema": AGENT_PLUGIN_MCP_SCHEMA, - "mcpServers": { - "public-http": {"type": "streamable-http", "url": "http://example.com/mcp"}, - "local": {"type": "stdio", "command": "./bin/server"}, - "escape": {"type": "stdio", "command": "../outside"}, - }, - } - ), - encoding="utf-8", + _write_mcp( + plugin, + { + "public-http": {"type": "streamable-http", "url": "http://example.com/mcp"}, + "local": {"type": "stdio", "command": "./bin/server"}, + "escape": {"type": "stdio", "command": "../outside"}, + }, ) set_agent_plugin_enabled(tmp_path, "network", True) diff --git a/tests/cli_apps/test_service.py b/tests/cli_apps/test_service.py index f61f547a7..00c18e13e 100644 --- a/tests/cli_apps/test_service.py +++ b/tests/cli_apps/test_service.py @@ -515,14 +515,7 @@ def test_install_records_available_cli_without_reinstalling( assert "entry_point_available" in payload["last_action"]["verification"] installed = json.loads(manager.installed_path.read_text(encoding="utf-8"))["apps"] assert installed["feishu"]["entry_point_path"] == str(resolved) - skill = ( - manager.workspace - / "plugins" - / "cli-app-feishu" - / "skills" - / "cli-app-feishu" - / "SKILL.md" - ) + skill = manager.workspace / "plugins/cli-app-feishu/skills/cli-app-feishu/SKILL.md" assert skill.is_file() assert 'run_cli_app` tool with `name="feishu"' in skill.read_text(encoding="utf-8") diff --git a/tests/webui/test_mcp_presets_api.py b/tests/webui/test_mcp_presets_api.py index 682f57ae7..5a817082f 100644 --- a/tests/webui/test_mcp_presets_api.py +++ b/tests/webui/test_mcp_presets_api.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import json import subprocess +from functools import partial from pathlib import Path import pytest @@ -132,36 +133,22 @@ def test_agent_plugin_reuses_mcp_catalog_and_runtime_action( async def reload() -> dict[str, object]: return {"ok": True, "message": "MCP reloaded.", "requires_restart": False} + plugin_action = partial( + mcp_presets_settings_action, + query={"name": ["plugin-desktop"]}, + ) with pytest.raises(McpPresetError, match="restricted") as restricted: - asyncio.run( - mcp_presets_settings_action( - "enable", - {"name": ["plugin-desktop"]}, - remote=True, - ) - ) + asyncio.run(plugin_action("enable", remote=True)) assert restricted.value.status == 403 - enabled = asyncio.run( - mcp_presets_settings_action( - "enable", - {"name": ["plugin-desktop"]}, - reload_mcp=reload, - ) - ) + enabled = asyncio.run(plugin_action("enable", reload_mcp=reload)) enabled_row = next(item for item in enabled["presets"] if item["name"] == "plugin-desktop") assert enabled_row["configured"] is True assert enabled_row["enabled"] is True assert enabled_row["status"] == "enabled" assert enabled["requires_restart"] is False - disabled = asyncio.run( - mcp_presets_settings_action( - "disable", - {"name": ["plugin-desktop"]}, - reload_mcp=reload, - ) - ) + disabled = asyncio.run(plugin_action("disable", reload_mcp=reload)) disabled_row = next(item for item in disabled["presets"] if item["name"] == "plugin-desktop") assert disabled_row["installed"] is True assert disabled_row["configured"] is True @@ -169,12 +156,7 @@ def test_agent_plugin_reuses_mcp_catalog_and_runtime_action( assert disabled_row["status"] == "disabled" with pytest.raises(McpPresetError, match="enable and disable"): - asyncio.run( - mcp_presets_settings_action( - "remove", - {"name": ["plugin-desktop"]}, - ) - ) + asyncio.run(plugin_action("remove")) def test_explicit_mcp_config_wins_over_plugin_catalog_name(