scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-03 02:22:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 80

fix(mcp): key discovery cache by the hashed token

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-27 04:13:19 +00:00
parent 25797cbf2a
commit b6f8480812
2 changed files with 16 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -4663,16 +4663,14 @@ class MCPServerManager:
if not (per_user or mcp_auth_header or extra_headers or stdio_env or subject_token):
return server.server_id, None
identity: Final = (
(user_api_key_auth.user_id, user_api_key_auth.api_key)
if per_user and user_api_key_auth is not None
else None
(user_api_key_auth.user_id, user_api_key_auth.token) if per_user and user_api_key_auth is not None else None
)
material: Final = json.dumps(
(identity, mcp_auth_header, extra_headers, stdio_env, subject_token, credential_fingerprint),
sort_keys=True,
separators=(",", ":"),
)
return server.server_id, hashlib.sha256(material.encode()).hexdigest()
return server.server_id, hashlib.sha256(material.encode(), usedforsecurity=False).hexdigest()
async def get_prompts_from_server(
self,

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

@ -14376,6 +14376,20 @@ def test_discovery_cache_keys_isolate_user_dependent_auth(auth_type: MCPAuth) ->
assert "first" not in str(first)
assert "second" not in str(second)
from litellm.proxy._types import hash_token
same_user_other_token: Final = manager._discovery_key(
server, UserAPIKeyAuth(user_id="first", token=hash_token("sk-second")), None, None, None, None
)
same_token_no_key: Final = manager._discovery_key(
server, UserAPIKeyAuth(user_id="first", token=hash_token("sk-first")), None, None, None, None
)
with_key: Final = manager._discovery_key(
server, UserAPIKeyAuth(user_id="first", api_key="sk-first"), None, None, None, None
)
assert same_user_other_token != with_key
assert same_token_no_key == with_key
@pytest.mark.asyncio
async def test_discovery_cache_retries_cancelled_fetches() -> None: