chore(mcp): validate outbound OAuth URLs and fail closed on missing client IP

The OAuth proxy endpoints /authorize, /token, and /register sit mid-OAuth-
handshake and can't require Depends(user_api_key_auth). They forward to
admin-configured token_url and registration_url. An unauthenticated caller
hitting /token or /register made the proxy POST to whatever URL the admin
configured for the MCP server, which let an internal IdP be probed via the
proxy if the admin had registered one.

Validate the outbound URL through litellm_core_utils.url_utils.validate_url
before each POST in register_client_with_server and exchange_token_with_server.
The helper resolves DNS, blocks RFC1918 / loopback / link-local destinations,
honours the existing user_url_allowed_hosts allowlist, and rewrites the URL
to the validated IP to defeat DNS rebinding. Operators who need to point at
an internal IdP can opt in via the same allowlist that other outbound
URL-validation sites use, or set litellm.user_url_validation = False.

Separately, _is_server_accessible_from_ip used to fail open when client_ip
was None, which let request handlers that couldn't determine a client IP
reach internal-only servers. Lock the contract: None now fails closed and
internal callers must pass the new INTERNAL_REQUEST sentinel to bypass IP
gating. get_mcp_server_by_name keeps its existing "None means internal"
wrapper convention by translating to the sentinel internally, so internal
callers (auth, debug, registry maintenance) continue to work unchanged.
The user-facing callsites in rest_endpoints that previously short-circuited
on client_ip is None now hit the gate and inherit the fail-closed behaviour.

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -5,7 +5,9 @@ from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 from fastapi import APIRouter, Form, HTTPException, Request
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
 
+import litellm
 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.url_utils import SSRFError, validate_url
 from litellm.llms.custom_httpx.http_handler import (
     get_async_httpx_client,
     httpxSpecialProvider,
@@ -25,6 +27,26 @@ from litellm.proxy.utils import get_server_root_path
 from litellm.types.mcp import MCPAuth
 from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
+
+def _validate_mcp_oauth_outbound_url(url: str, role: str) -> tuple[str, str]:
+    """Validate an admin-configured OAuth URL before the proxy makes a request to
+    it. The /token, /register and similar endpoints are reachable without a
+    LiteLLM API key (they sit in the middle of an OAuth handshake), so an
+    unauthenticated caller can trigger the proxy to POST to whatever host the
+    admin configured. Resolve and validate the host before sending so an
+    operator who points an MCP server at an internal IdP doesn't unintentionally
+    expose it as an SSRF probe via the proxy."""
+    if not getattr(litellm, "user_url_validation", True):
+        return url, urlparse(url).hostname or ""
+    try:
+        return validate_url(url)
+    except SSRFError as exc:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Configured MCP {role} URL is not safe to call: {exc}",
+        )
+
+
 router = APIRouter(
     tags=["mcp"],
 )
@@ -393,9 +415,12 @@ async def exchange_token_with_server(
             token_data["code_verifier"] = code_verifier
 
     async_client = get_async_httpx_client(llm_provider=httpxSpecialProvider.Oauth2Check)
+    target_url, host_header = _validate_mcp_oauth_outbound_url(
+        mcp_server.token_url, role="token"
+    )
     response = await async_client.post(
-        mcp_server.token_url,
-        headers={"Accept": "application/json"},
+        target_url,
+        headers={"Accept": "application/json", "Host": host_header},
         data=token_data,
     )
 
@@ -498,9 +523,12 @@ async def register_client_with_server(
     async_client = get_async_httpx_client(
         llm_provider=httpxSpecialProvider.Oauth2Register
     )
+    target_url, host_header = _validate_mcp_oauth_outbound_url(
+        mcp_server.registration_url, role="registration"
+    )
     response = await async_client.post(
-        mcp_server.registration_url,
-        headers=headers,
+        target_url,
+        headers={**headers, "Host": host_header},
         json=register_data,
     )
     response.raise_for_status()

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -166,6 +166,21 @@ def _deserialize_json_dict(data: Any) -> Optional[Dict[str, str]]:
         return data
 
 
+class _InternalRequest:
+    """Sentinel passed in place of a client IP to indicate that the call site
+    is internal (background reload, registry maintenance, admin debug) and
+    should bypass IP-based access control. External request handlers must pass
+    a real IP string instead — passing ``None`` now fails closed."""
+
+    __slots__ = ()
+
+    def __repr__(self) -> str:  # pragma: no cover - trivial
+        return "INTERNAL_REQUEST"
+
+
+INTERNAL_REQUEST = _InternalRequest()
+
+
 class MCPServerManager:
     _STDIO_ENV_TEMPLATE_PATTERN = re.compile(r"^\$\{(X-[^}]+)\}$")
 
@@ -3066,17 +3081,28 @@ class MCPServerManager:
             return {}
 
     def _is_server_accessible_from_ip(
-        self, server: MCPServer, client_ip: Optional[str]
+        self,
+        server: MCPServer,
+        client_ip: "Optional[str] | _InternalRequest",
     ) -> bool:
         """
         Check if a server is accessible from the given client IP.
 
-        - If client_ip is None, no IP filtering is applied (internal callers).
-        - If the server has available_on_public_internet=True, it's always accessible.
+        - ``INTERNAL_REQUEST`` bypasses IP filtering (internal callers, admin
+          debug, registry maintenance).
+        - ``None`` fails closed: external request handlers must extract a real
+          IP via ``IPAddressUtils.get_mcp_client_ip(request)`` and reject the
+          request when extraction fails. Earlier behaviour treated ``None`` as
+          "no filter," which let external callers reach internal-only servers
+          when IP extraction silently failed.
+        - If the server has ``available_on_public_internet=True``, it's
+          always accessible.
         - Otherwise, only internal/private IPs can access it.
         """
-        if client_ip is None:
+        if client_ip is INTERNAL_REQUEST:
             return True
+        if client_ip is None:
+            return False
         if server.available_on_public_internet:
             return True
         # Check backwards compat: litellm.public_mcp_servers
@@ -3193,24 +3219,31 @@ class MCPServerManager:
             server_name: The server name to look up.
             client_ip: Optional client IP for access control. When provided,
                        non-public servers are hidden from external IPs.
+                       ``None`` is treated as "internal context, no IP gating"
+                       to preserve the existing contract for internal callers
+                       (auth, debug, registry maintenance). External request
+                       handlers should pass a real IP.
         """
+        # Translate the wrapper-level "None means internal" convention into the
+        # gate function's explicit sentinel so it doesn't fail closed.
+        gate_arg = INTERNAL_REQUEST if client_ip is None else client_ip
         registry = self.get_registry()
         # Pass 1: Match by alias (highest priority)
         for server in registry.values():
             if server.alias == server_name:
-                if not self._is_server_accessible_from_ip(server, client_ip):
+                if not self._is_server_accessible_from_ip(server, gate_arg):
                     return None
                 return server
         # Pass 2: Match by server_name
         for server in registry.values():
             if server.server_name == server_name:
-                if not self._is_server_accessible_from_ip(server, client_ip):
+                if not self._is_server_accessible_from_ip(server, gate_arg):
                     return None
                 return server
         # Pass 3: Match by name (lowest priority)
         for server in registry.values():
             if server.name == server_name:
-                if not self._is_server_accessible_from_ip(server, client_ip):
+                if not self._is_server_accessible_from_ip(server, gate_arg):
                     return None
                 return server
         return None

litellm/proxy/_experimental/mcp_server/rest_endpoints.py

@@ -393,7 +393,6 @@ if MCP_AVAILABLE:
             )
             if (
                 _server is not None
-                and rest_client_ip is not None
                 and not global_mcp_server_manager._is_server_accessible_from_ip(
                     _server, rest_client_ip
                 )
@@ -483,7 +482,6 @@ if MCP_AVAILABLE:
             )
             if (
                 _server is not None
-                and rest_client_ip is not None
                 and not global_mcp_server_manager._is_server_accessible_from_ip(
                     _server, rest_client_ip
                 )

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

@@ -503,6 +503,13 @@ async def test_register_client_remote_registration_success():
                 "litellm.proxy._experimental.mcp_server.discoverable_endpoints.get_async_httpx_client",
                 return_value=mock_async_client,
             ),
+            patch(
+                "litellm.proxy._experimental.mcp_server.discoverable_endpoints.validate_url",
+                return_value=(
+                    "https://provider.example/oauth/register",
+                    "provider.example",
+                ),
+            ),
         ):
             response = await register_client(
                 request=mock_request, mcp_server_name=oauth2_server.server_name
@@ -522,6 +529,7 @@ async def test_register_client_remote_registration_success():
     assert call_args.kwargs["headers"] == {
         "Content-Type": "application/json",
         "Accept": "application/json",
+        "Host": "provider.example",
     }
     assert call_args.kwargs["json"]["redirect_uris"] == [
         "https://proxy.litellm.example/callback"
@@ -2473,3 +2481,73 @@ async def test_token_endpoint_sets_no_store_cache_control():
 
     assert response.headers["cache-control"] == "no-store"
     assert response.headers["pragma"] == "no-cache"
+
+
+class TestOutboundOAuthURLValidation:
+    """
+    /token and /register can be hit without a LiteLLM API key (the caller is
+    mid-OAuth-handshake). They forward the request to whatever URL the admin
+    configured for the MCP server. Lock the new contract: the outbound URL is
+    SSRF-validated via litellm_core_utils.url_utils.validate_url before any
+    proxy POST, so an admin-configured internal IdP can't be probed by an
+    unauthenticated caller through the proxy.
+    """
+
+    def test_validator_rejects_internal_url(self):
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        with pytest.raises(HTTPException) as exc_info:
+            _validate_mcp_oauth_outbound_url(
+                "http://127.0.0.1:8080/token", role="token"
+            )
+        assert exc_info.value.status_code == 400
+        assert "token" in str(exc_info.value.detail)
+
+    def test_validator_rejects_rfc1918_url(self):
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        with pytest.raises(HTTPException):
+            _validate_mcp_oauth_outbound_url(
+                "http://192.168.1.10/oauth/register", role="registration"
+            )
+
+    def test_validator_passes_when_validation_disabled(self):
+        # Operators who explicitly opt out via litellm.user_url_validation = False
+        # still get the URL passed through unchanged.
+        import litellm
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        original = getattr(litellm, "user_url_validation", True)
+        try:
+            litellm.user_url_validation = False
+            url, host = _validate_mcp_oauth_outbound_url(
+                "http://10.0.0.1:5000/token", role="token"
+            )
+            assert url == "http://10.0.0.1:5000/token"
+            assert host == "10.0.0.1"
+        finally:
+            litellm.user_url_validation = original
+
+    def test_validator_accepts_public_url(self):
+        # The public-URL acceptance path does live DNS resolution, which is
+        # flaky in CI. Patch validate_url to confirm the helper passes the
+        # validated tuple through.
+        from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+            _validate_mcp_oauth_outbound_url,
+        )
+
+        with patch(
+            "litellm.proxy._experimental.mcp_server.discoverable_endpoints.validate_url",
+            return_value=("https://provider.example/token", "provider.example"),
+        ):
+            url, host = _validate_mcp_oauth_outbound_url(
+                "https://provider.example/token", role="token"
+            )
+        assert url == "https://provider.example/token"
+        assert host == "provider.example"

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

@@ -3311,5 +3311,105 @@ class TestOAuthDiscoverySSRFGuard:
         mock_client.get.assert_not_called()
 
 
+class TestIPGatingFailClosed:
+    """
+    The IP-gating helper used to fail open when ``client_ip`` was ``None``,
+    which let request handlers that couldn't determine a client IP reach
+    internal-only MCP servers. Lock the new contract: ``None`` fails closed,
+    ``INTERNAL_REQUEST`` bypasses gating, real IPs apply the existing rules.
+    """
+
+    def _internal_server(self):
+        from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+        return MCPServer(
+            server_id="internal-1",
+            name="internal",
+            server_name="internal",
+            transport=MCPTransport.http,
+            url="https://internal.local/mcp",
+            available_on_public_internet=False,
+        )
+
+    def _public_server(self):
+        from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+        return MCPServer(
+            server_id="public-1",
+            name="public",
+            server_name="public",
+            transport=MCPTransport.http,
+            url="https://public.example/mcp",
+            available_on_public_internet=True,
+        )
+
+    def test_none_client_ip_fails_closed_for_internal_server(self):
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        assert (
+            manager._is_server_accessible_from_ip(self._internal_server(), None)
+            is False
+        )
+
+    def test_none_client_ip_fails_closed_for_public_server(self):
+        # Even public servers fail closed when client_ip is None — the missing
+        # IP signals an external request that couldn't be attributed, not an
+        # internal call site. Internal callers must use INTERNAL_REQUEST.
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        assert (
+            manager._is_server_accessible_from_ip(self._public_server(), None) is False
+        )
+
+    def test_internal_request_sentinel_bypasses_gating(self):
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            INTERNAL_REQUEST,
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        assert (
+            manager._is_server_accessible_from_ip(
+                self._internal_server(), INTERNAL_REQUEST
+            )
+            is True
+        )
+
+    def test_public_server_accessible_from_external_ip(self):
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        assert (
+            manager._is_server_accessible_from_ip(self._public_server(), "8.8.8.8")
+            is True
+        )
+
+    def test_get_mcp_server_by_name_preserves_internal_contract(self):
+        # Internal callers historically passed client_ip=None to mean "no IP
+        # gating." get_mcp_server_by_name translates None → INTERNAL_REQUEST
+        # so those callers keep working after the gate change.
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            MCPServerManager,
+        )
+
+        manager = MCPServerManager()
+        server = self._internal_server()
+        manager.registry[server.server_id] = server
+
+        result = manager.get_mcp_server_by_name("internal", client_ip=None)
+        assert result is server
+
+        result = manager.get_mcp_server_by_name("internal", client_ip="8.8.8.8")
+        assert result is None
+
+
 if __name__ == "__main__":
     pytest.main([__file__])