From d3e3fdad7fc22bf101b4d4a7dbc9cd4f29ef5755 Mon Sep 17 00:00:00 2001 From: derhornspieler <15236687+derhornspieler@users.noreply.github.com> Date: Tue, 25 Aug 2026 10:41:57 -0400 Subject: [PATCH] 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. --- .../common_utils/credential_hydration.py | 28 ++++++---- .../test_model_management_endpoints.py | 53 +++++++++++++++++++ 2 files changed, 71 insertions(+), 10 deletions(-) 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