fix(management_v1): authorize bulk member budget writes off the writer and reject unschedulable reset windows

The roster the authorization check reads came from the routed reader, so a
replica lagging behind a team-admin demotion could still grant that caller
member-budget writes. Pin that read to the writer, as the model reconcile does.

A budget_duration the reset job can never schedule from, a non-positive one
that leaves the row permanently due or an unparseable one that blew up mid
batch as a 500, is now a 422 naming the row it came from, with nothing written.
The check is the same one /team/member_update and /budget/new already run,
lifted out of validate_budget_duration so both surfaces share it.

litellm/proxy/common_utils/timezone_utils.py

@@ -78,3 +78,27 @@ def get_budget_reset_time(budget_duration: str) -> datetime:
     `BudgetResetSettings` by injection (creation/update endpoints, startup backfill).
     """
     return compute_budget_reset_at(budget_duration, get_budget_reset_settings())
+
+
+def _is_persistable_budget_duration(budget_duration: str) -> bool:
+    from litellm.litellm_core_utils.duration_parser import duration_in_seconds
+
+    try:
+        if duration_in_seconds(budget_duration) <= 0:
+            return False
+        get_budget_reset_time(budget_duration=budget_duration)
+    except (ValueError, OverflowError):
+        return False
+    return True
+
+
+def budget_duration_error(budget_duration: str | None) -> str | None:
+    """Why `budget_duration` cannot be persisted, or None when it is usable.
+
+    A non-positive duration resolves to a reset time of "now", which leaves the row
+    permanently due: the reset job re-reads it every tick and, once enough of them
+    exist, they fill each batch and starve every other tenant's reset.
+    """
+    if budget_duration is None or _is_persistable_budget_duration(budget_duration):
+        return None
+    return f"Invalid budget_duration '{budget_duration}'. Use a format like '1h', '24h', '7d', or '30d'."

litellm/proxy/management_endpoints/common_utils.py

@@ -34,23 +34,11 @@ def validate_budget_duration(budget_duration: str | None, status_code: int = 400
     enough of them exist, they fill each batch and starve every other tenant's
     reset.
     """
-    if budget_duration is None:
-        return
+    from litellm.proxy.common_utils.timezone_utils import budget_duration_error
 
-    from litellm.litellm_core_utils.duration_parser import duration_in_seconds
-    from litellm.proxy.common_utils.timezone_utils import get_budget_reset_time
-
-    try:
-        if duration_in_seconds(budget_duration) <= 0:
-            raise ValueError("budget_duration must be positive")
-        get_budget_reset_time(budget_duration=budget_duration)
-    except (ValueError, OverflowError):
-        raise HTTPException(
-            status_code=status_code,
-            detail={
-                "error": f"Invalid budget_duration '{budget_duration}'. Use a format like '1h', '24h', '7d', or '30d'."
-            },
-        )
+    error: Final = budget_duration_error(budget_duration)
+    if error is not None:
+        raise HTTPException(status_code=status_code, detail={"error": error})
 
 
 from litellm._logging import verbose_proxy_logger

litellm/proxy/management_helpers/bulk_team_member_budgets.py

@@ -14,6 +14,7 @@ from typing import TYPE_CHECKING, Final
 from litellm.proxy._types import LiteLLM_TeamTable, LitellmUserRoles, Member, UserAPIKeyAuth
 from litellm.proxy.auth.auth_checks import invalidate_team_member_spend_state
 from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
+from litellm.proxy.db.routing_prisma_wrapper import WriterPinnedClient
 from litellm.proxy.management_endpoints.common_utils import (
     _is_user_org_admin_for_team,  # pyright: ignore[reportPrivateUsage]  # same check /team/member_update uses
     _is_user_team_admin,  # pyright: ignore[reportPrivateUsage]  # same check /team/member_update uses
@@ -115,7 +116,7 @@ async def bulk_update_team_member_budgets(
     user_api_key_cache: UserApiKeyCache,
 ) -> tuple[TeamMemberBudgetUpdateResult, ...]:
     """Apply one merge patch of per-member limits per requested member, in one transaction."""
-    team: Final = await TeamRepository(prisma_client).find_by_id(team_id)
+    team: Final = await TeamRepository(WriterPinnedClient(prisma_client.db)).find_by_id(team_id)
     if team is None:
         raise _team_not_found(team_id)
 

litellm/types/proxy/management_endpoints/team_endpoints.py

@@ -1,6 +1,6 @@
 from typing import Any, Final, Literal
 
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from litellm.proxy._types import (
     KeyManagementRoutes,
@@ -10,6 +10,7 @@ from litellm.proxy._types import (
     Member,
     MemberDeleteRequest,
 )
+from litellm.proxy.common_utils.timezone_utils import budget_duration_error
 from litellm.types.proxy.management_endpoints.management_v1 import ResourceResponse
 
 TeamIdSearchMatch = Literal["exact", "prefix"]
@@ -168,6 +169,14 @@ class TeamMemberBudgetPatch(TeamMemberRef):
     budget_duration: str | None = None
     allowed_models: tuple[str, ...] | None = None
 
+    @field_validator("budget_duration")
+    @classmethod
+    def persistable_budget_duration(cls, value: str | None) -> str | None:
+        error: Final = budget_duration_error(value)
+        if error is not None:
+            raise ValueError(error)
+        return value
+
 
 class BulkTeamMemberBudgetUpdateRequest(BaseModel):
     """Body of `POST /management/v1/teams/{team_id}/members/bulk_update`."""

tests/test_litellm/proxy/management_endpoints/management_v1/test_teams.py

@@ -25,6 +25,7 @@ from litellm.proxy.common_utils.user_api_key_cache import (
     team_membership_auth_cache_key,
     team_membership_reservation_cache_key,
 )
+from litellm.proxy.db.routing_prisma_wrapper import RoutingPrismaWrapper
 from litellm.proxy.list_api.common import ManagementProblem, problem_response, request_validation_problem
 from litellm.proxy.management_endpoints.management_v1 import router
 from litellm.proxy.management_endpoints.management_v1.common import MANAGEMENT_V1_PREFIX
@@ -145,12 +146,23 @@ class _MembershipTable:
 
 
 class _TeamTable:
+    """`find_many` and `create` are what `RoutingPrismaWrapper` keys read routing off, so a fake
+    table without them would silently never route and pass a reader-staleness test on the writer."""
+
     def __init__(self, teams: Sequence[LiteLLM_TeamTable]) -> None:
         self.rows: dict[str, LiteLLM_TeamTable] = {t.team_id: t for t in teams}
 
     async def find_unique(self, where: Mapping[str, str]) -> LiteLLM_TeamTable | None:
         return self.rows.get(where["team_id"])
 
+    async def find_many(self, where: Mapping[str, object] | None = None) -> list[LiteLLM_TeamTable]:
+        return [t for t in self.rows.values() if where is None or _matches(t.model_dump(), where)]
+
+    async def create(self, data: Mapping[str, object]) -> LiteLLM_TeamTable:
+        row: Final = LiteLLM_TeamTable.model_validate(dict(data))
+        self.rows[row.team_id] = row
+        return row
+
 
 class _Db:
     def __init__(
@@ -183,6 +195,29 @@ class _FakePrisma:
             raise
 
 
+class _ReplicatedPrisma:
+    """A client whose reads route to a lagging replica, as a proxy with `DATABASE_URL_READ_REPLICA` does."""
+
+    def __init__(self, writer: _FakePrisma, reader: _FakePrisma) -> None:
+        self._writer = writer
+        self.db = RoutingPrismaWrapper(writer=writer.db, reader=reader.db)  # pyright: ignore[reportArgumentType]  # fake dbs stand in for PrismaWrapper
+
+    def tx(self, *, timeout: object = None):
+        return self._writer.tx(timeout=timeout)
+
+
+class _UnreachableDb:
+    """A `.db` whose every table access fails, as one behind a dropped connection does."""
+
+    def __getattr__(self, name: str) -> object:
+        raise RuntimeError("connection reset by peer")
+
+
+class _UnreachablePrisma:
+    def __init__(self) -> None:
+        self.db = _UnreachableDb()
+
+
 def _team(
     *members: str,
     team_id: str = TEAM_ID,
@@ -220,7 +255,7 @@ def _budget(
 
 
 async def _bulk_update(
-    prisma: _FakePrisma,
+    prisma: _FakePrisma | _ReplicatedPrisma,
     members: Sequence[Mapping[str, object]],
     team_id: str = TEAM_ID,
     caller: UserAPIKeyAuth = ADMIN,
@@ -575,6 +610,27 @@ async def test_a_row_that_names_nobody_on_the_team_reports_no_cap_and_no_source(
     ]
 
 
+@pytest.mark.asyncio
+async def test_the_roster_authz_read_runs_on_the_writer_so_a_lagging_replica_cannot_let_a_demoted_admin_write():
+    writer = _FakePrisma(
+        teams=[_team("lead", "m1")],
+        memberships=[_membership("m1", "priv-m1")],
+        budgets=[_budget("priv-m1", max_budget=1.0)],
+    )
+    replica = _FakePrisma(teams=[_team("lead", "m1", admins=("lead",))])
+    demoted = UserAPIKeyAuth(user_id="lead", user_role=LitellmUserRoles.INTERNAL_USER)
+
+    with pytest.raises(ManagementProblem) as raised:
+        await _bulk_update(
+            _ReplicatedPrisma(writer=writer, reader=replica),
+            [{"user_id": "m1", "max_budget_in_team": 99}],
+            caller=demoted,
+        )
+
+    assert raised.value.problem.status == 403
+    assert writer.db.litellm_budgettable.rows["priv-m1"].max_budget == 1.0
+
+
 app = FastAPI()
 
 
@@ -673,3 +729,44 @@ def test_a_team_admin_may_bulk_update_their_own_teams_members(prisma, monkeypatc
 
     assert response.status_code == 200
     assert [(r["user_id"], r["success"], r["max_budget"]) for r in response.json()["data"]] == [("m1", True, 10.0)]
+
+
+@pytest.mark.parametrize("duration", ("0d", "nonsense"))
+def test_a_budget_duration_no_reset_can_be_scheduled_from_is_a_422_naming_its_row_and_writes_nothing(
+    prisma, as_proxy_admin, duration
+):
+    response = _post(
+        {
+            "members": [
+                {"user_id": "m1", "max_budget_in_team": 10},
+                {"user_id": "m2", "budget_duration": duration},
+            ]
+        }
+    )
+
+    assert response.status_code == 422
+    assert response.headers["content-type"] == "application/problem+json"
+    assert response.json()["type"] == "urn:litellm:error:invalid-request-body"
+    assert "members.1.budget_duration" in response.json()["detail"]
+    assert prisma.db.litellm_budgettable.rows["priv-m1"].max_budget == 1.0
+
+
+def test_an_unconnected_database_is_a_503_problem_document(monkeypatch, as_proxy_admin):
+    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", None)
+
+    response = _post({"members": [{"user_id": "m1", "max_budget_in_team": 10}]})
+
+    assert response.status_code == 503
+    assert response.headers["content-type"] == "application/problem+json"
+    assert response.json()["type"] == "urn:litellm:error:database-not-connected"
+
+
+def test_a_driver_error_answers_as_a_problem_document_without_leaking_the_exception(monkeypatch, as_proxy_admin):
+    monkeypatch.setattr("litellm.proxy.proxy_server.prisma_client", _UnreachablePrisma())
+
+    response = _post({"members": [{"user_id": "m1", "max_budget_in_team": 10}]})
+
+    assert response.status_code == 500
+    assert response.headers["content-type"] == "application/problem+json"
+    assert response.json()["type"] == "urn:litellm:error:internal-server-error"
+    assert "connection reset by peer" not in response.text