fix(mcp): gate /public/mcp_hub strictly on litellm.public_mcp_servers

litellm/__init__.py

@@ -426,6 +426,7 @@
     False  # If false (default), converts all 'system' role messages to 'assistant' for GitHub Copilot compatibility. Set to true to disable this behavior.
 )
 public_mcp_servers: Optional[List[str]] = None
+public_mcp_hub_strict_whitelist: bool = True
 public_model_groups: Optional[List[str]] = None
 public_agent_groups: Optional[List[str]] = None
 # Supports both old format (Dict[str, str]) and new format (Dict[str, Dict[str, Any]])

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -3160,15 +3160,37 @@ def get_mcp_server_by_id(self, server_id: str) -> Optional[MCPServer]:
 
     def get_public_mcp_servers(self) -> List[MCPServer]:
         """
-        Get the public MCP servers (available_on_public_internet=True flag on server).
-        Also includes servers from litellm.public_mcp_servers for backwards compat.
-        """
-        servers: List[MCPServer] = []
+        Return the MCP servers published to the AI Hub via /v1/mcp/make_public.
+
+        Default (litellm.public_mcp_hub_strict_whitelist=True): mirrors
+        /public/model_hub and /public/agent_hub — gates strictly on the
+        litellm.public_mcp_servers whitelist. Returns an empty list when no
+        servers have been published. The per-server available_on_public_internet
+        flag is unrelated — it governs IP-based access in
+        _is_server_accessible_from_ip, not hub visibility.
+
+        Legacy (litellm.public_mcp_hub_strict_whitelist=False): preserves the
+        pre-fix behavior where any server with available_on_public_internet=True
+        is also included. Intended as a one-release migration window for
+        deployments that relied on the OR-with-default semantics; will be
+        removed in a future release.
+        """
+        if litellm.public_mcp_hub_strict_whitelist:
+            if litellm.public_mcp_servers is None:
+                return []
+            public_ids = set(litellm.public_mcp_servers)
+            return [
+                server
+                for server in self.get_registry().values()
+                if server.server_id in public_ids
+            ]
+
         public_ids = set(litellm.public_mcp_servers or [])
-        for server in self.get_registry().values():
-            if server.available_on_public_internet or server.server_id in public_ids:
-                servers.append(server)
-        return servers
+        return [
+            server
+            for server in self.get_registry().values()
+            if server.available_on_public_internet or server.server_id in public_ids
+        ]
 
     def expand_permission_list(self, identifiers: List[str]) -> List[str]:
         """

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -3436,5 +3436,140 @@ async def test_update_server_noop_for_unregistered_pending(self):
         assert "never-seen" not in manager.registry
 
 
+class TestGetPublicMCPServers:
+    """
+    /public/mcp_hub strict-whitelist semantics — mirrors /public/model_hub
+    and /public/agent_hub. Regression test for the PR #20607 OR-with-default
+    behavior that made `litellm.public_mcp_servers` ignored by the hub.
+    """
+
+    def _make_server(self, server_id, available_on_public_internet=True):
+        return MCPServer(
+            server_id=server_id,
+            name=server_id,
+            server_name=server_id,
+            transport=MCPTransport.http,
+            available_on_public_internet=available_on_public_internet,
+        )
+
+    def _make_manager(self, servers):
+        manager = MCPServerManager()
+        for s in servers:
+            manager.config_mcp_servers[s.server_id] = s
+        return manager
+
+    @patch("litellm.public_mcp_servers", None)
+    def test_returns_empty_when_whitelist_is_none(self):
+        """No /make_public call yet → hub returns nothing, regardless of
+        per-server flags."""
+        manager = self._make_manager(
+            [
+                self._make_server("a", available_on_public_internet=True),
+                self._make_server("b", available_on_public_internet=True),
+            ]
+        )
+        assert manager.get_public_mcp_servers() == []
+
+    @patch("litellm.public_mcp_servers", [])
+    def test_returns_empty_when_whitelist_is_empty(self):
+        """Explicit empty whitelist → hub returns nothing."""
+        manager = self._make_manager(
+            [self._make_server("a", available_on_public_internet=True)]
+        )
+        assert manager.get_public_mcp_servers() == []
+
+    @patch("litellm.public_mcp_servers", ["a"])
+    def test_returns_only_whitelisted_when_flag_defaults_to_true(self):
+        """
+        Regression: prior to the fix, every server with
+        available_on_public_internet=True (the default) leaked into the hub
+        regardless of the whitelist. Whitelist must be authoritative.
+        """
+        manager = self._make_manager(
+            [
+                self._make_server("a", available_on_public_internet=True),
+                self._make_server("b", available_on_public_internet=True),
+            ]
+        )
+        result = manager.get_public_mcp_servers()
+        assert [s.server_id for s in result] == ["a"]
+
+    @patch("litellm.public_mcp_servers", ["a"])
+    def test_does_not_leak_servers_via_internal_flag(self):
+        """
+        available_on_public_internet is an IP-gating flag, not a hub flag.
+        A server with the flag True that is not in the whitelist must not
+        appear in the hub.
+        """
+        manager = self._make_manager(
+            [
+                self._make_server("a", available_on_public_internet=False),
+                self._make_server("b", available_on_public_internet=True),
+            ]
+        )
+        result = manager.get_public_mcp_servers()
+        assert [s.server_id for s in result] == ["a"]
+
+    @patch("litellm.public_mcp_servers", ["does-not-exist"])
+    def test_stale_whitelist_id_returns_empty(self):
+        """Whitelist references an unknown server_id → no spurious results."""
+        manager = self._make_manager(
+            [self._make_server("a", available_on_public_internet=True)]
+        )
+        assert manager.get_public_mcp_servers() == []
+
+
+class TestGetPublicMCPServersLegacyMode:
+    """
+    Legacy migration knob: litellm.public_mcp_hub_strict_whitelist=False
+    preserves the pre-fix OR-with-default semantics for one release so
+    operators that relied on the old behavior have a window to call
+    /v1/mcp/make_public before /public/mcp_hub goes empty.
+    """
+
+    def _make_server(self, server_id, available_on_public_internet=True):
+        return MCPServer(
+            server_id=server_id,
+            name=server_id,
+            server_name=server_id,
+            transport=MCPTransport.http,
+            available_on_public_internet=available_on_public_internet,
+        )
+
+    def _make_manager(self, servers):
+        manager = MCPServerManager()
+        for s in servers:
+            manager.config_mcp_servers[s.server_id] = s
+        return manager
+
+    @patch("litellm.public_mcp_hub_strict_whitelist", False)
+    @patch("litellm.public_mcp_servers", None)
+    def test_legacy_returns_default_flag_servers_when_whitelist_is_none(self):
+        """Legacy mode + no whitelist → every server with the default
+        available_on_public_internet=True appears (old behavior)."""
+        manager = self._make_manager(
+            [
+                self._make_server("a", available_on_public_internet=True),
+                self._make_server("b", available_on_public_internet=False),
+            ]
+        )
+        result = manager.get_public_mcp_servers()
+        assert [s.server_id for s in result] == ["a"]
+
+    @patch("litellm.public_mcp_hub_strict_whitelist", False)
+    @patch("litellm.public_mcp_servers", ["b"])
+    def test_legacy_unions_whitelist_and_default_flag(self):
+        """Legacy mode unions the whitelist with any
+        available_on_public_internet=True server."""
+        manager = self._make_manager(
+            [
+                self._make_server("a", available_on_public_internet=True),
+                self._make_server("b", available_on_public_internet=False),
+            ]
+        )
+        result = manager.get_public_mcp_servers()
+        assert sorted(s.server_id for s in result) == ["a", "b"]
+
+
 if __name__ == "__main__":
     pytest.main([__file__])

tests/test_litellm/proxy/public_endpoints/test_public_endpoints.py

@@ -639,3 +639,67 @@ def test_clean_display_name_strips_suffix():
 def test_clean_display_name_passthrough_when_no_suffix():
     assert _clean_display_name("OpenAI") == "OpenAI"
     assert _clean_display_name("") == ""
+
+
+def test_public_mcp_hub_returns_only_whitelisted_servers():
+    """Regression: /public/mcp_hub must gate strictly on
+    litellm.public_mcp_servers, mirroring /public/model_hub and
+    /public/agent_hub. Servers with available_on_public_internet=True that
+    are not on the whitelist must not leak."""
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+    from litellm.proxy._types import MCPTransport
+
+    app = FastAPI()
+    app.include_router(router)
+    app.dependency_overrides[user_api_key_auth] = lambda: MagicMock()
+    client = TestClient(app)
+
+    listed = MCPServer(
+        server_id="listed",
+        name="listed",
+        server_name="listed",
+        transport=MCPTransport.http,
+        available_on_public_internet=True,
+    )
+
+    mock_manager = MagicMock()
+    mock_manager.get_public_mcp_servers.return_value = [listed]
+
+    with (
+        patch("litellm.public_mcp_servers", ["listed"]),
+        patch(
+            "litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager",
+            mock_manager,
+        ),
+    ):
+        response = client.get("/public/mcp_hub")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert [item["server_id"] for item in data] == ["listed"]
+    app.dependency_overrides.clear()
+
+
+def test_public_mcp_hub_returns_empty_when_whitelist_unset():
+    """When no servers have been published via /v1/mcp/make_public, the
+    hub returns an empty list (matches /public/agent_hub behavior)."""
+    app = FastAPI()
+    app.include_router(router)
+    app.dependency_overrides[user_api_key_auth] = lambda: MagicMock()
+    client = TestClient(app)
+
+    mock_manager = MagicMock()
+    mock_manager.get_public_mcp_servers.return_value = []
+
+    with (
+        patch("litellm.public_mcp_servers", None),
+        patch(
+            "litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager",
+            mock_manager,
+        ),
+    ):
+        response = client.get("/public/mcp_hub")
+
+    assert response.status_code == 200
+    assert response.json() == []
+    app.dependency_overrides.clear()