scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-04 02:31:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 71

fix(mcp): harden owner fingerprint hashing for non-str api keys

Browse source 
_owner_fingerprint_for assumed api_key/user_id supported .encode();
MagicMock-based tests (and any non-str truthy values) crashed with
TypeError before routing. Only hash str/bytes secrets; fall through
otherwise so MCP routing and session tests behave correctly.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Sameer Kankute 2026-05-11 21:06:45 +05:30
parent ea432f2042
commit 342adcd756
No known key found for this signature in database
1 changed files with 21 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
litellm/proxy/_experimental/mcp_server/server.py
View file

@ -2729,24 +2729,34 @@ if MCP_AVAILABLE:
single ``anonymous`` owner and end up able to drive each other's
stateful sessions.
"""
def _bytes_for_hash(value: Any) -> Optional[bytes]:
"""Only hash str/bytes secrets; skip mocks and other unexpected types."""
if value is None:
return None
if isinstance(value, (bytes, bytearray)):
return bytes(value)
if isinstance(value, str):
return value.encode("utf-8")
return None
if user_api_key_auth is not None:
if user_api_key_auth.api_key:
api_key_hash = hashlib.sha256(
user_api_key_auth.api_key.encode("utf-8")
).hexdigest()
key_material = _bytes_for_hash(getattr(user_api_key_auth, "api_key", None))
if key_material:
api_key_hash = hashlib.sha256(key_material).hexdigest()
return f"key:{api_key_hash}"
if user_api_key_auth.user_id:
user_id_hash = hashlib.sha256(
user_api_key_auth.user_id.encode("utf-8")
).hexdigest()
uid_material = _bytes_for_hash(getattr(user_api_key_auth, "user_id", None))
if uid_material:
user_id_hash = hashlib.sha256(uid_material).hexdigest()
return f"user:{user_id_hash}"
if oauth2_headers:
authz = oauth2_headers.get("Authorization") or oauth2_headers.get(
"authorization"
)
if authz:
return f"oauth:{hashlib.sha256(authz.encode('utf-8')).hexdigest()}"
if client_ip:
authz_bytes = _bytes_for_hash(authz)
if authz_bytes:
return f"oauth:{hashlib.sha256(authz_bytes).hexdigest()}"
if client_ip and isinstance(client_ip, str):
return f"ip:{hashlib.sha256(client_ip.encode('utf-8')).hexdigest()}"
return "anonymous"