（CVE-2020-11978）Apache Airflow 命令注入漏洞
=============================================

一、漏洞简介
------------

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

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

![2.png](./resource/(CVE-2020-11978)ApacheAirflow命令注入漏洞/media/rId22.png)

二、漏洞影响
------------

Apache Airflow \< 1.10.11

三、复现过程
------------

### 漏洞分析

首先看下下面两个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 object 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`：![1.png](./resource/(CVE-2020-11978)ApacheAirflow命令注入漏洞/media/rId26.png)

详细信息可以参考Airflow官方文档中对dag\_run的详细说明：

> http://airflow.apache.org/docs/stable/dag-run.html?highlight=dag\_run

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

![3.png](./resource/(CVE-2020-11978)ApacheAirflow命令注入漏洞/media/rId27.png)

![4.png](./resource/(CVE-2020-11978)ApacheAirflow命令注入漏洞/media/rId28.png)

参考链接
--------

> https://xz.aliyun.com/t/8037\#toc-4