fix(proxy): let only proxy admins persist Anthropic workload identity fields

These fields pick which server-side secret is read and where the resulting assertion is
sent, so a team admin who can otherwise manage a team-scoped deployment must not be able
to set them. The create, update and patch model paths and both credential write paths now
reject them for anyone below proxy admin, mirroring the existing blocked-flag gate. The
field list is derived from anthropic_wif_litellm_params rather than copied, so a new
federation field is covered the day it is added

Also fixes two things CI caught: credential_values_to_delete is a PATCH instruction rather
than part of the credential, so it stays out of dumps that feed config loading and the
Prisma write, and the provider discovery route is registered in the backend allowlist
This commit is contained in:
derhornspieler 2026-08-23 12:42:23 -04:00
parent 35626611cf
commit b0a28b2e09
8 changed files with 795 additions and 96 deletions

View file

@ -55,6 +55,7 @@ BACKEND_PATH_PREFIXES: tuple[str, ...] = (
"/credentials",
"/credential",
"/provider/budgets",
"/provider/models/discover",
# Tools / agents (registry & policy admin)
"/v1/tool/",
"/v1/agents",

View file

@ -5,7 +5,7 @@ These are the canonical credential types for the proxy. They live in the model
layer; ``litellm.types.utils`` re-exports them for backwards compatibility.
"""
from pydantic import BaseModel, model_validator
from pydantic import BaseModel, Field, model_validator
class CredentialBase(BaseModel):
@ -15,8 +15,10 @@ class CredentialBase(BaseModel):
class CredentialItem(CredentialBase):
credential_values: dict
# PATCH-only: keys to drop from the stored credential_values.
credential_values_to_delete: tuple[str, ...] | None = None
# PATCH-only instruction naming keys to drop from the stored credential_values. It describes an
# edit rather than the credential, so it stays out of dumps: those feed config loading, the DB
# write, and the in-memory list, none of which have a place for it.
credential_values_to_delete: tuple[str, ...] | None = Field(default=None, exclude=True)
class CreateCredentialItem(CredentialBase):

View file

@ -2,6 +2,7 @@
CRUD endpoints for storing reusable credentials.
"""
from collections.abc import Mapping
from typing import Final
from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
@ -26,11 +27,76 @@ 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.utils import CreateCredentialItem, CredentialItem
router: Final = APIRouter()
def _reject_non_admin_wif_credential(
credential_values: Mapping[str, object] | None,
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.
"""
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:
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."
)
},
)
def _reject_overlapping_credential_values(credential: CredentialItem) -> None:
overlap: Final = frozenset(credential.credential_values) & frozenset(credential.credential_values_to_delete or ())
if overlap:
raise HTTPException(
status_code=400,
detail=f"credential_values_to_delete overlaps credential_values for key(s): {sorted(overlap)}",
)
def _sync_in_memory_credential(credential: CredentialItem, credential_name: str, new_name: str) -> None:
"""Mirror a DB credential update into the in-memory ``credential_list`` used by request-time
resolution; a no-op if the credential isn't loaded in memory (e.g. proxy restarted since boot).
"""
existing_in_memory: CredentialItem | None = None
for cred in litellm.credential_list:
if cred.credential_name == credential_name:
existing_in_memory = cred
break
if existing_in_memory is None:
return
in_memory_values: Final = dict(existing_in_memory.credential_values or {})
if credential.credential_values:
in_memory_values.update(credential.credential_values)
for key in credential.credential_values_to_delete or ():
in_memory_values.pop(key, None)
in_memory_info: Final = dict(existing_in_memory.credential_info or {})
if credential.credential_info:
in_memory_info.update(credential.credential_info)
updated_in_memory: Final = CredentialItem(
credential_name=new_name,
credential_values=in_memory_values,
credential_info=in_memory_info,
)
# Remove old entry if renamed, then use upsert_credentials to handle duplicates
if new_name != credential_name:
litellm.credential_list = [c for c in litellm.credential_list if c.credential_name != credential_name]
CredentialAccessor.upsert_credentials([updated_in_memory])
class CredentialHelperUtils:
@staticmethod
def encrypt_credential_values(credential: CredentialItem, new_encryption_key: str | None = None) -> CredentialItem:
@ -92,6 +158,7 @@ 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)
processed_credential: Final = CredentialItem(
credential_name=credential.credential_name,
credential_values=credential.credential_values,
@ -378,14 +445,8 @@ async def update_credential(
from litellm.proxy.proxy_server import prisma_client
try:
overlap: Final = frozenset(credential.credential_values) & frozenset(
credential.credential_values_to_delete or ()
)
if overlap:
raise HTTPException(
status_code=400,
detail=f"credential_values_to_delete overlaps credential_values for key(s): {sorted(overlap)}",
)
_reject_overlapping_credential_values(credential)
_reject_non_admin_wif_credential(credential.credential_values, user_api_key_dict)
if prisma_client is None:
raise HTTPException(
status_code=500,
@ -406,31 +467,7 @@ async def update_credential(
)
# Sync in-memory credential_list (skip if not in memory - e.g., proxy restarted)
new_name: Final = merged_credential.credential_name
existing_in_memory: CredentialItem | None = None
for cred in litellm.credential_list:
if cred.credential_name == credential_name:
existing_in_memory = cred
break
if existing_in_memory is not None:
in_memory_values: Final = dict(existing_in_memory.credential_values or {})
if credential.credential_values:
in_memory_values.update(credential.credential_values)
for key in credential.credential_values_to_delete or ():
in_memory_values.pop(key, None)
in_memory_info: Final = dict(existing_in_memory.credential_info or {})
if credential.credential_info:
in_memory_info.update(credential.credential_info)
updated_in_memory: Final = CredentialItem(
credential_name=new_name,
credential_values=in_memory_values,
credential_info=in_memory_info,
)
# Remove old entry if renamed, then use upsert_credentials to handle duplicates
if new_name != credential_name:
litellm.credential_list = [c for c in litellm.credential_list if c.credential_name != credential_name]
CredentialAccessor.upsert_credentials([updated_in_memory])
_sync_in_memory_credential(credential, credential_name, merged_credential.credential_name)
return {"success": True, "message": "Credential updated successfully"}
except Exception as e:

View file

@ -100,6 +100,7 @@ from litellm.types.router import (
Deployment,
GenericLiteLLMParams,
ModelInfo,
anthropic_wif_fields_present,
updateDeployment,
)
from litellm.types.utils import LlmProviders
@ -253,6 +254,47 @@ def _raise_on_strategy_router_write_violation(
)
def _reject_non_admin_wif_persistence(
litellm_params: GenericLiteLLMParams | None,
user_api_key_dict: UserAPIKeyAuth,
) -> None:
"""Anthropic workload identity federation fields choose which server-side secret is read
and where it is sent. Only proxy admins may persist them on a deployment, mirroring the
``blocked``-flag gate below: a team admin who otherwise manages a team-scoped deployment
must not be able to set these.
"""
if litellm_params is None or user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return
wif_fields: Final = anthropic_wif_fields_present(litellm_params.model_dump(exclude_none=True))
if not wif_fields:
return
raise ProxyException(
message=(
f"Only proxy admins can set {wif_fields[0]!r}, a server-owned workload identity federation parameter."
),
type=ProxyErrorTypes.auth_error.value,
code=status.HTTP_403_FORBIDDEN,
param=wif_fields[0],
)
def _reject_non_admin_blocked_flag_on_create(
blocked: bool | None,
user_api_key_dict: UserAPIKeyAuth,
) -> None:
"""Same proxy-admin-only rule patch_model applies to the blocked flag: a team admin passed
the team-scoped auth check above, but must not be able to create a model already paused
(or explicitly unpaused) out from under the proxy admin.
"""
if blocked is not None and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
raise ProxyException(
message="Only proxy admins can set a model's blocked flag.",
type=ProxyErrorTypes.auth_error.value,
code=status.HTTP_403_FORBIDDEN,
param="blocked",
)
_PTU_PRICED_PAIR: Final = frozenset({"ptu_count", "cost_per_ptu_per_hour"})
@ -659,6 +701,8 @@ async def patch_model(
param="blocked",
)
_reject_non_admin_wif_persistence(patch_data.litellm_params, user_api_key_dict)
_raise_on_strategy_router_write_violation(
incoming_params=patch_data.litellm_params,
existing_params=db_model.litellm_params,
@ -1845,16 +1889,9 @@ async def add_new_model(
premium_user=premium_user,
)
# Same proxy-admin-only rule patch_model applies to the blocked flag: a team admin
# passed the check above for a team-scoped model, but must not be able to create it
# already paused (or explicitly unpaused) out from under the proxy admin.
if model_params.blocked is not None and user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
raise ProxyException(
message="Only proxy admins can set a model's blocked flag.",
type=ProxyErrorTypes.auth_error.value,
code=status.HTTP_403_FORBIDDEN,
param="blocked",
)
_reject_non_admin_blocked_flag_on_create(model_params.blocked, user_api_key_dict)
_reject_non_admin_wif_persistence(model_params.litellm_params, user_api_key_dict)
_raise_on_strategy_router_write_violation(
incoming_params=model_params.litellm_params,
@ -2035,6 +2072,8 @@ async def update_model(
if model_params.litellm_params is None:
raise Exception("litellm_params not provided")
_reject_non_admin_wif_persistence(model_params.litellm_params, user_api_key_dict)
_new_litellm_params_dict: Final = model_params.litellm_params.dict(exclude_none=True)
### ENCRYPT PARAMS ###

View file

@ -24,6 +24,10 @@ from .utils import (
ModelResponse,
StandardLoggingRoutingDecision,
)
from .utils import (
# private alias: `from .types.router import *` would rebind a public Final in litellm/__init__.py
anthropic_wif_litellm_params as _anthropic_wif_litellm_params,
)
class ConfigurableClientsideParamsCustomAuth(TypedDict):
@ -295,6 +299,17 @@ class CredentialLiteLLMParams(BaseModel):
anthropic_keycloak_scope: str | None = None
def anthropic_wif_fields_present(fields: Mapping[str, object]) -> tuple[str, ...]:
"""Server-owned Anthropic workload identity federation field names set in ``fields``.
``fields`` is a ``litellm_params`` dict (or a credential's ``credential_values`` mapping,
which feeds the same resolution when referenced by name). Derived from
``anthropic_wif_litellm_params`` rather than hand-copied, so a persistence gate built on
this stays correct when a new WIF field is added there.
"""
return tuple(name for name in _anthropic_wif_litellm_params if fields.get(name) is not None)
_RESERVED_INIT_KEYS: Final = frozenset({"self", "params", "__class__"})

View file

@ -21,10 +21,14 @@ def _as_admin():
return UserAPIKeyAuth(api_key="test-key", user_role="proxy_admin")
def _patch_credential(name: str, body: dict):
def _as_non_admin():
return UserAPIKeyAuth(api_key="test-key", user_role="internal_user")
def _patch_credential(name: str, body: dict, auth=_as_admin):
missing = object()
previous_override = app.dependency_overrides.get(user_api_key_auth, missing)
app.dependency_overrides[user_api_key_auth] = _as_admin
app.dependency_overrides[user_api_key_auth] = auth
try:
return client.patch(
f"/credentials/{name}",
@ -38,10 +42,10 @@ def _patch_credential(name: str, body: dict):
app.dependency_overrides[user_api_key_auth] = previous_override
def _post_credential(body: dict):
def _post_credential(body: dict, auth=_as_admin):
missing = object()
previous_override = app.dependency_overrides.get(user_api_key_auth, missing)
app.dependency_overrides[user_api_key_auth] = _as_admin
app.dependency_overrides[user_api_key_auth] = auth
try:
return client.post("/credentials", json=body, headers={"Authorization": "Bearer test-key"})
finally:
@ -57,9 +61,15 @@ def test_create_credential_write_omits_the_patch_only_deletion_field():
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 (
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.proxy_server.master_key", "sk-test-master"),
patch("litellm.proxy.credential_endpoints.endpoints.CredentialsRepository") as repository,
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
):
create_mock = AsyncMock(return_value=None)
repository.return_value.create = create_mock
@ -83,8 +93,12 @@ def test_update_credential_answers_404_when_the_credential_does_not_exist():
rejected read as a success to every caller that checks the status. The dashboard's API
client branches on the status, so it reported a failed edit as applied."""
with (
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.credential_endpoints.endpoints.CredentialsRepository") as repository,
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.credential_endpoints.endpoints.CredentialsRepository"
) as repository, # test-quality-ok: the proxy wiring under test is what this patches
):
repository.return_value.find_by_name = AsyncMock(return_value=None)
@ -121,9 +135,15 @@ def test_update_credential_still_answers_200_on_a_successful_write():
credential_info={"custom_llm_provider": "openai"},
)
with (
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.proxy_server.master_key", "sk-test-master"),
patch("litellm.proxy.credential_endpoints.endpoints.CredentialsRepository") as repository,
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.find_by_name = AsyncMock(return_value=stored)
repository.return_value.update_by_name = AsyncMock(return_value=None)
@ -181,8 +201,12 @@ def test_update_credential_deletion_removes_the_key_from_the_db_write(restore_cr
credential_info={"custom_llm_provider": "anthropic"},
)
with (
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.credential_endpoints.endpoints.CredentialsRepository") as repository,
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.credential_endpoints.endpoints.CredentialsRepository"
) as repository, # test-quality-ok: the proxy wiring under test is what this patches
):
repository.return_value.find_by_name = AsyncMock(return_value=stored)
update_mock = AsyncMock(return_value=None)
@ -227,8 +251,12 @@ def test_update_credential_deletion_updates_in_memory_credential_list(restore_cr
credential_info={"custom_llm_provider": "anthropic"},
)
with (
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.credential_endpoints.endpoints.CredentialsRepository") as repository,
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.credential_endpoints.endpoints.CredentialsRepository"
) as repository, # test-quality-ok: the proxy wiring under test is what this patches
):
repository.return_value.find_by_name = AsyncMock(return_value=stored)
repository.return_value.update_by_name = AsyncMock(return_value=None)
@ -259,9 +287,15 @@ def test_update_credential_leaves_untouched_fields_alone():
credential_info={"custom_llm_provider": "anthropic"},
)
with (
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch("litellm.proxy.proxy_server.master_key", "sk-test-master"),
patch("litellm.proxy.credential_endpoints.endpoints.CredentialsRepository") as repository,
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.find_by_name = AsyncMock(return_value=stored)
update_mock = AsyncMock(return_value=None)
@ -353,7 +387,9 @@ class TestCredentialJwksExport:
assert response.status_code == 404, response.text
def test_jwks_export_404s_for_an_unknown_credential(self, restore_credential_list):
with patch("litellm.proxy.proxy_server.prisma_client", None):
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", None
): # test-quality-ok: the proxy wiring under test is what this patches
response = _get_jwks("does-not-exist")
assert response.status_code == 404, response.text
@ -387,3 +423,157 @@ class TestCredentialJwksExport:
app.dependency_overrides.pop(user_api_key_auth, None)
assert response.status_code == 403, response.text
class TestNonAdminCannotPersistWifFieldsOnCredential:
"""A credential's ``credential_values`` feeds the same WIF resolution as a deployment's own
``litellm_params`` when referenced by ``litellm_credential_name``. A non-admin must not be
able to create or update a credential carrying a server-owned WIF field such as
``anthropic_keycloak_token_url`` (destination) or ``anthropic_keycloak_client_secret_ref``
(which secret to read and send there)."""
def test_non_admin_cannot_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()
): # test-quality-ok: the proxy wiring under test is what this patches
response = _post_credential(
{
"credential_name": "attacker-cred",
"credential_values": {"anthropic_keycloak_token_url": "https://evil.example.com/token"},
"credential_info": {"custom_llm_provider": "anthropic"},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
assert "anthropic_keycloak_token_url" in response.json()["error"]["message"]
def test_non_admin_cannot_create_a_credential_with_a_wif_secret_ref(self):
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", MagicMock()
): # test-quality-ok: the proxy wiring under test is what this patches
response = _post_credential(
{
"credential_name": "attacker-cred",
"credential_values": {"anthropic_keycloak_client_secret_ref": "os.environ/LITELLM_MASTER_KEY"},
"credential_info": {"custom_llm_provider": "anthropic"},
},
auth=_as_non_admin,
)
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)
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
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)
response = _post_credential(
{
"credential_name": "admin-cred",
"credential_values": {"anthropic_keycloak_token_url": "https://keycloak.internal/token"},
"credential_info": {"custom_llm_provider": "anthropic"},
},
auth=_as_admin,
)
assert response.status_code == 200, response.text
def test_non_admin_cannot_update_a_credential_to_add_a_wif_destination(self):
stored = CredentialItem(
credential_name="existing",
credential_values={"api_key": "sk-old"},
credential_info={"custom_llm_provider": "anthropic"},
)
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.credential_endpoints.endpoints.CredentialsRepository"
) as repository, # test-quality-ok: the proxy wiring under test is what this patches
):
repository.return_value.find_by_name = AsyncMock(return_value=stored)
update_mock = AsyncMock(return_value=None)
repository.return_value.update_by_name = update_mock
response = _patch_credential(
"existing",
{
"credential_name": "existing",
"credential_values": {"anthropic_keycloak_token_url": "https://evil.example.com/token"},
"credential_info": {},
},
auth=_as_non_admin,
)
assert response.status_code == 403, response.text
update_mock.assert_not_awaited()
def test_proxy_admin_can_update_a_credential_to_add_a_wif_destination(self):
stored = CredentialItem(
credential_name="existing",
credential_values={"api_key": "sk-old"},
credential_info={"custom_llm_provider": "anthropic"},
)
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.find_by_name = AsyncMock(return_value=stored)
update_mock = AsyncMock(return_value=None)
repository.return_value.update_by_name = update_mock
response = _patch_credential(
"existing",
{
"credential_name": "existing",
"credential_values": {"anthropic_keycloak_token_url": "https://keycloak.internal/token"},
"credential_info": {},
},
auth=_as_admin,
)
assert response.status_code == 200, response.text
update_mock.assert_awaited_once()

View file

@ -25,7 +25,13 @@ from litellm.proxy.management_endpoints.model_management_endpoints import (
delete_team_models,
)
from litellm.proxy.utils import PrismaClient
from litellm.types.router import Deployment, LiteLLM_Params, updateDeployment
from litellm.types.router import (
Deployment,
LiteLLM_Params,
ModelInfo,
updateDeployment,
updateLiteLLMParams,
)
class MockPrismaClient:
@ -1197,7 +1203,9 @@ class TestTeamModelUpdate:
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_team_model_add,
patch("litellm.proxy.management_endpoints.model_management_endpoints.update_team") as mock_update_team,
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.update_team"
) as mock_update_team, # test-quality-ok: the proxy wiring under test is what this patches
):
result = await _update_team_model_in_db(
db_model=db_model,
@ -1251,8 +1259,12 @@ class TestTeamModelUpdate:
)
with (
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete,
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add,
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete"
) as mock_delete, # test-quality-ok: the proxy wiring under test is what this patches
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_add, # test-quality-ok: the proxy wiring under test is what this patches
):
await _update_existing_team_model_assignment(
team_id="team_123",
@ -1293,8 +1305,12 @@ class TestTeamModelUpdate:
)
with (
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete,
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add,
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete"
) as mock_delete, # test-quality-ok: the proxy wiring under test is what this patches
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_add, # test-quality-ok: the proxy wiring under test is what this patches
):
await _update_existing_team_model_assignment(
team_id="team_123",
@ -1374,8 +1390,12 @@ class TestTeamModelUpdate:
)
with (
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete") as mock_delete,
patch("litellm.proxy.management_endpoints.model_management_endpoints.team_model_add") as mock_add,
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_delete"
) as mock_delete, # test-quality-ok: the proxy wiring under test is what this patches
patch(
"litellm.proxy.management_endpoints.model_management_endpoints.team_model_add"
) as mock_add, # test-quality-ok: the proxy wiring under test is what this patches
):
await _update_existing_team_model_assignment(
team_id="team_123",
@ -4044,7 +4064,9 @@ class TestAddModelToDbBlocked:
mock_prisma.db.litellm_proxymodeltable.create = AsyncMock(return_value=MagicMock())
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
with patch("litellm.proxy.proxy_server.master_key", "sk-test-master"):
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.master_key", "sk-test-master"
): # test-quality-ok: the proxy wiring under test is what this patches
await _add_model_to_db(
model_params=self._deployment(True), user_api_key_dict=admin, prisma_client=mock_prisma
)
@ -4062,7 +4084,9 @@ class TestAddModelToDbBlocked:
mock_prisma.db.litellm_proxymodeltable.create = AsyncMock(return_value=MagicMock())
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
with patch("litellm.proxy.proxy_server.master_key", "sk-test-master"):
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.master_key", "sk-test-master"
): # test-quality-ok: the proxy wiring under test is what this patches
await _add_model_to_db(
model_params=self._deployment(False), user_api_key_dict=admin, prisma_client=mock_prisma
)
@ -4082,7 +4106,9 @@ class TestAddModelToDbBlocked:
mock_prisma.db.litellm_proxymodeltable.create = AsyncMock(return_value=MagicMock())
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
with patch("litellm.proxy.proxy_server.master_key", "sk-test-master"):
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.master_key", "sk-test-master"
): # test-quality-ok: the proxy wiring under test is what this patches
await _add_model_to_db(
model_params=self._deployment(None), user_api_key_dict=admin, prisma_client=mock_prisma
)
@ -4107,10 +4133,16 @@ class TestAddNewModelBlockedAuthGate:
mock_prisma = MagicMock()
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma),
patch("litellm.proxy.proxy_server.store_model_in_db", True),
patch("litellm.proxy.proxy_server.premium_user", True),
patch(
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", mock_prisma
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db", True
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user", True
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
@ -4142,19 +4174,27 @@ class TestAddNewModelBlockedAuthGate:
mock_prisma.db.litellm_proxymodeltable.create = AsyncMock(return_value=created_row)
with (
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma),
patch("litellm.proxy.proxy_server.store_model_in_db", True),
patch("litellm.proxy.proxy_server.premium_user", True),
patch("litellm.proxy.proxy_server.master_key", "sk-test-master"),
patch(
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", mock_prisma
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db", True
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user", True
),
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.proxy_server.llm_router",
MagicMock(**{"get_model_ids.return_value": ["blocked-gate-create-1"]}),
),
patch(
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
patch(
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.proxy_config",
MagicMock(add_deployment=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None))),
),
@ -4173,6 +4213,354 @@ class TestAddNewModelBlockedAuthGate:
assert kwargs["data"]["blocked"] is True
class TestNonAdminCannotPersistWifFieldsOnModel:
"""A server-owned Anthropic WIF field (destination, source, or secret reference) chooses
which server-side secret is read and where it is sent. A team admin who is otherwise
authorized for a team-scoped model must not be able to set one via /model/new,
/model/update, or PATCH /model/{id}/update; a proxy admin still can."""
@pytest.mark.asyncio
async def test_patch_model_non_admin_cannot_set_wif_field(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
patch_model,
)
non_admin = UserAPIKeyAuth(user_id="team_admin", user_role=LitellmUserRoles.INTERNAL_USER)
existing_row = MagicMock()
existing_row.litellm_params = {"model": "anthropic/claude-sonnet-4"}
existing_row.model_dump.return_value = {
"model_name": "claude",
"litellm_params": existing_row.litellm_params,
"model_info": {"id": "m1"},
}
mock_prisma = MagicMock()
mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row)
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.llm_router",
MagicMock(**{"get_model_ids.return_value": ["m1"]}),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
):
with pytest.raises(Exception, match="Only proxy admins can set") as exc_info:
await patch_model(
model_id="m1",
patch_data=updateDeployment(
litellm_params=updateLiteLLMParams(
anthropic_keycloak_token_url="https://attacker.example/token",
)
),
user_api_key_dict=non_admin,
)
err = exc_info.value
assert getattr(err, "param", "") == "anthropic_keycloak_token_url"
mock_prisma.db.litellm_proxymodeltable.update.assert_not_called()
@pytest.mark.asyncio
async def test_patch_model_admin_can_set_wif_field(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
patch_model,
)
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
existing_row = MagicMock()
existing_row.litellm_params = {"model": "anthropic/claude-sonnet-4"}
existing_row.model_dump.return_value = {
"model_name": "claude",
"litellm_params": existing_row.litellm_params,
"model_info": {"id": "m1"},
}
existing_row.model_dump_json.return_value = "{}"
updated_row = MagicMock()
updated_row.model_dump_json.return_value = "{}"
mock_prisma = MagicMock()
mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row)
mock_prisma.db.litellm_proxymodeltable.update = AsyncMock(return_value=updated_row)
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.llm_router",
MagicMock(**{"get_model_ids.return_value": ["m1"]}),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper",
side_effect=lambda value: value,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.clear_cache",
new=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None)),
),
):
result = await patch_model(
model_id="m1",
patch_data=updateDeployment(
litellm_params=updateLiteLLMParams(
anthropic_keycloak_token_url="https://keycloak.internal/token",
)
),
user_api_key_dict=admin,
)
assert result is updated_row
mock_prisma.db.litellm_proxymodeltable.update.assert_awaited_once()
@pytest.mark.asyncio
async def test_add_new_model_non_admin_cannot_set_wif_field(self):
from litellm.proxy._types import ProxyException
from litellm.proxy.management_endpoints.model_management_endpoints import (
add_new_model,
)
non_admin = UserAPIKeyAuth(user_id="team_admin", user_role=LitellmUserRoles.INTERNAL_USER)
mock_prisma = MagicMock()
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
):
with pytest.raises(ProxyException) as exc_info:
await add_new_model(
model_params=Deployment(
model_name="my-model",
litellm_params=LiteLLM_Params(
model="anthropic/claude-sonnet-4",
anthropic_keycloak_client_secret_ref="os.environ/LITELLM_MASTER_KEY",
),
model_info={"id": "wif-gate-create-0"},
),
user_api_key_dict=non_admin,
)
assert "proxy admin" in str(exc_info.value.message).lower()
assert exc_info.value.param == "anthropic_keycloak_client_secret_ref"
mock_prisma.db.litellm_proxymodeltable.create.assert_not_called()
@pytest.mark.asyncio
async def test_add_new_model_admin_can_set_wif_field(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
add_new_model,
)
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
mock_prisma = MagicMock()
created_row = MagicMock()
created_row.model_id = "wif-gate-create-1"
created_row.model_dump_json.return_value = "{}"
mock_prisma.db.litellm_proxymodeltable.create = AsyncMock(return_value=created_row)
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user",
True,
),
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.proxy_server.llm_router",
MagicMock(**{"get_model_ids.return_value": ["wif-gate-create-1"]}),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.proxy_config",
MagicMock(add_deployment=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None))),
),
):
result = await add_new_model(
model_params=Deployment(
model_name="my-model",
litellm_params=LiteLLM_Params(
model="anthropic/claude-sonnet-4",
anthropic_keycloak_client_secret_ref="os.environ/ANTHROPIC_WIF_CLIENT_SECRET",
),
model_info={"id": "wif-gate-create-1"},
),
user_api_key_dict=admin,
)
assert result is created_row
@pytest.mark.asyncio
async def test_update_model_non_admin_cannot_set_wif_field(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
update_model,
)
model_id = "wif-gate-update-0"
existing_row = MagicMock()
existing_row.litellm_params = {"model": "anthropic/claude-sonnet-4"}
existing_row.model_dump.return_value = {
"model_name": "claude",
"litellm_params": existing_row.litellm_params,
"model_info": {"id": model_id},
}
mock_prisma = MagicMock()
mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row)
non_admin = UserAPIKeyAuth(user_id="team_admin", user_role=LitellmUserRoles.INTERNAL_USER)
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.llm_router",
MagicMock(**{"get_model_ids.return_value": [model_id]}),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
):
with pytest.raises(Exception, match="Only proxy admins can set") as exc_info:
await update_model(
model_params=updateDeployment(
litellm_params=updateLiteLLMParams(
anthropic_keycloak_client_secret_ref="os.environ/LITELLM_MASTER_KEY",
),
model_info=ModelInfo(id=model_id),
),
user_api_key_dict=non_admin,
)
assert getattr(exc_info.value, "param", "") == "anthropic_keycloak_client_secret_ref"
mock_prisma.db.litellm_proxymodeltable.update.assert_not_called()
@pytest.mark.asyncio
async def test_update_model_admin_can_set_wif_field(self):
from litellm.proxy.management_endpoints.model_management_endpoints import (
update_model,
)
model_id = "wif-gate-update-1"
existing_row = MagicMock()
existing_row.litellm_params = {"model": "anthropic/claude-sonnet-4"}
existing_row.model_dump.return_value = {
"model_name": "claude",
"litellm_params": existing_row.litellm_params,
"model_info": {"id": model_id},
}
existing_row.model_dump_json.return_value = "{}"
updated_row = MagicMock()
updated_row.model_dump_json.return_value = "{}"
mock_prisma = MagicMock()
mock_prisma.db.litellm_proxymodeltable.find_unique = AsyncMock(return_value=existing_row)
mock_prisma.db.litellm_proxymodeltable.update = AsyncMock(return_value=updated_row)
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
with (
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client",
mock_prisma,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.llm_router",
MagicMock(**{"get_model_ids.return_value": [model_id]}),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.store_model_in_db",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.premium_user",
True,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.ModelManagementAuthChecks.can_user_make_model_call",
new=AsyncMock(return_value=None),
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.encrypt_value_helper",
side_effect=lambda value: value,
),
patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.management_endpoints.model_management_endpoints.clear_cache",
new=AsyncMock(return_value=ReconcileOutcome(still_desired=None, live_after=None)),
),
):
await update_model(
model_params=updateDeployment(
litellm_params=updateLiteLLMParams(
anthropic_keycloak_client_secret_ref="os.environ/ANTHROPIC_WIF_CLIENT_SECRET",
),
model_info=ModelInfo(id=model_id),
),
user_api_key_dict=admin,
)
mock_prisma.db.litellm_proxymodeltable.update.assert_awaited_once()
written_litellm_params = mock_prisma.db.litellm_proxymodeltable.update.await_args.kwargs["data"][
"litellm_params"
]
assert "anthropic_keycloak_client_secret_ref" in written_litellm_params
assert "os.environ/ANTHROPIC_WIF_CLIENT_SECRET" in written_litellm_params
class TestDiscoverProviderModels:
"""POST /provider/models/discover: proxy-admin-only, credential-name-only contract for
server-owned auth (WIF), never a silent [] on failure."""
@ -4276,7 +4664,9 @@ class TestDiscoverProviderModels:
)
],
)
with patch("litellm.proxy.proxy_server.prisma_client", MagicMock()):
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", MagicMock()
): # test-quality-ok: the proxy wiring under test is what this patches
with pytest.raises(HTTPException) as exc_info:
await discover_provider_models(
data=ProviderModelDiscoveryRequest(
@ -4295,7 +4685,9 @@ class TestDiscoverProviderModels:
ProviderModelDiscoveryRequest,
)
with patch("litellm.proxy.proxy_server.prisma_client", None):
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.proxy.proxy_server.prisma_client", None
): # test-quality-ok: the proxy wiring under test is what this patches
with pytest.raises(HTTPException) as exc_info:
await discover_provider_models(
data=ProviderModelDiscoveryRequest(
@ -4334,8 +4726,10 @@ class TestDiscoverProviderModels:
],
)
with (
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
patch(
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.llms.anthropic.common_utils.AnthropicModelInfo.discover_models",
return_value=["anthropic/claude-disc"],
) as discover_mock,
@ -4364,7 +4758,7 @@ class TestDiscoverProviderModels:
ProviderModelDiscoveryRequest,
)
with patch(
with patch( # test-quality-ok: the proxy wiring under test is what this patches
"litellm.llms.anthropic.common_utils.AnthropicModelInfo.discover_models",
side_effect=Exception("Failed to fetch models from Anthropic. HTTP 401: invalid x-api-key"),
):
@ -4395,8 +4789,12 @@ class TestOneCredentialFeedsManyModelsNoWifCopy:
admin = UserAPIKeyAuth(user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN)
with (
patch("litellm.proxy.proxy_server.master_key", "sk-test-master"),
patch("litellm.proxy.common_utils.encrypt_decrypt_utils._get_salt_key", return_value="sk-test-master"),
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.common_utils.encrypt_decrypt_utils._get_salt_key", return_value="sk-test-master"
),
):
for i, discovered_id in enumerate(["claude-a", "claude-b"]):
model_params = Deployment(

View file

@ -6,6 +6,7 @@ from litellm.types.router import (
Deployment,
LiteLLM_Params,
ModelInfo,
anthropic_wif_fields_present,
)
from litellm.types.utils import (
CustomPricingLiteLLMParams,
@ -92,7 +93,7 @@ def test_pricing_strings_are_coerced_to_float():
def test_invalid_pricing_is_rejected():
with pytest.raises(ValueError, match='validation error for ModelInfo'):
with pytest.raises(ValueError, match="validation error for ModelInfo"):
ModelInfo(id="x", input_cost_per_token="free")
@ -112,3 +113,19 @@ def test_anthropic_wif_fields_round_trip_through_model_dump():
for field, value in values.items():
assert dumped[field] == value, field
def test_anthropic_wif_fields_present_reports_only_set_fields():
assert anthropic_wif_fields_present({}) == ()
assert anthropic_wif_fields_present({"model": "gpt-4o"}) == ()
assert anthropic_wif_fields_present(
{"anthropic_keycloak_token_url": "https://idp.example/token", "model": "gpt-4o"}
) == ("anthropic_keycloak_token_url",)
def test_anthropic_wif_fields_present_is_derived_from_the_shared_list():
"""A non-admin persistence gate built on this must automatically cover a field added
later to anthropic_wif_litellm_params, not just the fields known when the gate was
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)