airflow/www/security.py
def _sync_dag_view_permissions(self, dag_id, access_control):
        """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: string
        :param access_control: a dict where each key is a rolename and
            each value is a set() of permission names (e.g.,
            {'can_dag_read'}
        :type access_control: dict
        """
        def _get_or_create_dag_permission(perm_name):
            dag_perm = self.find_permission_view_menu(perm_name, dag_id)
            if not dag_perm:
                self.log.info(
                    "Creating new permission '%s' on view '%s'",
                    perm_name, dag_id
                )
                dag_perm = self.add_permission_view_menu(perm_name, dag_id)

            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_id, role.name
                        )
                        self.del_permission_role(role, perm)

        dag_view = self.find_view_menu(dag_id)
        if dag_view:
            _revoke_stale_permissions(dag_view)

        for rolename, perms in access_control.items():
            role = self.find_role(rolename)
            if not role:
                raise AirflowException(
                    "The access_control mapping for DAG '{}' includes a role "
                    "named '{}', but that role does not exist".format(
                        dag_id,
                        rolename))

            perms = set(perms)
            invalid_perms = perms - self.DAG_PERMS
            if invalid_perms:
                raise AirflowException(
                    "The access_control map for DAG '{}' includes the following "
                    "invalid permissions: {}; The set of valid permissions "
                    "is: {}".format(dag_id,
                                    (perms - self.DAG_PERMS),
                                    self.DAG_PERMS))

            for perm_name in perms:
                dag_perm = _get_or_create_dag_permission(perm_name)
                self.add_permission_role(role, dag_perm)
def _sync_dag_view_permissions(self, dag_id, access_control):
        """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: string
        :param access_control: a dict where each key is a rolename and
            each value is a set() of permission names (e.g.,
            {'can_dag_read'}
        :type access_control: dict
        """
        def _get_or_create_dag_permission(perm_name):
            dag_perm = self.find_permission_view_menu(perm_name, dag_id)
            if not dag_perm:
                self.log.info(
                    "Creating new permission '%s' on view '%s'",
                    perm_name, dag_id
                )
                dag_perm = self.add_permission_view_menu(perm_name, dag_id)

            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_id, role.name
                        )
                        self.del_permission_role(role, perm)

        dag_view = self.find_view_menu(dag_id)
        if dag_view:
            _revoke_stale_permissions(dag_view)

        for rolename, perms in access_control.items():
            role = self.find_role(rolename)
            if not role:
                raise AirflowException(
                    "The access_control mapping for DAG '{}' includes a role "
                    "named '{}', but that role does not exist".format(
                        dag_id,
                        rolename))

            perms = set(perms)
            invalid_perms = perms - self.DAG_PERMS
            if invalid_perms:
                raise AirflowException(
                    "The access_control map for DAG '{}' includes the following "
                    "invalid permissions: {}; The set of valid permissions "
                    "is: {}".format(dag_id,
                                    (perms - self.DAG_PERMS),
                                    self.DAG_PERMS))

            for perm_name in perms:
                dag_perm = _get_or_create_dag_permission(perm_name)
                self.add_permission_role(role, dag_perm)
def _sync_dag_view_permissions ( self , dag_id , access_control ) : def _get_or_create_dag_permission ( perm_name ) : dag_perm = self . find_permission_view_menu ( perm_name , dag_id ) if not dag_perm : self . log . info ( "Creating new permission '%s' on view '%s'" , perm_name , dag_id ) dag_perm = self . add_permission_view_menu ( perm_name , dag_id ) 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_id , role . name ) self . del_permission_role ( role , perm ) dag_view = self . find_view_menu ( dag_id ) if dag_view : _revoke_stale_permissions ( dag_view ) for rolename , perms in access_control . items ( ) : role = self . find_role ( rolename ) if not role : raise AirflowException ( "The access_control mapping for DAG '{}' includes a role " "named '{}', but that role does not exist" . format ( dag_id , rolename ) ) perms = set ( perms ) invalid_perms = perms - self . DAG_PERMS if invalid_perms : raise AirflowException ( "The access_control map for DAG '{}' includes the following " "invalid permissions: {}; The set of valid permissions " "is: {}" . format ( dag_id , ( perms - self . DAG_PERMS ) , self . DAG_PERMS ) ) for perm_name in perms : dag_perm = _get_or_create_dag_permission ( perm_name ) self . add_permission_role ( role , dag_perm )
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: string
        :param access_control: a dict where each key is a rolename and
            each value is a set() of permission names (e.g.,
            {'can_dag_read'}
        :type access_control: dict
Set the access policy on the given DAG s ViewModel .