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
35 changes: 35 additions & 0 deletions litellm/proxy/management_endpoints/internal_user_endpoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -618,6 +618,40 @@ def _normalize_user_info_user_id(
return user_id


def _enforce_user_info_access(
user_id: Optional[str], user_api_key_dict: UserAPIKeyAuth
) -> None:
"""Re-validate that the caller may read the resolved ``user_id`` after
URL-decoding has been finalized.

The route-level check in ``RouteChecks.non_proxy_admin_allowed_routes_check``
runs against ``request.query_params``, which decodes a literal ``+`` to a
space. ``_normalize_user_info_user_id`` then re-parses the raw query with
``unquote`` so the endpoint can return rows for user_ids that contain ``+``
(e.g. plus-addressed emails). That asymmetry let an attacker who registered
a username with a literal space pass the route check and then read another
user's row by sending the encoded ``+`` form. Re-checking ownership here
closes the gap without changing the supported user_id grammar.
"""
if user_id is None:
return
# Only true proxy admin bypasses ownership. PROXY_ADMIN_VIEW_ONLY is
# subject to the same `user_id == valid_token.user_id` rule that
# `RouteChecks.non_proxy_admin_allowed_routes_check` applies upstream
# for the `/user/info` route.
if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN:
return
if user_id == user_api_key_dict.user_id:
return
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=(
f"key not allowed to access this user's info. user_id={user_id}, "
f"key's user_id={user_api_key_dict.user_id}"
),
)


async def _get_user_info_teams(
prisma_client: Any,
user_id: Optional[str],
Expand Down Expand Up @@ -732,6 +766,7 @@ async def user_info( # noqa: PLR0915

try:
user_id = _normalize_user_info_user_id(request=request, user_id=user_id)
_enforce_user_info_access(user_id=user_id, user_api_key_dict=user_api_key_dict)

if prisma_client is None:
raise Exception(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -1949,11 +1949,13 @@ async def mock_find_memberships(*args, **kwargs):
assert exc.value.status_code == 403

# Critical: no delete_many calls should have executed.
assert not hasattr(
mock_prisma_client.db.litellm_verificationtoken.delete_many, "mock_calls"
) or len(
mock_prisma_client.db.litellm_verificationtoken.delete_many.mock_calls
) == 0
assert (
not hasattr(
mock_prisma_client.db.litellm_verificationtoken.delete_many, "mock_calls"
)
or len(mock_prisma_client.db.litellm_verificationtoken.delete_many.mock_calls)
== 0
)


@pytest.mark.asyncio
Expand Down Expand Up @@ -2631,3 +2633,107 @@ def test_allows_512_char_user_id(self):
request = self._make_request(f"user_id={exact_id}")
result = get_user_id_from_request(request)
assert result == exact_id


# ---------------------------------------------------------------------------
# VERIA-60: /user/info post-decode re-authorization
# ---------------------------------------------------------------------------


def test_enforce_user_info_access_admin_bypass():
"""Proxy admins must always be allowed past the re-check."""
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)

admin = UserAPIKeyAuth(
user_id="admin", user_role=LitellmUserRoles.PROXY_ADMIN.value
)
# Should not raise even when querying a different user
_enforce_user_info_access(user_id="someone_else", user_api_key_dict=admin)


def test_enforce_user_info_access_view_only_admin_blocked_from_other_users():
"""PROXY_ADMIN_VIEW_ONLY is not a true admin for /user/info — the upstream
route check applies the same `user_id == valid_token.user_id` rule, so the
re-check here must mirror that and deny cross-user lookups."""
import pytest
from fastapi import HTTPException

from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)

viewer = UserAPIKeyAuth(
user_id="viewer",
user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value,
)
with pytest.raises(HTTPException) as exc_info:
_enforce_user_info_access(user_id="someone_else", user_api_key_dict=viewer)
assert exc_info.value.status_code == 403


def test_enforce_user_info_access_view_only_admin_can_read_own():
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)

viewer = UserAPIKeyAuth(
user_id="viewer",
user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value,
)
_enforce_user_info_access(user_id="viewer", user_api_key_dict=viewer)


def test_enforce_user_info_access_owner_allowed():
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)

user = UserAPIKeyAuth(
user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER.value
)
_enforce_user_info_access(user_id="alice", user_api_key_dict=user)


def test_enforce_user_info_access_no_user_id_allowed():
"""No user_id in query → handler resolves to caller's own id later, so
this branch must not raise."""
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)

user = UserAPIKeyAuth(
user_id="alice", user_role=LitellmUserRoles.INTERNAL_USER.value
)
_enforce_user_info_access(user_id=None, user_api_key_dict=user)


def test_enforce_user_info_access_blocks_cross_user_lookup():
"""A non-admin caller may not query another user's row, even if URL
re-parsing produced a user_id that differs from the one the route check
saw (the VERIA-60 bypass)."""
import pytest
from fastapi import HTTPException

from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy.management_endpoints.internal_user_endpoints import (
_enforce_user_info_access,
)

attacker = UserAPIKeyAuth(
user_id="attacker space", # original (URL-decoded) id seen by route check
user_role=LitellmUserRoles.INTERNAL_USER.value,
)

with pytest.raises(HTTPException) as exc_info:
# Re-parsed id (with literal '+') belongs to the victim
_enforce_user_info_access(user_id="victim+target", user_api_key_dict=attacker)

assert exc_info.value.status_code == 403
assert "key not allowed to access this user's info" in str(exc_info.value.detail)
Loading