我之前维护Airflow集群时也碰到过几乎一模一样的难题——依赖服务的凭证每小时就过期，还只能通过内部命令行工具刷新，一开始确实卡了好一阵子。给你分享几个我亲测可行的解决方案，你可以根据自己的环境选最合适的：

1. 自定义Hook封装凭证刷新逻辑（最推荐） #

Airflow的Hook机制天生适合处理这种动态凭证场景，把获取最新密码的逻辑封装到自定义Hook里，每次任务执行时都会主动拉取有效凭证，完全不用手动干预。

举个SSH连接的例子，你可以这么写：

from airflow.hooks.base import BaseHook
from airflow.providers.ssh.hooks.ssh import SSHHook
import subprocess
from airflow.exceptions import AirflowException

class DynamicSSHHook(BaseHook):
    def __init__(self, conn_id: str):
        super().__init__()
        self.conn_id = conn_id

    def get_conn(self):
        # 调用内部工具获取最新密码
        try:
            result = subprocess.run(
                ["/path/to/your-internal-tool", "get-password"],
                capture_output=True,
                text=True,
                check=True
            )
            fresh_password = result.stdout.strip()
        except subprocess.CalledProcessError as e:
            raise AirflowException(f"获取密码失败: {e.stderr}") from e

        # 从Airflow连接管理中读取固定配置（主机、用户名等）
        base_conn = self.get_connection(self.conn_id)
        
        # 返回带最新密码的SSH连接对象
        return SSHHook(
            host=base_conn.host,
            username=base_conn.username,
            password=fresh_password,
            port=base_conn.port or 22
        )

使用的时候，在你的Operator里直接调用这个Hook就行：

from airflow.operators.python import PythonOperator

def your_task_func():
    hook = DynamicSSHHook(conn_id="your_ssh_conn")
    ssh_client = hook.get_conn()
    # 这里执行你的SSH操作，比如远程命令、文件传输等
    ssh_client.exec_command("echo 'Hello from dynamic connection'")

task = PythonOperator(
    task_id="dynamic_ssh_task",
    python_callable=your_task_func
)

这种方式的好处是逻辑解耦，所有和凭证相关的逻辑都集中在Hook里，后续修改工具调用方式也只需要改Hook代码，不用动业务任务。

2. 定时DAG刷新Airflow变量 #

如果不想写自定义代码，也可以用Airflow变量+定时任务的组合：

• 先创建一个Airflow变量（比如ssh_fresh_password）用来存储最新密码
• 写一个每隔55分钟执行的DAG，调用内部工具获取密码并更新变量
• 在你的SSH连接配置中，密码字段直接引用这个变量：{{ var.value.ssh_fresh_password }}

定时刷新的DAG示例：

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.models import Variable
import subprocess
from airflow.exceptions import AirflowException
from datetime import datetime, timedelta

def refresh_ssh_password():
    try:
        result = subprocess.run(
            ["/path/to/your-internal-tool", "get-password"],
            capture_output=True,
            text=True,
            check=True
        )
        fresh_pwd = result.stdout.strip()
        Variable.set("ssh_fresh_password", fresh_pwd)
        return f"密码已更新: {fresh_pwd[:4]}***"
    except subprocess.CalledProcessError as e:
        raise AirflowException(f"刷新密码失败: {e.stderr}") from e

with DAG(
    dag_id="refresh_ssh_password",
    schedule_interval=timedelta(minutes=55),
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=["infra", "credentials"]
) as dag:
    refresh_task = PythonOperator(
        task_id="fetch_and_update_password",
        python_callable=refresh_ssh_password
    )

⚠️ 注意：这种方式有个小风险——如果定时DAG执行失败，密码过期后业务任务会连接失败，所以一定要给这个DAG配置告警（比如邮件、Slack通知），确保能及时发现问题。

3. 连接配置的Extra字段存命令，运行时动态执行 #

如果你的场景比较简单，也可以在连接的Extra字段里存储获取密码的命令，然后在任务里动态执行：

1. 先在Airflow连接管理中，给你的SSH连接添加Extra字段：

   {"password_cmd": "/path/to/your-internal-tool get-password"}
   

2. 在任务中读取这个命令并执行：

   from airflow.providers.ssh.hooks.ssh import SSHHook
   from airflow.models import Connection
   import subprocess
   
   

def task_with_dynamic_pwd():
conn = Connection.get_connection_from_secrets(conn_id="your_ssh_conn")
# 从Extra中取出命令
password_cmd = conn.extra_dejson.get("password_cmd")
if not password_cmd:
raise ValueError("连接配置中缺少password_cmd字段")

# 执行命令获取密码
result = subprocess.run(password_cmd.split(), capture_output=True, text=True, check=True)
fresh_pwd = result.stdout.strip()

# 建立连接
ssh_hook = SSHHook(
    host=conn.host,
    username=conn.username,
    password=fresh_pwd
)
# 执行操作...

### 通用注意事项
- 确保Airflow Worker节点有权限执行你的内部工具，工具的路径最好用绝对路径，避免PATH问题
- 如果是多Worker集群，每个Worker都要能访问到这个内部工具（比如通过共享目录、容器镜像打包等方式）
- 所有调用工具的逻辑都要加异常处理，避免工具执行失败导致任务静默失败

内容的提问来源于stack exchange，提问作者Breathe