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

feat(team): let a team admin manage their own team's logging callbacks (#37667)

Browse source 
* feat(team): let a team admin manage their own team's logging callbacks

The team callback endpoints already authorize correctly: POST, GET and DELETE
each call _verify_team_access, which admits a proxy admin, an org admin for the
team, or an admin of that team, and 403s everyone else. The route-permission
layer never let a team admin reach them, so it answered 401 naming proxy admin
and the handler's own check was dead code for the caller it was written for.

Adding the two paths to self_managed_routes is how every other team-admin route
works: /team/member_add, /team/member_delete, /team/member_update and
/team/permissions_update all sit in that list and scope per team inside the
handler. The entries use the :path converter the routes are registered with, so
a team id containing a slash resolves the same way at the gate as at the router.

Because any authenticated caller now reaches these handlers, an unknown team had
to stop being distinguishable from one the caller may not manage. All three
handlers looked the team up and raised a distinct 'does not exist' before the
access check, which would have let any valid key probe for team ids. That branch
now returns the same 403 body _verify_team_access raises, and keeps the
diagnosable error for a proxy admin.

disable_logging stays out of the grant. That is a scope decision rather than a
security boundary, since a team admin holding DELETE can clear callbacks one at
a time; it differs only in also clearing the deprecated callback_settings shape.

* fix(team): reach the callback routes for a team id containing a colon

The route gate expands {team_id:path} to "[^:]+" so a colon-suffixed provider
route is not swallowed, which means the two entries added here matched a team
id with a slash but not one with a colon, while the router accepts both.
team_id is a free-form string, so a team whose id contains a colon kept the
old proxy-admin-only denial and its admin could not manage its own callbacks.
List both spellings rather than relaxing the shared matcher, which every
":path" route depends on. The comment claimed the two matchers agree; they do
not, so it now says what each placeholder actually accepts.

* fix(auth): match a :path placeholder the way the router's converter does

A team id may carry a slash, a colon, or both. The gate expanded {x:path}
to "[^:]+", so an id with a colon in it matched no self_managed_routes
entry and its team admin got the proxy-admin-only denial on a route the
router had already resolved for them. Listing a second {x} spelling covered
a colon or a slash but never both.

Expand {x:path} to ".+" instead, except when the template puts a ":"
literal of its own after the placeholder, which is where the narrower form
was earning its keep: the Google routes end in ":generateContent" and
friends, and there the value has to stop before that suffix rather than
swallow it and match a different verb.

That lets self_managed_routes drop back to the two :path spellings the
router itself mounts.

* test(auth): pin that the callback grant reaches no neighbouring team route

The grant is two templates ending in the callback suffix, and the
placeholder now takes slashes and colons. Every other route under
/team/{team_id} registers an ordinary single-segment placeholder, so no
URL the router sends to one of them can end in the callback suffix.

Pin that, so adding a path-converter route beside these fails here
rather than by handing a caller a handler the grant never covered.

* fix(team): make one entry own a credential family end to end

Every stored entry's callback_vars are flattened into one dict before a
request reads them, and that dict is what the exporter authenticates and
addresses with. So an entry naming only a destination is enough to
redirect a credential written somewhere else: a host on a second entry
pairs with the key pair from the first, and the request carries that key
pair to the new host. A team admin cannot read the team's masked Langfuse
secret, but could add such an entry and receive it.

Reject, for writers who are not proxy admins, an entry using a credential
family another entry already holds. Family rather than callback name,
because langfuse and langfuse_otel configure one Langfuse project and
would otherwise redirect each other, and because a destination like
dd_agent_host that no integration registry lists still pairs with the
Datadog credentials beside it.

A proxy admin already holds every credential the proxy has, so the rule
would buy nothing there and would break configs that predate it. A team
admin who does want to move a family deletes the entry holding it first,
which reveals nothing.

* fix(team): let one integration cover both callback events

The family rule compared variable names only, so a team admin who registered
an integration for the success event could not register the same integration,
with the same values, for the failure event.

Compare the values as well: repeating what the owning entry already stores
flattens to the same dict, so there is nothing to redirect. The stored side is
decrypted first, because the credentials are encrypted at rest and ciphertext
never equals the plaintext coming in.

* fix(team): compare the family's values, not its variable names

Comparing per variable rejected a credential written under its other spelling:
langfuse_secret and langfuse_secret_key are one key, so repeating the stored
secret under the other name read as a new value.

Ask instead whether the value is one the owning entries already carry. A
destination the caller controls is by definition not, so the redirect stays
closed, and no alias table has to stay complete for that to hold.

* fix(team): pin the family's configured variables as well as its values

Asking only whether a value is one the family holds let a held variable be
given another of the family's values, so the exporter would address or
authenticate with it.

Keep the value membership rule for a variable the family does not configure
yet, which is what lets one credential go in under its other spelling, and
require a variable it does configure to keep the value it has. Between them no
value the caller chose can enter the family.

* fix(auth): keep a newline in a :path value visible to the route gate

"." stops at a newline and the router's path converter does not, so a %0A
anywhere in a :path segment left the route unmatched here while still reaching
the handler. Every list built on this matcher inherited that: on a proxy with
DISABLE_ADMIN_ENDPOINTS set, DELETE /v1/mcp/server/abc%0Adef reached the MCP
handler instead of the 403 the same request gets without the %0A.

Match with a class that spans newlines.
This commit is contained in:
yucheng-berri 2026-09-08 15:58:58 -07:00 • committed by GitHub
parent 4b9c289a72
commit 90731576e3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 459 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/proxy/_types.py
View file

@ -835,6 +835,14 @@ class LiteLLMRoutes(enum.Enum):
"/team/daily/activity/aggregated",
"/team/spend/by_user",
"/team/{team_id}/members/me",
# POST/GET the team's logging callbacks, and DELETE one of them. Every
# handler calls _verify_team_access, which admits only a proxy admin, an
# org admin for the team, or an admin of this team.
#
# team_id is a free-form string, so it spells these with the same path
# converter the router uses; the gate matches that converter.
"/team/{team_id:path}/callback",
"/team/{team_id:path}/callback/{callback_name}",
"/model/new",
"/model/update",
"/model/delete",

20
litellm/proxy/auth/route_checks.py
View file

@ -497,10 +497,22 @@ class RouteChecks:
def _placeholder_to_regex(match: re.Match) -> str:
placeholder: Final = match.group(0).strip("{}")
if placeholder.endswith(":path"):
# allow "/" in the placeholder value, but don't eat the route suffix after ":"
return r"[^:]+"
return r"[^/]+"
if not placeholder.endswith(":path"):
return r"[^/]+"
# A ":path" placeholder takes whatever the router's own path
# converter takes, slashes and colons alike, so an id spelled with
# either (or both) still matches the template it was mounted under.
#
# Unless the template puts a ":" literal of its own after the
# placeholder: the Google routes end in ":generateContent" and
# friends, and there the value has to stop before that suffix
# rather than swallow it and match a different verb.
#
# "[\s\S]" rather than ".", because "." stops at a newline and the
# path converter does not: a %0A anywhere in the value would leave
# the route unmatched here while still reaching the handler, which
# turns this gate into a bypass for the lists built on it.
return r"[^:]+" if ":" in match.string[match.end() :] else r"[\s\S]+"
pattern = re.sub(r"\{[^}]+\}", _placeholder_to_regex, pattern)
# Anchor the pattern to match the entire string

85
litellm/proxy/common_utils/callback_config_validation.py
View file

@ -44,6 +44,91 @@ def _langfuse_environment_error(callback_vars: Mapping[str, str]) -> str | None:
return None
# Which credential family a dynamic variable belongs to. The families are the
# integrations that share one account: every langfuse_* variable configures the
# same Langfuse project whether it rides the classic callback or the OTel one,
# and every dd_* variable configures the same Datadog account.
_VAR_FAMILIES: Final[Mapping[str, str]] = MappingProxyType(
{
"arize_": "Arize",
"dd_": "Datadog",
"gcs_": "GCS",
"humanloop_": "Humanloop",
"langfuse_": "Langfuse",
"langsmith_": "LangSmith",
"newrelic_": "New Relic",
"posthog_": "PostHog",
"wandb_": "Weights & Biases",
"weave_": "Weights & Biases",
}
)
def _family_of(var: str) -> str | None:
"""The credential family ``var`` configures, or ``None`` if it configures none.
``turn_off_message_logging`` and friends belong to no backend, so they carry
no credentials anyone could redirect.
"""
return next((family for prefix, family in _VAR_FAMILIES.items() if var.startswith(prefix)), None)
def cross_entry_family_error(
callback_vars: Mapping[str, str] | None,
stored_vars_by_entry: Sequence[Mapping[str, str]],
) -> str | None:
"""Reject an entry that changes what a family another entry holds resolves to.
Every stored entry's variables are flattened into one dict before a request
reads them, and the flattened dict is what the exporter authenticates and
addresses with. So an entry naming only a destination is enough to redirect
credentials that were written somewhere else: a host on a second entry pairs
with the key from the first, and the request carries that key to the new
host.
Two rules together keep the flattened dict out of the caller's hands. A
variable the family already configures has to keep the value it has, so
nothing already in use can be moved. A variable the family does not yet
configure may only carry a value the family already holds, which is what lets
the same credential go in under its other spelling (``langfuse_secret`` and
``langfuse_secret_key`` are one key) without anything here having to list the
spellings. Between them, no value the caller chose can enter the family, and
repeating the family as it stands is still allowed -- that is how one
integration gets registered for both the success and the failure event.
A team admin who does want to move a family deletes the entry holding it
first, which reveals nothing.
Only the writers this endpoint newly admits are held to this, because a proxy
admin already holds every credential the proxy has.
``stored_vars_by_entry`` has to arrive decrypted; the credential values are
encrypted at rest and ciphertext never equals the plaintext coming in.
"""
if not callback_vars:
return None
stored_by_var: Final = {
var: value for entry in stored_vars_by_entry for var, value in entry.items() if _family_of(var) is not None
}
family_values: Final = frozenset(
(family, value)
for entry in stored_vars_by_entry
for var, value in entry.items()
if (family := _family_of(var)) is not None
)
held_families: Final = frozenset(family for family, _ in family_values)
return next(
(
f"{family} is already configured by another callback entry on this team. "
f"Remove that entry before setting {var} here."
for var, value, family in ((v, callback_vars[v], _family_of(v)) for v in callback_vars)
if family in held_families
and (stored_by_var[var] != value if var in stored_by_var else (family, value) not in family_values)
),
None,
)
def logging_metadata_config_error(metadata: Mapping[str, object] | None) -> str | None:
"""Validate every ``logging`` entry of a team/key metadata payload."""
if not metadata:

56
litellm/proxy/management_endpoints/team_callback_endpoints.py
View file

@ -20,6 +20,7 @@ from litellm.proxy._types import (
LiteLLM_AuditLogs,
LiteLLM_TeamTable,
LitellmTableNames,
LitellmUserRoles,
ProxyErrorTypes,
ProxyException,
TeamCallbackDeleteResponse,
@ -28,7 +29,10 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.callback_config_validation import callback_config_error
from litellm.proxy.common_utils.callback_config_validation import (
callback_config_error,
cross_entry_family_error,
)
from litellm.proxy.common_utils.callback_utils import (
_CALLBACK_VAR_ENCRYPTED_PREFIX,
decrypt_callback_vars,
@ -230,6 +234,22 @@ def _callback_error(status_code: int, message: str) -> HTTPException:
)
def _unknown_team_error(team_id: str, user_api_key_dict: UserAPIKeyAuth, status_code: int) -> HTTPException:
"""Report an unknown team without telling an unauthorized caller that it is unknown.
These routes are reachable by any authenticated caller so that a team admin can
get as far as _verify_team_access. A distinct "does not exist" would therefore let
any valid key probe which team ids exist, so a caller who could not have managed
the team either way gets the same 403 body _verify_team_access raises.
"""
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return _callback_error(status_code, f"Team id = {team_id} does not exist.")
return HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="You do not have access to this team",
)
@router.post(
"/team/{team_id:path}/callback",
tags=["team management"],
@ -304,10 +324,7 @@ async def add_team_callbacks(
# Check if team_id exists already
_existing_team = await prisma_client.get_data(team_id=team_id, table_name="team", query_type="find_unique")
if _existing_team is None:
raise HTTPException(
status_code=400,
detail={"error": f"Team id = {team_id} does not exist. Please use a different team id."},
)
raise _unknown_team_error(team_id, user_api_key_dict, status.HTTP_400_BAD_REQUEST)
# IDOR guard: only proxy admins / org admins / team admins of THIS
# team may write callback credentials. Without this, any
@ -326,6 +343,28 @@ async def add_team_callbacks(
if team_callback_settings is None or not isinstance(team_callback_settings, list):
team_callback_settings = []
# One entry has to own a credential family end to end. The entries are
# flattened into one dict before a request reads them, so an entry
# naming only a destination would pair with a key written on another
# entry and carry it to that destination -- a key a team admin can read
# back nowhere. Repeating a value the owning entry already stores is
# fine, which is how one integration covers both events. Proxy admins
# are exempt: they already hold every credential the proxy has.
if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN:
# Decrypted, because the check compares the incoming values against
# the stored ones and the credentials are encrypted at rest.
decrypted_logging: Final = decrypt_callback_vars(team_metadata).get("logging")
stored_entries: Final = decrypted_logging if isinstance(decrypted_logging, list) else ()
stored_entry_vars: Final = [ # mutable-ok: read-only input to the check, never stored
entry.get("callback_vars") or {} for entry in stored_entries
]
family_error: Final = cross_entry_family_error(data.callback_vars, stored_entry_vars)
if family_error is not None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=family_error,
)
## check if it already exists, for the same callback event
for callback in team_callback_settings:
if (
@ -452,7 +491,7 @@ async def delete_team_callback(
team_id=team_id, table_name="team", query_type="find_unique"
)
if _existing_team is None:
raise _callback_error(404, f"Team id = {team_id} does not exist.")
raise _unknown_team_error(team_id, user_api_key_dict, status.HTTP_404_NOT_FOUND)
# IDOR guard: only proxy admins / org admins / team admins of THIS team may
# deregister its callbacks, otherwise any authenticated key holder could
@ -726,10 +765,7 @@ async def get_team_callbacks(
# Check if team_id exists
_existing_team = await prisma_client.get_data(team_id=team_id, table_name="team", query_type="find_unique")
if _existing_team is None:
raise HTTPException(
status_code=404,
detail={"error": f"Team id = {team_id} does not exist."},
)
raise _unknown_team_error(team_id, user_api_key_dict, status.HTTP_404_NOT_FOUND)
# IDOR guard: callback metadata holds third-party API credentials
# (Langfuse / Langsmith / GCS). Only proxy admins / org admins /

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

@ -3638,3 +3638,191 @@ def test_agent_registry_route_gate_open_to_non_admin_roles(user_role, method, ro
valid_token=valid_token,
request_data={},
)
TEAM_CALLBACK_ROUTES = (
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/callback",
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/callback/langfuse",
# the routes register team_id with the :path converter, so a team id may
# contain a slash
"/team/tenant/06bda574/callback",
"/team/tenant/06bda574/callback/langfuse",
# team_id is a free-form string, so it may also contain a colon
"/team/tenant:06bda574/callback",
"/team/tenant:06bda574/callback/langfuse",
# or both, which is the shape neither a "[^:]+" nor a "[^/]+" expansion
# of the placeholder reaches on its own
"/team/tenant:acme/prod/callback",
"/team/tenant:acme/prod/callback/langfuse",
)
def _gate(route, role) -> str:
"""Drive the real route gate for a non-proxy-admin caller.
Reports "allowed" when the gate lets the request through to its handler, and
the denial message otherwise, so a caller asserts the verdict as a value
instead of on whether an exception escaped.
"""
user_obj = LiteLLM_UserTable(
user_id="team_admin_user",
user_email="team-admin@example.com",
user_role=role,
)
request = MagicMock(spec=Request)
request.query_params = {}
try:
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=role,
route=route,
request=request,
valid_token=UserAPIKeyAuth(user_id="team_admin_user", user_role=role),
request_data={},
)
except Exception as exc:
return f"denied: {exc}"
return "allowed"
def test_team_callback_routes_are_self_managed():
"""The grant has to come from self_managed_routes specifically.
That list is the one whose entries carry no role predicate, so the handler
decides. Granting the same paths through internal_user_routes instead would
look identical for an internal_user while silently denying the org admins and
view-only roles that list does not cover.
"""
for template in (
"/team/{team_id:path}/callback",
"/team/{team_id:path}/callback/{callback_name}",
):
assert template in LiteLLMRoutes.self_managed_routes.value
@pytest.mark.parametrize("route", TEAM_CALLBACK_ROUTES)
@pytest.mark.parametrize(
"role",
[
LitellmUserRoles.INTERNAL_USER.value,
LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value,
LitellmUserRoles.ORG_ADMIN.value,
],
)
def test_team_callback_routes_reach_their_handler_for_non_admins(route, role):
"""A team admin manages their own team's logging callbacks, so the route gate
must let a non-proxy-admin through to the handler.
The handler is what authorizes: every team callback endpoint calls
_verify_team_access, which admits only a proxy admin, an org admin for the
team, or an admin of that team, and 403s everyone else. Before this, the gate
rejected the team admin with a 401 naming proxy admin, so the handler's own
check was unreachable for them.
"""
assert _gate(route, role) == "allowed"
@pytest.mark.parametrize(
"pattern, route, matches",
[
# a :path placeholder takes what the router's path converter takes
("/team/{team_id:path}/callback", "/team/plain/callback", True),
("/team/{team_id:path}/callback", "/team/tenant/acme/callback", True),
("/team/{team_id:path}/callback", "/team/tenant:acme/callback", True),
("/team/{team_id:path}/callback", "/team/tenant:acme/prod/callback", True),
# and still has to reach the template's own suffix
("/team/{team_id:path}/callback", "/team/tenant:acme/disable_logging", False),
# a template with a ":" literal after the placeholder keeps the suffix
(
"/v1beta/models/{model_name:path}:generateContent",
"/v1beta/models/gemini-2.5-flash:generateContent",
True,
),
(
"/v1beta/models/{model_name:path}:generateContent",
"/v1beta/models/publishers/google/gemini-2.5-flash:generateContent",
True,
),
# the value must not swallow that suffix and match a different verb
(
"/v1beta/models/{model_name:path}:generateContent",
"/v1beta/models/gemini-2.5-flash:countTokens",
False,
),
# a %0A in the value reaches the handler through the path converter, so
# the gate has to see it too or DISABLE_ADMIN_ENDPOINTS is bypassable
("/v1/mcp/server/{path:path}", "/v1/mcp/server/abc\ndef", True),
("/team/{team_id:path}/callback", "/team/ten\nant/callback", True),
("/v1beta/models/{model_name:path}:generateContent", "/v1beta/models/gem\nini:generateContent", True),
# an ordinary placeholder stays one segment
("/team/{team_id}/members/me", "/team/abc/members/me", True),
("/team/{team_id}/members/me", "/team/tenant/abc/members/me", False),
("/team/{team_id}/members/me", "/team/ab\nc/members/me", True),
],
)
def test_path_placeholder_matches_what_the_router_accepts(pattern, route, matches):
"""The gate's placeholder expansion has to agree with the router's.
A team id may carry a slash, a colon, or both, and the router mounted these
paths with the same :path converter, so an id the router routes must not be
an id the gate fails to recognize. The one narrowing that stays is a template
whose own suffix begins with a colon: there the value stops before it, or
":generateContent" would also match a ":countTokens" request.
"""
assert RouteChecks._route_matches_pattern(route=route, pattern=pattern) is matches
# Every other route the proxy mounts under /team/{team_id}, spelled the way it
# is registered. None of them takes a path converter, so none can be reached by
# a URL that ends in the callback suffix.
PROTECTED_TEAM_ROUTES = (
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112",
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/disable_logging",
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/members/me",
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/member/u-1/reset_spend",
# the same routes with the callback suffix spliced in, which is the shape a
# caller would craft to make a protected route look self-managed
"/team/06bda574/callback/disable_logging/x",
"/team/06bda574/callback/member/u-1/reset_spend",
"/team/06bda574/callback/members/me",
)
@pytest.mark.parametrize("route", PROTECTED_TEAM_ROUTES)
def test_the_callback_grant_does_not_reach_another_team_route(route):
"""Widening the callback templates must not hand out any neighbouring route.
The grant is two templates ending in the callback suffix. Every other team
route registers an ordinary single-segment placeholder, so no URL the router
sends to one of them can end in "/callback" or "/callback/<name>" -- and the
gate must agree, or a crafted team id would carry a caller into a handler
the grant never covered.
"""
for template in (
"/team/{team_id:path}/callback",
"/team/{team_id:path}/callback/{callback_name}",
):
assert RouteChecks._route_matches_pattern(route=route, pattern=template) is False
def test_team_disable_logging_stays_proxy_admin_only():
"""disable_logging was left out of the grant, so it must still be rejected at
the gate. It is the one team callback route a team admin cannot reach."""
verdict = _gate(
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/disable_logging",
LitellmUserRoles.INTERNAL_USER.value,
)
assert "Only proxy admin" in verdict
assert "disable_logging" in verdict
@pytest.mark.parametrize(
"route",
[
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112",
"/team/update",
"/team/06bda574-5ca9-43d3-beb8-3b23c2f17112/model/add",
],
)
def test_neighbouring_team_routes_stay_closed(route):
"""The grant is the callback paths and nothing else on the team namespace."""
assert "Only proxy admin" in _gate(route, LitellmUserRoles.INTERNAL_USER.value)

116
tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py
View file

@ -19,6 +19,7 @@ from litellm.proxy._types import (
LitellmUserRoles,
UserAPIKeyAuth,
)
from litellm.proxy.common_utils.callback_config_validation import cross_entry_family_error
from litellm.proxy.management_endpoints.team_callback_endpoints import (
add_team_callbacks,
delete_team_callback,
@ -1443,3 +1444,118 @@ async def test_delete_team_callback_route_accepts_team_ids_containing_slashes():
assert response.json()["data"]["success_callbacks"] == ["langsmith"]
written = json.loads(mock_prisma.db.litellm_teamtable.update.await_args.kwargs["data"]["metadata"])
assert [entry["callback_name"] for entry in written["logging"]] == ["langsmith"]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"call_handler",
[
lambda caller: add_team_callbacks(
data=AddTeamCallback(
callback_name="langfuse",
callback_type="success",
callback_vars={"langfuse_public_key": "pk", "langfuse_secret_key": "sk"},
),
http_request=Mock(spec=Request),
team_id="team-does-not-exist",
user_api_key_dict=caller,
),
lambda caller: get_team_callbacks(
http_request=Mock(spec=Request),
team_id="team-does-not-exist",
user_api_key_dict=caller,
),
lambda caller: delete_team_callback(
http_request=Mock(spec=Request),
team_id="team-does-not-exist",
callback_name="langfuse",
user_api_key_dict=caller,
),
],
ids=["add", "get", "delete"],
)
async def test_unknown_team_is_indistinguishable_from_no_access(call_handler, unauthorized_caller):
"""An unauthorized caller must not learn whether a team id exists.
These routes are reachable by any authenticated caller so a team admin can get
as far as the access check, so a distinct "does not exist" would turn them into
a probe for valid team ids. The unknown-team response has to match the
no-access one exactly, status and body.
"""
with patch("litellm.proxy.proxy_server.prisma_client") as mock_client: # test-quality-ok: the handler imports prisma_client from proxy_server at call time, so there is no seam to inject through
mock_client.get_data = AsyncMock(return_value=None)
with pytest.raises(HTTPException) as unknown_team:
await call_handler(unauthorized_caller)
with patch("litellm.proxy.proxy_server.prisma_client") as mock_client: # test-quality-ok: the handler imports prisma_client from proxy_server at call time, so there is no seam to inject through
mock_client.get_data = AsyncMock(return_value=_team_row())
mock_client.db.litellm_teamtable.update = AsyncMock()
with patch( # test-quality-ok: _verify_team_access calls this module-level helper directly, so there is no seam to inject through
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
):
with pytest.raises(HTTPException) as no_access:
await call_handler(unauthorized_caller)
assert unknown_team.value.status_code == no_access.value.status_code == 403
assert unknown_team.value.detail == no_access.value.detail
assert "does not exist" not in str(unknown_team.value.detail)
@pytest.mark.asyncio
async def test_proxy_admin_still_told_the_team_is_unknown():
"""The masking is only for callers who could not have managed the team; a proxy
admin keeps the diagnosable error."""
admin = UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN, user_id="admin", api_key="sk-admin")
with patch("litellm.proxy.proxy_server.prisma_client") as mock_client: # test-quality-ok: the handler imports prisma_client from proxy_server at call time, so there is no seam to inject through
mock_client.get_data = AsyncMock(return_value=None)
with pytest.raises(HTTPException) as exc:
await get_team_callbacks(
http_request=Mock(spec=Request),
team_id="team-does-not-exist",
user_api_key_dict=admin,
)
assert exc.value.status_code == 404
assert "does not exist" in str(exc.value.detail)
@pytest.mark.parametrize(
"new_vars, stored, rejected",
[
# the redirect, in every carrier a caller could pick: an entry naming
# only a host, pairing with a key pair written on another entry
({"langfuse_host": "http://attacker.invalid"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], True),
# the sibling carrier -- langfuse and langfuse_otel are one account
({"langfuse_host": "http://attacker.invalid"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_secret_key": "sk"}], True),
# a destination variable no integration registry lists
({"dd_agent_host": "attacker.invalid"}, [{"dd_api_key": "k", "dd_site": "us5.datadoghq.com"}], True),
# one entry owning its family end to end is the feature
({"langfuse_host": "https://eu.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [], False),
# a different family alongside an existing one stays fine
({"gcs_bucket_name": "bucket"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False),
({"langsmith_api_key": "k"}, [{"dd_api_key": "k"}], False),
# variables that configure no backend carry nothing to redirect
({"turn_off_message_logging": "true"}, [{"langfuse_secret_key": "sk"}], False),
# the same integration registered for a second event: identical values
# flatten to the identical dict, so there is nothing to redirect
({"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False),
# the same credential under its other spelling is the same credential
({"langfuse_secret": "sk"}, [{"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], False),
# a value the family already holds cannot be moved into another of its
# variables either; the exporter would address or authenticate with it
({"langfuse_host": "pk"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk"}], True),
# the same shape with one value moved is the redirect again
({"langfuse_host": "http://attacker.invalid", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}, [{"langfuse_host": "https://us.cloud.langfuse.com", "langfuse_public_key": "pk", "langfuse_secret_key": "sk"}], True),
],
)
def test_one_entry_owns_a_credential_family(new_vars, stored, rejected):
"""A team admin must not be able to redirect a credential they cannot read.
The stored entries are flattened into one dict before a request reads them,
so an entry naming only a destination pairs with a key written elsewhere and
carries it to that destination.
"""
error = cross_entry_family_error(new_vars, stored)
assert (error is not None) is rejected