fix(mcp): close redirect bypass + empty-getaddrinfo gap on SSRF guard

Three follow-ups to the OAuth-discovery SSRF guard: 1. Greptile P1 (redirect bypass): the validated origin could return a 3xx whose ``Location`` points at an internal address, and httpx would follow without re-checking the new target. Pass ``follow_redirects=False`` to both gated httpx GETs. Spec-compliant OAuth/OIDC metadata endpoints serve the JSON directly, so this doesn't affect legitimate providers. 2. Greptile P2 (empty getaddrinfo): POSIX doesn't strictly forbid an empty success-list from ``getaddrinfo``. Add an explicit ``if not infos: return False`` so the guard fails closed instead of falling through to ``return True``. 3. Mypy: ``info[4][0]`` is typed ``str | int``; narrow at the boundary with an ``isinstance`` check (fail-closed if non-str). Adds two regression tests verifying ``follow_redirects=False`` is passed at both gated fetch sites, and one verifying the empty-list case rejects the URL.
2026-08-10 16:22:17 +00:00 · 2026-04-30 02:59:50 +00:00
parent df62dd8768
commit 1bb04cdc72
2 changed files with 92 additions and 4 deletions
@@ -1550,6 +1550,9 @@ class MCPServerManager:
        except socket.gaierror:
            return False

+        if not infos:
+            return False
+
        # Reuse the proxy-wide outbound block list (private / loopback /
        # link-local / multicast / reserved / cloud-fabric IPs).  Defence
        # in depth only — the resolution here and the one httpx performs at
@@ -1557,7 +1560,10 @@ class MCPServerManager:
        # also controls the resource server is already in scope, so the
        # primary mitigation is the same-authority pin above.
        for info in infos:
-            if _is_blocked_ip(info[4][0]):
+            sockaddr_host = info[4][0]
+            if not isinstance(sockaddr_host, str):
+                return False
+            if _is_blocked_ip(sockaddr_host):
                return False
        return True

@@ -1697,7 +1703,10 @@ class MCPServerManager:
                llm_provider=httpxSpecialProvider.MCP,
                params={"timeout": MCP_METADATA_TIMEOUT},
            )
-            response = await client.get(resource_metadata_url)
+            # Redirects bypass the SSRF guard (the new ``Location`` is not
+            # re-checked against ``server_url``), so refuse to follow them.
+            # Spec-compliant OAuth metadata endpoints serve the JSON directly.
+            response = await client.get(resource_metadata_url, follow_redirects=False)
            response.raise_for_status()
            data = response.json()
        except Exception as exc:  # pragma: no cover - network issues
@@ -1806,7 +1815,11 @@ class MCPServerManager:
                    llm_provider=httpxSpecialProvider.MCP,
                    params={"timeout": MCP_METADATA_TIMEOUT},
                )
-                response = await client.get(url)
+                # Disable redirects: a redirect to a private IP would bypass
+                # the SSRF guard (the ``Location`` target is not re-checked
+                # against ``server_url``).  Spec-compliant OAuth/OIDC
+                # metadata endpoints serve the JSON directly.
+                response = await client.get(url, follow_redirects=False)
                response.raise_for_status()
                data = response.json()
            except Exception as exc:  # pragma: no cover - network issues
@@ -786,7 +786,7 @@ class TestMCPServerManager:
        manager = MCPServerManager()
        issuer = "https://login.microsoftonline.com/test-tenant-id/v2.0"

-        def build_response(url: str):
+        def build_response(url: str, **kwargs):
            mock_response = MagicMock()
            if url == f"{issuer}/.well-known/openid-configuration":
                mock_response.json.return_value = {
@@ -3089,6 +3089,81 @@ class TestOAuthDiscoverySSRFGuard:
        assert scopes is None
        mock_client.get.assert_not_called()

+    def test_empty_getaddrinfo_result_blocks_url(self, monkeypatch):
+        # POSIX doesn't strictly forbid an empty success-list from getaddrinfo.
+        # The guard must fail closed rather than fall through to ``return True``.
+        monkeypatch.setattr(
+            "litellm.proxy._experimental.mcp_server.mcp_server_manager.socket.getaddrinfo",
+            lambda *a, **k: [],
+        )
+        assert not MCPServerManager._is_safe_metadata_url(
+            "https://no-records.example.com/.well-known/oauth-authorization-server",
+            "https://legit-mcp.example.com/mcp",
+        )
+
+    @pytest.mark.asyncio
+    async def test_fetch_oauth_metadata_does_not_follow_redirects(self):
+        # If the validated origin redirects to a loopback or other unsafe
+        # address, httpx must NOT follow — the new ``Location`` would not
+        # be re-checked against ``server_url``.
+        manager = MCPServerManager()
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "authorization_servers": ["https://auth.example.com"],
+        }
+        mock_response.raise_for_status = MagicMock()
+
+        captured_kwargs: Dict[str, Any] = {}
+
+        async def fake_get(url, **kwargs):
+            captured_kwargs.update(kwargs)
+            return mock_response
+
+        mock_client = MagicMock()
+        mock_client.get = AsyncMock(side_effect=fake_get)
+
+        with patch(
+            "litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
+            return_value=mock_client,
+        ):
+            await manager._fetch_oauth_metadata_from_resource(
+                "https://protected.example.com/.well-known/oauth",
+                "https://protected.example.com/mcp",
+            )
+        assert captured_kwargs.get("follow_redirects") is False
+
+    @pytest.mark.asyncio
+    async def test_fetch_single_auth_server_does_not_follow_redirects(self):
+        # Same redirect-bypass concern for the authorization-server fetch path.
+        manager = MCPServerManager()
+
+        mock_response = MagicMock()
+        mock_response.json.return_value = {
+            "authorization_endpoint": "https://provider.example.com/authorize",
+            "token_endpoint": "https://provider.example.com/token",
+        }
+        mock_response.raise_for_status = MagicMock()
+
+        captured_kwargs: Dict[str, Any] = {}
+
+        async def fake_get(url, **kwargs):
+            captured_kwargs.update(kwargs)
+            return mock_response
+
+        mock_client = MagicMock()
+        mock_client.get = AsyncMock(side_effect=fake_get)
+
+        with patch(
+            "litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client",
+            return_value=mock_client,
+        ):
+            await manager._fetch_single_authorization_server_metadata(
+                "https://provider.example.com",
+                "https://provider.example.com",
+            )
+        assert captured_kwargs.get("follow_redirects") is False
+
    @pytest.mark.asyncio
    async def test_fetch_authorization_server_refuses_unsafe_issuer(self, monkeypatch):
        # Mirrors the GHSA-mrfv repro: PRM lists a loopback issuer URL.