scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

Merge pull request #39068 from BerriAI/litellm_spend_log_request_id_call_id

Browse source 
fix(spend_logs): store litellm_call_id and match it in request_id lookups
This commit is contained in:
Mateo Wang 2026-09-12 21:12:57 -07:00 • committed by GitHub
commit a978ad2227
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
19 changed files with 1205 additions and 167 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm-proxy-extras/litellm_proxy_extras/migrations/20260831120000_add_litellm_call_id_spend_logs/migration.sql Normal file
View file

@ -0,0 +1,2 @@
-- AlterTable
ALTER TABLE "LiteLLM_SpendLogs" ADD COLUMN IF NOT EXISTS "litellm_call_id" TEXT;

12
litellm-proxy-extras/litellm_proxy_extras/migrations/20260831120001_spend_logs_litellm_call_id_index/migration.sql Normal file
View file

@ -0,0 +1,12 @@
-- CreateIndex (CONCURRENTLY)
--
-- Disclaimer:
-- - CREATE INDEX CONCURRENTLY cannot run inside a transaction. This migration must stay a
-- single statement so Prisma Migrate on PostgreSQL can apply it outside a transaction.
-- - Builds are slower and use more I/O than a blocking CREATE INDEX; if the build is
-- interrupted, Postgres may leave an INVALID index that must be dropped and recreated.
-- - Do not edit this file after it has been applied to any database: Prisma checksums
-- migrations; add a new migration instead.
-- - Requires PostgreSQL that supports CONCURRENTLY with IF NOT EXISTS (use a new migration
-- without IF NOT EXISTS if you must support older versions).
CREATE INDEX CONCURRENTLY IF NOT EXISTS "LiteLLM_SpendLogs_litellm_call_id_idx" ON "LiteLLM_SpendLogs"("litellm_call_id");

2
litellm-proxy-extras/litellm_proxy_extras/schema.prisma
View file

@ -659,12 +659,14 @@ model LiteLLM_SpendLogs {
mcp_namespaced_tool_name String?
agent_id String?
proxy_server_request Json? @default("{}")
litellm_call_id String?
created_at DateTime @default(now()) @map("created_at")
updated_at DateTime @default(now()) @updatedAt @map("updated_at")
@@index([startTime])
@@index([startTime, request_id])
@@index([end_user])
@@index([session_id])
@@index([litellm_call_id])
}
model LiteLLM_BudgetWindowSpend {

1
litellm/constants.py
View file

@ -143,6 +143,7 @@ DEFAULT_MCP_SEMANTIC_FILTER_SIMILARITY_THRESHOLD: Final = float(
os.getenv("DEFAULT_MCP_SEMANTIC_FILTER_SIMILARITY_THRESHOLD", 0.3)
)
MAX_MCP_SEMANTIC_FILTER_TOOLS_HEADER_LENGTH: Final = int(os.getenv("MAX_MCP_SEMANTIC_FILTER_TOOLS_HEADER_LENGTH", 150))
MAX_LITELLM_CALL_ID_LENGTH: Final = 256
MAX_GUARDRAIL_SCAN_METADATA_HEADER_LENGTH: Final = 2048
DEFAULT_AUTO_ROUTER_MAX_INPUT_CHARS: Final = 2000

1
litellm/proxy/_types.py
View file

@ -3882,6 +3882,7 @@ class SpendLogsPayload(TypedDict):
session_id: str | None
request_duration_ms: int | None
status: Literal["success", "failure"]
litellm_call_id: ReadOnly[str | None]
class SpanAttributes(str, enum.Enum):

9
litellm/proxy/common_request_processing.py
View file

@ -24,6 +24,7 @@ from litellm.constants import (
DEFAULT_MAX_RECURSE_DEPTH,
LITELLM_DETAILED_TIMING,
LITELLM_HTTP_STATUS_CLIENT_DISCONNECTED,
MAX_LITELLM_CALL_ID_LENGTH,
MAX_PAYLOAD_SIZE_FOR_DEBUG_LOG,
NON_INFERENCE_CALL_TYPES,
RETURN_RAW_MODEL_NAME_METADATA_KEY,
@ -217,6 +218,12 @@ def _withheld_provider_output(response: object) -> bool:
return getattr(response, "has_buffered_provider_output", False) is True
def resolve_litellm_call_id(client_call_id: str | None) -> str:
if client_call_id is not None and 0 < len(client_call_id) <= MAX_LITELLM_CALL_ID_LENGTH:
return client_call_id
return str(uuid.uuid4())
def _should_return_raw_model_name(request_data: dict[str, object]) -> bool:
return any(
isinstance(metadata, dict) and metadata.get(RETURN_RAW_MODEL_NAME_METADATA_KEY) is True
@ -1938,7 +1945,7 @@ class ProxyBaseLLMRequestProcessing:
if alias_target is not None:
self.data["model"] = alias_target
self.data["litellm_call_id"] = request.headers.get("x-litellm-call-id", str(uuid.uuid4()))
self.data["litellm_call_id"] = resolve_litellm_call_id(request.headers.get("x-litellm-call-id"))
DDSpanTagger.tag_call_id(self.data.get("litellm_call_id"))
DDSpanTagger.tag_request(
user_api_key_dict=user_api_key_dict,

2
litellm/proxy/schema.prisma
View file

@ -659,12 +659,14 @@ model LiteLLM_SpendLogs {
mcp_namespaced_tool_name String?
agent_id String?
proxy_server_request Json? @default("{}")
litellm_call_id String?
created_at DateTime @default(now()) @map("created_at")
updated_at DateTime @default(now()) @updatedAt @map("updated_at")
@@index([startTime])
@@index([startTime, request_id])
@@index([end_user])
@@index([session_id])
@@index([litellm_call_id])
}
model LiteLLM_BudgetWindowSpend {

327
litellm/proxy/spend_tracking/spend_management_endpoints.py
View file

@ -3,6 +3,7 @@ import collections
import json
import os
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import date, datetime, timedelta, timezone
from itertools import groupby
from types import MappingProxyType
@ -17,6 +18,7 @@ from typing import (
TypeAlias,
TypedDict,
TypeVar,
cast, # noqa: TID251 # custom-logger and cold-storage payloads are untyped JSON
)
import fastapi
@ -75,6 +77,7 @@ _SPEND_LOG_LIST_COLUMNS: Final = """
cache_hit, cache_key, request_tags, team_id,
organization_id, end_user, requester_ip_address,
session_id, status, mcp_namespaced_tool_name, agent_id,
litellm_call_id,
COALESCE(request_duration_ms,
(EXTRACT(EPOCH FROM ("endTime" - "startTime")) * 1000)::INTEGER) AS request_duration_ms
"""
@ -91,9 +94,9 @@ class _SupportsModelDump(Protocol):
def model_dump(self) -> Mapping[str, object]: ...
class _SpendLogOwnershipRow(Protocol):
user: str | None
team_id: str | None
class _SpendLogOwnerRow(TypedDict):
user: ReadOnly[str | None]
team_id: ReadOnly[str | None]
class _ActivityRow(TypedDict):
@ -330,12 +333,37 @@ async def _find_spend_logs(
return rows
async def _find_spend_log_row(prisma_client: PrismaClient, request_id: str) -> _SpendLogOwnershipRow | None:
"""Read the single spend log row identified by ``request_id``."""
return await _spend_logs_table(prisma_client).find_unique(
where={"request_id": request_id},
include=None,
)
class _RequestIdEquals(TypedDict):
request_id: ReadOnly[str]
class _LitellmCallIdEquals(TypedDict):
litellm_call_id: ReadOnly[str]
def _request_id_or_call_id_clause(request_id: str) -> tuple[_RequestIdEquals, _LitellmCallIdEquals]:
request_id_clause: Final[_RequestIdEquals] = {"request_id": request_id}
call_id_clause: Final[_LitellmCallIdEquals] = {"litellm_call_id": request_id}
return (request_id_clause, call_id_clause)
async def _find_spend_log_owners(prisma_client: PrismaClient, request_id: str) -> Sequence[_SpendLogOwnerRow]:
"""Read the distinct ``(user, team_id)`` owner pairs across 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: any number of rows
can match one id. The read is uncapped because a flood of another tenant's rows
carrying the caller's id could otherwise push the caller's own owner pair past a
row-sample cap and lock them out of their own lookup.
"""
sql_query: Final = """
SELECT DISTINCT "user", team_id
FROM "LiteLLM_SpendLogs"
WHERE request_id = $1 OR litellm_call_id = $1
"""
owners: Final[Sequence[_SpendLogOwnerRow] | None] = await _query_raw_or_none(prisma_client, sql_query, request_id)
return owners if owners is not None else ()
async def _count_spend_logs(prisma_client: PrismaClient, where: Mapping[str, object]) -> int:
@ -2599,10 +2627,11 @@ async def ui_view_spend_logs(
if max_spend is not None:
where_conditions["spend"]["lte"] = max_spend
# A request_id lookup drops the date window, so a non-admin could otherwise
# reach any single row by id; require they own it, mirroring the detail
# endpoint. That ownership check fully authorizes the one row, so the
# general scoping below is skipped for id lookups. Scoped to the UI route
# so the public v2 contract is unchanged.
# reach any single row by id; require they own one of the matches, mirroring
# the detail endpoint, and keep the general scoping below so a colliding
# foreign row is filtered out rather than served or allowed to deny the
# caller their own row. Scoped to the UI route so the public v2 contract is
# unchanged.
if request_id is not None and not is_v2 and not is_admin_view:
await _assert_user_can_view_request_id(
prisma_client=prisma_client,
@ -2610,10 +2639,9 @@ async def ui_view_spend_logs(
request_id=request_id,
)
user_scope_applies: Final = (
not is_request_id_lookup
and not is_admin_view
not is_admin_view
and team_id is None
and _can_user_view_spend_log(user_api_key_dict=user_api_key_dict)
and (is_request_id_lookup or _can_user_view_spend_log(user_api_key_dict=user_api_key_dict))
)
permitted_team_ids: Final = (
await _get_permitted_team_ids_for_spend_logs_or_empty(
@ -2626,7 +2654,7 @@ async def ui_view_spend_logs(
explicit_user_requires_caller_scope: Final = (
user_scope_applies and not permitted_team_ids and user_id is not None
)
if not is_request_id_lookup and not is_admin_view:
if not is_admin_view:
if team_id is not None:
can_view_team: Final = await _can_team_member_view_log(
prisma_client=prisma_client,
@ -2696,7 +2724,6 @@ async def ui_view_spend_logs(
("team_id", "team_id"),
('"user"', "user"),
("api_key", "api_key"),
("request_id", "request_id"),
("model", "model"),
("model_id", "model_id"),
("model_group", "model_group"),
@ -2708,6 +2735,13 @@ async def ui_view_spend_logs(
sql_params.append(val)
p += 1
request_id_filter: Final = where_conditions.get("request_id")
exact_request_id_first: Final = f"(request_id = ${p}) DESC, " if isinstance(request_id_filter, str) else ""
if isinstance(request_id_filter, str):
sql_conditions.append(f"(request_id = ${p} OR litellm_call_id = ${p})")
sql_params.append(request_id_filter)
p += 1
# Multi-team OR filter: (user = $X OR team_id = ANY($Y))
if permitted_team_ids:
or_clause: Final = f'("user" = ${p} OR team_id = ANY(${p + 1}::text[]))'
@ -2837,7 +2871,7 @@ async def ui_view_spend_logs(
WHERE {joined_conditions}
ORDER BY {_SESSION_GROUP_KEY_SQL}, call_type IN {_MCP_CALL_TYPES_SQL}, "startTime" DESC
) AS session_representatives
ORDER BY {_order_expr} {_sql_dir}{_nulls_clause}, request_id
ORDER BY {exact_request_id_first}{_order_expr} {_sql_dir}{_nulls_clause}, request_id
LIMIT ${p} OFFSET ${p + 1}
"""
if session_grouping
@ -2846,7 +2880,7 @@ async def ui_view_spend_logs(
{_SPEND_LOG_LIST_COLUMNS}
FROM "LiteLLM_SpendLogs"
WHERE {joined_conditions}
ORDER BY {_order_expr} {_sql_dir}{_nulls_clause}
ORDER BY {exact_request_id_first}{_order_expr} {_sql_dir}{_nulls_clause}
LIMIT ${p} OFFSET ${p + 1}
"""
)
@ -2854,6 +2888,14 @@ async def ui_view_spend_logs(
data: Final = await prisma_client.db.query_raw(sql_query, *sql_params)
if request_id is not None and not is_v2 and not is_admin_view:
await _assert_user_owns_fetched_spend_rows(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
rows=data,
request_id=request_id,
)
_hydrate_spend_log_metadata(data)
# Calculate total pages
@ -3104,7 +3146,7 @@ def _hydrate_spend_log_metadata(rows: Sequence[Mapping[str, object]]) -> None:
def _cold_storage_object_key_from_metadata(
metadata: str | dict | None,
metadata: str | Mapping[str, object] | None,
) -> str | None:
if isinstance(metadata, str):
try:
@ -3209,7 +3251,8 @@ async def ui_view_request_response_for_request_id(
"""
from litellm.proxy.proxy_server import prisma_client
if not _is_admin_view_safe(user_api_key_dict=user_api_key_dict):
caller_is_admin: Final = _is_admin_view_safe(user_api_key_dict=user_api_key_dict)
if not caller_is_admin:
if prisma_client is None:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
@ -3234,38 +3277,45 @@ async def ui_view_request_response_for_request_id(
if end_date is not None:
end_date_obj = datetime.strptime(end_date, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone.utc)
spend_log_row: Final = (
None
if prisma_client is None
else await _resolve_spend_log_payload_row(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
request_id=request_id,
caller_is_admin=caller_is_admin,
)
)
stored_request_id: Final = _stored_request_id(spend_log_row, request_id)
for custom_logger in custom_loggers:
payload = await custom_logger.get_request_response_payload(
request_id=request_id,
request_id=stored_request_id,
start_time_utc=start_date_obj,
end_time_utc=end_date_obj,
)
if payload is not None:
if not caller_is_admin and prisma_client is not None:
await _assert_user_owns_cold_storage_payload(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
payload=cast(Mapping[str, object], payload), # cast-ok: custom-logger payload is untyped
request_id=request_id,
)
return payload
if spend_log_row is None:
return None
# Fallback: the list endpoint omits the heavy columns for performance, so
# serve them here. When prompts were offloaded to cold storage the DB holds
# only placeholders, so _resolve_request_response_payload fetches the real
# payload from the configured cold storage backend by object key.
if prisma_client is not None:
from litellm.proxy.spend_tracking.cold_storage_handler import (
ColdStorageHandler,
)
from litellm.proxy.spend_tracking.cold_storage_handler import ColdStorageHandler
sql_query: Final = """
SELECT messages, response, proxy_server_request, metadata
FROM "LiteLLM_SpendLogs"
WHERE request_id = $1
LIMIT 1
"""
db_result: Final[Sequence[Mapping[str, object]] | None] = await _query_raw_or_none(
prisma_client, sql_query, request_id
)
if db_result and len(db_result) > 0:
resolved = await _resolve_request_response_payload(db_result[0], cold_storage_handler=ColdStorageHandler())
return resolved._asdict()
return None
resolved: Final = await _resolve_request_response_payload(spend_log_row, cold_storage_handler=ColdStorageHandler())
return resolved._asdict()
@router.get(
@ -3391,7 +3441,7 @@ async def view_spend_logs(
if api_key is not None and isinstance(api_key, str):
filter_query["api_key"] = summary_api_key
if request_id is not None and isinstance(request_id, str):
filter_query["request_id"] = request_id
filter_query["OR"] = _request_id_or_call_id_clause(request_id)
if user_id is not None and isinstance(user_id, str):
filter_query["user"] = user_id
@ -3439,7 +3489,7 @@ async def view_spend_logs(
return [*summary_items, *padding]
else:
scoped_filter: Final[dict[str, str]] = {}
scoped_filter: Final[dict[str, object]] = {}
if api_key is not None and isinstance(api_key, str):
if api_key.startswith("sk-"):
hashed_token = prisma_client.hash_token(token=api_key)
@ -3447,7 +3497,7 @@ async def view_spend_logs(
hashed_token = api_key
scoped_filter["api_key"] = hashed_token
if request_id is not None and isinstance(request_id, str):
scoped_filter["request_id"] = request_id
scoped_filter["OR"] = _request_id_or_call_id_clause(request_id)
if user_id is not None and isinstance(user_id, str):
scoped_filter["user"] = user_id
@ -4676,39 +4726,190 @@ def _can_user_view_spend_log(user_api_key_dict: UserAPIKeyAuth) -> bool:
)
async def _user_can_view_spend_log_owner(
prisma_client: PrismaClient,
user_api_key_dict: UserAPIKeyAuth,
owner_user: str | None,
owner_team_id: str | None,
) -> bool:
if owner_user is not None and owner_user == user_api_key_dict.user_id:
return True
if owner_team_id:
return await _can_team_member_view_log(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
team_id=owner_team_id,
)
return False
def _spend_log_forbidden(request_id: str) -> HTTPException:
return HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={"error": f"Not authorized to view spend log for request_id={request_id}"},
)
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, including when no spend-log row exists for the
request_id (e.g. it was pruned by retention), so a missing row can't be
used to read a payload out of cold storage via the detail endpoint.
Verify the requesting non-admin user is allowed to view at least one spend-log
row identified by ``request_id`` or ``litellm_call_id``. The latter is
client-settable, so an id lookup can match rows across different tenants; the
data queries scope a non-admin's results to rows they own directly or via a
permitted team, so a colliding foreign row can neither be served nor deny the
caller their own. Raises HTTP 403 when none of the matching rows is theirs to
view, including when no row exists at all (e.g. it was pruned by retention),
so a missing row can't be used to read a payload out of cold storage via the
detail endpoint.
"""
row: Final = await _find_spend_log_row(prisma_client, request_id)
owners: Final = await _find_spend_log_owners(prisma_client, request_id)
for owner in owners:
if await _user_can_view_spend_log_owner(prisma_client, user_api_key_dict, owner["user"], owner["team_id"]):
return
raise _spend_log_forbidden(request_id)
if row is not None and row.user is not None and row.user == user_api_key_dict.user_id:
return
if row is not None and row.team_id:
can_view: Final = await _can_team_member_view_log(
@dataclass(frozen=True, slots=True)
class _SpendLogViewer:
user_id: str | None
team_ids: tuple[str, ...]
async def _spend_log_viewer(prisma_client: PrismaClient, user_api_key_dict: UserAPIKeyAuth) -> _SpendLogViewer:
return _SpendLogViewer(
user_id=user_api_key_dict.user_id,
team_ids=await _get_permitted_team_ids_for_spend_logs_or_empty(
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}"},
),
)
def _viewer_scope_clause(viewer: _SpendLogViewer | None) -> tuple[str, tuple[object, ...]]:
match viewer:
case None:
return ("", ())
case _SpendLogViewer(user_id=user_id, team_ids=()):
return (' AND "user" = $2', (user_id,))
case _SpendLogViewer(user_id=user_id, team_ids=team_ids):
return (' AND ("user" = $2 OR team_id = ANY($3::text[]))', (user_id, team_ids))
def _spend_log_payload_query(request_id: str, viewer: _SpendLogViewer | None) -> tuple[str, tuple[object, ...]]:
"""
Fetch the one row an id lookup resolves to, preferring the exact ``request_id``
match over rows that merely carry the id as their client-set ``litellm_call_id``.
A non-admin viewer only ever gets rows they own or rows of a team they may view.
"""
scope, scope_params = _viewer_scope_clause(viewer)
return (
f"""
SELECT request_id, messages, response, proxy_server_request, metadata, "user", team_id
FROM "LiteLLM_SpendLogs"
WHERE (request_id = $1 OR litellm_call_id = $1){scope}
ORDER BY (request_id = $1) DESC
LIMIT 1
""",
(request_id, *scope_params),
)
async def _resolve_spend_log_payload_row(
prisma_client: PrismaClient,
user_api_key_dict: UserAPIKeyAuth,
request_id: str,
caller_is_admin: bool,
) -> Mapping[str, object] | None:
"""
Resolve an id lookup to the caller's own spend-log row before any payload
store is consulted. Cold storage is keyed by the provider ``request_id``, so
asking it for the raw lookup id could hand back another tenant's payload when
that id is only the caller's ``litellm_call_id``; the row's stored
``request_id`` is the key that names the caller's own request.
"""
viewer: Final = None if caller_is_admin else await _spend_log_viewer(prisma_client, user_api_key_dict)
sql_query, sql_params = _spend_log_payload_query(request_id, viewer)
rows: Final[Sequence[Mapping[str, object]] | None] = await _query_raw_or_none(prisma_client, sql_query, *sql_params)
if not rows:
return None
if not caller_is_admin:
await _assert_user_owns_fetched_spend_rows(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
rows=rows,
request_id=request_id,
)
return rows[0]
def _stored_request_id(row: Mapping[str, object] | None, lookup_id: str) -> str:
stored: Final = None if row is None else row.get("request_id")
return stored if isinstance(stored, str) else lookup_id
def _fetched_row_owner(row: Mapping[str, object]) -> tuple[str | None, str | None]:
user: Final = row.get("user")
team_id: Final = row.get("team_id")
return (
user if isinstance(user, str) else None,
team_id if isinstance(team_id, str) else None,
)
async def _assert_user_owns_fetched_spend_rows(
prisma_client: PrismaClient,
user_api_key_dict: UserAPIKeyAuth,
rows: Sequence[Mapping[str, object]],
request_id: str,
) -> None:
"""
Re-verify ownership on the rows an id lookup actually fetched.
``_assert_user_can_view_request_id`` and the data query read the table at
different moments, so a foreign row inserted between them could otherwise be
returned even though the pre-check passed. Checking the fetched rows
themselves means no interleaving can return another tenant's row.
"""
for user, team_id in frozenset(_fetched_row_owner(row) for row in rows):
if not await _user_can_view_spend_log_owner(prisma_client, user_api_key_dict, user, team_id):
raise _spend_log_forbidden(request_id)
def _cold_storage_payload_owner(payload: Mapping[str, object]) -> tuple[str | None, str | None]:
metadata: Final = payload.get("metadata")
if not isinstance(metadata, Mapping):
return (None, None)
owner: Final = cast(Mapping[str, object], metadata) # cast-ok: cold-storage JSON is untyped
user: Final = owner.get("user_api_key_user_id")
team_id: Final = owner.get("user_api_key_team_id")
return (
user if isinstance(user, str) else None,
team_id if isinstance(team_id, str) else None,
)
async def _assert_user_owns_cold_storage_payload(
prisma_client: PrismaClient,
user_api_key_dict: UserAPIKeyAuth,
payload: Mapping[str, object],
request_id: str,
) -> None:
"""
Authorize a cold-storage payload against the owner recorded inside it.
The custom logger reads the payload straight from cold storage, written
independently of the spend-log table and able to outlive its row, so a
request_id lookup could otherwise hand back another tenant's stored payload
when no row exists for the pre-check to catch. Verifying the payload's own
owner closes that gap, and a payload that records no owner fails closed.
"""
owner_user, owner_team_id = _cold_storage_payload_owner(payload)
if not await _user_can_view_spend_log_owner(prisma_client, user_api_key_dict, owner_user, owner_team_id):
raise _spend_log_forbidden(request_id)
async def _get_permitted_team_ids_for_spend_logs(
prisma_client: PrismaClient,
user_api_key_dict: UserAPIKeyAuth,

1
litellm/proxy/spend_tracking/spend_tracking_utils.py
View file

@ -621,6 +621,7 @@ def get_logging_payload(kwargs, response_obj, start_time, end_time) -> SpendLogs
status=_get_status_for_spend_log(
metadata=metadata,
),
litellm_call_id=litellm_call_id,
)
verbose_proxy_logger.debug(

2
schema.prisma
View file

@ -659,12 +659,14 @@ model LiteLLM_SpendLogs {
mcp_namespaced_tool_name String?
agent_id String?
proxy_server_request Json? @default("{}")
litellm_call_id String?
created_at DateTime @default(now()) @map("created_at")
updated_at DateTime @default(now()) @updatedAt @map("updated_at")
@@index([startTime])
@@index([startTime, request_id])
@@index([end_user])
@@index([session_id])
@@index([litellm_call_id])
}
model LiteLLM_BudgetWindowSpend {

1
tests/logging_callback_tests/test_gcs_pub_sub.py
View file

@ -28,6 +28,7 @@ verbose_logger.setLevel(logging.DEBUG)
ignored_keys = [
"request_id",
"litellm_call_id",
"metadata.litellm_call_id",
"session_id",
"startTime",

814
tests/test_litellm/proxy/spend_tracking/test_spend_management_endpoints.py
View file

@ -124,7 +124,10 @@ def _reconstruct_ui_where_from_sql(sql_query, params):
sess = re.fullmatch(r"session_id LIKE \$(\d+)", cond)
status = re.fullmatch(r"status = \$(\d+)", cond)
api_key_not_in = re.fullmatch(r"api_key NOT IN \(\$(\d+), \$(\d+)\)", cond)
if gte:
req_or_call = re.fullmatch(r"\(request_id = \$(\d+) OR litellm_call_id = \$\1\)", cond)
if req_or_call:
where["request_id_or_call_id"] = params[int(req_or_call.group(1)) - 1]
elif gte:
date_bounds["gte"] = _iso(params[int(gte.group(1)) - 1])
elif lte:
date_bounds["lte"] = _iso(params[int(lte.group(1)) - 1])
@ -213,6 +216,8 @@ def make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn, team_lookup_fn=No
query_observer(sql_query, params)
if "mcp_tool_call_count" in sql_query:
return []
if 'SELECT DISTINCT "user", team_id' in sql_query:
return _emulate_spend_log_owner_lookup(mock_spend_logs, sql_query, params)
filtered = filter_fn(_reconstruct_ui_where_from_sql(sql_query, params))
total = len(filtered)
if "COUNT(*)" in sql_query:
@ -220,7 +225,13 @@ def make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn, team_lookup_fn=No
return [{"total_count": min(total, cap_plus_one)}]
page_size = params[-2] if len(params) >= 2 else 50
skip = params[-1] if len(params) >= 1 else 0
return [row for row in filtered[skip : skip + page_size]]
exact_first = re.search(r"ORDER BY \(request_id = \$(\d+)\) DESC", sql_query)
ordered = (
sorted(filtered, key=lambda row: row["request_id"] == params[int(exact_first.group(1)) - 1], reverse=True)
if exact_first
else filtered
)
return [row for row in ordered[skip : skip + page_size]]
class MockPrismaClient:
def __init__(self):
@ -424,37 +435,173 @@ def test_can_user_view_spend_log_false_for_other_roles():
assert spend_management_endpoints._can_user_view_spend_log(auth) is False
def _emulate_spend_log_owner_lookup(rows, sql_query, params):
"""Emulate the ownership lookup SQL over an in-memory spend-log corpus,
honoring DISTINCT and any literal LIMIT the query carries so a capped or
non-distinct query produces the truncated result it would in Postgres."""
lookup_id = params[0]
matches = [
{"user": row.get("user"), "team_id": row.get("team_id")}
for row in rows
if lookup_id in (row.get("request_id"), row.get("litellm_call_id"))
]
if "DISTINCT" in sql_query:
deduped = []
for match in matches:
if match not in deduped:
deduped.append(match)
matches = deduped
limit = re.search(r"LIMIT\s+(\d+)", sql_query, re.IGNORECASE)
if limit is not None:
matches = matches[: int(limit.group(1))]
return matches
def _make_owner_lookup_prisma(rows):
class MockDB:
async def query_raw(self, sql_query, *params):
return _emulate_spend_log_owner_lookup(rows, sql_query, params)
class MockPrisma:
def __init__(self):
self.db = MockDB()
return MockPrisma()
@pytest.mark.asyncio
async def test_assert_user_can_view_request_id_rejects_both_users_none():
"""
API keys with user_id=None must not be treated as owning a log whose user
field is None (avoid None == None bypass).
"""
class MockRow:
user = None
team_id = None
class MockSpendLogs:
async def find_unique(self, where, include=None):
return MockRow()
class MockDB:
def __init__(self):
self.litellm_spendlogs = MockSpendLogs()
class MockPrisma:
def __init__(self):
self.db = MockDB()
prisma = _make_owner_lookup_prisma(
[{"request_id": "req-none-user", "litellm_call_id": None, "user": None, "team_id": None}]
)
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id=None)
with pytest.raises(HTTPException) as exc_info:
await spend_management_endpoints._assert_user_can_view_request_id(
MockPrisma(), auth, "req-none-user"
prisma, auth, "req-none-user"
)
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_assert_user_can_view_request_id_rejects_when_no_match_is_owned():
"""An id whose every matching row belongs to other tenants is refused outright,
so the relaxed date window of an id lookup cannot reach a foreign row."""
prisma = _make_owner_lookup_prisma(
[
{"request_id": "foreign-request", "litellm_call_id": "shared-id", "user": "tenant_a", "team_id": None},
{"request_id": "shared-id", "litellm_call_id": "other-call-id", "user": "tenant_b", "team_id": None},
]
)
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(prisma, auth, "shared-id")
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_assert_user_can_view_request_id_allows_owner_despite_foreign_collision():
"""
litellm_call_id comes from the client-settable x-litellm-call-id header, so
another tenant can mint a row whose call id equals the caller's request_id.
That collision must not lock the caller out of their own row: the pre-check
passes once one match is theirs, and the scoped data queries keep the foreign
row out of the result. Regression for the every-match-must-be-owned rule that
let any tenant deny another's lookup by reusing their id.
"""
prisma = _make_owner_lookup_prisma(
[
{
"request_id": "attacker-own-request",
"litellm_call_id": "victim-request-id",
"user": "attacker",
"team_id": None,
},
{
"request_id": "victim-request-id",
"litellm_call_id": "victim-call-id",
"user": "victim",
"team_id": None,
},
]
)
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="victim")
result = await spend_management_endpoints._assert_user_can_view_request_id(prisma, auth, "victim-request-id")
assert result is None
@pytest.mark.asyncio
async def test_assert_user_can_view_request_id_finds_owner_past_any_row_cap():
"""
An attacker can mint hundreds of rows carrying the victim's request_id as
their litellm_call_id, so a capped or sampled ownership read could exhaust
its cap on attacker-owned rows and never see the victim's own row, locking
the victim out of their lookup. The ownership read must consider every
matching row's owner no matter how many rows match. Regression for the
find_many(take=100) sample the first fix used.
"""
rows = [
{
"request_id": f"attacker-request-{i}",
"litellm_call_id": "victim-request-id",
"user": "attacker",
"team_id": None,
}
for i in range(150)
]
rows.append(
{
"request_id": "victim-request-id",
"litellm_call_id": "victim-call-id",
"user": "victim",
"team_id": None,
}
)
prisma = _make_owner_lookup_prisma(rows)
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="victim")
result = await spend_management_endpoints._assert_user_can_view_request_id(prisma, auth, "victim-request-id")
assert result is None
@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."""
prisma = _make_owner_lookup_prisma(
[
{
"request_id": "shared-request-id",
"litellm_call_id": "caller-call-a",
"user": "caller",
"team_id": None,
},
{
"request_id": "caller-request-b",
"litellm_call_id": "shared-request-id",
"user": "caller",
"team_id": None,
},
]
)
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="caller")
result = await spend_management_endpoints._assert_user_can_view_request_id(
prisma, auth, "shared-request-id"
)
assert result is None
@pytest.mark.asyncio
async def test_assert_user_can_view_request_id_rejects_missing_row():
"""
@ -462,24 +609,11 @@ async def test_assert_user_can_view_request_id_rejects_missing_row():
authorize reading the payload from cold storage; a missing row is not
the same as an owned row.
"""
class MockSpendLogs:
async def find_unique(self, where, include=None):
return None
class MockDB:
def __init__(self):
self.litellm_spendlogs = MockSpendLogs()
class MockPrisma:
def __init__(self):
self.db = MockDB()
prisma = _make_owner_lookup_prisma([])
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
with pytest.raises(HTTPException) as exc_info:
await spend_management_endpoints._assert_user_can_view_request_id(
MockPrisma(), auth, "req-missing-row"
)
await spend_management_endpoints._assert_user_can_view_request_id(prisma, auth, "req-missing-row")
assert exc_info.value.status_code == 403
@ -507,6 +641,7 @@ def test_ui_view_request_response_forbids_non_admin_without_db(client, monkeypat
ignored_keys = [
"request_id",
"litellm_call_id",
"metadata.litellm_call_id",
"session_id",
"startTime",
@ -2216,7 +2351,10 @@ async def test_ui_view_spend_logs_request_id_lookup_ignores_date_window(
def filter_fn(where):
captured["where"] = where
rows = _filter_logs_by_date_range(mock_spend_logs, where)
if where.get("request_id"):
rid_either = where.get("request_id_or_call_id")
if rid_either:
rows = [r for r in rows if rid_either in (r["request_id"], r.get("litellm_call_id"))]
elif where.get("request_id"):
rows = [r for r in rows if r["request_id"] == where["request_id"]]
return rows
@ -2246,9 +2384,82 @@ async def test_ui_view_spend_logs_request_id_lookup_ignores_date_window(
data = response.json()
assert data["total"] == 1
assert data["data"][0]["request_id"] == "req-old"
# Query dropped the time window and scoped solely by the primary key.
# Query dropped the time window and scoped solely by the id lookup.
assert "startTime" not in captured["where"]
assert captured["where"]["request_id"] == "req-old"
assert captured["where"]["request_id_or_call_id"] == "req-old"
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_spend_logs_request_id_lookup_matches_litellm_call_id(
client, monkeypatch
):
"""
LIT-6302: success rows are keyed by the upstream provider response id, so a
lookup with the x-litellm-call-id response header value found nothing. The id
lookup now matches request_id OR litellm_call_id, resolving the header value.
"""
today = datetime.datetime.now(timezone.utc)
mock_spend_logs = [
{
"id": "log_provider_keyed",
"request_id": "chatcmpl-9ZKMURhVYSi9D6r6PJ9vLcayIK0Vm",
"litellm_call_id": "b980eea9-5cd9-4099-93cd-8291e46c76fd",
"api_key": "sk-test-key",
"user": "test_user_1",
"team_id": "team1",
"spend": 0.05,
"startTime": today.isoformat(),
"model": "gpt-4",
},
{
"id": "log_other",
"request_id": "chatcmpl-other",
"litellm_call_id": "11111111-2222-3333-4444-555555555555",
"api_key": "sk-test-key",
"user": "test_user_1",
"team_id": "team1",
"spend": 0.01,
"startTime": today.isoformat(),
"model": "gpt-4",
},
]
def filter_fn(where):
rid_either = where.get("request_id_or_call_id")
if rid_either:
return [
r
for r in mock_spend_logs
if rid_either in (r["request_id"], r.get("litellm_call_id"))
]
if where.get("request_id"):
return [
r for r in mock_spend_logs if r["request_id"] == where["request_id"]
]
return list(mock_spend_logs)
monkeypatch.setattr(
"litellm.proxy.proxy_server.prisma_client",
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn),
)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN
)
try:
response = client.get(
"/spend/logs/ui",
params={"request_id": "b980eea9-5cd9-4099-93cd-8291e46c76fd"},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 1
assert (
data["data"][0]["request_id"] == "chatcmpl-9ZKMURhVYSi9D6r6PJ9vLcayIK0Vm"
)
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@ -2303,23 +2514,18 @@ async def test_ui_view_spend_logs_request_id_blocks_non_owner(client, monkeypatc
"""A non-admin looking up a request_id they do not own is rejected (403), so
the relaxed date window cannot read another tenant's log by id."""
class _ForeignRow:
user = "other_user"
team_id = None
prisma = _make_owner_lookup_prisma(
[
{
"request_id": "foreign-req",
"litellm_call_id": None,
"user": "other_user",
"team_id": None,
}
]
)
class _SpendLogs:
async def find_unique(self, where, include=None):
return _ForeignRow()
class _DB:
def __init__(self):
self.litellm_spendlogs = _SpendLogs()
class _Prisma:
def __init__(self):
self.db = _DB()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", _Prisma())
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1"
)
@ -2335,13 +2541,437 @@ async def test_ui_view_spend_logs_request_id_blocks_non_owner(client, monkeypatc
@pytest.mark.asyncio
async def test_ui_view_spend_logs_request_id_owner_scoped_by_id_only(
async def test_ui_view_spend_logs_request_id_collision_serves_only_callers_rows(client, monkeypatch):
"""Two tenants share one id: the attacker minted a row whose client-set
litellm_call_id equals the victim's request_id. Each side's lookup of that id
returns only their own row, so the collision neither leaks the other tenant's
row nor denies the victim theirs (Veria: identifier collision could deny access)."""
now_iso = datetime.datetime.now(timezone.utc).isoformat()
corpus = [
{
"id": "log_attacker",
"request_id": "attacker-req",
"litellm_call_id": "victim-req",
"api_key": "sk-attacker-key",
"user": "attacker_user",
"team_id": None,
"spend": 0.05,
"startTime": now_iso,
"model": "gpt-4",
},
{
"id": "log_victim",
"request_id": "victim-req",
"litellm_call_id": "victim-call-id",
"api_key": "sk-victim-key",
"user": "victim_user",
"team_id": None,
"spend": 0.07,
"startTime": now_iso,
"model": "gpt-4",
},
]
def filter_fn(where):
rid_either = where.get("request_id_or_call_id")
rows = [r for r in corpus if rid_either in (r["request_id"], r["litellm_call_id"])]
return [r for r in rows if where.get("user") is None or r["user"] == where["user"]]
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", make_ui_spend_logs_mock_prisma(corpus, filter_fn))
try:
for caller, own_request_id, other in (
("victim_user", "victim-req", "attacker_user"),
("attacker_user", "attacker-req", "victim_user"),
):
app.dependency_overrides[ps.user_api_key_auth] = lambda caller=caller: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id=caller
)
response = client.get(
"/spend/logs/ui",
params={"request_id": "victim-req"},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 200, response.text
data = response.json()
assert data["total"] == 1
assert data["data"][0]["request_id"] == own_request_id
assert other not in response.text
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_spend_logs_id_lookup_scopes_every_non_admin_role(client, monkeypatch):
"""An org admin reaches /spend/logs/ui without the internal-user row scope. An
id lookup still fetches only rows they own, so another tenant's row carrying
that id as its client-set litellm_call_id neither leaks nor turns the
org admin's own lookup into a 403 (Bugbot: non-internal id lookup 403s on collision)."""
now_iso = datetime.datetime.now(timezone.utc).isoformat()
corpus = [
{
"id": "log_attacker",
"request_id": "attacker-req",
"litellm_call_id": "victim-req",
"api_key": "sk-attacker-key",
"user": "attacker_user",
"team_id": None,
"spend": 0.05,
"startTime": now_iso,
"model": "gpt-4",
},
{
"id": "log_victim",
"request_id": "victim-req",
"litellm_call_id": "victim-call-id",
"api_key": "sk-victim-key",
"user": "victim_user",
"team_id": None,
"spend": 0.07,
"startTime": now_iso,
"model": "gpt-4",
},
]
def filter_fn(where):
rid_either = where.get("request_id_or_call_id")
rows = [r for r in corpus if rid_either in (r["request_id"], r["litellm_call_id"])]
return [r for r in rows if where.get("user") is None or r["user"] == where["user"]]
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", make_ui_spend_logs_mock_prisma(corpus, filter_fn))
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.ORG_ADMIN, user_id="victim_user"
)
try:
response = client.get(
"/spend/logs/ui",
params={"request_id": "victim-req"},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 200, response.text
data = response.json()
assert data["total"] == 1
assert [row["request_id"] for row in data["data"]] == ["victim-req"]
assert "attacker_user" not in response.text
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_spend_logs_request_id_rejects_foreign_row_inserted_after_owner_check(client, monkeypatch):
"""The SQL scope keeps foreign rows out of an id lookup; this backstop covers a
row the scope did not filter (the mock ignores it on purpose). The rows actually
fetched are ownership-checked again, so the lookup answers 403 instead of serving
the other tenant's row."""
now_iso = datetime.datetime.now(timezone.utc).isoformat()
owned_row = {
"id": "log_owned",
"request_id": "attacker-req",
"litellm_call_id": "shared-id",
"api_key": "sk-test-key",
"user": "user_1",
"team_id": None,
"spend": 0.05,
"startTime": now_iso,
"model": "gpt-4",
}
foreign_row = {
"id": "log_foreign",
"request_id": "shared-id",
"litellm_call_id": None,
"api_key": "sk-victim-key",
"user": "victim_user",
"team_id": None,
"spend": 0.07,
"startTime": now_iso,
"model": "gpt-4",
}
mock_prisma = make_ui_spend_logs_mock_prisma([owned_row], lambda where: [owned_row, foreign_row])
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", mock_prisma)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1"
)
try:
response = client.get(
"/spend/logs/ui",
params={"request_id": "shared-id"},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 403
assert "victim_user" not in response.text
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
def _make_payload_lookup_prisma(rows):
"""Emulate the detail endpoint's SQL over an in-memory corpus: the owner
pre-check, the caller scope on ``"user"`` and permitted teams, and the
exact-request_id-first ordering with LIMIT 1."""
class MockDB:
async def query_raw(self, sql_query, *params):
if 'SELECT DISTINCT "user", team_id' in sql_query:
return _emulate_spend_log_owner_lookup(rows, sql_query, params)
lookup_id = params[0]
matches = [r for r in rows if lookup_id in (r["request_id"], r["litellm_call_id"])]
if '"user" = $2' in sql_query:
team_ids = params[2] if "ANY($3::text[])" in sql_query else ()
matches = [r for r in matches if r["user"] == params[1] or r["team_id"] in team_ids]
if "ORDER BY (request_id = $1) DESC" in sql_query:
matches = sorted(matches, key=lambda r: r["request_id"] == lookup_id, reverse=True)
return matches[:1]
class MockPrisma:
def __init__(self):
self.db = MockDB()
return MockPrisma()
def _payload_row(request_id, litellm_call_id, user, prompt):
return {
"request_id": request_id,
"litellm_call_id": litellm_call_id,
"messages": [{"role": "user", "content": prompt}],
"response": {"id": request_id},
"proxy_server_request": None,
"metadata": None,
"user": user,
"team_id": None,
}
@pytest.mark.asyncio
async def test_ui_view_request_response_collision_serves_callers_own_row(client, monkeypatch):
"""The attacker's row carries the victim's request_id as its client-set call id
and was written first. Each tenant's detail lookup of that id serves only their
own payload, and an admin's lookup resolves the exact request_id match rather
than whichever colliding row the database happens to return first."""
prisma = _make_payload_lookup_prisma(
[
_payload_row("attacker-req", "victim-req", "attacker_user", "attacker prompt"),
_payload_row("victim-req", "victim-call-id", "victim_user", "victim prompt"),
]
)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma)
try:
for role, user_id, own_prompt, other_prompt in (
(LitellmUserRoles.INTERNAL_USER, "victim_user", "victim prompt", "attacker prompt"),
(LitellmUserRoles.INTERNAL_USER, "attacker_user", "attacker prompt", "victim prompt"),
(LitellmUserRoles.PROXY_ADMIN, "admin", "victim prompt", "attacker prompt"),
):
app.dependency_overrides[ps.user_api_key_auth] = lambda role=role, user_id=user_id: UserAPIKeyAuth(
user_role=role, user_id=user_id
)
response = client.get("/spend/logs/ui/victim-req", headers={"Authorization": "Bearer sk-test"})
assert response.status_code == 200, response.text
assert own_prompt in response.text
assert other_prompt not in response.text
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_request_response_rejects_foreign_row_inserted_after_owner_check(client, monkeypatch):
"""Backstop behind the SQL scope on the detail endpoint (the mock ignores the
scope on purpose): the payload row fetched by id is itself ownership-checked, so
a foreign row the scope did not filter cannot have its payload served."""
class MockDB:
async def query_raw(self, sql_query, *params):
if 'SELECT DISTINCT "user", team_id' in sql_query:
return [{"user": "user_1", "team_id": None}]
return [
{
"messages": [{"role": "user", "content": "victim prompt"}],
"response": {"id": "resp-1"},
"proxy_server_request": None,
"metadata": None,
"user": "victim_user",
"team_id": None,
}
]
class MockPrisma:
def __init__(self):
self.db = MockDB()
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", MockPrisma())
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1"
)
try:
response = client.get(
"/spend/logs/ui/shared-id",
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 403
assert "victim prompt" not in response.text
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_request_response_custom_logger_denies_foreign_payload_owner(client, monkeypatch):
"""The custom-logger payload comes straight from cold storage, written independently
of the spend-log table and able to outlive its row. When an id lookup matches no row,
the DB owner pre-check has nothing to verify, so the payload is authorized against the
owner recorded inside it. A foreign tenant's stored payload is denied even though no
spend-log row exists for the pre-check to catch."""
class MockDB:
async def query_raw(self, sql_query, *params):
return []
class MockPrisma:
def __init__(self):
self.db = MockDB()
class ColdStorageLogger:
async def get_request_response_payload(self, request_id, start_time_utc, end_time_utc):
return {
"messages": [{"role": "user", "content": "victim prompt"}],
"response": {"id": "r"},
"metadata": {"user_api_key_user_id": "victim_user", "user_api_key_team_id": None},
}
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", MockPrisma())
monkeypatch.setattr(
litellm.logging_callback_manager,
"get_active_additional_logging_utils_from_custom_logger",
lambda: [ColdStorageLogger()],
)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1"
)
try:
response = client.get(
"/spend/logs/ui/shared-id",
params={"start_date": "2026-01-01 00:00:00"},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 403
assert "victim prompt" not in response.text
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_request_response_custom_logger_is_keyed_by_callers_own_request_id(client, monkeypatch):
"""Cold storage is keyed by the provider request_id. The caller's row carries the
lookup id only as its client-set litellm_call_id while another tenant's row owns
that id as its request_id. The custom logger is asked for the caller's own stored
request_id, so the caller gets their payload rather than a 403 from the foreign
payload's owner check, and the foreign payload is never fetched."""
prisma = _make_payload_lookup_prisma(
[
_payload_row("shared-id", "other-call-id", "other_user", "other tenant prompt"),
_payload_row("caller-req", "shared-id", "caller_user", "caller prompt"),
]
)
cold_storage = {
"shared-id": {
"messages": [{"role": "user", "content": "other tenant prompt"}],
"response": {"id": "shared-id"},
"metadata": {"user_api_key_user_id": "other_user", "user_api_key_team_id": None},
},
"caller-req": {
"messages": [{"role": "user", "content": "caller prompt"}],
"response": {"id": "caller-req"},
"metadata": {"user_api_key_user_id": "caller_user", "user_api_key_team_id": None},
},
}
requested_ids = []
class ColdStorageLogger:
async def get_request_response_payload(self, request_id, start_time_utc, end_time_utc):
requested_ids.append(request_id)
return cold_storage.get(request_id)
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", prisma)
monkeypatch.setattr(
litellm.logging_callback_manager,
"get_active_additional_logging_utils_from_custom_logger",
lambda: [ColdStorageLogger()],
)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="caller_user"
)
try:
response = client.get("/spend/logs/ui/shared-id", headers={"Authorization": "Bearer sk-test"})
assert response.status_code == 200, response.text
assert "caller prompt" in response.text
assert "other tenant prompt" not in response.text
assert requested_ids == ["caller-req"]
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
@pytest.mark.parametrize("group_by_session", [False, True])
async def test_ui_view_spend_logs_id_lookup_lists_exact_request_id_row_first(client, monkeypatch, group_by_session):
"""The dashboard's deep link fetches a single row for ``?log_id=``. When a newer
row carries that id as its client-set litellm_call_id, the row whose request_id
is the id still comes first, so the link opens the request it names. The
session-grouped page orders its representatives the same way."""
today = datetime.datetime.now(timezone.utc)
corpus = [
{
"id": "log_colliding",
"request_id": "colliding-req",
"litellm_call_id": "victim-req",
"api_key": "sk-test-key",
"user": "other_user",
"team_id": None,
"spend": 0.01,
"startTime": today.isoformat(),
"model": "gpt-4",
},
{
"id": "log_victim",
"request_id": "victim-req",
"litellm_call_id": "victim-call-id",
"api_key": "sk-test-key",
"user": "victim_user",
"team_id": None,
"spend": 0.02,
"startTime": (today - datetime.timedelta(minutes=5)).isoformat(),
"model": "gpt-4",
},
]
def filter_fn(where):
rows = _filter_logs_by_date_range(corpus, where)
rid_either = where.get("request_id_or_call_id")
if rid_either:
return [r for r in rows if rid_either in (r["request_id"], r["litellm_call_id"])]
return rows
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", make_ui_spend_logs_mock_prisma(corpus, filter_fn))
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.PROXY_ADMIN, user_id="admin"
)
try:
response = client.get(
"/spend/logs/ui",
params={"request_id": "victim-req", "page_size": 1, "group_by_session": str(group_by_session).lower()},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 200, response.text
data = response.json()
assert data["total"] == 2
assert [row["request_id"] for row in data["data"]] == ["victim-req"]
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_spend_logs_request_id_owner_lookup_drops_window_keeps_scope(
client, monkeypatch
):
"""A non-admin owner looking up their own request_id resolves across all time.
The ownership check authorizes the single row, so the query drops both the date
window and the general user/team scoping and filters by the primary key alone;
without that skip an internal user would have a `user`/`OR` clause added."""
"""A non-admin owner looking up their own request_id resolves across all time:
the query drops the date window the dashboard sends, while the caller's own-user
scope stays on the id lookup so a colliding foreign row can never be served."""
today = datetime.datetime.now(timezone.utc)
mock_spend_logs = [
{
@ -2361,20 +2991,14 @@ async def test_ui_view_spend_logs_request_id_owner_scoped_by_id_only(
def filter_fn(where):
captured["where"] = where
rows = _filter_logs_by_date_range(mock_spend_logs, where)
if where.get("request_id"):
rid_either = where.get("request_id_or_call_id")
if rid_either:
rows = [r for r in rows if rid_either in (r["request_id"], r.get("litellm_call_id"))]
elif where.get("request_id"):
rows = [r for r in rows if r["request_id"] == where["request_id"]]
return rows
mock_prisma = make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn)
class _OwnedRow:
user = "user_1"
team_id = "team1"
async def _find_unique(where, include=None):
return _OwnedRow()
mock_prisma.db.find_unique = _find_unique
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.
@ -2399,9 +3023,8 @@ async def test_ui_view_spend_logs_request_id_owner_scoped_by_id_only(
assert data["total"] == 1
assert data["data"][0]["request_id"] == "req-old"
assert "startTime" not in captured["where"]
assert captured["where"]["request_id"] == "req-old"
assert "user" not in captured["where"]
assert "OR" not in captured["where"]
assert captured["where"]["request_id_or_call_id"] == "req-old"
assert captured["where"]["user"] == "user_1"
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@ -5095,7 +5718,7 @@ async def test_view_spend_logs_internal_user_combines_user_with_request_id(
where = mock_client.db.captured_where
assert where is not None
assert where["user"] == "internal-user-2"
assert where["request_id"] == "req-abc"
assert where["OR"] == ({"request_id": "req-abc"}, {"litellm_call_id": "req-abc"})
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@ -5122,7 +5745,7 @@ async def test_view_spend_logs_non_date_range_combines_user_with_request_id(
where = mock_client.db.captured_where
assert where is not None
assert where["user"] == "internal-user-3"
assert where["request_id"] == "req-xyz"
assert where["OR"] == ({"request_id": "req-xyz"}, {"litellm_call_id": "req-xyz"})
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@ -7052,25 +7675,19 @@ async def test_ui_view_spend_logs_search_returns_flat_rows_when_grouping_by_sess
def _fake_prisma_with_owned_spend_log(owner_user_id, messages_json, response_json):
class _Row:
user = owner_user_id
team_id = None
class _SpendLogs:
async def find_unique(self, where, include=None):
return _Row()
class _DB:
def __init__(self):
self.litellm_spendlogs = _SpendLogs()
async def query_raw(self, _sql, *_args):
async def query_raw(self, sql, *_args):
if 'SELECT DISTINCT "user", team_id' in sql:
return [{"user": owner_user_id, "team_id": None}]
return [
{
"request_id": "req-owned-by-user-a",
"messages": messages_json,
"response": response_json,
"proxy_server_request": "{}",
"metadata": "{}",
"user": owner_user_id,
"team_id": None,
}
]
@ -7127,8 +7744,8 @@ def test_ui_view_request_response_internal_user_non_owner_forbidden(client, monk
"""
A different internal_user requesting someone else's row is forbidden;
guards against _assert_user_can_view_request_id being skipped in the
detail-drawer handler. Also proves the handler stops before it ever asks
a custom logger or the DB for the payload.
detail-drawer handler. Also proves the handler stops at the owner lookup,
before it ever asks a custom logger or the DB for the payload.
"""
messages_json = json.dumps([{"role": "user", "content": "hi"}])
response_json = json.dumps({"choices": [{"message": {"content": "hello"}}]})
@ -7160,7 +7777,7 @@ def test_ui_view_request_response_internal_user_non_owner_forbidden(client, monk
)
assert response.status_code == 403
assert custom_logger.requested_ids == []
assert query_raw_calls == []
assert [args[0] for args, _kwargs in query_raw_calls if "messages" in args[0]] == []
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@ -7171,24 +7788,27 @@ def test_ui_view_request_response_internal_user_missing_row_forbidden(client, mo
request_id with no spend-log row (e.g. pruned by retention) must be
denied before the handler ever consults a custom logger, otherwise a
non-admin who guesses/obtains a request_id could read another tenant's
payload out of cold storage. Fails if `if row is None: return` is
reintroduced.
payload out of cold storage, and an existing payload that is not theirs
would still confirm the id exists. Even a payload recorded as the caller's
own is never fetched once the row is gone. Fails if an empty owner lookup
is allowed to fall through to the loggers.
"""
class _SpendLogs:
async def find_unique(self, where, include=None):
return None
class _DB:
def __init__(self):
self.litellm_spendlogs = _SpendLogs()
async def query_raw(self, _sql, *_args):
return []
from types import SimpleNamespace
fake_prisma = SimpleNamespace(db=_DB())
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", fake_prisma)
custom_logger = _RecordingAdditionalLoggingUtils({"messages": "should-not-be-returned"})
custom_logger = _RecordingAdditionalLoggingUtils(
{
"messages": "should-not-be-returned",
"metadata": {"user_api_key_user_id": "user_a", "user_api_key_team_id": None},
}
)
monkeypatch.setattr(
litellm.logging_callback_manager,
"get_active_additional_logging_utils_from_custom_logger",

27
tests/test_litellm/proxy/spend_tracking/test_spend_tracking_utils.py
View file

@ -1329,6 +1329,33 @@ def test_get_logging_payload_includes_agent_id_from_kwargs():
assert payload["agent_id"] == test_agent_id, f"Expected agent_id '{test_agent_id}', got '{payload.get('agent_id')}'"
def test_get_logging_payload_populates_litellm_call_id_alongside_provider_request_id():
"""
LIT-6302: request_id stays the provider response id, so clients holding the
x-litellm-call-id header value could never find their row. The payload now
also carries litellm_call_id as its own column for lookups by either id.
"""
call_id = "b980eea9-5cd9-4099-93cd-8291e46c76fd"
payload = get_logging_payload(
kwargs={
"model": "gpt-4o-mini",
"litellm_call_id": call_id,
"litellm_params": {"metadata": {"user_api_key": "test-key"}},
},
response_obj=litellm.ModelResponse(
id="chatcmpl-provider-id",
choices=[],
usage=litellm.Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2),
),
start_time=datetime.datetime.now(timezone.utc),
end_time=datetime.datetime.now(timezone.utc),
)
assert payload["request_id"] == "chatcmpl-provider-id"
assert payload["litellm_call_id"] == call_id
@patch("litellm.proxy.proxy_server.master_key", None)
@patch("litellm.proxy.proxy_server.general_settings", {})
def test_get_logging_payload_includes_overhead_in_spend_logs_metadata():

16
tests/test_litellm/proxy/test_common_request_processing.py
View file

@ -13,7 +13,7 @@ from fastapi.responses import JSONResponse, StreamingResponse
import litellm
from litellm._uuid import uuid
from litellm.constants import RETURN_RAW_MODEL_NAME_METADATA_KEY
from litellm.constants import MAX_LITELLM_CALL_ID_LENGTH, RETURN_RAW_MODEL_NAME_METADATA_KEY
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.opentelemetry import UserAPIKeyAuth
from litellm.proxy.common_request_processing import (
@ -30,6 +30,7 @@ from litellm.proxy.common_request_processing import (
_has_attribute_error_in_chain,
_is_azure_model_router_request,
open_sse_before_first_byte,
resolve_litellm_call_id,
ttft_keepalive_interval,
_override_openai_response_model,
_parse_event_data_for_error,
@ -8060,6 +8061,19 @@ def test_log_llm_api_exception_traceback_only_for_unexpected_errors(exc, expect_
assert (records[0].exc_info is not None) is expect_traceback
class TestResolveLitellmCallId:
def test_client_call_id_within_the_bound_is_kept(self):
assert resolve_litellm_call_id("req-abc-123") == "req-abc-123"
at_bound: Final = "y" * MAX_LITELLM_CALL_ID_LENGTH
assert resolve_litellm_call_id(at_bound) == at_bound
@pytest.mark.parametrize("client_call_id", [None, "", "x" * (MAX_LITELLM_CALL_ID_LENGTH + 1), "z" * 3000])
def test_missing_empty_or_oversized_client_call_id_gets_a_generated_uuid(self, client_call_id):
resolved: Final = resolve_litellm_call_id(client_call_id)
assert resolved != client_call_id
assert uuid.UUID(resolved).version == 4
class _FailureHookRecorder:
"""Stands in for ProxyLogging.post_call_failure_hook, recording what the detached-failure closure hands it."""

97
ui/litellm-dashboard/src/components/GuardrailsMonitor/LogViewer.test.tsx Normal file
View file

@ -0,0 +1,97 @@
import userEvent from "@testing-library/user-event";
import React from "react";
import { beforeEach, describe, expect, it, vi } from "vitest";
import { renderWithProviders, screen, testQueryClient, waitFor } from "../../../tests/test-utils";
import type { LogEntry as SpendLogEntry } from "@/components/view_logs/columns";
import { LogViewer } from "./LogViewer";
vi.mock("@/components/networking", async (importOriginal) => {
const actual = await importOriginal<typeof import("@/components/networking")>();
return { ...actual, uiSpendLogsCall: vi.fn() };
});
vi.mock("@/components/view_logs/LogDetailsDrawer", () => ({
LogDetailsDrawer: function LogDetailsDrawerMock({
open,
logEntry,
}: {
open: boolean;
logEntry?: { request_id: string } | null;
}) {
return (
<div data-testid="log-details-drawer" data-log-id={logEntry?.request_id ?? ""}>
{open ? "open" : "closed"}
</div>
);
},
}));
import { uiSpendLogsCall } from "@/components/networking";
const spendLog = (overrides: Partial<SpendLogEntry>): SpendLogEntry => ({
request_id: "req-1",
api_key: "key-1",
team_id: "team-1",
model: "gpt-4o",
model_id: "model-1",
call_type: "acompletion",
spend: 0.01,
total_tokens: 10,
prompt_tokens: 5,
completion_tokens: 5,
startTime: "2026-09-02T09:50:13Z",
endTime: "2026-09-02T09:50:14Z",
cache_hit: "false",
messages: [],
response: {},
...overrides,
});
const guardrailLog = {
id: "provider-victim",
timestamp: "2026-09-02 09:50:13",
action: "passed" as const,
input_snippet: "victim prompt",
};
describe("GuardrailsMonitor LogViewer drawer", () => {
beforeEach(() => {
vi.mocked(uiSpendLogsCall).mockReset();
testQueryClient.clear();
});
it("opens the row whose request_id is the clicked log id even when a newer row carries that id as its call id", async () => {
vi.mocked(uiSpendLogsCall).mockResolvedValue({
data: [
spendLog({ request_id: "provider-attacker", litellm_call_id: "provider-victim" }),
spendLog({ request_id: "provider-victim", litellm_call_id: "call-victim" }),
],
total: 2,
});
renderWithProviders(<LogViewer logs={[guardrailLog]} accessToken="sk-test" />);
await userEvent.click(screen.getByText("victim prompt"));
await waitFor(() => {
expect(screen.getByTestId("log-details-drawer")).toHaveAttribute("data-log-id", "provider-victim");
});
expect(vi.mocked(uiSpendLogsCall)).toHaveBeenCalledWith(
expect.objectContaining({ params: { request_id: "provider-victim" } }),
);
});
it("falls back to the first returned row when none carries the clicked id as its request_id", async () => {
vi.mocked(uiSpendLogsCall).mockResolvedValue({
data: [spendLog({ request_id: "provider-other", litellm_call_id: "provider-victim" })],
total: 1,
});
renderWithProviders(<LogViewer logs={[guardrailLog]} accessToken="sk-test" />);
await userEvent.click(screen.getByText("victim prompt"));
await waitFor(() => {
expect(screen.getByTestId("log-details-drawer")).toHaveAttribute("data-log-id", "provider-other");
});
});
});

3
ui/litellm-dashboard/src/components/GuardrailsMonitor/LogViewer.tsx
View file

@ -92,7 +92,8 @@ export function LogViewer({
enabled: Boolean(accessToken && selectedRequestId && drawerOpen),
});
const selectedLog: ViewLogsLogEntry | null = fullLogResponse?.data?.[0] ?? null;
const selectedLog: ViewLogsLogEntry | null =
fullLogResponse?.data?.find((log) => log.request_id === selectedRequestId) ?? fullLogResponse?.data?.[0] ?? null;
const handleLogClick = (log: LogEntry) => {
setSelectedRequestId(log.id);

43
ui/litellm-dashboard/src/components/view_logs/RequestLogsPanel.test.tsx
View file

@ -577,6 +577,49 @@ describe("RequestLogsPanel", () => {
expect(byIdCall.params?.group_by_session).toBeUndefined();
});
it("opens the drawer when ?log_id= is the log's litellm_call_id rather than its request_id", async () => {
respondWith([logEntry({ request_id: "chatcmpl-provider", litellm_call_id: "call-1" })]);
renderPanel("?log_id=call-1");
await waitFor(() => {
expect(drawer()).toHaveTextContent("open");
});
expect(drawer()).toHaveAttribute("data-log-id", "chatcmpl-provider");
});
it("fetches by litellm_call_id and opens the drawer when that log is not in the loaded page", async () => {
vi.mocked(uiSpendLogsCall).mockImplementation(async ({ params }) =>
params?.request_id === "call-old"
? {
data: [logEntry({ request_id: "chatcmpl-old", litellm_call_id: "call-old" })],
total: 1,
page: 1,
page_size: 1,
total_pages: 1,
}
: { data: [], total: 0, page: 1, page_size: 50, total_pages: 0 },
);
renderPanel("?log_id=call-old");
await waitFor(() => {
expect(drawer()).toHaveTextContent("open");
});
expect(drawer()).toHaveAttribute("data-log-id", "chatcmpl-old");
});
it("opens the exact request_id row when another log in the page carries that id as its litellm_call_id", async () => {
respondWith([
logEntry({ request_id: "chatcmpl-other", litellm_call_id: "victim-req" }),
logEntry({ request_id: "victim-req", litellm_call_id: "victim-call" }),
]);
renderPanel("?log_id=victim-req");
await waitFor(() => {
expect(drawer()).toHaveTextContent("open");
});
expect(drawer()).toHaveAttribute("data-log-id", "victim-req");
});
it("closing the drawer removes ?log_id= from the URL and closes the drawer", async () => {
const user = userEvent.setup();
respondWith([logEntry({ request_id: "req-1" })]);

11
ui/litellm-dashboard/src/components/view_logs/RequestLogsPanel.tsx
View file

@ -28,6 +28,9 @@ import { RequestLogsTable } from "./RequestLogsTable";
const PAGE_SIZE = DEFAULT_PAGE_SIZE_OPTIONS[0];
const DEFAULT_INTERVAL = { value: 24, unit: "hours" };
const matchesLogId = (log: LogEntry, logId: string) => log.request_id === logId || log.litellm_call_id === logId;
const findLogById = (logs: readonly LogEntry[], logId: string): LogEntry | null =>
logs.find((log) => log.request_id === logId) ?? logs.find((log) => log.litellm_call_id === logId) ?? null;
interface RequestLogsPanelProps {
accessToken: string;
@ -141,9 +144,9 @@ export default function RequestLogsPanel({ accessToken, token, userRole, userID,
page_size: 1,
params: { request_id: urlLogId },
});
return response.data.find((log) => log.request_id === urlLogId) ?? null;
return findLogById(response.data, urlLogId);
},
enabled: urlLogId !== null && selectedLog?.request_id !== urlLogId,
enabled: urlLogId !== null && !(selectedLog !== null && matchesLogId(selectedLog, urlLogId)),
staleTime: Infinity,
};
@ -151,8 +154,8 @@ export default function RequestLogsPanel({ accessToken, token, userRole, userID,
const displayLog = useMemo<LogEntry | null>(() => {
if (urlLogId === null) return null;
if (selectedLog?.request_id === urlLogId) return selectedLog;
return filteredLogs.data.find((log) => log.request_id === urlLogId) ?? urlLog ?? null;
if (selectedLog !== null && matchesLogId(selectedLog, urlLogId)) return selectedLog;
return findLogById(filteredLogs.data, urlLogId) ?? urlLog ?? null;
}, [urlLogId, selectedLog, filteredLogs.data, urlLog]);
const displaySessionId = useMemo<string | null>(() => {

1
ui/litellm-dashboard/src/components/view_logs/columns.tsx
View file

@ -12,6 +12,7 @@ export type LogsSortField = keyof typeof LOGS_SORT_FIELD_MAP;
export type LogEntry = {
request_id: string;
litellm_call_id?: string | null;
api_key: string;
team_id: string;
model: string;