scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-25 01:02:15 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 80

fix(mcp): keep oauth scopes in admin api credential redaction

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
jesus 2026-09-04 20:51:41 +00:00 • committed by Yucheng He
parent 59c24abcbe
commit 25c8926c48
2 changed files with 98 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

34
litellm/proxy/management_endpoints/mcp_management_endpoints.py
View file

@ -22,7 +22,14 @@ import os
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Annotated, Final, Literal, Protocol
from typing import (
TYPE_CHECKING,
Annotated,
Final,
Literal,
Protocol,
cast, # noqa: TID251 # validated JSON values need explicit narrowing
)
from fastapi import (
APIRouter,
@ -628,8 +635,8 @@ if MCP_AVAILABLE:
def _preserved_admin_config_credentials(
credentials: "MCPCredentials | str | None",
) -> "dict[str, str] | None":
"""Keep only the non-secret admin-config keys, which are stored unencrypted so they lift out
) -> "dict[str, str | list[str]] | None":
"""Keep non-secret admin-config keys and scopes, which are stored unencrypted so they lift out
as plaintext; every secret and minted-token key is dropped.
Total over every stored shape: a dict is read directly, a JSON-object string is parsed, and
@ -639,15 +646,26 @@ if MCP_AVAILABLE:
parsed: object = credentials
if isinstance(credentials, str):
try:
parsed = json.loads(credentials)
parsed = cast(object, json.loads(credentials))
except (ValueError, TypeError):
return None
if not isinstance(parsed, dict):
return None
preserved: Final = {
key: value
for key in MCP_ADMIN_CONFIG_CREDENTIAL_KEYS
if isinstance((value := parsed.get(key)), str) and value
parsed_credentials: Final = cast(Mapping[str, object], parsed)
scopes: Final[object] = parsed_credentials.get("scopes")
scopes_as_objects: Final[list[object]] = cast(list[object], scopes) if isinstance(scopes, list) else []
preserved_scopes: Final[dict[str, list[str]]] = (
{"scopes": cast(list[str], scopes_as_objects)}
if scopes_as_objects and all(isinstance(scope, str) and scope for scope in scopes_as_objects)
else {}
)
preserved: Final[dict[str, str | list[str]]] = {
**{
key: value
for key in MCP_ADMIN_CONFIG_CREDENTIAL_KEYS
if isinstance((value := parsed_credentials.get(key)), str) and value
},
**preserved_scopes,
}
return preserved or None

75
tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py
View file

@ -6,7 +6,7 @@ import logging
from contextlib import ExitStack
from datetime import datetime, timedelta
from types import SimpleNamespace
from typing import List, Optional
from typing import List, Optional, cast
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -29,7 +29,7 @@ from litellm.proxy._types import (
UpdateMCPServerRequest,
UserAPIKeyAuth,
)
from litellm.types.mcp import MCPAuth
from litellm.types.mcp import MCPAuth, MCPCredentials
from litellm.types.mcp_server.mcp_server_manager import MCPServer
@ -864,6 +864,75 @@ class TestListMCPServers:
assert result.credentials == expected
@pytest.mark.parametrize(
"stored_credentials, expected",
[
(
{
"client_id": "cid",
"client_secret": "csecret",
"scopes": ["read", "write"],
"upstream_token_header": "esb-oauth",
},
{"scopes": ["read", "write"], "upstream_token_header": "esb-oauth"},
),
(
'{"client_id": "cid", "client_secret": "csecret", "scopes": ["read", "write"], '
'"upstream_token_header": "esb-oauth"}',
{"scopes": ["read", "write"], "upstream_token_header": "esb-oauth"},
),
(
{"client_id": "cid", "client_secret": "csecret", "scopes": ["read", ""]},
None,
),
(
{"client_id": "cid", "client_secret": "csecret", "scopes": "read"},
None,
),
],
)
@pytest.mark.asyncio
async def test_fetch_single_mcp_server_preserves_valid_oauth_scopes(
self, stored_credentials: object, expected: object
):
mock_server = generate_mock_mcp_server_db_record(server_id="server-scopes", alias="Scopes")
mock_server.credentials = cast(MCPCredentials, stored_credentials)
mock_health_result = generate_mock_mcp_server_db_record(server_id="server-scopes", alias="Scopes")
mock_health_result.status = "healthy"
mock_health_result.last_health_check = datetime.now()
mock_health_result.health_check_error = None
mock_user_auth = generate_mock_user_api_key_auth(user_role=LitellmUserRoles.PROXY_ADMIN)
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=MagicMock(),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
AsyncMock(return_value=mock_server),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.global_mcp_server_manager.health_check_server",
AsyncMock(return_value=mock_health_result),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints._user_has_admin_view",
return_value=True,
),
):
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
fetch_mcp_server,
)
result = await fetch_mcp_server(
request=_make_mock_request(),
server_id="server-scopes",
user_api_key_dict=mock_user_auth,
)
assert result.credentials == expected
@pytest.mark.asyncio
async def test_fetch_single_mcp_server_strips_upstream_resource_for_non_admin(self):
"""A non-full-admin viewer gets the whole blob nulled, including the non-secret admin config,
@ -2339,7 +2408,7 @@ class TestTemporaryMCPSessionEndpoints:
"client_secret": "client-secret",
"scopes": ["scope1"],
}
assert response.credentials is None
assert response.credentials == {"scopes": ["scope1"]}
@pytest.mark.asyncio
async def test_add_session_mcp_server_rejects_non_admins(self):