Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 25 additions & 0 deletions litellm/proxy/_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,7 @@ class LiteLLMRoutes(enum.Enum):
"/organization/list",
"/team/available",
"/user/info",
"/v2/user/info",
"/model/info",
"/v1/model/info",
"/v2/model/info",
Expand Down Expand Up @@ -2552,6 +2553,30 @@ class UserInfoResponse(LiteLLMPydanticObjectBase):
teams: List


class UserInfoV2Response(LiteLLMPydanticObjectBase):
"""
Response model for GET /v2/user/info

Returns ONLY the user object - no keys, no teams objects.
This is a lightweight alternative to UserInfoResponse.
"""

user_id: str
user_email: Optional[str] = None
user_alias: Optional[str] = None
user_role: Optional[str] = None
spend: float = 0.0
max_budget: Optional[float] = None
models: List[str] = []
budget_duration: Optional[str] = None
budget_reset_at: Optional[datetime] = None
metadata: Optional[dict] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
sso_user_id: Optional[str] = None
teams: List[str] = [] # Just team IDs, not full team objects


class LiteLLM_Config(LiteLLMPydanticObjectBase):
param_name: str
param_value: Dict
Expand Down
3 changes: 3 additions & 0 deletions litellm/proxy/auth/route_checks.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,9 @@ def non_proxy_admin_allowed_routes_check(
user_id, valid_token.user_id
),
)
elif route == "/v2/user/info":
# handled by the endpoint itself (full RBAC in handler)
pass
elif route == "/model/info":
# /model/info just shows models user has access to
pass
Expand Down
12 changes: 12 additions & 0 deletions litellm/proxy/client/users.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,18 @@ def get_user(self, user_id: Optional[str] = None) -> Dict[str, Any]:
response.raise_for_status()
return response.json()

def get_user_v2(self, user_id: Optional[str] = None) -> Dict[str, Any]:
"""Get user info v2 - lightweight, returns only user object (GET /v2/user/info)"""
url = f"{self.base_url}/v2/user/info"
params = {"user_id": user_id} if user_id else {}
response = requests.get(url, headers=self._get_headers(), params=params)
if response.status_code == 401:
raise UnauthorizedError(response.text)
if response.status_code == 404:
raise NotFoundError(response.text)
response.raise_for_status()
return response.json()

def create_user(self, user_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create a new user (POST /user/new)"""
url = f"{self.base_url}/user/new"
Expand Down
165 changes: 164 additions & 1 deletion litellm/proxy/management_endpoints/internal_user_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,10 @@
get_daily_activity_aggregated,
)
from litellm.proxy.auth.auth_checks import get_team_object, get_user_object
from litellm.proxy.management_endpoints.common_utils import _user_has_admin_view
from litellm.proxy.management_endpoints.common_utils import (
_is_user_team_admin,
_user_has_admin_view,
)
from litellm.proxy.management_endpoints.key_management_endpoints import (
generate_key_helper_fn,
prepare_metadata_fields,
Expand Down Expand Up @@ -715,6 +718,166 @@ async def user_info(
raise handle_exception_on_proxy(e)


async def _check_user_info_v2_access(
user_api_key_dict: UserAPIKeyAuth,
target_user_id: str,
) -> Optional["LiteLLM_UserTable"]:
"""
Check if the caller is allowed to access the target user's info.

Returns the target user's DB row if access is allowed, None otherwise.
Returning the row avoids a redundant DB fetch in the caller.

Access rules:
1. Proxy admins / proxy admin viewers can access any user
2. User can access their own info
3. Team admins can access info of users in their teams

Raises on unexpected DB errors so they surface as 500s, not silent 404s.
"""
from litellm.proxy.proxy_server import prisma_client

if prisma_client is None:
return None
Comment on lines +738 to +741

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

prisma_client is None returns None, causing misleading 404

When prisma_client is None, _check_user_info_v2_access returns None. The caller (user_info_v2) treats None as "access denied / user not found" and raises a 404. However, user_info_v2 already guards against this with its own if prisma_client is None: raise HTTPException(500) check before calling this function.

The issue is that _check_user_info_v2_access is a standalone helper function. If it is ever called from a context that does not first check prisma_client, a DB-unavailable condition would silently produce a None return indistinguishable from a legitimate "access denied" result.

Consider raising an exception here to make the failure loud:

if prisma_client is None:
    raise HTTPException(
        status_code=500,
        detail=CommonProxyErrors.db_not_connected_error.value,
    )


# Helper: fetch the target user row (reused across branches)
async def _fetch_target_user():
return await prisma_client.db.litellm_usertable.find_unique(
where={"user_id": target_user_id}
)

# Rule 1: Proxy admins — fetch and return the target row directly
if _user_has_admin_view(user_api_key_dict):
return await _fetch_target_user()

# Rule 2: Self-lookup
if user_api_key_dict.user_id == target_user_id:
return await _fetch_target_user()

# Rule 3: Team admins can look up users in their teams
if user_api_key_dict.user_id is not None:
# Get caller's teams
caller_user = await prisma_client.db.litellm_usertable.find_unique(
where={"user_id": user_api_key_dict.user_id}
)
if caller_user is not None and caller_user.teams:
# Fetch the target user ONCE, before the loop
target_user = await _fetch_target_user()
if target_user is None:
return None

# Get all teams the caller belongs to
teams = await prisma_client.db.litellm_teamtable.find_many(
where={"team_id": {"in": caller_user.teams}}
)
for team in teams:
team_obj = LiteLLM_TeamTable(**team.model_dump())
if _is_user_team_admin(
user_api_key_dict=user_api_key_dict, team_obj=team_obj
):
# Check if target user is in this team
if team.team_id in (target_user.teams or []):
return target_user

return None


@router.get(
"/v2/user/info",
tags=["Internal User management"],
dependencies=[Depends(user_api_key_auth)],
response_model=UserInfoV2Response,
)
@management_endpoint_wrapper
async def user_info_v2(
request: Request,
user_id: Optional[str] = fastapi.Query(
default=None, description="User ID in the request parameters"
),
user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
):
"""
Lightweight endpoint to get user info. Returns only the user object — no keys, no teams objects.

This is the v2 replacement for /user/info, designed to avoid the "god endpoint" problem
where the old endpoint loaded all keys and teams into memory.

Access control:
- Proxy admins can query any user
- Team admins can query users within their teams
- Internal users can only query themselves (omit user_id or pass own)
- Returns 404 for non-existent users or unauthorized access

Example request:
```
curl -X GET 'http://localhost:4000/v2/user/info?user_id=user123' \\
--header 'Authorization: Bearer sk-1234'
```
"""
from litellm.proxy.proxy_server import prisma_client

try:
if prisma_client is None:
raise HTTPException(
status_code=500,
detail=CommonProxyErrors.db_not_connected_error.value,
)

# Handle URL encoding for + characters
if user_id is not None and " " in user_id:
user_id = get_user_id_from_request(request=request)

# Default to self-lookup if no user_id provided
if user_id is None:
user_id = user_api_key_dict.user_id

if user_id is None:
raise HTTPException(
status_code=400,
detail="user_id is required. Either pass it as a query parameter or authenticate with a user-bound key.",
)

# Check access — returns the user row if allowed, None otherwise.
# This avoids a redundant DB fetch since the access check already
# loads the target user for team-admin verification.
user_row = await _check_user_info_v2_access(
user_api_key_dict=user_api_key_dict,
target_user_id=user_id,
)

if user_row is None:
raise HTTPException(
status_code=404,
detail=f"User not found: {user_id}",
)

user_data = user_row.model_dump()

return UserInfoV2Response(
user_id=user_data.get("user_id", user_id),
user_email=user_data.get("user_email"),
user_alias=user_data.get("user_alias"),
user_role=user_data.get("user_role"),
spend=user_data.get("spend", 0.0),
max_budget=user_data.get("max_budget"),
models=user_data.get("models") or [],
budget_duration=user_data.get("budget_duration"),
budget_reset_at=user_data.get("budget_reset_at"),
metadata=user_data.get("metadata"),
created_at=user_data.get("created_at"),
updated_at=user_data.get("updated_at"),
sso_user_id=user_data.get("sso_user_id"),
teams=user_data.get("teams") or [],
)
except Exception as e:
verbose_proxy_logger.exception(
"litellm.proxy.proxy_server.user_info_v2(): Exception occured - {}".format(
str(e)
)
)
raise handle_exception_on_proxy(e)


async def _get_user_info_for_proxy_admin(user_api_key_dict: UserAPIKeyAuth):
"""
Admin UI Endpoint - Returns All Teams and Keys when Proxy Admin is querying
Expand Down
27 changes: 27 additions & 0 deletions tests/test_litellm/proxy/auth/test_info_routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -117,3 +117,30 @@ def test_team_info_route_access():
valid_token=valid_token,
request_data={},
)


def test_v2_user_info_route_in_info_routes():
"""Test that /v2/user/info is in the info_routes list"""
assert "/v2/user/info" in LiteLLMRoutes.info_routes.value


def test_v2_user_info_route_access():
"""Test access control for /v2/user/info route - handled by endpoint itself"""
user_obj = LiteLLM_UserTable(
user_id="test_user",
user_email="test@example.com",
user_role=LitellmUserRoles.INTERNAL_USER,
)
valid_token = UserAPIKeyAuth(user_id="test_user")
request = MagicMock(spec=Request)
request.query_params = {"user_id": "other_user"}

# Should not raise exception as /v2/user/info handles its own RBAC logic in the handler
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=LitellmUserRoles.INTERNAL_USER,
route="/v2/user/info",
request=request,
valid_token=valid_token,
request_data={},
)
Loading
Loading