Merge 8fd6f39fa3 into c850ed3c8e

litellm/proxy/auth/route_checks.py

@@ -13,6 +13,9 @@ from litellm.proxy._types import (
     LitellmUserRoles,
     UserAPIKeyAuth,
 )
+from litellm.types.passthrough_endpoints.pass_through_endpoints import (
+    LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
+)
 
 from .auth_checks_organization import _user_is_org_admin
 
@@ -262,7 +265,11 @@ class RouteChecks:
             method=RouteChecks._get_request_method(request=request),
         ):
             RouteChecks._require_auth_pass_through_access(route=route, valid_token=valid_token)
-        elif RouteChecks.is_llm_api_route(route=route):
+        elif RouteChecks.is_llm_api_route(route=route) or RouteChecks.is_public_pass_through_route(
+            route=route,
+            method=RouteChecks._get_request_method(request=request),
+            request=request,
+        ):
             pass
         elif RouteChecks.is_info_route(route=route):
             # check if user allowed to call an info route
@@ -645,6 +652,30 @@ class RouteChecks:
             return False
         return route_info.get("auth") is True
 
+    @staticmethod
+    def is_public_pass_through_route(
+        route: str,
+        method: str | None = None,
+        request: Request | None = None,
+    ) -> bool:
+        if request is None:
+            return False
+        scope: Final = getattr(request, "scope", None)
+        if not isinstance(scope, dict):
+            return False
+        endpoint: Final = scope.get("endpoint")
+        if getattr(endpoint, LITELLM_PASS_THROUGH_ENDPOINT_MARKER, False) is not True:
+            return False
+
+        from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
+            InitPassThroughEndpointHelpers,
+        )
+
+        route_info: Final = InitPassThroughEndpointHelpers.get_registered_pass_through_route(route=route, method=method)
+        if route_info is None:
+            return False
+        return route_info.get("auth") is False
+
     @staticmethod
     def _auth_pass_through_denied_exception(route: str) -> HTTPException:
         return HTTPException(

tests/test_litellm/proxy/auth/test_route_checks.py

@@ -1180,6 +1180,141 @@ def test_non_proxy_admin_allows_auth_pass_through_with_team_allowlist():
         )
 
 
+@pytest.mark.parametrize(
+    "route,registered_type",
+    [
+        ("/myprefix", "exact"),
+        ("/myprefix/models", "subpath"),
+    ],
+)
+def test_non_proxy_admin_allows_non_auth_pass_through_with_llm_api_routes(
+    route, registered_type
+):
+    mock_registered_routes = {
+        f"test-uuid-1:{registered_type}:/myprefix:GET": {
+            "endpoint_id": "test-uuid-1",
+            "path": "/myprefix",
+            "type": registered_type,
+            "methods": ["GET"],
+            "auth": False,
+        },
+    }
+    valid_token = UserAPIKeyAuth(
+        user_id="test_user",
+        user_role=LitellmUserRoles.INTERNAL_USER.value,
+        allowed_routes=["llm_api_routes"],
+    )
+    from litellm.types.passthrough_endpoints.pass_through_endpoints import (
+        LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
+    )
+
+    def pass_through_endpoint(): ...
+
+    setattr(pass_through_endpoint, LITELLM_PASS_THROUGH_ENDPOINT_MARKER, True)
+    request = Request(
+        scope={
+            "type": "http",
+            "headers": [],
+            "method": "GET",
+            "query_string": b"",
+            "endpoint": pass_through_endpoint,
+        }
+    )
+
+    with (
+        patch(
+            "litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
+            mock_registered_routes,
+        ),
+        patch(
+            "litellm.proxy.utils.get_server_root_path",
+            return_value="/",
+        ),
+    ):
+        assert (
+            RouteChecks.is_virtual_key_allowed_to_call_route(
+                route=route,
+                valid_token=valid_token,
+                request=request,
+            )
+            is True
+        )
+        RouteChecks.non_proxy_admin_allowed_routes_check(
+            user_obj=None,
+            _user_role=LitellmUserRoles.INTERNAL_USER.value,
+            route=route,
+            request=request,
+            valid_token=valid_token,
+            request_data={},
+        )
+
+
+@pytest.mark.parametrize(
+    "registered_path,registered_type",
+    [
+        ("/config/update", "exact"),
+        ("/config", "subpath"),
+    ],
+)
+def test_public_pass_through_registration_does_not_bypass_builtin_route_auth(
+    registered_path, registered_type
+):
+    mock_registered_routes = {
+        f"test-uuid-1:{registered_type}:{registered_path}:GET": {
+            "endpoint_id": "test-uuid-1",
+            "path": registered_path,
+            "type": registered_type,
+            "methods": ["GET"],
+            "auth": False,
+        },
+    }
+    valid_token = UserAPIKeyAuth(
+        user_id="test_user",
+        user_role=LitellmUserRoles.INTERNAL_USER.value,
+        allowed_routes=["llm_api_routes"],
+    )
+
+    def builtin_config_update(): ...
+
+    request = Request(
+        scope={
+            "type": "http",
+            "headers": [],
+            "method": "GET",
+            "query_string": b"",
+            "endpoint": builtin_config_update,
+        }
+    )
+
+    with (
+        patch(
+            "litellm.proxy.pass_through_endpoints.pass_through_endpoints._registered_pass_through_routes",
+            mock_registered_routes,
+        ),
+        patch(
+            "litellm.proxy.utils.get_server_root_path",
+            return_value="/",
+        ),
+    ):
+        assert (
+            RouteChecks.is_virtual_key_allowed_to_call_route(
+                route="/config/update",
+                valid_token=valid_token,
+                request=request,
+            )
+            is True
+        )
+        with pytest.raises(Exception, match="Only proxy admin"):
+            RouteChecks.non_proxy_admin_allowed_routes_check(
+                user_obj=None,
+                _user_role=LitellmUserRoles.INTERNAL_USER.value,
+                route="/config/update",
+                request=request,
+                valid_token=valid_token,
+                request_data={},
+            )
+
+
 def test_virtual_key_without_llm_api_routes_cannot_access_pass_through():
     """
     Test that virtual keys without llm_api_routes permission cannot access registered pass-through endpoints.