diff --git a/claude-code/hooks/test_pretool_mcp.py b/claude-code/hooks/test_pretool_mcp.py index d6b17ea..ae15908 100644 --- a/claude-code/hooks/test_pretool_mcp.py +++ b/claude-code/hooks/test_pretool_mcp.py @@ -146,6 +146,51 @@ def test_string_form_absolute_path_rejected(self): cfg = unbound._resolve_plugin_mcp_config("plugin_absplug_evil", cache_dir=self.cache) self.assertIsNone(cfg) + def test_array_form_resolves_all_servers(self): + # `mcpServers` as an array of config paths must resolve every listed file. + _make_plugin( + self.cache, "mkt", "toolkit", "1.0.0", + { + ".claude-plugin/plugin.json": {"mcpServers": ["./a.json", "./b.json"]}, + "a.json": {"mcpServers": {"alpha": {"url": "https://alpha.example/mcp"}}}, + "b.json": {"mcpServers": {"beta": {"url": "https://beta.example/mcp"}}}, + }, + ) + self.assertEqual( + unbound._resolve_plugin_mcp_config("plugin_toolkit_alpha", cache_dir=self.cache), + {"url": "https://alpha.example/mcp"}, + ) + self.assertEqual( + unbound._resolve_plugin_mcp_config("plugin_toolkit_beta", cache_dir=self.cache), + {"url": "https://beta.example/mcp"}, + ) + + def test_array_form_traversal_escape_rejected(self): + # A ../ element in the array must not read a file outside the version dir. + ver_dir = _make_plugin( + self.cache, "mkt", "arrevil", "1.0.0", + {".claude-plugin/plugin.json": {"mcpServers": ["../../../evil.json"]}}, + ) + outside = ver_dir.parent.parent.parent / "evil.json" + _write_json(outside, {"mcpServers": {"evil": {"url": "https://evil.example/mcp"}}}) + self.assertIsNone( + unbound._resolve_plugin_mcp_config("plugin_arrevil_evil", cache_dir=self.cache) + ) + + def test_array_form_drops_bad_element_keeps_valid(self): + # One escaping element must not poison the rest of the array. + _make_plugin( + self.cache, "mkt", "mixed", "1.0.0", + { + ".claude-plugin/plugin.json": {"mcpServers": ["./ok.json", "../../../evil.json"]}, + "ok.json": {"mcpServers": {"good": {"url": "https://good.example/mcp"}}}, + }, + ) + self.assertEqual( + unbound._resolve_plugin_mcp_config("plugin_mixed_good", cache_dir=self.cache), + {"url": "https://good.example/mcp"}, + ) + def test_identical_config_collision_resolves(self): # Two distinct (plugin, server) pairs mangle to the same candidate AND # share the SAME config -> one distinct entry -> resolves (benign). diff --git a/claude-code/hooks/unbound.py b/claude-code/hooks/unbound.py index 3e3c3c8..3f409a7 100644 --- a/claude-code/hooks/unbound.py +++ b/claude-code/hooks/unbound.py @@ -685,6 +685,30 @@ def _norm_mcp_token(s: Optional[str]) -> str: return re.sub(r'_+', '_', s or '').strip('_') +def _load_plugin_mcp_map(version_dir: Path, rel_path) -> Dict: + # Read a plugin-relative config file's `mcpServers` map. Contain the path to + # the version dir: reject absolute paths and ../ traversal (and symlink + # escapes via resolve()). + if not isinstance(rel_path, str): + return {} + candidate = (version_dir / rel_path).resolve() + try: + candidate.relative_to(version_dir.resolve()) + except ValueError: + log_error(f"mcp plugin path rejected (containment): {rel_path}", 'mcp_plugin') + return {} + if not candidate.is_file(): + return {} + try: + with open(candidate, 'r', encoding='utf-8') as f: + data = json.loads(f.read()) + except Exception as exc: + log_error(f"mcp plugin source unreadable: {candidate}: {exc}", 'mcp_plugin') + return {} + servers = data.get('mcpServers') if isinstance(data, dict) else None + return servers if isinstance(servers, dict) else {} + + def _plugin_mcp_server_map(version_dir: Path) -> Dict: servers = {} sources = [version_dir / ".mcp.json", version_dir / ".claude-plugin" / "plugin.json"] @@ -711,23 +735,21 @@ def _plugin_mcp_server_map(version_dir: Path) -> Dict: } if root_map: mcp_servers = root_map + # Per the plugin spec `mcpServers` is `string | array | object`: a config + # path, a list of config paths, or an inline map. Resolve paths to maps. if isinstance(mcp_servers, str): - # Contain the path to the version dir: reject absolute paths and - # ../ traversal (and symlink escapes via resolve()). - candidate = (version_dir / mcp_servers).resolve() - try: - candidate.relative_to(version_dir.resolve()) - except ValueError: - continue - if candidate.is_file(): - try: - with open(candidate, 'r', encoding='utf-8') as f: - rel_data = json.loads(f.read()) - except Exception as exc: - log_error(f"mcp plugin source unreadable: {candidate}: {exc}", 'mcp_plugin') - continue - if isinstance(rel_data, dict): - mcp_servers = rel_data.get('mcpServers') + mcp_servers = _load_plugin_mcp_map(version_dir, mcp_servers) + elif isinstance(mcp_servers, list): + # First-wins on a duplicate key (consistent with the cross-source + # setdefault below); log a real conflict so a wrong-server fingerprint + # is diagnosable rather than silent. + merged = {} + for elem in mcp_servers: + for key, entry in _load_plugin_mcp_map(version_dir, elem).items(): + if key in merged and merged[key] != entry: + log_error(f"mcp plugin array key collision: {key}", 'mcp_plugin') + merged.setdefault(key, entry) + mcp_servers = merged if isinstance(mcp_servers, dict): for key, entry in mcp_servers.items(): servers.setdefault(key, entry)