fix(mcp): caller sign-in subject follows the single-space Bearer rule; true_passthrough moved-shape regression tests

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -3964,10 +3964,9 @@ class MCPServerManager:
         authorization: Final = (oauth2_headers or {}).get("Authorization") or _raw_header_value(
             raw_headers, "authorization"
         )
-        scheme_and_credential: Final = (authorization or "").split(None, 1)
-        if len(scheme_and_credential) != 2 or scheme_and_credential[0].lower() != "bearer":
+        if not authorization or not authorization.lower().startswith("bearer "):
             return None
-        bearer: Final = scheme_and_credential[1]
+        bearer: Final = authorization[len("bearer ") :]
         if bearer.startswith(LITELLM_VIRTUAL_KEY_PREFIX) or _is_master_key(bearer, master_key):
             return None
         admission_header: Final = _raw_header_value(raw_headers, "x-litellm-api-key")

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

@@ -6978,6 +6978,20 @@ class TestMCPServerManager:
                 None,
                 id="scheme-less-value-is-not-a-subject",
             ),
+            pytest.param(
+                {"x-litellm-api-key": "sk-1234", "authorization": "Bearer\teyJ.x.y"},
+                "sk-1234",
+                None,
+                None,
+                id="tab-separated-bearer-is-not-a-subject",
+            ),
+            pytest.param(
+                {"x-litellm-api-key": "sk-1234", "authorization": "Bearer  eyJ.x.y"},
+                "sk-1234",
+                " eyJ.x.y",
+                " eyJ.x.y",
+                id="bearer-credential-is-taken-verbatim-after-one-space",
+            ),
         ],
     )
     async def test_pre_call_tool_check_separates_raw_bearer_from_subject(

tests/unit/proxy/_experimental/mcp_server/test_mcp_server_tool_calls_and_headers.py

@@ -11060,6 +11060,36 @@ class TestPreemptive401ModeAware:
         assert "/obx" in exact_header
         assert moved_header == exact_header.replace("/obx", f"/{requested}")
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("shape", ["alias_case", "server_id", "x_mcp_servers"])
+    async def test_moved_true_passthrough_shapes_relay_the_exact_name_routes_upstream_challenge(self, shape):
+        from litellm.proxy._experimental.mcp_server import server as server_module
+
+        server = MCPServer(
+            server_id="id-tpx",
+            name="tpx",
+            alias="tpx",
+            server_name="tpx",
+            url="https://tpx.test/mcp",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.true_passthrough,
+            mcp_info={"server_name": "tpx"},
+        )
+        requested, path = {
+            "alias_case": ("TPX", "/mcp/TPX"),
+            "server_id": (server.server_id, f"/mcp/{server.server_id}"),
+            "x_mcp_servers": ("TPX", "/mcp"),
+        }[shape]
+        probe = AsyncMock(return_value=(401, 'Bearer realm="upstream"'))
+
+        with patch.object(server_module, "_probe_upstream_auth", probe):
+            exact = await self._connect_with_a_grant(server, "tpx", "/mcp/tpx")
+            moved = await self._connect_with_a_grant(server, requested, path)
+
+        assert exact.status_code == 401
+        assert (moved.status_code, moved.detail, moved.headers) == (exact.status_code, exact.detail, exact.headers)
+        assert probe.await_args_list == [call(server.url, ""), call(server.url, "")]
+
     @pytest.mark.asyncio
     async def test_aggregate_connect_without_a_server_selection_is_not_challenged(self):
         from litellm.proxy._experimental.mcp_server import server as server_module