Airflow SFTPHook私钥认证失败：UI测试正常DAG执行报错

问题核心 #

使用apache/airflow:2.3.2-python3.10镜像运行Airflow，已指定paramiko==2.7.2，SFTP连接在UI测试时成功，但DAG执行时触发Authentication (password) failed错误——日志显示系统尝试密码认证而非配置的私钥。

关键原因 #

UI测试连接与DAG执行的环境/配置解析逻辑存在差异：UI对私钥格式的容错性更高，但DAG执行时JSON配置中的换行符会导致私钥解析失败，Paramiko会自动 fallback 到密码认证流程。

解决方案 #

1. 修正Connections Extra中的私钥格式 #

JSON不允许直接包含换行符，需将私钥中的换行替换为\n转义字符，或对整个私钥进行Base64编码：

方式一：转义换行符

{
  "private_key": "-----BEGIN OPENSSH PRIVATE KEY-----\nabcdefghijklmnopqrstuvwxyz...\n-----END OPENSSH PRIVATE KEY-----",
  "no_host_key_check": true
}

方式二：Base64编码私钥

将本地私钥文件编码为Base64字符串（Linux/macOS可执行base64 -w0 your_private_key），然后配置：

{
  "private_key": "YOUR_BASE64_ENCODED_PRIVATE_KEY",
  "no_host_key_check": true,
  "private_key_passphrase": "YOUR_PASSPHRASE_IF_ANY"
}

2. 自定义SFTPHook强制密钥认证 #

若默认Hook仍自动切换认证方式，可重写Hook逻辑，强制仅使用私钥认证：

from airflow.providers.sftp.hooks.sftp import SFTPHook
import paramiko

class CustomSFTPHook(SFTPHook):
    def get_conn(self):
        conn = self.get_connection(self.conn_id)
        ssh_client = paramiko.SSHClient()
        
        # 关闭主机密钥检查
        if conn.extra_dejson.get("no_host_key_check"):
            ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        
        # 加载私钥
        private_key_str = conn.extra_dejson.get("private_key")
        try:
            # 处理带密码的私钥
            passphrase = conn.extra_dejson.get("private_key_passphrase")
            private_key = paramiko.RSAKey.from_private_key_str(private_key_str, password=passphrase)
        except paramiko.PasswordRequiredException:
            # 无密码的私钥
            private_key = paramiko.RSAKey.from_private_key_str(private_key_str)
        
        # 强制使用私钥认证，禁用本地密钥查找和代理
        ssh_client.connect(
            hostname=conn.host,
            username=conn.login,
            pkey=private_key,
            port=conn.port or 22,
            look_for_keys=False,
            allow_agent=False,
            password=None  # 显式禁用密码认证
        )
        self.ssh_client = ssh_client
        return ssh_client.open_sftp()

在DAG中使用自定义Hook替代默认SFTPHook：

hook = CustomSFTPHook("your_sftp_connection_id")

3. 确认Worker环境一致性 #

确保Airflow Worker容器与Webserver使用相同镜像，且paramiko==2.7.2已正确安装——版本不一致可能导致私钥解析逻辑差异。

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