如何用Airflow FTPS Hook配置证书/密钥连接外部FTPS服务器？

在使用Airflow的FTPSHook连接外部FTPS服务器时出现握手错误，原因是该服务器要求连接时提供证书文件（在WinSCP中不使用证书连接也会出现同类错误）。以下是Python Operator中运行的精简测试代码：

def ftps_connection_test(conn_id, source_path, 
                source_connection_type):

    if source_connection_type == 'FTPS':

        ftps_hook=FTPSHook(ftp_conn_id=conn_id)
        ftps_hook.get_conn().prot_p()

        print('Testing Connection:', ftps_hook.test_connection())

已尝试搜索同类错误、查阅Airflow官方文档，但仍未找到通过该Hook携带密钥/证书文件建立连接的方法，求解决方案。

解决方案 #

Airflow的FTPSHook底层基于Python标准库ftplib.FTP_TLS，默认未直接支持传入证书文件的参数，可通过以下两种方式解决：

方法1：自定义FTPSHook扩展证书支持 #

继承原FTPSHook，重写get_conn方法，在建立连接时传入证书相关参数：

from airflow.providers.ftp.hooks.ftp import FTPSHook
from ftplib import FTP_TLS

class CustomFTPSHook(FTPSHook):
    def __init__(self, ftp_conn_id: str, certfile: str = None, keyfile: str = None, *args, **kwargs):
        super().__init__(ftp_conn_id=ftp_conn_id, *args, **kwargs)
        self.certfile = certfile
        self.keyfile = keyfile

    def get_conn(self):
        conn = FTP_TLS()
        conn.connect(self.host, self.port, self.timeout)
        conn.login(self.login, self.password)
        # 传入证书文件完成握手
        if self.certfile and self.keyfile:
            conn.auth()
            conn.prot_p()
            conn.certfile = self.certfile
            conn.keyfile = self.keyfile
        return conn

使用时实例化自定义Hook即可：

def ftps_connection_test(conn_id, source_path, source_connection_type, cert_path, key_path):
    if source_connection_type == 'FTPS':
        ftps_hook = CustomFTPSHook(ftp_conn_id=conn_id, certfile=cert_path, keyfile=key_path)
        conn = ftps_hook.get_conn()
        print('连接测试结果:', ftps_hook.test_connection())

方法2：利用Airflow连接配置的额外参数传递证书路径 #

在Airflow的FTPS连接配置中，通过extra字段传入证书和密钥文件路径，再在代码中读取这些参数并动态设置到连接对象上：

from airflow.providers.ftp.hooks.ftp import FTPSHook

def ftps_connection_test(conn_id, source_path, source_connection_type):
    if source_connection_type == 'FTPS':
        ftps_hook = FTPSHook(ftp_conn_id=conn_id)
        conn = ftps_hook.get_conn()
        # 从连接配置的extra中获取证书路径
        connection = ftps_hook.get_connection(conn_id)
        certfile = connection.extra_dejson.get('certfile')
        keyfile = connection.extra_dejson.get('keyfile')
        if certfile and keyfile:
            conn.certfile = certfile
            conn.keyfile = keyfile
            # 重新执行认证握手
            conn.auth()
        print('连接测试结果:', ftps_hook.test_connection())

Airflow连接配置的extra示例：

{"certfile": "/path/to/certificate.pem", "keyfile": "/path/to/privatekey.pem"}

注意：需确保Airflow Worker进程对证书和密钥文件有读取权限，避免因权限问题导致加载失败。

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