[Fix] Constrain feature_name to Literal, deduplicate runtime flags, fix silent test swallowing

- rbac_utils.py: change feature_name from str to Literal["agents", "vector_stores"]
  so typos are caught by type checkers at import time
- proxy_setting_endpoints.py: extract _RUNTIME_GENERAL_SETTINGS_FLAGS as a module-level
  constant, replacing duplicated inline lists in get_ui_settings and update_ui_settings
- test_vector_store_rbac.py: remove try/except pattern that silently swallowed non-403
  HTTPExceptions; tests now let any unexpected exception propagate as a test failure

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
yuneng-jiang
2026-03-05 15:42:17 -08:00
co-authored by Claude Opus 4.6
parent b3c092f489
commit 79817ff796
3 changed files with 22 additions and 31 deletions
+5 -1
View File
@@ -5,14 +5,18 @@ These helpers are used by agent and vector store endpoints to enforce
proxy-admin-configurable toggles that restrict access for internal users.
"""
from typing import Literal
from fastapi import HTTPException
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
FeatureName = Literal["agents", "vector_stores"]
async def check_feature_access_for_user(
user_api_key_dict: UserAPIKeyAuth,
feature_name: str,
feature_name: FeatureName,
) -> None:
"""
Raise HTTP 403 if the user's role is blocked from accessing the given feature
@@ -145,6 +145,16 @@ ALLOWED_UI_SETTINGS_FIELDS = {
"allow_vector_stores_for_team_admins",
}
# Flags that must be synced from the persisted UISettings into
# general_settings at runtime (on both read and write).
_RUNTIME_GENERAL_SETTINGS_FLAGS = [
"forward_client_headers_to_llm_api",
"disable_agents_for_internal_users",
"allow_agents_for_team_admins",
"disable_vector_stores_for_internal_users",
"allow_vector_stores_for_team_admins",
]
class MCPSemanticFilterSettings(BaseModel):
"""Configuration for MCP Semantic Tool Filter"""
@@ -1002,14 +1012,7 @@ async def get_ui_settings():
# Sync runtime flags into general_settings so the proxy picks them up
# at runtime (covers server restart scenarios).
_runtime_flags = [
"forward_client_headers_to_llm_api",
"disable_agents_for_internal_users",
"allow_agents_for_team_admins",
"disable_vector_stores_for_internal_users",
"allow_vector_stores_for_team_admins",
]
_flags_to_sync = {k: ui_settings[k] for k in _runtime_flags if k in ui_settings}
_flags_to_sync = {k: ui_settings[k] for k in _RUNTIME_GENERAL_SETTINGS_FLAGS if k in ui_settings}
if _flags_to_sync:
from litellm.proxy.proxy_server import general_settings
@@ -1080,14 +1083,7 @@ async def update_ui_settings(
# Sync runtime flags to general_settings so the proxy picks them up
# at runtime (general_settings is checked in pre-call utils).
_runtime_flags = [
"forward_client_headers_to_llm_api",
"disable_agents_for_internal_users",
"allow_agents_for_team_admins",
"disable_vector_stores_for_internal_users",
"allow_vector_stores_for_team_admins",
]
_flags_to_sync = {k: ui_settings[k] for k in _runtime_flags if k in ui_settings}
_flags_to_sync = {k: ui_settings[k] for k in _RUNTIME_GENERAL_SETTINGS_FLAGS if k in ui_settings}
if _flags_to_sync:
from litellm.proxy.proxy_server import general_settings
@@ -53,7 +53,6 @@ async def test_list_vector_stores_allowed_when_not_disabled():
mock_prisma = MagicMock()
mock_prisma.db.litellm_managedvectorstorestable.find_many = AsyncMock(return_value=[])
raised_403 = False
with patch.dict("litellm.proxy.proxy_server.general_settings", _ENABLED_GS, clear=True):
with patch("litellm.proxy.proxy_server.prisma_client", mock_prisma):
with patch.object(litellm, "vector_store_registry", None):
@@ -61,12 +60,9 @@ async def test_list_vector_stores_allowed_when_not_disabled():
"litellm.proxy.vector_store_endpoints.management_endpoints.VectorStoreRegistry._get_vector_stores_from_db",
new=AsyncMock(return_value=[]),
):
try:
await list_vector_stores(user_api_key_dict=user)
except HTTPException as e:
if e.status_code == 403:
raised_403 = True
assert not raised_403, "Should not raise 403 when vector stores are not disabled"
# Must not raise any HTTPException — if mocking is incomplete the
# test should fail loudly rather than silently swallowing errors.
await list_vector_stores(user_api_key_dict=user)
# ---------------------------------------------------------------------------
@@ -105,7 +101,6 @@ async def test_list_vector_stores_admin_not_blocked():
mock_prisma = MagicMock()
mock_prisma.db.litellm_managedvectorstorestable.find_many = AsyncMock(return_value=[])
raised_403 = False
with patch.dict("litellm.proxy.proxy_server.general_settings", _DISABLED_GS, clear=True):
with patch("litellm.proxy.proxy_server.prisma_client", mock_prisma):
with patch.object(litellm, "vector_store_registry", None):
@@ -113,9 +108,5 @@ async def test_list_vector_stores_admin_not_blocked():
"litellm.proxy.vector_store_endpoints.management_endpoints.VectorStoreRegistry._get_vector_stores_from_db",
new=AsyncMock(return_value=[]),
):
try:
await list_vector_stores(user_api_key_dict=admin)
except HTTPException as e:
if e.status_code == 403:
raised_403 = True
assert not raised_403, "Admin should not be blocked even when vector stores are disabled"
# Must not raise any HTTPException — admin is always allowed.
await list_vector_stores(user_api_key_dict=admin)