fix(team): gate /team/{id}/callback endpoints behind _verify_team_access

Three endpoints in ``team_callback_endpoints.py`` accept a ``team_id``
from the URL but never check whether the authenticated caller can
manage that team:

* ``POST /team/{team_id}/callback`` — write Langfuse / Langsmith / GCS
  credentials to any team
* ``POST /team/{team_id}/disable_logging`` — silence audit logging
  for any team
* ``GET /team/{team_id}/callback`` — read back another team's stored
  third-party API credentials

Each handler now runs the existing ``_verify_team_access`` helper
(proxy-admin / org-admin / team-admin hierarchy already used by sibling
endpoints in ``team_endpoints.py``) on the resolved team row before
the read or write.

Tests:

- ``test_add_team_callbacks_rejects_unauthorized_caller`` — internal
  user not on the team gets 403; DB write never happens.
- ``test_disable_team_logging_rejects_unauthorized_caller`` — same.
- ``test_get_team_callbacks_rejects_unauthorized_caller`` — same on the
  read path; victim team's callback data stays inaccessible.
- ``test_proxy_admin_can_add_team_callbacks`` — proxy admin still
  passes through to the DB write (sanity that the guard didn't
  over-rotate).
- ``test_team_admin_of_target_team_can_add_callbacks`` — team admin of
  the target team still passes through.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
user 2026-04-29 21:39:39 +00:00
parent 9b3cd5ca25
commit 140628063c
No known key found for this signature in database
2 changed files with 238 additions and 0 deletions

View file

@ -13,12 +13,14 @@ from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
from litellm._logging import verbose_proxy_logger
from litellm.proxy._types import (
AddTeamCallback,
LiteLLM_TeamTable,
ProxyErrorTypes,
ProxyException,
TeamCallbackMetadata,
UserAPIKeyAuth,
)
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management_endpoints.team_endpoints import _verify_team_access
from litellm.proxy.management_helpers.utils import management_endpoint_wrapper
router = APIRouter()
@ -100,6 +102,15 @@ async def add_team_callbacks(
},
)
# IDOR guard: only proxy admins / org admins / team admins of THIS
# team may write callback credentials. Without this, any
# authenticated key holder could overwrite another team's logging
# config (and read back the credentials they wrote).
await _verify_team_access(
team_obj=LiteLLM_TeamTable(**_existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
# store team callback settings in metadata
team_metadata = _existing_team.metadata
team_callback_settings: List[dict] = team_metadata.get(
@ -196,6 +207,14 @@ async def disable_team_logging(
detail={"error": f"Team id = {team_id} does not exist."},
)
# IDOR guard: only proxy admins / org admins / team admins of THIS
# team may disable its logging — otherwise any authenticated key
# holder can silence audit logging for any team.
await _verify_team_access(
team_obj=LiteLLM_TeamTable(**_existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
# Update team metadata to disable logging
team_metadata = _existing_team.metadata
team_callback_settings = team_metadata.get("callback_settings", {})
@ -305,6 +324,14 @@ async def get_team_callbacks(
detail={"error": f"Team id = {team_id} does not exist."},
)
# IDOR guard: callback metadata holds third-party API credentials
# (Langfuse / Langsmith / GCS). Only proxy admins / org admins /
# team admins of THIS team may read them.
await _verify_team_access(
team_obj=LiteLLM_TeamTable(**_existing_team.model_dump()),
user_api_key_dict=user_api_key_dict,
)
# Retrieve team callback settings from metadata
team_metadata = _existing_team.metadata
team_callback_settings = team_metadata.get("callback_settings", {})

View file

@ -0,0 +1,211 @@
"""
Regression tests for the IDOR fix on team callback endpoints
(GHSA-xxv2-fprq-9x93).
The three endpoints below previously authenticated the caller but never
checked whether the caller could manage the target team any
authenticated key holder could write callback credentials to any team,
disable any team's logging, or read back another team's stored
third-party API credentials (Langfuse / Langsmith / GCS).
The fix routes each handler through ``_verify_team_access``, which
enforces the proxy-admin / org-admin / team-admin hierarchy used by
sibling endpoints in ``team_endpoints.py``.
"""
import os
import sys
from unittest.mock import AsyncMock, MagicMock, Mock, patch
import pytest
from fastapi import HTTPException, Request
sys.path.insert(0, os.path.abspath("../../../.."))
from litellm.proxy._types import (
AddTeamCallback,
LitellmUserRoles,
Member,
UserAPIKeyAuth,
)
from litellm.proxy.management_endpoints.team_callback_endpoints import (
add_team_callbacks,
disable_team_logging,
get_team_callbacks,
)
def _other_team_existing_row():
"""Return a mock team row owned by someone other than the test caller."""
row = MagicMock()
row.model_dump.return_value = {
"team_id": "team-victim",
"team_alias": "victim-team",
"members_with_roles": [
{"role": "admin", "user_id": "victim_admin"},
],
"organization_id": "org-victim",
}
row.metadata = {}
return row
@pytest.fixture
def unauthorized_caller():
return UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="random_authenticated_user",
api_key="sk-random",
)
@pytest.fixture
def patched_prisma():
"""
Patch the proxy_server.prisma_client used inside each handler with a
mock that returns a victim-team row from get_data().
"""
with (
patch(
"litellm.proxy.proxy_server.prisma_client",
) as mock_client,
patch(
"litellm.proxy.management_endpoints.team_endpoints._is_user_org_admin_for_team",
new_callable=AsyncMock,
return_value=False,
),
):
mock_client.get_data = AsyncMock(return_value=_other_team_existing_row())
mock_client.db.litellm_teamtable.update = AsyncMock()
yield mock_client
@pytest.mark.asyncio
async def test_add_team_callbacks_rejects_unauthorized_caller(
patched_prisma, unauthorized_caller
):
data = AddTeamCallback(
callback_name="langfuse",
callback_type="success",
callback_vars={
"langfuse_public_key": "pk-attacker",
"langfuse_secret_key": "sk-attacker",
},
)
with pytest.raises(HTTPException) as exc:
await add_team_callbacks(
data=data,
http_request=Mock(spec=Request),
team_id="team-victim",
user_api_key_dict=unauthorized_caller,
)
assert exc.value.status_code == 403
# The unauthorized caller must NOT have written to the victim team.
patched_prisma.db.litellm_teamtable.update.assert_not_called()
@pytest.mark.asyncio
async def test_disable_team_logging_rejects_unauthorized_caller(
patched_prisma, unauthorized_caller
):
# The endpoint catches HTTPException and re-wraps it as ProxyException
# with the original status code preserved.
from litellm.proxy._types import ProxyException
with pytest.raises((HTTPException, ProxyException)) as exc:
await disable_team_logging(
http_request=Mock(spec=Request),
team_id="team-victim",
user_api_key_dict=unauthorized_caller,
)
code = getattr(exc.value, "status_code", None) or getattr(exc.value, "code", None)
assert int(code) == 403
patched_prisma.db.litellm_teamtable.update.assert_not_called()
@pytest.mark.asyncio
async def test_get_team_callbacks_rejects_unauthorized_caller(
patched_prisma, unauthorized_caller
):
# The endpoint catches generic Exception and re-wraps as ProxyException;
# an HTTPException raised by the access guard surfaces as a 403
# ProxyException — both shapes are acceptable failure modes, what
# matters is that the caller does NOT receive the team's callback data.
from litellm.proxy._types import ProxyException
with pytest.raises((HTTPException, ProxyException)) as exc:
await get_team_callbacks(
http_request=Mock(spec=Request),
team_id="team-victim",
user_api_key_dict=unauthorized_caller,
)
code = getattr(exc.value, "status_code", None) or getattr(exc.value, "code", None)
assert int(code) == 403
@pytest.mark.asyncio
async def test_proxy_admin_can_add_team_callbacks(patched_prisma):
"""
A proxy admin should pass the access guard and reach the DB write.
Sanity check that the guard didn't over-rotate.
"""
proxy_admin = UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN,
user_id="admin",
api_key="sk-admin",
)
data = AddTeamCallback(
callback_name="langfuse",
callback_type="success",
callback_vars={
"langfuse_public_key": "pk-admin",
"langfuse_secret_key": "sk-admin",
},
)
await add_team_callbacks(
data=data,
http_request=Mock(spec=Request),
team_id="team-victim",
user_api_key_dict=proxy_admin,
)
patched_prisma.db.litellm_teamtable.update.assert_awaited_once()
@pytest.mark.asyncio
async def test_team_admin_of_target_team_can_add_callbacks(patched_prisma):
"""
A team admin OF THE TARGET team should pass the access guard.
"""
# Override the victim row so the caller IS the team admin.
row = MagicMock()
row.model_dump.return_value = {
"team_id": "team-victim",
"team_alias": "victim-team",
"members_with_roles": [
{"role": "admin", "user_id": "team_admin_user"},
],
"organization_id": "org-victim",
}
row.metadata = {}
patched_prisma.get_data = AsyncMock(return_value=row)
team_admin = UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER,
user_id="team_admin_user",
api_key="sk-team-admin",
)
data = AddTeamCallback(
callback_name="langfuse",
callback_type="success",
callback_vars={
"langfuse_public_key": "pk-team",
"langfuse_secret_key": "sk-team",
},
)
await add_team_callbacks(
data=data,
http_request=Mock(spec=Request),
team_id="team-victim",
user_api_key_dict=team_admin,
)
patched_prisma.db.litellm_teamtable.update.assert_awaited_once()