fix(spend-logs): require every ambiguous request_id match to be owned

litellm_call_id is populated from the client-settable x-litellm-call-id
header, so a request_id lookup can match more than one row across
tenants. Authorizing on a single arbitrary match let an attacker reuse
a victim's request_id as their own call id and read the victim's spend
log row. Widen the ownership check to require every matching row to
belong to the caller, failing closed on any foreign match.

litellm/proxy/spend_tracking/spend_management_endpoints.py

@@ -243,11 +243,19 @@ def _request_id_or_call_id_clause(request_id: str) -> tuple[_RequestIdEquals, _L
     return (request_id_clause, call_id_clause)
 
 
-async def _find_spend_log_row(prisma_client: PrismaClient, request_id: str) -> _SpendLogOwnershipRow | None:
-    """Read the single spend log row identified by ``request_id`` or ``litellm_call_id``."""
-    return await _spend_logs_table(prisma_client).find_first(
+_SPEND_LOG_ID_LOOKUP_ROW_CAP: Final = 100
+
+
+async def _find_spend_log_rows(prisma_client: PrismaClient, request_id: str) -> Sequence[_SpendLogOwnershipRow]:
+    """Read every spend log row identified by ``request_id`` or ``litellm_call_id``.
+
+    ``litellm_call_id`` is populated from the client-settable ``x-litellm-call-id``
+    request header, so it is not guaranteed unique to one tenant: more than one row
+    can match. Callers must authorize every returned row, not just one of them.
+    """
+    return await _spend_logs_table(prisma_client).find_many(
         where={"OR": _request_id_or_call_id_clause(request_id)},
-        include=None,
+        take=_SPEND_LOG_ID_LOOKUP_ROW_CAP,
     )
 
 
@@ -4295,37 +4303,41 @@ def _can_user_view_spend_log(user_api_key_dict: UserAPIKeyAuth) -> bool:
     )
 
 
+async def _user_can_view_spend_log_row(
+    prisma_client: PrismaClient,
+    user_api_key_dict: UserAPIKeyAuth,
+    row: _SpendLogOwnershipRow,
+) -> bool:
+    if row.user is not None and row.user == user_api_key_dict.user_id:
+        return True
+    if row.team_id:
+        return await _can_team_member_view_log(
+            prisma_client=prisma_client,
+            user_api_key_dict=user_api_key_dict,
+            team_id=row.team_id,
+        )
+    return False
+
+
 async def _assert_user_can_view_request_id(
     prisma_client: PrismaClient,
     user_api_key_dict: UserAPIKeyAuth,
     request_id: str,
 ) -> None:
     """
-    Verify the requesting non-admin user is allowed to view this spend-log row.
-    Allowed when the log belongs to the user directly, or to one of their
-    permitted teams (admin or ``/spend/logs`` permission).
-    Raises HTTP 403 if not.
+    Verify the requesting non-admin user is allowed to view every spend-log row
+    identified by ``request_id`` or ``litellm_call_id``. The latter is client-settable,
+    so an id lookup can match more than one row across different tenants; access is
+    granted only when the user owns all of them directly or via a permitted team.
+    Raises HTTP 403 if any matching row is not the user's to view.
     """
-    row: Final = await _find_spend_log_row(prisma_client, request_id)
-    if row is None:
-        return
-
-    if row.user is not None and row.user == user_api_key_dict.user_id:
-        return
-
-    if row.team_id:
-        can_view: Final = await _can_team_member_view_log(
-            prisma_client=prisma_client,
-            user_api_key_dict=user_api_key_dict,
-            team_id=row.team_id,
-        )
-        if can_view:
-            return
-
-    raise HTTPException(
-        status_code=status.HTTP_403_FORBIDDEN,
-        detail={"error": f"Not authorized to view spend log for request_id={request_id}"},
-    )
+    rows: Final = await _find_spend_log_rows(prisma_client, request_id)
+    for row in rows:
+        if not await _user_can_view_spend_log_row(prisma_client, user_api_key_dict, row):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail={"error": f"Not authorized to view spend log for request_id={request_id}"},
+            )
 
 
 async def _get_permitted_team_ids_for_spend_logs(

tests/test_litellm/proxy/spend_tracking/test_spend_management_endpoints.py

@@ -413,8 +413,8 @@ async def test_assert_user_can_view_request_id_rejects_both_users_none():
         team_id = None
 
     class MockSpendLogs:
-        async def find_first(self, where=None, include=None):
-            return MockRow()
+        async def find_many(self, where=None, take=None):
+            return [MockRow()]
 
     class MockDB:
         def __init__(self):
@@ -432,6 +432,79 @@ async def test_assert_user_can_view_request_id_rejects_both_users_none():
     assert exc_info.value.status_code == 403
 
 
+@pytest.mark.asyncio
+async def test_assert_user_can_view_request_id_rejects_spoofed_call_id_collision():
+    """
+    litellm_call_id comes from the client-settable x-litellm-call-id header, so an
+    id lookup can match a row the caller owns AND a different tenant's row (the
+    caller set their own call id to the victim's request_id). Owning one of the
+    matching rows must not authorize the whole ambiguous id: every match has to
+    belong to the caller, or the whole lookup is rejected. Regression for the
+    cross-tenant spend-log read this OR clause introduced.
+    """
+
+    class _OwnRow:
+        user = "caller"
+        team_id = None
+
+    class _VictimRow:
+        user = "victim"
+        team_id = None
+
+    class MockSpendLogs:
+        async def find_many(self, where=None, take=None):
+            return [_OwnRow(), _VictimRow()]
+
+    class MockDB:
+        def __init__(self):
+            self.litellm_spendlogs = MockSpendLogs()
+
+    class MockPrisma:
+        def __init__(self):
+            self.db = MockDB()
+
+    auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="caller")
+    with pytest.raises(HTTPException) as exc_info:
+        await spend_management_endpoints._assert_user_can_view_request_id(
+            MockPrisma(), auth, "victim-request-id"
+        )
+    assert exc_info.value.status_code == 403
+
+
+@pytest.mark.asyncio
+async def test_assert_user_can_view_request_id_allows_when_every_match_is_owned():
+    """The same ambiguous id matching more than one row is fine when every match
+    belongs to the caller (e.g. two of the caller's own requests happen to share
+    a request_id/litellm_call_id pairing); only a foreign match should block it."""
+
+    class _OwnRowA:
+        user = "caller"
+        team_id = None
+
+    class _OwnRowB:
+        user = "caller"
+        team_id = None
+
+    class MockSpendLogs:
+        async def find_many(self, where=None, take=None):
+            return [_OwnRowA(), _OwnRowB()]
+
+    class MockDB:
+        def __init__(self):
+            self.litellm_spendlogs = MockSpendLogs()
+
+    class MockPrisma:
+        def __init__(self):
+            self.db = MockDB()
+
+    auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="caller")
+    result = await spend_management_endpoints._assert_user_can_view_request_id(
+        MockPrisma(), auth, "shared-request-id"
+    )
+
+    assert result is None
+
+
 def test_ui_view_request_response_forbids_non_admin_without_db(client, monkeypatch):
     """
     Without prisma, non-admins cannot be authorized to read request/response
@@ -2334,8 +2407,8 @@ async def test_ui_view_spend_logs_request_id_blocks_non_owner(client, monkeypatc
         team_id = None
 
     class _SpendLogs:
-        async def find_first(self, where=None, include=None):
-            return _ForeignRow()
+        async def find_many(self, where=None, take=None):
+            return [_ForeignRow()]
 
     class _DB:
         def __init__(self):
@@ -2400,10 +2473,10 @@ async def test_ui_view_spend_logs_request_id_owner_scoped_by_id_only(
         user = "user_1"
         team_id = "team1"
 
-    async def _find_first(where=None, include=None):
-        return _OwnedRow()
+    async def _find_many(where=None, take=None):
+        return [_OwnedRow()]
 
-    mock_prisma.db.find_first = _find_first
+    mock_prisma.db.find_many = _find_many
     monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
 
     # A 5-day window that EXCLUDES the 90-day-old log, as the dashboard sends.