Honor AUTH_ROLE_PUBLIC in FastAPI API server

airflow-core/src/airflow/api_fastapi/auth/managers/base_auth_manager.py

@@ -158,6 +158,20 @@ async def get_user_from_token(self, token: str) -> BaseUser:
             log.error("Couldn't deserialize user from token, JWT token is not valid: %s", e)
             raise InvalidTokenError(str(e))
 
+    def get_public_user(self) -> BaseUser | None:
+        """
+        Return a user representing anonymous/public access, or ``None`` if not supported.
+
+        Auth managers that support unauthenticated access (for example, the FAB auth manager's
+        ``[fab] auth_role_public`` configuration) should override this method to return a user
+        object when public access is enabled. When a user is returned, the API server will use it
+        for requests that do not carry any authentication token instead of returning 401.
+
+        By default this method returns ``None``, meaning the auth manager does not permit
+        anonymous access and a valid token is always required.
+        """
+        return None
+
     def generate_jwt(
         self, user: T, *, expiration_time_in_seconds: int = conf.getint("api_auth", "jwt_expiration_time")
     ) -> str:

airflow-core/src/airflow/api_fastapi/core_api/security.py

@@ -114,11 +114,18 @@ def auth_manager_from_app(request: Request) -> BaseAuthManager:
 
 
 async def resolve_user_from_token(token_str: str | None) -> BaseUser:
+    auth_manager = get_auth_manager()
     if not token_str:
+        # When the auth manager supports anonymous/public access (e.g. FAB's
+        # ``[fab] auth_role_public`` setting), fall back to the public user instead of
+        # rejecting the request with 401.
+        public_user = auth_manager.get_public_user()
+        if public_user is not None:
+            return public_user
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated")
 
     try:
-        return await get_auth_manager().get_user_from_token(token_str)
+        return await auth_manager.get_user_from_token(token_str)
     except ExpiredSignatureError:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token Expired")
     except InvalidTokenError:

airflow-core/tests/unit/api_fastapi/core_api/test_security.py

@@ -110,6 +110,34 @@ async def test_get_user_expired_token(self, mock_get_auth_manager):
 
         auth_manager.get_user_from_token.assert_called_once_with(token_str)
 
+    @patch("airflow.api_fastapi.core_api.security.get_auth_manager")
+    async def test_resolve_user_from_token_no_token_raises_when_no_public_user(self, mock_get_auth_manager):
+        """No token and auth manager has no public user → 401."""
+        auth_manager = AsyncMock()
+        auth_manager.get_public_user = Mock(return_value=None)
+        mock_get_auth_manager.return_value = auth_manager
+
+        with pytest.raises(HTTPException, match="Not authenticated"):
+            await resolve_user_from_token(None)
+
+        auth_manager.get_public_user.assert_called_once_with()
+        auth_manager.get_user_from_token.assert_not_called()
+
+    @patch("airflow.api_fastapi.core_api.security.get_auth_manager")
+    async def test_resolve_user_from_token_no_token_returns_public_user(self, mock_get_auth_manager):
+        """No token but auth manager exposes a public user → return it instead of 401."""
+        public_user = SimpleAuthManagerUser(username="public", role="admin")
+
+        auth_manager = AsyncMock()
+        auth_manager.get_public_user = Mock(return_value=public_user)
+        mock_get_auth_manager.return_value = auth_manager
+
+        result = await resolve_user_from_token(None)
+
+        assert result is public_user
+        auth_manager.get_public_user.assert_called_once_with()
+        auth_manager.get_user_from_token.assert_not_called()
+
     @patch("airflow.api_fastapi.core_api.security.resolve_user_from_token")
     async def test_get_user_with_request_state(self, mock_resolve_user_from_token):
         user = Mock()

providers/fab/docs/auth-manager/webserver-authentication.rst

@@ -42,9 +42,27 @@ following CLI commands to create an account:
         --role Admin \
         --email spiderman@superhero.org
 
-To deactivate the authentication and allow users to be identified as Anonymous, the following entry
-in ``$AIRFLOW_HOME/webserver_config.py`` needs to be set with the desired role that the Anonymous
-user will have by default:
+To deactivate the authentication and allow users to be identified as Anonymous, set the
+``[fab] auth_role_public`` Airflow configuration to the desired role that anonymous users will have
+by default. For example, in ``airflow.cfg``:
+
+.. code-block:: ini
+
+    [fab]
+    auth_role_public = Admin
+
+or via the equivalent environment variable:
+
+.. code-block:: bash
+
+    export AIRFLOW__FAB__AUTH_ROLE_PUBLIC=Admin
+
+This makes both the FastAPI-based API server and the legacy Flask-based views honor anonymous
+access consistently.
+
+For backwards compatibility, setting ``AUTH_ROLE_PUBLIC`` in ``$AIRFLOW_HOME/webserver_config.py``
+is still supported, but the ``[fab] auth_role_public`` Airflow config takes precedence when both
+are set and is the recommended configuration going forward:
 
 .. code-block:: ini
 

providers/fab/newsfragments/60897.bugfix.rst

@@ -0,0 +1 @@
+Honor ``AUTH_ROLE_PUBLIC`` in the FastAPI-based API server. Add a new ``[fab] auth_role_public`` Airflow configuration that is used consistently by both the FastAPI and legacy Flask FAB auth stacks so that setting a public role actually removes the login prompt. Setting ``AUTH_ROLE_PUBLIC`` in ``webserver_config.py`` keeps working for backwards compatibility.

providers/fab/provider.yaml

@@ -263,6 +263,22 @@ config:
         type: integer
         example: ~
         default: "30"
+      auth_role_public:
+        description: |
+          Role that Anonymous (unauthenticated) users are granted. When set, the FAB auth manager
+          will allow access to the API server and UI without requiring a login, and anonymous
+          requests will be treated as members of the given role. Leave empty (the default) to
+          require authentication.
+
+          This replaces the previous ``AUTH_ROLE_PUBLIC`` setting in ``webserver_config.py``. When
+          both are set, this ``[fab] auth_role_public`` config takes precedence. Setting this
+          config also applies the equivalent ``AUTH_ROLE_PUBLIC`` to the Flask app used by the FAB
+          auth manager, so all FAB auth code paths (FastAPI-based API server and legacy Flask
+          views) honor it consistently.
+        version_added: 3.6.2
+        type: string
+        example: "Admin"
+        default: ""
 
 auth-managers:
   - airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager

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

@@ -300,11 +300,53 @@ def is_logged_in(self) -> bool:
         """Return whether the user is logged in."""
         user = self.get_user()
         return bool(
-            self.appbuilder
-            and self.appbuilder.app.config.get("AUTH_ROLE_PUBLIC", None)
-            or (not user.is_anonymous and user.is_active)
+            (self.appbuilder and self._get_auth_role_public()) or (not user.is_anonymous and user.is_active)
         )
 
+    def _get_auth_role_public(self) -> str | None:
+        """
+        Return the role granted to anonymous users, or ``None`` if public access is disabled.
+
+        Prefers the Airflow ``[fab] auth_role_public`` configuration and falls back to the legacy
+        ``AUTH_ROLE_PUBLIC`` entry in ``webserver_config.py`` so existing deployments continue to
+        work.
+        """
+        role = conf.get("fab", "auth_role_public", fallback="") or ""
+        if role:
+            return role
+        if self.appbuilder is not None:
+            return self.appbuilder.app.config.get("AUTH_ROLE_PUBLIC", None)
+        return None
+
+    def get_public_user(self) -> AnonymousUser | None:
+        """
+        Return an :class:`AnonymousUser` when public access is enabled, else ``None``.
+
+        Public access is enabled when the Airflow ``[fab] auth_role_public`` config (or, for
+        backwards compatibility, ``AUTH_ROLE_PUBLIC`` in ``webserver_config.py``) is set. The
+        returned user inherits the role configured there and is used by the FastAPI API server
+        to authorize requests that do not carry a JWT token.
+        """
+        public_role_name = self._get_auth_role_public()
+        if not public_role_name:
+            return None
+
+        user = AnonymousUser()
+        flask_app = self.flask_app or (self.appbuilder.app if self.appbuilder else None)
+        if flask_app is not None:
+            with flask_app.app_context():
+                flask_app.config["AUTH_ROLE_PUBLIC"] = public_role_name
+                role = flask_app.appbuilder.sm.find_role(public_role_name)
+                if role is not None:
+                    # FAB's ``AnonymousUser.roles`` is a lazy property that calls
+                    # ``security_manager.get_public_role()`` on every access, which needs a Flask
+                    # request context we do not have under FastAPI. Writing ``_roles``/``_perms``
+                    # directly freezes a snapshot of the public role's permissions for the
+                    # lifetime of a single FastAPI authorization check (see #60897).
+                    user._roles = {role}
+                    user._perms = {(perm.action.name, perm.resource.name) for perm in role.permissions}
+        return user
+
     def create_token(self, headers: dict[str, str], body: dict[str, Any]) -> User | None:
         """
         Create a new token from a payload.

providers/fab/src/airflow/providers/fab/www/app.py

@@ -84,6 +84,15 @@ def remove_duplicate_date_header(response):
     with flask_app.app_context():
         flask_app.config.from_pyfile(webserver_config, silent=True)
 
+    # Bridge ``[fab] auth_role_public`` Airflow config into the Flask app config so legacy FAB
+    # code paths that read ``AUTH_ROLE_PUBLIC`` from ``current_app.config`` (e.g.
+    # ``AnonymousUser.roles``, basic_auth, kerberos_auth, security manager) stay in sync with
+    # the FastAPI-based auth flow. The Airflow config takes precedence over
+    # ``webserver_config.py`` when both are set so there is a single source of truth.
+    auth_role_public_conf = conf.get("fab", "auth_role_public", fallback="") or ""
+    if auth_role_public_conf:
+        flask_app.config["AUTH_ROLE_PUBLIC"] = auth_role_public_conf
+
     url = make_url(flask_app.config["SQLALCHEMY_DATABASE_URI"])
     if url.drivername == "sqlite" and url.database and not isabs(url.database):
         raise AirflowConfigException(

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

@@ -262,6 +262,90 @@ def test_is_logged_in_with_inactive_user(self, mock_get_user, auth_manager_with_
 
         assert auth_manager_with_appbuilder.is_logged_in() is False
 
+    @mock.patch.object(FabAuthManager, "get_user")
+    def test_is_logged_in_with_auth_role_public_conf(self, mock_get_user, auth_manager_with_appbuilder):
+        """``[fab] auth_role_public`` Airflow config is enough to be 'logged in'."""
+        user = Mock()
+        user.is_anonymous.return_value = True
+        user.is_active.return_value = False
+        mock_get_user.return_value = user
+
+        with conf_vars({("fab", "auth_role_public"): "Admin"}):
+            assert auth_manager_with_appbuilder.is_logged_in() is True
+
+    def test_get_public_user_returns_none_when_not_configured(self, auth_manager_with_appbuilder):
+        """Without ``[fab] auth_role_public`` or ``AUTH_ROLE_PUBLIC``, there is no public user."""
+        flask_app = auth_manager_with_appbuilder.flask_app or (auth_manager_with_appbuilder.appbuilder.app)
+        previous = flask_app.config.get("AUTH_ROLE_PUBLIC")
+        flask_app.config["AUTH_ROLE_PUBLIC"] = None
+        try:
+            with conf_vars({("fab", "auth_role_public"): ""}):
+                assert auth_manager_with_appbuilder.get_public_user() is None
+        finally:
+            flask_app.config["AUTH_ROLE_PUBLIC"] = previous
+
+    def test_get_public_user_returns_anonymous_user_from_conf(self, auth_manager_with_appbuilder):
+        """``[fab] auth_role_public`` yields an :class:`AnonymousUser` with the role resolved."""
+        from airflow.providers.fab.auth_manager.models.anonymous_user import AnonymousUser
+
+        with conf_vars({("fab", "auth_role_public"): "Admin"}):
+            user = auth_manager_with_appbuilder.get_public_user()
+
+        assert isinstance(user, AnonymousUser)
+        assert len(user.roles) == 1
+        assert user.roles[0].name == "Admin"
+        # ``perms`` must be pre-populated so FastAPI can evaluate authorization outside a
+        # Flask request/app context.
+        assert user._perms, "Expected permissions to be pre-populated on the public user"
+
+    def test_get_public_user_falls_back_to_flask_config(self, auth_manager_with_appbuilder):
+        """Legacy ``AUTH_ROLE_PUBLIC`` in ``webserver_config.py`` is still honored."""
+        from airflow.providers.fab.auth_manager.models.anonymous_user import AnonymousUser
+
+        flask_app = auth_manager_with_appbuilder.flask_app or (auth_manager_with_appbuilder.appbuilder.app)
+        previous = flask_app.config.get("AUTH_ROLE_PUBLIC")
+        flask_app.config["AUTH_ROLE_PUBLIC"] = "Admin"
+        try:
+            with conf_vars({("fab", "auth_role_public"): ""}):
+                user = auth_manager_with_appbuilder.get_public_user()
+        finally:
+            flask_app.config["AUTH_ROLE_PUBLIC"] = previous
+
+        assert isinstance(user, AnonymousUser)
+        assert len(user.roles) == 1
+        assert user.roles[0].name == "Admin"
+
+    def test_get_public_user_with_unknown_role_returns_user_with_empty_perms(
+        self, auth_manager_with_appbuilder
+    ):
+        """A misconfigured role name still yields an :class:`AnonymousUser` with empty perms."""
+        from airflow.providers.fab.auth_manager.models.anonymous_user import AnonymousUser
+
+        with conf_vars({("fab", "auth_role_public"): "DoesNotExist"}):
+            user = auth_manager_with_appbuilder.get_public_user()
+
+        assert isinstance(user, AnonymousUser)
+        # No pre-populated perms means the user has no authorization; the role will also not
+        # be resolved since ``find_role`` returned None for a non-existent role.
+        assert user._perms == set()
+
+    def test_get_public_user_airflow_conf_takes_precedence(self, auth_manager_with_appbuilder):
+        """When both ``[fab] auth_role_public`` and the Flask config are set, Airflow config wins."""
+        from airflow.providers.fab.auth_manager.models.anonymous_user import AnonymousUser
+
+        flask_app = auth_manager_with_appbuilder.flask_app or (auth_manager_with_appbuilder.appbuilder.app)
+        previous = flask_app.config.get("AUTH_ROLE_PUBLIC")
+        flask_app.config["AUTH_ROLE_PUBLIC"] = "Viewer"
+        try:
+            with conf_vars({("fab", "auth_role_public"): "Admin"}):
+                user = auth_manager_with_appbuilder.get_public_user()
+        finally:
+            flask_app.config["AUTH_ROLE_PUBLIC"] = previous
+
+        assert isinstance(user, AnonymousUser)
+        assert len(user.roles) == 1
+        assert user.roles[0].name == "Admin"
+
     @pytest.mark.parametrize(
         ("auth_type", "method"),
         [