airflow/www/decorators.py
def has_dag_access(**dag_kwargs):
    """
    Decorator to check whether the user has read / write permission on the dag.
    """
    def decorator(f):
        @functools.wraps(f)
        def wrapper(self, *args, **kwargs):
            has_access = self.appbuilder.sm.has_access
            dag_id = request.args.get('dag_id')
            # if it is false, we need to check whether user has write access on the dag
            can_dag_edit = dag_kwargs.get('can_dag_edit', False)

            # 1. check whether the user has can_dag_edit permissions on all_dags
            # 2. if 1 false, check whether the user
            #    has can_dag_edit permissions on the dag
            # 3. if 2 false, check whether it is can_dag_read view,
            #    and whether user has the permissions
            if (
                has_access('can_dag_edit', 'all_dags') or
                has_access('can_dag_edit', dag_id) or (not can_dag_edit and
                                                       (has_access('can_dag_read',
                                                                   'all_dags') or
                                                        has_access('can_dag_read',
                                                                   dag_id)))):
                return f(self, *args, **kwargs)
            else:
                flash("Access is Denied", "danger")
                return redirect(url_for(self.appbuilder.sm.auth_view.
                                        __class__.__name__ + ".login"))
        return wrapper
    return decorator
def has_dag_access(**dag_kwargs):
    """
    Decorator to check whether the user has read / write permission on the dag.
    """
    def decorator(f):
        @functools.wraps(f)
        def wrapper(self, *args, **kwargs):
            has_access = self.appbuilder.sm.has_access
            dag_id = request.args.get('dag_id')
            # if it is false, we need to check whether user has write access on the dag
            can_dag_edit = dag_kwargs.get('can_dag_edit', False)

            # 1. check whether the user has can_dag_edit permissions on all_dags
            # 2. if 1 false, check whether the user
            #    has can_dag_edit permissions on the dag
            # 3. if 2 false, check whether it is can_dag_read view,
            #    and whether user has the permissions
            if (
                has_access('can_dag_edit', 'all_dags') or
                has_access('can_dag_edit', dag_id) or (not can_dag_edit and
                                                       (has_access('can_dag_read',
                                                                   'all_dags') or
                                                        has_access('can_dag_read',
                                                                   dag_id)))):
                return f(self, *args, **kwargs)
            else:
                flash("Access is Denied", "danger")
                return redirect(url_for(self.appbuilder.sm.auth_view.
                                        __class__.__name__ + ".login"))
        return wrapper
    return decorator
Decorator to check whether the user has read / write permission on the dag.