Fix revoke Dag stale permission on airflow < 2.10 (apache#42844)

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

@@ -1107,7 +1107,7 @@ def is_dag_resource(self, resource_name: str) -> bool:
     def sync_perm_for_dag(
         self,
         dag_id: str,
-        access_control: dict[str, dict[str, Collection[str]]] | None = None,
+        access_control: dict[str, dict[str, Collection[str]] | Collection[str]] | None = None,
     ) -> None:
         """
         Sync permissions for given dag id.
@@ -1149,7 +1149,7 @@ def _resource_name(self, dag_id: str, resource_name: str) -> str:
     def _sync_dag_view_permissions(
         self,
         dag_id: str,
-        access_control: dict[str, dict[str, Collection[str]]],
+        access_control: dict[str, dict[str, Collection[str]] | Collection[str]],
     ) -> None:
         """
         Set the access policy on the given DAG's ViewModel.
@@ -1175,7 +1175,13 @@ def _get_or_create_dag_permission(action_name: str, dag_resource_name: str) -> P
                 for perm in existing_dag_perms:
                     non_admin_roles = [role for role in perm.role if role.name != "Admin"]
                     for role in non_admin_roles:
-                        target_perms_for_role = access_control.get(role.name, {}).get(resource_name, set())
+                        access_control_role = access_control.get(role.name)
+                        target_perms_for_role = set()
+                        if access_control_role:
+                            if isinstance(access_control_role, set):
+                                target_perms_for_role = access_control_role
+                            elif isinstance(access_control_role, dict):
+                                target_perms_for_role = access_control_role.get(resource_name, set())
                         if perm.action.name not in target_perms_for_role:
                             self.log.info(
                                 "Revoking '%s' on DAG '%s' for role '%s'",
@@ -1194,7 +1200,7 @@ def _get_or_create_dag_permission(action_name: str, dag_resource_name: str) -> P
                     f"'{rolename}', but that role does not exist"
                 )
 
-            if isinstance(resource_actions, (set, list)):
+            if not isinstance(resource_actions, dict):
                 # Support for old-style access_control where only the actions are specified
                 resource_actions = {permissions.RESOURCE_DAG: set(resource_actions)}
 

providers/tests/fab/auth_manager/test_security.py

@@ -865,11 +865,22 @@ def test_access_control_is_set_on_init(
             )
 
 
+@pytest.mark.parametrize(
+    "access_control_before, access_control_after",
+    [
+        (READ_WRITE, READ_ONLY),
+        # old access control format
+        ({permissions.ACTION_CAN_READ, permissions.ACTION_CAN_EDIT}, {permissions.ACTION_CAN_READ}),
+    ],
+    ids=["new_access_control_format", "old_access_control_format"],
+)
 def test_access_control_stale_perms_are_revoked(
     app,
     security_manager,
     assert_user_has_dag_perms,
     assert_user_does_not_have_dag_perms,
+    access_control_before,
+    access_control_after,
 ):
     username = "access_control_stale_perms_are_revoked"
     role_name = "team-a"
@@ -882,12 +893,12 @@ def test_access_control_stale_perms_are_revoked(
         ) as user:
             set_user_single_role(app, user, role_name="team-a")
             security_manager._sync_dag_view_permissions(
-                "access_control_test", access_control={"team-a": READ_WRITE}
+                "access_control_test", access_control={"team-a": access_control_before}
             )
             assert_user_has_dag_perms(perms=["GET", "PUT"], dag_id="access_control_test", user=user)
 
             security_manager._sync_dag_view_permissions(
-                "access_control_test", access_control={"team-a": READ_ONLY}
+                "access_control_test", access_control={"team-a": access_control_after}
             )
             # Clear the cache, to make it pick up new rol perms
             user._perms = None