feat(proxy): gate fallback model access enforcement behind enforce_fallback_model_access

litellm/proxy/_types.py

@@ -2589,6 +2589,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
         None,
         description="By default, the user calling /team/new is automatically added to the new team as a team admin. If True, proxy admins are no longer auto-added; members explicitly listed in members_with_roles are unaffected. Default is False.",
     )
+    enforce_fallback_model_access: bool | None = Field(
+        None,
+        description="If True, router fallbacks configured in router_settings are only attempted when the calling key (and its team and project) is allowed to call the fallback model; unauthorized fallback targets are skipped and the primary model's error is returned. Default is False.",
+    )
     scheduled_job_stagger: ScheduledJobStaggerSettings | None = Field(
         None,
         description=(

litellm/proxy/auth/fallback_model_access.py

@@ -4,10 +4,12 @@ Authorize router fallback targets against the caller's key, team and project mod
 `_enforce_key_and_fallback_model_access` only sees fallbacks the client sends in the request body.
 Fallbacks configured on the router (`router_settings.fallbacks` and friends) are chosen after auth,
 inside the router, so this predicate is injected into the router to re-run the same model access
-checks for each fallback target before it is attempted.
+checks for each fallback target before it is attempted. Opt-in via
+`general_settings.enforce_fallback_model_access: true`.
 """
 
-from collections.abc import Mapping
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
 from typing import Final
 
 from pydantic import BaseModel, ValidationError
@@ -22,6 +24,10 @@ class _RequestMetadata(BaseModel):
     user_api_key_auth: UserAPIKeyAuth | None = None
 
 
+class _FallbackAccessSettings(BaseModel):
+    enforce_fallback_model_access: bool = False
+
+
 async def is_model_authorized_for_token(*, model: str, valid_token: UserAPIKeyAuth, llm_router: Router) -> bool:
     try:
         await can_key_call_resolved_model(
@@ -56,13 +62,29 @@ def _user_api_key_auth_from_request(request_kwargs: Mapping[str, object]) -> Use
     )
 
 
-async def router_fallback_access_check(*, model: str, request_kwargs: Mapping[str, object], llm_router: Router) -> bool:
+def _enforced_by_general_settings() -> bool:
+    from litellm.proxy.proxy_server import general_settings
+
+    return _FallbackAccessSettings.model_validate(general_settings).enforce_fallback_model_access
+
+
+@dataclass(frozen=True, slots=True)
+class RouterFallbackAccessCheck:
     """
-    `FallbackAccessCheck` for the proxy's router: a fallback target is attempted only when the
-    key behind the request could have requested it directly. Requests that carry no key (for
-    example internal health checks) are not restricted.
+    `FallbackAccessCheck` for the proxy's router: while `is_enforced()` is true, a fallback target
+    is attempted only when the key behind the request could have requested it directly. Requests
+    that carry no key (for example internal health checks) are not restricted.
     """
-    valid_token: Final = _user_api_key_auth_from_request(request_kwargs)
-    if valid_token is None:
-        return True
-    return await is_model_authorized_for_token(model=model, valid_token=valid_token, llm_router=llm_router)
+
+    is_enforced: Callable[[], bool]
+
+    async def __call__(self, *, model: str, request_kwargs: Mapping[str, object], llm_router: Router) -> bool:
+        if not self.is_enforced():
+            return True
+        valid_token: Final = _user_api_key_auth_from_request(request_kwargs)
+        if valid_token is None:
+            return True
+        return await is_model_authorized_for_token(model=model, valid_token=valid_token, llm_router=llm_router)
+
+
+router_fallback_access_check: Final = RouterFallbackAccessCheck(is_enforced=_enforced_by_general_settings)

tests/test_litellm/proxy/auth/test_fallback_model_access.py

@@ -3,6 +3,7 @@ import pytest
 from litellm import Router
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.auth.fallback_model_access import (
+    RouterFallbackAccessCheck,
     is_model_authorized_for_token,
     router_fallback_access_check,
 )
@@ -29,6 +30,14 @@ def _key_limited_to(access_group: str) -> UserAPIKeyAuth:
     return UserAPIKeyAuth(api_key="hashed", models=[access_group])
 
 
+def _request_with_key(metadata_field: str = "metadata") -> dict:
+    return {metadata_field: {"user_api_key_auth": _key_limited_to("open-group")}}
+
+
+ENFORCED = RouterFallbackAccessCheck(is_enforced=lambda: True)
+NOT_ENFORCED = RouterFallbackAccessCheck(is_enforced=lambda: False)
+
+
 @pytest.mark.asyncio
 async def test_is_model_authorized_for_token_follows_the_key_access_groups():
     router = _router()
@@ -57,18 +66,42 @@ async def test_is_model_authorized_for_token_fails_closed_when_the_lookup_breaks
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("metadata_field", ["metadata", "litellm_metadata"])
-async def test_router_fallback_access_check_authorizes_the_key_carried_in_request_metadata(metadata_field: str):
+async def test_enforced_check_authorizes_the_key_carried_in_request_metadata(metadata_field: str):
     router = _router()
-    request_kwargs = {metadata_field: {"user_api_key_auth": _key_limited_to("open-group")}}
+    request_kwargs = _request_with_key(metadata_field)
 
-    assert await router_fallback_access_check(model="open-model", request_kwargs=request_kwargs, llm_router=router)
-    assert not await router_fallback_access_check(
-        model="secret-model", request_kwargs=request_kwargs, llm_router=router
-    )
+    assert await ENFORCED(model="open-model", request_kwargs=request_kwargs, llm_router=router)
+    assert not await ENFORCED(model="secret-model", request_kwargs=request_kwargs, llm_router=router)
 
 
 @pytest.mark.asyncio
-async def test_router_fallback_access_check_does_not_restrict_requests_without_a_key():
-    assert await router_fallback_access_check(
-        model="secret-model", request_kwargs={"metadata": {}}, llm_router=_router()
+async def test_enforced_check_does_not_restrict_requests_without_a_key():
+    assert await ENFORCED(model="secret-model", request_kwargs={"metadata": {}}, llm_router=_router())
+
+
+@pytest.mark.asyncio
+async def test_check_allows_every_fallback_while_not_enforced():
+    assert await NOT_ENFORCED(model="secret-model", request_kwargs=_request_with_key(), llm_router=_router())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("general_settings", "expected"),
+    [
+        ({}, True),
+        ({"enforce_fallback_model_access": False}, True),
+        ({"enforce_fallback_model_access": True}, False),
+        ({"enforce_fallback_model_access": "true"}, False),
+    ],
+)
+async def test_proxy_check_reads_enforce_fallback_model_access_from_general_settings(
+    monkeypatch: pytest.MonkeyPatch, general_settings: dict, expected: bool
+):
+    monkeypatch.setattr("litellm.proxy.proxy_server.general_settings", general_settings)
+
+    assert (
+        await router_fallback_access_check(
+            model="secret-model", request_kwargs=_request_with_key(), llm_router=_router()
+        )
+        is expected
     )