added option to allow team user to see logs of team

This commit is contained in:
shivam
2026-04-09 17:43:20 -07:00
parent 072d4108c3
commit 31f750146b
5 changed files with 412 additions and 16 deletions
+4
View File
@@ -247,6 +247,9 @@ class KeyManagementRoutes(str, enum.Enum):
# team usage routes
TEAM_DAILY_ACTIVITY = "/team/daily/activity"
# team spend-log viewing
SPEND_LOGS = "/spend/logs"
class LiteLLMRoutes(enum.Enum):
openai_route_names = [
@@ -520,6 +523,7 @@ class LiteLLMRoutes(enum.Enum):
KeyManagementRoutes.KEY_UNBLOCK.value,
KeyManagementRoutes.KEY_BULK_UPDATE.value,
KeyManagementRoutes.TEAM_DAILY_ACTIVITY.value,
KeyManagementRoutes.SPEND_LOGS.value,
KeyManagementRoutes.KEY_RESET_SPEND.value,
KeyManagementRoutes.KEY_ALIASES.value,
]
@@ -15,6 +15,7 @@ from litellm.proxy._types import ProviderBudgetResponse, ProviderBudgetResponseO
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin,
_team_member_has_permission,
_user_has_admin_view,
)
from litellm.proxy.spend_tracking.spend_tracking_utils import (
@@ -1870,6 +1871,7 @@ async def ui_view_spend_logs( # noqa: PLR0915
if max_spend is not None:
where_conditions["spend"]["lte"] = max_spend
is_admin_view = _is_admin_view_safe(user_api_key_dict=user_api_key_dict)
permitted_team_ids: Optional[List[str]] = None
if not is_admin_view:
if team_id is not None:
can_view_team = await _can_team_member_view_log(
@@ -1887,9 +1889,26 @@ async def ui_view_spend_logs( # noqa: PLR0915
},
)
where_conditions["team_id"] = team_id
where_conditions.pop("user", None)
else:
if _can_user_view_spend_log(user_api_key_dict=user_api_key_dict):
where_conditions["user"] = user_api_key_dict.user_id
try:
permitted_team_ids = (
await _get_permitted_team_ids_for_spend_logs(
prisma_client=prisma_client,
user_api_key_dict=user_api_key_dict,
)
)
except Exception:
permitted_team_ids = []
if permitted_team_ids:
where_conditions.pop("user", None)
where_conditions["OR"] = [
{"user": user_api_key_dict.user_id},
{"team_id": {"in": permitted_team_ids}},
]
else:
where_conditions["user"] = user_api_key_dict.user_id
where_conditions.pop("team_id", None)
# Calculate skip value for pagination
skip = (page - 1) * page_size
@@ -1934,6 +1953,14 @@ async def ui_view_spend_logs( # noqa: PLR0915
sql_params.append(val)
p += 1
# Multi-team OR filter: (user = $X OR team_id = ANY($Y))
if permitted_team_ids is not None and len(permitted_team_ids) > 0:
or_clause = f'("user" = ${p} OR team_id = ANY(${p + 1}::text[]))'
sql_params.append(user_api_key_dict.user_id)
sql_params.append(permitted_team_ids)
p += 2
sql_conditions.append(or_clause)
# Status filter
if status_filter is not None:
if status_filter == "success":
@@ -2033,6 +2060,7 @@ async def ui_view_request_response_for_request_id(
default=None,
description="Time till which to view key spend",
),
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
View request / response for a specific request_id
@@ -2040,6 +2068,16 @@ async def ui_view_request_response_for_request_id(
- goes through all callbacks, checks if any of them have a @property -> has_request_response_payload
- if so, it will return the request and response payload
"""
from litellm.proxy.proxy_server import prisma_client
if not _is_admin_view_safe(user_api_key_dict=user_api_key_dict):
if 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,
)
custom_loggers = (
litellm.logging_callback_manager.get_active_additional_logging_utils_from_custom_logger()
)
@@ -2068,8 +2106,6 @@ async def ui_view_request_response_for_request_id(
# response, and proxy_server_request for performance. When no custom
# logger (S3, GCS, etc.) is configured, we still need to serve these
# fields from the DB for the detail/drawer view.
from litellm.proxy.proxy_server import prisma_client
if prisma_client is not None:
sql_query = """
SELECT messages, response, proxy_server_request
@@ -3419,16 +3455,24 @@ async def _can_team_member_view_log(
) -> bool:
"""
Check if the requesting user can view spend logs for the given team.
Returns True only if the team exists and the user is a team admin.
Returns True if the team exists and the user is either a team admin or
a team member with the ``/spend/logs`` permission.
"""
if team_id is None:
return False
team_obj = await prisma_client.db.litellm_teamtable.find_unique(
team_row = await prisma_client.db.litellm_teamtable.find_unique(
where={"team_id": team_id}
)
if team_obj is None:
if team_row is None:
return False
return _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
team_obj = LiteLLM_TeamTable(**team_row.model_dump())
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
return True
return _team_member_has_permission(
user_api_key_dict=user_api_key_dict,
team_obj=team_obj,
permission=KeyManagementRoutes.SPEND_LOGS.value,
)
def _can_user_view_spend_log(user_api_key_dict: UserAPIKeyAuth) -> bool:
@@ -3445,3 +3489,84 @@ def _can_user_view_spend_log(user_api_key_dict: UserAPIKeyAuth) -> bool:
)
and user_id is not None
)
async def _assert_user_can_view_request_id(
prisma_client,
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.
"""
row = await prisma_client.db.litellm_spendlogs.find_unique(
where={"request_id": request_id},
include=None,
)
if row is None:
return
if row.user == user_api_key_dict.user_id:
return
if row.team_id:
can_view = 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": "Not authorized to view spend log for request_id={}".format(
request_id
)
},
)
async def _get_permitted_team_ids_for_spend_logs(
prisma_client,
user_api_key_dict: UserAPIKeyAuth,
) -> List[str]:
"""
Return team IDs where the user is either a team admin or has the
``/spend/logs`` permission, allowing them to view team-wide spend logs.
"""
from litellm.proxy.auth.auth_checks import get_user_object
from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache
user_obj = await get_user_object(
user_id=user_api_key_dict.user_id,
prisma_client=prisma_client,
user_api_key_cache=user_api_key_cache,
user_id_upsert=False,
proxy_logging_obj=proxy_logging_obj,
)
if user_obj is None or not user_obj.teams:
return []
team_rows = await prisma_client.db.litellm_teamtable.find_many(
where={"team_id": {"in": user_obj.teams}}
)
permitted: List[str] = []
for team_row in team_rows:
team_obj = LiteLLM_TeamTable(**team_row.model_dump())
if _is_user_team_admin(
user_api_key_dict=user_api_key_dict, team_obj=team_obj
):
permitted.append(team_obj.team_id)
elif _team_member_has_permission(
user_api_key_dict=user_api_key_dict,
team_obj=team_obj,
permission=KeyManagementRoutes.SPEND_LOGS.value,
):
permitted.append(team_obj.team_id)
return permitted
@@ -97,6 +97,8 @@ def make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn, team_lookup_fn=No
from litellm.proxy._types import (
LiteLLM_TeamTable,
LiteLLM_UserTable,
LitellmUserRoles,
Member,
SpendLogsPayload,
@@ -198,9 +200,18 @@ async def test_can_team_member_view_log_team_not_found(monkeypatch):
@pytest.mark.asyncio
async def test_can_team_member_view_log_not_admin(monkeypatch):
# Existing team but caller is not a team admin -> False
# Existing team but caller is not a team admin and no /spend/logs permission -> False
class MockTeam:
pass
team_id = "team_x"
members_with_roles = [Member(user_id="user_1", role="user")]
team_member_permissions = None
def model_dump(self):
return {
"team_id": self.team_id,
"members_with_roles": [{"user_id": "user_1", "role": "user"}],
"team_member_permissions": self.team_member_permissions,
}
class MockPrisma:
class DB:
@@ -231,7 +242,16 @@ async def test_can_team_member_view_log_not_admin(monkeypatch):
async def test_can_team_member_view_log_admin(monkeypatch):
# Existing team and caller is team admin -> True
class MockTeam:
pass
team_id = "team_x"
members_with_roles = [Member(user_id="user_1", role="admin")]
team_member_permissions = None
def model_dump(self):
return {
"team_id": self.team_id,
"members_with_roles": [{"user_id": "user_1", "role": "admin"}],
"team_member_permissions": self.team_member_permissions,
}
class MockPrisma:
class DB:
@@ -246,11 +266,6 @@ async def test_can_team_member_view_log_admin(monkeypatch):
self.db = self.DB()
prisma = MockPrisma()
monkeypatch.setattr(
spend_management_endpoints,
"_is_user_team_admin",
lambda user_api_key_dict, team_obj: True,
)
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
allowed = await spend_management_endpoints._can_team_member_view_log(
prisma, auth, "team_x"
@@ -866,7 +881,16 @@ async def test_ui_view_spend_logs_team_admin_can_view_team_spend(client, monkeyp
return mock_spend_logs
class TeamTable:
team_id = "team_admin_team"
members_with_roles = [Member(user_id="admin_user", role="admin")]
team_member_permissions = None
def model_dump(self):
return {
"team_id": self.team_id,
"members_with_roles": [{"user_id": "admin_user", "role": "admin"}],
"team_member_permissions": self.team_member_permissions,
}
async def team_lookup(where):
return TeamTable() if where == {"team_id": "team_admin_team"} else None
@@ -2473,3 +2497,225 @@ async def test_build_ui_spend_logs_response_dict_rows_session_counts():
where={"session_id": {"in": [session_id]}},
count={"session_id": True},
)
# ---------------------------------------------------------------------------
# Tests for /spend/logs team-member permission
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_can_team_member_view_log_with_spend_logs_permission(monkeypatch):
"""
Non-admin team member WITH /spend/logs permission should be allowed.
"""
class MockTeam:
team_id = "team_abc"
members_with_roles = [Member(user_id="member_1", role="user")]
team_member_permissions = ["/spend/logs"]
def model_dump(self):
return {
"team_id": self.team_id,
"members_with_roles": [{"user_id": "member_1", "role": "user"}],
"team_member_permissions": self.team_member_permissions,
}
class MockPrisma:
class DB:
class TeamTable:
async def find_unique(self, where: dict):
return MockTeam()
def __init__(self):
self.litellm_teamtable = self.TeamTable()
def __init__(self):
self.db = self.DB()
prisma = MockPrisma()
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="member_1")
allowed = await spend_management_endpoints._can_team_member_view_log(
prisma, auth, "team_abc"
)
assert allowed is True
@pytest.mark.asyncio
async def test_can_team_member_view_log_without_spend_logs_permission(monkeypatch):
"""
Non-admin team member WITHOUT /spend/logs permission should be denied.
"""
class MockTeam:
team_id = "team_abc"
members_with_roles = [Member(user_id="member_1", role="user")]
team_member_permissions = ["/key/info"]
def model_dump(self):
return {
"team_id": self.team_id,
"members_with_roles": [{"user_id": "member_1", "role": "user"}],
"team_member_permissions": self.team_member_permissions,
}
class MockPrisma:
class DB:
class TeamTable:
async def find_unique(self, where: dict):
return MockTeam()
def __init__(self):
self.litellm_teamtable = self.TeamTable()
def __init__(self):
self.db = self.DB()
prisma = MockPrisma()
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="member_1")
allowed = await spend_management_endpoints._can_team_member_view_log(
prisma, auth, "team_abc"
)
assert allowed is False
@pytest.mark.asyncio
async def test_ui_view_spend_logs_team_member_with_spend_logs_permission(
client, monkeypatch
):
"""
A non-admin team member with /spend/logs permission should see team-wide
spend logs when filtering by that team_id.
"""
mock_spend_logs = [
{
"id": "log1",
"request_id": "req1",
"api_key": "sk-key-1",
"user": "member_1",
"team_id": "team_perm",
"spend": 0.05,
"startTime": datetime.datetime.now(timezone.utc).isoformat(),
"model": "gpt-4",
},
{
"id": "log2",
"request_id": "req2",
"api_key": "sk-key-2",
"user": "member_2",
"team_id": "team_perm",
"spend": 0.10,
"startTime": datetime.datetime.now(timezone.utc).isoformat(),
"model": "gpt-4",
},
]
def filter_by_team(where):
if "team_id" in where and where["team_id"] == "team_perm":
return mock_spend_logs
return []
class TeamTable:
team_id = "team_perm"
members_with_roles = [Member(user_id="member_1", role="user")]
team_member_permissions = ["/spend/logs"]
def model_dump(self):
return {
"team_id": self.team_id,
"members_with_roles": [{"user_id": "member_1", "role": "user"}],
"team_member_permissions": self.team_member_permissions,
}
async def team_lookup(where):
return TeamTable() if where == {"team_id": "team_perm"} else None
monkeypatch.setattr(
"litellm.proxy.proxy_server.prisma_client",
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_by_team, team_lookup),
)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="member_1"
)
try:
start_date, end_date = _default_date_range()
response = client.get(
"/spend/logs/ui",
params={
"team_id": "team_perm",
"start_date": start_date,
"end_date": end_date,
},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 200
data = response.json()
assert data["total"] == 2
assert len(data["data"]) == 2
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@pytest.mark.asyncio
async def test_ui_view_spend_logs_team_member_no_permission_blocked(
client, monkeypatch
):
"""
A non-admin team member WITHOUT /spend/logs permission should be
rejected when filtering by team_id.
"""
mock_spend_logs = [
{
"id": "log1",
"request_id": "req1",
"api_key": "sk-key-1",
"user": "member_1",
"team_id": "team_noperm",
"spend": 0.05,
"startTime": datetime.datetime.now(timezone.utc).isoformat(),
"model": "gpt-4",
},
]
def filter_fn(where):
return mock_spend_logs
class TeamTable:
team_id = "team_noperm"
members_with_roles = [Member(user_id="member_1", role="user")]
team_member_permissions = ["/key/info"]
def model_dump(self):
return {
"team_id": self.team_id,
"members_with_roles": [{"user_id": "member_1", "role": "user"}],
"team_member_permissions": self.team_member_permissions,
}
async def team_lookup(where):
return TeamTable() if where == {"team_id": "team_noperm"} else None
monkeypatch.setattr(
"litellm.proxy.proxy_server.prisma_client",
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn, team_lookup),
)
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
user_role=LitellmUserRoles.INTERNAL_USER, user_id="member_1"
)
try:
start_date, end_date = _default_date_range()
response = client.get(
"/spend/logs/ui",
params={
"team_id": "team_noperm",
"start_date": start_date,
"end_date": end_date,
},
headers={"Authorization": "Bearer sk-test"},
)
assert response.status_code == 403
finally:
app.dependency_overrides.pop(ps.user_api_key_auth, None)
@@ -74,5 +74,24 @@ describe("permission_definitions", () => {
expect(PERMISSION_DESCRIPTIONS["/team/daily/activity"]).toBeDefined();
expect(PERMISSION_DESCRIPTIONS["/team/daily/activity"]).toContain("team usage");
});
it("should include spend logs permission", () => {
expect(PERMISSION_DESCRIPTIONS["/spend/logs"]).toBeDefined();
expect(PERMISSION_DESCRIPTIONS["/spend/logs"]).toContain("spend logs");
});
});
describe("spend/logs permission", () => {
it("should return GET method for /spend/logs", () => {
expect(getMethodForEndpoint("/spend/logs")).toBe("GET");
});
it("should return correct info for /spend/logs permission", () => {
const result = getPermissionInfo("/spend/logs");
expect(result.method).toBe("GET");
expect(result.endpoint).toBe("/spend/logs");
expect(result.description).toBe(PERMISSION_DESCRIPTIONS["/spend/logs"]);
expect(result.route).toBe("/spend/logs");
});
});
});
@@ -22,13 +22,15 @@ export const PERMISSION_DESCRIPTIONS: Record<string, string> = {
"/key/unblock": "Member can unblock a virtual key belonging to this team",
"/team/daily/activity":
"Member can view all team usage data (not just their own)",
"/spend/logs":
"Member can view spend logs for the entire team (not just their own)",
};
/**
* Determines the HTTP method for a given permission endpoint
*/
export const getMethodForEndpoint = (endpoint: string): string => {
if (endpoint.includes("/info") || endpoint.includes("/list") || endpoint.includes("/activity")) {
if (endpoint.includes("/info") || endpoint.includes("/list") || endpoint.includes("/activity") || endpoint === "/spend/logs") {
return "GET";
}
return "POST";