mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-22 00:31:44 +00:00
fix(mcp): reject scheme-only Basic credentials
This commit is contained in:
parent
13553473aa
commit
8a3add3c6a
2 changed files with 14 additions and 2 deletions
|
|
@ -29,7 +29,8 @@ def _usable_credential_value(auth_type: MCPAuthType, name: str, value: str) -> b
|
|||
if len(parts) != 2 or parts[0].lower() != "basic":
|
||||
return False
|
||||
try:
|
||||
return bool(base64.b64decode(parts[1], validate=True).strip())
|
||||
decoded: Final = base64.b64decode(parts[1], validate=True).strip()
|
||||
return bool(decoded) and decoded.lower() != b"basic"
|
||||
except ValueError:
|
||||
return False
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -13638,10 +13638,21 @@ class TestProtectedCredentialPreparation:
|
|||
assert exc.value.status_code == 500
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("header", ["Basic", "Basic @@@", "Other abc"])
|
||||
@pytest.mark.parametrize("header", ["Basic", "Basic @@@", "Other abc", "Basic QmFzaWM="])
|
||||
async def test_basic_headers_without_usable_credentials_reject(self, header: str) -> None:
|
||||
server = MCPServer(server_id="bad-basic", name="bad-basic", url="https://upstream.example/mcp",
|
||||
transport=MCPTransport.http, auth_type=MCPAuth.basic)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await MCPServerManager()._create_mcp_client(server, extra_headers={"Authorization": header})
|
||||
assert exc.value.status_code == 500
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("value", ["Basic", "Basic ", "basic"])
|
||||
@pytest.mark.parametrize("source", ["configured", "caller"])
|
||||
async def test_basic_scheme_alone_is_not_a_credential(self, value: str, source: str) -> None:
|
||||
server = MCPServer(server_id="basic-scheme", name="basic-scheme", url="https://upstream.example/mcp",
|
||||
transport=MCPTransport.http, auth_type=MCPAuth.basic,
|
||||
authentication_token=value if source == "configured" else None)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await MCPServerManager()._create_mcp_client(server, mcp_auth_header=value if source == "caller" else None)
|
||||
assert exc.value.status_code == 500
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue