diff --git a/litellm/proxy/common_utils/credential_hydration.py b/litellm/proxy/common_utils/credential_hydration.py index 446fb8544d7..a9579e48d15 100644 --- a/litellm/proxy/common_utils/credential_hydration.py +++ b/litellm/proxy/common_utils/credential_hydration.py @@ -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))) diff --git a/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py index 991cce82578..b21c010b2d0 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py +++ b/tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py @@ -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