scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

fix(credentials): scope the proxy-admin gate to trace destinations

Browse source 
Credential create/update/delete and reads were gated to the proxy admin for every
credential, which tightened provider-credential management for keys an admin had
delegated /credentials to via allowed_routes; a backward-incompatible change beyond
the OTEL trace-destination feature.

Restrict the gate to trace destinations (credential_type="logging"). Those require the
proxy admin regardless of allowed_routes, so a non-admin cannot create a global
destination and receive other tenants' traces, nor convert a provider credential into
one. Provider credentials keep their existing route-level authorization. Reads follow
suit: the list hides destinations from non-admins and by_name returns 403 for a
destination, while provider credentials read as before.
This commit is contained in:
Yucheng Zhu 2026-07-29 16:34:17 -07:00
parent 440277c97b
commit 6829988c70
4 changed files with 247 additions and 187 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
litellm/proxy/credential_endpoints/endpoints.py
View file

@ -14,6 +14,7 @@ from litellm.litellm_core_utils.litellm_logging import _get_masked_values
from litellm.proxy._types import CommonProxyErrors, LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.encrypt_decrypt_utils import encrypt_value_helper
from litellm.proxy.management_endpoints.logging_exporter_access import is_logging_credential
from litellm.proxy.management_endpoints.logging_exporter_validation import (
validate_credential_access,
)
@ -40,6 +41,22 @@ def _is_proxy_admin(user_api_key_dict: UserAPIKeyAuth) -> bool:
return user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
def _is_admin_tier(user_api_key_dict: UserAPIKeyAuth) -> bool:
return _is_proxy_admin(user_api_key_dict) or (user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY)
def _require_proxy_admin_for_logging(user_api_key_dict: UserAPIKeyAuth, *credential_infos: object) -> None:
"""Require proxy admin only when a trace destination is involved.
Trace destinations (``credential_type == "logging"``) are admin-managed regardless of a
key's ``allowed_routes``; provider credentials keep their existing route-level authorization.
Passing several infos (e.g. the stored credential and an update patch) gates the write when
any of them is a destination, which also blocks converting a provider credential into one.
"""
if any(is_logging_credential(info) for info in credential_infos):
_require_proxy_admin(user_api_key_dict)
class CredentialHelperUtils:
@staticmethod
def encrypt_credential_values(
@ -77,7 +94,7 @@ async def create_credential(
"""
from litellm.proxy.proxy_server import llm_router, prisma_client
_require_proxy_admin(user_api_key_dict)
_require_proxy_admin_for_logging(user_api_key_dict, credential.credential_info)
validate_credential_access(credential.credential_info)
try:
@ -144,26 +161,24 @@ async def get_credentials(
"""
[BETA] endpoint. This might change unexpectedly.
Proxy-admin only (a proxy-admin-viewer may read). Credentials, including
admin-owned logging destinations, are managed exclusively by the proxy admin;
tenants never read them over the API. Secret values are masked for both
admin-tier readers, exactly as they were before this feature.
Lists credentials with secret values masked. Trace destinations
(``credential_type == "logging"``) are shown only to the proxy admin or a
proxy-admin-viewer; other callers see provider credentials only. Destinations
and their ``access`` scoping stay proxy-admin information.
"""
try:
is_proxy_admin = _is_proxy_admin(user_api_key_dict)
is_admin_viewer = user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
if not (is_proxy_admin or is_admin_viewer):
raise HTTPException(
status_code=403,
detail={"error": CommonProxyErrors.not_allowed_access.value},
)
visible_credentials = (
litellm.credential_list
if _is_admin_tier(user_api_key_dict)
else [c for c in litellm.credential_list if not is_logging_credential(c.credential_info)]
)
masked_credentials = [
{
"credential_name": credential.credential_name,
"credential_values": _get_masked_values(credential.credential_values),
"credential_info": credential.credential_info,
}
for credential in litellm.credential_list
for credential in visible_credentials
]
return {"success": True, "credentials": masked_credentials}
except HTTPException:
@ -190,6 +205,11 @@ async def get_credential_by_name(
try:
for credential in litellm.credential_list:
if credential.credential_name == credential_name:
if is_logging_credential(credential.credential_info) and not _is_admin_tier(user_api_key_dict):
raise HTTPException(
status_code=403,
detail={"error": CommonProxyErrors.not_allowed_access.value},
)
masked_credential = CredentialItem(
credential_name=credential.credential_name,
credential_values=_get_masked_values(
@ -204,6 +224,8 @@ async def get_credential_by_name(
status_code=404,
detail="Credential not found. Got credential name: " + credential_name,
)
except HTTPException:
raise
except Exception as e:
verbose_proxy_logger.exception(e)
raise handle_exception_on_proxy(e)
@ -267,19 +289,25 @@ async def delete_credential(
"""
from litellm.proxy.proxy_server import prisma_client
_require_proxy_admin(user_api_key_dict)
try:
if prisma_client is None:
raise HTTPException(
status_code=500,
detail={"error": CommonProxyErrors.db_not_connected_error.value},
)
await CredentialsRepository(prisma_client).delete_by_name(credential_name)
credentials_repository = CredentialsRepository(prisma_client)
db_credential = await credentials_repository.find_by_name(credential_name)
_require_proxy_admin_for_logging(
user_api_key_dict,
db_credential.credential_info if db_credential is not None else None,
)
await credentials_repository.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)
@ -371,10 +399,10 @@ async def update_credential(
"""
[BETA] endpoint. This might change unexpectedly.
Proxy-admin only. Credentials, including admin-owned logging destinations and
their ``access`` scoping, are managed exclusively by the proxy admin.
Updating a trace destination (``credential_type == "logging"``), or converting a
credential into one, requires the proxy admin. Provider credentials keep their
existing route-level authorization.
"""
_require_proxy_admin(user_api_key_dict)
from litellm.proxy.proxy_server import prisma_client
validate_credential_access(credential.credential_info)
@ -389,6 +417,7 @@ async def update_credential(
db_credential = await credentials_repository.find_by_name(credential_name)
if db_credential is None:
raise HTTPException(status_code=404, detail="Credential not found in DB.")
_require_proxy_admin_for_logging(user_api_key_dict, db_credential.credential_info, credential.credential_info)
merged_credential = update_db_credential(db_credential, _patch_to_credential_item(credential, credential_name))
credential_object_jsonified = jsonify_object(merged_credential.model_dump())
await credentials_repository.update_by_name(
@ -404,6 +433,8 @@ async def update_credential(
patch=credential,
)
return {"success": True, "message": "Credential updated successfully"}
except HTTPException:
raise
except Exception as e:
return handle_exception_on_proxy(e)

10
litellm/proxy/management_endpoints/logging_exporter_access.py
View file

@ -32,6 +32,16 @@ def parse_credential_info(raw: object) -> CredentialInfo | None:
return None
def is_logging_credential(raw: object) -> bool:
"""Whether ``credential_info`` marks a trace destination (``credential_type == "logging"``).
Trace destinations are proxy-admin-managed regardless of a key's ``allowed_routes``;
provider credentials are not logging and keep their existing route-level authorization.
"""
info = parse_credential_info(raw)
return info is not None and info.credential_type == "logging"
def identity_scope(team_id: str | None, org_id: str | None) -> tuple[frozenset[str], frozenset[str]]:
"""A single request identity's scope as ``(team_ids, org_ids)`` for
``access_grants``."""

342
tests/test_litellm/proxy/credential_endpoints/test_endpoints.py
View file

@ -1,11 +1,14 @@
"""Admin-gating on credential mutations.
"""Authorization on credential endpoints, scoped to trace destinations.
Every ``/credentials`` operation -- GET, POST, PATCH, DELETE, for both logging
destinations and provider credentials -- is proxy-admin only (a proxy-admin-viewer
may read). Admin-owned OTEL logging destinations and their ``access`` scoping are
managed exclusively by the proxy admin; tenants never read or mutate them over the
API. Trace routing to identity-scoped destinations happens server-side in the
resolver, independent of this surface.
Trace destinations (``credential_type == "logging"``) are proxy-admin-managed
regardless of a key's ``allowed_routes``: create/update/delete require the proxy
admin, and reads of a destination are limited to the proxy admin (or a
proxy-admin-viewer). Provider credentials are not trace destinations, so the
handlers do not gate them by role; their authorization stays at the route layer
(a non-admin only reaches ``/credentials`` when an admin delegated it via
``allowed_routes``), exactly as it was before the destinations feature. Trace
routing to identity-scoped destinations happens server-side in the resolver,
independent of this surface.
"""
import os
@ -33,6 +36,23 @@ def _member():
_LOGGING_INFO = {"credential_type": "logging", "description": "langfuse_otel"}
_PROVIDER_INFO = {"custom_llm_provider": "openai"}
def _logging_cred(name="dest"):
return CredentialItem(
credential_name=name,
credential_values={"langfuse_host": "h"},
credential_info=_LOGGING_INFO,
)
def _provider_cred(name="openai-prod"):
return CredentialItem(
credential_name=name,
credential_values={"api_key": "sk-real"},
credential_info=_PROVIDER_INFO,
)
@pytest.fixture
@ -43,16 +63,19 @@ def _connected_db(monkeypatch):
monkeypatch.setenv("LITELLM_SALT_KEY", "sk-test-salt-key")
monkeypatch.setattr(proxy_server, "prisma_client", MagicMock())
monkeypatch.setattr(proxy_server, "llm_router", None)
monkeypatch.setattr(litellm, "credential_list", [])
repo = MagicMock()
repo.create = AsyncMock()
repo.delete_by_name = AsyncMock()
repo.update_by_name = AsyncMock()
repo.find_by_name = AsyncMock(return_value=None)
monkeypatch.setattr(endpoints, "CredentialsRepository", lambda _client: repo)
monkeypatch.setattr(
endpoints.CredentialAccessor, "upsert_credentials", lambda creds: None
)
monkeypatch.setattr(endpoints.CredentialAccessor, "upsert_credentials", lambda creds: None)
return repo
# --- create ------------------------------------------------------------------
@pytest.mark.asyncio
async def test_create_logging_credential_forbidden_for_non_admin(_connected_db):
with pytest.raises(HTTPException) as exc:
@ -87,16 +110,36 @@ async def test_create_logging_credential_allowed_for_admin(_connected_db):
@pytest.mark.asyncio
async def test_create_provider_credential_forbidden_for_non_admin(_connected_db):
"""POST is proxy-admin only for provider and logging credentials alike."""
async def test_create_provider_credential_allowed_for_non_admin(_connected_db):
"""Provider credentials are not trace destinations, so create is not gated by
role in the handler: a key that route-auth admitted (delegated via
``allowed_routes``) keeps its pre-feature ability to create one."""
result = await endpoints.create_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CreateCredentialItem(
credential_name="openai",
credential_values={"api_key": "sk"},
credential_info=_PROVIDER_INFO,
),
user_api_key_dict=_member(),
)
assert result["success"] is True
_connected_db.create.assert_awaited_once()
@pytest.mark.asyncio
async def test_create_provider_credential_cannot_smuggle_destination(_connected_db):
"""A non-admin cannot create a destination by tagging a 'provider' create with
credential_type=logging: the gate keys off the payload's type, not its name."""
with pytest.raises(HTTPException) as exc:
await endpoints.create_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CreateCredentialItem(
credential_name="openai",
credential_values={"api_key": "sk"},
credential_info={"custom_llm_provider": "openai"},
credential_name="looks-like-provider",
credential_values={"otel_endpoint": "https://attacker/v1/traces"},
credential_info={"credential_type": "logging", "access": {"global": True}},
),
user_api_key_dict=_member(),
)
@ -104,53 +147,74 @@ async def test_create_provider_credential_forbidden_for_non_admin(_connected_db)
_connected_db.create.assert_not_awaited()
# --- update ------------------------------------------------------------------
@pytest.mark.asyncio
async def test_update_logging_credential_forbidden_for_non_admin(_connected_db):
_connected_db.find_by_name = AsyncMock(return_value=_logging_cred())
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="dest",
credential_values={},
credential_info={"access": {"global": True}},
),
credential=UpdateCredentialItem(credential_info={"access": {"global": True}}),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.update_by_name.assert_not_awaited()
@pytest.mark.asyncio
async def test_update_existing_logging_credential_forbidden_even_without_logging_patch(
_connected_db, monkeypatch
):
"""A non-admin cannot edit a stored logging credential's values, even with a patch
that omits credential_info (the gate consults the in-memory credential too)."""
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "h"},
credential_info=_LOGGING_INFO,
)
],
)
async def test_update_existing_logging_credential_forbidden_even_without_logging_patch(_connected_db):
"""A non-admin cannot edit a stored destination's values, even with a patch that
omits credential_info: the gate consults the stored (DB) credential too."""
_connected_db.find_by_name = AsyncMock(return_value=_logging_cred())
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "evil"},
credential_info={},
),
credential=UpdateCredentialItem(credential_values={"langfuse_host": "evil"}),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.update_by_name.assert_not_awaited()
@pytest.mark.asyncio
async def test_update_provider_credential_allowed_for_non_admin(_connected_db):
"""A non-admin (delegated via allowed_routes) may still patch a provider
credential; only destinations are handler-gated."""
_connected_db.find_by_name = AsyncMock(return_value=_provider_cred())
result = await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=UpdateCredentialItem(credential_values={"api_key": "sk-rotated"}),
credential_name="openai-prod",
user_api_key_dict=_member(),
)
assert result["success"] is True
_connected_db.update_by_name.assert_awaited_once()
@pytest.mark.asyncio
async def test_update_provider_to_destination_forbidden_for_non_admin(_connected_db):
"""Converting a provider credential into a destination (patch adds a logging
credential_type / access) requires the proxy admin, so a non-admin can't
escalate a delegated provider credential into a global trace sink."""
_connected_db.find_by_name = AsyncMock(return_value=_provider_cred())
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=UpdateCredentialItem(
credential_info={"credential_type": "logging", "access": {"global": True}},
),
credential_name="openai-prod",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.update_by_name.assert_not_awaited()
def test_update_db_credential_preserves_existing_info_on_partial_patch():
@ -222,131 +286,52 @@ def test_update_db_credential_preserves_untouched_access_subfields():
}
# --- delete ------------------------------------------------------------------
@pytest.mark.asyncio
async def test_delete_logging_credential_forbidden_for_non_admin(
_connected_db, monkeypatch
):
async def test_delete_logging_credential_forbidden_for_non_admin(_connected_db):
_connected_db.find_by_name = AsyncMock(return_value=_logging_cred())
with pytest.raises(HTTPException) as exc:
await endpoints.delete_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.delete_by_name.assert_not_awaited()
@pytest.mark.asyncio
async def test_delete_provider_credential_allowed_for_non_admin(_connected_db):
"""Deleting a provider credential is not handler-gated (route-auth governs it),
so a delegated non-admin keeps its pre-feature ability to delete one."""
_connected_db.find_by_name = AsyncMock(return_value=_provider_cred())
result = await endpoints.delete_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential_name="openai-prod",
user_api_key_dict=_member(),
)
assert result["success"] is True
_connected_db.delete_by_name.assert_awaited_once()
# --- reads -------------------------------------------------------------------
@pytest.mark.asyncio
async def test_get_credentials_hides_destinations_from_non_admin(monkeypatch):
"""A non-admin list returns provider credentials (pre-feature behavior) but
never trace destinations, which stay proxy-admin information."""
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="dest",
credential_values={},
credential_info=_LOGGING_INFO,
)
],
)
with pytest.raises(HTTPException) as exc:
await endpoints.delete_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.delete_by_name.assert_not_awaited()
@pytest.mark.asyncio
async def test_update_db_only_logging_credential_forbidden_for_non_admin(
_connected_db, monkeypatch
):
"""A logging credential that exists ONLY in the DB (not resident in the
in-memory ``credential_list`` -- e.g. created on another scaled instance or
before a restart) must still gate a non-admin update. The gate falls back to
the DB so a credential_values-only patch can't redirect a logging
destination's endpoint without the proxy-admin check."""
monkeypatch.setattr(litellm, "credential_list", []) # nothing in memory
_connected_db.find_by_name = AsyncMock(
return_value=CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "h"},
credential_info=_LOGGING_INFO,
)
)
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "evil"},
credential_info={},
credential_name="openai",
credential_values={"api_key": "sk-secret"},
credential_info=_PROVIDER_INFO,
),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
@pytest.mark.asyncio
async def test_delete_db_only_logging_credential_forbidden_for_non_admin(
_connected_db, monkeypatch
):
"""Same DB-only fallback for delete: a non-admin can't delete a logging
credential that is resident only in the DB."""
monkeypatch.setattr(litellm, "credential_list", [])
_connected_db.find_by_name = AsyncMock(
return_value=CredentialItem(
credential_name="dest",
credential_values={},
credential_info=_LOGGING_INFO,
)
)
with pytest.raises(HTTPException) as exc:
await endpoints.delete_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential_name="dest",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.delete_by_name.assert_not_awaited()
# --- PATCH gating (proxy-admin only) ----------------------------------------
@pytest.mark.asyncio
async def test_provider_credential_patch_forbidden_for_non_admin(
_connected_db, monkeypatch
):
"""A non-admin cannot PATCH a provider credential (or any credential):
without the gate a non-admin could rotate the upstream api_key."""
provider_cred = CredentialItem(
credential_name="openai-prod",
credential_values={"api_key": "sk-real"},
credential_info={"custom_llm_provider": "openai"},
)
monkeypatch.setattr(litellm, "credential_list", [provider_cred])
_connected_db.find_by_name = AsyncMock(return_value=provider_cred)
_connected_db.update_by_name = AsyncMock()
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="openai-prod",
credential_values={"api_key": "sk-stolen"},
credential_info={},
),
credential_name="openai-prod",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
_connected_db.update_by_name.assert_not_awaited()
@pytest.mark.asyncio
async def test_get_credentials_forbidden_for_non_admin(monkeypatch):
"""A non-proxy-admin (team-admin, org-admin, or plain internal_user) gets 403.
Credentials, including admin-owned logging destinations, are proxy-admin only;
the list is never exposed to a tenant over the API."""
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="poc-langfuse",
credential_values={"public_key": "pk-1"},
@ -354,15 +339,42 @@ async def test_get_credentials_forbidden_for_non_admin(monkeypatch):
),
],
)
response = await endpoints.get_credentials(
request=MagicMock(),
fastapi_response=MagicMock(),
user_api_key_dict=_member(),
)
names = [c["credential_name"] for c in response["credentials"]]
assert names == ["openai"] # destination hidden, provider visible
@pytest.mark.asyncio
async def test_get_credential_by_name_destination_forbidden_for_non_admin(monkeypatch):
monkeypatch.setattr(litellm, "credential_list", [_logging_cred("poc-langfuse")])
with pytest.raises(HTTPException) as exc:
await endpoints.get_credentials(
await endpoints.get_credential_by_name(
request=MagicMock(),
fastapi_response=MagicMock(),
credential_name="poc-langfuse",
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
@pytest.mark.asyncio
async def test_get_credential_by_name_provider_allowed_for_non_admin(monkeypatch):
"""A provider credential read by name is not handler-gated (masked as before)."""
monkeypatch.setattr(litellm, "credential_list", [_provider_cred("openai-prod")])
result = await endpoints.get_credential_by_name(
request=MagicMock(),
fastapi_response=MagicMock(),
credential_name="openai-prod",
user_api_key_dict=_member(),
)
assert result.credential_name == "openai-prod"
assert result.credential_values["api_key"] != "sk-real" # still masked
def test_patch_credentials_route_targets_update_credential():
"""Regression: the @router.patch decorator on /credentials/{name:path} must
decorate update_credential, not one of the extracted helpers. A misplaced
@ -420,9 +432,7 @@ async def test_get_credentials_returns_all_for_proxy_admin(monkeypatch):
)
names = sorted(c["credential_name"] for c in response["credentials"])
assert names == ["generic-otel", "openai", "poc-langfuse"]
generic = next(
c for c in response["credentials"] if c["credential_name"] == "generic-otel"
)
generic = next(c for c in response["credentials"] if c["credential_name"] == "generic-otel")
# otel_headers carries the collector auth token, so the masker treats it as a
# secret key: readable prefix only, never the full value.
assert generic["credential_values"]["otel_headers"] != raw_headers
@ -457,9 +467,7 @@ async def test_get_credentials_admin_viewer_reads_same_masked_list_as_admin(monk
viewer_response = await endpoints.get_credentials(
request=MagicMock(),
fastapi_response=MagicMock(),
user_api_key_dict=UserAPIKeyAuth(
api_key="k", user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
),
user_api_key_dict=UserAPIKeyAuth(api_key="k", user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY),
)
admin_response = await endpoints.get_credentials(
request=MagicMock(),
@ -471,5 +479,3 @@ async def test_get_credentials_admin_viewer_reads_same_masked_list_as_admin(monk
assert names == ["generic-otel", "openai"]
assert "collector-secret" not in str(viewer_response)
assert "sk-secret-value" not in str(viewer_response)

13
tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py
View file

@ -15,6 +15,7 @@ from litellm.models.credentials import CredentialAccess, CredentialInfo, Credent
from litellm.proxy.management_endpoints.logging_exporter_access import (
access_grants,
identity_scope,
is_logging_credential,
parse_credential_info,
resolved_logging_exporter_names,
)
@ -192,3 +193,15 @@ def test_resolved_names_empty_scope_gets_global_only(monkeypatch):
def test_resolved_names_empty_registry(monkeypatch):
monkeypatch.setattr(litellm, "credential_list", [])
assert resolved_logging_exporter_names("t1", "o1") == ()
def test_is_logging_credential_distinguishes_destinations_from_provider_creds():
# A trace destination is tagged credential_type=logging.
assert is_logging_credential({"credential_type": "logging", "access": {"global": True}}) is True
# Provider credentials are not destinations.
assert is_logging_credential({"custom_llm_provider": "openai"}) is False
# An access map without the logging type must not count (can't smuggle a destination).
assert is_logging_credential({"access": {"global": True}}) is False
# Absent / malformed info fails closed to "not a destination".
assert is_logging_credential(None) is False
assert is_logging_credential("not-a-dict") is False