scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 123

feat(mcp): resolve team/key MCP permissions by server name or alias

Browse source 
team.object_permission.mcp_servers (and the per-key equivalent) previously
only accepted server_id strings. For config-loaded MCP servers, the id is
derived from a hash that includes the server URL, so the same logical
server in two regions ends up with two different ids in a shared database.
Permission lists had to enumerate every region's id.

Add a single MCPServerManager.expand_permission_list() helper that resolves
each entry against the current region's config + DB registry union: entries
that match a server_id pass through, entries that match an alias/server_name/
name expand to every matching id, and unresolved entries drop with a debug
log so stale or typo entries are diagnosable. Wire it into the four
_get_allowed_mcp_servers_for_* helpers so direct server entries and
mcp_tool_permissions dict keys are both expanded before the intersection.

Access-check outcomes are unchanged for existing id-based permissions;
name-based entries now resolve instead of being silently denied.
This commit is contained in:
Ryan Crabbe 2026-04-23 11:13:30 -07:00
parent e9e86ed956
commit 85f9c5e83f
No known key found for this signature in database
4 changed files with 339 additions and 98 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py
View file

@ -643,8 +643,14 @@ class MCPRequestHandler:
if key_object_permission is None:
return []
# Get direct MCP servers
direct_mcp_servers = key_object_permission.mcp_servers or []
# Permission entries may be server_ids OR names/aliases — expand to ids.
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
key_object_permission.mcp_servers or []
)
# Get MCP servers from access groups
access_group_servers = (
@ -654,8 +660,8 @@ class MCPRequestHandler:
)
# servers referenced in tool permissions should also be accessible
tool_perm_servers = list(
(key_object_permission.mcp_tool_permissions or {}).keys()
tool_perm_servers = global_mcp_server_manager.expand_permission_list(
list((key_object_permission.mcp_tool_permissions or {}).keys())
)
# Combine all lists
@ -685,8 +691,14 @@ class MCPRequestHandler:
if object_permissions is None:
return []
# Get direct MCP servers
direct_mcp_servers = object_permissions.mcp_servers or []
# Permission entries may be server_ids OR names/aliases — expand to ids.
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
object_permissions.mcp_servers or []
)
# Get MCP servers from access groups
access_group_servers = (
@ -696,8 +708,8 @@ class MCPRequestHandler:
)
# servers referenced in tool permissions should also be accessible
tool_perm_servers = list(
(object_permissions.mcp_tool_permissions or {}).keys()
tool_perm_servers = global_mcp_server_manager.expand_permission_list(
list((object_permissions.mcp_tool_permissions or {}).keys())
)
# Combine all lists
@ -746,8 +758,14 @@ class MCPRequestHandler:
if end_user_obj is None or end_user_obj.object_permission is None:
return []
# Get direct MCP servers
direct_mcp_servers = end_user_obj.object_permission.mcp_servers or []
# Permission entries may be server_ids OR names/aliases — expand to ids.
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
direct_mcp_servers = global_mcp_server_manager.expand_permission_list(
end_user_obj.object_permission.mcp_servers or []
)
# Get MCP servers from access groups
access_group_servers = (
@ -757,8 +775,8 @@ class MCPRequestHandler:
)
# servers referenced in tool permissions should also be accessible
tool_perm_servers = list(
(end_user_obj.object_permission.mcp_tool_permissions or {}).keys()
tool_perm_servers = global_mcp_server_manager.expand_permission_list(
list((end_user_obj.object_permission.mcp_tool_permissions or {}).keys())
)
# Combine all lists
@ -836,12 +854,21 @@ class MCPRequestHandler:
if isinstance(mcp_access_groups, str):
mcp_access_groups = []
# Permission entries may be server_ids OR names/aliases — expand to ids.
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
expanded_direct_servers = global_mcp_server_manager.expand_permission_list(
list(direct_mcp_servers)
)
access_group_servers = (
await MCPRequestHandler._get_mcp_servers_from_access_groups(
mcp_access_groups
)
)
all_servers = list(direct_mcp_servers) + access_group_servers
all_servers = expanded_direct_servers + access_group_servers
return list(set(all_servers))
except Exception as e:
verbose_logger.warning(

36
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
View file

@ -2765,6 +2765,42 @@ class MCPServerManager:
servers.append(server)
return servers
def expand_permission_list(self, identifiers: List[str]) -> List[str]:
"""
Expand a permission list of server_ids/names/aliases into concrete
server_ids against the current region's config + DB registry union.
Entries that match a server_id pass through unchanged. Entries that
don't are looked up by alias/server_name/name and replaced with every
matching server_id (duplicate names grant access to all matches).
Entries that resolve to nothing are dropped and a debug log is
emitted so admins can diagnose stale/typo permission entries — the
final access-check still denies them either way.
"""
if not identifiers:
return []
registry = self.get_registry()
expanded: Set[str] = set()
for identifier in identifiers:
if identifier in registry:
expanded.add(identifier)
continue
matches: List[str] = [
server_id
for server_id, server in registry.items()
if server.alias == identifier
or server.server_name == identifier
or server.name == identifier
]
if matches:
expanded.update(matches)
else:
verbose_logger.debug(
f"MCP permission entry '{identifier}' does not resolve "
"to any known server (config + DB union). Skipping."
)
return list(expanded)
def get_mcp_server_by_name(
self, server_name: str, client_ip: Optional[str] = None
) -> Optional[MCPServer]:

224
tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
View file

@ -1391,55 +1391,73 @@ async def test_get_allowed_mcp_servers_for_team_uses_helper():
Test that _get_allowed_mcp_servers_for_team properly uses _get_team_object_permission
helper which handles both loaded and unloaded object_permission cases.
"""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
from litellm.proxy._types import LiteLLM_ObjectPermissionTable
from litellm.types.mcp import MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPServer
# Create mock object permission with servers and access groups
mock_object_permission = LiteLLM_ObjectPermissionTable(
object_permission_id="perm-789",
mcp_servers=["direct-server1", "direct-server2"],
mcp_access_groups=["dev-group"],
vector_stores=[],
)
# Register placeholder ids in the manager so expand_permission_list resolves them.
for sid in ("direct-server1", "direct-server2"):
global_mcp_server_manager.registry[sid] = MCPServer(
server_id=sid,
name=sid,
server_name=sid,
url=f"https://{sid}.example.com",
transport=MCPTransport.http,
)
try:
# Create mock object permission with servers and access groups
mock_object_permission = LiteLLM_ObjectPermissionTable(
object_permission_id="perm-789",
mcp_servers=["direct-server1", "direct-server2"],
mcp_access_groups=["dev-group"],
vector_stores=[],
)
# Create mock user auth
mock_user_auth = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
team_id="team-789",
)
# Create mock user auth
mock_user_auth = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
team_id="team-789",
)
# Mock the helper methods
with patch.object(
MCPRequestHandler, "_get_team_object_permission"
) as mock_get_team_perm:
# Mock the helper methods
with patch.object(
MCPRequestHandler, "_get_mcp_servers_from_access_groups"
) as mock_get_access_group_servers:
# Configure mocks
mock_get_team_perm.return_value = mock_object_permission
mock_get_access_group_servers.return_value = [
"group-server1",
"group-server2",
]
MCPRequestHandler, "_get_team_object_permission"
) as mock_get_team_perm:
with patch.object(
MCPRequestHandler, "_get_mcp_servers_from_access_groups"
) as mock_get_access_group_servers:
# Configure mocks
mock_get_team_perm.return_value = mock_object_permission
mock_get_access_group_servers.return_value = [
"group-server1",
"group-server2",
]
# Call the method
result = await MCPRequestHandler._get_allowed_mcp_servers_for_team(
mock_user_auth
)
# Call the method
result = await MCPRequestHandler._get_allowed_mcp_servers_for_team(
mock_user_auth
)
# Assert the result contains both direct and access group servers
assert set(result) == {
"direct-server1",
"direct-server2",
"group-server1",
"group-server2",
}
# Assert the result contains both direct and access group servers
assert set(result) == {
"direct-server1",
"direct-server2",
"group-server1",
"group-server2",
}
# Verify _get_team_object_permission was called (the helper we fixed)
mock_get_team_perm.assert_called_once_with(mock_user_auth)
# Verify _get_team_object_permission was called (the helper we fixed)
mock_get_team_perm.assert_called_once_with(mock_user_auth)
# Verify access groups were resolved
mock_get_access_group_servers.assert_called_once_with(["dev-group"])
# Verify access groups were resolved
mock_get_access_group_servers.assert_called_once_with(["dev-group"])
finally:
for sid in ("direct-server1", "direct-server2"):
global_mcp_server_manager.registry.pop(sid, None)
@pytest.mark.asyncio
@ -1569,35 +1587,53 @@ async def test_get_allowed_mcp_servers_for_key_returns_empty_when_db_returns_non
async def test_get_allowed_mcp_servers_for_key_prefers_in_memory_permission():
"""Ensure in-memory object_permission is used without hitting the DB."""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
from litellm.proxy._types import LiteLLM_ObjectPermissionTable
from litellm.types.mcp import MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPServer
perms = LiteLLM_ObjectPermissionTable(
object_permission_id="perm-in-memory",
mcp_servers=["direct-server"],
mcp_access_groups=["grp-alpha"],
)
user_api_key_auth = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
object_permission=perms,
# Register "direct-server" in the manager so permission expansion resolves it.
# Without this, expand_permission_list drops unknown ids as stale — which is
# the correct production behavior but unrelated to what this test asserts.
global_mcp_server_manager.registry["direct-server"] = MCPServer(
server_id="direct-server",
name="direct-server",
server_name="direct-server",
url="https://direct-server.example.com",
transport=MCPTransport.http,
)
try:
perms = LiteLLM_ObjectPermissionTable(
object_permission_id="perm-in-memory",
mcp_servers=["direct-server"],
mcp_access_groups=["grp-alpha"],
)
user_api_key_auth = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
object_permission=perms,
)
with patch(
"litellm.proxy.auth.auth_checks.get_object_permission",
new_callable=AsyncMock,
) as mock_get_perm:
with patch.object(
MCPRequestHandler, "_get_mcp_servers_from_access_groups"
) as mock_access_groups:
mock_access_groups.return_value = ["group-server"]
with patch(
"litellm.proxy.auth.auth_checks.get_object_permission",
new_callable=AsyncMock,
) as mock_get_perm:
with patch.object(
MCPRequestHandler, "_get_mcp_servers_from_access_groups"
) as mock_access_groups:
mock_access_groups.return_value = ["group-server"]
result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
user_api_key_auth
)
result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
user_api_key_auth
)
assert set(result) == {"direct-server", "group-server"}
mock_get_perm.assert_not_called()
mock_access_groups.assert_called_once_with(["grp-alpha"])
assert set(result) == {"direct-server", "group-server"}
mock_get_perm.assert_not_called()
mock_access_groups.assert_called_once_with(["grp-alpha"])
finally:
global_mcp_server_manager.registry.pop("direct-server", None)
@pytest.mark.asyncio
@ -1753,28 +1789,46 @@ async def test_tool_permission_servers_included_in_allowed_servers():
Regression test for https://github.com/BerriAI/litellm/issues/21954
"""
perm = MagicMock()
perm.mcp_servers = []
perm.mcp_access_groups = []
perm.mcp_tool_permissions = {"server_id_123": ["tool_a", "tool_b"]}
user_api_key_auth = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
global_mcp_server_manager,
)
from litellm.types.mcp import MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPServer
with (
patch.object(
MCPRequestHandler, "_get_key_object_permission", return_value=perm
),
patch.object(
MCPRequestHandler,
"_get_mcp_servers_from_access_groups",
new_callable=AsyncMock,
return_value=[],
),
):
result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
user_api_key_auth=user_api_key_auth,
# Register the server id so expand_permission_list resolves it rather than
# dropping it as stale.
global_mcp_server_manager.registry["server_id_123"] = MCPServer(
server_id="server_id_123",
name="server_id_123",
server_name="server_id_123",
url="https://server-id-123.example.com",
transport=MCPTransport.http,
)
try:
perm = MagicMock()
perm.mcp_servers = []
perm.mcp_access_groups = []
perm.mcp_tool_permissions = {"server_id_123": ["tool_a", "tool_b"]}
user_api_key_auth = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
)
assert "server_id_123" in result
with (
patch.object(
MCPRequestHandler, "_get_key_object_permission", return_value=perm
),
patch.object(
MCPRequestHandler,
"_get_mcp_servers_from_access_groups",
new_callable=AsyncMock,
return_value=[],
),
):
result = await MCPRequestHandler._get_allowed_mcp_servers_for_key(
user_api_key_auth=user_api_key_auth,
)
assert "server_id_123" in result
finally:
global_mcp_server_manager.registry.pop("server_id_123", None)

124
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
View file

@ -2610,5 +2610,129 @@ class TestMCPServerManagerUpstreamInstructionsCache:
assert "srv_b" in by_name and by_name["srv_b"].instructions is None
class TestMCPServerManagerExpandPermissionList:
"""Tests for the alias/name-aware permission list expansion used by team MCP permissions."""
def _make_server(
self,
server_id: str,
server_name: str,
alias=None,
name=None,
) -> MCPServer:
return MCPServer(
server_id=server_id,
name=name if name is not None else (alias or server_name),
alias=alias,
server_name=server_name,
url=f"https://{server_id}.example.com",
transport=MCPTransport.http,
)
def test_empty_list_returns_empty(self):
manager = MCPServerManager()
assert manager.expand_permission_list([]) == []
def test_expands_server_name(self):
manager = MCPServerManager()
manager.config_mcp_servers["id-usw1"] = self._make_server(
"id-usw1", server_name="a"
)
assert manager.expand_permission_list(["a"]) == ["id-usw1"]
def test_expands_alias(self):
manager = MCPServerManager()
manager.config_mcp_servers["id-1"] = self._make_server(
"id-1", server_name="internal_name", alias="public_alias"
)
assert manager.expand_permission_list(["public_alias"]) == ["id-1"]
def test_drops_unknown_entry_silently(self):
manager = MCPServerManager()
manager.config_mcp_servers["id-1"] = self._make_server("id-1", server_name="b")
assert manager.expand_permission_list(["a"]) == []
def test_name_collision_expands_to_all_matches(self):
"""Two servers sharing a server_name both resolve — the documented behavior."""
manager = MCPServerManager()
manager.config_mcp_servers["id-config"] = self._make_server(
"id-config", server_name="shared"
)
manager.registry["id-db"] = self._make_server("id-db", server_name="shared")
assert sorted(manager.expand_permission_list(["shared"])) == [
"id-config",
"id-db",
]
def test_searches_config_and_registry_union(self):
manager = MCPServerManager()
manager.config_mcp_servers["cfg-id"] = self._make_server("cfg-id", server_name="a")
manager.registry["reg-id"] = self._make_server("reg-id", server_name="b")
assert manager.expand_permission_list(["a"]) == ["cfg-id"]
assert manager.expand_permission_list(["b"]) == ["reg-id"]
def test_id_match_takes_precedence_over_name_match(self):
"""
If a permission entry matches a server_id directly, don't also add
servers whose server_name happens to equal that id.
"""
manager = MCPServerManager()
manager.config_mcp_servers["id-1"] = self._make_server(
"id-1", server_name="other_name"
)
manager.config_mcp_servers["id-2"] = self._make_server(
"id-2", server_name="id-1"
)
assert manager.expand_permission_list(["id-1"]) == ["id-1"]
def test_mixed_ids_and_names_in_same_list(self):
manager = MCPServerManager()
manager.config_mcp_servers["uuid-1"] = self._make_server(
"uuid-1", server_name="a"
)
manager.config_mcp_servers["uuid-2"] = self._make_server(
"uuid-2", server_name="b"
)
# ["uuid-1", "b"] -> uuid-1 passes through, "b" resolves to uuid-2
assert sorted(manager.expand_permission_list(["uuid-1", "b"])) == [
"uuid-1",
"uuid-2",
]
def test_deduplicates_overlapping_id_and_name_entries(self):
"""If a list references the same server by both id and name, return it once."""
manager = MCPServerManager()
manager.config_mcp_servers["uuid-1"] = self._make_server(
"uuid-1", server_name="a"
)
assert manager.expand_permission_list(["uuid-1", "a"]) == ["uuid-1"]
def test_simulates_cross_region_portability(self):
"""
Same permission entry "a" resolves to different concrete IDs per region —
the cross-region portability the customer is asking for.
"""
usw1 = MCPServerManager()
usw1.config_mcp_servers["hash-usw1"] = self._make_server(
"hash-usw1", server_name="a"
)
usc1 = MCPServerManager()
usc1.config_mcp_servers["hash-usc1"] = self._make_server(
"hash-usc1", server_name="a"
)
assert usw1.expand_permission_list(["a"]) == ["hash-usw1"]
assert usc1.expand_permission_list(["a"]) == ["hash-usc1"]
if __name__ == "__main__":
pytest.main([__file__])