Apache Airflow 接入 Akeyless 密钥后端Connections、Variables 与配置项的统一托管实践【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow本指南以 Apache Airflow 官方 Akeyless Providerapache-airflow-providers-akeyless提供的AkeylessBackend为核心讲解如何将 Airflow 的 Connections、Variables 与配置项Configuration直接托管到 Akeyless Vault Platform实现「代码零硬编码密钥」。读完本文你将掌握airflow.cfg与环境变量两种接入方式、Akeyless 侧的密钥命名约定、三种 Connection 存储格式、五种认证方式以及 Amazon MWAA、Google 托管 Airflow 等云环境下的免静态凭据接入方案并深入理解底层实现原理。一、AkeylessBackend 是什么Apache Airflow 提供了可插拔的 Secrets Backend 机制通过实现BaseSecretsBackend接口任何密钥管理系统都可以成为 Airflow 的 Connection / Variable / 配置项来源。Akeyless Provider 中的AkeylessBackend源码位于 secrets/akeyless.py正是这样一个实现它继承自BaseSecretsBackend与LoggingMixin可以直接从 Akeyless Vault Platform 拉取三类数据Connections例如postgres_default、smtp_default等连接对象VariablesDAG 中使用的 Airflow 变量ConfigurationAirflow 自身的配置选项如smtp_host、sql_alchemy_conn。从 provider.yaml 可以看到该后端在 Provider 元数据中被显式声明为secrets-backends安装 Provider 后即被 Airflow 识别。二、快速接入两种配置方式2.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 }2.2 通过环境变量配置Airflow 支持用环境变量覆盖所有配置项secrets.backend与secrets.backend_kwargs也不例外export AIRFLOW__SECRETS__BACKENDairflow.providers.akeyless.secrets.akeyless.AkeylessBackend export AIRFLOW__SECRETS__BACKEND_KWARGS{connections_path: /airflow/connections, ...}环境变量方式特别适合容器化部署与托管服务场景——无需修改配置文件即可完成注入。2.3 前置条件安装 Providerpip install apache-airflow-providers-akeyless要求apache-airflow2.11.0、akeyless5.0.0详见 README.rst若使用云厂商认证aws_iam/gcp/azure_ad还需安装可选依赖pip install apache-airflow-providers-akeyless[cloud_id]对应akeyless-cloud-id0.3.0。三、密钥命名约定base_path/key后端解析密钥的方式是拼接base_path sep key默认分隔符sep为/。也就是说Akeyless 中的文件夹结构天然对应 Airflow 的密钥命名空间类型查找路径示例Connectionpostgres_default/airflow/connections/postgres_defaultVariablemy_var/airflow/variables/my_varConfigsmtp_host/airflow/config/smtp_host对应源码中的_get_secret方法secrets/akeyless.pypath f{base_path}{self.sep}{key} token self._authenticate() res self._client.get_secret_value(akeyless.GetSecretValue(names[path], tokentoken)) return res.get(path)理解这一点后你只需在 Akeyless 控制台按上述路径创建静态密钥即可无需在 Airflow 侧维护任何映射关系。四、在 Akeyless 中存储 ConnectionsConnection 支持三种存储格式后端会智能解析对应 get_connection 的实现逻辑。4.1 格式一URI 字符串直接存储连接 URI例如postgresql://user:passwordhost:5432/dbname后端会将该原始字符串作为Connection的uri传入Connection(conn_id, uriraw)构造。4.2 格式二带conn_uri的 JSON 字典{conn_uri: postgresql://user:passwordhost:5432/dbname}后端从 JSON 中取出conn_uri键并构造 Connection。4.3 格式三字段展开的 JSON 字典{ conn_type: postgres, host: db.example.com, login: admin, password: secret, schema: mydb, port: 5432 }后端将剩余字段作为关键字参数传给Connection(conn_id, **data)与 Airflow 元数据库中的 Connection 字段一一对应。这套解析逻辑在单元测试 test_akeyless.py 中有完整覆盖URI 格式解析出host、loginJSONconn_uri格式与字段展开格式均能正确还原 Connection 对象。实用建议字段展开格式可读性最强且支持conn_type之外的任意 Connection 字段如extra、port推荐在团队中使用。五、认证方式详解后端支持五种access_type在 secrets/akeyless.py 中由_SUPPORTED_BACKEND_AUTH_TYPES常量约束传入不支持的取值会直接抛出ValueErroraccess_type说明api_key使用 Access ID Access Key 认证默认方法uid使用预先存在的 Universal Identity 令牌aws_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 身份认证适合 Azure 托管的负载其中云厂商认证aws_iam、gcp、azure_ad依赖akeyless_cloud_id包未安装时_get_cloud_id方法会抛出带安装提示的ImportErrorsecrets/akeyless.pyraise ImportError( fakeyless_cloud_id is required for {self._access_type} authentication. Install it with: pip install apache-airflow-providers-akeyless[cloud_id] )5.1 认证与令牌缓存原理_authenticate方法secrets/akeyless.py实现了令牌缓存首次认证后缓存 token 与过期时间token_ttl秒内默认 600 秒直接复用缓存避免每次密钥读取都重新认证。这一行为在测试test_token_cachingtest_akeyless.py中得到验证连续两次读取变量auth只被调用一次。5.2 云厂商认证细节aws_iam通过CloudId().generate()生成云身份gcp通过CloudId().generateGcp(gcp_audience)生成gcp_audience即 Akeyless 侧配置的 audienceazure_ad通过CloudId().generateAzure(azure_object_id)生成azure_object_id为 Azure AD 对象 ID。认证时构造akeyless.Auth(access_id..., access_type..., cloud_id...)请求体调用client.auth()换取 API token。六、云端托管平台接入实战6.1 使用 Amazon MWAA在 MWAA 上可以复用环境的 IAM 执行角色完成 Akeyless 认证全程无需静态 API Key添加依赖在上传到 S3 的requirements.txt中加入apache-airflow-providers-akeyless[cloud_id]配置 Airflow configuration options在 MWAA 控制台添加配置键值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即可让执行角色自动获得访问权限。6.2 使用 Google Managed Service for Apache Airflow通过 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 }其中gcp_audience需要与 Akeyless 侧 GCP Auth Method 配置的 audience 保持一致。七、参数参考backend_kwargs 全集以下参数在 AkeylessBackend.init中定义参数默认值说明connections_path/airflow/connectionsAkeyless 中存储 Connection 的文件夹路径设为None可禁用variables_path/airflow/variablesAkeyless 中存储 Variable 的文件夹路径设为None可禁用config_path/airflow/configAkeyless 中存储配置项的文件夹路径设为None可禁用sep/基础路径与密钥名之间的分隔符api_urlhttps://api.akeyless.ioAkeyless API 端点SaaS 或自建 Gatewayaccess_id无Akeyless Access IDaccess_key无Akeyless Access Keyapi_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 缓存秒数到期后重新认证此外源码还额外支持两个多团队相关参数详见下节参数默认值说明use_team_secrets_pathTrue多团队模式下是否先按{base}/{team}/{key}查找global_secrets_pathNone多团队模式下全局回退路径段如global注意三个*_path参数传入时会被自动去除末尾/rstrip(/)避免路径拼接时出现双斜杠。八、源码级深入多团队模式与值解析8.1 多团队multi-team部署的路径策略当 Airflow 以多团队模式运行core.multi_team True时AkeylessBackend会按以下顺序解析密钥见_get_team_or_global_secretsecrets/akeyless.py团队路径{base_path}/{team_name}/{key}命中即返回全局回退路径未命中且设置了global_secrets_path时尝试{base_path}/{global_secrets_path}/{key}默认回退否则回退到{base_path}/{key}。若设置use_team_secrets_path False则跳过团队前缀直接按全局路径查找。多团队模式下get_connection与get_variable的team_name参数由 Airflow 运行时注入测试覆盖了团队命中、全局回退、全局路径回退、禁用团队路径等全部场景test_akeyless.py。8.2 跨团队命名空间防护源码中有一处值得关注的安全设计在多团队模式且启用团队路径时若密钥名本身包含分隔符sep如beta/db_password后端会拒绝查询并返回None同时记录告警日志见_escapes_its_namespace与_log_refusalsecrets/akeyless.py。原因在于这类密钥在团队路径未命中后会通过全局回退解析到{base}/{key}而该前缀正是其他团队密钥所在命名空间存在越权读取风险。对应测试test_get_variable_cannot_reach_another_teams_namespacetest_akeyless.py验证了即便目标路径存在值该查询也不会发出。8.3 Variable 与 Config 的 JSON 值解析get_variable与get_config支持「纯文本」与「JSON 包裹」两种存储方式若存储内容是 JSON 且为字典会优先取其中的value键否则返回原始字符串secrets/akeyless.py。例如 Akeyless 中存储{value: my-json-wrapped-value}Airflow 侧读到的就是my-json-wrapped-value便于与 CLI 或 UI 写入的 JSON 结构兼容。8.4 未命中与禁用的行为密钥不存在akeyless.ApiException时返回None并记录 debug 日志不会抛错中断 DAG将对应*_path设为None即禁用该类密钥的解析get_*方法直接返回None。这些边界行为均有对应单元测试佐证test_akeyless.py。九、延伸AkeylessHook 与连接类型除 Secrets Backend 外该 Provider 还提供AkeylessHookhooks/akeyless.py用于在 DAG 中以编程方式读写密钥支持静态密钥的读取、创建、更新、删除以及动态密钥与轮转密钥的获取。它与 Secrets Backend 的差别在于Backend 面向「Airflow 系统级凭据解析」Hook 面向「DAG 内的任意密钥操作」。Hook 支持更丰富的认证类型api_key、aws_iam、gcp、azure_ad、uid、jwt、k8s、certificate通过akeyless类型连接配置字段映射为Host → API URL、Login → Access ID、Password → Access Key、Extra → 认证相关 JSON。系统示例 DAG 位于 example_dag_akeyless.py演示了get_secret_value、list_items、get_dynamic_secret_value的典型用法连接配置的完整字段说明见 connections.rst。十、接入流程小结安装 Provider云厂商认证追加[cloud_id]extra在 Akeyless 中按{base_path}/{key}约定创建静态密钥Connection 建议使用字段展开的 JSON 格式通过airflow.cfg或环境变量配置secrets.backend与secrets.backend_kwargs选择认证方式静态环境用api_key云托管环境优先aws_iam/gcp/azure_ad免静态凭据重启 Airflow 组件后DAG 中即可直接按conn_id/ 变量名 / 配置键引用密钥统一由 Akeyless 托管配合其权限策略与审计能力实现安全闭环。相关参考文档secrets-backend.rst本文原始出处、connections.rst、provider.yaml。【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考