fix(proxy): decrypt stored litellm_params before the WIF write gate

This commit is contained in:
mateo-berri 2026-09-05 14:56:33 -07:00
parent 7930c8de77
commit b414494fc8
2 changed files with 69 additions and 1 deletions

View file

@ -339,6 +339,20 @@ def _effective_complexity_router_config(
return existing_params.complexity_router_config
def _decrypted_litellm_params(litellm_params: GenericLiteLLMParams) -> Mapping[str, object]:
dumped: Final[Mapping[str, object]] = litellm_params.model_dump(exclude_none=True)
return MappingProxyType(
{
name: (
decrypt_value_helper(value=value, key=name, exception_type="debug", return_original_value=True)
if isinstance(value, str)
else value
)
for name, value in dumped.items()
}
)
def _effective_model(
incoming_params: GenericLiteLLMParams | None, existing_params: GenericLiteLLMParams | None
) -> str | None:
@ -1725,7 +1739,7 @@ class ModelManagementAuthChecks:
) -> None:
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return
stored: Final = model_params.litellm_params.model_dump(exclude_none=True)
stored: Final = _decrypted_litellm_params(model_params.litellm_params)
wif_fields: Final = await effective_server_owned_wif_fields(stored, incoming_params, prisma_client)
if wif_fields:
# ProxyException rather than HTTPException so the offending field stays a structured

View file

@ -5835,6 +5835,60 @@ class TestWifBoundaryReadsTheResultingDeployment:
user_api_key_dict=non_admin,
)
@pytest.mark.asyncio
async def test_non_admin_cannot_modify_a_deployment_whose_stored_credential_name_is_encrypted(self, monkeypatch):
"""Rows written through /model/new hold every litellm_params value encrypted, so a gate that
looks the stored credential name up as written asks about a ciphertext, finds no such
credential, and lets the write through."""
from litellm.proxy.management_endpoints.model_management_endpoints import patch_model
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-1234")
non_admin = UserAPIKeyAuth(user_id="team_admin", user_role=LitellmUserRoles.INTERNAL_USER)
federated_row = MagicMock()
federated_row.litellm_params = {
"model": encrypt_value_helper(value="anthropic/claude-sonnet-4"),
"litellm_credential_name": encrypt_value_helper(value="admin-wif"),
}
assert federated_row.litellm_params["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 = {
"credential_name": "admin-wif",
"credential_values": {
"anthropic_federation_rule_id": "fdrl_admin",
"anthropic_organization_id": "org-admin",
},
"credential_info": {"custom_llm_provider": "anthropic"},
}
def credential_by_exact_name(**kwargs):
return admin_credential_row if kwargs["where"].get("credential_name") == "admin-wif" else None
mock_prisma = MagicMock()
mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=federated_row)
mock_prisma.db.litellm_credentialstable.find_unique = AsyncMock(side_effect=credential_by_exact_name)
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 workload identity"
):
await patch_model(
model_id="m1",
patch_data=updateDeployment(litellm_params=updateLiteLLMParams(rpm=5)),
user_api_key_dict=non_admin,
)
mock_prisma.db.litellm_proxymodeltable.update.assert_not_called()
class TestEnforceRpmTpmOnModelAdd:
def test_passes_when_disabled_even_without_limits(self):