refactor: consolidate route auth for UI and API tokens

Unify UI and API token authorization through the shared RBAC path
and backfill missing routes in role-based route lists.

litellm/proxy/_types.py

@@ -494,10 +494,12 @@ class LiteLLMRoutes(enum.Enum):
         "/v2/key/info",
         "/model_group/info",
         "/health",
+        "/health/services",
         "/key/list",
         "/user/filter/ui",
         "/models",
         "/v1/models",
+        "/sso/get/ui_settings",
     ]
 
     # NOTE: ROUTES ONLY FOR MASTER KEY - only the Master Key should be able to Reset Spend
@@ -566,6 +568,8 @@ class LiteLLMRoutes(enum.Enum):
         "/spend/tags",
         "/spend/calculate",
         "/spend/logs",
+        "/spend/logs/ui",
+        "/spend/logs/session/ui",
         "/cost/estimate",
     ]
 
@@ -581,6 +585,7 @@ class LiteLLMRoutes(enum.Enum):
         "/global/spend/report",
         "/global/spend/provider",
         "/global/spend/tags",
+        "/global/spend/all_tag_names",
     ]
 
     public_routes = set(
@@ -602,44 +607,18 @@ class LiteLLMRoutes(enum.Enum):
         ]
     )
 
-    ui_routes = [
-        "/sso",
-        "/sso/get/ui_settings",
-        "/get/ui_settings",
-        "/login",
-        "/key/info",
-        "/config",
-        "/spend",
-        "/model/info",
-        "/v2/model/info",
-        "/v2/key/info",
-        "/models",
-        "/v1/models",
-        "/global/spend",
-        "/global/spend/logs",
-        "/global/spend/keys",
-        "/global/spend/models",
-        "/global/spend/tags",
-        "/global/predict/spend/logs",
-        "/global/activity",
-        "/health/services",
-    ] + info_routes
-
     internal_user_routes = (
         [
-            "/global/spend/tags",
-            "/global/spend/keys",
-            "/global/spend/models",
-            "/global/spend/provider",
-            "/global/spend/end_users",
             "/global/activity",
             "/global/activity/model",
+            "/global/activity/cache_hits",
             "/v1/models/{model_id}",
             "/models/{model_id}",
             "/guardrails/list",
             "/v2/guardrails/list",
         ]
         + spend_tracking_routes
+        + global_spend_tracking_routes
         + key_management_routes
     )
 
@@ -694,6 +673,9 @@ class LiteLLMRoutes(enum.Enum):
         "/tag/list",
         "/audit",
         "/audit/{id}",
+        "/global/activity",
+        "/global/activity/model",
+        "/global/activity/cache_hits",
     ] + info_routes
 
     # All routes accesible by an Org Admin
@@ -892,9 +874,9 @@ class GenerateRequestBase(LiteLLMPydanticObjectBase):
     allowed_cache_controls: Optional[list] = []
     config: Optional[dict] = {}
     permissions: Optional[dict] = {}
-    model_max_budget: Optional[dict] = (
-        {}
-    )  # {"gpt-4": 5.0, "gpt-3.5-turbo": 5.0}, defaults to {}
+    model_max_budget: Optional[
+        dict
+    ] = {}  # {"gpt-4": 5.0, "gpt-3.5-turbo": 5.0}, defaults to {}
 
     model_config = ConfigDict(protected_namespaces=())
     model_rpm_limit: Optional[dict] = None
@@ -1036,9 +1018,9 @@ class RegenerateKeyRequest(GenerateKeyRequest):
     spend: Optional[float] = None
     metadata: Optional[dict] = None
     new_master_key: Optional[str] = None
-    grace_period: Optional[str] = (
-        None  # Duration to keep old key valid (e.g. "24h", "2d"); None = immediate revoke
-    )
+    grace_period: Optional[
+        str
+    ] = None  # Duration to keep old key valid (e.g. "24h", "2d"); None = immediate revoke
 
 
 class ResetSpendRequest(LiteLLMPydanticObjectBase):
@@ -1562,12 +1544,12 @@ class NewCustomerRequest(BudgetNewRequest):
     blocked: bool = False  # allow/disallow requests for this end-user
     budget_id: Optional[str] = None  # give either a budget_id or max_budget
     spend: Optional[float] = None
-    allowed_model_region: Optional[AllowedModelRegion] = (
-        None  # require all user requests to use models in this specific region
-    )
-    default_model: Optional[str] = (
-        None  # if no equivalent model in allowed region - default all requests to this model
-    )
+    allowed_model_region: Optional[
+        AllowedModelRegion
+    ] = None  # require all user requests to use models in this specific region
+    default_model: Optional[
+        str
+    ] = None  # if no equivalent model in allowed region - default all requests to this model
     object_permission: Optional[LiteLLM_ObjectPermissionBase] = None
 
     @model_validator(mode="before")
@@ -1590,12 +1572,12 @@ class UpdateCustomerRequest(LiteLLMPydanticObjectBase):
     blocked: bool = False  # allow/disallow requests for this end-user
     max_budget: Optional[float] = None
     budget_id: Optional[str] = None  # give either a budget_id or max_budget
-    allowed_model_region: Optional[AllowedModelRegion] = (
-        None  # require all user requests to use models in this specific region
-    )
-    default_model: Optional[str] = (
-        None  # if no equivalent model in allowed region - default all requests to this model
-    )
+    allowed_model_region: Optional[
+        AllowedModelRegion
+    ] = None  # require all user requests to use models in this specific region
+    default_model: Optional[
+        str
+    ] = None  # if no equivalent model in allowed region - default all requests to this model
     object_permission: Optional[LiteLLM_ObjectPermissionBase] = None
 
 
@@ -1685,15 +1667,15 @@ class NewTeamRequest(TeamBase):
     ] = None  # raise an error if 'guaranteed_throughput' is set and we're overallocating tpm
 
     model_tpm_limit: Optional[Dict[str, int]] = None
-    team_member_budget: Optional[float] = (
-        None  # allow user to set a budget for all team members
-    )
-    team_member_rpm_limit: Optional[int] = (
-        None  # allow user to set RPM limit for all team members
-    )
-    team_member_tpm_limit: Optional[int] = (
-        None  # allow user to set TPM limit for all team members
-    )
+    team_member_budget: Optional[
+        float
+    ] = None  # allow user to set a budget for all team members
+    team_member_rpm_limit: Optional[
+        int
+    ] = None  # allow user to set RPM limit for all team members
+    team_member_tpm_limit: Optional[
+        int
+    ] = None  # allow user to set TPM limit for all team members
     team_member_key_duration: Optional[str] = None  # e.g. "1d", "1w", "1m"
     team_member_budget_duration: Optional[str] = None  # e.g. "30d", "1mo"
     allowed_vector_store_indexes: Optional[List[AllowedVectorStoreIndexItem]] = None
@@ -1790,9 +1772,9 @@ class BlockKeyRequest(LiteLLMPydanticObjectBase):
 
 class AddTeamCallback(LiteLLMPydanticObjectBase):
     callback_name: str
-    callback_type: Optional[Literal["success", "failure", "success_and_failure"]] = (
-        "success_and_failure"
-    )
+    callback_type: Optional[
+        Literal["success", "failure", "success_and_failure"]
+    ] = "success_and_failure"
     callback_vars: Dict[str, str]
 
     @model_validator(mode="before")
@@ -2134,9 +2116,9 @@ class ConfigList(LiteLLMPydanticObjectBase):
     stored_in_db: Optional[bool]
     field_default_value: Any
     premium_field: bool = False
-    nested_fields: Optional[List[FieldDetail]] = (
-        None  # For nested dictionary or Pydantic fields
-    )
+    nested_fields: Optional[
+        List[FieldDetail]
+    ] = None  # For nested dictionary or Pydantic fields
 
 
 class UserHeaderMapping(LiteLLMPydanticObjectBase):
@@ -2495,9 +2477,9 @@ class UserAPIKeyAuth(
     user_max_budget: Optional[float] = None
     request_route: Optional[str] = None
     user: Optional[Any] = None  # Expanded user object when expand=user is used
-    created_by_user: Optional[Any] = (
-        None  # Expanded created_by user when expand=user is used
-    )
+    created_by_user: Optional[
+        Any
+    ] = None  # Expanded created_by user when expand=user is used
     end_user_object_permission: Optional[LiteLLM_ObjectPermissionTable] = None
     # Decoded upstream IdP claims (groups, roles, etc.) propagated by JWT auth machinery
     # and forwarded into outbound tokens by guardrails such as MCPJWTSigner.
@@ -2636,9 +2618,9 @@ class LiteLLM_OrganizationMembershipTable(LiteLLMPydanticObjectBase):
     budget_id: Optional[str] = None
     created_at: datetime
     updated_at: datetime
-    user: Optional[Any] = (
-        None  # You might want to replace 'Any' with a more specific type if available
-    )
+    user: Optional[
+        Any
+    ] = None  # You might want to replace 'Any' with a more specific type if available
     litellm_budget_table: Optional[LiteLLM_BudgetTable] = None
     user_email: Optional[str] = None
 
@@ -3793,9 +3775,9 @@ class TeamModelDeleteRequest(BaseModel):
 # Organization Member Requests
 class OrganizationMemberAddRequest(OrgMemberAddRequest):
     organization_id: str
-    max_budget_in_organization: Optional[float] = (
-        None  # Users max budget within the organization
-    )
+    max_budget_in_organization: Optional[
+        float
+    ] = None  # Users max budget within the organization
 
 
 class OrganizationMemberDeleteRequest(MemberDeleteRequest):
@@ -4050,9 +4032,9 @@ class ProviderBudgetResponse(LiteLLMPydanticObjectBase):
     Maps provider names to their budget configs.
     """
 
-    providers: Dict[str, ProviderBudgetResponseObject] = (
-        {}
-    )  # Dictionary mapping provider names to their budget configurations
+    providers: Dict[
+        str, ProviderBudgetResponseObject
+    ] = {}  # Dictionary mapping provider names to their budget configurations
 
 
 class ProxyStateVariables(TypedDict):
@@ -4214,9 +4196,9 @@ class LiteLLM_JWTAuth(LiteLLMPydanticObjectBase):
     enforce_rbac: bool = False
     roles_jwt_field: Optional[str] = None  # v2 on role mappings
     role_mappings: Optional[List[RoleMapping]] = None
-    object_id_jwt_field: Optional[str] = (
-        None  # can be either user / team, inferred from the role mapping
-    )
+    object_id_jwt_field: Optional[
+        str
+    ] = None  # can be either user / team, inferred from the role mapping
     scope_mappings: Optional[List[ScopeMapping]] = None
     enforce_scope_based_access: bool = False
     enforce_team_based_model_access: bool = False

litellm/proxy/auth/auth_checks.py

@@ -196,9 +196,7 @@ def _is_model_cost_zero(
     return True
 
 
-def _is_cost_explicitly_configured(
-    model: str, llm_router: "Router"
-) -> bool:
+def _is_cost_explicitly_configured(model: str, llm_router: "Router") -> bool:
     """
     Check if any deployment in the model group has cost fields explicitly
     set in its litellm.model_cost entry.
@@ -215,10 +213,7 @@ def _is_cost_explicitly_configured(
         if model_id is None:
             continue
         raw_entry = litellm.model_cost.get(model_id, {})
-        if (
-            "input_cost_per_token" in raw_entry
-            or "output_cost_per_token" in raw_entry
-        ):
+        if "input_cost_per_token" in raw_entry or "output_cost_per_token" in raw_entry:
             return True
     return False
 
@@ -596,17 +591,12 @@ async def common_checks(  # noqa: PLR0915
         user_object=user_object, route=route, request_body=request_body
     )
 
-    token_team = getattr(valid_token, "team_id", None)
-    token_type: Literal["ui", "api"] = (
-        "ui" if token_team is not None and token_team == "litellm-dashboard" else "api"
-    )
-    _is_route_allowed = _is_allowed_route(
+    _is_route_allowed = _is_api_route_allowed(
         route=route,
-        token_type=token_type,
-        user_obj=user_object,
         request=request,
         request_data=request_body,
         valid_token=valid_token,
+        user_obj=user_object,
     )
 
     # 11. [OPTIONAL] Vector store checks - is the object allowed to access the vector store
@@ -629,31 +619,6 @@ async def common_checks(  # noqa: PLR0915
     return True
 
 
-def _is_ui_route(
-    route: str,
-    user_obj: Optional[LiteLLM_UserTable] = None,
-) -> bool:
-    """
-    - Check if the route is a UI used route
-    """
-    # this token is only used for managing the ui
-    allowed_routes = LiteLLMRoutes.ui_routes.value
-    # check if the current route startswith any of the allowed routes
-    if (
-        route is not None
-        and isinstance(route, str)
-        and any(route.startswith(allowed_route) for allowed_route in allowed_routes)
-    ):
-        # Do something if the current route starts with any of the allowed routes
-        return True
-    elif any(
-        RouteChecks._route_matches_pattern(route=route, pattern=allowed_route)
-        for allowed_route in allowed_routes
-    ):
-        return True
-    return False
-
-
 def _get_user_role(
     user_obj: Optional[LiteLLM_UserTable],
 ) -> Optional[LitellmUserRoles]:
@@ -717,30 +682,6 @@ def _is_user_proxy_admin(user_obj: Optional[LiteLLM_UserTable]):
     return False
 
 
-def _is_allowed_route(
-    route: str,
-    token_type: Literal["ui", "api"],
-    request: Request,
-    request_data: dict,
-    valid_token: Optional[UserAPIKeyAuth],
-    user_obj: Optional[LiteLLM_UserTable] = None,
-) -> bool:
-    """
-    - Route b/w ui token check and normal token check
-    """
-
-    if token_type == "ui" and _is_ui_route(route=route, user_obj=user_obj):
-        return True
-    else:
-        return _is_api_route_allowed(
-            route=route,
-            request=request,
-            request_data=request_data,
-            valid_token=valid_token,
-            user_obj=user_obj,
-        )
-
-
 def _allowed_routes_check(user_route: str, allowed_routes: list) -> bool:
     """
     Return if a user is allowed to access route. Helper function for `allowed_routes_check`.

tests/proxy_unit_tests/test_jwt.py

@@ -715,7 +715,7 @@ async def aaaatest_user_token_output(
     assert team_result.user_id == user_id
 
 
-@pytest.mark.parametrize("admin_allowed_routes", [None, ["ui_routes"]])
+@pytest.mark.parametrize("admin_allowed_routes", [None, ["info_routes"]])
 @pytest.mark.parametrize("audience", [None, "litellm-proxy"])
 @pytest.mark.asyncio
 async def test_allowed_routes_admin(
@@ -934,10 +934,7 @@ async def mock_user_object(*args, **kwargs):
     user_id = kwargs.get("user_id")
     user_email = kwargs.get("user_email")
     return LiteLLM_UserTable(
-        spend=0, 
-        user_id=user_id, 
-        max_budget=None, 
-        user_email=user_email
+        spend=0, user_id=user_id, max_budget=None, user_email=user_email
     )
 
 
@@ -1170,15 +1167,13 @@ async def test_end_user_jwt_auth(monkeypatch):
     # use generated key to auth in
     from litellm import Router
     from litellm.types.router import RouterGeneralSettings
-    
+
     # Create a router with pass_through_all_models enabled
     router = Router(
         model_list=[],
-        router_general_settings=RouterGeneralSettings(
-            pass_through_all_models=True
-        ),
+        router_general_settings=RouterGeneralSettings(pass_through_all_models=True),
     )
-    
+
     setattr(litellm.proxy.proxy_server, "premium_user", True)
     setattr(
         litellm.proxy.proxy_server,
@@ -1196,7 +1191,7 @@ async def test_end_user_jwt_auth(monkeypatch):
 
     cost_tracking()
     result = await user_api_key_auth(request=request, api_key=bearer_token)
-    
+
     # Assert that end_user_id is correctly extracted from JWT token's 'sub' field
     assert result.end_user_id == "81b3e52a-67a6-4efb-9645-70527e101479"
 
@@ -1228,7 +1223,9 @@ async def test_end_user_jwt_auth(monkeypatch):
         ),
     )
 
-    with patch("litellm.acompletion", new=AsyncMock(return_value=mock_response)) as mock_completion:
+    with patch(
+        "litellm.acompletion", new=AsyncMock(return_value=mock_response)
+    ) as mock_completion:
         resp = await chat_completion(
             request=request,
             fastapi_response=temp_response,
@@ -1243,10 +1240,13 @@ async def test_end_user_jwt_auth(monkeypatch):
         # Verify the completion was called with correct end_user_id
         mock_completion.assert_called_once()
         call_kwargs = mock_completion.call_args.kwargs
-        
+
         # end_user_id is passed in metadata as 'user_api_key_end_user_id'
         metadata = call_kwargs.get("metadata", {})
-        assert metadata.get("user_api_key_end_user_id") == "81b3e52a-67a6-4efb-9645-70527e101479"
+        assert (
+            metadata.get("user_api_key_end_user_id")
+            == "81b3e52a-67a6-4efb-9645-70527e101479"
+        )
 
 
 def test_can_rbac_role_call_route():
@@ -1278,13 +1278,13 @@ def test_user_api_key_auth_jwt_hashing():
     """
     from litellm.proxy._types import UserAPIKeyAuth
     from litellm.proxy.auth.handle_jwt import JWTHandler
-    
+
     # Test with a JWT token (3 parts separated by dots)
     jwt_token = "test-jwt-token-header.payload.signature"
-    
+
     # Create UserAPIKeyAuth instance with JWT
     user_auth = UserAPIKeyAuth(api_key=jwt_token)
-    
+
     # Verify that the API key is hashed with "hashed-jwt-" prefix
     # critical - the raw JWT token should not be in the api_key or token
     assert user_auth.api_key.startswith("hashed-jwt-")
@@ -1292,19 +1292,18 @@ def test_user_api_key_auth_jwt_hashing():
     assert jwt_token not in user_auth.api_key
     assert jwt_token not in user_auth.token
 
-    
     # Test with a regular API key (should not be hashed)
     regular_api_key = "sk-1234567890abcdef"
     user_auth_regular = UserAPIKeyAuth(api_key=regular_api_key)
-    
+
     # Verify that regular API key is hashed normally (without "hashed-jwt-" prefix)
     assert not user_auth_regular.api_key.startswith("hashed-jwt-")
     assert not user_auth_regular.token.startswith("hashed-jwt-")
-    
+
     # Test with a non-JWT, non-sk string (should not be hashed)
     non_jwt_key = "some-random-key"
     user_auth_non_jwt = UserAPIKeyAuth(api_key=non_jwt_key)
-    
+
     # Verify that non-JWT key is not hashed
     assert user_auth_non_jwt.api_key == non_jwt_key
     assert user_auth_non_jwt.token == non_jwt_key
@@ -1315,19 +1314,19 @@ def test_jwt_handler_is_jwt_static_method():
     Test that JWTHandler.is_jwt is a static method and works correctly
     """
     from litellm.proxy.auth.handle_jwt import JWTHandler
-    
+
     # Test with valid JWT format
     valid_jwt = "test-jwt-token-header.payload.signature"
     assert JWTHandler.is_jwt(valid_jwt) == True
-    
+
     # Test with invalid JWT format (only 2 parts)
     invalid_jwt = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ"
     assert JWTHandler.is_jwt(invalid_jwt) == False
-    
+
     # Test with regular API key
     regular_key = "sk-1234567890abcdef"
     assert JWTHandler.is_jwt(regular_key) == False
-    
+
     # Test with empty string
     assert JWTHandler.is_jwt("") == False
 
@@ -1461,7 +1460,13 @@ async def test_auth_jwt_es256_jwk_path(monkeypatch):
 
     now = int(time.time())
     token = jwt.encode(
-        {"sub": "alice", "aud": "litellm-proxy", "iss": "http://example", "iat": now, "exp": now + 300},
+        {
+            "sub": "alice",
+            "aud": "litellm-proxy",
+            "iss": "http://example",
+            "iat": now,
+            "exp": now + 300,
+        },
         ec_priv_pem,
         algorithm="ES256",
         headers={"kid": "ec1"},
@@ -1508,7 +1513,13 @@ async def test_auth_jwt_rs256_regression(monkeypatch):
 
     now = int(time.time())
     token = jwt.encode(
-        {"sub": "bob", "aud": "litellm-proxy", "iss": "http://example", "iat": now, "exp": now + 300},
+        {
+            "sub": "bob",
+            "aud": "litellm-proxy",
+            "iss": "http://example",
+            "iat": now,
+            "exp": now + 300,
+        },
         rsa_priv_pem,
         algorithm="RS256",
         headers={"kid": "rsa1"},
@@ -1540,7 +1551,13 @@ async def test_auth_jwt_mismatched_key_fails(monkeypatch):
     )
     now = int(time.time())
     token = jwt.encode(
-        {"sub": "mallory", "aud": "litellm-proxy", "iss": "http://example", "iat": now, "exp": now + 300},
+        {
+            "sub": "mallory",
+            "aud": "litellm-proxy",
+            "iss": "http://example",
+            "iat": now,
+            "exp": now + 300,
+        },
         ec_priv_pem,
         algorithm="ES256",
         headers={"kid": "ec1"},
@@ -1566,4 +1583,4 @@ async def test_auth_jwt_mismatched_key_fails(monkeypatch):
     with patch.object(h, "get_public_key", new=AsyncMock(return_value=rsa_jwk)):
         with pytest.raises(Exception) as exc:
             await h.auth_jwt(token)
-        assert "Validation fails" in str(exc.value)
+        assert "Validation fails" in str(exc.value)

tests/proxy_unit_tests/test_user_api_key_auth.py

@@ -359,27 +359,38 @@ async def test_auth_with_allowed_routes(route, should_raise_error):
 
 
 @pytest.mark.parametrize(
-    "route, user_role, expected_result",
+    "route, user_role, should_be_allowed",
     [
-        # Proxy Admin checks
+        # Admin can access everything
+        ("/config/update", "proxy_admin", True),
         ("/global/spend/logs", "proxy_admin", True),
-        ("/key/delete", "proxy_admin", False),
-        ("/key/generate", "proxy_admin", False),
-        ("/key/regenerate", "proxy_admin", False),
-        # Internal User checks - allowed routes
+        ("/global/activity/cache_hits", "proxy_admin", True),
+        # Internal User - allowed read-only routes
         ("/global/spend/logs", "internal_user", True),
-        ("/key/delete", "internal_user", False),
-        ("/key/generate", "internal_user", False),
-        ("/key/82akk800000000jjsk/regenerate", "internal_user", False),
-        # Internal User Viewer
-        ("/key/generate", "internal_user_viewer", False),
-        # Internal User checks - disallowed routes
+        ("/spend/logs/ui", "internal_user", True),
+        ("/global/activity/cache_hits", "internal_user", True),
+        ("/health/services", "internal_user", True),
+        # Internal User - BLOCKED from admin routes (security fix)
+        ("/config/update", "internal_user", False),
+        ("/config/pass_through_endpoint", "internal_user", False),
+        ("/config/field/update", "internal_user", False),
         ("/organization/member_add", "internal_user", False),
+        # Internal User Viewer - allowed spend routes only
+        ("/spend/logs/ui", "internal_user_viewer", True),
+        ("/global/spend/all_tag_names", "internal_user_viewer", True),
+        # Internal User Viewer - blocked from admin routes
+        ("/config/update", "internal_user_viewer", False),
+        ("/key/generate", "internal_user_viewer", False),
     ],
 )
-def test_is_ui_route_allowed(route, user_role, expected_result):
-    from litellm.proxy.auth.auth_checks import _is_ui_route
-    from litellm.proxy._types import LiteLLM_UserTable
+def test_ui_token_route_access(route, user_role, should_be_allowed):
+    """
+    Verify that UI tokens (team_id=litellm-dashboard) go through the same
+    RBAC checks as API tokens. Non-admin dashboard users must not be able
+    to access admin-only routes like /config/update.
+    """
+    from litellm.proxy.auth.auth_checks import _is_api_route_allowed
+    from litellm.proxy._types import LiteLLM_UserTable, UserAPIKeyAuth
 
     user_obj = LiteLLM_UserTable(
         user_id="3b803c0e-666e-4e99-bd5c-6e534c07e297",
@@ -395,18 +406,36 @@ def test_is_ui_route_allowed(route, user_role, expected_result):
         organization_memberships=[],
     )
 
-    received_args: dict = {
-        "route": route,
-        "user_obj": user_obj,
-    }
-    try:
-        assert _is_ui_route(**received_args) == expected_result
-    except Exception as e:
-        # If expected result is False, we expect an error
-        if expected_result is False:
-            pass
-        else:
-            raise e
+    valid_token = UserAPIKeyAuth(
+        user_id="3b803c0e-666e-4e99-bd5c-6e534c07e297",
+        team_id="litellm-dashboard",
+        user_role=user_role,
+    )
+
+    from starlette.datastructures import URL
+    from fastapi import Request
+
+    request = Request(scope={"type": "http"})
+    request._url = URL(url=route)
+
+    if should_be_allowed:
+        result = _is_api_route_allowed(
+            route=route,
+            request=request,
+            request_data={},
+            valid_token=valid_token,
+            user_obj=user_obj,
+        )
+        assert result is True
+    else:
+        with pytest.raises(Exception):
+            _is_api_route_allowed(
+                route=route,
+                request=request,
+                request_data={},
+                valid_token=valid_token,
+                user_obj=user_obj,
+            )
 
 
 @pytest.mark.parametrize(
@@ -684,7 +713,7 @@ async def test_soft_budget_alert():
 
 
 def test_is_allowed_route():
-    from litellm.proxy.auth.auth_checks import _is_allowed_route
+    from litellm.proxy.auth.auth_checks import _is_api_route_allowed
     from litellm.proxy._types import UserAPIKeyAuth
     import datetime
 
@@ -692,7 +721,6 @@ def test_is_allowed_route():
 
     args = {
         "route": "/embeddings",
-        "token_type": "api",
         "request": request,
         "request_data": {"input": ["hello world"], "model": "embedding-small"},
         "valid_token": UserAPIKeyAuth(
@@ -752,7 +780,7 @@ def test_is_allowed_route():
         "user_obj": None,
     }
 
-    assert _is_allowed_route(**args)
+    assert _is_api_route_allowed(**args)
 
 
 @pytest.mark.parametrize(
@@ -836,7 +864,6 @@ async def test_user_api_key_auth_websocket():
     with patch(
         "litellm.proxy.auth.user_api_key_auth.user_api_key_auth", autospec=True
     ) as mock_user_api_key_auth:
-
         # Make the call to the WebSocket function
         await user_api_key_auth_websocket(mock_websocket)
 
@@ -845,10 +872,14 @@ async def test_user_api_key_auth_websocket():
 
         # Get the request object that was passed to user_api_key_auth
         request_arg = mock_user_api_key_auth.call_args.kwargs["request"]
-        
+
         # Verify that the request has headers set
-        assert hasattr(request_arg, "headers"), "Request object should have headers attribute"
-        assert "authorization" in request_arg.headers, "Request headers should contain authorization"
+        assert hasattr(
+            request_arg, "headers"
+        ), "Request object should have headers attribute"
+        assert (
+            "authorization" in request_arg.headers
+        ), "Request headers should contain authorization"
         assert request_arg.headers["authorization"] == "Bearer some_api_key"
 
         assert (
@@ -1036,7 +1067,10 @@ async def test_jwt_non_admin_team_route_access(monkeypatch):
 
     # Create request
     request = Request(
-        scope={"type": "http", "headers": [(b"authorization", b"Bearer fake.jwt.token")]}
+        scope={
+            "type": "http",
+            "headers": [(b"authorization", b"Bearer fake.jwt.token")],
+        }
     )
     request._url = URL(url="/team/new")
 
@@ -1101,14 +1135,14 @@ async def test_x_litellm_api_key():
     ignored_key = "aj12445"
 
     # Create request with headers as bytes
-    request = Request(
-        scope={
-            "type": "http"
-        }
-    )
+    request = Request(scope={"type": "http"})
     request._url = URL(url="/chat/completions")
 
-    valid_token = await user_api_key_auth(request=request, api_key="Bearer " + ignored_key, custom_litellm_key_header=master_key)
+    valid_token = await user_api_key_auth(
+        request=request,
+        api_key="Bearer " + ignored_key,
+        custom_litellm_key_header=master_key,
+    )
     assert valid_token.token == hash_token(master_key)
 
 
@@ -1123,7 +1157,9 @@ async def test_user_api_key_from_query_param():
     from litellm.proxy.proxy_server import hash_token, user_api_key_cache
 
     user_key = "sk-query-1234"
-    user_api_key_cache.set_cache(key=hash_token(user_key), value=UserAPIKeyAuth(token=hash_token(user_key)))
+    user_api_key_cache.set_cache(
+        key=hash_token(user_key), value=UserAPIKeyAuth(token=hash_token(user_key))
+    )
 
     setattr(litellm.proxy.proxy_server, "user_api_key_cache", user_api_key_cache)
     setattr(litellm.proxy.proxy_server, "master_key", "sk-1234")
@@ -1136,7 +1172,9 @@ async def test_user_api_key_from_query_param():
             "query_string": f"alt=sse&key={user_key}".encode(),
         }
     )
-    request._url = URL(url=f"/v1beta/models/gemini:streamGenerateContent?alt=sse&key={user_key}")
+    request._url = URL(
+        url=f"/v1beta/models/gemini:streamGenerateContent?alt=sse&key={user_key}"
+    )
 
     async def return_body():
         return b"{}"
@@ -1145,4 +1183,3 @@ async def test_user_api_key_from_query_param():
 
     valid_token = await user_api_key_auth(request=request, api_key="")
     assert valid_token.token == hash_token(user_key)
-