Fix AUTH_ROLE_PUBLIC returning 401 in FastAPI API server

providers/fab/src/airflow/providers/fab/auth_manager/middleware.py

@@ -23,6 +23,9 @@
 
 from airflow.api_fastapi.app import get_auth_manager
 from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN
+from airflow.api_fastapi.core_api import security as core_api_security
+
+
 
 if TYPE_CHECKING:
     from fastapi import Request
@@ -44,9 +47,19 @@ class FabAuthRolePublicMiddleware(BaseHTTPMiddleware):
     async def dispatch(self, request: Request, call_next):
         if not self._has_auth_token(request):
             auth_manager = cast("FabAuthManager", get_auth_manager())
+
             public_user = auth_manager.build_public_user()
             if public_user is not None:
                 request.state.user = public_user
+
+                user_injected = getattr(
+                    core_api_security,
+                    "USER_INJECTED_BY_TRUSTED_MIDDLEWARE",
+                    None,
+                )
+                if user_injected is not None:
+                    request.state.user_authenticated_via = user_injected
+
         return await call_next(request)
 
     @staticmethod

providers/fab/tests/unit/fab/auth_manager/test_middleware.py

@@ -21,6 +21,7 @@
 import pytest
 
 from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN
+from airflow.api_fastapi.core_api import security as core_api_security
 from airflow.providers.fab.auth_manager.middleware import FabAuthRolePublicMiddleware
 
 
@@ -47,6 +48,18 @@ async def test_sets_public_user_when_no_auth_present(self, mock_get_auth_manager
         await middleware.dispatch(request, call_next)
 
         assert request.state.user is public_user
+
+        trusted_marker = getattr(
+            core_api_security,
+            "USER_INJECTED_BY_TRUSTED_MIDDLEWARE",
+            None,
+        )
+
+        if trusted_marker is not None:
+            assert request.state.user_authenticated_via is trusted_marker
+        else:
+            assert not hasattr(request.state, "user_authenticated_via")
+
         auth_manager.build_public_user.assert_called_once_with()
         call_next.assert_awaited_once_with(request)