Update is_authorized_dag method in FabAuthManager

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

@@ -103,6 +103,7 @@ def inner(
         user: GetUserDep,
     ) -> None:
         dag_id: str | None = request.path_params.get("dag_id")
+        dag_id = dag_id if dag_id != "~" else None
         team_name = DagModel.get_team_name(dag_id) if dag_id else None
 
         _requires_access(

airflow-core/src/airflow/security/permissions.py

@@ -41,7 +41,6 @@
 RESOURCE_DAG_DEPENDENCIES = "DAG Dependencies"
 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"
@@ -98,30 +97,30 @@ class ResourceDetails(TypedDict):
     ),
     RESOURCE_DAG_RUN: ResourceDetails(
         actions={ACTION_CAN_READ, ACTION_CAN_CREATE, ACTION_CAN_DELETE, ACTION_CAN_ACCESS_MENU},
-        prefix=RESOURCE_DAG_RUN_PREFIX,
+        prefix="DAG Run:",
     ),
 }
 PREFIX_LIST = [details["prefix"] for details in RESOURCE_DETAILS_MAP.values()]
 PREFIX_RESOURCES_MAP = {details["prefix"]: resource for resource, details in RESOURCE_DETAILS_MAP.items()}
 
 
-def resource_name(root_dag_id: str, resource: str) -> str:
+def resource_name(dag_id: str, resource: str) -> str:
     """Return the resource name for a DAG id."""
-    if root_dag_id in RESOURCE_DETAILS_MAP.keys():
-        return root_dag_id
-    if root_dag_id.startswith(tuple(PREFIX_RESOURCES_MAP.keys())):
-        return root_dag_id
-    return f"{RESOURCE_DETAILS_MAP[resource]['prefix']}{root_dag_id}"
+    if dag_id in RESOURCE_DETAILS_MAP.keys():
+        return dag_id
+    if dag_id.startswith(tuple(PREFIX_RESOURCES_MAP.keys())):
+        return dag_id
+    return f"{RESOURCE_DETAILS_MAP[resource]['prefix']}{dag_id}"
 
 
-def resource_name_for_dag(root_dag_id: str) -> str:
+def resource_name_for_dag(dag_id: str) -> str:
     """
     Return the resource name for a DAG id.
 
     Note: This function is kept for backwards compatibility.
     """
-    if root_dag_id == RESOURCE_DAG:
-        return root_dag_id
-    if root_dag_id.startswith(RESOURCE_DAG_PREFIX):
-        return root_dag_id
-    return f"{RESOURCE_DAG_PREFIX}{root_dag_id}"
+    if dag_id == RESOURCE_DAG:
+        return dag_id
+    if dag_id.startswith(RESOURCE_DAG_PREFIX):
+        return dag_id
+    return f"{RESOURCE_DAG_PREFIX}{dag_id}"

providers/fab/docs/auth-manager/access-control.rst

@@ -139,9 +139,9 @@ Dag-level permissions
 ^^^^^^^^^^^^^^^^^^^^^
 
 For Dag-level permissions exclusively, access can be controlled at the level of all Dags or individual Dag objects.
-This includes ``DAGs.can_read``, ``DAGs.can_edit``, ``DAGs.can_delete``, ``DAG Runs.can_read``, ``DAG Runs.can_create``, ``DAG Runs.can_delete``, and ``DAG Runs.menu_access``.
+This includes ``DAGs.can_read``, ``DAGs.can_edit`` and ``DAGs.can_delete``.
 When these permissions are listed, access is granted to users who either have the listed permission or the same permission for the specific Dag being acted upon.
-For individual Dags, the resource name is ``Dag:`` + the Dag ID, or for the Dag Runs resource the resource name is ``Dag Run:``.
+For individual Dags, the resource name is ``Dag:`` + the Dag ID.
 
 For example, if a user is trying to view Dag information for the ``example_dag_id``, and the endpoint requires ``DAGs.can_read`` access, access will be granted if the user has either ``DAGs.can_read`` or ``DAG:example_dag_id.can_read`` access.
 

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

@@ -75,13 +75,15 @@
 )
 from airflow.providers.fab.www.security import permissions
 from airflow.providers.fab.www.security.permissions import (
+    ACTION_CAN_READ,
     RESOURCE_AUDIT_LOG,
     RESOURCE_CLUSTER_ACTIVITY,
     RESOURCE_CONFIG,
     RESOURCE_CONNECTION,
     RESOURCE_DAG,
     RESOURCE_DAG_CODE,
     RESOURCE_DAG_DEPENDENCIES,
+    RESOURCE_DAG_PREFIX,
     RESOURCE_DAG_RUN,
     RESOURCE_DAG_VERSION,
     RESOURCE_DAG_WARNING,
@@ -323,40 +325,52 @@ def is_authorized_dag(
 
         There are multiple scenarios:
 
-        1. ``access_entity`` is not provided which means the user wants to access the DAG itself and not a sub
-        entity (e.g. DAG runs).
-        2. ``access_entity`` is provided which means the user wants to access a sub entity of the DAG
-        (e.g. DAG runs).
+        1. ``method`` is "GET" and no details are provided which means the user wants to list Dags (or sub entities of Dags).
+        2. ``access_entity`` is not provided which means the user wants to access the DAG itself and not a sub
+            entity (e.g. Task instances).
+        3. ``access_entity`` is provided which means the user wants to access a sub entity of the DAG
+            (e.g. DAG runs).
 
             a. If ``method`` is GET, then check the user has READ permissions on the DAG and the sub entity.
-            b. Else, check the user has EDIT permissions on the DAG and ``method`` on the sub entity. However,
-                if no specific DAG is targeted, just check the sub entity.
+            b. Else, check the user has EDIT permissions on the DAG and ``method`` on the sub entity.
 
         :param method: The method to authorize.
         :param user: The user performing the action.
         :param access_entity: The dag access entity.
         :param details: The dag details.
         """
-        if not access_entity:
+        if access_entity:
+            # If a sub-Dag entity is specified, check whether the user has access to it
+            resource_types = self._get_fab_resource_types(access_entity)
+            access_entity_authorized = all(
+                self._is_authorized(method=method, resource_type=resource_type, user=user)
+                for resource_type in resource_types
+            )
+            if access_entity == DagAccessEntity.RUN and details and details.id:
+                # Check using the deprecated permission prefix "DAG Run" to check whether the user has access to dag runs
+                is_authorized_run = self._is_authorized(
+                    method=method,
+                    resource_type=permissions.resource_name(details.id, RESOURCE_DAG_RUN),
+                    user=user,
+                )
+                if not (is_authorized_run or access_entity_authorized):
+                    return False
+            elif not access_entity_authorized:
+                return False
+
+        if method == "GET" and (not details or not details.id):
             # Scenario 1
+            return self._is_authorized_list_dags(user=user)
+        if not access_entity:
+            # Scenario 2
             return self._is_authorized_dag(method=method, details=details, user=user)
-        # Scenario 2
-        resource_types = self._get_fab_resource_types(access_entity)
+        # Scenario 3
         dag_method: ResourceMethod = "GET" if method == "GET" else "PUT"
-
         if (details and details.id) and not self._is_authorized_dag(
             method=dag_method, details=details, user=user
         ):
             return False
-
-        return all(
-            (
-                self._is_authorized(method=method, resource_type=resource_type, user=user)
-                if resource_type != RESOURCE_DAG_RUN or not hasattr(permissions, "resource_name")
-                else self._is_authorized_dag_run(method=method, details=details, user=user)
-            )
-            for resource_type in resource_types
-        )
+        return True
 
     def is_authorized_backfill(
         self,
@@ -545,7 +559,7 @@ def _is_authorized(
 
         :param method: the method to perform
         :param resource_type: the type of resource the user attempts to perform the action on
-        :param user: the user to performing the action
+        :param user: the user performing the action
 
         :meta private:
         """
@@ -565,7 +579,7 @@ def _is_authorized_dag(
 
         :param method: the method to perform
         :param details: details about the DAG
-        :param user: the user to performing the action
+        :param user: the user performing the action
 
         :meta private:
         """
@@ -577,34 +591,25 @@ def _is_authorized_dag(
             # Check whether the user has permissions to access a specific DAG
             resource_dag_name = permissions.resource_name(details.id, RESOURCE_DAG)
             return self._is_authorized(method=method, resource_type=resource_dag_name, user=user)
-        authorized_dags = self.get_authorized_dag_ids(user=user, method=method)
-        return len(authorized_dags) > 0
+        return False
 
-    def _is_authorized_dag_run(
-        self,
-        method: ResourceMethod,
-        details: DagDetails | None,
-        user: User,
-    ) -> bool:
+    def _is_authorized_list_dags(self, *, user: User) -> bool:
         """
-        Return whether the user is authorized to perform a given action on a DAG Run.
+        Return whether the user is authorized to list Dags.
 
-        :param method: the method to perform
-        :param details: details about the DAG
-        :param user: the user to performing the action
+        :param user: the user performing the action
 
         :meta private:
         """
-        is_global_authorized = self._is_authorized(method=method, resource_type=RESOURCE_DAG_RUN, user=user)
+        is_global_authorized = self._is_authorized(method="GET", resource_type=RESOURCE_DAG, user=user)
         if is_global_authorized:
             return True
 
-        if details and details.id:
-            # Check whether the user has permissions to access a specific DAG Run permission on a DAG Level
-            resource_dag_name = permissions.resource_name(details.id, RESOURCE_DAG_RUN)
-            return self._is_authorized(method=method, resource_type=resource_dag_name, user=user)
-        authorized_dags = self.get_authorized_dag_ids(user=user, method=method)
-        return len(authorized_dags) > 0
+        user_permissions = self._get_user_permissions(user)
+        for action, resource in user_permissions:
+            if action == ACTION_CAN_READ and resource.startswith(RESOURCE_DAG_PREFIX):
+                return True
+        return False
 
     @staticmethod
     def _get_fab_action(method: ExtendedResourceMethod) -> str:

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

@@ -31,7 +31,6 @@
 RESOURCE_DAG_DEPENDENCIES = "DAG Dependencies"
 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"
@@ -88,17 +87,17 @@ class ResourceDetails(TypedDict):
     ),
     RESOURCE_DAG_RUN: ResourceDetails(
         actions={ACTION_CAN_READ, ACTION_CAN_CREATE, ACTION_CAN_DELETE, ACTION_CAN_ACCESS_MENU},
-        prefix=RESOURCE_DAG_RUN_PREFIX,
+        prefix="DAG Run:",
     ),
 }
 PREFIX_LIST = [details["prefix"] for details in RESOURCE_DETAILS_MAP.values()]
 PREFIX_RESOURCES_MAP = {details["prefix"]: resource for resource, details in RESOURCE_DETAILS_MAP.items()}
 
 
-def resource_name(root_dag_id: str, resource: str) -> str:
+def resource_name(dag_id: str, resource: str) -> str:
     """Return the resource name for a DAG id."""
-    if root_dag_id in RESOURCE_DETAILS_MAP.keys():
-        return root_dag_id
-    if root_dag_id.startswith(tuple(PREFIX_RESOURCES_MAP.keys())):
-        return root_dag_id
-    return f"{RESOURCE_DETAILS_MAP[resource]['prefix']}{root_dag_id}"
+    if dag_id in RESOURCE_DETAILS_MAP.keys():
+        return dag_id
+    if dag_id.startswith(tuple(PREFIX_RESOURCES_MAP.keys())):
+        return dag_id
+    return f"{RESOURCE_DETAILS_MAP[resource]['prefix']}{dag_id}"

providers/fab/src/airflow/providers/fab/www/static/dist/manifest.json

@@ -1,17 +1,17 @@
 {
-  "airflowDefaultTheme.css": "airflowDefaultTheme.56d4475fdae7883d3454.css",
-  "airflowDefaultTheme.js": "airflowDefaultTheme.56d4475fdae7883d3454.js",
-  "flash.css": "flash.0951d47c62bc8906be65.css",
-  "flash.js": "flash.0951d47c62bc8906be65.js",
-  "loadingDots.css": "loadingDots.deaad0ce0e7691ed6251.css",
-  "loadingDots.js": "loadingDots.deaad0ce0e7691ed6251.js",
-  "main.css": "main.810554d06c3e30f2484e.css",
-  "main.js": "main.810554d06c3e30f2484e.js",
-  "materialIcons.css": "materialIcons.b0c6cc32cdacff89f7c2.css",
-  "materialIcons.js": "materialIcons.b0c6cc32cdacff89f7c2.js",
-  "moment.js": "moment.518a43bcfaf149ae2836.js",
-  "runtime.js": "runtime.4a925577de9ab84d8e00.js",
-  "743.js": "743.27a753a06671118f1c5c.js",
+  "airflowDefaultTheme.css": "airflowDefaultTheme.ff5a35f322070b094aa2.css",
+  "airflowDefaultTheme.js": "airflowDefaultTheme.ff5a35f322070b094aa2.js",
+  "flash.css": "flash.5583a9e0cf11f2be93da.css",
+  "flash.js": "flash.5583a9e0cf11f2be93da.js",
+  "loadingDots.css": "loadingDots.2e5f555f0753107b0300.css",
+  "loadingDots.js": "loadingDots.2e5f555f0753107b0300.js",
+  "main.css": "main.3cf3be1a0c5439bb640d.css",
+  "main.js": "main.3cf3be1a0c5439bb640d.js",
+  "materialIcons.css": "materialIcons.3e67dd6fbfcc4f3b5105.css",
+  "materialIcons.js": "materialIcons.3e67dd6fbfcc4f3b5105.js",
+  "moment.js": "moment.9baee5ec3d7639a10897.js",
+  "runtime.js": "runtime.ad800fc1845ad5c6ddeb.js",
+  "743.js": "743.fc7a7c6ef9d09365976e.js",
   "jquery-ui.min.js": "jquery-ui.min.js",
   "jquery-ui.min.css": "jquery-ui.min.css",
   "oss-licenses.json": "oss-licenses.json",