fix(proxy): gate every credential write path on the effective workload identity surface

The admin gate only inspected the incoming credential_values, so it was fail-open three
ways. A non-admin could PATCH with an empty value map and name a federation field in
credential_values_to_delete, dropping it off the stored credential. A PATCH carrying no
federation field could still mutate a credential whose stored values held them, because
the stored state was never consulted. DELETE had no gate at all

The gate now takes the effective surface of the operation: the fields the payload sets,
the keys it names for deletion, and the fields already on the credential being changed.
Presence is keyed on the field name rather than a non-null value, matching how the
resolver and get_litellm_params decide a field is set, so clearing a field to null no
longer walks past the gate and wedges every deployment referencing that credential

Create, delete and a rename that targets another credential resolve the existing record
through hydrate_named_credential, which reads the in-memory config list before the
database. A credential declared in config.yaml has no row, so a DB-only lookup let a
non-admin shadow or drop one. POST semantics for credentials carrying no federation
field are unchanged
This commit is contained in:
derhornspieler 2026-08-23 16:06:38 -04:00
parent db268de4f7
commit 6c89af0720
4 changed files with 455 additions and 51 deletions

View file

@ -2,7 +2,6 @@
CRUD endpoints for storing reusable credentials.
"""
from collections.abc import Mapping
from typing import Final
from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
@ -27,35 +26,49 @@ from litellm.proxy.common_utils.credential_hydration import hydrate_named_creden
from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper
from litellm.proxy.utils import handle_exception_on_proxy, jsonify_object
from litellm.repositories.credentials_repository import CredentialsRepository
from litellm.types.router import anthropic_wif_fields_present
from litellm.types.router import anthropic_wif_fields_named
from litellm.types.utils import CreateCredentialItem, CredentialItem
router: Final = APIRouter()
def _reject_non_admin_wif_credential(
credential_values: Mapping[str, object] | None,
def _reject_non_admin_wif_fields(
wif_fields: tuple[str, ...],
user_api_key_dict: UserAPIKeyAuth,
) -> None:
"""A credential referenced by ``litellm_credential_name`` feeds its values into the same
workload identity federation resolution as a deployment's own ``litellm_params``. Only
proxy admins may create or update a credential that carries a server-owned WIF field.
workload identity federation resolution as a deployment's own ``litellm_params``. Only proxy
admins may touch a server-owned WIF field, whether they write it, drop it, or edit a stored
credential that already carries one.
"""
if credential_values is None or user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return
wif_fields: Final = anthropic_wif_fields_present(credential_values)
if not wif_fields:
if not wif_fields or user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return
raise HTTPException(
status_code=403,
detail={ # mutable-ok: starlette json.dumps()s HTTPException.detail raw, needs a real dict
"error": (
f"Only proxy admins can set {wif_fields[0]!r}, a server-owned workload identity federation parameter."
f"Only proxy admins can change {wif_fields[0]!r}, a server-owned workload identity federation "
"parameter."
)
},
)
def _incoming_wif_fields(credential: CredentialItem) -> tuple[str, ...]:
"""WIF fields the request payload itself touches: the ones it sets (to any value, ``None``
included, since the key alone is what the federation resolver reacts to), plus the ones it
names in ``credential_values_to_delete``, since dropping a federation field off the stored
credential breaks every deployment referencing it just as installing one would redirect them.
"""
return anthropic_wif_fields_named(credential.credential_values) + anthropic_wif_fields_named(
credential.credential_values_to_delete or ()
)
def _stored_wif_fields(stored_credential: CredentialItem) -> tuple[str, ...]:
return anthropic_wif_fields_named(stored_credential.credential_values)
def _reject_overlapping_credential_values(credential: CredentialItem) -> None:
overlap: Final = frozenset(credential.credential_values) & frozenset(credential.credential_values_to_delete or ())
if overlap:
@ -158,7 +171,10 @@ async def create_credential(
status_code=400,
detail="Credential values are required. Unable to infer credential values from model ID.",
)
_reject_non_admin_wif_credential(credential.credential_values, user_api_key_dict)
_reject_non_admin_wif_fields(anthropic_wif_fields_named(credential.credential_values), user_api_key_dict)
existing_credential: Final = await hydrate_named_credential(credential.credential_name, prisma_client)
if existing_credential is not None:
_reject_non_admin_wif_fields(_stored_wif_fields(existing_credential), user_api_key_dict)
processed_credential: Final = CredentialItem(
credential_name=credential.credential_name,
credential_values=credential.credential_values,
@ -376,11 +392,16 @@ async def delete_credential(
status_code=500,
detail={"error": CommonProxyErrors.db_not_connected_error.value},
)
existing_credential: Final = await hydrate_named_credential(credential_name, prisma_client)
if existing_credential is not None:
_reject_non_admin_wif_fields(_stored_wif_fields(existing_credential), user_api_key_dict)
await CredentialsRepository(prisma_client).delete_by_name(credential_name)
## DELETE FROM LITELLM ##
litellm.credential_list = [cred for cred in litellm.credential_list if cred.credential_name != credential_name]
return {"success": True, "message": "Credential deleted successfully"}
except HTTPException:
raise
except Exception as e:
return handle_exception_on_proxy(e)
@ -446,7 +467,7 @@ async def update_credential(
try:
_reject_overlapping_credential_values(credential)
_reject_non_admin_wif_credential(credential.credential_values, user_api_key_dict)
_reject_non_admin_wif_fields(_incoming_wif_fields(credential), user_api_key_dict)
if prisma_client is None:
raise HTTPException(
status_code=500,
@ -456,6 +477,11 @@ async def update_credential(
db_credential: Final = await credentials_repository.find_by_name(credential_name)
if db_credential is None:
raise HTTPException(status_code=404, detail="Credential not found in DB.")
_reject_non_admin_wif_fields(_stored_wif_fields(db_credential), user_api_key_dict)
if credential.credential_name != credential_name:
shadowed_credential: Final = await hydrate_named_credential(credential.credential_name, prisma_client)
if shadowed_credential is not None:
_reject_non_admin_wif_fields(_stored_wif_fields(shadowed_credential), user_api_key_dict)
merged_credential: Final = update_db_credential(db_credential, credential)
credential_object_jsonified: Final = jsonify_object(merged_credential.model_dump(exclude_none=True))
await credentials_repository.update_by_name(

View file

@ -4,7 +4,7 @@ litellm.Router Types - includes RouterConfig, UpdateRouterConfig, ModelInfo etc
import datetime
import enum
from collections.abc import Mapping
from collections.abc import Container, Mapping
from dataclasses import dataclass
from typing import Any, ClassVar, Final, Generic, Literal, TypeVar, get_type_hints
@ -310,6 +310,19 @@ def anthropic_wif_fields_present(fields: Mapping[str, object]) -> tuple[str, ...
return tuple(name for name in _anthropic_wif_litellm_params if fields.get(name) is not None)
def anthropic_wif_fields_named(keys: Container[str]) -> tuple[str, ...]:
"""Server-owned Anthropic workload identity federation field names that appear in ``keys``,
whatever value they carry.
The write gates on credentials need this key-based sibling of ``anthropic_wif_fields_present``:
``get_litellm_params`` forwards a WIF kwarg on key presence and the federation resolver rejects
a foreign variant's field by key, so a persisted ``{"anthropic_issuer_url": None}`` wedges every
deployment that references the credential even though no value is set. Pass a mapping (its keys
are tested) or a plain collection of key names.
"""
return tuple(name for name in _anthropic_wif_litellm_params if name in keys)
_RESERVED_INIT_KEYS: Final = frozenset({"self", "params", "__class__"})

View file

@ -1,6 +1,7 @@
"""Tests for the credential management endpoints."""
import json
from contextlib import contextmanager
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
@ -55,11 +56,12 @@ def _post_credential(body: dict, auth=_as_admin):
app.dependency_overrides[user_api_key_auth] = previous_override
def test_create_credential_write_omits_the_patch_only_deletion_field():
"""Regression: CredentialItem.credential_values_to_delete is a PATCH-only field that
defaults to None on every other construction path. A bare .model_dump() (without
exclude_none) on the create path put a `credential_values_to_delete: null` key into the
Prisma write, which litellm_credentialstable has no column for."""
@contextmanager
def _repository_holding(stored: CredentialItem | None):
"""The credentials repository seam, answering ``find_by_name`` with ``stored`` and recording
the writes the handler attempts. Patched at both import sites, since the handlers resolve an
existing credential through ``hydrate_named_credential`` (memory first, then this repository)
and then write through their own ``CredentialsRepository`` binding."""
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", MagicMock()
@ -69,11 +71,24 @@ def test_create_credential_write_omits_the_patch_only_deletion_field():
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
) as repository, # test-quality-ok: the proxy wiring under test is what this patches
) as repository,
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.common_utils.credential_hydration.CredentialsRepository", repository
),
):
create_mock = AsyncMock(return_value=None)
repository.return_value.create = create_mock
repository.return_value.find_by_name = AsyncMock(return_value=stored)
repository.return_value.create = AsyncMock(return_value=None)
repository.return_value.update_by_name = AsyncMock(return_value=None)
repository.return_value.delete_by_name = AsyncMock(return_value=None)
yield repository.return_value
def test_create_credential_write_omits_the_patch_only_deletion_field(restore_credential_list):
"""Regression: CredentialItem.credential_values_to_delete is a PATCH-only field that
defaults to None on every other construction path. A bare .model_dump() (without
exclude_none) on the create path put a `credential_values_to_delete: null` key into the
Prisma write, which litellm_credentialstable has no column for."""
with _repository_holding(None) as repository:
response = _post_credential(
{
"credential_name": "new-cred",
@ -83,7 +98,7 @@ def test_create_credential_write_omits_the_patch_only_deletion_field():
)
assert response.status_code == 200, response.text
written_data = create_mock.await_args.kwargs["data"]
written_data = repository.create.await_args.kwargs["data"]
assert "credential_values_to_delete" not in written_data
@ -463,20 +478,8 @@ class TestNonAdminCannotPersistWifFieldsOnCredential:
assert response.status_code == 403, response.text
def test_non_admin_can_create_a_credential_without_wif_fields(self):
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", MagicMock()
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.master_key", "sk-test-master"
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
) as repository, # test-quality-ok: the proxy wiring under test is what this patches
):
repository.return_value.create = AsyncMock(return_value=None)
def test_non_admin_can_create_a_credential_without_wif_fields(self, restore_credential_list):
with _repository_holding(None) as repository:
response = _post_credential(
{
"credential_name": "ordinary-cred",
@ -487,21 +490,10 @@ class TestNonAdminCannotPersistWifFieldsOnCredential:
)
assert response.status_code == 200, response.text
repository.create.assert_awaited_once()
def test_proxy_admin_can_create_a_credential_with_a_wif_destination(self):
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", MagicMock()
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.master_key", "sk-test-master"
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.credential_endpoints.endpoints.CredentialsRepository"
) as repository, # test-quality-ok: the proxy wiring under test is what this patches
):
repository.return_value.create = AsyncMock(return_value=None)
def test_proxy_admin_can_create_a_credential_with_a_wif_destination(self, restore_credential_list):
with _repository_holding(None) as repository:
response = _post_credential(
{
"credential_name": "admin-cred",
@ -512,6 +504,7 @@ class TestNonAdminCannotPersistWifFieldsOnCredential:
)
assert response.status_code == 200, response.text
repository.create.assert_awaited_once()
def test_non_admin_cannot_update_a_credential_to_add_a_wif_destination(self):
stored = CredentialItem(
@ -577,3 +570,359 @@ class TestNonAdminCannotPersistWifFieldsOnCredential:
assert response.status_code == 200, response.text
update_mock.assert_awaited_once()
def _delete_credential(name: str, auth=_as_admin):
missing = object()
previous_override = app.dependency_overrides.get(user_api_key_auth, missing)
app.dependency_overrides[user_api_key_auth] = auth
try:
return client.delete(f"/credentials/{name}", headers={"Authorization": "Bearer test-key"})
finally:
if previous_override is missing:
app.dependency_overrides.pop(user_api_key_auth, None)
else:
app.dependency_overrides[user_api_key_auth] = previous_override
def _wif_credential(name: str = "federated-cred") -> CredentialItem:
return CredentialItem(
credential_name=name,
credential_values={
"anthropic_keycloak_token_url": "https://keycloak.internal/token",
"api_key": "sk-old",
},
credential_info={"custom_llm_provider": "anthropic"},
)
def _plain_credential(name: str = "ordinary-cred") -> CredentialItem:
return CredentialItem(
credential_name=name,
credential_values={"api_key": "sk-old"},
credential_info={"custom_llm_provider": "openai"},
)
class TestNonAdminCannotTouchAStoredWifCredential:
"""The WIF gate used to read only the incoming ``credential_values``, so a non-admin could
drop a federation field by naming it in ``credential_values_to_delete`` (breaking every
deployment that references the credential), or edit a stored admin-owned WIF credential by
sending a payload carrying no WIF field at all. The gate is evaluated against the effective
surface of the operation: incoming keys (a ``null`` value still persists the key), deleted
keys, and the stored credential, wherever it lives (DB row or config-only ``credential_list``
entry)."""
def test_non_admin_cannot_delete_a_wif_field_off_a_credential(self, restore_credential_list):
with _repository_holding(_plain_credential("some-cred")) as repository:
response = _patch_credential(
"some-cred",
{
"credential_name": "some-cred",
"credential_values": {},
"credential_values_to_delete": ["anthropic_keycloak_token_url"],
"credential_info": {},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_keycloak_token_url" in response.text
repository.update_by_name.assert_not_awaited()
def test_non_admin_cannot_patch_a_stored_wif_credential(self, restore_credential_list):
with _repository_holding(_wif_credential("federated-cred")) as repository:
response = _patch_credential(
"federated-cred",
{
"credential_name": "federated-cred",
"credential_values": {"api_key": "sk-attacker"},
"credential_info": {},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_keycloak_token_url" in response.text
repository.update_by_name.assert_not_awaited()
def test_proxy_admin_can_delete_a_wif_field_off_a_credential(self, restore_credential_list):
with _repository_holding(_wif_credential("federated-cred")) as repository:
response = _patch_credential(
"federated-cred",
{
"credential_name": "federated-cred",
"credential_values": {},
"credential_values_to_delete": ["anthropic_keycloak_token_url"],
"credential_info": {},
},
auth=_as_admin,
)
assert response.status_code == 200, response.text
written_values = json.loads(repository.update_by_name.await_args.kwargs["data"]["credential_values"])
assert "anthropic_keycloak_token_url" not in written_values
def test_proxy_admin_can_patch_a_stored_wif_credential(self, restore_credential_list):
with _repository_holding(_wif_credential("federated-cred")) as repository:
response = _patch_credential(
"federated-cred",
{
"credential_name": "federated-cred",
"credential_values": {"api_key": "sk-rotated"},
"credential_info": {},
},
auth=_as_admin,
)
assert response.status_code == 200, response.text
written_values = json.loads(repository.update_by_name.await_args.kwargs["data"]["credential_values"])
assert written_values["anthropic_keycloak_token_url"] is not None
def test_non_admin_can_still_patch_a_credential_with_no_wif_fields_anywhere(self, restore_credential_list):
with _repository_holding(_plain_credential("ordinary-cred")) as repository:
response = _patch_credential(
"ordinary-cred",
{
"credential_name": "ordinary-cred",
"credential_values": {"api_key": "sk-rotated"},
"credential_info": {},
},
auth=_as_non_admin,
)
assert response.status_code == 200, response.text
repository.update_by_name.assert_awaited_once()
def test_non_admin_cannot_delete_a_stored_wif_credential(self, restore_credential_list):
"""DELETE takes the whole row, so it drops the admin-owned federation settings as surely
as a targeted key deletion would."""
with _repository_holding(_wif_credential("federated-cred")) as repository:
response = _delete_credential("federated-cred", auth=_as_non_admin)
assert response.status_code == 403, response.text
assert "anthropic_keycloak_token_url" in response.text
repository.delete_by_name.assert_not_awaited()
def test_proxy_admin_can_delete_a_stored_wif_credential(self, restore_credential_list):
with _repository_holding(_wif_credential("federated-cred")) as repository:
response = _delete_credential("federated-cred", auth=_as_admin)
assert response.status_code == 200, response.text
repository.delete_by_name.assert_awaited_once_with("federated-cred")
def test_non_admin_can_still_delete_a_credential_with_no_wif_fields(self, restore_credential_list):
with _repository_holding(_plain_credential("ordinary-cred")) as repository:
response = _delete_credential("ordinary-cred", auth=_as_non_admin)
assert response.status_code == 200, response.text
repository.delete_by_name.assert_awaited_once_with("ordinary-cred")
def test_non_admin_cannot_null_out_a_wif_field_on_a_credential(self, restore_credential_list):
"""A JSON ``null`` still lands as a key in ``credential_values``. ``get_litellm_params``
forwards a WIF kwarg on key presence and the federation resolver rejects a foreign
variant's field by key, so a value-based gate let a non-admin persist the key and wedge
every deployment referencing the credential at request time."""
with _repository_holding(_plain_credential("some-cred")) as repository:
response = _patch_credential(
"some-cred",
{
"credential_name": "some-cred",
"credential_values": {"anthropic_issuer_url": None},
"credential_info": {},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_issuer_url" in response.text
repository.update_by_name.assert_not_awaited()
def test_non_admin_cannot_patch_a_credential_storing_a_null_wif_field(self, restore_credential_list):
stored = CredentialItem(
credential_name="nulled-cred",
credential_values={"anthropic_issuer_url": None, "api_key": "sk-old"},
credential_info={"custom_llm_provider": "anthropic"},
)
with _repository_holding(stored) as repository:
response = _patch_credential(
"nulled-cred",
{
"credential_name": "nulled-cred",
"credential_values": {"api_key": "sk-attacker"},
"credential_info": {},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_issuer_url" in response.text
repository.update_by_name.assert_not_awaited()
def test_proxy_admin_can_null_out_a_wif_field_on_a_credential(self, restore_credential_list):
with _repository_holding(_wif_credential("federated-cred")) as repository:
response = _patch_credential(
"federated-cred",
{
"credential_name": "federated-cred",
"credential_values": {"anthropic_keycloak_token_url": None},
"credential_info": {},
},
auth=_as_admin,
)
assert response.status_code == 200, response.text
repository.update_by_name.assert_awaited_once()
def test_non_admin_cannot_delete_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
"""A ``credential_list`` entry from config.yaml has no DB row, so a gate that consulted
only the DB let a non-admin evict the admin-owned federation settings from memory."""
config_credential = _wif_credential("config-wif")
monkeypatch.setattr(litellm, "credential_list", [config_credential])
with _repository_holding(None) as repository:
response = _delete_credential("config-wif", auth=_as_non_admin)
assert response.status_code == 403, response.text
assert "anthropic_keycloak_token_url" in response.text
repository.delete_by_name.assert_not_awaited()
assert litellm.credential_list == [config_credential]
def test_proxy_admin_can_delete_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
monkeypatch.setattr(litellm, "credential_list", [_wif_credential("config-wif")])
with _repository_holding(None) as repository:
response = _delete_credential("config-wif", auth=_as_admin)
assert response.status_code == 200, response.text
repository.delete_by_name.assert_awaited_once_with("config-wif")
assert litellm.credential_list == []
def test_non_admin_cannot_shadow_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
"""POST with the same name carries no WIF field and collides with no DB row, yet
``CredentialAccessor.upsert_credentials`` would replace the admin entry in memory and
the periodic config sync would then make the takeover permanent."""
config_credential = _wif_credential("config-wif")
monkeypatch.setattr(litellm, "credential_list", [config_credential])
with _repository_holding(None) as repository:
response = _post_credential(
{
"credential_name": "config-wif",
"credential_values": {"api_key": "sk-attacker"},
"credential_info": {"custom_llm_provider": "anthropic"},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_keycloak_token_url" in response.text
repository.create.assert_not_awaited()
assert litellm.credential_list == [config_credential]
assert litellm.credential_list[0].credential_values["api_key"] == "sk-old"
def test_proxy_admin_can_post_over_a_config_only_wif_credential(self, restore_credential_list, monkeypatch):
monkeypatch.setattr(litellm, "credential_list", [_wif_credential("config-wif")])
with _repository_holding(None) as repository:
response = _post_credential(
{
"credential_name": "config-wif",
"credential_values": {"api_key": "sk-rotated"},
"credential_info": {"custom_llm_provider": "anthropic"},
},
auth=_as_admin,
)
assert response.status_code == 200, response.text
repository.create.assert_awaited_once()
assert litellm.credential_list[0].credential_values == {"api_key": "sk-rotated"}
def test_non_admin_cannot_rename_a_credential_onto_a_config_only_wif_credential(
self, restore_credential_list, monkeypatch
):
"""PATCH is the other way to shadow: renaming an ordinary credential onto the WIF
credential's name makes ``_sync_in_memory_credential`` upsert the attacker's values over
the admin entry, with no WIF field in the payload and no DB row to collide with."""
config_credential = _wif_credential("config-wif")
monkeypatch.setattr(litellm, "credential_list", [_plain_credential("mine"), config_credential])
with _repository_holding(_plain_credential("mine")) as repository:
response = _patch_credential(
"mine",
{
"credential_name": "config-wif",
"credential_values": {"api_key": "sk-attacker"},
"credential_info": {},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_keycloak_token_url" in response.text
repository.update_by_name.assert_not_awaited()
assert config_credential in litellm.credential_list
assert litellm.credential_list[1].credential_values["api_key"] == "sk-old"
def test_proxy_admin_can_rename_a_credential_onto_a_config_only_wif_credential(
self, restore_credential_list, monkeypatch
):
monkeypatch.setattr(litellm, "credential_list", [_plain_credential("mine"), _wif_credential("config-wif")])
with _repository_holding(_plain_credential("mine")) as repository:
response = _patch_credential(
"mine",
{
"credential_name": "config-wif",
"credential_values": {"api_key": "sk-rotated"},
"credential_info": {},
},
auth=_as_admin,
)
assert response.status_code == 200, response.text
repository.update_by_name.assert_awaited_once()
assert [c.credential_name for c in litellm.credential_list] == ["config-wif"]
def test_non_admin_cannot_post_a_null_wif_field(self, restore_credential_list):
"""Same key-presence rule on the create path: ``{"anthropic_issuer_url": null}`` persists
the key, and the resolver reacts to the key."""
with _repository_holding(None) as repository:
response = _post_credential(
{
"credential_name": "nulled-cred",
"credential_values": {"anthropic_issuer_url": None, "api_key": "sk-new"},
"credential_info": {"custom_llm_provider": "anthropic"},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_issuer_url" in response.text
repository.create.assert_not_awaited()
assert litellm.credential_list == []
def test_non_admin_cannot_shadow_a_db_stored_wif_credential(self, restore_credential_list):
"""Same hole for a WIF credential another pod wrote to the DB before this pod's in-memory
list caught up: the existing-credential lookup falls through to the DB."""
with _repository_holding(_wif_credential("federated-cred")) as repository:
response = _post_credential(
{
"credential_name": "federated-cred",
"credential_values": {"api_key": "sk-attacker"},
"credential_info": {"custom_llm_provider": "anthropic"},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
repository.create.assert_not_awaited()
def test_non_admin_can_still_post_a_credential_with_no_wif_fields_anywhere(self, restore_credential_list):
with _repository_holding(None) as repository:
response = _post_credential(
{
"credential_name": "ordinary-cred",
"credential_values": {"api_key": "sk-new"},
"credential_info": {"custom_llm_provider": "openai"},
},
auth=_as_non_admin,
)
assert response.status_code == 200, response.text
repository.create.assert_awaited_once()
assert litellm.credential_list[0].credential_name == "ordinary-cred"

View file

@ -6,6 +6,7 @@ from litellm.types.router import (
Deployment,
LiteLLM_Params,
ModelInfo,
anthropic_wif_fields_named,
anthropic_wif_fields_present,
)
from litellm.types.utils import (
@ -129,3 +130,18 @@ def test_anthropic_wif_fields_present_is_derived_from_the_shared_list():
written -- so this must read the shared list rather than a hand-copied one."""
values = {field: "set" for field in anthropic_wif_litellm_params}
assert set(anthropic_wif_fields_present(values)) == set(anthropic_wif_litellm_params)
def test_anthropic_wif_fields_named_reports_keys_whatever_their_value():
"""The credential write gates must see a key a caller sets to ``None``: the federation
resolver reacts to the key's presence, not its value, so ``{"anthropic_issuer_url": None}``
wedges every deployment referencing the credential once persisted."""
assert anthropic_wif_fields_named({}) == ()
assert anthropic_wif_fields_named({"model": "gpt-4o"}) == ()
assert anthropic_wif_fields_named({"anthropic_issuer_url": None}) == ("anthropic_issuer_url",)
assert anthropic_wif_fields_present({"anthropic_issuer_url": None}) == ()
assert anthropic_wif_fields_named(("anthropic_keycloak_token_url", "api_key")) == ("anthropic_keycloak_token_url",)
def test_anthropic_wif_fields_named_is_derived_from_the_shared_list():
assert set(anthropic_wif_fields_named(frozenset(anthropic_wif_litellm_params))) == set(anthropic_wif_litellm_params)