跳转至

(CVE-2020-11978)Apache Airflow 命令注入漏洞

一、漏洞简介

默认情况下Airflow Web UI是未授权访问的,直接可以登录,而登录后,只能查看DAG的调度状态等,无法进行更多操作。

但Airflow Web UI中提供了触发DAG运行的功能,以便测试DAG,同时Airflow为了让使用者可以快速熟悉其DAG开发流程和功能,为了更好的示例这些DAG覆盖了大多的执行器。而其中两个DAG组合起来可触发命令注入导致漏洞产生。

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

二、漏洞影响

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:

详细信息可以参考Airflow官方文档中队dag_run的详细说明:

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

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

参考链接

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