Apache Airflow 集成 Akeyless Vault：Hook、连接类型与 Secrets Backend 实战指南

Apache Airflow 集成 Akeyless VaultHook、连接类型与 Secrets Backend 实战指南【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本指南以 Apache Airflow 官方apache-airflow-providers-akeyless0.3.1Provider 为核心系统讲解如何将 Akeyless Vault Platform 接入 Airflow既可以在 DAG 中通过AkeylessHook直接读写静态/动态/轮转密钥也可以将 Akeyless 配置为全局 Secrets Backend让 Connections、Variables 与配置项直接从密钥库解析。读完本文你将掌握从安装、连接配置、认证方式到云托管平台Amazon MWAA、Google Managed Service落地部署的完整方案。一、Provider 概览它解决了什么问题apache-airflow-providers-akeyless是 Apache Airflow 官方 Provider 体系中用于对接 Akeyless Vault Platform 的定义它提供三类核心能力Hook封装akeylessPython SDK用于管理静态密钥static secrets、动态密钥dynamic secrets、轮转密钥rotated secrets等Connection type在 Airflow UI / API 中注册akeyless连接类型供 Hook 及下游算子复用Secrets Backend将 Akeyless 挂载为 Airflow 的密钥后端自动为 Connections、Variables 和 Configuration 提供取值来源。全部类位于airflow.providers.akeylessPython 包内当前 Release 版本为0.3.1。Provider 的集成清单、Hook 模块、连接类型字段与 Secrets Backend 声明可参见 providers/akeyless/provider.yaml。二、安装与依赖2.1 基础安装在已安装的 Airflow 环境之上直接通过 PyPI 安装pip install apache-airflow-providers-akeyless2.2 版本要求该 Provider 发行版要求的最小 Apache Airflow 版本为2.11.0依赖关系如下来自 providers/akeyless/docs/index.rstPIP 包要求版本apache-airflow2.11.0apache-airflow-providers-common-compat1.12.0akeyless5.0.0同时支持 Python 3.10、3.11、3.12、3.13、3.14见 providers/akeyless/README.rst。2.3 可选依赖cloud_idProvider 提供一个名为cloud_id的 extra用于安装云身份认证所需的akeyless-cloud-id库要求akeyless-cloud-id0.3.0pip install apache-airflow-providers-akeyless[cloud_id]只有使用aws_iam、gcp、azure_ad三种云身份认证时才需要该 extra仅使用api_key或uid认证则无需安装。2.4 从源码安装与官方发行包从源码安装的步骤参见 providers/akeyless/docs/installing-providers-from-sources.rst。官方正式发布包sdist 与 wheel含.asc签名与.sha512校验文件可在 Apache 官方下载站点获取安装后建议校验包完整性。Provider 的变更历史可查阅 providers/akeyless/docs/changelog.rst。三、配置 Akeyless 连接Connection type3.1 连接字段映射akeyless连接类型将 Airflow 的标准连接字段映射为 Akeyless 的认证参数详见 providers/akeyless/docs/connections.rst 与 Hook 源码Airflow 字段含义HostAkeyless API URL例如https://api.akeyless.io或自建 Gateway URL端口8081留空时默认https://api.akeyless.ioLoginAkeylessAccess ID形如p-xxxxxxxxxPasswordAkeylessAccess Key用于api_key认证Extra (JSON)按认证方式补充的额外字段见下表3.2 Extra 字段说明字段说明access_type认证方式可选api_key默认、aws_iam、gcp、azure_ad、uid、jwt、k8s、certificateuid_tokenUniversal-Identity token用于uid认证gcp_audienceGCP audience 字符串用于gcp认证azure_object_idAzure AD Object ID用于azure_ad认证jwt原始 JWT token用于jwt认证k8s_auth_config_nameKubernetes Auth Config 名称用于k8s认证certificate_dataPEM 编码的客户端证书用于certificate认证private_key_dataPEM 编码的私钥用于certificate认证3.3 连接示例API Key 认证默认Login填 Access ID、Password填 Access Key{ access_type: api_key }AWS IAM 认证需安装cloud_idextra{ access_type: aws_iam }Kubernetes 认证{ access_type: k8s, k8s_auth_config_name: my-k8s-config }3.4 UI 表单行为在 Airflow 连接管理界面中该连接类型会隐藏extra、schema、port字段并将login、password、host重命名为 Access ID、Access Key、API URL同时为access_type、uid_token、gcp_audience、azure_object_id、jwt_token、k8s_auth_config_name、certificate_data、private_key_data提供独立输入框。这些 UI 行为同时声明在 provider.yaml 与 Hook 的get_connection_form_widgets/get_ui_field_behaviour中。四、AkeylessHook源码级剖析AkeylessHook位于 providers/akeyless/src/airflow/providers/akeyless/hooks/akeyless.py是对akeylessPython SDK 的薄封装默认连接 ID 为akeyless_default连接类型为akeyless。4.1 认证流程authenticate()返回一个 API token其分支逻辑如下uid直接返回 Extra 中的uid_token不发起网络调用api_key使用连接中的 Access IDLogin与 Access KeyPassword调用 SDK 的Authaws_iam/gcp/azure_ad通过akeyless_cloud_id的CloudId生成云身份 ID 后认证分别调用generate()、generateGcp(gcp_audience)、generateAzure(azure_object_id)jwt将 Extra 中的jwt放入认证体k8s使用 Extra 中的k8s_auth_config_namecertificate使用 Extra 中的certificate_data与private_key_dataSDK 侧映射为cert认证。若access_type不在合法集合(api_key, aws_iam, gcp, azure_ad, uid, jwt, k8s, certificate)内会抛出ValueError。client属性为缓存的akeyless.V2Api实例API URL 取自连接 Host未以http开头时自动补全https://前缀缺省为https://api.akeyless.io。4.2 密钥操作 APIHook 提供以下直接可用的方法均为对 SDK 的委托方法功能get_secret_value(name)按路径读取单个静态密钥get_secret_values(names)批量读取多个静态密钥create_secret(name, value, descriptionNone)创建静态密钥update_secret_value(name, value)更新静态密钥值delete_item(name)删除密钥/条目describe_item(name)获取密钥/条目元数据list_items(path/)列出路径下的条目get_dynamic_secret_value(name)生成动态密钥值如数据库临时凭据get_rotated_secret_value(name)获取轮转密钥值test_connection()供 Airflow UI 校验连接内部执行一次认证单元测试 tests/unit/akeyless/hooks/test_akeyless.py 覆盖了test_connection成功/失败、静态密钥读写、动态/轮转密钥获取、uid认证不走网络、非法access_type抛错、jwt认证透传等场景可作为调用方式的参考。五、将 Akeyless 配置为 Secrets BackendSecrets Backend 类为airflow.providers.akeyless.secrets.akeyless.AkeylessBackend见 providers/akeyless/docs/secrets-backend.rst 与 源码用于让 Airflow 直接从 Akeyless 解析 Connections、Variables 与配置项。5.1 airflow.cfg 配置在airflow.cfg的[secrets]段添加[secrets] backend airflow.providers.akeyless.secrets.akeyless.AkeylessBackend backend_kwargs { connections_path: /airflow/connections, variables_path: /airflow/variables, config_path: /airflow/config, api_url: https://api.akeyless.io, access_id: p-xxxxxxxxx, access_key: your-access-key, access_type: api_key }5.2 环境变量配置也可以不修改配置文件改用 Airflow 的环境变量覆盖机制export AIRFLOW__SECRETS__BACKENDairflow.providers.akeyless.secrets.akeyless.AkeylessBackend export AIRFLOW__SECRETS__BACKEND_KWARGS{connections_path: /airflow/connections, variables_path: /airflow/variables, config_path: /airflow/config, api_url: https://api.akeyless.io, access_id: p-xxxxxxxxx, access_key: your-access-key, access_type: api_key}5.3 密钥命名约定密钥按base_path/key拼接解析类型示例查找路径Connectionpostgres_default/airflow/connections/postgres_defaultVariablemy_var/airflow/variables/my_varConfigsmtp_host/airflow/config/smtp_host5.4 存储 Connections 的三种格式Connections 在 Akeyless 中支持三种存储格式解析逻辑见 secrets/akeyless.py 的get_connection对应测试见 tests/unit/akeyless/secrets/test_akeyless.pyURI 字符串postgresql://user:passwordhost:5432/dbnameJSON 字典含conn_uri{conn_uri: postgresql://user:passwordhost:5432/dbname}JSON 字典逐字段{ conn_type: postgres, host: db.example.com, login: admin, password: secret, schema: mydb, port: 5432 }对于 Variables 与 Config若存储的 JSON 字典含value键则取value字段作为值否则返回原始字符串见get_variable/get_config实现。六、Secrets Backend 认证方法Backend 支持的认证类型与 Hook 略有差异仅限五种api_key、uid、aws_iam、gcp、azure_ad构造时传入其他类型会直接抛出ValueError源码注释提示其他认证方式请直接使用AkeylessHookaccess_type说明api_key使用 Access ID Access Key 认证默认方式uid使用已有的 Universal Identity tokenaws_iam使用宿主机 AWS IAM 角色认证适合Amazon MWAA以及 EC2/ECS/EKS 工作负载无需静态凭据gcp使用 GCP workload identity 认证适合Google Managed Service for Apache Airflow原 Cloud Composer以及 GCE/GKE 工作负载azure_ad使用 Azure AD identity 认证适合 Azure 托管的 Airflowaws_iam、gcp、azure_ad三种云认证需要akeyless_cloud_id包pip install apache-airflow-providers-akeyless[cloud_id]认证成功后Backend 会按token_ttl默认600秒缓存 API token在有效期内复用避免每个密钥查询都重复认证——这一行为在 test_token_caching 中有明确验证。七、云托管平台实践7.1 Amazon MWAA在 Amazon MWAA 上可利用环境自身的 IAM 执行角色认证 Akeyless无需静态 API Key在上传到 S3 的requirements.txt中加入apache-airflow-providers-akeyless[cloud_id]在 MWAA 控制台添加以下Airflow configuration options配置项值secrets.backendairflow.providers.akeyless.secrets.akeyless.AkeylessBackendsecrets.backend_kwargs{api_url: https://api.akeyless.io, access_id: p-xxxxxxxxx, access_type: aws_iam}确保 MWAA VPC 对 Akeyless API 端点api.akeyless.io或自建 Akeyless Gateway有出站 HTTPS 访问权限在 Akeyless 中创建与 MWAA 执行角色 ARN 关联的aws_iamAuth Method。7.2 Google Managed Service for Apache Airflow使用 GCP workload identity 认证配置示例[secrets] backend airflow.providers.akeyless.secrets.akeyless.AkeylessBackend backend_kwargs { api_url: https://api.akeyless.io, access_id: p-xxxxxxxxx, access_type: gcp, gcp_audience: akeyless.io }八、Backend 参数详解AkeylessBackend的全部构造参数默认值与说明见 secrets/akeyless.py参数默认值说明connections_path/airflow/connectionsConnections 的 Akeyless 文件夹路径设为None禁用variables_path/airflow/variablesVariables 的 Akeyless 文件夹路径设为None禁用config_path/airflow/config配置项的 Akeyless 文件夹路径设为None禁用sep/基础路径与密钥名之间的分隔符use_team_secrets_pathTrue多团队模式下是否先按{base}/{team}/{key}查找global_secrets_pathNone多团队模式下全局回退路径段如globalapi_urlhttps://api.akeyless.ioAkeyless API 端点access_id无Akeyless Access IDaccess_key无Akeyless Access Key用于api_key认证access_typeapi_key认证方式api_key、uid、aws_iam、gcp、azure_adgcp_audience无GCP audience仅gcp认证azure_object_id无Azure AD Object ID仅azure_ad认证token_ttl600API token 缓存秒数过期后重新认证注意路径参数在构造时会被rstrip(/)规范化将某一*_path设为None即关闭对应类型的密钥解析对应测试test_get_connection_disabled、test_get_variable_disabled、test_get_config_disabled。九、多团队模式与命名空间隔离当 Airflow 以多团队模式部署core.multi_team True时Backend 的查找顺序为先尝试团队作用域路径{base_path}/{team_name}/{key}未命中则回退到全局路径若设置了global_secrets_path则查{base_path}/{global_secrets_path}/{key}否则查{base_path}/{key}。use_team_secrets_path False可关闭团队前缀查找。值得注意的是Backend 实现了一套跨团队命名空间逃逸防护见 secrets/akeyless.py 的_escapes_its_namespace当多团队开启、use_team_secrets_path为真、且查询带team_name的密钥名本身包含分隔符如beta/db_password时团队回退可能解析到其他团队的密钥因此这类查询会被拒绝并返回None。该行为在 tests/unit/akeyless/secrets/test_akeyless.py 的 Cross-team namespace escape 用例组中做了完整验证团队作用域查询、全局回退、global_secrets_path回退、关闭团队路径、无 team 调用方、非多团队模式下的嵌套密钥等场景均有对应断言。十、在 DAG 中使用 AkeylessHook仓库提供了可直接运行的系统测试 DAG providers/akeyless/tests/system/akeyless/example_dag_akeyless.py。运行前需创建连接连接 ID 为akeyless_default、类型为akeyless、Host 为https://api.akeyless.io、Login/Password 为 Access ID/Key、Extra 为{access_type: api_key}。该 DAG 的流水线为get_static_secret list_secrets get_dynamic_secret展示了三类典型用法from airflow.providers.akeyless.hooks.akeyless import AkeylessHook # 读取静态密钥 hook AkeylessHook(akeyless_conn_idakeyless_default) value hook.get_secret_value(/example/my-secret) # 列出路径下的条目 items hook.list_items(/example) # 生成动态密钥如数据库临时凭据 creds hook.get_dynamic_secret_value(/example/dynamic-db-producer)十一、常见问题与排查要点云认证报ImportError: akeyless_cloud_id确认已安装apache-airflow-providers-akeyless[cloud_id]Hook 与 Backend 的_get_cloud_id在缺包时都会给出明确的安装提示。access_type不受支持Hook 侧支持 8 种类型Backend 侧仅支持 5 种若 Backend 需要jwt、k8s、certificate等认证应改用AkeylessHook在 DAG 内自行认证。连接测试失败在 Airflow UI 连接页点击 Test 会触发一次真实认证失败信息即为底层异常请核对 API URL 可达性、Access ID/Key 有效性以及认证方式对应的 Extra 字段。多团队模式下密钥查不到检查core.multi_team是否开启、团队路径是否正确以及密钥名是否包含分隔符触发了命名空间逃逸防护。十二、参考资料Provider 文档主页providers/akeyless/docs/index.rst连接配置指南providers/akeyless/docs/connections.rstSecrets Backend 指南providers/akeyless/docs/secrets-backend.rstHook 实现providers/akeyless/src/airflow/providers/akeyless/hooks/akeyless.pySecrets Backend 实现providers/akeyless/src/airflow/providers/akeyless/secrets/akeyless.pyProvider 元数据providers/akeyless/provider.yaml单元测试providers/akeyless/tests/unit/akeyless/hooks/test_akeyless.py、providers/akeyless/tests/unit/akeyless/secrets/test_akeyless.py系统测试 DAGproviders/akeyless/tests/system/akeyless/example_dag_akeyless.py【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考