Implement simple_auth_manager_all_admins in simple auth manager with new auth flow

airflow/api_fastapi/auth/managers/simple/openapi/v1-generated.yaml

@@ -7,6 +7,25 @@ info:
   version: 0.1.0
 paths:
   /auth/token:
+    get:
+      tags:
+      - SimpleAuthManagerLogin
+      summary: Create Token All Admins
+      description: Create a token with no credentials only if ``simple_auth_manager_all_admins``
+        is True.
+      operationId: create_token_all_admins
+      responses:
+        '307':
+          description: Successful Response
+          content:
+            application/json:
+              schema: {}
+        '403':
+          description: Forbidden
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
     post:
       tags:
       - SimpleAuthManagerLogin

airflow/api_fastapi/auth/managers/simple/routes/login.py

@@ -17,10 +17,13 @@
 
 from __future__ import annotations
 
-from fastapi import status
+from fastapi import HTTPException, status
+from starlette.responses import RedirectResponse
 
+from airflow.api_fastapi.app import get_auth_manager
 from airflow.api_fastapi.auth.managers.simple.datamodels.login import LoginBody, LoginResponse
 from airflow.api_fastapi.auth.managers.simple.services.login import SimpleAuthManagerLogin
+from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
 from airflow.api_fastapi.common.router import AirflowRouter
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.configuration import conf
@@ -40,6 +43,27 @@ def create_token(
     return SimpleAuthManagerLogin.create_token(body=body)
 
 
+@login_router.get(
+    "/token",
+    status_code=status.HTTP_307_TEMPORARY_REDIRECT,
+    responses=create_openapi_http_exception_doc([status.HTTP_403_FORBIDDEN]),
+)
+def create_token_all_admins() -> RedirectResponse:
+    """Create a token with no credentials only if ``simple_auth_manager_all_admins`` is True."""
+    is_simple_auth_manager_all_admins = conf.getboolean("core", "simple_auth_manager_all_admins")
+    if not is_simple_auth_manager_all_admins:
+        raise HTTPException(
+            status.HTTP_403_FORBIDDEN,
+            "This method is only allowed if ``[core] simple_auth_manager_all_admins`` is True",
+        )
+    user = SimpleAuthManagerUser(
+        username="Anonymous",
+        role="ADMIN",
+    )
+    url = f"{conf.get('api', 'base_url')}/?token={get_auth_manager().get_jwt_token(user)}"
+    return RedirectResponse(url=url)
+
+
 @login_router.post(
     "/token/cli",
     status_code=status.HTTP_201_CREATED,

airflow/api_fastapi/auth/managers/simple/simple_auth_manager.py

@@ -112,6 +112,9 @@ def get_passwords(users: list[dict[str, str]]) -> dict[str, str]:
         }
 
     def init(self) -> None:
+        is_simple_auth_manager_all_admins = conf.getboolean("core", "simple_auth_manager_all_admins")
+        if is_simple_auth_manager_all_admins:
+            return
         users = self.get_users()
         passwords = self.get_passwords(users)
         for user in users:
@@ -126,7 +129,11 @@ def init(self) -> None:
 
     def get_url_login(self, **kwargs) -> str:
         """Return the login page url."""
-        return "/auth/webapp/login"
+        is_simple_auth_manager_all_admins = conf.getboolean("core", "simple_auth_manager_all_admins")
+        if is_simple_auth_manager_all_admins:
+            return "/auth/token"
+
+        return "/auth/login"
 
     def deserialize_user(self, token: dict[str, Any]) -> SimpleAuthManagerUser:
         return SimpleAuthManagerUser(username=token["username"], role=token["role"])
@@ -258,7 +265,7 @@ def get_fastapi_app(self) -> FastAPI | None:
             name="simple_auth_manager_ui_folder",
         )
 
-        @app.get("/webapp/{rest_of_path:path}", response_class=HTMLResponse, include_in_schema=False)
+        @app.get("/{rest_of_path:path}", response_class=HTMLResponse, include_in_schema=False)
         def webapp(request: Request, rest_of_path: str):
             return templates.TemplateResponse("/index.html", {"request": request}, media_type="text/html")
 

airflow/api_fastapi/auth/managers/simple/ui/src/router.tsx

@@ -32,6 +32,6 @@ export const router = createBrowserRouter(
     },
   ],
   {
-    basename: "/auth/webapp",
+    basename: "/auth",
   },
 );

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

@@ -25,6 +25,7 @@
 from sqlalchemy.engine.url import make_url
 
 from airflow import settings
+from airflow.api_fastapi.app import get_auth_manager
 from airflow.configuration import conf
 from airflow.exceptions import AirflowConfigException
 from airflow.logging_config import configure_logging
@@ -49,6 +50,8 @@
 
 def create_app(enable_plugins: bool):
     """Create a new instance of Airflow WWW app."""
+    from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
+
     flask_app = Flask(__name__)
     flask_app.secret_key = conf.get("webserver", "SECRET_KEY")
     flask_app.config["SQLALCHEMY_DATABASE_URI"] = conf.get("database", "SQL_ALCHEMY_CONN")
@@ -77,7 +80,14 @@ def create_app(enable_plugins: bool):
     with flask_app.app_context():
         init_appbuilder(flask_app, enable_plugins=enable_plugins)
         init_error_handlers(flask_app)
-        if enable_plugins:
+        # In two scenarios a Flask application can be created:
+        # - To support Airflow 2 plugins relying on Flask (``enable_plugins`` is True)
+        # - To support FAB auth manager (``enable_plugins`` is False)
+        # There are some edge cases where ``enable_plugins`` is False but the auth manager configured is not
+        # FAB auth manager. One example is ``run_update_fastapi_api_spec``, it calls
+        # ``FabAuthManager().get_fastapi_app()`` to generate the openapi documentation regardless of the
+        # configured auth manager.
+        if enable_plugins or not isinstance(get_auth_manager(), FabAuthManager):
             init_plugins(flask_app)
         else:
             init_api_auth_provider(flask_app)

tests/api_fastapi/auth/managers/simple/conftest.py

@@ -18,16 +18,24 @@
 
 from __future__ import annotations
 
+import os
+
 import pytest
 from fastapi.testclient import TestClient
 
+from airflow.api_fastapi.app import create_app
 from airflow.api_fastapi.auth.managers.simple.simple_auth_manager import SimpleAuthManager
 from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser
 
+from tests_common.test_utils.config import conf_vars
+
 
 @pytest.fixture
 def auth_manager():
-    return SimpleAuthManager(None)
+    auth_manager = SimpleAuthManager()
+    if os.path.exists(auth_manager.get_generated_password_file()):
+        os.remove(auth_manager.get_generated_password_file())
+    return auth_manager
 
 
 @pytest.fixture
@@ -41,5 +49,13 @@ def test_admin():
 
 
 @pytest.fixture
-def test_client(auth_manager):
-    return TestClient(auth_manager.get_fastapi_app())
+def test_client():
+    with conf_vars(
+        {
+            (
+                "core",
+                "auth_manager",
+            ): "airflow.api_fastapi.auth.managers.simple.simple_auth_manager.SimpleAuthManager"
+        }
+    ):
+        return TestClient(create_app("core"))

tests/api_fastapi/auth/managers/simple/routes/test_login.py

@@ -24,6 +24,8 @@
 
 from airflow.api_fastapi.auth.managers.simple.datamodels.login import LoginResponse
 
+from tests_common.test_utils.config import conf_vars
+
 TEST_USER_1 = "test1"
 TEST_USER_2 = "test2"
 
@@ -41,20 +43,31 @@ def test_create_token(self, mock_simple_auth_manager_login, test_client, auth_ma
         mock_simple_auth_manager_login.create_token.return_value = LoginResponse(jwt_token="DUMMY_TOKEN")
 
         response = test_client.post(
-            "/token",
+            "/auth/token",
             json={"username": test_user, "password": "DUMMY_PASS"},
         )
         assert response.status_code == 201
         assert response.json()["jwt_token"]
 
     def test_create_token_invalid_user_password(self, test_client):
         response = test_client.post(
-            "/token",
+            "/auth/token",
             json={"username": "INVALID_USER", "password": "INVALID_PASS"},
         )
         assert response.status_code == 401
         assert response.json()["detail"] == "Invalid credentials"
 
+    def test_create_token_all_admins(self, test_client):
+        with conf_vars({("core", "simple_auth_manager_all_admins"): "true"}):
+            response = test_client.get("/auth/token", follow_redirects=False)
+            assert response.status_code == 307
+            assert "location" in response.headers
+            assert response.headers["location"].startswith("http://localhost:8080/?token=")
+
+    def test_create_token_all_admins_config_disabled(self, test_client):
+        response = test_client.get("/auth/token")
+        assert response.status_code == 403
+
     @pytest.mark.parametrize(
         "test_user",
         [
@@ -67,15 +80,15 @@ def test_create_token_cli(self, mock_simple_auth_manager_login, test_client, aut
         mock_simple_auth_manager_login.create_token.return_value = LoginResponse(jwt_token="DUMMY_TOKEN")
 
         response = test_client.post(
-            "/token/cli",
+            "/auth/token/cli",
             json={"username": test_user, "password": "DUMMY_PASS"},
         )
         assert response.status_code == 201
         assert response.json()["jwt_token"]
 
     def test_create_token_invalid_user_password_cli(self, test_client):
         response = test_client.post(
-            "/token/cli",
+            "/auth/token/cli",
             json={"username": "INVALID_USER", "password": "INVALID_PASS"},
         )
         assert response.status_code == 401

tests/api_fastapi/auth/managers/simple/test_simple_auth_manager.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import json
+import os
 
 import pytest
 
@@ -57,9 +58,19 @@ def test_init_with_users(self, auth_manager):
 
                 assert len(user_passwords_from_file) == 2
 
+    def test_init_with_all_admins(self, auth_manager):
+        with conf_vars({("core", "simple_auth_manager_all_admins"): "true"}):
+            auth_manager.init()
+            assert not os.path.exists(auth_manager.get_generated_password_file())
+
     def test_get_url_login(self, auth_manager):
         result = auth_manager.get_url_login()
-        assert result == "/auth/webapp/login"
+        assert result == "/auth/login"
+
+    def test_get_url_login_with_all_admins(self, auth_manager):
+        with conf_vars({("core", "simple_auth_manager_all_admins"): "true"}):
+            result = auth_manager.get_url_login()
+            assert result == "/auth/token"
 
     def test_deserialize_user(self, auth_manager):
         result = auth_manager.deserialize_user({"username": "test", "role": "admin"})