mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-16 23:41:43 +00:00
fix(otel/v2): scope credential list visibility to caller admin scope; centralize on typed predicate
GET /credentials returned every logging destination's name, host, and access scope to any team/org admin, filtering only on credential type and not on the caller's grantable scope. Route the list through the same is_destination_visible predicate the assignment validator and the request-time resolver use, so a non-proxy-admin sees only destinations granted to a team or org they administer. The predicate now operates on the typed CredentialInfo / CredentialAccess models instead of raw dict .get(), so visibility, assignment, and resolution share one typed source of truth.
This commit is contained in:
yucheng-berriai
parent
190190cce6
commit
8aebda7972
6 changed files with 392 additions and 99 deletions
|
|
@ -2,6 +2,8 @@
|
|||
CRUD endpoints for storing reusable credentials.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Optional
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Path, Request, Response
|
||||
|
|
@ -25,6 +27,10 @@ from litellm.proxy.credential_endpoints.access_decision import (
|
|||
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,
|
||||
|
|
@ -57,26 +63,36 @@ def _summarize_validation_error(ve: ValidationError) -> str:
|
|||
return "; ".join(parts)
|
||||
|
||||
|
||||
async def _caller_grantable_team_ids(
|
||||
@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: "Optional[PrismaClient]"
|
||||
) -> frozenset[str]:
|
||||
"""Team ids the caller may add to / remove from a destination's access.teams.
|
||||
) -> CallerAdminScope:
|
||||
"""The teams and orgs the caller administers.
|
||||
|
||||
Two paths to grantability:
|
||||
|
||||
1. Direct team-admin: caller is admin of the team (role=admin in
|
||||
``members_with_roles``).
|
||||
2. Via org-admin: caller is ORG_ADMIN of the team's organization. Org
|
||||
admins manage every team in their org, even teams they aren't a
|
||||
direct member of.
|
||||
|
||||
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.
|
||||
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 frozenset()
|
||||
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
|
||||
|
||||
|
|
@ -90,44 +106,60 @@ async def _caller_grantable_team_ids(
|
|||
proxy_logging_obj=proxy_logging_obj,
|
||||
)
|
||||
if user_obj is None:
|
||||
return frozenset()
|
||||
return CallerAdminScope(frozenset(), frozenset())
|
||||
|
||||
# Direct team-admin grants: walk the caller's own team list.
|
||||
team_admin_of: set[str] = set()
|
||||
for team_id in [tid for tid in (getattr(user_obj, "teams", None) or []) if isinstance(tid, str)]:
|
||||
team_obj = await get_team_object(
|
||||
team_id=team_id,
|
||||
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,
|
||||
# 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 [])
|
||||
):
|
||||
team_admin_of.add(team_id)
|
||||
)
|
||||
)
|
||||
|
||||
# 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: list[str] = [
|
||||
# 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_grantable: set[str] = set()
|
||||
if org_admin_of:
|
||||
org_teams = await prisma_client.db.litellm_teamtable.find_many( # type: ignore[union-attr]
|
||||
where={"organization_id": {"in": org_admin_of}}
|
||||
)
|
||||
org_teams = (
|
||||
await prisma_client.db.litellm_teamtable.find_many( # type: ignore[union-attr]
|
||||
where={"organization_id": {"in": list(org_admin_of)}}
|
||||
)
|
||||
org_grantable = {t.team_id for t in org_teams if t.team_id}
|
||||
if org_admin_of
|
||||
else []
|
||||
)
|
||||
org_grantable = frozenset(t.team_id for t in org_teams if t.team_id)
|
||||
|
||||
return frozenset(team_admin_of | org_grantable)
|
||||
return CallerAdminScope(team_admin_of | org_grantable, org_admin_of)
|
||||
except Exception: # noqa: BLE001
|
||||
# Best-effort lookup. A miss here means the PATCH decider will deny any
|
||||
# patch other than a no-op, which is the safe fallback.
|
||||
verbose_proxy_logger.exception("team-admin lookup failed")
|
||||
return frozenset()
|
||||
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: "Optional[PrismaClient]"
|
||||
) -> 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) -> Optional[CredentialItem]:
|
||||
|
|
@ -267,12 +299,13 @@ async def get_credentials(
|
|||
"""
|
||||
[BETA] endpoint. This might change unexpectedly.
|
||||
|
||||
Proxy admins see every credential (values masked). Team-admins and
|
||||
org-admins see only logging-typed destinations so they can self-assign
|
||||
them; provider credentials stay invisible to non-PROXY_ADMINs. Plain
|
||||
internal users with no team-admin or org-admin status get 403 — they
|
||||
have no use for the list and shouldn't see destination names, hosts,
|
||||
or scope metadata (Veria F2).
|
||||
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).
|
||||
"""
|
||||
from litellm.proxy.proxy_server import prisma_client
|
||||
|
||||
|
|
@ -280,8 +313,8 @@ async def get_credentials(
|
|||
if _is_proxy_admin(user_api_key_dict):
|
||||
visible = list(litellm.credential_list)
|
||||
else:
|
||||
grantable = await _caller_grantable_team_ids(user_api_key_dict, prisma_client)
|
||||
if not grantable:
|
||||
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={
|
||||
|
|
@ -295,7 +328,9 @@ async def get_credentials(
|
|||
visible = [
|
||||
credential
|
||||
for credential in litellm.credential_list
|
||||
if is_admin_gated_credential_info(credential.credential_info)
|
||||
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)
|
||||
]
|
||||
masked_credentials = [
|
||||
{
|
||||
|
|
|
|||
|
|
@ -619,22 +619,24 @@ async def _resolve_logging_exporters(
|
|||
from litellm.integrations.otel.presets.destinations import build_destination
|
||||
from litellm.proxy.management_endpoints.logging_exporter_access import (
|
||||
access_grants,
|
||||
is_auto_enable,
|
||||
identity_scope,
|
||||
parse_credential_info,
|
||||
)
|
||||
|
||||
team_id = user_api_key_dict.team_id
|
||||
org_id = await _effective_org_id(user_api_key_dict)
|
||||
names = await _union_logging_exporter_names(user_api_key_dict, org_id)
|
||||
team_ids, org_ids = identity_scope(team_id, org_id)
|
||||
|
||||
def _selected(credential: "CredentialItem") -> bool:
|
||||
info = credential.credential_info or {}
|
||||
if info.get("credential_type") != "logging":
|
||||
info = parse_credential_info(credential.credential_info)
|
||||
if info is None or info.credential_type != "logging":
|
||||
return False
|
||||
if is_auto_enable(info):
|
||||
if info.auto_enable:
|
||||
return True
|
||||
if credential.credential_name not in names:
|
||||
return False
|
||||
return access_grants(info.get("access"), team_id, org_id)
|
||||
return access_grants(info.access, team_ids, org_ids)
|
||||
|
||||
def _build(
|
||||
credential: "CredentialItem",
|
||||
|
|
|
|||
|
|
@ -1,41 +1,79 @@
|
|||
"""Shared access predicate for admin-owned logging destinations.
|
||||
"""Shared visibility predicate for admin-owned logging destinations.
|
||||
|
||||
``credential_info.access`` answers "who may see/assign this destination" — it is
|
||||
visibility, decoupled from enablement (which lives in ``metadata.logging_exporters``
|
||||
and the explicit ``auto_enable`` flag). The request-time resolver and the write-time
|
||||
validator gate on the SAME predicate so "visible" means the same thing on both sides:
|
||||
a team/org admin can only assign destinations they can see, and the resolver
|
||||
defensively re-checks visibility at request time.
|
||||
``credential_info.access`` answers "who may see and assign this destination". It
|
||||
is visibility, decoupled from enablement (a named assignment plus the explicit
|
||||
``auto_enable`` default-on flag). One predicate serves three callers so "visible"
|
||||
means the same thing everywhere: the ``GET /credentials`` list filter, the
|
||||
assignment validator, and the request-time resolver. When these disagree a
|
||||
non-admin can be shown, or route traffic to, a destination it should never see;
|
||||
keeping the check in one place is what prevents that.
|
||||
|
||||
``access_grants`` is the primitive: does this ``access`` reach a caller whose
|
||||
admin scope is the given set of team ids and org ids. Single-identity callers
|
||||
(the resolver, the per-write assignment gate) pass a one-element scope built with
|
||||
``identity_scope``; the list endpoint, whose caller may administer several teams
|
||||
and orgs, passes the full scope.
|
||||
"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
AUTO_ENABLE_KEY = "auto_enable"
|
||||
from pydantic import ValidationError
|
||||
|
||||
from litellm.models.credentials import CredentialAccess, CredentialInfo
|
||||
|
||||
|
||||
def access_grants(access: object, team_id: Optional[str], org_id: Optional[str]) -> bool:
|
||||
"""Whether a destination's ``access`` makes it visible to this identity.
|
||||
def parse_credential_info(raw: object) -> Optional[CredentialInfo]:
|
||||
"""Parse stored ``credential_info`` into the typed model, or ``None`` when it is
|
||||
absent or malformed.
|
||||
|
||||
``global`` reaches everyone; otherwise the identity's team or org must be listed.
|
||||
A missing or malformed ``access`` grants no one (fail closed): visibility must be
|
||||
an explicit admin grant, never an accident of an absent field.
|
||||
Callers fail closed on ``None``: a destination whose stored ``access`` cannot be
|
||||
parsed (a legacy shape the strict read model rejects) is treated as invisible
|
||||
rather than granted to everyone.
|
||||
"""
|
||||
if not isinstance(access, dict):
|
||||
if not isinstance(raw, dict):
|
||||
return None
|
||||
try:
|
||||
return CredentialInfo.model_validate(raw)
|
||||
except ValidationError:
|
||||
return None
|
||||
|
||||
|
||||
def identity_scope(team_id: Optional[str], org_id: Optional[str]) -> tuple[frozenset[str], frozenset[str]]:
|
||||
"""A single request identity's admin scope as ``(team_ids, org_ids)`` for
|
||||
``access_grants`` / ``is_destination_visible``."""
|
||||
return (
|
||||
frozenset({team_id}) if team_id else frozenset(),
|
||||
frozenset({org_id}) if org_id else frozenset(),
|
||||
)
|
||||
|
||||
|
||||
def access_grants(
|
||||
access: Optional[CredentialAccess],
|
||||
team_ids: frozenset[str],
|
||||
org_ids: frozenset[str],
|
||||
) -> bool:
|
||||
"""Whether ``access`` makes a destination visible to a caller admin-scoped to
|
||||
``team_ids`` / ``org_ids``.
|
||||
|
||||
``global`` reaches everyone; otherwise one of the caller's admin teams or orgs
|
||||
must be granted. A missing ``access`` grants no one (fail closed): visibility is
|
||||
an explicit admin grant, never the accident of an absent field.
|
||||
"""
|
||||
if access is None:
|
||||
return False
|
||||
if access.get("global") is True:
|
||||
if access.global_:
|
||||
return True
|
||||
teams = access.get("teams")
|
||||
if team_id is not None and isinstance(teams, (list, tuple)) and team_id in teams:
|
||||
if not team_ids.isdisjoint(access.teams):
|
||||
return True
|
||||
orgs = access.get("orgs")
|
||||
return org_id is not None and isinstance(orgs, (list, tuple)) and org_id in orgs
|
||||
return not org_ids.isdisjoint(access.orgs)
|
||||
|
||||
|
||||
def is_auto_enable(credential_info: object) -> bool:
|
||||
"""Whether a destination is an explicit global/default (auto-enabled everywhere).
|
||||
|
||||
This is the deliberate replacement for the old behavior where ``access.global``
|
||||
implicitly auto-enabled a destination for every request. Enablement is now opt-in:
|
||||
only ``auto_enable`` turns a destination on without being named.
|
||||
def is_destination_visible(
|
||||
info: CredentialInfo,
|
||||
team_ids: frozenset[str],
|
||||
org_ids: frozenset[str],
|
||||
) -> bool:
|
||||
"""Whether a caller admin-scoped to ``team_ids`` / ``org_ids`` may see and assign
|
||||
this destination: an auto-enabled default, or a grant that reaches their scope.
|
||||
"""
|
||||
return isinstance(credential_info, dict) and credential_info.get(AUTO_ENABLE_KEY) is True
|
||||
return info.auto_enable or access_grants(info.access, team_ids, org_ids)
|
||||
|
|
|
|||
|
|
@ -18,8 +18,9 @@ from fastapi import HTTPException, status
|
|||
import litellm
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
from litellm.proxy.management_endpoints.logging_exporter_access import (
|
||||
access_grants,
|
||||
is_auto_enable,
|
||||
identity_scope,
|
||||
is_destination_visible,
|
||||
parse_credential_info,
|
||||
)
|
||||
|
||||
LOGGING_EXPORTERS_KEY = "logging_exporters"
|
||||
|
|
@ -98,12 +99,13 @@ def _reject_unassignable_destinations(
|
|||
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 (
|
||||
is_auto_enable(by_name.get(name))
|
||||
or access_grants((by_name.get(name) or {}).get("access"), scope_team_id, scope_org_id)
|
||||
(info := parse_credential_info(by_name.get(name))) is not None
|
||||
and is_destination_visible(info, team_ids, org_ids)
|
||||
)
|
||||
]
|
||||
if unassignable:
|
||||
|
|
|
|||
|
|
@ -339,14 +339,23 @@ def _team_admin_of(team_ids):
|
|||
|
||||
@pytest.fixture
|
||||
def _patch_team_admin_lookup(monkeypatch):
|
||||
"""Substitute the DB-backed grant-capability lookup with a configurable mock."""
|
||||
"""Substitute the DB-backed admin-scope lookup with a configurable mock.
|
||||
|
||||
holder = {"ids": frozenset()}
|
||||
``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 holder["ids"]
|
||||
return endpoints.CallerAdminScope(
|
||||
team_ids=frozenset(holder["ids"]),
|
||||
org_ids=frozenset(holder["org_ids"]),
|
||||
)
|
||||
|
||||
monkeypatch.setattr(endpoints, "_caller_grantable_team_ids", _fake)
|
||||
monkeypatch.setattr(endpoints, "_caller_admin_scope", _fake)
|
||||
return holder
|
||||
|
||||
|
||||
|
|
@ -566,33 +575,107 @@ async def test_team_admin_cannot_flip_global(
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_credentials_filters_to_logging_for_non_admin(
|
||||
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_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."""
|
||||
monkeypatch.setattr(
|
||||
litellm,
|
||||
"credential_list",
|
||||
[
|
||||
CredentialItem(
|
||||
credential_name="openai",
|
||||
credential_values={"api_key": "sk-secret"},
|
||||
credential_info={"custom_llm_provider": "openai"},
|
||||
),
|
||||
CredentialItem(
|
||||
credential_name="poc-langfuse",
|
||||
credential_values={"public_key": "pk-1"},
|
||||
credential_info=_DEST_WITH_TEAMS,
|
||||
credential_info=_DEST_WITH_TEAMS, # granted to team-existing
|
||||
),
|
||||
],
|
||||
)
|
||||
_patch_team_admin_lookup["ids"] = frozenset({"team-T"})
|
||||
_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 == ["poc-langfuse"]
|
||||
assert names == ["org-dest"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
|
|||
|
|
@ -0,0 +1,133 @@
|
|||
"""The shared visibility predicate for admin-owned logging destinations.
|
||||
|
||||
This is the single chokepoint the list endpoint, the assignment validator, and
|
||||
the request-time resolver all route through, so a mutation here would let a
|
||||
non-admin see or route to a destination outside their scope. Each case is written
|
||||
to fail if the corresponding branch is flipped.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.abspath("../../../.."))
|
||||
|
||||
from litellm.models.credentials import CredentialAccess, CredentialInfo
|
||||
from litellm.proxy.management_endpoints.logging_exporter_access import (
|
||||
access_grants,
|
||||
identity_scope,
|
||||
is_destination_visible,
|
||||
parse_credential_info,
|
||||
)
|
||||
|
||||
|
||||
# --- parse_credential_info: fail closed on bad input -----------------------
|
||||
|
||||
|
||||
def test_parse_none_for_non_dict():
|
||||
assert parse_credential_info(None) is None
|
||||
assert parse_credential_info("not a dict") is None
|
||||
assert parse_credential_info(["a"]) is None
|
||||
|
||||
|
||||
def test_parse_typed_access_and_auto_enable():
|
||||
info = parse_credential_info(
|
||||
{
|
||||
"credential_type": "logging",
|
||||
"description": "arize",
|
||||
"auto_enable": True,
|
||||
"access": {"global": True, "teams": ["t1"], "orgs": ["o1"]},
|
||||
}
|
||||
)
|
||||
assert info is not None
|
||||
assert info.credential_type == "logging"
|
||||
assert info.auto_enable is True
|
||||
assert info.access is not None
|
||||
assert info.access.global_ is True
|
||||
assert info.access.teams == ("t1",)
|
||||
assert info.access.orgs == ("o1",)
|
||||
|
||||
|
||||
def test_parse_missing_access_is_none_not_error():
|
||||
info = parse_credential_info({"credential_type": "logging"})
|
||||
assert info is not None
|
||||
assert info.access is None
|
||||
assert info.auto_enable is False
|
||||
|
||||
|
||||
def test_parse_malformed_access_fails_closed():
|
||||
"""A stored access with an unknown field is rejected by the strict read model;
|
||||
the parse must return None (invisible) rather than raise or grant."""
|
||||
assert parse_credential_info({"access": {"legacy_field": "x"}}) is None
|
||||
assert parse_credential_info({"access": "not-an-object"}) is None
|
||||
|
||||
|
||||
# --- access_grants: the primitive ------------------------------------------
|
||||
|
||||
|
||||
def _access(**kw) -> CredentialAccess:
|
||||
return CredentialAccess.model_validate(kw)
|
||||
|
||||
|
||||
def test_access_grants_global_reaches_empty_scope():
|
||||
assert access_grants(_access(**{"global": True}), frozenset(), frozenset()) is True
|
||||
|
||||
|
||||
def test_access_grants_none_denies():
|
||||
assert access_grants(None, frozenset({"t1"}), frozenset({"o1"})) is False
|
||||
|
||||
|
||||
def test_access_grants_team_match():
|
||||
a = _access(teams=["t1", "t2"])
|
||||
assert access_grants(a, frozenset({"t2"}), frozenset()) is True
|
||||
assert access_grants(a, frozenset({"t3"}), frozenset()) is False
|
||||
|
||||
|
||||
def test_access_grants_org_match():
|
||||
a = _access(orgs=["o1"])
|
||||
assert access_grants(a, frozenset(), frozenset({"o1"})) is True
|
||||
assert access_grants(a, frozenset(), frozenset({"o2"})) is False
|
||||
|
||||
|
||||
def test_access_grants_disjoint_denies():
|
||||
a = _access(teams=["t1"], orgs=["o1"])
|
||||
assert access_grants(a, frozenset({"t9"}), frozenset({"o9"})) is False
|
||||
|
||||
|
||||
def test_access_grants_not_global_when_false():
|
||||
"""global=False must not short-circuit to visible."""
|
||||
a = _access(**{"global": False})
|
||||
assert access_grants(a, frozenset({"t1"}), frozenset({"o1"})) is False
|
||||
|
||||
|
||||
# --- is_destination_visible: auto_enable OR grant --------------------------
|
||||
|
||||
|
||||
def test_visible_auto_enable_ignores_access():
|
||||
info = CredentialInfo(credential_type="logging", auto_enable=True)
|
||||
assert is_destination_visible(info, frozenset(), frozenset()) is True
|
||||
|
||||
|
||||
def test_visible_delegates_to_access_when_not_auto_enable():
|
||||
info = CredentialInfo(credential_type="logging", access=_access(teams=["t1"]))
|
||||
assert is_destination_visible(info, frozenset({"t1"}), frozenset()) is True
|
||||
assert is_destination_visible(info, frozenset({"t2"}), frozenset()) is False
|
||||
|
||||
|
||||
def test_visible_denies_when_neither():
|
||||
info = CredentialInfo(credential_type="logging")
|
||||
assert is_destination_visible(info, frozenset({"t1"}), frozenset({"o1"})) is False
|
||||
|
||||
|
||||
# --- identity_scope --------------------------------------------------------
|
||||
|
||||
|
||||
def test_identity_scope_single_elements():
|
||||
teams, orgs = identity_scope("t1", "o1")
|
||||
assert teams == frozenset({"t1"})
|
||||
assert orgs == frozenset({"o1"})
|
||||
|
||||
|
||||
def test_identity_scope_empty_for_none():
|
||||
teams, orgs = identity_scope(None, None)
|
||||
assert teams == frozenset()
|
||||
assert orgs == frozenset()
|
||||
Loading…
Add table
Reference in a new issue