From a2979c3a4ba3abd75c829562d5f6504a95909448 Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Tue, 11 Aug 2026 17:05:07 +0900 Subject: [PATCH] fix(plugins): harden activation boundaries --- docs/configuration.md | 2 + nanobot/agent/plugins.py | 51 ++++++++++- nanobot/agent/tools/filesystem.py | 13 ++- tests/agent/test_agent_plugins.py | 139 ++++++++++++++++++++++++++++++ 4 files changed, 201 insertions(+), 4 deletions(-) diff --git a/docs/configuration.md b/docs/configuration.md index 48ec4b8f2..6a8e010b2 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -2354,6 +2354,8 @@ nanobot discovers [Agent Plugins](https://agent-plugins.org/) under ` Directory presence means installed; activation is explicit in **Apps**. Skills use progressive loading and `$skill-name` invocation, with workspace > plugin > built-in precedence. Enabled `stdio` servers receive contained `PLUGIN_ROOT` and isolated `PLUGIN_DATA` paths; explicit `tools.mcpServers` entries win collisions. Invalid or escaping components are ignored. +An enabled package is treated as immutable: changing any packaged file disables it until the user +reviews and enables it again. Runtime state belongs under `PLUGIN_DATA`, not the package root. Enabled plugins run as the nanobot user; permissions are descriptive, not an OS sandbox. The optional `extensions.dev.nanobot.logo` accepts a contained PNG, JPEG, or WebP up to 256 KiB. diff --git a/nanobot/agent/plugins.py b/nanobot/agent/plugins.py index 12a1225eb..b9615f875 100644 --- a/nanobot/agent/plugins.py +++ b/nanobot/agent/plugins.py @@ -118,7 +118,9 @@ def agent_plugin_mcp_servers( continue plugin_servers = _plugin_mcp_servers(workspace, plugin) for name, server in plugin_servers.items(): - host_name = plugin.name if len(plugin_servers) == 1 else f"{plugin.name}-{name}" + # ``--`` cannot occur in a valid plugin identity, so multi-server + # namespaces cannot collide with a single-server plugin name. + host_name = plugin.name if len(plugin_servers) == 1 else f"{plugin.name}--{name}" servers[host_name] = server configured = configured or {} if collisions := servers.keys() & configured.keys(): @@ -146,7 +148,10 @@ def set_agent_plugin_enabled(workspace: Path, name: str, enabled: bool) -> None: data = _plugin_data_dir(workspace, plugin.name, create=True) marker = data / "enabled" if enabled: - marker.write_text(str(plugin.root), encoding="utf-8") + activation = _activation_marker(plugin) + if activation is None: + raise RuntimeError(f"Agent Plugin '{name}' changed while it was being enabled") + marker.write_text(activation, encoding="utf-8") marker.chmod(0o600) else: marker.unlink(missing_ok=True) @@ -238,6 +243,7 @@ def _plugin_mcp_server(raw: object, root: Path, data: Path) -> MCPServerConfig | "args": [_expand(item, root, data) for item in server.args], "env": { **{key: _expand(value, root, data) for key, value in server.env.items()}, + "PYTHONDONTWRITEBYTECODE": "1", "PLUGIN_ROOT": str(root), "PLUGIN_DATA": str(data), }, @@ -303,11 +309,50 @@ def _plugin_data_dir(workspace: Path, name: str, *, create: bool) -> Path: def _enabled(workspace: Path, plugin: AgentPlugin) -> bool: marker = _plugin_data_dir(workspace, plugin.name, create=False) / "enabled" try: - return marker.is_file() and marker.read_text(encoding="utf-8") == str(plugin.root) + if not marker.is_file(): + return False + current = marker.read_text(encoding="utf-8") + activation = _activation_marker(plugin) + if activation is None: + return False + if current == activation: + return True + if current == str(plugin.root): + marker.write_text(activation, encoding="utf-8") + marker.chmod(0o600) + return True + return False except OSError: return False +def _activation_marker(plugin: AgentPlugin) -> str | None: + """Bind activation to one immutable package snapshot.""" + digest = sha256() + try: + for candidate in sorted(plugin.root.rglob("*")): + relative = candidate.relative_to(plugin.root).as_posix() + digest.update(relative.encode()) + if candidate.is_symlink(): + digest.update(b"\0link\0") + digest.update(candidate.readlink().as_posix().encode()) + elif candidate.is_file(): + digest.update(b"\0file\0") + digest.update(candidate.read_bytes()) + elif candidate.is_dir(): + digest.update(b"\0dir\0") + else: + return None + digest.update(b"\0") + except OSError: + return None + return json.dumps( + {"fingerprint": digest.hexdigest(), "root": str(plugin.root)}, + separators=(",", ":"), + sort_keys=True, + ) + + def _discover_plugin_skills(plugin_name: str, plugin_root: Path) -> list[tuple[str, Path]]: skills_root = _contained(plugin_root / "skills", plugin_root, directory=True) if skills_root is None: diff --git a/nanobot/agent/tools/filesystem.py b/nanobot/agent/tools/filesystem.py index ea84fedb1..397f4e420 100644 --- a/nanobot/agent/tools/filesystem.py +++ b/nanobot/agent/tools/filesystem.py @@ -148,9 +148,20 @@ class _FsTool(Tool): ) def _resolve_read(self, path: str) -> Path: + plugin_skill_dirs: list[Path] = [] + if self._workspace is not None: + from nanobot.agent.plugins import enabled_agent_plugin_skills + + try: + plugin_skill_dirs = [ + skill.parent + for _name, skill in enabled_agent_plugin_skills(Path(self._workspace)) + ] + except (OSError, RuntimeError): + pass return self._resolve_with_extra( path, - self._extra_read_allowed_dirs, + [*self._extra_read_allowed_dirs, *plugin_skill_dirs], self._extra_read_allowed_files, include_media_dir=True, extra_files_require_allowed_root=True, diff --git a/tests/agent/test_agent_plugins.py b/tests/agent/test_agent_plugins.py index 5daafb823..740bddd82 100644 --- a/tests/agent/test_agent_plugins.py +++ b/tests/agent/test_agent_plugins.py @@ -14,6 +14,14 @@ from nanobot.agent.plugins import ( set_agent_plugin_enabled, ) from nanobot.agent.skills import SkillsLoader +from nanobot.agent.tools.context import ToolContext +from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool +from nanobot.config.schema import ToolsConfig +from nanobot.security.workspace_access import ( + bind_workspace_scope, + reset_workspace_scope, + validate_workspace_scope_payload, +) @pytest.fixture(autouse=True) @@ -169,6 +177,72 @@ def test_plugin_mcp_requires_explicit_enable(tmp_path: Path) -> None: assert agent_plugin_mcp_servers(tmp_path) == {} +def test_plugin_mcp_namespaces_cannot_shadow_plugin_identities(tmp_path: Path) -> None: + single = _plugin(tmp_path, "foo-bar") + multi = _plugin(tmp_path, "foo") + for root, servers in ( + (single, {"main": {"type": "stdio", "command": "echo", "args": ["single"]}}), + ( + multi, + { + "bar": {"type": "stdio", "command": "echo", "args": ["multi"]}, + "other": {"type": "stdio", "command": "echo"}, + }, + ), + ): + _write_json( + root / "mcp.json", + {"$schema": AGENT_PLUGIN_MCP_SCHEMA, "mcpServers": servers}, + ) + set_agent_plugin_enabled(tmp_path, "foo-bar", True) + set_agent_plugin_enabled(tmp_path, "foo", True) + + servers = agent_plugin_mcp_servers(tmp_path) + + assert set(servers) == {"foo-bar", "foo--bar", "foo--other"} + assert servers["foo-bar"].args == ["single"] + assert servers["foo--bar"].args == ["multi"] + + +@pytest.mark.asyncio +async def test_restricted_project_can_read_only_enabled_plugin_skill( + tmp_path: Path, +) -> None: + agent_workspace = tmp_path / "agent" + project = tmp_path / "project" + project.mkdir() + plugin = _plugin(agent_workspace) + skill = _skill(plugin / "skills", "demo-skill") + resource = skill / "reference.md" + resource.write_text("plugin reference", encoding="utf-8") + ctx = ToolContext( + config=ToolsConfig(restrict_to_workspace=True), + workspace=str(agent_workspace), + ) + read_tool = ReadFileTool.create(ctx) + write_tool = WriteFileTool.create(ctx) + set_agent_plugin_enabled(agent_workspace, "demo", True) + scope = validate_workspace_scope_payload( + {"project_path": str(project), "access_mode": "restricted"}, + default_workspace=agent_workspace, + default_restrict_to_workspace=True, + ) + + token = bind_workspace_scope(scope) + try: + read_result = await read_tool.execute(path=str(resource)) + write_result = await write_tool.execute(path=str(resource), content="changed") + set_agent_plugin_enabled(agent_workspace, "demo", False) + disabled_result = await read_tool.execute(path=str(resource)) + finally: + reset_workspace_scope(token) + + assert "plugin reference" in read_result + assert "outside allowed directory" in write_result + assert "outside allowed directory" in disabled_result + assert resource.read_text(encoding="utf-8") == "plugin reference" + + def test_plugin_state_symlink_cannot_escape_config_root( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -213,3 +287,68 @@ def test_plugin_activation_requires_one_stable_package_identity(tmp_path: Path) roots[0].rename(moved) assert discover_agent_plugins(tmp_path)[0].enabled is False assert agent_plugin_mcp_servers(tmp_path) == {} + + +def test_legacy_path_activation_is_upgraded_to_package_fingerprint(tmp_path: Path) -> None: + plugin = _plugin(tmp_path) + set_agent_plugin_enabled(tmp_path, "demo", True) + marker = next((tmp_path / "config" / "plugin-data").glob("*/demo/enabled")) + marker.write_text(str(plugin), encoding="utf-8") + + assert discover_agent_plugins(tmp_path)[0].enabled is True + assert marker.read_text(encoding="utf-8").startswith('{"fingerprint":') + + +def test_plugin_activation_does_not_survive_in_place_contract_replacement( + tmp_path: Path, +) -> None: + plugin = _plugin(tmp_path, "desktop") + mcp = plugin / "mcp.json" + + def write_server(marker: str) -> None: + _write_json( + mcp, + { + "$schema": AGENT_PLUGIN_MCP_SCHEMA, + "mcpServers": { + "server": {"type": "stdio", "command": "echo", "args": [marker]} + }, + }, + ) + + write_server("trusted") + set_agent_plugin_enabled(tmp_path, "desktop", True) + assert agent_plugin_mcp_servers(tmp_path)["desktop"].args == ["trusted"] + + write_server("replacement") + + assert discover_agent_plugins(tmp_path)[0].enabled is False + assert agent_plugin_mcp_servers(tmp_path) == {} + + +def test_plugin_activation_does_not_survive_in_place_code_replacement( + tmp_path: Path, +) -> None: + plugin = _plugin(tmp_path, "desktop") + executable = plugin / "server.py" + executable.write_text("print('trusted')\n", encoding="utf-8") + _write_json( + plugin / "mcp.json", + { + "$schema": AGENT_PLUGIN_MCP_SCHEMA, + "mcpServers": { + "server": { + "type": "stdio", + "command": "python", + "args": ["${PLUGIN_ROOT}/server.py"], + } + }, + }, + ) + set_agent_plugin_enabled(tmp_path, "desktop", True) + assert discover_agent_plugins(tmp_path)[0].enabled is True + + executable.write_text("print('replacement')\n", encoding="utf-8") + + assert discover_agent_plugins(tmp_path)[0].enabled is False + assert agent_plugin_mcp_servers(tmp_path) == {}