chore(proxy): close router-settings-override fallback smuggling path

Two changes that together prevent a caller from smuggling unauthorized
models past the API key's allowlist via per-request router overrides.

1. ``_enforce_key_and_fallback_model_access``: also walk fallback models
   nested inside ``router_settings_override.fallbacks`` /
   ``context_window_fallbacks`` / ``content_policy_fallbacks``.
   ``route_llm_request.py`` promotes those to per-request kwargs after
   auth, so without this they bypassed the model allowlist entirely.
   New ``iter_router_fallback_model_names`` helper extracts leaf names
   from both the simple top-level shape (str | {"model": str}) and the
   nested router-config shape ({primary: [fallbacks]}). The two fallback
   validation loops are unified — every name (top-level + override) is
   deduplicated and validated once via ``can_key_call_model`` +
   ``is_valid_fallback_model``.

2. ``route_request``: strip router-internal ``mock_testing_*`` flags
   from user-supplied data. These are testing-only flags that
   deterministically force the router into fallback logic by raising a
   synthetic ``InternalServerError`` etc. Combined with override
   fallbacks they made the smuggling path trivially exploitable. Test
   code that calls the router directly bypasses the strip and is
   unaffected. The strip list is derived from ``MockRouterTestingParams``
   so a new ``mock_testing_*`` flag added to that dataclass is
   automatically covered.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

litellm/proxy/auth/user_api_key_auth.py

@@ -11,7 +11,7 @@ import asyncio
 import re
 import secrets
 from datetime import datetime, timezone
-from typing import Any, List, Optional, Tuple, cast
+from typing import Any, Iterator, List, Optional, Tuple, cast
 
 import fastapi
 from fastapi import HTTPException, Request, WebSocket, status
@@ -2131,10 +2131,6 @@ async def _enforce_key_and_fallback_model_access(
         pass
     else:
         model = get_model_from_request(request_data, route)
-        fallback_models = cast(
-            Optional[List[ALL_FALLBACK_MODEL_VALUES]],
-            request_data.get("fallbacks", None),
-        )
 
         if model is not None:
             await can_key_call_model(
@@ -2144,20 +2140,65 @@ async def _enforce_key_and_fallback_model_access(
                 llm_router=llm_router,
             )
 
-        if fallback_models is not None:
-            for m in fallback_models:
-                await can_key_call_model(
-                    model=m["model"] if isinstance(m, dict) else m,
-                    llm_model_list=llm_model_list,
-                    valid_token=valid_token,
-                    llm_router=llm_router,
-                )
-                await is_valid_fallback_model(
-                    model=m["model"] if isinstance(m, dict) else m,
-                    llm_router=llm_router,
-                    user_model=None,
+        # Validate every fallback model name reachable by this request:
+        # top-level ``fallbacks`` AND fallbacks nested inside
+        # ``router_settings_override`` (which ``route_llm_request.py``
+        # promotes to per-request kwargs *after* this check). VERIA-44.
+        fallback_names: List[str] = list(
+            iter_router_fallback_model_names(request_data.get("fallbacks"))
+        )
+        override_settings = request_data.get("router_settings_override")
+        if isinstance(override_settings, dict):
+            for _fb_key in ROUTER_FALLBACK_FIELDS:
+                fallback_names.extend(
+                    iter_router_fallback_model_names(override_settings.get(_fb_key))
                 )
 
+        for _name in dict.fromkeys(fallback_names):  # dedupe, preserve order
+            await can_key_call_model(
+                model=_name,
+                llm_model_list=llm_model_list,
+                valid_token=valid_token,
+                llm_router=llm_router,
+            )
+            await is_valid_fallback_model(
+                model=_name,
+                llm_router=llm_router,
+                user_model=None,
+            )
+
+
+ROUTER_FALLBACK_FIELDS: Tuple[str, ...] = (
+    "fallbacks",
+    "context_window_fallbacks",
+    "content_policy_fallbacks",
+)
+
+
+def iter_router_fallback_model_names(fallbacks: Any) -> Iterator[str]:
+    """Yield leaf model names from any of the supported fallbacks shapes.
+
+    Handles the simple top-level shape (``str`` or ``{"model": str}``) and
+    the nested router-config shape (``[{primary: [fallback_list]}]``).
+    """
+    if not isinstance(fallbacks, list):
+        return
+    for entry in fallbacks:
+        if isinstance(entry, str):
+            yield entry
+        elif isinstance(entry, dict):
+            if isinstance(entry.get("model"), str):
+                yield entry["model"]
+                continue
+            for fallback_list in entry.values():
+                if not isinstance(fallback_list, list):
+                    continue
+                for m in fallback_list:
+                    if isinstance(m, str):
+                        yield m
+                    elif isinstance(m, dict) and isinstance(m.get("model"), str):
+                        yield m["model"]
+
 
 async def _run_post_custom_auth_checks(
     valid_token: UserAPIKeyAuth,

litellm/proxy/route_llm_request.py

@@ -1,10 +1,18 @@
 import asyncio
+from dataclasses import fields as _dc_fields
 from typing import TYPE_CHECKING, Any, Literal, Optional
 
 from fastapi import HTTPException, status
 
 import litellm
 from litellm.proxy._types import UserAPIKeyAuth
+from litellm.types.router import MockRouterTestingParams
+
+# Router-internal mock_testing_* flag names. Single source of truth so a
+# new flag added to ``MockRouterTestingParams`` is automatically stripped.
+_MOCK_TESTING_KWARG_NAMES: tuple = tuple(
+    f.name for f in _dc_fields(MockRouterTestingParams)
+)
 
 if TYPE_CHECKING:
     from litellm.router import Router as _Router
@@ -322,6 +330,13 @@ async def route_request(  # noqa: PLR0915 - Complex routing function, refactorin
     """
     await add_shared_session_to_data(data)
 
+    # Strip router-internal mock_testing_* flags. Combined with an
+    # unauthorized fallback in ``router_settings_override`` they let a
+    # caller deterministically execute requests against restricted
+    # models. VERIA-44.
+    for _key in _MOCK_TESTING_KWARG_NAMES:
+        data.pop(_key, None)
+
     team_id = get_team_id_from_data(data)
     router_model_names = llm_router.model_names if llm_router is not None else []
 

tests/test_litellm/proxy/auth/test_router_override_fallback_auth.py

@@ -0,0 +1,187 @@
+"""
+VERIA-44: ``router_settings_override.fallbacks`` must be validated
+against the API key's model allowlist at auth time. Without this, the
+override is promoted to per-request kwargs after auth and lets a caller
+execute requests against models their API key cannot call.
+"""
+
+from typing import List
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+from litellm.proxy.auth.user_api_key_auth import (
+    _enforce_key_and_fallback_model_access,
+    iter_router_fallback_model_names,
+)
+
+
+def _key_with_models(models: List[str]) -> UserAPIKeyAuth:
+    return UserAPIKeyAuth(
+        api_key="hashed",
+        user_id="caller",
+        user_role=LitellmUserRoles.INTERNAL_USER,
+        models=models,
+    )
+
+
+# ── iter_router_fallback_model_names ─────────────────────────────────────────
+
+
+def testiter_router_fallback_model_names_router_config_shape():
+    """Router-config shape: ``[{primary: [fallback_list]}]``."""
+    assert list(
+        iter_router_fallback_model_names(
+            [{"gpt-3.5-turbo": ["gpt-4", "claude-3"]}, {"gpt-4o": ["o1"]}]
+        )
+    ) == ["gpt-4", "claude-3", "o1"]
+
+
+def testiter_router_fallback_model_names_simple_string_shape():
+    """Simple top-level shape: list of strings."""
+    assert list(iter_router_fallback_model_names(["gpt-4", "claude-3"])) == [
+        "gpt-4",
+        "claude-3",
+    ]
+
+
+def testiter_router_fallback_model_names_client_side_shape():
+    """ClientSideFallbackModel shape: ``[{"model": "..."}]``."""
+    assert list(
+        iter_router_fallback_model_names([{"model": "gpt-4"}, {"model": "claude-3"}])
+    ) == ["gpt-4", "claude-3"]
+
+
+def testiter_router_fallback_model_names_empty_or_none():
+    assert list(iter_router_fallback_model_names(None)) == []
+    assert list(iter_router_fallback_model_names([])) == []
+    assert list(iter_router_fallback_model_names("not a list")) == []
+
+
+# ── _enforce_key_and_fallback_model_access ────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_router_override_fallbacks_validated_against_key_allowlist():
+    """A fallback nested inside ``router_settings_override`` is validated
+    against the API key's allowed models — not just the top-level
+    ``fallbacks`` field."""
+    valid_token = _key_with_models(["gpt-3.5-turbo"])
+    request_data = {
+        "model": "gpt-3.5-turbo",
+        "router_settings_override": {
+            "fallbacks": [{"gpt-3.5-turbo": ["unauthorized-model"]}],
+        },
+    }
+
+    seen_models: List[str] = []
+
+    async def fake_can_key_call_model(model, llm_model_list, valid_token, llm_router):
+        seen_models.append(model)
+
+    with (
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.can_key_call_model",
+            side_effect=fake_can_key_call_model,
+        ),
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.is_valid_fallback_model",
+            new=AsyncMock(),
+        ),
+    ):
+        await _enforce_key_and_fallback_model_access(
+            valid_token=valid_token,
+            request_data=request_data,
+            route="/v1/chat/completions",
+            llm_model_list=None,
+            llm_router=None,
+        )
+
+    # Both the primary model and the override-nested fallback must be
+    # checked against the API key's allowlist.
+    assert "gpt-3.5-turbo" in seen_models
+    assert "unauthorized-model" in seen_models
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "fallback_field",
+    [
+        "fallbacks",
+        "context_window_fallbacks",
+        "content_policy_fallbacks",
+    ],
+)
+async def test_router_override_all_fallback_fields_validated(fallback_field):
+    """All three fallback fields the router accepts as per-request kwargs
+    are validated — context_window_fallbacks and content_policy_fallbacks
+    are promoted in route_llm_request.py too."""
+    valid_token = _key_with_models(["gpt-3.5-turbo"])
+    request_data = {
+        "model": "gpt-3.5-turbo",
+        "router_settings_override": {
+            fallback_field: [{"gpt-3.5-turbo": ["smuggled-model"]}],
+        },
+    }
+
+    seen: List[str] = []
+
+    async def fake_can_key_call_model(model, llm_model_list, valid_token, llm_router):
+        seen.append(model)
+
+    with (
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.can_key_call_model",
+            side_effect=fake_can_key_call_model,
+        ),
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.is_valid_fallback_model",
+            new=AsyncMock(),
+        ),
+    ):
+        await _enforce_key_and_fallback_model_access(
+            valid_token=valid_token,
+            request_data=request_data,
+            route="/v1/chat/completions",
+            llm_model_list=None,
+            llm_router=None,
+        )
+
+    assert "smuggled-model" in seen
+
+
+@pytest.mark.asyncio
+async def test_router_override_without_fallbacks_does_not_break_auth():
+    """``router_settings_override`` set without any fallback fields is a
+    no-op for the auth check — only the primary model is validated."""
+    valid_token = _key_with_models(["gpt-3.5-turbo"])
+    request_data = {
+        "model": "gpt-3.5-turbo",
+        "router_settings_override": {"num_retries": 3, "timeout": 30},
+    }
+
+    seen: List[str] = []
+
+    async def fake_can_key_call_model(model, llm_model_list, valid_token, llm_router):
+        seen.append(model)
+
+    with (
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.can_key_call_model",
+            side_effect=fake_can_key_call_model,
+        ),
+        patch(
+            "litellm.proxy.auth.user_api_key_auth.is_valid_fallback_model",
+            new=AsyncMock(),
+        ),
+    ):
+        await _enforce_key_and_fallback_model_access(
+            valid_token=valid_token,
+            request_data=request_data,
+            route="/v1/chat/completions",
+            llm_model_list=None,
+            llm_router=None,
+        )
+
+    assert seen == ["gpt-3.5-turbo"]

tests/test_litellm/proxy/test_route_llm_request.py

@@ -239,3 +239,35 @@ async def test_route_request_with_router_settings_override_preserves_existing():
     assert call_kwargs["num_retries"] == 10
     # Key/team timeout should be applied since not in request
     assert call_kwargs["timeout"] == 30
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "mock_flag",
+    [
+        "mock_testing_fallbacks",
+        "mock_testing_context_fallbacks",
+        "mock_testing_content_policy_fallbacks",
+    ],
+)
+async def test_route_request_strips_mock_testing_flags(mock_flag):
+    """VERIA-44: router-internal testing flags must not survive a
+    user-supplied request body. Without this strip, an attacker can
+    combine ``mock_testing_fallbacks=true`` with an unauthorized fallback
+    in ``router_settings_override`` to deterministically execute requests
+    against restricted models."""
+    data = {
+        "model": "gpt-3.5-turbo",
+        "messages": [{"role": "user", "content": "Hello"}],
+        mock_flag: True,
+    }
+    llm_router = MagicMock()
+    llm_router.acompletion.return_value = "ok"
+
+    await route_request(data, llm_router, None, "acompletion")
+
+    call_kwargs = llm_router.acompletion.call_args[1]
+    assert mock_flag not in call_kwargs
+    # The flag is also gone from the original data dict so any subsequent
+    # processing (e.g. logging) doesn't see it either.
+    assert mock_flag not in data