feat(AIP-84): add auth to /ui/backfills

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -358,6 +358,8 @@ paths:
       - Backfill
       summary: List Backfills
       operationId: list_backfills
+      security:
+      - OAuth2PasswordBearer: []
       parameters:
       - name: limit
         in: query

airflow/api_fastapi/core_api/routes/public/backfills.py

@@ -116,7 +116,10 @@ def get_backfill(
             status.HTTP_409_CONFLICT,
         ]
     ),
-    dependencies=[Depends(requires_access_backfill(method="PUT"))],
+    dependencies=[
+        Depends(requires_access_backfill(method="PUT")),
+        Depends(requires_access_dag(method="PUT", access_entity=DagAccessEntity.RUN)),
+    ],
 )
 def pause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
     b = session.get(Backfill, backfill_id)
@@ -138,7 +141,10 @@ def pause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
             status.HTTP_409_CONFLICT,
         ]
     ),
-    dependencies=[Depends(requires_access_backfill(method="PUT"))],
+    dependencies=[
+        Depends(requires_access_backfill(method="PUT")),
+        Depends(requires_access_dag(method="PUT", access_entity=DagAccessEntity.RUN)),
+    ],
 )
 def unpause_backfill(backfill_id, session: SessionDep) -> BackfillResponse:
     b = session.get(Backfill, backfill_id)

airflow/api_fastapi/core_api/routes/ui/backfills.py

@@ -35,6 +35,7 @@
 from airflow.api_fastapi.core_api.openapi.exceptions import (
     create_openapi_http_exception_doc,
 )
+from airflow.api_fastapi.core_api.security import requires_access_backfill, requires_access_dag
 from airflow.models.backfill import Backfill
 
 backfills_router = AirflowRouter(tags=["Backfill"], prefix="/backfills")
@@ -43,6 +44,10 @@
 @backfills_router.get(
     path="",
     responses=create_openapi_http_exception_doc([status.HTTP_404_NOT_FOUND]),
+    dependencies=[
+        Depends(requires_access_backfill(method="GET")),
+        Depends(requires_access_dag(method="GET")),
+    ],
 )
 def list_backfills(
     limit: QueryLimit,

airflow/assets/manager.py

@@ -17,7 +17,7 @@
 # under the License.
 from __future__ import annotations
 
-from collections.abc import Collection, Iterable
+from collections.abc import Collection
 from typing import TYPE_CHECKING
 
 from sqlalchemy import exc, or_, select

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

@@ -63,7 +63,10 @@
 from airflow.providers.fab.auth_manager.models.anonymous_user import AnonymousUser
 from airflow.providers.fab.www.app import create_app
 from airflow.providers.fab.www.constants import SWAGGER_BUNDLE, SWAGGER_ENABLED
-from airflow.providers.fab.www.extensions.init_views import _CustomErrorRequestBodyValidator, _LazyResolver
+from airflow.providers.fab.www.extensions.init_views import (
+    _CustomErrorRequestBodyValidator,
+    _LazyResolver,
+)
 from airflow.providers.fab.www.security import permissions
 from airflow.providers.fab.www.security.permissions import (
     RESOURCE_AUDIT_LOG,
@@ -90,7 +93,10 @@
     RESOURCE_WEBSITE,
     RESOURCE_XCOM,
 )
-from airflow.providers.fab.www.utils import get_fab_action_from_method_map, get_method_from_fab_action_map
+from airflow.providers.fab.www.utils import (
+    get_fab_action_from_method_map,
+    get_method_from_fab_action_map,
+)
 from airflow.security.permissions import RESOURCE_BACKFILL
 from airflow.utils.session import NEW_SESSION, create_session, provide_session
 from airflow.utils.yaml import safe_load
@@ -101,7 +107,9 @@
         CLICommand,
     )
     from airflow.providers.common.compat.assets import AssetAliasDetails, AssetDetails
-    from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
+    from airflow.providers.fab.auth_manager.security_manager.override import (
+        FabAirflowSecurityManagerOverride,
+    )
     from airflow.providers.fab.www.extensions.init_appbuilder import AirflowAppBuilder
     from airflow.providers.fab.www.security.permissions import (
         RESOURCE_ASSET,
@@ -188,7 +196,9 @@ def get_cli_commands() -> list[CLICommand]:
 
     def get_fastapi_app(self) -> FastAPI | None:
         """Get the FastAPI app."""
-        from airflow.providers.fab.auth_manager.api_fastapi.routes.login import login_router
+        from airflow.providers.fab.auth_manager.api_fastapi.routes.login import (
+            login_router,
+        )
 
         flask_app = create_app(enable_plugins=False)
 
@@ -217,7 +227,10 @@ def get_api_endpoints(self) -> None | Blueprint:
             specification=specification,
             resolver=_LazyResolver(),
             base_path="/fab/v1",
-            options={"swagger_ui": SWAGGER_ENABLED, "swagger_path": SWAGGER_BUNDLE.__fspath__()},
+            options={
+                "swagger_ui": SWAGGER_ENABLED,
+                "swagger_path": SWAGGER_BUNDLE.__fspath__(),
+            },
             strict_validation=True,
             validate_responses=True,
             validator_map={"body": _CustomErrorRequestBodyValidator},
@@ -324,7 +337,11 @@ def is_authorized_dag(
             )
 
     def is_authorized_backfill(
-        self, *, method: ResourceMethod, user: User, details: BackfillDetails | None = None
+        self,
+        *,
+        method: ResourceMethod,
+        user: User,
+        details: BackfillDetails | None = None,
     ) -> bool:
         return self._is_authorized(method=method, resource_type=RESOURCE_BACKFILL, user=user)
 
@@ -334,7 +351,11 @@ def is_authorized_asset(
         return self._is_authorized(method=method, resource_type=RESOURCE_ASSET, user=user)
 
     def is_authorized_asset_alias(
-        self, *, method: ResourceMethod, user: User, details: AssetAliasDetails | None = None
+        self,
+        *,
+        method: ResourceMethod,
+        user: User,
+        details: AssetAliasDetails | None = None,
     ) -> bool:
         return self._is_authorized(method=method, resource_type=RESOURCE_ASSET_ALIAS, user=user)
 
@@ -344,15 +365,21 @@ def is_authorized_pool(
         return self._is_authorized(method=method, resource_type=RESOURCE_POOL, user=user)
 
     def is_authorized_variable(
-        self, *, method: ResourceMethod, user: User, details: VariableDetails | None = None
+        self,
+        *,
+        method: ResourceMethod,
+        user: User,
+        details: VariableDetails | None = None,
     ) -> bool:
         return self._is_authorized(method=method, resource_type=RESOURCE_VARIABLE, user=user)
 
     def is_authorized_view(self, *, access_view: AccessView, user: User) -> bool:
         # "Docs" are only links in the menu, there is no page associated
         method: ResourceMethod = "MENU" if access_view == AccessView.DOCS else "GET"
         return self._is_authorized(
-            method=method, resource_type=_MAP_ACCESS_VIEW_TO_FAB_RESOURCE_TYPE[access_view], user=user
+            method=method,
+            resource_type=_MAP_ACCESS_VIEW_TO_FAB_RESOURCE_TYPE[access_view],
+            user=user,
         )
 
     def is_authorized_custom_view(

providers/fab/src/airflow/providers/fab/auth_manager/security_manager/override.py

@@ -110,18 +110,17 @@
     AirflowDatabaseSessionInterface,
     AirflowDatabaseSessionInterface as FabAirflowDatabaseSessionInterface,
 )
+from airflow.security.permissions import RESOURCE_BACKFILL
 
 if TYPE_CHECKING:
     from airflow.providers.fab.www.security.permissions import (
         RESOURCE_ASSET,
         RESOURCE_ASSET_ALIAS,
-        RESOURCE_BACKFILL,
     )
 else:
     from airflow.providers.common.compat.security.permissions import (
         RESOURCE_ASSET,
         RESOURCE_ASSET_ALIAS,
-        RESOURCE_BACKFILL,
     )
 
 log = logging.getLogger(__name__)

tests/api_fastapi/core_api/routes/ui/test_backfills.py

@@ -87,7 +87,7 @@ class TestListBackfills(TestBackfillEndpoint):
             ({"dag_id": "TEST_DAG_1"}, ["backfill1"], 1),
         ],
     )
-    def test_list_backfill(self, test_params, response_params, total_entries, test_client, session):
+    def test_should_response_200(self, test_params, response_params, total_entries, test_client, session):
         dags = self._create_dag_models()
         from_date = timezone.utcnow()
         to_date = timezone.utcnow()
@@ -150,3 +150,11 @@ def test_list_backfill(self, test_params, response_params, total_entries, test_c
             "backfills": expected_response,
             "total_entries": total_entries,
         }
+
+    def test_should_response_401(self, unauthenticated_test_client):
+        response = unauthenticated_test_client.get("/ui/backfills", params={})
+        assert response.status_code == 401
+
+    def test_should_response_403(self, unauthorized_test_client):
+        response = unauthorized_test_client.get("/ui/backfills", params={})
+        assert response.status_code == 403