在 BashOperator 中将密码作为参数发送时如何屏蔽密码

Question

我正在使用 BashOperator 运行需要多个参数才能工作的 bash 命令。在这些参数中，我发送了一个密码。

bash_task = BashOperator(
    task_id='bash_operator_example',
    bash_command=f'echo {param1} {password}',
)

这里的问题是，即使我从应保密的气流连接中获取该密码，但当操作员运行命令时，密码会在日志中以明确的测试显示。

查看下面生成的日志：

[2021-03-06 17:42:14,079] {bash_operator.py:136} INFO - Temporary script location: /tmp/airflowtmp_mez12q_/bash_operator_examplena7hxdup
[2021-03-06 17:42:14,079] {bash_operator.py:146} INFO - Running command: echo my_param_1 my_password
[2021-03-06 17:42:14,092] {bash_operator.py:153} INFO - Output:
[2021-03-06 17:42:14,094] {bash_operator.py:157} INFO - my_param_1 my_password
[2021-03-06 17:42:14,094] {bash_operator.py:161} INFO - Command exited with return code 0
[2021-03-06 17:42:14,101] {taskinstance.py:1070} INFO - Marking task as SUCCESS.dag_id=TEST_running_DAG, task_id=bash_operator_example, execution_date=20210101T000000, start_date=20210306T174214, end_date=20210306T174214

我尝试了一些方法，例如，使用 bash 运算符中可用的参数，例如：

bash_task = BashOperator(
        task_id='bash_operator_example',
        bash_command='echo {{ params.param1 }} {{ params.password }}',
        params={
            "param1": param1,
            "password": password
        }
    )

但结果还是一样，它会打印在日志中。我也尝试过使用 env，但虽然不在日志中，但完整的 env 在 UI 的“任务实例详细信息”选项中可用，因此密码也可以在那里看到。

似乎可行的一件事是使用子进程库在 PythonOperator 中运行 bash 命令。但我认为 Airflow 应该有一个更简单的选项，我只是不知道。

如果任何有 Airflow 经验的人能指出我正确的方向，将不胜感激。

2020-03-07 更新：

最后，我所做的是编写自己的 Operator。我仍然认为必须有一个更简单的解决方案，因为这应该是一个非常常见的用例。 这就是我所做的：

创建一个新的 Operator，您可以使用本指南了解如何创建自己的 Operator：

https://airflow.apache.org/docs/apache-airflow/stable/howto/custom-operator.html

然后，我基本上将 init 和 execute 都替换为基本运算符的代码，您可以在此处找到：

https://airflow.apache.org/docs/apache-airflow/1.10.3/_modules/airflow/operators/bash_operator.html

最后，我在 init 方法中添加了一个新的参数密码，并使用该参数替换了 self.bash_command，如下所示：

with NamedTemporaryFile(dir=tmp_dir, prefix=self.task_id) as f:
        #Custom Code to replace the password
        final_bash_command = self.bash_command.replace(':password', 
        self.password) if self.password else self.bash_command
    
        f.write(bytes(final_bash_command, 'utf_8'))
        ...

如果参数密码存在，它会搜索占位符 :password 并将其替换为参数“password”

不是最优雅的解决方案，但它使技巧在后面的代码中打印出 self.bash_command 而不是我的新变量 final_bash_command ，这是实际执行的。

这就是我使用新运算符的方式：

hello_task = PasswordBashOperator(
        task_id='sample-task',
        bash_command='echo {{ params.param1 }} :password',
        params={
            "param1": param1
        },
        password=password
)

现在，如果我运行它打印的任务：

[2021-03-07 16:16:55,786] {password_bash_operator.py:64} INFO - Temporary script location: /tmp/airflowtmpfos7yapf/sample-task_da9_lrr
[2021-03-07 16:16:55,786] {password_bash_operator.py:74} INFO - Running command: echo my_param_1 :password
[2021-03-07 16:16:55,792] {password_bash_operator.py:83} INFO - Output:
[2021-03-07 16:16:55,793] {password_bash_operator.py:87} INFO - my_param_1 my_password
[2021-03-07 16:16:55,793] {password_bash_operator.py:91} INFO - Command exited with return code 0
[2021-03-07 16:16:55,799] {taskinstance.py:1070} INFO - Marking task as SUCCESS.dag_id=TEST_running_DAG, task_id=sample-task, execution_date=20210101T000000, start_date=20210307T161655, end_date=20210307T161655

所以，这行：

[2021-03-07 16:16:55,786] {password_bash_operator.py:74} INFO - 运行命令：echo my_param_1 :password

不再打印密码。

如果有人能找到更好、更优雅的解决方案，我仍然会感兴趣。

Answer 1

我也遇到了这个问题，并设法以不同的方式解决它。我想我会分享我所做的，因为这是我在寻找解决方案时发现的第一篇文章。

我使用BashOperator 中的env 参数来设置一个可由bash 命令使用的变量，如下所示：

PASSWORD = "my_password"

some_task = BashOperator(task_id="push_files",
                         bash_command="echo $password,
                         env={'password': PASSWORD})

这会导致 Airflow 日志在将要运行的命令打印到日志时不显示密码字符串：

[2021-11-29 11:19:07,527] {bash_operator.py:147} INFO - Running command: echo $PASSWORD
[2021-11-29 11:19:07,674] {bash_operator.py:154} INFO - Output: 
[2021-11-29 11:19:07,678] {bash_operator.py:158} INFO - my_password

我在 GCPs Cloud Composer 中使用 Airflow 1.10.15 并找到了这个解决方案 in the docs。