scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-12 23:01: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

Browse source 
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.
This commit is contained in:
Yucheng Zhu 2026-08-20 10:43:22 -07:00
parent c50d83ece2
commit e64b6f6a68
4 changed files with 196 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/proxy/_types.py
View file

@ -830,6 +830,12 @@ class LiteLLMRoutes(enum.Enum):
"/team/daily/activity",
"/team/daily/activity/aggregated",
"/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. The :path converter
# mirrors the route registration, which accepts a team id containing "/".
"/team/{team_id:path}/callback",
"/team/{team_id:path}/callback/{callback_name}",
"/model/new",
"/model/update",
"/model/delete",

29
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,
@ -230,6 +231,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 +321,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
@ -452,7 +466,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 +740,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 /

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

@ -3544,3 +3544,98 @@ 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",
)
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.
"""
assert "/team/{team_id:path}/callback" in LiteLLMRoutes.self_managed_routes.value
assert "/team/{team_id:path}/callback/{callback_name}" 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"
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)

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

@ -1443,3 +1443,78 @@ 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)