fix(proxy): refuse /fallback writes to a fallbacks list set in config.yaml

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/proxy/management_endpoints/fallback_management_endpoints.py

@@ -11,10 +11,10 @@ DELETE /fallback/{model} - Delete fallbacks for a specific model
 # pyright: reportMissingImports=false
 
 import json
-from collections.abc import Mapping
+from collections.abc import Mapping, Sequence
 from typing import TYPE_CHECKING, Annotated, Final, Literal
 
-from pydantic import Field, TypeAdapter
+from pydantic import Field, JsonValue, TypeAdapter
 
 from litellm._logging import verbose_proxy_logger
 from litellm.proxy._types import UserAPIKeyAuth
@@ -47,6 +47,7 @@ ROUTER_SETTINGS_PARAM: Final = "router_settings"
 FallbackRule = dict[str, list[str]]
 StoredFallback = Annotated[FallbackRule | dict[str, object] | str, Field(union_mode="left_to_right")]
 _STORED_FALLBACKS: Final = TypeAdapter(list[StoredFallback])
+_JSON_RULES: Final = TypeAdapter(list[JsonValue])
 
 
 def _rule_covers(entry: StoredFallback, model: str) -> bool:
@@ -59,6 +60,12 @@ async def _router_settings_fresh_from_db(proxy_config: "ProxyConfig") -> dict[st
     return config.get(ROUTER_SETTINGS_PARAM, {})
 
 
+def _reject_config_owned_rules(proxy_config: "ProxyConfig", fallback_key: str, rules: Sequence[StoredFallback]) -> None:
+    proxy_config.reject_config_owned_writes(
+        section_name=ROUTER_SETTINGS_PARAM, changed_keys={fallback_key: _JSON_RULES.validate_python(rules)}
+    )
+
+
 async def _persist_router_settings(prisma_client: PrismaClient, router_settings: Mapping[str, object]) -> None:
     router_settings_json: Final = json.dumps(router_settings)
     await ConfigRepository(prisma_client).table.upsert(
@@ -180,6 +187,8 @@ async def create_fallback(
             # Add new fallback
             existing_fallbacks.append({data.model: data.fallback_models})
 
+        _reject_config_owned_rules(proxy_config, fallback_key, existing_fallbacks)
+
         # Update router settings
         router_settings[fallback_key] = existing_fallbacks
 
@@ -323,21 +332,16 @@ async def delete_fallback(
         # Get existing fallbacks
         existing_fallbacks: Final = _STORED_FALLBACKS.validate_python(router_settings.get(fallback_key) or [])
 
-        # Find and remove the fallback configuration
-        fallback_found = False
-        updated_fallbacks: Final = []
-        for rule in existing_fallbacks:
-            if not _rule_covers(rule, model):
-                updated_fallbacks.append(rule)
-            else:
-                fallback_found = True
+        updated_fallbacks: Final = [rule for rule in existing_fallbacks if not _rule_covers(rule, model)]
 
-        if not fallback_found:
+        if len(updated_fallbacks) == len(existing_fallbacks):
             raise HTTPException(
                 status_code=status.HTTP_404_NOT_FOUND,
                 detail={"error": f"No {fallback_type} fallbacks configured for model '{model}'"},
             )
 
+        _reject_config_owned_rules(proxy_config, fallback_key, updated_fallbacks)
+
         # Update router settings
         router_settings[fallback_key] = updated_fallbacks
 

tests/unit/proxy/test_fallback_management_endpoints.py

@@ -17,9 +17,9 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
+from pydantic import JsonValue, TypeAdapter
 
 from litellm import Router
-from litellm.proxy.utils import evict_config_param, get_config_param
 from litellm.proxy.management_endpoints.fallback_management_endpoints import (
     FallbackCreateRequest,
     FallbackDeleteResponse,
@@ -28,6 +28,8 @@ from litellm.proxy.management_endpoints.fallback_management_endpoints import (
     delete_fallback,
     get_fallback,
 )
+from litellm.proxy.proxy_server import ProxyConfig
+from litellm.proxy.utils import evict_config_param, get_config_param
 from litellm.types.router import DeploymentTypedDict, LiteLLMParamsTypedDict
 
 
@@ -122,6 +124,7 @@ FALLBACK_DEPLOYMENT_ID: Final = "team-fallback-id"
 
 FallbackRules = list[dict[str, list[str]] | dict[str, object] | str]
 RouterSettings = dict[str, FallbackRules | None]
+_ROUTER_SETTINGS_JSON: Final = TypeAdapter(dict[str, JsonValue])
 
 
 def _litellm_params(mock_response: str | None) -> LiteLLMParamsTypedDict:
@@ -254,12 +257,14 @@ def _config_row(router_settings: RouterSettings) -> SimpleNamespace:
 class _StoredRouterSettings:
     """The LiteLLM_Config router_settings row, with the proxy objects that read it through the config cache"""
 
-    def __init__(self, router_settings: RouterSettings | None) -> None:
+    def __init__(self, router_settings: RouterSettings | None, config_file: RouterSettings | None = None) -> None:
         self.row: SimpleNamespace | None = None if router_settings is None else _config_row(router_settings)
         self.prisma_client: Final = MagicMock()
         self.prisma_client.get_generic_data = AsyncMock(side_effect=lambda **_: self.row)
         self.prisma_client.db.litellm_config.upsert = AsyncMock(side_effect=self._upsert)
-        self.proxy_config: Final = MagicMock()
+        self.proxy_config: Final = MagicMock() if config_file is None else ProxyConfig()
+        if config_file is not None:
+            self.proxy_config.router_settings.load_yaml(_ROUTER_SETTINGS_JSON.validate_python(config_file))
         self.proxy_config.get_config = AsyncMock(side_effect=self._get_config)
 
     async def _upsert(self, where: dict[str, str], data: dict[str, dict[str, str]]) -> None:
@@ -376,6 +381,102 @@ class TestFallbackWritesSeeTheLatestStoredRules:
         assert stored.stored_fallbacks() == [GATEWAY_RULE]
 
 
+CONFIG_RULE: Final = {"gpt-5.4-mini": ["team-fallback"]}
+CONFIG_WINDOW_RULE: Final = {"team-primary": ["gpt-5.4-mini"]}
+
+
+class TestFallbackWritesHonourTheConfigFile:
+    """A fallbacks list set in config.yaml owns the key: /fallback writes that would change it are refused and
+    the router keeps serving the file's rules"""
+
+    @pytest.fixture(autouse=True)
+    async def clean_config_cache(self) -> AsyncIterator[None]:
+        await evict_config_param("router_settings")
+        yield
+        await evict_config_param("router_settings")
+
+    @pytest.fixture
+    def router(self) -> Router:
+        return _team_router()
+
+    @pytest.fixture
+    def stored(self) -> _StoredRouterSettings:
+        return _StoredRouterSettings({"fallbacks": [CONFIG_RULE]}, config_file={"fallbacks": [CONFIG_RULE]})
+
+    async def _create(
+        self, request: FallbackCreateRequest, router: Router, stored: _StoredRouterSettings
+    ) -> FallbackResponse:
+        with (
+            patch("litellm.proxy.proxy_server.llm_router", router),
+            patch("litellm.proxy.proxy_server.prisma_client", stored.prisma_client),
+            patch("litellm.proxy.proxy_server.proxy_config", stored.proxy_config),
+            patch("litellm.proxy.proxy_server.store_model_in_db", True),
+        ):
+            return await create_fallback(request, MagicMock())
+
+    async def _delete(self, model: str, router: Router, stored: _StoredRouterSettings) -> FallbackDeleteResponse:
+        with (
+            patch("litellm.proxy.proxy_server.llm_router", router),
+            patch("litellm.proxy.proxy_server.prisma_client", stored.prisma_client),
+            patch("litellm.proxy.proxy_server.proxy_config", stored.proxy_config),
+            patch("litellm.proxy.proxy_server.store_model_in_db", True),
+        ):
+            return await delete_fallback(model, "general", MagicMock())
+
+    async def test_a_create_that_adds_a_rule_to_a_config_owned_list_is_refused(
+        self, router: Router, stored: _StoredRouterSettings
+    ) -> None:
+        router.fallbacks = [CONFIG_RULE]
+        request: Final = FallbackCreateRequest(model="team-primary", fallback_models=["team-fallback"])
+
+        with pytest.raises(HTTPException) as refused:
+            await self._create(request, router, stored)
+
+        assert refused.value.status_code == 400
+        assert "key 'fallbacks' is set in the config file" in str(refused.value.detail)
+        assert stored.stored_fallbacks() == [CONFIG_RULE]
+        stored.prisma_client.db.litellm_config.upsert.assert_not_awaited()
+        assert router.fallbacks == [CONFIG_RULE]
+
+    async def test_a_delete_of_a_config_owned_rule_is_refused(
+        self, router: Router, stored: _StoredRouterSettings
+    ) -> None:
+        router.fallbacks = [CONFIG_RULE]
+
+        with pytest.raises(HTTPException) as refused:
+            await self._delete("gpt-5.4-mini", router, stored)
+
+        assert refused.value.status_code == 400
+        assert "key 'fallbacks' is set in the config file" in str(refused.value.detail)
+        assert stored.stored_fallbacks() == [CONFIG_RULE]
+        stored.prisma_client.db.litellm_config.upsert.assert_not_awaited()
+        assert router.fallbacks == [CONFIG_RULE]
+
+    async def test_a_create_that_restates_the_config_rule_is_accepted(
+        self, router: Router, stored: _StoredRouterSettings
+    ) -> None:
+        request: Final = FallbackCreateRequest(model="gpt-5.4-mini", fallback_models=["team-fallback"])
+
+        response: Final = await self._create(request, router, stored)
+
+        assert response.fallback_models == ["team-fallback"]
+        assert stored.stored_fallbacks() == [CONFIG_RULE]
+
+    async def test_a_sibling_list_the_config_file_does_not_set_stays_writable(
+        self, router: Router, stored: _StoredRouterSettings
+    ) -> None:
+        request: Final = FallbackCreateRequest(
+            model="team-primary", fallback_models=["gpt-5.4-mini"], fallback_type="context_window"
+        )
+
+        response: Final = await self._create(request, router, stored)
+
+        assert response.fallback_type == "context_window"
+        assert stored.row is not None
+        assert stored.row.param_value["context_window_fallbacks"] == [CONFIG_WINDOW_RULE]
+        assert router.context_window_fallbacks == [CONFIG_WINDOW_RULE]
+
+
 @pytest.mark.asyncio
 class TestCreateFallback:
     """Test the create_fallback endpoint"""