红帽社区是一个垂直网络安全社区，融合“红帽先锋”正能量精神，每日分享最新安全资讯，提供安全问答、靶场、众测、漏洞库等功能，是网络安全爱好者学习、交流的优质社区。

前言

Airflow 是一个使用 python 语言编写的 data pipeline 调度和监控工作流的平台。 Airflow 是通过 DAG（Directed acyclic graph 有向无环图）来管理任务流程的任务调度工具， 不需要知道业务数据的具体内容，设置任务的依赖关系即可实现任务调度。

这个平台拥有和 Hive、Presto、MySQL、HDFS、Postgres 等数据源之间交互的能力，并且提供了钩子（hook）使其拥有很好地扩展性。 除了一个命令行界面，该工具还提供了一个基于 Web 的用户界面可以可视化管道的依赖关系、监控进度、触发任务等。

项目地址：https://github.com/apache/airflow

漏洞环境

#启动容器
docker run --rm --entrypoint '' -v /tmp/airflow/:/root/airflow  --name airflow -it apache/airflow:master-ci /bin/bash
#进入容器进行初始化配置
airflow db init
nohup airflow webserver &
nohup airflow scheduler &


#默认使用Sqlite如需要使用mysql修改配置后重新初始化DB即可
vi /root/airflow/airflow.cfg
修改：sql_alchemy_conn = mysql+mysqldb://root:password@mysqlip:3306/airflow
#高版本下Mysql需要注意DB需要为UTF8mb3，否则会创建索引的时候超过长度
CREATE DATABASE airflow CHARACTER SET UTF8mb3 COLLATE utf8_general_ci

参考：https://airflow.apache.org/docs/stable/start.html

漏洞分析

漏洞描述

参考：
https://lists.apache.org/thread.html/r7255cf0be3566f23a768e2a04b40fb09e52fcd1872695428ba9afe91%40%3Cusers.airflow.apache.org%3E

默认情况下Airflow Web UI是未授权访问的，直接可以登录，而登录后，只能查看DAG的调度状态等，无法进行更多操作。
但Airflow Web UI中提供了触发DAG运行的功能，以便测试DAG，同时Airflow为了让使用者可以快速熟悉其DAG开发流程和功能，为了更好的示例这些DAG覆盖了大多的执行器。而其中两个DAG组合起来可触发命令注入导致漏洞产生。

详细分析

首先看下下面两个DAG

#airflow/example_dags/example_trigger_target_dag.py
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago

dag = DAG(
    dag_id="example_trigger_target_dag"
    default_args={"start_date": days_ago(2) "owner": "airflow"}
    schedule_interval=None
    tags=['example']
)


def run_this_func(**context):
    """
    Print the payload "message" passed to the DagRun conf attribute.

    :param context: The execution context
    :type context: dict
    """
    print("Remotely received value of {} for key=message".format(context["dag_run"].conf["message"]))


run_this = PythonOperator(task_id="run_this" python_callable=run_this_func dag=dag)

bash_task = BashOperator(
    task_id="bash_task"
    bash_command='echo "Here is the message: \'{{ dag_run.conf["message"] if dag_run else "" }}\'"'
    dag=dag
)

#airflow/example_dags/example_trigger_controller_dag.py
from airflow import DAG
from airflow.operators.dagrun_operator import TriggerDagRunOperator
from airflow.utils.dates import days_ago

dag = DAG(
    dag_id="example_trigger_controller_dag"
    default_args={"owner": "airflow" "start_date": days_ago(2)}
    schedule_interval="@once"
    tags=['example']
)

trigger = TriggerDagRunOperator(
    task_id="test_trigger_dagrun"
    trigger_dag_id="example_trigger_target_dag"  # Ensure this equals the dag_id of the DAG to trigger
    conf={"message": "Hello World"}
    dag=dag
)

官方对这两个DAG的说明如下：

Example usage of the TriggerDagRunOperator. This example holds 2 DAGs:
1. 1st DAG (example_trigger_controller_dag) holds a TriggerDagRunOperator which will trigger the 2nd DAG
2. 2nd DAG (example_trigger_target_dag) which will be triggered by the TriggerDagRunOperator in the 1st DAG

可以看出Airflow希望通过这两个DAG组合来展示如果通过一个DAG（example_trigger_controller_dag）来动态的调用另外一个DAG（example_trigger_target_dag）。即通过example_trigger_controller_dag内部定义的conf={"message": "Hello World"}来触发example_trigger_target_dag中bash_command='echo "Here is the message: \'{{ dag_run.conf["message"] if dag_run else "" }}\'"'的运行，此处看起来：
存在命令执行点'echo "Here is the message: \'{{ dag_run.conf["message"] if dag_run else "" }}\'"'
这边是Python下面的Jinja模板，因此会根据后面的if...else逻辑来执行dag_run.conf["message"] 来动态加载内容，此处如果dag_run.conf["message"] 可控，则可以通过Jinja模板注入恶意命令。

但根据上面信息可以看出，输入dag_run.conf["message"]由第一个DGA传递过来的，看起来无法控制。而实际上熟悉下Airflow相关代码即可发现，Airflow中A DAG Run is an representing an instantiation of the DAG in time.而其中conf 正是用于传递参数的方式， Airflow提供了多渠道可以修改conf，包括命令行例如：

airflow dags trigger --conf '{"conf1": "value1"}' example_parametrized_dag

同时也包含Web UI 上直接触发任意DAG并传递dag_run.conf：

详细信息可以参考Airflow官方文档中队dag_run的详细说明：
http://airflow.apache.org/docs/stable/dag-run.html?highlight=dag_run

因此可以直接利用此接口触发example_trigger_target_dag.py的调度，这样就可以绕过example_trigger_controller_dag中写死的配置。

注意：
要在WEB UI中先执行下启用DAG，然后才可以执行运行，如下所示

修复方式

1、升级到1.10.10之后版本
2、删除或禁用默认DAG（可自行删除或在配置文件中禁用默认DAGload_examples=False）

时间线

2020-05-31 发现并上报漏洞给Apache Airflow安全团队
2020-06-01 Apache Airflow安全团队确认漏洞
2020-07-10 发布修复版本Airflow 1.10.11（https://github.com/apache/airflow/releases/tag/1.10.11）
2020-07-14 分配CVE-2020-11978