fix(proxy): validate model credential name only when it changes (#42701)

PATCH /model/{id}/update rejected read-modify-write edits that resent an unchanged but dangling litellm_credential_name. Existence validation now runs only when the requested name differs from the stored one; empty string and non-admin detach rejections are unchanged

Co-authored-by: yuneng <yuneng@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/management_endpoints/model_management_endpoints.py

@@ -340,8 +340,21 @@ def _raise_on_strategy_router_write_violation(
     )
 
 
+def _stored_credential_name(existing_litellm_params: GenericLiteLLMParams | None) -> str | None:
+    if existing_litellm_params is None or existing_litellm_params.litellm_credential_name is None:
+        return None
+    return decrypt_value_helper(
+        value=existing_litellm_params.litellm_credential_name,
+        key="litellm_credential_name",
+        exception_type="debug",
+        return_original_value=True,
+    )
+
+
 async def _raise_on_invalid_credential_name(
-    litellm_params: updateLiteLLMParams | None, prisma_client: PrismaClient
+    litellm_params: updateLiteLLMParams | None,
+    existing_litellm_params: GenericLiteLLMParams | None,
+    prisma_client: PrismaClient,
 ) -> None:
     if litellm_params is None or "litellm_credential_name" not in litellm_params.model_fields_set:
         return
@@ -355,6 +368,8 @@ async def _raise_on_invalid_credential_name(
             code=status.HTTP_400_BAD_REQUEST,
             param="litellm_credential_name",
         )
+    if credential_name == _stored_credential_name(existing_litellm_params):
+        return
     if CredentialAccessor.find_credential(credential_name) is not None:
         return
     stored_credential: Final = await CredentialsRepository(WriterPinnedClient(prisma_client.db)).find_by_name(
@@ -1192,7 +1207,7 @@ async def patch_model(
             existing_litellm_params=db_model.litellm_params,
             null_detaches=True,
         )
-        await _raise_on_invalid_credential_name(patch_data.litellm_params, prisma_client)
+        await _raise_on_invalid_credential_name(patch_data.litellm_params, db_model.litellm_params, prisma_client)
 
         ModelManagementAuthChecks.can_user_set_aws_session_tags(
             litellm_params=patch_data.litellm_params,
@@ -2012,18 +2027,8 @@ class ModelManagementAuthChecks:
             return True
         if litellm_params.litellm_credential_name is None and not null_detaches:
             return True
-        existing_credential_name: Final = (
-            decrypt_value_helper(
-                value=existing_litellm_params.litellm_credential_name,
-                key="litellm_credential_name",
-                exception_type="debug",
-                return_original_value=True,
-            )
-            if existing_litellm_params is not None and existing_litellm_params.litellm_credential_name is not None
-            else None
-        )
         requested_credential_name: Final = litellm_params.litellm_credential_name
-        if requested_credential_name == existing_credential_name:
+        if requested_credential_name == _stored_credential_name(existing_litellm_params):
             return True
         if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
             return True

tests/integration/contracts.json

@@ -33,6 +33,16 @@
     "tests/integration/management/test_key_updates.py::test_update_preserves_independent_fields_and_serving": [
       "mgmt.key.update.preserves_independent_fields"
     ],
+    "tests/integration/management/test_model_credential_name_updates.py::test_unrelated_patch_succeeds_when_resent_credential_name_is_dangling": [
+      "mgmt.model.update.unchanged_credential_name_is_not_revalidated"
+    ],
+    "tests/integration/management/test_model_credential_name_updates.py::test_non_admin_detach_and_empty_credential_name_still_rejected": [
+      "mgmt.model.update.non_admin_detach_is_rejected",
+      "mgmt.model.update.empty_credential_name_is_rejected"
+    ],
+    "tests/integration/management/test_model_credential_name_updates.py::test_changing_credential_name_to_missing_credential_is_rejected": [
+      "mgmt.model.update.changed_missing_credential_name_is_rejected"
+    ],
     "tests/integration/pricing/test_configured_prices.py::test_custom_price_is_reported_and_charged": [
       "quota_management.spend_tracking.custom_price.matches_input_rates"
     ],

tests/integration/management/test_model_credential_name_updates.py

@@ -0,0 +1,136 @@
+import uuid
+from typing import Final
+
+import httpx
+import pytest
+from pydantic import JsonValue
+
+from tests.integration._support.client import JSON_OBJECT, Gateway, Scenario, object_value, string_value
+from tests.integration._support.database import read_rows
+
+
+def _dangling_credential(gateway: Gateway, scenario: Scenario) -> str:
+    name: Final = f"credential-{uuid.uuid4().hex}"
+    gateway.post(
+        "/credentials",
+        {"credential_name": name, "credential_values": {"api_key": "synthetic-credential"}, "credential_info": {}},
+    )
+    scenario.cleanups.callback(_delete_credential_if_present, gateway, name)
+    return name
+
+
+def _delete_credential_if_present(gateway: Gateway, name: str) -> None:
+    response: Final = gateway.request("DELETE", f"/credentials/{name}")
+    assert response.status_code in (200, 404), response.text
+
+
+def _delete_credential(gateway: Gateway, name: str) -> None:
+    response: Final = gateway.request("DELETE", f"/credentials/{name}")
+    assert response.status_code == 200, response.text
+    assert read_rows('SELECT credential_name FROM "LiteLLM_CredentialsTable" WHERE credential_name = %s', (name,)) == []
+
+
+def _model_with_credential(gateway: Gateway, scenario: Scenario, credential: str, **model_info: JsonValue) -> str:
+    created: Final = gateway.post(
+        "/model/new",
+        {
+            "model_name": f"integration-{uuid.uuid4().hex}",
+            "litellm_params": {
+                "model": "openai/gpt-4o-mini",
+                "api_base": f"{gateway.upstream_url}/v1",
+                "litellm_credential_name": credential,
+                "rpm": 5,
+            },
+            "model_info": dict(model_info),
+        },
+    )
+    identity: Final = string_value(object_value(created["model_info"])["id"])
+    scenario.cleanups.callback(scenario.delete_model, identity)
+    return identity
+
+
+def _stored_params(gateway: Gateway, identity: str) -> dict[str, JsonValue]:
+    entries: Final = gateway.get("/model/info", {"litellm_model_id": identity})["data"]
+    assert isinstance(entries, list) and len(entries) == 1, entries
+    return object_value(object_value(entries[0])["litellm_params"])
+
+
+def _error(response: httpx.Response) -> dict[str, JsonValue]:
+    return object_value(JSON_OBJECT.validate_json(response.content)["error"])
+
+
+@pytest.mark.covers("mgmt.model.update.unchanged_credential_name_is_not_revalidated")
+def test_unrelated_patch_succeeds_when_resent_credential_name_is_dangling(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        credential: Final = _dangling_credential(gateway, scenario)
+        identity: Final = _model_with_credential(gateway, scenario, credential)
+        _delete_credential(gateway, credential)
+        before: Final = _stored_params(gateway, identity)
+        assert before["litellm_credential_name"] == credential
+        assert before["rpm"] == 5
+        patched: Final = gateway.request(
+            "PATCH",
+            f"/model/{identity}/update",
+            {"litellm_params": {"litellm_credential_name": before["litellm_credential_name"], "rpm": 7}},
+        )
+        assert patched.status_code == 200, patched.text
+        after: Final = _stored_params(gateway, identity)
+        assert after == {**before, "rpm": 7}
+
+
+@pytest.mark.covers(
+    "mgmt.model.update.non_admin_detach_is_rejected",
+    "mgmt.model.update.empty_credential_name_is_rejected",
+)
+def test_non_admin_detach_and_empty_credential_name_still_rejected(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        credential: Final = _dangling_credential(gateway, scenario)
+        user: Final = scenario.user(user_role="internal_user")
+        team: Final = scenario.team(members_with_roles=[{"user_id": user, "role": "admin"}])
+        team_admin: Final = scenario.key(user_id=user, team_id=team)
+        identity: Final = _model_with_credential(gateway, scenario, credential, team_id=team)
+        before: Final = _stored_params(gateway, identity)
+        detached: Final = gateway.request(
+            "PATCH", f"/model/{identity}/update", {"litellm_params": {"litellm_credential_name": None}}, key=team_admin
+        )
+        assert detached.status_code == 403, detached.text
+        assert _error(detached) == {
+            "message": "Only a proxy admin can detach a stored credential (litellm_credential_name) on a model. "
+            "Your role=internal_user.",
+            "type": "auth_error",
+            "param": "litellm_credential_name",
+            "code": "403",
+        }
+        emptied: Final = gateway.request(
+            "PATCH", f"/model/{identity}/update", {"litellm_params": {"litellm_credential_name": ""}}
+        )
+        assert emptied.status_code == 400, emptied.text
+        assert _error(emptied) == {
+            "message": "litellm_credential_name cannot be an empty string. Send null to detach the stored credential "
+            "or omit the field to leave it unchanged.",
+            "type": "validation_error",
+            "param": "litellm_credential_name",
+            "code": "400",
+        }
+        assert _stored_params(gateway, identity) == before
+
+
+@pytest.mark.covers("mgmt.model.update.changed_missing_credential_name_is_rejected")
+def test_changing_credential_name_to_missing_credential_is_rejected(gateway: Gateway) -> None:
+    with gateway.scenario() as scenario:
+        credential: Final = _dangling_credential(gateway, scenario)
+        identity: Final = _model_with_credential(gateway, scenario, credential)
+        _delete_credential(gateway, credential)
+        before: Final = _stored_params(gateway, identity)
+        missing: Final = f"credential-{uuid.uuid4().hex}"
+        rejected: Final = gateway.request(
+            "PATCH", f"/model/{identity}/update", {"litellm_params": {"litellm_credential_name": missing, "rpm": 7}}
+        )
+        assert rejected.status_code == 400, rejected.text
+        assert _error(rejected) == {
+            "message": f"Credential '{missing}' not found. Create it via /credentials before attaching it to a model.",
+            "type": "validation_error",
+            "param": "litellm_credential_name",
+            "code": "400",
+        }
+        assert _stored_params(gateway, identity) == before

tests/test_litellm/proxy/management_endpoints/test_model_management_endpoints.py

@@ -4796,6 +4796,30 @@ class TestPatchModelCredentialName:
         assert "not found" in exc_info.value.message.lower()
         credentials_repository.find_by_name.assert_awaited_once_with("ghost-credential")
 
+    @pytest.mark.asyncio
+    async def test_patch_model_resending_unchanged_dangling_credential_name_is_not_validated(self, monkeypatch):
+        credentials_repository = MagicMock()
+        db_model: Final = Deployment(
+            model_name="gpt-4",
+            litellm_params=LiteLLM_Params(
+                model="openai/gpt-4o",
+                api_base="https://api.openai.com/v1",
+                litellm_credential_name="ghost-credential",
+            ),
+            model_info=ModelInfo(id="dep-cred-1"),
+        )
+
+        persisted: Final = await self._patch_model(
+            monkeypatch,
+            db_model,
+            self._admin_user(),
+            "ghost-credential",
+            credentials_repository=credentials_repository,
+        )
+        params: Final = json.loads(persisted[0]["litellm_params"])
+        assert params["litellm_credential_name"] == "ghost-credential"
+        credentials_repository.find_by_name.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_patch_model_accepts_credential_known_only_in_db(self, monkeypatch):
         db_model: Final = Deployment(