airflow/www/security.py
"""Set the access policy on the given DAG's ViewModel.
        :param dag_id: the ID of the DAG whose permissions should be updated
        :type dag_id: str
        :param access_control: a dict where each key is a rolename and
            each value is a set() of permission names (e.g.,
            {'can_read'}
        :type access_control: dict
        """
        dag_resource_name = permissions.resource_name_for_dag(dag_id)
        def _get_or_create_dag_permission(perm_name):
            dag_perm = self.find_permission_view_menu(perm_name, dag_resource_name)
            if not dag_perm:
                self.log.info("Creating new permission '%s' on view '%s'", perm_name, dag_resource_name)
                dag_perm = self.add_permission_view_menu(perm_name, dag_resource_name)
            return dag_perm
        def _revoke_stale_permissions(dag_view):
            existing_dag_perms = self.find_permissions_view_menu(dag_view)
            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, {})
                    if perm.permission.name not in target_perms_for_role:
                        self.log.info(
                            "Revoking '%s' on DAG '%s' for role '%s'",
                            perm.permission,
                            dag_resource_name,
                            role.name,