scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-15 23:31:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 165

refactor(otel/v2): scope admin-owned trace destinations to proxy admin

Browse source 
Admin-owned OTEL v2 logging destinations and their access scoping are now managed
only by the proxy admin. Trace routing to identity-scoped destinations is unchanged
because it runs server-side in the resolver; this removes only the tenant-facing
read/write surface the earlier revision exposed.

GET/POST/PATCH/DELETE /credentials are proxy-admin only again (a proxy-admin-viewer
may read); the two credential routes leave self_managed_routes, and the non-admin
scoped list, the team-admin PATCH self-service grant, and the access_decision decider
are deleted. Assigning logging_exporters on a key, team, or org is proxy-admin only,
dropping the team-admin and org-admin widening. In the UI the logging destinations
table and the exporter picker render only for a proxy admin, so non-admins no longer
call GET /credentials.
This commit is contained in:
Yucheng Zhu 2026-07-27 19:41:03 -07:00
parent b0484bae70
commit 04e4effba3
14 changed files with 97 additions and 1652 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/proxy/_types.py
View file

@ -777,12 +777,6 @@ class LiteLLMRoutes(enum.Enum):
# Team guardrail submissions - endpoint scopes results to caller's teams (non-admin)
"/guardrails/submissions",
"/guardrails/submissions/{guardrail_id}",
# Logging-credential routes. GET filters to logging-typed for non-admins;
# PATCH delegates to decide_credential_patch in credential_endpoints, which
# only allows a team-admin to append their own team_id to access.teams.
# POST and DELETE stay proxy-admin only via is_admin_gated_credential_info.
"/credentials",
"/credentials/{credential_name}",
] # routes that manage their own allowed/disallowed logic
## Org Admin Routes ##

133
litellm/proxy/credential_endpoints/access_decision.py
View file

@ -1,133 +0,0 @@
"""Pure tagged-union decision for who can patch a logging-credential.
A logging credential controls where other tenants' traces are exported, so
``credential_info.access`` is the only field a non-admin caller may touch, and
only to add their own team_id(s) to ``access.teams``. Everything else
(values, host, type, description, ``global``, ``orgs``, foreign team_ids,
or removing existing grants) stays proxy-admin only.
The decision is a value (Allow vs Deny(reason)), kept separate from the
endpoint so it can be unit-tested exhaustively without spinning up FastAPI.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal, Mapping
from litellm.models.credentials import CredentialAccess, CredentialInfo
OPAQUE_DENY_REASON = "Only the proxy admin can manage logging credentials"
@dataclass(frozen=True, slots=True)
class Allow:
tag: Literal["allow"] = "allow"
@dataclass(frozen=True, slots=True)
class Deny:
reason: str
# When True the reason is derived from caller input (e.g. they typed a
# foreign team_id) and is safe to surface. When False the reason would
# confirm the stored credential is a logging destination; the endpoint
# collapses these to OPAQUE_DENY_REASON so PATCH /credentials/{name}
# can't be used as an existence oracle by a non-admin caller.
from_user_input: bool = False
tag: Literal["deny"] = "deny"
Decision = Allow | Deny
_IMMUTABLE_INFO_FIELDS = frozenset({"credential_type", "description", "host", "endpoint"})
def _patched_fields(info: CredentialInfo | None) -> frozenset[str]:
"""Names of credential_info fields the caller actually set in their patch."""
if info is None:
return frozenset()
return frozenset(info.model_fields_set) | frozenset(info.model_extra.keys() if info.model_extra else ())
def _access_teams(access: CredentialAccess | None) -> frozenset[str]:
return frozenset(access.teams) if access is not None else frozenset()
def decide_credential_patch(
*,
is_proxy_admin: bool,
caller_team_admin_ids: frozenset[str],
existing_info: CredentialInfo | None,
patch_info: CredentialInfo | None,
patch_values: Mapping[str, object] | None,
patch_name_changed: bool,
) -> Decision:
"""Return Allow or Deny(reason) for a PATCH /credentials/{name} request.
Proxy admins always pass. A team admin only passes when the patch (a) does
not change ``credential_values`` or ``credential_name``, (b) does not
modify any immutable ``credential_info`` field, and (c) limits its
``access`` change to appending team_ids the caller is team-admin of to
``access.teams`` (no removals, no foreign ids, no ``global``/``orgs``
edits). Emptying the last team grant is allowed: empty access is deny-all,
so it disables the destination rather than widening it.
"""
if is_proxy_admin:
return Allow()
if not caller_team_admin_ids:
return Deny(OPAQUE_DENY_REASON)
if patch_name_changed:
return Deny("credential_name is proxy-admin only")
if patch_values:
return Deny("credential_values is proxy-admin only")
touched = _patched_fields(patch_info)
if not touched:
return Deny("patch must set credential_info.access for team-admin writes")
forbidden = touched & _IMMUTABLE_INFO_FIELDS
if forbidden:
return Deny("credential_info fields are proxy-admin only: " + ", ".join(sorted(forbidden)))
if touched - {"access"}:
return Deny("team-admin may only patch credential_info.access; got: " + ", ".join(sorted(touched)))
assert patch_info is not None
patch_access = patch_info.access
if patch_access is None:
return Deny("credential_info.access must be set for team-admin writes")
# Touching global/orgs is allowed when the value matches the stored state
# (the UI's edit modal sends the full access object back so unchecking
# revokes; a no-op resend of global=false / orgs=[] must not be rejected).
# Only block when the caller would actually CHANGE these.
existing_access = existing_info.access if existing_info is not None else None
access_touched = frozenset(patch_access.model_fields_set)
existing_global = existing_access.global_ if existing_access is not None else False
existing_orgs = frozenset(existing_access.orgs) if existing_access is not None else frozenset()
if "global_" in access_touched and patch_access.global_ != existing_global:
return Deny("access.global is proxy-admin only")
if "orgs" in access_touched and frozenset(patch_access.orgs) != existing_orgs:
return Deny("access.orgs is proxy-admin only")
existing_teams = _access_teams(existing_access)
patch_teams = _access_teams(patch_access)
foreign_removed = (existing_teams - patch_teams) - caller_team_admin_ids
if foreign_removed:
# Do NOT echo the foreign team_ids -- they're stored values the
# caller didn't send, so naming them would leak access list members.
return Deny("team-admin may only revoke their own team grants")
foreign_added = (patch_teams - existing_teams) - caller_team_admin_ids
if foreign_added:
return Deny(
"team-admin may only grant their own team_ids: " + ", ".join(sorted(foreign_added)),
from_user_input=True,
)
return Allow()

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

@ -2,15 +2,10 @@
CRUD endpoints for storing reusable credentials.
"""
import asyncio
from dataclasses import dataclass
from typing import TYPE_CHECKING, Optional
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
if TYPE_CHECKING:
from litellm.proxy.utils import PrismaClient
import litellm
from litellm._logging import verbose_proxy_logger
from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
@ -18,21 +13,8 @@ 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 pydantic import ValidationError
from litellm.models.credentials import CredentialInfo
from litellm.proxy.credential_endpoints.access_decision import (
OPAQUE_DENY_REASON,
Allow,
Deny,
decide_credential_patch,
)
from litellm.proxy.management_endpoints.logging_exporter_access import (
is_destination_visible,
parse_credential_info,
)
from litellm.proxy.management_endpoints.logging_exporter_validation import (
is_admin_gated_credential_info,
validate_credential_access,
)
from litellm.proxy.utils import handle_exception_on_proxy, jsonify_object
@ -58,136 +40,6 @@ def _is_proxy_admin(user_api_key_dict: UserAPIKeyAuth) -> bool:
return user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
def _summarize_validation_error(ve: ValidationError) -> str:
parts = (".".join(str(loc) for loc in err["loc"]) + ": " + err["msg"] for err in ve.errors())
return "; ".join(parts)
@dataclass(frozen=True, slots=True)
class CallerAdminScope:
"""The teams and orgs a caller administers, for destination visibility.
``team_ids`` are the teams the caller admins directly (role=admin in
``members_with_roles``) unioned with every team in an org the caller is
org-admin of, since an org admin manages their org's teams even ones they
aren't a direct member of. ``org_ids`` are the orgs the caller is org-admin
of. The list endpoint matches a destination's ``access.teams`` against
``team_ids`` and ``access.orgs`` against ``org_ids``; the PATCH decider uses
``team_ids`` alone, since a team admin may only grant team ids.
"""
team_ids: frozenset[str]
org_ids: frozenset[str]
async def _caller_admin_scope(
user_api_key_dict: UserAPIKeyAuth, prisma_client: "PrismaClient | None"
) -> CallerAdminScope:
"""The teams and orgs the caller administers.
Empty when the caller has no user_id, no DB connection, or admins nothing.
Uses cached ``get_user_object`` / ``get_team_object`` plus one bounded query
for org teams; the role match is done in Python because ``members_with_roles``
is a JSON column. Best-effort: a lookup miss returns the empty scope, which
denies visibility and reduces the PATCH decider to no-ops (the safe fallback).
"""
if user_api_key_dict.user_id is None or prisma_client is None:
return CallerAdminScope(frozenset(), frozenset())
from litellm.proxy.auth.auth_checks import get_team_object, get_user_object
from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache
try:
user_obj = await get_user_object(
user_id=user_api_key_dict.user_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
user_id_upsert=False,
parent_otel_span=user_api_key_dict.parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
)
if user_obj is None:
return CallerAdminScope(frozenset(), frozenset())
# Direct team-admin grants: fetch the caller's teams concurrently, keep the
# ones where they hold the admin role.
own_team_ids = tuple(tid for tid in (getattr(user_obj, "teams", None) or []) if isinstance(tid, str))
team_objs = await asyncio.gather(
*(
get_team_object(
team_id=tid,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
parent_otel_span=user_api_key_dict.parent_otel_span,
proxy_logging_obj=proxy_logging_obj,
)
for tid in own_team_ids
)
)
team_admin_of = frozenset(
tid
for tid, team_obj in zip(own_team_ids, team_objs)
if any(
member.user_id == user_api_key_dict.user_id and member.role == "admin"
for member in (team_obj.members_with_roles or [])
)
)
# Org-admin grants: every team in any org the caller admins, even if the
# caller isn't a direct member of that team.
org_admin_of = frozenset(
m.organization_id
for m in (user_obj.organization_memberships or [])
if m.organization_id and m.user_role == LitellmUserRoles.ORG_ADMIN.value
)
org_teams = (
await prisma_client.db.litellm_teamtable.find_many(where={"organization_id": {"in": list(org_admin_of)}})
if org_admin_of
else []
)
org_grantable = frozenset(t.team_id for t in org_teams if t.team_id)
return CallerAdminScope(team_admin_of | org_grantable, org_admin_of)
except Exception: # noqa: BLE001 # fail closed to an empty admin scope so a lookup error never widens access
verbose_proxy_logger.exception("caller admin-scope lookup failed")
return CallerAdminScope(frozenset(), frozenset())
async def _caller_grantable_team_ids(
user_api_key_dict: UserAPIKeyAuth, prisma_client: "PrismaClient | None"
) -> frozenset[str]:
"""Team ids the caller may add to / remove from a destination's ``access.teams``
(the PATCH decider's grant scope)."""
return (await _caller_admin_scope(user_api_key_dict, prisma_client)).team_ids
def _credential_in_memory(credential_name: str) -> CredentialItem | None:
return next(
(cred for cred in litellm.credential_list if cred.credential_name == credential_name),
None,
)
async def _credential_for_admin_gate(credential_name: str, prisma_client: object) -> CredentialItem | None:
"""Authoritative credential lookup for the admin gate on update/delete.
The in-process ``litellm.credential_list`` can be stale: a credential created
via the API on another horizontally-scaled instance, or before a restart,
exists only in the DB. Gating on the in-memory copy alone would let a logging
credential that isn't resident be updated/deleted without the proxy-admin
check. Prefer the in-memory copy, fall back to the DB so the gate sees the
real ``credential_info``.
"""
existing = _credential_in_memory(credential_name)
if existing is not None:
return existing
if prisma_client is None:
return None
try:
return await CredentialsRepository(prisma_client).find_by_name(credential_name)
except Exception: # noqa: BLE001 # treat any lookup failure as credential-not-found
return None
class CredentialHelperUtils:
@staticmethod
def encrypt_credential_values(
@ -225,9 +77,6 @@ async def create_credential(
"""
from litellm.proxy.proxy_server import llm_router, prisma_client
# POST stays proxy-admin only across the board: route gate was widened so
# team-admins can PATCH access on existing logging destinations, but
# creation of any credential (logging or provider) remains admin-only.
_require_proxy_admin(user_api_key_dict)
validate_credential_access(credential.credential_info)
@ -295,41 +144,19 @@ async def get_credentials(
"""
[BETA] endpoint. This might change unexpectedly.
Proxy admins see every credential (values masked). A non-proxy-admin sees
only the logging destinations actually visible to a scope they administer:
the same ``is_destination_visible`` predicate the assignment validator and
the request-time resolver use, so the list can never show a destination a
caller could neither assign nor route to. Provider credentials, and logging
destinations scoped to other tenants, stay invisible. A caller who
administers nothing gets 403 (Veria F2).
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. Values are masked for the admin and
fully redacted for the admin-viewer.
"""
from litellm.proxy.proxy_server import prisma_client
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 is_proxy_admin or is_admin_viewer:
visible = list(litellm.credential_list)
else:
scope = await _caller_admin_scope(user_api_key_dict, prisma_client)
if not scope.team_ids and not scope.org_ids:
raise HTTPException(
status_code=403,
detail={
"error": (
"Listing logging destinations requires team-admin or "
"org-admin status. Ask your proxy admin to add you to a "
"team or org."
)
},
)
visible = [
credential
for credential in litellm.credential_list
if (info := parse_credential_info(credential.credential_info)) is not None
and info.credential_type == "logging"
and is_destination_visible(info, scope.team_ids, scope.org_ids)
]
if not (is_proxy_admin or is_admin_viewer):
raise HTTPException(
status_code=403,
detail={"error": CommonProxyErrors.not_allowed_access.value},
)
masked_credentials = [
{
"credential_name": credential.credential_name,
@ -340,7 +167,7 @@ async def get_credentials(
),
"credential_info": credential.credential_info,
}
for credential in visible
for credential in litellm.credential_list
]
return {"success": True, "credentials": masked_credentials}
except HTTPException:
@ -528,59 +355,6 @@ def _merge_credential_info(into: dict, patch: dict) -> None:
into["access"] = patch_access
async def _authorize_credential_patch(
*,
credential_name: str,
patch: UpdateCredentialItem,
existing: CredentialItem | None,
user_api_key_dict: UserAPIKeyAuth,
prisma_client: "PrismaClient | None",
) -> None:
"""Raise 403 unless the caller is allowed to apply ``patch`` to ``existing``.
The decider widening only applies when the STORED credential is a logging
destination -- a patch body alone can't promote a provider credential into
the decider's allowed paths (Cursor BugBot bypass: ``is_admin_gated_credential_info``
returned True for any patch carrying ``access``, so a team-admin could PATCH
``access.teams`` onto a provider credential and reach the decider).
"""
existing_is_logging_gated = existing is not None and is_admin_gated_credential_info(existing.credential_info)
if not existing_is_logging_gated:
_require_proxy_admin(user_api_key_dict)
return
is_admin = _is_proxy_admin(user_api_key_dict)
team_admin_ids = frozenset() if is_admin else await _caller_grantable_team_ids(user_api_key_dict, prisma_client)
try:
patch_info_typed = (
CredentialInfo.model_validate(patch.credential_info) if patch.credential_info is not None else None
)
except ValidationError as ve:
raise HTTPException(status_code=400, detail={"error": _summarize_validation_error(ve)})
assert existing is not None # narrowed by existing_is_logging_gated
try:
existing_info_typed = CredentialInfo.model_validate(existing.credential_info)
except ValidationError:
# Stored info the strict access model can't parse (e.g. a legacy access key)
# can't be run through the field-level decider. Fail closed: only the proxy
# admin may patch such a row, instead of 500-ing every caller.
if not is_admin:
raise HTTPException(status_code=403, detail={"error": OPAQUE_DENY_REASON})
return
decision = decide_credential_patch(
is_proxy_admin=is_admin,
caller_team_admin_ids=team_admin_ids,
existing_info=existing_info_typed,
patch_info=patch_info_typed,
patch_values=patch.credential_values,
patch_name_changed=(patch.credential_name is not None and patch.credential_name != credential_name),
)
if isinstance(decision, Deny):
reason = decision.reason if decision.from_user_input else OPAQUE_DENY_REASON
raise HTTPException(status_code=403, detail={"error": reason})
assert isinstance(decision, Allow)
def _patch_to_credential_item(patch: UpdateCredentialItem, credential_name: str) -> CredentialItem:
"""Translate the partial PATCH body into the legacy CredentialItem shape
the downstream merge expects (non-None dicts)."""
@ -606,21 +380,12 @@ async def update_credential(
"""
[BETA] endpoint. This might change unexpectedly.
Both ``credential_values`` and ``credential_info`` are optional; a team-admin
typically patches only ``credential_info.access`` to grant or revoke their
own team. A proxy admin may patch either or both. See
``decide_credential_patch`` for the exact contract.
Proxy-admin only. Credentials, including admin-owned logging destinations and
their ``access`` scoping, are managed exclusively by the proxy admin.
"""
_require_proxy_admin(user_api_key_dict)
from litellm.proxy.proxy_server import prisma_client
existing = await _credential_for_admin_gate(credential_name, prisma_client)
await _authorize_credential_patch(
credential_name=credential_name,
patch=credential,
existing=existing,
user_api_key_dict=user_api_key_dict,
prisma_client=prisma_client,
)
validate_credential_access(credential.credential_info)
try:

71
litellm/proxy/management_endpoints/key_management_endpoints.py
View file

@ -1557,10 +1557,6 @@ async def generate_key_fn(
"""
try:
from litellm.proxy._types import CommonProxyErrors
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
_is_user_team_admin,
)
from litellm.proxy.management_endpoints.logging_exporter_validation import (
validate_logging_exporter_field,
)
@ -1657,20 +1653,7 @@ async def generate_key_fn(
# proxy-admin only. Skip the role lookup when the field isn't in the payload
# to keep /key/generate cheap for the common case.
if data.logging_exporters is not None:
validate_logging_exporter_field(
data.logging_exporters,
user_api_key_dict,
caller_is_team_admin=(
team_table is not None
and _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_table)
),
caller_is_org_admin=(
team_table is not None
and await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_table)
),
scope_team_id=getattr(team_table, "team_id", None),
scope_org_id=getattr(team_table, "organization_id", None),
)
validate_logging_exporter_field(data.logging_exporters, user_api_key_dict)
if team_table is not None:
await _check_team_key_limits(
@ -1853,28 +1836,12 @@ async def generate_service_account_key_fn(
# eligible for service-account creation could set metadata.logging_exporters
# and route future traces to a destination they aren't allowed to assign
# (Veria F3). Skip the lookup unless the field is being written.
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
_is_user_team_admin,
)
from litellm.proxy.management_endpoints.logging_exporter_validation import (
validate_logging_exporter_field,
)
if data.logging_exporters is not None:
validate_logging_exporter_field(
data.logging_exporters,
user_api_key_dict,
caller_is_team_admin=(
team_table is not None and _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_table)
),
caller_is_org_admin=(
team_table is not None
and await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=team_table)
),
scope_team_id=getattr(team_table, "team_id", None),
scope_org_id=getattr(team_table, "organization_id", None),
)
validate_logging_exporter_field(data.logging_exporters, user_api_key_dict)
data.user_id = None # do not allow user_id to be set for service account keys
@ -2631,10 +2598,6 @@ async def update_key_fn( # noqa: C901 # single endpoint handling many optional
}'
```
"""
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
_is_user_team_admin,
)
from litellm.proxy.management_endpoints.logging_exporter_validation import (
validate_logging_exporter_field,
)
@ -2685,17 +2648,7 @@ async def update_key_fn( # noqa: C901 # single endpoint handling many optional
validate_logging_exporter_field(
data.logging_exporters,
user_api_key_dict,
caller_is_team_admin=(
_key_team is not None
and _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=_key_team)
),
caller_is_org_admin=(
_key_team is not None
and await _is_user_org_admin_for_team(user_api_key_dict=user_api_key_dict, team_obj=_key_team)
),
existing_exporters=getattr(existing_key_row, "logging_exporters", None),
scope_team_id=getattr(_key_team, "team_id", None),
scope_org_id=getattr(_key_team, "organization_id", None),
)
await _validate_update_key_data(
@ -4898,10 +4851,6 @@ async def regenerate_key_fn( # noqa: C901 # single endpoint handling many opti
# effective value doesn't change; pass the stored column value so a
# non-admin cannot clear an admin-assigned one.
if data is not None and data.logging_exporters is not None:
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
_is_user_team_admin,
)
from litellm.proxy.management_endpoints.logging_exporter_validation import (
validate_logging_exporter_field,
)
@ -4909,23 +4858,7 @@ async def regenerate_key_fn( # noqa: C901 # single endpoint handling many opti
validate_logging_exporter_field(
data.logging_exporters,
user_api_key_dict,
caller_is_team_admin=(
regenerate_team_table is not None
and _is_user_team_admin(
user_api_key_dict=user_api_key_dict,
team_obj=regenerate_team_table,
)
),
caller_is_org_admin=(
regenerate_team_table is not None
and await _is_user_org_admin_for_team(
user_api_key_dict=user_api_key_dict,
team_obj=regenerate_team_table,
)
),
existing_exporters=getattr(_key_in_db, "logging_exporters", None),
scope_team_id=getattr(regenerate_team_table, "team_id", None),
scope_org_id=getattr(regenerate_team_table, "organization_id", None),
)
verbose_proxy_logger.info(

108
litellm/proxy/management_endpoints/logging_exporter_validation.py
View file

@ -1,25 +1,16 @@
"""Validation for admin-owned logging-exporter assignment on key/team/org.
An identity's ``metadata.logging_exporters`` binds it to admin-owned trace
destinations. Every name must be a registered logging credential, the caller must
hold a role that authorizes the write (proxy admin always; team admin and org admin
in specific contexts), AND a non-proxy-admin may only name destinations whose
``credential_info.access`` makes them visible to the scope being written (the key's
team, or the team/org being updated). Visibility and enablement are separate: a
destination granted to a team is assignable by that team's admin, but assigning it is
what enables it. The resolver (``litellm_pre_call_utils``) re-checks visibility at
request time, so this gate and the resolver agree on what "visible" means.
destinations. Only the proxy admin may write it, and every name must be a registered
logging credential. Which identities a destination actually fires for is governed by
the destination's own ``credential_info.access``; the resolver
(``litellm_pre_call_utils``) evaluates that at request time.
"""
from fastapi import HTTPException, status
import litellm
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.logging_exporter_access import (
identity_scope,
is_destination_visible,
parse_credential_info,
)
LOGGING_EXPORTERS_KEY = "logging_exporters"
@ -82,42 +73,6 @@ def _logging_credential_names() -> set[str]:
return set(_logging_credentials_by_name())
def _reject_unassignable_destinations(
exporters: list[str],
*,
scope_team_id: str | None,
scope_org_id: str | None,
) -> None:
"""Reject names a non-proxy-admin cannot assign in this scope.
A destination is assignable when it is an explicit global/default
(``auto_enable``) or its ``access`` grants the scope being written (the key's
team, or the team/org being updated). Names are already known logging
credentials by the time this runs, so a missing entry means a benign race; we
fail closed on it.
"""
by_name = _logging_credentials_by_name()
team_ids, org_ids = identity_scope(scope_team_id, scope_org_id)
unassignable = [
name
for name in exporters
if not (
(info := parse_credential_info(by_name.get(name))) is not None
and is_destination_visible(info, team_ids, org_ids)
)
]
if unassignable:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": (
"You can only assign logging destinations granted to your team "
f"or organization. Not granted: {unassignable}"
)
},
)
def _validate_exporters_shape_and_names(exporters: object) -> None:
"""Common shape + registry check shared by every entry point."""
if not isinstance(exporters, list):
@ -174,13 +129,9 @@ def validate_logging_exporter_field(
requested_exporters: list | None,
user_api_key_dict: UserAPIKeyAuth,
*,
caller_is_team_admin: bool = False,
caller_is_org_admin: bool = False,
existing_exporters: list | None = None,
scope_team_id: str | None = None,
scope_org_id: str | None = None,
) -> None:
"""Authorize a typed ``logging_exporters`` write (the column-backed field).
"""Authorize a typed ``logging_exporters`` write (proxy-admin only).
Adapts the typed list to the metadata-shaped input the shared assignment
validator expects, so the authorization logic lives in one place.
@ -194,11 +145,7 @@ def validate_logging_exporter_field(
validate_logging_exporter_assignment(
requested_metadata,
user_api_key_dict,
caller_is_team_admin=caller_is_team_admin,
caller_is_org_admin=caller_is_org_admin,
existing_metadata=existing_metadata,
scope_team_id=scope_team_id,
scope_org_id=scope_org_id,
)
@ -206,53 +153,26 @@ def validate_logging_exporter_assignment(
metadata: dict | None,
user_api_key_dict: UserAPIKeyAuth,
*,
caller_is_team_admin: bool = False,
caller_is_org_admin: bool = False,
existing_metadata: dict | None = None,
scope_team_id: str | None = None,
scope_org_id: str | None = None,
) -> None:
"""Validate a ``metadata.logging_exporters`` write on key / team / org endpoints.
No-op when the update does not change the effective ``logging_exporters``
value. Proxy admins always pass. Caller-provided flags widen the allow-list
per endpoint:
Proxy-admin only. No-op when the update does not change the effective
``logging_exporters`` value; otherwise a non-proxy-admin is rejected.
- ``/team/update``: pass ``caller_is_org_admin`` from the loaded team's org.
- ``/key/generate``/``/key/update``: pass both flags from the key's team.
- ``/team/new``/``/organization/*``: pass neither (proxy-admin only).
``scope_team_id``/``scope_org_id`` are the team and org the write lands in (the
key's team, or the team/org being updated). A non-proxy-admin may only name
destinations visible to that scope: this is what stops a team admin from routing a
key's traces to a destination scoped to a different team. Proxy admins skip the
scope check; they can assign anything, but the resolver still only fires a named
destination for identities it is visible to.
Update paths replace stored metadata wholesale, so a caller can drop an
admin-assigned exporter by sending ``metadata`` without
``logging_exporters``. Pass ``existing_metadata`` from the loaded row so
removal-via-omission is gated too (Veria F4). On create paths the existing
value is implicitly ``None`` and the validator behaves as before.
Every exporter name (when present) must resolve to a registered logging credential.
Update paths replace stored metadata wholesale, so a caller could drop an
admin-assigned exporter by sending ``metadata`` without ``logging_exporters``.
Pass ``existing_metadata`` from the loaded row so removal-via-omission is gated
too (Veria F4). Every exporter name (when present) must resolve to a registered
logging credential.
"""
if not _exporter_value_changes(metadata, existing_metadata):
return
is_proxy_admin = user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN
if not (is_proxy_admin or caller_is_team_admin or caller_is_org_admin):
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": (
"Only the proxy admin, a team admin of this team, or an "
"org admin of this team's organization can assign logging "
"exporters"
)
},
detail={"error": "Only the proxy admin can assign logging exporters"},
)
requested = metadata.get(LOGGING_EXPORTERS_KEY) if isinstance(metadata, dict) else None
if requested is not None:
_validate_exporters_shape_and_names(requested)
if not is_proxy_admin:
_reject_unassignable_destinations(requested, scope_team_id=scope_team_id, scope_org_id=scope_org_id)

31
litellm/proxy/management_endpoints/team_endpoints.py
View file

@ -1004,9 +1004,6 @@ async def new_team(
```
"""
try:
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_org_id,
)
from litellm.proxy.management_endpoints.logging_exporter_validation import (
validate_logging_exporter_field,
)
@ -1027,15 +1024,7 @@ async def new_team(
# Skip the org-admin lookup entirely when the field isn't being
# written, to avoid hitting the cache for unrelated /team/new calls.
if data.logging_exporters is not None:
validate_logging_exporter_field(
data.logging_exporters,
user_api_key_dict,
caller_is_org_admin=await _is_user_org_admin_for_org_id(
user_api_key_dict=user_api_key_dict,
organization_id=data.organization_id,
),
scope_org_id=data.organization_id,
)
validate_logging_exporter_field(data.logging_exporters, user_api_key_dict)
if prisma_client is None:
raise HTTPException(status_code=500, detail={"error": "No db connected"})
@ -1688,9 +1677,6 @@ async def update_team(
```
"""
try:
from litellm.proxy.management_endpoints.common_utils import (
_is_user_org_admin_for_team,
)
from litellm.proxy.management_endpoints.logging_exporter_validation import (
validate_logging_exporter_field,
)
@ -1748,23 +1734,14 @@ async def update_team(
user_api_key_dict=user_api_key_dict,
)
# logging_exporters on /team/update is proxy-admin or org-admin only:
# team-admins are blocked at the route gate (test_team_update_authz_
# matrix pins this) and the role matrix documents ❌ for team-admin on
# this path. Pass only the org-admin flag so the validator can't
# silently grant team-admins if the route gate is ever widened. The
# validator no-ops when the effective value doesn't change; pass the
# stored column value so a non-admin cannot clear an admin-assigned one.
# logging_exporters on /team/update is proxy-admin only. Pass the stored
# column value so the validator's no-op check sees a real change and a
# non-admin cannot clear an admin-assigned value.
if data.logging_exporters is not None:
validate_logging_exporter_field(
data.logging_exporters,
user_api_key_dict,
caller_is_org_admin=await _is_user_org_admin_for_team(
user_api_key_dict=user_api_key_dict, team_obj=team_for_auth
),
existing_exporters=getattr(existing_team_row, "logging_exporters", None),
scope_team_id=getattr(team_for_auth, "team_id", None),
scope_org_id=getattr(team_for_auth, "organization_id", None),
)
_check_passthrough_routes_caller_permission(data, user_api_key_dict, entity="team")

12
ruff-strict-budget.json
View file

@ -24,7 +24,7 @@
"limit": 130
},
"ANN401": {
"limit": 2015
"limit": 2014
},
"ASYNC230": {
"limit": 14
@ -60,7 +60,7 @@
"limit": 4
},
"BLE001": {
"limit": 2902
"limit": 2901
},
"C401": {
"limit": 11
@ -237,7 +237,7 @@
"limit": 41
},
"RUF022": {
"limit": 85
"limit": 84
},
"RUF023": {
"limit": 5
@ -306,7 +306,7 @@
"limit": 9
},
"TID251": {
"limit": 2652
"limit": 2651
},
"TRY002": {
"limit": 548
@ -354,7 +354,7 @@
"limit": 4
},
"UP035": {
"limit": 2232
"limit": 2231
},
"UP036": {
"limit": 4
@ -363,6 +363,6 @@
"limit": 105
},
"UP045": {
"limit": 17824
"limit": 17820
}
}

32
tests/test_litellm/proxy/auth/test_route_checks.py
View file

@ -3094,36 +3094,20 @@ def test_internal_user_blocked_from_search_tool_writes(route):
# --- Credential route gating (PR #30873: /credentials opened to self-managed) --- #
def test_self_managed_routes_includes_credentials_entries():
"""The PR adds exactly the credential list + single-name routes to the
self-managed set (handlers do their own authz). No wildcard is added, so the
by_name/by_model subpaths are NOT covered here."""
routes = LiteLLMRoutes.self_managed_routes.value
assert "/credentials" in routes
assert "/credentials/{credential_name}" in routes
assert "/credentials/by_name/{credential_name}" not in routes
assert "/credentials/*" not in routes
@pytest.mark.parametrize(
"route,expected",
[
("/credentials", True),
("/credentials/my-dest", True), # single segment -> {credential_name}
("/credentials/by_name/my-dest", False), # extra segment -> no match
("/credentials/by_model/model-123", False), # extra segment -> no match
],
"route",
["/credentials", "/credentials/{credential_name}", "/credentials/my-dest"],
)
def test_credentials_self_managed_pattern_matches_single_segment_only(route, expected):
"""`/credentials/{credential_name}` compiles to ^/credentials/[^/]+$, so a caller
who only reaches self-managed routes (team-admin, org-admin, plain internal user)
can hit the list/single-name routes but NOT the two-segment by_name/by_model
endpoints."""
def test_credentials_routes_are_not_self_managed(route):
"""Credentials are proxy-admin only: no ``/credentials`` route is in the
self-managed set, so a non-admin never reaches the handler for any method
(GET/POST/PATCH/DELETE). Admin-owned logging destinations are managed
exclusively by the proxy admin."""
assert (
RouteChecks.check_route_access(
route=route, allowed_routes=LiteLLMRoutes.self_managed_routes.value
)
is expected
is False
)

360
tests/test_litellm/proxy/credential_endpoints/test_access_decision.py
View file

@ -1,360 +0,0 @@
"""Exhaustive tests for the pure access-decision function.
Each test names one specific reason a team-admin patch should be denied (or
allowed). Together they pin the security contract: changing this code with
the tests in place should fail a case named after what you broke.
"""
from __future__ import annotations
import pytest
from litellm.models.credentials import CredentialInfo
from litellm.proxy.credential_endpoints.access_decision import (
Allow,
Deny,
decide_credential_patch,
)
_EXISTING_INFO = {
"credential_type": "logging",
"description": "tenant Langfuse",
"host": "https://cloud.langfuse.com",
"access": {"teams": ["team-A", "team-B"], "orgs": ["org-1"], "global": False},
}
def _info(value):
return None if value is None else CredentialInfo.model_validate(value)
def _decision(
*,
is_proxy_admin: bool = False,
caller_team_admin_ids: frozenset[str] = frozenset({"team-T"}),
existing_info=_EXISTING_INFO,
patch_info=None,
patch_values=None,
patch_name_changed: bool = False,
):
return decide_credential_patch(
is_proxy_admin=is_proxy_admin,
caller_team_admin_ids=caller_team_admin_ids,
existing_info=_info(existing_info),
patch_info=_info(patch_info),
patch_values=patch_values,
patch_name_changed=patch_name_changed,
)
class TestProxyAdminAllow:
def test_proxy_admin_allowed_on_value_change(self):
d = _decision(
is_proxy_admin=True,
patch_values={"api_key": "rotated"},
patch_info={"credential_type": "logging"},
)
assert isinstance(d, Allow)
def test_proxy_admin_allowed_on_global_flip(self):
d = _decision(
is_proxy_admin=True,
patch_info={"access": {"global": True}},
)
assert isinstance(d, Allow)
def test_proxy_admin_allowed_on_rename(self):
d = _decision(is_proxy_admin=True, patch_name_changed=True)
assert isinstance(d, Allow)
class TestTeamAdminAllow:
def test_appending_own_team_id(self):
d = _decision(
patch_info={
"access": {"teams": ["team-A", "team-B", "team-T"]},
},
)
assert isinstance(d, Allow)
def test_appending_only_own_team_id_when_no_prior_teams(self):
existing = {**_EXISTING_INFO, "access": {"global": False}}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": ["team-T"]}},
)
assert isinstance(d, Allow)
def test_idempotent_when_already_granted(self):
existing = {**_EXISTING_INFO, "access": {"teams": ["team-T"]}}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": ["team-T"]}},
)
assert isinstance(d, Allow)
class TestTeamAdminDeny:
def test_not_team_admin_anywhere(self):
d = _decision(
caller_team_admin_ids=frozenset(),
patch_info={"access": {"teams": ["team-T"]}},
)
assert isinstance(d, Deny)
assert "proxy admin" in d.reason
def test_rename(self):
d = _decision(
patch_name_changed=True,
patch_info={"access": {"teams": ["team-T"]}},
)
assert isinstance(d, Deny)
assert "credential_name" in d.reason
def test_changing_credential_values(self):
d = _decision(
patch_values={"api_key": "stolen"},
patch_info={"access": {"teams": ["team-T"]}},
)
assert isinstance(d, Deny)
assert "credential_values" in d.reason
def test_empty_patch(self):
d = _decision(patch_info=None)
assert isinstance(d, Deny)
@pytest.mark.parametrize(
"field", ["credential_type", "description", "host", "endpoint"]
)
def test_changing_immutable_info_field(self, field):
d = _decision(patch_info={field: "x", "access": {"teams": ["team-T"]}})
assert isinstance(d, Deny)
assert field in d.reason
def test_patch_info_with_unknown_keys(self):
d = _decision(patch_info={"weird_field": 1})
assert isinstance(d, Deny)
assert "weird_field" in d.reason
def test_flipping_global(self):
d = _decision(patch_info={"access": {"global": True}})
assert isinstance(d, Deny)
assert "global" in d.reason
def test_editing_orgs(self):
d = _decision(patch_info={"access": {"orgs": ["org-new"]}})
assert isinstance(d, Deny)
assert "orgs" in d.reason
def test_no_op_resend_of_existing_global_is_allowed(self):
"""The UI's Edit-access modal always sends the FULL access object
(so unchecking a team revokes). A non-admin re-sending the existing
global=false alongside their team patch must NOT be rejected as if
they were trying to flip the toggle.
Before this fix the decider checked only "is global_ in the patch?"
which broke every UI save that included the unchanged global toggle
plus a team edit.
"""
d = _decision(
patch_info={
"access": {
"global": False,
"teams": ["team-A", "team-B", "team-T"],
"orgs": ["org-1"],
}
}
)
assert isinstance(d, Allow)
def test_no_op_resend_of_existing_orgs_is_allowed(self):
"""Same shape as global: a patch that includes the unchanged orgs
list alongside a team edit must pass."""
d = _decision(
patch_info={
"access": {
"global": False,
"teams": ["team-A", "team-B", "team-T"],
"orgs": ["org-1"],
}
}
)
assert isinstance(d, Allow)
def test_attempt_to_flip_global_when_existing_is_false(self):
"""Direct flip from stored False to True still rejected."""
d = _decision(
patch_info={"access": {"global": True, "teams": ["team-A", "team-B"]}}
)
assert isinstance(d, Deny)
assert "global" in d.reason
def test_attempt_to_flip_global_when_existing_is_true(self):
"""Direct flip from stored True to False still rejected (it's still
a global mutation; only proxy-admin can change destination-wide reach)."""
existing = {
**_EXISTING_INFO,
"access": {**_EXISTING_INFO["access"], "global": True},
}
d = _decision(
existing_info=existing,
patch_info={"access": {"global": False, "teams": ["team-A", "team-B"]}},
)
assert isinstance(d, Deny)
assert "global" in d.reason
def test_attempt_to_change_orgs_is_rejected(self):
"""Adding an org_id different from stored is still rejected."""
d = _decision(
patch_info={
"access": {"orgs": ["org-1", "org-2"], "teams": ["team-A", "team-B"]}
}
)
assert isinstance(d, Deny)
assert "orgs" in d.reason
def test_adding_foreign_team_id(self):
"""foreign team_ids in the patch ARE caller input -- safe to echo."""
d = _decision(
patch_info={
"access": {"teams": ["team-A", "team-B", "team-foreign"]},
},
)
assert isinstance(d, Deny)
assert d.from_user_input is True
assert "team-foreign" in d.reason
def test_removing_foreign_team_grant(self):
"""team-admin may NOT remove a team they don't admin.
Reason intentionally does NOT echo the stored team_id (it's not
caller-typed; surfacing it would leak access list membership).
"""
d = _decision(
patch_info={
"access": {"teams": ["team-A", "team-T"]},
},
)
assert isinstance(d, Deny)
assert d.from_user_input is False
assert "team-B" not in d.reason
assert "may only revoke" in d.reason
def test_replacing_teams_wholesale_with_foreign_remaining(self):
"""Wholesale replacement that removes foreign grants is rejected.
The stored team_ids that were dropped must not appear in the reason --
they're access list contents, not caller input.
"""
d = _decision(patch_info={"access": {"teams": ["team-T"]}})
assert isinstance(d, Deny)
assert d.from_user_input is False
assert "team-A" not in d.reason
assert "team-B" not in d.reason
class TestTeamAdminRevoke:
"""A team-admin may revoke their OWN team's grant; never another's."""
def test_revoking_own_team_is_allowed(self):
existing = {
**_EXISTING_INFO,
"access": {"teams": ["team-A", "team-T"]},
}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": ["team-A"]}},
)
assert isinstance(d, Allow)
def test_revoking_own_team_when_only_grant(self):
"""Saving an empty teams list when the caller was the sole grant."""
existing = {**_EXISTING_INFO, "access": {"teams": ["team-T"]}}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": []}},
)
assert isinstance(d, Allow)
def test_revoke_attempt_on_foreign_team_denied(self):
"""A patch that removes a foreign team is still rejected, even if
the caller is also revoking their own. The foreign team_id MUST
NOT appear in the reason (stored access list content)."""
existing = {
**_EXISTING_INFO,
"access": {"teams": ["team-A", "team-B", "team-T"]},
}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": ["team-A"]}}, # drops team-B AND team-T
)
assert isinstance(d, Deny)
assert d.from_user_input is False
assert "team-B" not in d.reason
class TestEmptyingGrantsIsAllowed:
"""Empty access is deny-all (a destination with no grants routes to no one),
so a team-admin emptying their own grant merely DISABLES the destination and
can never widen it. The decider allows it for every combination; there is no
special auto_enable case, because empty access is not proxy-wide."""
def test_emptying_sole_own_grant_on_auto_enable_destination_is_allowed(self):
existing = {
**_EXISTING_INFO,
"auto_enable": True,
"access": {"global": False, "teams": ["team-T"], "orgs": []},
}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": []}},
)
assert isinstance(d, Allow)
def test_emptying_sole_own_grant_without_auto_enable_is_allowed(self):
existing = {
**_EXISTING_INFO,
"auto_enable": False,
"access": {"global": False, "teams": ["team-T"], "orgs": []},
}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": []}},
)
assert isinstance(d, Allow)
def test_emptying_teams_still_cannot_drop_a_foreign_grant(self):
"""Allowing empty-out does not weaken the foreign-revoke guard: a team-admin
still cannot remove a team they do not administer, auto_enable or not."""
existing = {
**_EXISTING_INFO,
"auto_enable": True,
"access": {"global": False, "teams": ["team-T", "team-A"], "orgs": []},
}
d = _decision(
existing_info=existing,
patch_info={"access": {"teams": []}}, # drops team-A (foreign) too
)
assert isinstance(d, Deny)
assert "team-A" not in d.reason
class TestTeamAdminMultipleTeams:
def test_can_add_multiple_own_team_ids(self):
d = _decision(
caller_team_admin_ids=frozenset({"team-T1", "team-T2"}),
patch_info={
"access": {"teams": ["team-A", "team-B", "team-T1", "team-T2"]},
},
)
assert isinstance(d, Allow)
def test_one_own_one_foreign_is_deny(self):
d = _decision(
caller_team_admin_ids=frozenset({"team-T1"}),
patch_info={
"access": {"teams": ["team-A", "team-B", "team-T1", "team-T2"]},
},
)
assert isinstance(d, Deny)
assert "team-T2" in d.reason

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

@ -1,12 +1,11 @@
"""Admin-gating on credential mutations.
POST and DELETE on ``/credentials`` are proxy-admin only across the board
(both logging and provider credentials). The route gate was widened so
team-admins can PATCH ``access.teams`` on existing logging destinations,
which requires also reaching the GET endpoint — but creation and deletion
of any credential remain admin-only to keep platform infrastructure under
the platform admin's control. PATCH is widened for logging credentials only
via the pure ``decide_credential_patch`` decider tested separately.
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.
"""
import os
@ -22,7 +21,6 @@ import litellm
import litellm.proxy.credential_endpoints.endpoints as endpoints
from litellm.models.credentials import CredentialItem, UpdateCredentialItem
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.credential_endpoints.access_decision import OPAQUE_DENY_REASON
from litellm.types.utils import CreateCredentialItem
@ -318,79 +316,6 @@ async def test_delete_db_only_logging_credential_forbidden_for_non_admin(
# --- team-admin self-assign tests (LIT-3850 follow-up) ----------------------
_DEST_WITH_TEAMS = {
"credential_type": "logging",
"description": "tenant Langfuse",
"host": "https://cloud.langfuse.com",
"access": {"teams": ["team-existing"]},
}
def _team_admin_of(team_ids):
"""A non-admin caller whose user_id is admin of the named teams.
Combined with ``_patch_team_admin_lookup`` it mimics the real
``_caller_grantable_team_ids`` resolution without touching the DB.
"""
return UserAPIKeyAuth(
api_key="k", user_role=LitellmUserRoles.INTERNAL_USER, user_id="ta-demo"
)
@pytest.fixture
def _patch_team_admin_lookup(monkeypatch):
"""Substitute the DB-backed admin-scope lookup with a configurable mock.
``ids`` is the caller's team-admin scope; ``org_ids`` the org-admin scope.
Patching the single source ``_caller_admin_scope`` covers both the list
endpoint and the PATCH decider, which reads team ids through the
``_caller_grantable_team_ids`` wrapper.
"""
holder = {"ids": frozenset(), "org_ids": frozenset()}
async def _fake(user_api_key_dict, prisma_client):
return endpoints.CallerAdminScope(
team_ids=frozenset(holder["ids"]),
org_ids=frozenset(holder["org_ids"]),
)
monkeypatch.setattr(endpoints, "_caller_admin_scope", _fake)
return holder
def _resident_logging_dest():
return CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "h"},
credential_info=_DEST_WITH_TEAMS,
)
@pytest.mark.asyncio
async def test_team_admin_can_append_own_team_to_access(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
monkeypatch.setattr(litellm, "credential_list", [_resident_logging_dest()])
_connected_db.find_by_name = AsyncMock(return_value=_resident_logging_dest())
_connected_db.update_by_name = AsyncMock()
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
result = await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="dest",
credential_values={},
credential_info={"access": {"teams": ["team-existing", "team-T"]}},
),
credential_name="dest",
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert result["success"] is True
_connected_db.update_by_name.assert_awaited_once()
@pytest.mark.asyncio
async def test_provider_credential_patch_forbidden_for_non_admin(
_connected_db, monkeypatch
@ -428,242 +353,10 @@ async def test_provider_credential_patch_forbidden_for_non_admin(
@pytest.mark.asyncio
async def test_provider_credential_access_patch_bypass_forbidden(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
"""Cursor BugBot regression: a team-admin can't sneak `access.teams` onto
a PROVIDER credential to route through the decider instead of the admin
gate.
`is_admin_gated_credential_info(patch)` returns True for any patch
containing an `access` field, so previously a team-admin could PATCH a
provider credential with `{credential_info: {access: {teams: [...]}}}`
and reach `decide_credential_patch`, which would Allow because the patch
is just "add own team to access.teams". Gate must look at the STORED
credential's type, not the patch body.
"""
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()
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=UpdateCredentialItem(
credential_info={"access": {"teams": ["team-T"]}},
),
credential_name="openai-prod",
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert exc.value.status_code == 403
_connected_db.update_by_name.assert_not_awaited()
@pytest.mark.asyncio
async def test_team_admin_can_revoke_own_team_grant(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
"""A team-admin saving an access list without their own team_id revokes it."""
existing = CredentialItem(
credential_name="dest",
credential_values={"langfuse_host": "h"},
credential_info={
"credential_type": "logging",
"description": "tenant Langfuse",
"host": "https://cloud.langfuse.com",
"access": {"teams": ["team-existing", "team-T"]},
},
)
monkeypatch.setattr(litellm, "credential_list", [existing])
_connected_db.find_by_name = AsyncMock(return_value=existing)
_connected_db.update_by_name = AsyncMock()
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
result = await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="dest",
credential_values={},
credential_info={"access": {"teams": ["team-existing"]}},
),
credential_name="dest",
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert result["success"] is True
_connected_db.update_by_name.assert_awaited_once()
@pytest.mark.asyncio
async def test_team_admin_cannot_grant_foreign_team(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
monkeypatch.setattr(litellm, "credential_list", [_resident_logging_dest()])
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
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": {"teams": ["team-existing", "team-foreign"]}
},
),
credential_name="dest",
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert exc.value.status_code == 403
assert "team-foreign" in exc.value.detail["error"]
@pytest.mark.asyncio
async def test_team_admin_cannot_rotate_credential_values(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
monkeypatch.setattr(litellm, "credential_list", [_resident_logging_dest()])
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=CredentialItem(
credential_name="dest",
credential_values={"public_key": "pk-stolen"},
credential_info={"access": {"teams": ["team-existing", "team-T"]}},
),
credential_name="dest",
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert exc.value.status_code == 403
# Endpoint collapses non-caller-input Deny reasons to the opaque message
# so PATCH /credentials/{name} can't be used as an existence oracle.
assert exc.value.detail["error"] == OPAQUE_DENY_REASON
@pytest.mark.asyncio
async def test_team_admin_cannot_flip_global(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
monkeypatch.setattr(litellm, "credential_list", [_resident_logging_dest()])
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
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_name="dest",
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert exc.value.status_code == 403
assert exc.value.detail["error"] == OPAQUE_DENY_REASON
@pytest.mark.asyncio
async def test_get_credentials_shows_only_in_scope_destinations_for_non_admin(
monkeypatch, _patch_team_admin_lookup
):
"""Leak regression (Veria #1): a non-proxy-admin sees only destinations granted
to a scope they administer, not every logging destination. The caller admins
team-existing, so they see the destination granted to team-existing but never
the provider credential and never a destination scoped to another team."""
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem( # provider credential: never visible to a non-admin
credential_name="openai",
credential_values={"api_key": "sk-secret"},
credential_info={"custom_llm_provider": "openai"},
),
CredentialItem( # granted to team-existing
credential_name="poc-langfuse",
credential_values={"public_key": "pk-1"},
credential_info=_DEST_WITH_TEAMS,
),
CredentialItem( # granted to a DIFFERENT team: must stay hidden
credential_name="other-team-dest",
credential_values={},
credential_info={
"credential_type": "logging",
"description": "arize",
"access": {"teams": ["team-other"]},
},
),
],
)
_patch_team_admin_lookup["ids"] = frozenset({"team-existing"})
response = await endpoints.get_credentials(
request=MagicMock(),
fastapi_response=MagicMock(),
user_api_key_dict=_team_admin_of(["team-existing"]),
)
names = [c["credential_name"] for c in response["credentials"]]
assert names == ["poc-langfuse"]
@pytest.mark.asyncio
async def test_get_credentials_masks_all_values_for_non_admin(
monkeypatch, _patch_team_admin_lookup
):
raw_headers = "Authorization=Bearer collector-secret,x-api-key=api-secret"
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="generic-otel",
credential_values={
"otel_endpoint": "https://collector.example.com/v1/traces",
"otel_headers": raw_headers,
},
credential_info={
"credential_type": "logging",
"description": "generic",
"access": {"teams": ["team-existing"]},
},
),
],
)
_patch_team_admin_lookup["ids"] = frozenset({"team-existing"})
response = await endpoints.get_credentials(
request=MagicMock(),
fastapi_response=MagicMock(),
user_api_key_dict=_team_admin_of(["team-existing"]),
)
values = response["credentials"][0]["credential_values"]
assert values == {
"otel_endpoint": "********",
"otel_headers": "********",
}
assert "collector-secret" not in str(response)
assert "api-secret" not in str(response)
@pytest.mark.asyncio
async def test_get_credentials_hides_out_of_scope_destination(
monkeypatch, _patch_team_admin_lookup
):
"""The exact leak: a team-admin of an unrelated team must see none of another
team's destinations. Pre-fix, get_credentials returned every logging
destination to any team/org admin regardless of the destination's access."""
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",
@ -671,88 +364,17 @@ async def test_get_credentials_hides_out_of_scope_destination(
CredentialItem(
credential_name="poc-langfuse",
credential_values={"public_key": "pk-1"},
credential_info=_DEST_WITH_TEAMS, # granted to team-existing
credential_info=_LOGGING_INFO,
),
],
)
_patch_team_admin_lookup["ids"] = frozenset({"team-T"}) # NOT team-existing
response = await endpoints.get_credentials(
request=MagicMock(),
fastapi_response=MagicMock(),
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert response["credentials"] == []
@pytest.mark.asyncio
async def test_get_credentials_shows_org_scoped_destination_to_org_admin(
monkeypatch, _patch_team_admin_lookup
):
"""An org-admin sees a destination granted to their org via access.orgs, matched
against the org-admin scope (not just team ids)."""
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="org-dest",
credential_values={},
credential_info={
"credential_type": "logging",
"description": "arize",
"access": {"orgs": ["org-1"]},
},
),
],
)
_patch_team_admin_lookup["ids"] = frozenset()
_patch_team_admin_lookup["org_ids"] = frozenset({"org-1"})
response = await endpoints.get_credentials(
request=MagicMock(),
fastapi_response=MagicMock(),
user_api_key_dict=UserAPIKeyAuth(
api_key="k", user_role=LitellmUserRoles.INTERNAL_USER, user_id="oa"
),
)
names = [c["credential_name"] for c in response["credentials"]]
assert names == ["org-dest"]
@pytest.mark.asyncio
async def test_get_credentials_forbidden_for_plain_user(
monkeypatch, _patch_team_admin_lookup
):
"""Veria F2 regression: a plain internal_user (no team-admin or
org-admin status anywhere) gets 403, NOT a filtered list. The previous
handler returned destination names, hosts, and scope metadata to any
authenticated caller because the route gate was widened to support
team-admin self-assignment.
"""
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="poc-langfuse",
credential_values={"public_key": "pk-1"},
credential_info=_DEST_WITH_TEAMS,
),
],
)
_patch_team_admin_lookup["ids"] = frozenset() # admins nothing
with pytest.raises(HTTPException) as exc:
await endpoints.get_credentials(
request=MagicMock(),
fastapi_response=MagicMock(),
user_api_key_dict=UserAPIKeyAuth(
api_key="k",
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="plain-user",
),
user_api_key_dict=_member(),
)
assert exc.value.status_code == 403
assert "team-admin" in exc.value.detail["error"]
def test_patch_credentials_route_targets_update_credential():
@ -774,92 +396,6 @@ def test_patch_credentials_route_targets_update_credential():
assert patch_route.endpoint is endpoints.update_credential
@pytest.mark.asyncio
async def test_patch_credentials_does_not_leak_credential_type(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
"""Existence-oracle regression: a team-admin probing a credential they don't own
must NOT be able to distinguish "logging credential, not yours" from
"provider credential" or "doesn't exist" by comparing 403 detail strings.
Pre-fix the decider returned reasons like "access.global is proxy-admin only"
while `_require_proxy_admin` returned a fixed string, so the same probe
(e.g. PATCH `{credential_info: {access: {global: true}}}`) would yield
different bodies depending on the stored type. All three paths now return
the same opaque message.
"""
provider = CredentialItem(
credential_name="openai-prod",
credential_values={"api_key": "sk-real"},
credential_info={"custom_llm_provider": "openai"},
)
logging_other = CredentialItem(
credential_name="other-langfuse",
credential_values={"langfuse_host": "h"},
credential_info={
"credential_type": "logging",
"description": "tenant Langfuse",
"host": "https://cloud.langfuse.com",
"access": {"teams": ["team-other"]},
},
)
monkeypatch.setattr(litellm, "credential_list", [provider, logging_other])
_connected_db.find_by_name = AsyncMock(
side_effect=lambda name: provider if name == "openai-prod" else logging_other
)
# Caller is team-admin of team-T (NOT team-other, so can't legitimately
# edit logging_other either), probing with a patch that touches a
# decider-protected field to maximally expose any branch divergence.
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
probe_patch = UpdateCredentialItem(
credential_info={"access": {"global": True}},
)
bodies: list[object] = []
for name in ("openai-prod", "other-langfuse", "does-not-exist"):
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=probe_patch,
credential_name=name,
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert exc.value.status_code == 403
bodies.append(exc.value.detail)
assert bodies[0] == bodies[1] == bodies[2] == {"error": OPAQUE_DENY_REASON}
@pytest.mark.asyncio
async def test_patch_credentials_echoes_foreign_team_id_to_legit_team_admin(
_connected_db, _patch_team_admin_lookup, monkeypatch
):
"""The one accepted leak: when a team-admin tries to grant a team_id they
typed in the patch and don't admin, the response names that team_id so
the UI can render a useful error. The team_id was caller input, so it
isn't an existence oracle (the caller already knew the value).
"""
monkeypatch.setattr(litellm, "credential_list", [_resident_logging_dest()])
_connected_db.find_by_name = AsyncMock(return_value=_resident_logging_dest())
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
with pytest.raises(HTTPException) as exc:
await endpoints.update_credential(
request=MagicMock(),
fastapi_response=MagicMock(),
credential=UpdateCredentialItem(
credential_info={
"access": {"teams": ["team-existing", "team-foreign"]}
},
),
credential_name="dest",
user_api_key_dict=_team_admin_of(["team-T"]),
)
assert exc.value.status_code == 403
assert "team-foreign" in exc.value.detail["error"]
@pytest.mark.asyncio
async def test_get_credentials_returns_all_for_proxy_admin(monkeypatch):
raw_headers = "Authorization=Bearer collector-secret,x-api-key=api-secret"
@ -875,7 +411,11 @@ async def test_get_credentials_returns_all_for_proxy_admin(monkeypatch):
CredentialItem(
credential_name="poc-langfuse",
credential_values={"public_key": "pk-1"},
credential_info=_DEST_WITH_TEAMS,
credential_info={
"credential_type": "logging",
"description": "langfuse_otel",
"access": {"teams": ["team-A"]},
},
),
CredentialItem(
credential_name="generic-otel",
@ -943,44 +483,3 @@ async def test_get_credentials_admin_viewer_gets_full_list_fully_masked(monkeypa
assert "sk-secret" not in str(response)
@pytest.mark.asyncio
async def test_authorize_patch_malformed_stored_access_does_not_500(
_patch_team_admin_lookup,
):
"""A stored logging destination whose ``access`` carries a key the strict model
forbids (e.g. legacy data, or data written before the write gate rejected unknown
keys) must not 500 every PATCH. Fail closed: a non-admin gets 403, the proxy admin
is still allowed through. Pre-fix the unguarded ``CredentialInfo.model_validate``
on the stored row raised an uncaught ``ValidationError`` for all callers."""
malformed = CredentialItem(
credential_name="legacy-dest",
credential_values={"langfuse_host": "h"},
credential_info={
"credential_type": "logging",
"access": {"global": True, "legacy_field": "x"},
},
)
patch = UpdateCredentialItem(credential_info={"access": {"teams": ["team-T"]}})
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
with pytest.raises(HTTPException) as exc:
await endpoints._authorize_credential_patch(
credential_name="legacy-dest",
patch=patch,
existing=malformed,
user_api_key_dict=_team_admin_of(["team-T"]),
prisma_client=MagicMock(),
)
assert exc.value.status_code == 403
assert exc.value.detail == {"error": OPAQUE_DENY_REASON}
assert (
await endpoints._authorize_credential_patch(
credential_name="legacy-dest",
patch=patch,
existing=malformed,
user_api_key_dict=_admin(),
prisma_client=MagicMock(),
)
is None
)

163
tests/test_litellm/proxy/management_endpoints/test_logging_exporter_validation.py
View file

@ -1,10 +1,10 @@
"""Validation for admin-owned logging-exporter assignment on key/team/org.
The single ``validate_logging_exporter_assignment`` runs across all four
endpoints (``/team/new``, ``/team/update``, ``/key/generate``, ``/key/update``,
``/organization/*``); each call site computes the relevant
``caller_is_team_admin`` / ``caller_is_org_admin`` flags from the loaded
team or org and passes them in. Proxy admin always passes.
The single ``validate_logging_exporter_assignment`` runs across every write path
(``/team/new``, ``/team/update``, ``/key/generate``, ``/key/update``,
``/organization/*``). Assigning ``logging_exporters`` is proxy-admin only: a
non-proxy-admin write is rejected, and every named exporter must resolve to a
registered logging credential.
"""
import os
@ -94,34 +94,6 @@ def test_proxy_admin_always_allowed(_registry):
validate_logging_exporter_assignment(_ok(["langfuse-eu"]), _admin())
def test_team_admin_flag_allows_non_admin(_registry):
"""A non-admin caller flagged caller_is_team_admin=True passes."""
validate_logging_exporter_assignment(
_ok(["langfuse-eu"]),
_non_admin(),
caller_is_team_admin=True,
)
def test_org_admin_flag_allows_non_admin(_registry):
"""A non-admin caller flagged caller_is_org_admin=True passes."""
validate_logging_exporter_assignment(
_ok(["langfuse-eu"]),
_non_admin(),
caller_is_org_admin=True,
)
def test_both_flags_set_allows_non_admin(_registry):
"""Setting both flags is fine; they're independent OR-ed allows."""
validate_logging_exporter_assignment(
_ok(["langfuse-eu"]),
_non_admin(),
caller_is_team_admin=True,
caller_is_org_admin=True,
)
def test_non_admin_with_no_flags_is_forbidden(_registry):
"""The headline deny: internal_user with no team/org admin context."""
with pytest.raises(HTTPException) as exc:
@ -129,95 +101,9 @@ def test_non_admin_with_no_flags_is_forbidden(_registry):
assert exc.value.status_code == 403
def test_proxy_admin_overrides_falsy_flags(_registry):
"""proxy_admin role wins even when both flags are False."""
validate_logging_exporter_assignment(
_ok(["langfuse-eu"]),
_admin(),
caller_is_team_admin=False,
caller_is_org_admin=False,
)
# --- Scope checks: a non-proxy-admin may only name destinations granted to them -
def test_team_admin_can_assign_destination_granted_to_their_team(_registry):
"""arize-ds is granted to ds-team; a team admin writing in ds-team's scope may
name it."""
validate_logging_exporter_assignment(
_ok(["arize-ds"]),
_non_admin(),
caller_is_team_admin=True,
scope_team_id="ds-team",
)
def test_team_admin_cannot_assign_destination_not_granted_to_their_team(_registry):
"""The headline leak: a team admin of another team names ds-team's destination.
Pre-fix this passed (only the name was checked); now it is a 403."""
with pytest.raises(HTTPException) as exc:
validate_logging_exporter_assignment(
_ok(["arize-ds"]),
_non_admin(),
caller_is_team_admin=True,
scope_team_id="platform-team",
)
assert exc.value.status_code == 403
def test_org_admin_can_assign_destination_granted_to_their_org(_registry):
validate_logging_exporter_assignment(
_ok(["arize-ds"]),
_non_admin(),
caller_is_org_admin=True,
scope_org_id="ds-org",
)
def test_org_admin_cannot_assign_destination_not_granted_to_their_org(_registry):
with pytest.raises(HTTPException) as exc:
validate_logging_exporter_assignment(
_ok(["arize-ds"]),
_non_admin(),
caller_is_org_admin=True,
scope_org_id="other-org",
)
assert exc.value.status_code == 403
def test_proxy_admin_can_assign_any_destination(_registry):
"""Proxy admin skips the scope check entirely; arize-ds is granted to no scope
the admin is in, yet the write is allowed."""
validate_logging_exporter_assignment(
_ok(["arize-ds"]),
_admin(),
scope_team_id="platform-team",
)
def test_team_admin_can_assign_auto_enable_default(_registry):
"""A proxy-wide auto default (access.global + auto_enable) is visible to every
scope, so a team admin in any team may name it."""
validate_logging_exporter_assignment(
_ok(["central-default"]),
_non_admin(),
caller_is_team_admin=True,
scope_team_id="platform-team",
)
def test_team_admin_can_assign_global_destination(_registry):
"""access.global makes a destination visible to every scope, so a team admin
in any team may name it."""
validate_logging_exporter_assignment(
_ok(["langfuse-eu"]),
_non_admin(),
caller_is_team_admin=True,
scope_team_id="platform-team",
)
# --- Shape / registry checks (run regardless of who's calling) --------------
@ -227,16 +113,6 @@ def test_unknown_credential_rejected_for_admin(_registry):
assert exc.value.status_code == 400
def test_unknown_credential_rejected_for_team_admin(_registry):
with pytest.raises(HTTPException) as exc:
validate_logging_exporter_assignment(
_ok(["does-not-exist"]),
_non_admin(),
caller_is_team_admin=True,
)
assert exc.value.status_code == 400
def test_provider_credential_rejected(_registry):
"""openai-key exists but is provider-typed, not a logging destination."""
with pytest.raises(HTTPException) as exc:
@ -288,16 +164,6 @@ def test_removal_via_omission_allowed_for_proxy_admin(_registry):
)
def test_removal_via_omission_allowed_for_team_admin(_registry):
"""A team-admin of the owning team may drop the exporter."""
validate_logging_exporter_assignment(
{"some_other_key": 1},
_non_admin(),
caller_is_team_admin=True,
existing_metadata={"logging_exporters": ["langfuse-eu"]},
)
def test_explicit_empty_list_blocked_for_non_admin(_registry):
"""A non-admin submitting `logging_exporters: []` over a non-empty stored
value is a removal write and must be gated."""
@ -410,23 +276,15 @@ def test_field_none_is_noop_for_non_admin(_registry):
validate_logging_exporter_field(None, _non_admin())
def test_field_set_by_non_admin_without_flags_is_forbidden(_registry):
def test_field_set_by_non_admin_is_forbidden(_registry):
with pytest.raises(HTTPException) as exc:
validate_logging_exporter_field(["langfuse-eu"], _non_admin())
assert exc.value.status_code == 403
def test_field_scoped_rejection_for_out_of_scope_team(_registry):
"""arize-ds is granted to ds-team; a team admin writing in another team's scope
cannot name it, exactly as the metadata path gated it."""
with pytest.raises(HTTPException) as exc:
validate_logging_exporter_field(
["arize-ds"],
_non_admin(),
caller_is_team_admin=True,
scope_team_id="platform-team",
)
assert exc.value.status_code == 403
def test_field_proxy_admin_can_assign(_registry):
"""Proxy admin may assign any registered logging destination."""
validate_logging_exporter_field(["arize-ds"], _admin())
def test_field_empty_clear_over_existing_is_gated_for_non_admin(_registry):
@ -450,6 +308,3 @@ def test_field_unchanged_value_is_noop(_registry):
)
def test_field_admin_can_assign_out_of_scope(_registry):
"""Proxy admin skips the scope check (parity with the metadata path)."""
validate_logging_exporter_field(["arize-ds"], _admin(), scope_team_id="platform-team")

4
ui/litellm-dashboard/src/app/(dashboard)/hooks/credentials/useCredentials.ts
View file

@ -5,11 +5,11 @@ import useAuthorized from "@/app/(dashboard)/hooks/useAuthorized";
const credentialsKeys = createQueryKeys("credentials");
export const useCredentials = () => {
export const useCredentials = (enabled: boolean = true) => {
const { accessToken } = useAuthorized();
return useQuery<CredentialsResponse>({
queryKey: credentialsKeys.list({}),
queryFn: async () => await credentialListCall(accessToken!),
enabled: Boolean(accessToken),
enabled: enabled && Boolean(accessToken),
});
};

18
ui/litellm-dashboard/src/components/logging_credentials/LoggingExportersSelect.tsx
View file

@ -2,6 +2,8 @@ import { Select } from "antd";
import React from "react";
import { useCredentials } from "@/app/(dashboard)/hooks/credentials/useCredentials";
import useAuthorized from "@/app/(dashboard)/hooks/useAuthorized";
import { isProxyAdminRole } from "@/utils/roles";
interface LoggingExportersSelectProps {
value?: string[];
@ -14,15 +16,17 @@ interface LoggingExportersSelectProps {
* the identity's logging_exporters column; the proxy unions them across the identity
* chain and fans out.
*
* The options are exactly what GET /credentials returns for the caller, which the backend
* already scopes: a proxy admin receives every destination, while a team or org admin
* receives only the destinations granted to a scope they administer. Visibility is
* enforced server-side by the same predicate the assignment gate and the request-time
* resolver use, so this component does no role-based filtering of its own; doing so would
* risk disagreeing with the backend in either direction.
* Assigning logging exporters is proxy-admin only (GET /credentials and the assignment
* gate both reject non-proxy-admins), so this control renders only for a proxy admin.
*/
const LoggingExportersSelect: React.FC<LoggingExportersSelectProps> = ({ value, onChange }) => {
const { data } = useCredentials();
const { userRole } = useAuthorized();
const isProxyAdmin = userRole ? isProxyAdminRole(userRole) : false;
const { data } = useCredentials(isProxyAdmin);
if (!isProxyAdmin) {
return null;
}
const options = (data?.credentials ?? [])
.filter((credential) => credential.credential_info?.credential_type === "logging")

13
ui/litellm-dashboard/src/components/settings.tsx
View file

@ -42,6 +42,7 @@ import {
import { LoggingCallbacksTable } from "./Settings/LoggingAndAlerts/LoggingCallbacks/LoggingCallbacksTable";
import { AlertingObject, CredentialAccess, ResolvedScope } from "./Settings/LoggingAndAlerts/LoggingCallbacks/types";
import { useCredentials } from "@/app/(dashboard)/hooks/credentials/useCredentials";
import { isProxyAdminRole } from "@/utils/roles";
import { useTeams } from "@/app/(dashboard)/hooks/teams/useTeams";
import { useOrganizations } from "@/app/(dashboard)/hooks/organizations/useOrganizations";
import EditLoggingCredentialModal from "./logging_credentials/EditLoggingCredentialModal";
@ -259,9 +260,15 @@ const Settings: React.FC<SettingsPageProps> = ({ accessToken, userRole, userID,
const [isAddingCallback, setIsAddingCallback] = useState(false);
const [isDeletingCallback, setIsDeletingCallback] = useState(false);
// OTEL trace destinations are credentials tagged credential_type=logging; they share
// the one Active Logging Callbacks table as rows alongside config callbacks.
const { data: credentialData, refetch: refetchCredentials } = useCredentials();
// OTEL trace destinations are proxy-admin-managed credentials tagged
// credential_type=logging; they share the one Active Logging Callbacks table as
// rows alongside config callbacks. Only a proxy admin (or admin-viewer, read-only)
// may read them, so non-admins skip the fetch entirely.
const canReadCredentials =
(userRole ? isProxyAdminRole(userRole) : false) ||
userRole === "Admin Viewer" ||
userRole === "proxy_admin_viewer";
const { data: credentialData, refetch: refetchCredentials } = useCredentials(canReadCredentials);
const { data: teamsData } = useTeams();
const { data: orgsData } = useOrganizations();
const [editAccessFor, setEditAccessFor] = useState<{ name: string; access?: CredentialAccess } | null>(null);