AIP-84 Add Auth for DAG Versioning

airflow/api_fastapi/auth/managers/models/resource_details.py

@@ -100,7 +100,8 @@ class DagAccessEntity(Enum):
     SLA_MISS = "SLA_MISS"
     TASK = "TASK"
     TASK_INSTANCE = "TASK_INSTANCE"
-    TASK_RESCHEDULE = "TASK_RESCHEDULE"
     TASK_LOGS = "TASK_LOGS"
+    TASK_RESCHEDULE = "TASK_RESCHEDULE"
+    VERSION = "VERSION"
     WARNING = "WARNING"
     XCOM = "XCOM"

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

@@ -7036,6 +7036,8 @@ paths:
       summary: Get Dag Version
       description: Get one Dag Version.
       operationId: get_dag_version
+      security:
+      - OAuth2PasswordBearer: []
       parameters:
       - name: dag_id
         in: path
@@ -7091,6 +7093,8 @@ paths:
         This endpoint allows specifying `~` as the dag_id to retrieve DAG Versions
         for all DAGs.'
       operationId: get_dag_versions
+      security:
+      - OAuth2PasswordBearer: []
       parameters:
       - name: dag_id
         in: path

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

@@ -21,6 +21,7 @@
 from fastapi import Depends, HTTPException, Request, status
 from sqlalchemy import select
 
+from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
 from airflow.api_fastapi.common.parameters import (
     FilterParam,
@@ -35,6 +36,7 @@
     DagVersionResponse,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
+from airflow.api_fastapi.core_api.security import requires_access_dag
 from airflow.models.dag import DAG
 from airflow.models.dag_version import DagVersion
 
@@ -48,6 +50,7 @@
             status.HTTP_404_NOT_FOUND,
         ]
     ),
+    dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.VERSION))],
 )
 def get_dag_version(
     dag_id: str,
@@ -71,8 +74,9 @@ def get_dag_version(
     responses=create_openapi_http_exception_doc(
         [
             status.HTTP_404_NOT_FOUND,
-        ]
+        ],
     ),
+    dependencies=[Depends(requires_access_dag(method="GET", access_entity=DagAccessEntity.VERSION))],
 )
 def get_dag_versions(
     dag_id: str,

airflow/security/permissions.py

@@ -32,6 +32,7 @@
 RESOURCE_DAG_PREFIX = "DAG:"
 RESOURCE_DAG_RUN = "DAG Runs"
 RESOURCE_DAG_RUN_PREFIX = "DAG Run:"
+RESOURCE_DAG_VERSION = "DAG Versions"
 RESOURCE_DAG_WARNING = "DAG Warnings"
 RESOURCE_CLUSTER_ACTIVITY = "Cluster Activity"
 RESOURCE_ASSET = "Assets"

providers/amazon/tests/unit/amazon/aws/auth_manager/test_aws_auth_manager.py

@@ -412,7 +412,10 @@ def test_batch_is_authorized_dag(
             requests=[
                 {"method": "GET"},
                 {"method": "GET", "details": DagDetails(id="dag_1")},
-                {"method": "GET", "details": DagDetails(id="dag_1"), "access_entity": DagAccessEntity.CODE},
+            ]
+            + [
+                {"method": "GET", "details": DagDetails(id="dag_1"), "access_entity": dag_access_entity}
+                for dag_access_entity in DagAccessEntity
             ],
             user=mock,
         )
@@ -431,16 +434,28 @@ def test_batch_is_authorized_dag(
                     "entity_id": "dag_1",
                     "context": None,
                 },
+            ]
+            + [
                 {
                     "method": "GET",
                     "entity_type": AvpEntities.DAG,
                     "entity_id": "dag_1",
-                    "context": {
-                        "dag_entity": {
-                            "string": DagAccessEntity.CODE.value,
-                        },
-                    },
-                },
+                    "context": {"dag_entity": {"string": dag_entity}},
+                }
+                for dag_entity in (
+                    DagAccessEntity.AUDIT_LOG.value,
+                    DagAccessEntity.CODE.value,
+                    DagAccessEntity.DEPENDENCIES.value,
+                    DagAccessEntity.RUN.value,
+                    DagAccessEntity.SLA_MISS.value,
+                    DagAccessEntity.TASK.value,
+                    DagAccessEntity.TASK_INSTANCE.value,
+                    DagAccessEntity.TASK_LOGS.value,
+                    DagAccessEntity.TASK_RESCHEDULE.value,
+                    DagAccessEntity.VERSION.value,
+                    DagAccessEntity.WARNING.value,
+                    DagAccessEntity.XCOM.value,
+                )
             ],
             user=ANY,
         )

providers/common/compat/src/airflow/providers/common/compat/security/permissions.py

@@ -19,12 +19,22 @@
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
-    from airflow.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS, RESOURCE_BACKFILL
+    from airflow.security.permissions import (
+        RESOURCE_ASSET,
+        RESOURCE_ASSET_ALIAS,
+        RESOURCE_BACKFILL,
+        RESOURCE_DAG_VERSION,
+    )
 else:
     try:
-        from airflow.security.permissions import RESOURCE_ASSET, RESOURCE_ASSET_ALIAS, RESOURCE_BACKFILL
+        from airflow.security.permissions import (
+            RESOURCE_ASSET,
+            RESOURCE_ASSET_ALIAS,
+            RESOURCE_BACKFILL,
+            RESOURCE_DAG_VERSION,
+        )
     except ImportError:
         from airflow.security.permissions import RESOURCE_DATASET as RESOURCE_ASSET
 
 
-__all__ = ["RESOURCE_ASSET", "RESOURCE_ASSET_ALIAS", "RESOURCE_BACKFILL"]
+__all__ = ["RESOURCE_ASSET", "RESOURCE_ASSET_ALIAS", "RESOURCE_BACKFILL", "RESOURCE_DAG_VERSION"]

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

@@ -74,6 +74,7 @@
     RESOURCE_DAG_CODE,
     RESOURCE_DAG_DEPENDENCIES,
     RESOURCE_DAG_RUN,
+    RESOURCE_DAG_VERSION,
     RESOURCE_DAG_WARNING,
     RESOURCE_DOCS,
     RESOURCE_IMPORT_ERROR,
@@ -129,6 +130,7 @@
     DagAccessEntity.TASK_INSTANCE: (RESOURCE_DAG_RUN, RESOURCE_TASK_INSTANCE),
     DagAccessEntity.TASK_LOGS: (RESOURCE_TASK_LOG,),
     DagAccessEntity.TASK_RESCHEDULE: (RESOURCE_TASK_RESCHEDULE,),
+    DagAccessEntity.VERSION: (RESOURCE_DAG_VERSION,),
     DagAccessEntity.WARNING: (RESOURCE_DAG_WARNING,),
     DagAccessEntity.XCOM: (RESOURCE_XCOM,),
 }

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

@@ -246,13 +246,14 @@ class FabAirflowSecurityManagerOverride(AirflowSecurityManagerV2):
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_DEPENDENCIES),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_CODE),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_VERSION),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_WARNING),
         (permissions.ACTION_CAN_READ, RESOURCE_ASSET),
         (permissions.ACTION_CAN_READ, RESOURCE_ASSET_ALIAS),
         (permissions.ACTION_CAN_READ, RESOURCE_BACKFILL),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_CLUSTER_ACTIVITY),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_POOL),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_IMPORT_ERROR),
-        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_WARNING),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_JOB),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_MY_PASSWORD),
         (permissions.ACTION_CAN_EDIT, permissions.RESOURCE_MY_PASSWORD),

providers/fab/src/airflow/providers/fab/www/security/permissions.py

@@ -32,6 +32,7 @@
 RESOURCE_DAG_PREFIX = "DAG:"
 RESOURCE_DAG_RUN = "DAG Runs"
 RESOURCE_DAG_RUN_PREFIX = "DAG Run:"
+RESOURCE_DAG_VERSION = "DAG Versions"
 RESOURCE_DAG_WARNING = "DAG Warnings"
 RESOURCE_CLUSTER_ACTIVITY = "Cluster Activity"
 RESOURCE_ASSET = "Assets"

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

@@ -426,12 +426,13 @@ def test_get_user_roles_for_anonymous_user(app, security_manager):
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_DEPENDENCIES),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_CODE),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_RUN),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_VERSION),
+        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_WARNING),
         (permissions.ACTION_CAN_READ, RESOURCE_ASSET),
         (permissions.ACTION_CAN_READ, RESOURCE_ASSET_ALIAS),
         (permissions.ACTION_CAN_READ, RESOURCE_BACKFILL),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_CLUSTER_ACTIVITY),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_IMPORT_ERROR),
-        (permissions.ACTION_CAN_READ, permissions.RESOURCE_DAG_WARNING),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_JOB),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_POOL),
         (permissions.ACTION_CAN_READ, permissions.RESOURCE_SLA_MISS),

tests/api_fastapi/core_api/routes/public/test_dag_versions.py

@@ -117,6 +117,18 @@ def test_get_dag_version_404(self, test_client):
             "detail": "The DagVersion with dag_id: `dag_with_multiple_versions` and version_number: `99` was not found",
         }
 
+    def test_should_respond_401(self, unauthenticated_test_client):
+        response = unauthenticated_test_client.get(
+            "/public/dags/dag_with_multiple_versions/dagVersions/99", params={}
+        )
+        assert response.status_code == 401
+
+    def test_should_respond_403(self, unauthorized_test_client):
+        response = unauthorized_test_client.get(
+            "/public/dags/dag_with_multiple_versions/dagVersions/99", params={}
+        )
+        assert response.status_code == 403
+
 
 class TestGetDagVersions(TestDagVersionEndpoint):
     @pytest.mark.parametrize(
@@ -298,3 +310,11 @@ def test_get_dag_versions_should_return_404_for_missing_dag(self, test_client):
         assert response.json() == {
             "detail": "The DAG with dag_id: `MISSING_ID` was not found",
         }
+
+    def test_should_respond_401(self, unauthenticated_test_client):
+        response = unauthenticated_test_client.get("/public/dags/~/dagVersions", params={})
+        assert response.status_code == 401
+
+    def test_should_respond_403(self, unauthorized_test_client):
+        response = unauthorized_test_client.get("/public/dags/~/dagVersions", params={})
+        assert response.status_code == 403