diff --git a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py index f1fcc95c532..5a271625c07 100644 --- a/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py +++ b/litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py @@ -25,6 +25,14 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.bridge_credenti from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import ( EnvelopeIdentity, ) +from litellm.proxy._experimental.mcp_server.outbound_credentials.session_credentials import ( + NotSessionBearer, + SessionBearerAdmitted, + SessionBearerInvalid, + is_session_bearer_shaped, + resolve_session_bearer, + session_keys_from_master_key, +) from litellm.proxy._types import ( UI_TEAM_ID, LiteLLM_TeamTable, @@ -124,6 +132,25 @@ def _has_client_supplied_mcp_auth( return bool(mcp_auth_header) or bool(mcp_server_auth_headers) +def _is_aggregate_mcp_scope( + route: str, + mcp_servers: list[str] | None, + mcp_auth_header: str | None, + mcp_server_auth_headers: dict[str, dict[str, str]] | None, +) -> bool: + """True when a request targets the aggregate ``/mcp`` endpoint rather than any named server. + + Aggregate scope means no target resolves from the path or ``x-mcp-servers``, and the caller + supplied no per-server MCP auth headers (which would make it a scripted client, not a + cold-start DCR one). Both the 401 challenge and the gateway session arm consume this, so the + scope the challenge advertises and the scope the session is admitted at cannot drift.""" + if mcp_servers: + return False + if _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers): + return False + return len(MCPRequestHandler._resolve_target_server_names(route, mcp_servers)) == 0 + + def _is_aggregate_gateway_dcr_challenge_scope( route: str, mcp_servers: list[str] | None, @@ -135,17 +162,11 @@ def _is_aggregate_gateway_dcr_challenge_scope( should receive the RFC 9728 401 challenge that advertises the gateway as the authorization server. - Fires only for a genuine 401 on the aggregate scope: any named target - (path or ``x-mcp-servers``) belongs to the per-server challenge paths, and - client-supplied MCP auth headers mean the caller is not a cold-start DCR - client. Fails closed to the original admission error otherwise.""" + Fires only for a genuine 401 on the aggregate scope. Fails closed to the + original admission error otherwise.""" if not _is_litellm_auth_admission_error(exc): return False - if mcp_servers: - return False - if _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers): - return False - return len(MCPRequestHandler._extract_target_server_names_from_path(route)) == 0 + return _is_aggregate_mcp_scope(route, mcp_servers, mcp_auth_header, mcp_server_auth_headers) def _aggregate_gateway_dcr_challenge(request: Request, invalid_token: bool) -> HTTPException: @@ -317,6 +338,20 @@ class MCPRequestHandler: # for a delegated server, so validate it: identity / spend / rate # limits resolve and any stored upstream token can be forwarded. validated_user_api_key_auth = await user_api_key_auth(api_key=litellm_api_key, request=request) + elif ( + oauth2_headers + and is_session_bearer_shaped(oauth2_headers["Authorization"]) + and _is_aggregate_mcp_scope(request_route, mcp_servers, mcp_auth_header, mcp_server_auth_headers) + ): + # A gateway DCR session bearer at the aggregate scope. Placed after the explicit + # x-litellm-api-key arm, which keeps winning, and before the generic Authorization arm, + # which would otherwise feed this bearer to user_api_key_auth as if it were a litellm + # key and 401 on it. + validated_user_api_key_auth = await MCPRequestHandler._admit_gateway_session( + authorization_value=oauth2_headers["Authorization"], + request=request, + route=request_route, + ) elif MCPRequestHandler._target_servers_delegate_auth_to_upstream( path=request_route, mcp_servers=mcp_servers, @@ -626,6 +661,52 @@ class MCPRequestHandler: case _: assert_never(result) + @staticmethod + async def _admit_gateway_session( + authorization_value: str, + request: Request, + route: str, + ) -> UserAPIKeyAuth: + """Admit a gateway DCR session bearer at the aggregate ``/mcp`` scope. + + The bearer proves the user completed SSO when it was minted; it carries no authorization, + so everything that decides what the request may reach is resolved fresh here. The recovered + ``user_id`` reloads the live user record and the same centralized policy gate the key path + uses runs over it, which is what makes deactivating a user, blocking their team, or + exhausting a budget kill outstanding sessions on the next call instead of at expiry. + + No upstream credential is injected, unlike the bridge envelope arm: the custody model vaults + every upstream token server-side, and egress resolves it by user at call time. + + Fails closed with the aggregate challenge rather than a bare 401, so a spec-compliant client + re-runs the authorization flow instead of giving up. A session REFRESH token presented here + is rejected the same way: it is a valid gateway credential, but only ever at the token + endpoint. + """ + from litellm.proxy.proxy_server import master_key + + if not master_key: + raise HTTPException(status_code=500, detail="Server misconfigured: master_key is not set") + + await MCPRequestHandler._run_pre_db_read_auth_checks(request=request, route=route) + + result = resolve_session_bearer( + authorization_value, + session_keys_from_master_key(master_key), + datetime.now(timezone.utc), + ) + match result: + case SessionBearerAdmitted(): + admitted = await MCPRequestHandler._reload_admitted_user( + result.principal.user_id, is_mcp_gateway_session=True + ) + await MCPRequestHandler._enforce_admitted_live_policy(admitted=admitted, request=request, route=route) + return admitted + case SessionBearerInvalid() | NotSessionBearer(): + raise _aggregate_gateway_dcr_challenge(request, invalid_token=True) + case _: + assert_never(result) + @staticmethod async def _run_pre_db_read_auth_checks(request: Request, route: str) -> None: """Run the proxy-wide gates ``user_api_key_auth`` applies before any key lookup: the @@ -665,7 +746,7 @@ class MCPRequestHandler: assert_never(identity.subject_type) @staticmethod - async def _reload_admitted_user(user_id: str) -> UserAPIKeyAuth: + async def _reload_admitted_user(user_id: str, is_mcp_gateway_session: bool = False) -> UserAPIKeyAuth: """Reload the live user an interactively-minted envelope references and admit them as themselves. @@ -726,6 +807,20 @@ class MCPRequestHandler: user_role=user_object.user_role, object_permission=object_permission, object_permission_id=user_object.object_permission_id, + # Bind the org so the org MCP ceiling and the org budget check, both of which return + # early without an org_id, actually apply to an admitted subject. A ceiling only ever + # narrows, so this cannot widen what the subject reaches. + org_id=user_object.organization_id, + # The user-level limits are the ONLY rate limits this principal can bind: every other + # limiter descriptor keys on api_key, team_id, or end_user_id, none of which a keyless + # subject has. Without them _create_rate_limit_descriptors returns an empty list and + # the limiter silently enforces nothing. + user_rpm_limit=user_object.rpm_limit, + user_tpm_limit=user_object.tpm_limit, + user_email=user_object.user_email, + user_max_budget=user_object.max_budget, + user_spend=user_object.spend, + is_mcp_gateway_session=is_mcp_gateway_session, ) @staticmethod diff --git a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py index 90b70dd01f2..e6ada852886 100644 --- a/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py +++ b/litellm/proxy/_experimental/mcp_server/mcp_server_manager.py @@ -2234,8 +2234,17 @@ class MCPServerManager: ) try: - # If admin but NO explicit object permission, get all servers - if user_api_key_auth and _user_has_admin_view(user_api_key_auth) and not has_explicit_object_permission: + # If admin but NO explicit object permission, get all servers. + # A gateway DCR session is excluded: it is a keyless subject that a public OAuth client + # holds, so letting it inherit the role-based view-all short-circuit would hand an + # admin's MCP client every server on the proxy, and the connect grid the user + # authorized servers on would no longer describe what their client can reach. + if ( + user_api_key_auth + and not user_api_key_auth.is_mcp_gateway_session + and _user_has_admin_view(user_api_key_auth) + and not has_explicit_object_permission + ): verbose_logger.debug("Admin user without explicit object_permission - returning all servers") return list(self.get_registry().keys()) diff --git a/litellm/proxy/_types.py b/litellm/proxy/_types.py index 20dc14aa5a3..f6b620cb6d2 100644 --- a/litellm/proxy/_types.py +++ b/litellm/proxy/_types.py @@ -2605,6 +2605,10 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView): # the expected response ob user_max_budget: Optional[float] = None request_route: Optional[str] = None is_session_token: bool = False + # Admitted through the aggregate gateway DCR front door: a keyless subject whose reach comes + # entirely from its own grants. Distinct from is_session_token, which means the dashboard UI + # session and carries budget semantics of its own. + is_mcp_gateway_session: bool = False budget_reservation: Optional[Dict[str, Any]] = Field(default=None, exclude=True) budget_throttle_pct: Optional[float] = Field(default=None, exclude=True) user: Optional[Any] = None # Expanded user object when expand=user is used diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py b/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py index 7b05b8c9dd0..caafeba9543 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py @@ -5069,6 +5069,26 @@ class TestMCPDcrBridgeDelegateAdmission: _MASTER_KEY = "sk-bridge-master-key-for-envelope-derivation" + @staticmethod + def _user_row(**overrides): + """A real user row rather than a MagicMock: the reload reads typed fields off it, and a + mock silently answers every attribute, so assertions against it can pass vacuously.""" + from litellm.proxy._types import LiteLLM_UserTable + + return LiteLLM_UserTable( + **{ + "user_id": "sso-user-7", + "user_role": None, + "max_budget": None, + "spend": 0.0, + "models": [], + "metadata": {"scim_active": True}, + "object_permission": None, + "object_permission_id": None, + **overrides, + } + ) + @staticmethod def _bridge_delegate_server(server_name="bridge_delegate_server", dcr_bridge=True, alias=None): from litellm.types.mcp import MCPAuth @@ -5299,15 +5319,7 @@ class TestMCPDcrBridgeDelegateAdmission: ) as mock_auth, patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager") as mock_mgr, patch("litellm.proxy.proxy_server.master_key", self._MASTER_KEY), - self._patch_user_reload( - return_value=MagicMock( - user_id="sso-user-7", - metadata={"scim_active": True}, - user_role=None, - object_permission=None, - object_permission_id=None, - ) - ) as get_user_object, + self._patch_user_reload(return_value=self._user_row()) as get_user_object, ): mock_mgr.get_mcp_server_by_name.return_value = self._bridge_delegate_server() (auth_result, _h, _s, mcp_server_auth_headers, _o, _r) = await MCPRequestHandler.process_mcp_request(scope) @@ -5342,10 +5354,7 @@ class TestMCPDcrBridgeDelegateAdmission: patch("litellm.proxy._experimental.mcp_server.mcp_server_manager.global_mcp_server_manager") as mock_mgr, patch("litellm.proxy.proxy_server.master_key", self._MASTER_KEY), self._patch_user_reload( - return_value=MagicMock( - user_id="sso-user-7", - metadata={"scim_active": True}, - user_role=None, + return_value=self._user_row( object_permission=object_permission, object_permission_id="op-user-7", ) @@ -6261,3 +6270,193 @@ class TestAggregateGatewayDcrChallenge: with pytest.raises(ProxyException) as exc_info: await MCPRequestHandler.process_mcp_request(self._scope()) assert str(exc_info.value.code) == "500" + + +@pytest.mark.asyncio +class TestGatewayDCRSessionAdmission: + """The aggregate front door's admission arm. + + A gateway DCR session bearer carries identity only, so these tests pin that authorization is + resolved fresh from the live user record on every call, that the arm fails closed with the + challenge a spec client can act on, and that it never fires outside the aggregate scope. + """ + + _MASTER_KEY = "sk-gateway-session-tests" + _AUTH_PATCH_TARGET = "litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.user_api_key_auth" + + def _bearer(self, user_id: str = "user-1", client_id: str = "llm_dcrc_client") -> str: + from datetime import datetime, timezone + + from litellm.proxy._experimental.mcp_server.outbound_credentials.session_credentials import ( + session_keys_from_master_key, + ) + from litellm.proxy._experimental.mcp_server.outbound_credentials.session_token import ( + SessionPrincipal, + mint_session_token, + ) + + minted = mint_session_token( + SessionPrincipal(user_id=user_id, client_id=client_id), + session_keys_from_master_key(self._MASTER_KEY), + datetime.now(timezone.utc), + ) + return minted.token.get_secret_value() + + def _refresh_bearer(self, user_id: str = "user-1") -> str: + from datetime import datetime, timezone + + from litellm.proxy._experimental.mcp_server.outbound_credentials.session_credentials import ( + session_keys_from_master_key, + ) + from litellm.proxy._experimental.mcp_server.outbound_credentials.session_token import ( + SessionPrincipal, + mint_session_refresh_token, + ) + + minted = mint_session_refresh_token( + SessionPrincipal(user_id=user_id, client_id="llm_dcrc_client"), + session_keys_from_master_key(self._MASTER_KEY), + datetime.now(timezone.utc), + ) + return minted.token.get_secret_value() + + def _scope(self, bearer: str, path: str = "/mcp", extra_headers=None): + headers = [(b"authorization", f"Bearer {bearer}".encode())] + headers.extend(extra_headers or []) + return {"type": "http", "method": "POST", "path": path, "headers": headers} + + def _user_row(self, **overrides): + from litellm.proxy._types import LiteLLM_UserTable + + return LiteLLM_UserTable( + **{ + "user_id": "user-1", + "user_role": "internal_user", + "max_budget": None, + "spend": 0.0, + "models": [], + "organization_id": "org-42", + "rpm_limit": 11, + "tpm_limit": 2222, + "user_email": "user@example.com", + **overrides, + } + ) + + @contextlib.contextmanager + def _rig(self, user_row=None): + """Real session crypto, stubbed persistence and policy gate. + + Only the DB read and the shared policy gate are stubbed; the bearer is genuinely minted and + genuinely opened, so a break in the crypto or the arm's wiring surfaces here. + """ + from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler + + async def _no_pre_db_checks(request, route): + return None + + async def _no_policy(admitted, request, route): + return None + + async def _get_user_object(**kwargs): + return user_row if user_row is not None else self._user_row() + + with ( + patch("litellm.proxy.proxy_server.master_key", self._MASTER_KEY), + patch("litellm.proxy.proxy_server.prisma_client", MagicMock()), + patch.object(MCPRequestHandler, "_run_pre_db_read_auth_checks", _no_pre_db_checks), + patch.object(MCPRequestHandler, "_enforce_admitted_live_policy", _no_policy), + patch("litellm.proxy.auth.auth_checks.get_user_object", _get_user_object), + ): + yield + + async def test_a_valid_session_bearer_is_admitted_as_the_sealed_user(self): + with self._rig(): + auth, *_rest = await MCPRequestHandler.process_mcp_request(self._scope(self._bearer())) + + assert auth.user_id == "user-1" + assert auth.is_mcp_gateway_session is True + assert auth.api_key is None + + async def test_the_admitted_subject_binds_its_org_so_the_org_ceiling_applies(self): + """get_allowed_mcp_servers and the org budget check both return early without an org_id, + so an unbound org means the ceiling silently never applies.""" + with self._rig(): + auth, *_rest = await MCPRequestHandler.process_mcp_request(self._scope(self._bearer())) + + assert auth.org_id == "org-42" + + async def test_the_admitted_subject_binds_the_only_rate_limits_it_can(self): + """Every other limiter descriptor keys on api_key, team_id or end_user_id, none of which a + keyless subject has; without the user-level limits the limiter enforces nothing at all.""" + with self._rig(): + auth, *_rest = await MCPRequestHandler.process_mcp_request(self._scope(self._bearer())) + + assert (auth.user_rpm_limit, auth.user_tpm_limit) == (11, 2222) + + async def test_an_expired_or_tampered_bearer_gets_the_reauthorize_challenge(self): + tampered = self._bearer()[:-3] + "AAA" + + with self._rig(): + with pytest.raises(HTTPException) as exc_info: + await MCPRequestHandler.process_mcp_request(self._scope(tampered)) + + assert exc_info.value.status_code == 401 + assert 'error="invalid_token"' in exc_info.value.headers["WWW-Authenticate"] + assert "resource_metadata=" in exc_info.value.headers["WWW-Authenticate"] + + async def test_a_refresh_token_is_never_usable_at_the_tool_call_edge(self): + """A refresh token is a valid gateway credential, but only at the token endpoint. It must + fail closed here rather than fall through to another admission arm.""" + with self._rig(): + with pytest.raises(HTTPException) as exc_info: + await MCPRequestHandler.process_mcp_request(self._scope(self._refresh_bearer())) + + assert exc_info.value.status_code == 401 + assert 'error="invalid_token"' in exc_info.value.headers["WWW-Authenticate"] + + async def test_a_session_bearer_does_not_admit_at_a_named_server_scope(self): + """The session grants the aggregate scope. A named target belongs to the per-server paths, + so the bearer must fall through to normal admission there instead of being honored.""" + called = {} + + async def _record(api_key, request): + called["api_key"] = api_key + return UserAPIKeyAuth(user_id="fell-through") + + with self._rig(), patch(self._AUTH_PATCH_TARGET, side_effect=_record): + auth, *_rest = await MCPRequestHandler.process_mcp_request( + self._scope(self._bearer(), path="/mcp/some_server") + ) + + assert auth.user_id == "fell-through" + assert auth.is_mcp_gateway_session is False + + async def test_an_explicit_litellm_key_still_wins_over_a_session_bearer(self): + async def _key_auth(api_key, request): + return UserAPIKeyAuth(user_id="key-owner", api_key="sk-real") + + with self._rig(), patch(self._AUTH_PATCH_TARGET, side_effect=_key_auth): + auth, *_rest = await MCPRequestHandler.process_mcp_request( + self._scope(self._bearer(), extra_headers=[(b"x-litellm-api-key", b"sk-real")]) + ) + + assert auth.user_id == "key-owner" + assert auth.is_mcp_gateway_session is False + + async def test_a_bearer_signed_under_a_different_master_key_is_rejected(self): + with self._rig(): + with patch("litellm.proxy.proxy_server.master_key", "sk-a-totally-different-master-key"): + with pytest.raises(HTTPException) as exc_info: + await MCPRequestHandler.process_mcp_request(self._scope(self._bearer())) + + assert exc_info.value.status_code == 401 + + async def test_a_scim_deactivated_user_cannot_use_an_outstanding_session(self): + """The bearer is a reference, not an authorization: offboarding must land on the next call + rather than at token expiry.""" + with self._rig(user_row=self._user_row(metadata={"scim_active": False})): + with pytest.raises(HTTPException) as exc_info: + await MCPRequestHandler.process_mcp_request(self._scope(self._bearer())) + + assert exc_info.value.status_code == 401 diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py index a5cb16822cf..015169707e5 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py @@ -9028,3 +9028,68 @@ class TestUrllessIssuerDiscovery: anchored.assert_awaited_once_with("https://idp.example.com", None) resource_rooted.assert_not_awaited() assert built.token_url == "https://idp.example.com/token" + + +@pytest.mark.asyncio +class TestGatewayDCRSessionDoesNotInheritAdminViewAll: + """A gateway DCR session is a keyless subject held by a public OAuth client. + + ``get_allowed_mcp_servers`` short-circuits to the entire registry for a principal with an admin + role and no explicit object permission. That is the dashboard's view-all behavior, and letting a + gateway session inherit it would hand an admin's MCP client every server on the proxy, so the + connect grid the user authorized servers on would stop describing what their client can reach. + """ + + @staticmethod + def _admin(is_mcp_gateway_session: bool): + from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth + + return UserAPIKeyAuth( + user_id="admin-user", + user_role=LitellmUserRoles.PROXY_ADMIN, + is_mcp_gateway_session=is_mcp_gateway_session, + ) + + @staticmethod + def _registry(): + from litellm.types.mcp import MCPAuth + from litellm.types.mcp_server.mcp_server_manager import MCPServer + + return { + server_id: MCPServer( + server_id=server_id, + name=server_id, + server_name=server_id, + transport="http", + auth_type=MCPAuth.none, + allow_all_keys=False, + ) + for server_id in ("granted-server", "other-server") + } + + async def _allowed(self, manager, auth): + from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler + + with patch.object( + MCPRequestHandler, "get_allowed_mcp_servers", new=AsyncMock(return_value=["granted-server"]) + ): + return await manager.get_allowed_mcp_servers(user_api_key_auth=auth) + + async def test_an_admin_dashboard_principal_still_sees_every_server(self): + """The existing view-all behavior must be untouched for everything that is not a session.""" + manager = MCPServerManager() + manager.get_registry = MagicMock(return_value=self._registry()) + + assert sorted(await self._allowed(manager, self._admin(is_mcp_gateway_session=False))) == [ + "granted-server", + "other-server", + ] + + async def test_an_admin_connecting_through_the_gateway_gets_only_their_grants(self): + manager = MCPServerManager() + manager.get_registry = MagicMock(return_value=self._registry()) + + allowed = await self._allowed(manager, self._admin(is_mcp_gateway_session=True)) + + assert allowed == ["granted-server"] + assert "other-server" not in allowed