mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-07 02:59:05 +00:00
Add team member visibility permissions for spend logs
Register /team/member/view_all_keys and /team/member/view_all_logs permission strings. Refactor _can_team_member_view_log() to use cached get_team_object() and check team admin, member permission, and org admin. Members without the permission now see their own logs within the team instead of getting 403. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
yuneng-jiang
parent
1c0c504877
commit
8f696a6071
3 changed files with 534 additions and 86 deletions
|
|
@ -245,6 +245,10 @@ class KeyManagementRoutes(str, enum.Enum):
|
|||
# team usage routes
|
||||
TEAM_DAILY_ACTIVITY = "/team/daily/activity"
|
||||
|
||||
# team member visibility routes
|
||||
TEAM_MEMBER_VIEW_ALL_KEYS = "/team/member/view_all_keys"
|
||||
TEAM_MEMBER_VIEW_ALL_LOGS = "/team/member/view_all_logs"
|
||||
|
||||
|
||||
class LiteLLMRoutes(enum.Enum):
|
||||
openai_route_names = [
|
||||
|
|
@ -516,6 +520,8 @@ class LiteLLMRoutes(enum.Enum):
|
|||
KeyManagementRoutes.KEY_UNBLOCK.value,
|
||||
KeyManagementRoutes.KEY_BULK_UPDATE.value,
|
||||
KeyManagementRoutes.TEAM_DAILY_ACTIVITY.value,
|
||||
KeyManagementRoutes.TEAM_MEMBER_VIEW_ALL_KEYS.value,
|
||||
KeyManagementRoutes.TEAM_MEMBER_VIEW_ALL_LOGS.value,
|
||||
KeyManagementRoutes.KEY_RESET_SPEND.value,
|
||||
KeyManagementRoutes.KEY_ALIASES.value,
|
||||
]
|
||||
|
|
@ -649,6 +655,8 @@ class LiteLLMRoutes(enum.Enum):
|
|||
"/team/permissions_list",
|
||||
"/team/permissions_update",
|
||||
"/team/daily/activity",
|
||||
"/team/member/view_all_keys",
|
||||
"/team/member/view_all_logs",
|
||||
"/model/new",
|
||||
"/model/update",
|
||||
"/model/delete",
|
||||
|
|
|
|||
|
|
@ -14,7 +14,9 @@ from litellm.proxy._types import *
|
|||
from litellm.proxy._types import ProviderBudgetResponse, ProviderBudgetResponseObject
|
||||
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
|
||||
from litellm.proxy.management_endpoints.common_utils import (
|
||||
_is_user_org_admin_for_team,
|
||||
_is_user_team_admin,
|
||||
_team_member_has_permission,
|
||||
_user_has_admin_view,
|
||||
)
|
||||
from litellm.proxy.spend_tracking.spend_tracking_utils import (
|
||||
|
|
@ -1842,12 +1844,20 @@ async def ui_view_spend_logs( # noqa: PLR0915
|
|||
is_admin_view = _is_admin_view_safe(user_api_key_dict=user_api_key_dict)
|
||||
if not is_admin_view:
|
||||
if team_id is not None:
|
||||
can_view_team = await _can_team_member_view_log(
|
||||
view_result = await _can_team_member_view_log(
|
||||
prisma_client=prisma_client,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
team_id=team_id,
|
||||
)
|
||||
if not can_view_team:
|
||||
if view_result.can_view_all:
|
||||
# Admin / permission holder / org admin → see all team logs
|
||||
where_conditions["team_id"] = team_id
|
||||
elif view_result.is_member:
|
||||
# Member without permission → see only own logs in team
|
||||
where_conditions["team_id"] = team_id
|
||||
where_conditions["user"] = user_api_key_dict.user_id
|
||||
else:
|
||||
# Not a member of this team → 403
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail={
|
||||
|
|
@ -1856,7 +1866,6 @@ async def ui_view_spend_logs( # noqa: PLR0915
|
|||
)
|
||||
},
|
||||
)
|
||||
where_conditions["team_id"] = team_id
|
||||
else:
|
||||
if _can_user_view_spend_log(user_api_key_dict=user_api_key_dict):
|
||||
where_conditions["user"] = user_api_key_dict.user_id
|
||||
|
|
@ -3368,23 +3377,89 @@ def _is_admin_view_safe(user_api_key_dict: UserAPIKeyAuth) -> bool:
|
|||
return False
|
||||
|
||||
|
||||
class _TeamLogViewResult:
|
||||
"""Result of checking a user's log-viewing permissions for a team."""
|
||||
|
||||
__slots__ = ("can_view_all", "is_member", "team_obj")
|
||||
|
||||
def __init__(self, can_view_all: bool, is_member: bool, team_obj: Optional[Any]):
|
||||
self.can_view_all = can_view_all
|
||||
self.is_member = is_member
|
||||
self.team_obj = team_obj
|
||||
|
||||
|
||||
def _is_user_member_of_team(
|
||||
user_api_key_dict: UserAPIKeyAuth, team_obj: Any
|
||||
) -> bool:
|
||||
"""Check if the user appears in the team's members_with_roles list."""
|
||||
for member in team_obj.members_with_roles:
|
||||
if member.user_id is not None and member.user_id == user_api_key_dict.user_id:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
async def _can_team_member_view_log(
|
||||
prisma_client,
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
team_id: Optional[str],
|
||||
) -> bool:
|
||||
) -> _TeamLogViewResult:
|
||||
"""
|
||||
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 a _TeamLogViewResult with:
|
||||
- can_view_all: True if the user can see ALL team logs (team admin,
|
||||
has /team/member/view_all_logs permission, or is org admin).
|
||||
- is_member: True if the user is a member of the team (any role).
|
||||
- team_obj: The fetched team object, or None if the team was not found.
|
||||
"""
|
||||
if team_id is None:
|
||||
return False
|
||||
team_obj = await prisma_client.db.litellm_teamtable.find_unique(
|
||||
where={"team_id": team_id}
|
||||
return _TeamLogViewResult(can_view_all=False, is_member=False, team_obj=None)
|
||||
|
||||
from litellm.proxy.proxy_server import proxy_logging_obj, user_api_key_cache
|
||||
|
||||
try:
|
||||
from litellm.proxy.auth.auth_checks import get_team_object
|
||||
|
||||
team_obj = await get_team_object(
|
||||
team_id=team_id,
|
||||
prisma_client=prisma_client,
|
||||
user_api_key_cache=user_api_key_cache,
|
||||
proxy_logging_obj=proxy_logging_obj,
|
||||
)
|
||||
except HTTPException:
|
||||
return _TeamLogViewResult(can_view_all=False, is_member=False, team_obj=None)
|
||||
|
||||
is_member = _is_user_member_of_team(
|
||||
user_api_key_dict=user_api_key_dict, team_obj=team_obj
|
||||
)
|
||||
|
||||
# Team admin → can view all
|
||||
if _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj):
|
||||
return _TeamLogViewResult(
|
||||
can_view_all=True, is_member=True, team_obj=team_obj
|
||||
)
|
||||
|
||||
# Member with explicit permission → can view all
|
||||
if is_member and _team_member_has_permission(
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
team_obj=team_obj,
|
||||
permission="/team/member/view_all_logs",
|
||||
):
|
||||
return _TeamLogViewResult(
|
||||
can_view_all=True, is_member=True, team_obj=team_obj
|
||||
)
|
||||
|
||||
# Org admin for the team's org → can view all
|
||||
if await _is_user_org_admin_for_team(
|
||||
user_api_key_dict=user_api_key_dict, team_obj=team_obj
|
||||
):
|
||||
return _TeamLogViewResult(
|
||||
can_view_all=True, is_member=is_member, team_obj=team_obj
|
||||
)
|
||||
|
||||
return _TeamLogViewResult(
|
||||
can_view_all=False, is_member=is_member, team_obj=team_obj
|
||||
)
|
||||
if team_obj is None:
|
||||
return False
|
||||
return _is_user_team_admin(user_api_key_dict=user_api_key_dict, team_obj=team_obj)
|
||||
|
||||
|
||||
def _can_user_view_spend_log(user_api_key_dict: UserAPIKeyAuth) -> bool:
|
||||
|
|
|
|||
|
|
@ -140,110 +140,190 @@ async def test_is_admin_view_safe_exception(monkeypatch):
|
|||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_team_member_view_log_none_team_id():
|
||||
# team_id=None should immediately return False
|
||||
class MockPrisma:
|
||||
class DB:
|
||||
class TeamTable:
|
||||
async def find_unique(self, where: dict):
|
||||
return None
|
||||
|
||||
def __init__(self):
|
||||
self.litellm_teamtable = self.TeamTable()
|
||||
|
||||
def __init__(self):
|
||||
self.db = self.DB()
|
||||
|
||||
prisma = MockPrisma()
|
||||
# team_id=None should immediately return result with all False
|
||||
prisma = MagicMock()
|
||||
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
|
||||
allowed = await spend_management_endpoints._can_team_member_view_log(
|
||||
result = await spend_management_endpoints._can_team_member_view_log(
|
||||
prisma, auth, None
|
||||
)
|
||||
assert allowed is False
|
||||
assert result.can_view_all is False
|
||||
assert result.is_member is False
|
||||
assert result.team_obj is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_team_member_view_log_team_not_found(monkeypatch):
|
||||
# Non-existent team should return False
|
||||
class MockPrisma:
|
||||
class DB:
|
||||
class TeamTable:
|
||||
async def find_unique(self, where: dict):
|
||||
return None
|
||||
# Non-existent team should return all False
|
||||
from fastapi import HTTPException as _HTTPException
|
||||
|
||||
def __init__(self):
|
||||
self.litellm_teamtable = self.TeamTable()
|
||||
async def mock_get_team_object(**kwargs):
|
||||
raise _HTTPException(status_code=404, detail="Team not found")
|
||||
|
||||
def __init__(self):
|
||||
self.db = self.DB()
|
||||
|
||||
prisma = MockPrisma()
|
||||
# Even if admin check would return True, no team means False
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_team_admin", lambda user_api_key_dict, team_obj: True
|
||||
"litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object
|
||||
)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
prisma = MagicMock()
|
||||
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
|
||||
allowed = await spend_management_endpoints._can_team_member_view_log(
|
||||
result = await spend_management_endpoints._can_team_member_view_log(
|
||||
prisma, auth, "team_x"
|
||||
)
|
||||
assert allowed is False
|
||||
assert result.can_view_all is False
|
||||
assert result.is_member is False
|
||||
assert result.team_obj is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_team_member_view_log_not_admin(monkeypatch):
|
||||
# Existing team but caller is not a team admin -> False
|
||||
class MockTeam:
|
||||
pass
|
||||
# Existing team, user is member but not admin, no permission → can_view_all=False, is_member=True
|
||||
mock_team = MagicMock()
|
||||
mock_team.members_with_roles = [Member(user_id="user_1", role="user")]
|
||||
mock_team.team_member_permissions = []
|
||||
mock_team.organization_id = None
|
||||
|
||||
class MockPrisma:
|
||||
class DB:
|
||||
class TeamTable:
|
||||
async def find_unique(self, where: dict):
|
||||
return MockTeam()
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
def __init__(self):
|
||||
self.litellm_teamtable = self.TeamTable()
|
||||
|
||||
def __init__(self):
|
||||
self.db = self.DB()
|
||||
|
||||
prisma = MockPrisma()
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_team_admin", lambda user_api_key_dict, team_obj: False
|
||||
"litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object
|
||||
)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
async def mock_org_admin(user_api_key_dict, team_obj):
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_org_admin_for_team", mock_org_admin
|
||||
)
|
||||
|
||||
prisma = MagicMock()
|
||||
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
|
||||
allowed = await spend_management_endpoints._can_team_member_view_log(
|
||||
result = await spend_management_endpoints._can_team_member_view_log(
|
||||
prisma, auth, "team_x"
|
||||
)
|
||||
assert allowed is False
|
||||
assert result.can_view_all is False
|
||||
assert result.is_member is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_team_member_view_log_admin(monkeypatch):
|
||||
# Existing team and caller is team admin -> True
|
||||
class MockTeam:
|
||||
pass
|
||||
# Existing team and caller is team admin → can_view_all=True, is_member=True
|
||||
mock_team = MagicMock()
|
||||
mock_team.members_with_roles = [Member(user_id="user_1", role="admin")]
|
||||
mock_team.team_member_permissions = []
|
||||
mock_team.organization_id = None
|
||||
|
||||
class MockPrisma:
|
||||
class DB:
|
||||
class TeamTable:
|
||||
async def find_unique(self, where: dict):
|
||||
return MockTeam()
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
def __init__(self):
|
||||
self.litellm_teamtable = self.TeamTable()
|
||||
|
||||
def __init__(self):
|
||||
self.db = self.DB()
|
||||
|
||||
prisma = MockPrisma()
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_team_admin", lambda user_api_key_dict, team_obj: True
|
||||
"litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object
|
||||
)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
prisma = MagicMock()
|
||||
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
|
||||
allowed = await spend_management_endpoints._can_team_member_view_log(
|
||||
result = await spend_management_endpoints._can_team_member_view_log(
|
||||
prisma, auth, "team_x"
|
||||
)
|
||||
assert allowed is True
|
||||
assert result.can_view_all is True
|
||||
assert result.is_member is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_team_member_view_log_member_with_permission(monkeypatch):
|
||||
"""Member with /team/member/view_all_logs permission → can_view_all=True."""
|
||||
mock_team = MagicMock()
|
||||
mock_team.members_with_roles = [Member(user_id="user_1", role="user")]
|
||||
mock_team.team_member_permissions = ["/team/member/view_all_logs"]
|
||||
mock_team.organization_id = None
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object
|
||||
)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
prisma = MagicMock()
|
||||
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
|
||||
result = await spend_management_endpoints._can_team_member_view_log(
|
||||
prisma, auth, "team_x"
|
||||
)
|
||||
assert result.can_view_all is True
|
||||
assert result.is_member is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_team_member_view_log_org_admin(monkeypatch):
|
||||
"""Org admin (not a team member) → can_view_all=True, is_member=False."""
|
||||
mock_team = MagicMock()
|
||||
mock_team.members_with_roles = [Member(user_id="other_user", role="admin")]
|
||||
mock_team.team_member_permissions = []
|
||||
mock_team.organization_id = "org_1"
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object
|
||||
)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
async def mock_org_admin(user_api_key_dict, team_obj):
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_org_admin_for_team", mock_org_admin
|
||||
)
|
||||
|
||||
prisma = MagicMock()
|
||||
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
|
||||
result = await spend_management_endpoints._can_team_member_view_log(
|
||||
prisma, auth, "team_x"
|
||||
)
|
||||
assert result.can_view_all is True
|
||||
assert result.is_member is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_can_team_member_view_log_non_member(monkeypatch):
|
||||
"""User not in team members → can_view_all=False, is_member=False."""
|
||||
mock_team = MagicMock()
|
||||
mock_team.members_with_roles = [Member(user_id="other_user", role="admin")]
|
||||
mock_team.team_member_permissions = []
|
||||
mock_team.organization_id = None
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object
|
||||
)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
async def mock_org_admin(user_api_key_dict, team_obj):
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_org_admin_for_team", mock_org_admin
|
||||
)
|
||||
|
||||
prisma = MagicMock()
|
||||
auth = UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER, user_id="user_1")
|
||||
result = await spend_management_endpoints._can_team_member_view_log(
|
||||
prisma, auth, "team_x"
|
||||
)
|
||||
assert result.can_view_all is False
|
||||
assert result.is_member is False
|
||||
|
||||
|
||||
def test_can_user_view_spend_log_true_for_internal_user():
|
||||
|
|
@ -797,6 +877,7 @@ async def test_ui_view_spend_logs_internal_user_scoped_without_user_id(client, m
|
|||
async def test_ui_view_spend_logs_team_admin_can_view_team_spend(client, monkeypatch):
|
||||
"""
|
||||
Team admins should be able to view team-wide spend when team_id is provided.
|
||||
Uses get_team_object (cached) instead of raw DB query.
|
||||
"""
|
||||
mock_spend_logs = [
|
||||
{"id": "log1", "request_id": "req1", "api_key": "sk-test-key", "user": "member1", "team_id": "team_admin_team", "spend": 0.05, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-3.5-turbo"},
|
||||
|
|
@ -808,16 +889,27 @@ async def test_ui_view_spend_logs_team_admin_can_view_team_spend(client, monkeyp
|
|||
return [mock_spend_logs[0]]
|
||||
return mock_spend_logs
|
||||
|
||||
class TeamTable:
|
||||
members_with_roles = [Member(user_id="admin_user", role="admin")]
|
||||
|
||||
async def team_lookup(where):
|
||||
return TeamTable() if where == {"team_id": "team_admin_team"} else None
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.proxy_server.prisma_client",
|
||||
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_by_team, team_lookup),
|
||||
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_by_team),
|
||||
)
|
||||
|
||||
# Mock get_team_object to return a team where the user is admin
|
||||
mock_team = MagicMock()
|
||||
mock_team.members_with_roles = [Member(user_id="admin_user", role="admin")]
|
||||
mock_team.team_member_permissions = []
|
||||
mock_team.organization_id = None
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
from fastapi import HTTPException as _HTTPException
|
||||
if kwargs.get("team_id") == "team_admin_team":
|
||||
return mock_team
|
||||
raise _HTTPException(status_code=404, detail="Team not found")
|
||||
|
||||
monkeypatch.setattr("litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER, user_id="admin_user"
|
||||
)
|
||||
|
|
@ -2403,3 +2495,276 @@ async def test_build_ui_spend_logs_response_dict_rows_session_counts():
|
|||
where={"session_id": {"in": [session_id]}},
|
||||
count={"session_id": True},
|
||||
)
|
||||
|
||||
|
||||
# ── Endpoint-level tests for team visibility permissions ──
|
||||
|
||||
|
||||
def _mock_team_with_members(members, permissions=None, organization_id=None):
|
||||
"""Helper to create a mock team object for visibility permission tests."""
|
||||
team = MagicMock()
|
||||
team.members_with_roles = members
|
||||
team.team_member_permissions = permissions or []
|
||||
team.organization_id = organization_id
|
||||
return team
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spend_logs_member_with_view_all_logs_sees_team_logs(client, monkeypatch):
|
||||
"""Member with /team/member/view_all_logs permission sees all team logs (no user filter)."""
|
||||
mock_spend_logs = [
|
||||
{"id": "log1", "request_id": "req1", "api_key": "sk-key1", "user": "member1", "team_id": "team1", "spend": 0.05, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
{"id": "log2", "request_id": "req2", "api_key": "sk-key2", "user": "member2", "team_id": "team1", "spend": 0.10, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
]
|
||||
|
||||
def filter_fn(where):
|
||||
result = mock_spend_logs
|
||||
if "team_id" in where:
|
||||
result = [l for l in result if l["team_id"] == where["team_id"]]
|
||||
if "user" in where:
|
||||
result = [l for l in result if l["user"] == where["user"]]
|
||||
return result
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.proxy_server.prisma_client",
|
||||
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn),
|
||||
)
|
||||
|
||||
mock_team = _mock_team_with_members(
|
||||
members=[Member(user_id="member1", role="user")],
|
||||
permissions=["/team/member/view_all_logs"],
|
||||
)
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
monkeypatch.setattr("litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER, user_id="member1"
|
||||
)
|
||||
|
||||
try:
|
||||
start_date, end_date = _default_date_range()
|
||||
response = client.get(
|
||||
"/spend/logs/ui",
|
||||
params={"team_id": "team1", "start_date": start_date, "end_date": end_date},
|
||||
headers={"Authorization": "Bearer sk-test"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
# Should see both logs (no user filter applied)
|
||||
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_spend_logs_member_without_permission_sees_own_logs_in_team(client, monkeypatch):
|
||||
"""Member without permission sees only own logs within the team (team_id + user_id filter)."""
|
||||
mock_spend_logs = [
|
||||
{"id": "log1", "request_id": "req1", "api_key": "sk-key1", "user": "member1", "team_id": "team1", "spend": 0.05, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
{"id": "log2", "request_id": "req2", "api_key": "sk-key2", "user": "member2", "team_id": "team1", "spend": 0.10, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
]
|
||||
|
||||
def filter_fn(where):
|
||||
result = mock_spend_logs
|
||||
if "team_id" in where:
|
||||
result = [l for l in result if l["team_id"] == where["team_id"]]
|
||||
if "user" in where:
|
||||
result = [l for l in result if l["user"] == where["user"]]
|
||||
return result
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.proxy_server.prisma_client",
|
||||
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn),
|
||||
)
|
||||
|
||||
# Team with no view_all_logs permission
|
||||
mock_team = _mock_team_with_members(
|
||||
members=[Member(user_id="member1", role="user"), Member(user_id="member2", role="user")],
|
||||
)
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
monkeypatch.setattr("litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
async def mock_org_admin(user_api_key_dict, team_obj):
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_org_admin_for_team", mock_org_admin
|
||||
)
|
||||
|
||||
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER, user_id="member1"
|
||||
)
|
||||
|
||||
try:
|
||||
start_date, end_date = _default_date_range()
|
||||
response = client.get(
|
||||
"/spend/logs/ui",
|
||||
params={"team_id": "team1", "start_date": start_date, "end_date": end_date},
|
||||
headers={"Authorization": "Bearer sk-test"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
# Should see only own log (user filter applied)
|
||||
assert data["total"] == 1
|
||||
assert len(data["data"]) == 1
|
||||
assert data["data"][0]["user"] == "member1"
|
||||
finally:
|
||||
app.dependency_overrides.pop(ps.user_api_key_auth, None)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spend_logs_org_admin_sees_team_logs(client, monkeypatch):
|
||||
"""Org admin for the team's org sees all team logs."""
|
||||
mock_spend_logs = [
|
||||
{"id": "log1", "request_id": "req1", "api_key": "sk-key1", "user": "member1", "team_id": "team1", "spend": 0.05, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
{"id": "log2", "request_id": "req2", "api_key": "sk-key2", "user": "member2", "team_id": "team1", "spend": 0.10, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
]
|
||||
|
||||
def filter_fn(where):
|
||||
result = mock_spend_logs
|
||||
if "team_id" in where:
|
||||
result = [l for l in result if l["team_id"] == where["team_id"]]
|
||||
if "user" in where:
|
||||
result = [l for l in result if l["user"] == where["user"]]
|
||||
return result
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.proxy_server.prisma_client",
|
||||
make_ui_spend_logs_mock_prisma(mock_spend_logs, filter_fn),
|
||||
)
|
||||
|
||||
# Org admin is NOT a team member
|
||||
mock_team = _mock_team_with_members(
|
||||
members=[Member(user_id="member1", role="user")],
|
||||
)
|
||||
mock_team.organization_id = "org_1"
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
monkeypatch.setattr("litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
async def mock_org_admin(user_api_key_dict, team_obj):
|
||||
return True
|
||||
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_org_admin_for_team", mock_org_admin
|
||||
)
|
||||
|
||||
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER, user_id="org_admin_user"
|
||||
)
|
||||
|
||||
try:
|
||||
start_date, end_date = _default_date_range()
|
||||
response = client.get(
|
||||
"/spend/logs/ui",
|
||||
params={"team_id": "team1", "start_date": start_date, "end_date": end_date},
|
||||
headers={"Authorization": "Bearer sk-test"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
# Should see all team logs (no user filter)
|
||||
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_spend_logs_non_member_gets_403(client, monkeypatch):
|
||||
"""User not in team at all gets 403."""
|
||||
mock_spend_logs = [
|
||||
{"id": "log1", "request_id": "req1", "api_key": "sk-key1", "user": "member1", "team_id": "team1", "spend": 0.05, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
]
|
||||
|
||||
monkeypatch.setattr(
|
||||
"litellm.proxy.proxy_server.prisma_client",
|
||||
make_ui_spend_logs_mock_prisma(mock_spend_logs, lambda where: mock_spend_logs),
|
||||
)
|
||||
|
||||
# User "outsider" is not in the team
|
||||
mock_team = _mock_team_with_members(
|
||||
members=[Member(user_id="member1", role="user")],
|
||||
)
|
||||
|
||||
async def mock_get_team_object(**kwargs):
|
||||
return mock_team
|
||||
|
||||
monkeypatch.setattr("litellm.proxy.auth.auth_checks.get_team_object", mock_get_team_object)
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.user_api_key_cache", MagicMock())
|
||||
monkeypatch.setattr("litellm.proxy.proxy_server.proxy_logging_obj", MagicMock())
|
||||
|
||||
async def mock_org_admin(user_api_key_dict, team_obj):
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(
|
||||
spend_management_endpoints, "_is_user_org_admin_for_team", mock_org_admin
|
||||
)
|
||||
|
||||
app.dependency_overrides[ps.user_api_key_auth] = lambda: UserAPIKeyAuth(
|
||||
user_role=LitellmUserRoles.INTERNAL_USER, user_id="outsider"
|
||||
)
|
||||
|
||||
try:
|
||||
start_date, end_date = _default_date_range()
|
||||
response = client.get(
|
||||
"/spend/logs/ui",
|
||||
params={"team_id": "team1", "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)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_spend_logs_no_team_id_shows_own_logs(client, monkeypatch):
|
||||
"""Without team_id, internal user sees only own logs (unchanged behavior)."""
|
||||
mock_spend_logs = [
|
||||
{"id": "log1", "request_id": "req1", "api_key": "sk-key1", "user": "user_1", "team_id": "team1", "spend": 0.05, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
{"id": "log2", "request_id": "req2", "api_key": "sk-key2", "user": "user_2", "team_id": "team1", "spend": 0.10, "startTime": datetime.datetime.now(timezone.utc).isoformat(), "model": "gpt-4"},
|
||||
]
|
||||
|
||||
def filter_fn(where):
|
||||
result = mock_spend_logs
|
||||
if "user" in where:
|
||||
result = [l for l in result if l["user"] == where["user"]]
|
||||
return result
|
||||
|
||||
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.INTERNAL_USER, user_id="user_1"
|
||||
)
|
||||
|
||||
try:
|
||||
start_date, end_date = _default_date_range()
|
||||
response = client.get(
|
||||
"/spend/logs/ui",
|
||||
params={"start_date": start_date, "end_date": end_date},
|
||||
headers={"Authorization": "Bearer sk-test"},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["total"] == 1
|
||||
assert len(data["data"]) == 1
|
||||
assert data["data"][0]["user"] == "user_1"
|
||||
finally:
|
||||
app.dependency_overrides.pop(ps.user_api_key_auth, None)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue