From 140628063ce7737421823dc3a166e3c212d07208 Mon Sep 17 00:00:00 2001 From: user <70670632+stuxf@users.noreply.github.com> Date: Wed, 29 Apr 2026 21:39:39 +0000 Subject: [PATCH] fix(team): gate /team/{id}/callback endpoints behind _verify_team_access MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- .../team_callback_endpoints.py | 27 +++ .../test_team_callback_endpoints.py | 211 ++++++++++++++++++ 2 files changed, 238 insertions(+) create mode 100644 tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py diff --git a/litellm/proxy/management_endpoints/team_callback_endpoints.py b/litellm/proxy/management_endpoints/team_callback_endpoints.py index 4eec7c6b7c0..f28db70ef5d 100644 --- a/litellm/proxy/management_endpoints/team_callback_endpoints.py +++ b/litellm/proxy/management_endpoints/team_callback_endpoints.py @@ -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", {}) diff --git a/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py b/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py new file mode 100644 index 00000000000..745c0583bbd --- /dev/null +++ b/tests/test_litellm/proxy/management_endpoints/test_team_callback_endpoints.py @@ -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()