From e99b1a11aad01bb20940e3470f1884fa0669b8ac Mon Sep 17 00:00:00 2001 From: yu-xin-c <2182712990@qq.com> Date: Thu, 13 Aug 2026 16:06:57 +0800 Subject: [PATCH] fix(plugins): revalidate cached skill roots --- nanobot/agent/plugins.py | 96 +++++++++++++++++++++++++++---- tests/agent/test_agent_plugins.py | 37 ++++++++++++ 2 files changed, 122 insertions(+), 11 deletions(-) diff --git a/nanobot/agent/plugins.py b/nanobot/agent/plugins.py index 37ae3976c..450b166e6 100644 --- a/nanobot/agent/plugins.py +++ b/nanobot/agent/plugins.py @@ -23,7 +23,33 @@ AGENT_PLUGIN_MCP_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.js _PLUGIN_NAME = re.compile(r"^(?!.*(?:--|\.\.))[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?$") _MCP_SERVER_FIELDS = {"type", "command", "args", "env", "cwd"} _MAX_LOGO_BYTES = 256 * 1024 -_SKILL_CACHE: dict[tuple[Path, Path], tuple[tuple[str, Path], ...]] = {} + + +@dataclass(frozen=True, slots=True) +class _PackageEntry: + path: str + mode: int + size: int + mtime_ns: int + ctime_ns: int + device: int + inode: int + link_target: str | None + + +@dataclass(frozen=True, slots=True) +class _PackageSnapshot: + root: Path + entries: tuple[_PackageEntry, ...] + + +@dataclass(frozen=True, slots=True) +class _SkillCacheEntry: + skills: tuple[tuple[str, Path], ...] + packages: tuple[_PackageSnapshot, ...] + + +_SKILL_CACHE: dict[tuple[Path, Path], _SkillCacheEntry] = {} @dataclass(frozen=True) @@ -66,22 +92,42 @@ def _installed_plugins(workspace: Path) -> list[AgentPlugin]: def enabled_agent_plugin_skills(workspace: Path) -> list[tuple[str, Path]]: """Verify and return skills from plugins the user has explicitly enabled.""" - skills = [ - skill - for plugin in _installed_plugins(workspace) - if _enabled(workspace, plugin) - for skill in _discover_plugin_skills(plugin.name, plugin.root) - ] - _SKILL_CACHE[_skill_cache_key(workspace)] = tuple(skills) + skills: list[tuple[str, Path]] = [] + packages: list[_PackageSnapshot] = [] + cacheable = True + for plugin in _installed_plugins(workspace): + before = _package_snapshot(plugin.root) + if before is None: + cacheable = False + continue + if not _enabled(workspace, plugin): + continue + plugin_skills = _discover_plugin_skills(plugin.name, plugin.root) + after = _package_snapshot(plugin.root) + if after is None or after != before: + cacheable = False + continue + skills.extend(plugin_skills) + packages.append(after) + + key = _skill_cache_key(workspace) + if cacheable: + _SKILL_CACHE[key] = _SkillCacheEntry(tuple(skills), tuple(packages)) + else: + _SKILL_CACHE.pop(key, None) return skills def enabled_agent_plugin_skill_dirs(workspace: Path) -> tuple[Path, ...]: - """Return the last verified skill roots, verifying once on a cache miss.""" + """Return verified skill roots, revalidating packages when they change.""" key = _skill_cache_key(workspace) - skills = _SKILL_CACHE.get(key) - if skills is None: + cached = _SKILL_CACHE.get(key) + if cached is None or any( + _package_snapshot(package.root) != package for package in cached.packages + ): skills = tuple(enabled_agent_plugin_skills(workspace)) + else: + skills = cached.skills return tuple(path.parent for _name, path in skills) @@ -96,6 +142,34 @@ def _invalidate_skill_cache(workspace: Path) -> None: _SKILL_CACHE.pop(_skill_cache_key(workspace), None) +def _package_snapshot(root: Path) -> _PackageSnapshot | None: + """Capture cheap package metadata used to guard cached authorization.""" + try: + candidates = [root, *sorted(root.rglob("*"))] + entries: list[_PackageEntry] = [] + for candidate in candidates: + stat = candidate.lstat() + entries.append( + _PackageEntry( + path="." if candidate == root else candidate.relative_to(root).as_posix(), + mode=stat.st_mode, + size=stat.st_size, + mtime_ns=stat.st_mtime_ns, + ctime_ns=stat.st_ctime_ns, + device=stat.st_dev, + inode=stat.st_ino, + link_target=( + candidate.readlink().as_posix() + if candidate.is_symlink() + else None + ), + ) + ) + except OSError: + return None + return _PackageSnapshot(root=root, entries=tuple(entries)) + + def _load_manifest(plugin_root: Path) -> AgentPlugin | None: payload = _read_object(plugin_root / "plugin.json", plugin_root) if payload is None: diff --git a/tests/agent/test_agent_plugins.py b/tests/agent/test_agent_plugins.py index 2bcf35af0..370e472b5 100644 --- a/tests/agent/test_agent_plugins.py +++ b/tests/agent/test_agent_plugins.py @@ -256,6 +256,43 @@ async def test_restricted_project_can_read_only_enabled_plugin_skill( assert resource.read_text(encoding="utf-8") == "plugin reference" +@pytest.mark.asyncio +async def test_restricted_project_revokes_cached_plugin_read_after_replacement( + 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("trusted plugin reference", encoding="utf-8") + read_tool = ReadFileTool.create( + ToolContext( + config=ToolsConfig(restrict_to_workspace=True), + workspace=str(agent_workspace), + ) + ) + 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: + trusted_result = await read_tool.execute(path=str(resource)) + resource.write_text("replacement plugin reference", encoding="utf-8") + replacement_result = await read_tool.execute(path=str(resource)) + finally: + reset_workspace_scope(token) + + assert "trusted plugin reference" in trusted_result + assert "outside allowed directory" in replacement_result + assert discover_agent_plugins(agent_workspace)[0].enabled is False + + def test_plugin_state_symlink_cannot_escape_config_root( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: