fix: re-verify ownership on custom-logger payload branch for id lookups

This commit is contained in:
mateo-berri 2026-09-01 11:33:20 -07:00
parent 5382f9b720
commit 61cef45d8e
2 changed files with 53 additions and 0 deletions

View file

@ -2896,6 +2896,12 @@ async def ui_view_request_response_for_request_id(
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_can_view_request_id(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
request_id=request_id,
)
return payload
# Fallback: the list endpoint omits the heavy columns for performance, so

View file

@ -2566,6 +2566,53 @@ async def test_ui_view_request_response_rejects_foreign_row_inserted_after_owner
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_request_response_custom_logger_rechecks_after_fetch(client, monkeypatch):
"""The custom-logger payload branch re-verifies ownership after fetching. A row
that appears between the pre-check and the payload read (so the pre-check saw only
owned rows) is caught on the post-fetch check, so the foreign payload is not served."""
owner_states = iter(
[
[{"user": "user_1", "team_id": None}],
[{"user": "user_1", "team_id": None}, {"user": "victim_user", "team_id": None}],
]
)
class MockDB:
async def query_raw(self, sql_query, *params):
if 'SELECT DISTINCT "user", team_id' in sql_query:
return next(owner_states)
return []
class MockPrisma:
def __init__(self):
self.db = MockDB()
class LeakyLogger:
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"}}
monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", MockPrisma())
monkeypatch.setattr(
litellm.logging_callback_manager,
"get_active_additional_logging_utils_from_custom_logger",
lambda: [LeakyLogger()],
)
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_spend_logs_request_id_owner_scoped_by_id_only(
client, monkeypatch