fix(mcp): validate loopback redirect_uri in discoverable OAuth proxy

Extract the BYOK loopback redirect_uri check into a shared
oauth_utils.validate_loopback_redirect_uri helper. Call it in
discoverable_endpoints.authorize_with_server before the client-supplied
redirect_uri is encrypted into the OAuth state.

Without this check, a non-loopback redirect_uri was encoded into the
state parameter and decoded on /callback to 302 the user back to the
attacker's URL with the authorization code attached — an open-redirect
+ code-theft primitive (VERIA-57 root cause B). The /callback handler
is already safe because state is HMAC-signed via encrypt_value_helper,
so validating at /authorize before encoding is sufficient.

Also updates existing tests to use loopback client redirect_uris and
adds regression tests for non-loopback rejection, IPv4 127.0.0.0/8
range acceptance, and full-form IPv6 loopback acceptance.

litellm/proxy/_experimental/mcp_server/byok_oauth_endpoints.py

@@ -18,9 +18,8 @@ import hashlib
 import html as _html_module
 import time
 import uuid
-from ipaddress import ip_address
 from typing import Dict, Optional, cast
-from urllib.parse import urlencode, urlparse
+from urllib.parse import urlencode
 
 import jwt
 from fastapi import APIRouter, Depends, Form, HTTPException, Request
@@ -31,6 +30,9 @@ from litellm.proxy._experimental.mcp_server.db import store_user_credential
 from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
     get_request_base_url,
 )
+from litellm.proxy._experimental.mcp_server.oauth_utils import (
+    validate_loopback_redirect_uri,
+)
 from litellm.proxy._types import UserAPIKeyAuth
 
 # ---------------------------------------------------------------------------
@@ -86,32 +88,6 @@ def _oauth_token_error(code: str, status: int = 400) -> JSONResponse:
     )
 
 
-def _validate_redirect_uri(redirect_uri: str) -> None:
-    """Require a loopback redirect_uri (OAuth 2.1 §4.1.2.1 + RFC 8252
-    native-app pattern). MCP clients are native apps that listen on a
-    localhost port; rejecting non-loopback URIs prevents a malicious MCP
-    client from pointing the callback at its own server to capture the
-    code after a legitimate user enters their upstream API key.
-
-    Accepts the literal ``localhost`` plus any IP in the loopback ranges
-    (IPv4 ``127.0.0.0/8`` and IPv6 ``::1``) per RFC 8252 §7.3 — a string
-    match on ``"127.0.0.1"`` would miss ``127.0.0.2`` and full-form IPv6
-    (``0:0:0:0:0:0:0:1``).
-    """
-    parsed = urlparse(redirect_uri)
-    if parsed.scheme not in ("http", "https"):
-        raise HTTPException(status_code=400, detail="invalid_request")
-    host = (parsed.hostname or "").lower()
-    if host == "localhost":
-        return
-    try:
-        if ip_address(host).is_loopback:
-            return
-    except ValueError:
-        pass
-    raise HTTPException(status_code=400, detail="invalid_request")
-
-
 def _user_id_from_session_cookie(request: Request) -> Optional[str]:
     """Return user_id from the UI ``token`` cookie (HS256-signed with
     ``master_key``), or None if missing/invalid.
@@ -677,7 +653,7 @@ async def byok_authorize_get(
         raise HTTPException(status_code=400, detail="redirect_uri is required")
     # Validate here too so the user sees the rejection before typing their
     # API key into the HTML form (the POST handler also validates).
-    _validate_redirect_uri(redirect_uri)
+    validate_loopback_redirect_uri(redirect_uri)
     if not code_challenge:
         raise HTTPException(status_code=400, detail="code_challenge is required")
 
@@ -737,7 +713,7 @@ async def byok_authorize_post(
     """
     _purge_expired_codes()
 
-    _validate_redirect_uri(redirect_uri)
+    validate_loopback_redirect_uri(redirect_uri)
 
     # Reject new codes if the store is at capacity (prevents memory exhaustion
     # from a burst of abandoned OAuth flows).

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -10,6 +10,9 @@ from litellm.llms.custom_httpx.http_handler import (
     get_async_httpx_client,
     httpxSpecialProvider,
 )
+from litellm.proxy._experimental.mcp_server.oauth_utils import (
+    validate_loopback_redirect_uri,
+)
 from litellm.proxy.auth.ip_address_utils import IPAddressUtils
 from litellm.proxy.common_utils.encrypt_decrypt_utils import (
     decrypt_value_helper,
@@ -322,15 +325,13 @@ async def authorize_with_server(
             status_code=400, detail="MCP server authorization url is not set"
         )
 
+    # Loopback-only redirect_uri. The URI is encrypted into the OAuth
+    # state and decoded on /callback to redirect the user back; a non-
+    # loopback URI would be an open-redirect + code-theft primitive
+    # (VERIA-57 root cause B). MCP clients are native apps — loopback is
+    # the spec-compliant callback pattern.
+    validate_loopback_redirect_uri(redirect_uri)
     parsed = urlparse(redirect_uri)
-    if parsed.scheme not in ("http", "https"):
-        raise HTTPException(
-            status_code=400,
-            detail={
-                "error": "invalid_redirect_uri",
-                "message": "redirect_uri must use http or https scheme",
-            },
-        )
     base_url = urlunparse(parsed._replace(query=""))
     request_base_url = get_request_base_url(request)
     encoded_state = encode_state_with_base_url(

litellm/proxy/_experimental/mcp_server/oauth_utils.py

@@ -0,0 +1,34 @@
+"""Shared helpers for the MCP OAuth authorization endpoints
+(BYOK + discoverable / pass-through OAuth proxy)."""
+
+from ipaddress import ip_address
+from urllib.parse import urlparse
+
+from fastapi import HTTPException
+
+
+def validate_loopback_redirect_uri(redirect_uri: str) -> None:
+    """Require a loopback ``redirect_uri`` (OAuth 2.1 §4.1.2.1 + RFC 8252
+    §7.3 native-app pattern). MCP clients are native apps that listen on
+    a localhost port; rejecting non-loopback URIs prevents a malicious
+    client from pointing the callback at its own server to capture the
+    authorization code — the credential-theft primitive behind VERIA-57
+    and pNr1PHa9.
+
+    Accepts the literal ``localhost`` plus any IP in the loopback ranges
+    (IPv4 ``127.0.0.0/8`` and IPv6 ``::1``). A string match on
+    ``"127.0.0.1"`` alone would miss ``127.0.0.2`` and the full-form
+    IPv6 loopback ``0:0:0:0:0:0:0:1``.
+    """
+    parsed = urlparse(redirect_uri)
+    if parsed.scheme not in ("http", "https"):
+        raise HTTPException(status_code=400, detail="invalid_request")
+    host = (parsed.hostname or "").lower()
+    if host == "localhost":
+        return
+    try:
+        if ip_address(host).is_loopback:
+            return
+    except ValueError:
+        pass
+    raise HTTPException(status_code=400, detail="invalid_request")

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

@@ -76,7 +76,7 @@ async def test_authorize_endpoint_includes_response_type():
             request=mock_request,
             client_id="test_client_id",
             mcp_server_name="test_oauth",
-            redirect_uri="https://client.example.com/callback",
+            redirect_uri="http://127.0.0.1:60108/callback",
             state="test_state",
         )
 
@@ -139,7 +139,7 @@ async def test_authorize_endpoint_preserves_existing_query_params():
             request=mock_request,
             client_id="test_client_id",
             mcp_server_name="test_oauth",
-            redirect_uri="https://client.example.com/callback",
+            redirect_uri="http://127.0.0.1:60108/callback",
             state="test_state",
         )
 
@@ -558,7 +558,7 @@ async def test_authorize_endpoint_respects_x_forwarded_proto():
             request=mock_request,
             client_id="test_client_id",
             mcp_server_name="test_oauth",
-            redirect_uri="https://client.example.com/callback",
+            redirect_uri="http://127.0.0.1:60108/callback",
             state="test_state",
         )
 
@@ -855,7 +855,7 @@ async def test_authorize_endpoint_respects_x_forwarded_host():
             request=mock_request,
             client_id="test_client_id",
             mcp_server_name="test_oauth",
-            redirect_uri="https://client.example.com/callback",
+            redirect_uri="http://127.0.0.1:60108/callback",
             state="test_state",
         )
 
@@ -1817,3 +1817,106 @@ async def test_token_endpoint_authorization_code_missing_code():
         )
     assert exc_info.value.status_code == 400
     assert "code is required" in str(exc_info.value.detail)
+
+
+@pytest.mark.asyncio
+async def test_authorize_endpoint_rejects_non_loopback_redirect_uri():
+    """VERIA-57 root cause B regression. The client-supplied redirect_uri
+    is encrypted into the OAuth state and decoded on /callback to 302 the
+    user back. A non-loopback value is an open-redirect + code-theft
+    primitive — reject with 400 before encoding anything into state."""
+    from fastapi import Request
+
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+        authorize,
+    )
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
+    from litellm.proxy._types import MCPTransport
+    from litellm.types.mcp import MCPAuth
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+    global_mcp_server_manager.registry.clear()
+    oauth2_server = MCPServer(
+        server_id="test_oauth_server",
+        name="test_oauth",
+        server_name="test_oauth",
+        alias="test_oauth",
+        transport=MCPTransport.http,
+        auth_type=MCPAuth.oauth2,
+        client_id="cid",
+        client_secret="cs",
+        authorization_url="https://provider.com/oauth/authorize",
+        token_url="https://provider.com/oauth/token",
+    )
+    global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://litellm.example.com/"
+    mock_request.headers = {}
+
+    with pytest.raises(HTTPException) as exc_info:
+        await authorize(
+            request=mock_request,
+            client_id="cid",
+            mcp_server_name="test_oauth",
+            redirect_uri="https://attacker.example.com/cb",
+            state="s",
+        )
+    assert exc_info.value.status_code == 400
+
+
+@pytest.mark.asyncio
+async def test_authorize_endpoint_accepts_ipv4_loopback_range_and_ipv6_full_form():
+    """RFC 8252 §7.3 + RFC 4291: full 127.0.0.0/8 and full-form IPv6
+    loopback must be accepted — string match on ``127.0.0.1`` alone
+    would miss ``127.0.0.2`` and ``0:0:0:0:0:0:0:1``."""
+    from fastapi import Request
+
+    from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
+        authorize,
+    )
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+        global_mcp_server_manager,
+    )
+    from litellm.proxy._types import MCPTransport
+    from litellm.types.mcp import MCPAuth
+    from litellm.types.mcp_server.mcp_server_manager import MCPServer
+
+    global_mcp_server_manager.registry.clear()
+    oauth2_server = MCPServer(
+        server_id="test_oauth_server",
+        name="test_oauth",
+        server_name="test_oauth",
+        alias="test_oauth",
+        transport=MCPTransport.http,
+        auth_type=MCPAuth.oauth2,
+        client_id="cid",
+        client_secret="cs",
+        authorization_url="https://provider.com/oauth/authorize",
+        token_url="https://provider.com/oauth/token",
+    )
+    global_mcp_server_manager.registry[oauth2_server.server_id] = oauth2_server
+
+    mock_request = MagicMock(spec=Request)
+    mock_request.base_url = "https://litellm.example.com/"
+    mock_request.headers = {}
+
+    for uri in (
+        "http://127.0.0.2:3000/cb",
+        "http://[0:0:0:0:0:0:0:1]:3000/cb",
+        "http://localhost:3000/cb",
+    ):
+        with patch(
+            "litellm.proxy._experimental.mcp_server.discoverable_endpoints.encrypt_value_helper"
+        ) as mock_encrypt:
+            mock_encrypt.return_value = "mocked_encrypted_state"
+            response = await authorize(
+                request=mock_request,
+                client_id="cid",
+                mcp_server_name="test_oauth",
+                redirect_uri=uri,
+                state="s",
+            )
+        assert response.status_code == 307, f"{uri} should be accepted"