fix(proxy): count the credential a deployment already carries, not only the one a write names

A deployment federated through a named credential holds no federation field of its
own, and the gate resolved only the name the write supplied. So a patch that cleared
litellm_credential_name, alongside an api_base or api_key of the caller's choosing,
left nothing federated to find and the write was allowed. Clearing was the way out of
a rule meant to have no way out.

Both names count now, the one already on the deployment and the one being written,
since detaching an administrator's federated credential is itself an administrator's
action.
This commit is contained in:
derhornspieler 2026-08-25 10:41:57 -04:00
parent f3b2a461c1
commit d3e3fdad7f
2 changed files with 71 additions and 10 deletions

View file

@ -6,6 +6,7 @@ in-memory list has not yet picked up a credential another pod just wrote or upda
"""
from collections.abc import Mapping
from itertools import chain
from types import MappingProxyType
from typing import Final
@ -117,19 +118,26 @@ async def effective_anthropic_wif_fields(
"""
from_stored: Final = () if stored is None else anthropic_wif_fields_present(stored)
from_incoming: Final = () if incoming is None else anthropic_wif_fields_named(incoming.model_fields_set)
credential_name: Final = _effective_credential_name(stored, incoming)
from_credential: Final = (
() if credential_name is None else await named_credential_wif_fields(credential_name, prisma_client)
)
per_credential: Final = [
await named_credential_wif_fields(credential_name, prisma_client)
for credential_name in _effective_credential_names(stored, incoming)
]
from_credential: Final = tuple(chain.from_iterable(per_credential))
return tuple(dict.fromkeys(from_stored + from_incoming + from_credential))
def _effective_credential_name(
def _effective_credential_names(
stored: Mapping[str, object] | None,
incoming: GenericLiteLLMParams | None,
) -> str | None:
if incoming is not None and "litellm_credential_name" in incoming.model_fields_set:
named: Final = incoming.litellm_credential_name
return named if isinstance(named, str) else None
) -> tuple[str, ...]:
"""Both the credential the deployment already carries and the one this write names.
Taking only the incoming name would let a write clear its way out: detaching a federated
credential, by sending ``litellm_credential_name: null`` alongside an api_key or api_base of
the caller's choosing, would leave nothing federated to find and the write would be allowed.
Detaching an administrator's federated credential is itself an administrator's action, so the
stored name counts whatever the write says.
"""
from_stored: Final = None if stored is None else stored.get("litellm_credential_name")
return from_stored if isinstance(from_stored, str) else None
from_incoming: Final = None if incoming is None else incoming.litellm_credential_name
return tuple(dict.fromkeys(name for name in (from_stored, from_incoming) if isinstance(name, str)))

View file

@ -4856,6 +4856,59 @@ class TestWifBoundaryReadsTheResultingDeployment:
user_api_key_dict=non_admin,
)
@pytest.mark.asyncio
async def test_non_admin_cannot_detach_a_federated_credential_to_escape_the_gate(self):
"""Clearing the credential name must not be the way out. A deployment federated through a
named credential carries no federation field of its own, so a patch that sends
litellm_credential_name: null alongside an api_base of the caller's choosing would leave
nothing federated to find, and the write would be allowed."""
from litellm.proxy.management_endpoints.model_management_endpoints import patch_model
non_admin = UserAPIKeyAuth(user_id="team_admin", user_role=LitellmUserRoles.INTERNAL_USER)
federated_row = MagicMock()
federated_row.litellm_params = {
"model": "anthropic/claude-sonnet-4",
"litellm_credential_name": "admin-wif",
}
federated_row.model_dump.return_value = {
"model_name": "claude",
"litellm_params": federated_row.litellm_params,
"model_info": {"id": "m1"},
}
admin_credential_row = MagicMock()
admin_credential_row.dict.return_value = {
"credential_name": "admin-wif",
"credential_values": {
"anthropic_federation_rule_id": "fdrl_admin",
"anthropic_organization_id": "org-admin",
},
"credential_info": {"custom_llm_provider": "anthropic"},
}
mock_prisma = MagicMock()
mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=federated_row)
mock_prisma.db.litellm_credentialstable.find_unique = AsyncMock(return_value=admin_credential_row)
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma), # test-quality-ok: proxy wiring under test
patch( # test-quality-ok: proxy wiring under test
"litellm.proxy.proxy_server.llm_router", MagicMock(**{"get_model_ids.return_value": ["m1"]})
),
patch("litellm.proxy.proxy_server.store_model_in_db", True), # test-quality-ok: proxy wiring under test
patch("litellm.proxy.proxy_server.premium_user", True), # test-quality-ok: proxy wiring under test
):
with pytest.raises(Exception, match="Only proxy admins can modify a deployment configured for Anthropic"):
await patch_model(
model_id="m1",
patch_data=updateDeployment(
litellm_params=updateLiteLLMParams(
litellm_credential_name=None, api_base="https://gateway.internal"
)
),
user_api_key_dict=non_admin,
)
@pytest.mark.asyncio
async def test_non_admin_cannot_attach_a_federated_credential_by_name(self):
"""litellm_credential_name names no federation field itself, but request-time hydration