fix(scim): clamp collection page size

litellm/proxy/_lazy_openapi_snapshot.json

@@ -38680,8 +38680,7 @@
               "required": false,
               "schema": {
                 "default": 10,
-                "maximum": 100,
-                "minimum": 1,
+                "minimum": 0,
                 "title": "Count",
                 "type": "integer"
               }
@@ -39385,8 +39384,7 @@
               "required": false,
               "schema": {
                 "default": 10,
-                "maximum": 100,
-                "minimum": 1,
+                "minimum": 0,
                 "title": "Count",
                 "type": "integer"
               }

litellm/proxy/management_endpoints/scim/scim_v2.py

@@ -264,6 +264,8 @@ scim_router: Final = APIRouter(
     dependencies=[Depends(_premium_user_check)],
 )
 
+SCIM_MAX_PAGE_SIZE: Final = 100
+
 
 # Helper functions for common operations
 async def _get_prisma_client_or_raise_exception():
@@ -1572,12 +1574,13 @@ def _parse_scim_eq_filter(scim_filter: str) -> tuple[str, str] | None:
 )
 async def get_users(
     startIndex: int = Query(1, ge=1),
-    count: int = Query(10, ge=1, le=100),
+    count: int = Query(10, ge=0),
     filter: str | None = Query(None),
 ):
     """
     Get a list of users according to SCIM v2 protocol
     """
+    page_size: Final = min(count, SCIM_MAX_PAGE_SIZE)
     verbose_proxy_logger.debug(
         "SCIM GET USERS request: startIndex=%s count=%s filter=%s",
         startIndex,
@@ -1607,7 +1610,7 @@ async def get_users(
         users: Final[Sequence[LiteLLM_UserTable]] = await _table(UserRepository(prisma_client)).find_many(
             where=where_conditions,
             skip=(startIndex - 1),
-            take=count,
+            take=page_size,
             order={"created_at": "desc"},
         )
 
@@ -1623,7 +1626,7 @@ async def get_users(
         return SCIMListResponse(
             totalResults=total_count,
             startIndex=startIndex,
-            itemsPerPage=min(count, len(scim_users)),
+            itemsPerPage=len(scim_users),
             Resources=scim_users,
         )
 
@@ -2399,12 +2402,13 @@ class _TeamWhereConditions(TypedDict, total=False):
 )
 async def get_groups(
     startIndex: int = Query(1, ge=1),
-    count: int = Query(10, ge=1, le=100),
+    count: int = Query(10, ge=0),
     filter: str | None = Query(None),
 ):
     """
     Get a list of groups according to SCIM v2 protocol
     """
+    page_size: Final = min(count, SCIM_MAX_PAGE_SIZE)
     verbose_proxy_logger.debug(
         "SCIM GET GROUPS request: startIndex=%s count=%s filter=%s",
         startIndex,
@@ -2425,7 +2429,7 @@ async def get_groups(
         teams: Final = await _table(TeamRepository(prisma_client)).find_many(
             where=where_conditions,
             skip=(startIndex - 1),
-            take=count,
+            take=page_size,
             order={"created_at": "desc"},
         )
 
@@ -2462,7 +2466,7 @@ async def get_groups(
         return SCIMListResponse(
             totalResults=total_count,
             startIndex=startIndex,
-            itemsPerPage=min(count, len(scim_groups)),
+            itemsPerPage=len(scim_groups),
             Resources=scim_groups,
         )
 

tests/test_litellm/proxy/management_endpoints/scim/test_scim_v2_endpoints.py

@@ -7,7 +7,8 @@ from typing import Final
 from unittest.mock import AsyncMock, MagicMock, call
 
 import pytest
-from fastapi import HTTPException
+from fastapi import FastAPI, HTTPException
+from httpx import ASGITransport, AsyncClient
 from pytest_mock import MockerFixture
 
 from litellm.proxy._types import (
@@ -31,6 +32,7 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import (
     _handle_group_membership_changes,
     _handle_team_membership_changes,
     _parse_member_entries,
+    _premium_user_check,
     _process_group_patch_operations,
     _recompute_scim_member_roles,
     _resolve_group_member_ids,
@@ -45,8 +47,10 @@ from litellm.proxy.management_endpoints.scim.scim_v2 import (
     patch_group,
     patch_team_membership,
     patch_user,
+    scim_router,
     update_group,
     update_user,
+    user_api_key_auth,
 )
 from litellm.types.proxy.management_endpoints.scim_v2 import (
     SCIM_ENTERPRISE_USER_SCHEMA,
@@ -484,6 +488,48 @@ async def test_scim_create_user_respects_default_role_set_via_ui(mocker, monkeyp
     )
 
 
+@pytest.fixture
+def scim_test_client():
+    """An in-process SCIM application with authorization dependencies bypassed."""
+    app = FastAPI()
+    app.dependency_overrides[_premium_user_check] = lambda: None
+    app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN)
+    app.include_router(scim_router)
+    return AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("endpoint", ["Users", "Groups"])
+@pytest.mark.parametrize(("requested_count", "effective_count"), [(0, 0), (200, 100), (1000, 100)])
+async def test_scim_collection_endpoints_clamp_requested_page_size(
+    scim_test_client, endpoint, requested_count, effective_count, mocker
+):
+    """SCIM list endpoints accept zero and cap larger client page requests."""
+    mock_prisma_client = MagicMock()
+    mock_prisma_client.db = MagicMock()
+    table = MagicMock()
+    table.find_many = AsyncMock(return_value=[])
+    table.count = AsyncMock(return_value=0)
+    mock_prisma_client.db.litellm_usertable = table
+    mock_prisma_client.db.litellm_teamtable = table
+    mocker.patch(  # test-quality-ok: HTTP validation requires an in-memory database boundary.
+        "litellm.proxy.management_endpoints.scim.scim_v2._get_prisma_client_or_raise_exception",
+        AsyncMock(return_value=mock_prisma_client),
+    )
+
+    async with scim_test_client as client:
+        response = await client.get(f"/scim/v2/{endpoint}?startIndex=1&count={requested_count}")
+
+    assert response.status_code == 200
+    table.find_many.assert_awaited_once_with(
+        where={},
+        skip=0,
+        take=effective_count,
+        order={"created_at": "desc"},
+    )
+    assert response.json()["itemsPerPage"] == 0
+
+
 @pytest.mark.asyncio
 async def test_get_users_filters_username_by_exposed_scim_username_for_okta(mocker):
     """