企业官网建设流程全解析

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 Provider（apache-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等连接对象；
• Variables：DAG 中使用的 Airflow 变量；
• Configuration：Airflow 自身的配置选项（如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__BACKEND="airflow.providers.akeyless.secrets.akeyless.AkeylessBackend" export AIRFLOW__SECRETS__BACKEND_KWARGS='{"connections_path": "/airflow/connections", ...}'

环境变量方式特别适合容器化部署与托管服务场景——无需修改配置文件即可完成注入。

2.3 前置条件

• 安装 Provider：pip install apache-airflow-providers-akeyless（要求apache-airflow>=2.11.0、akeyless>=5.0.0，详见 README.rst）；
• 若使用云厂商认证（aws_iam/gcp/azure_ad），还需安装可选依赖：pip install apache-airflow-providers-akeyless[cloud_id]（对应akeyless-cloud-id>=0.3.0）。

三、密钥命名约定：<base_path>/<key>

后端解析密钥的方式是拼接base_path + sep + key（默认分隔符sep为/）。也就是说，Akeyless 中的文件夹结构天然对应 Airflow 的密钥命名空间：

对应源码中的_get_secret方法（secrets/akeyless.py）：

path = f"{base_path}{self.sep}{key}" token = self._authenticate() res = self._client.get_secret_value(akeyless.GetSecretValue(names=[path], token=token)) return res.get(path)

理解这一点后，你只需在 Akeyless 控制台按上述路径创建静态密钥即可，无需在 Airflow 侧维护任何映射关系。

四、在 Akeyless 中存储 Connections

Connection 支持三种存储格式，后端会智能解析（对应 get_connection 的实现逻辑）。

4.1 格式一：URI 字符串

直接存储连接 URI，例如：

postgresql://user:password@host:5432/dbname

后端会将该原始字符串作为Connection的uri传入Connection(conn_id, uri=raw)构造。

4.2 格式二：带conn_uri的 JSON 字典

{"conn_uri": "postgresql://user:password@host: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、login，JSONconn_uri格式与字段展开格式均能正确还原 Connection 对象。

实用建议：字段展开格式可读性最强，且支持conn_type之外的任意 Connection 字段（如extra、port），推荐在团队中使用。

五、认证方式详解

后端支持五种access_type，在 secrets/akeyless.py 中由_SUPPORTED_BACKEND_AUTH_TYPES常量约束，传入不支持的取值会直接抛出ValueError：

其中云厂商认证（aws_iam、gcp、azure_ad）依赖akeyless_cloud_id包，未安装时_get_cloud_id方法会抛出带安装提示的ImportError（secrets/akeyless.py）：

raise ImportError( f"`akeyless_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_caching（test_akeyless.py）中得到验证：连续两次读取变量，auth只被调用一次。

5.2 云厂商认证细节

• aws_iam：通过CloudId().generate()生成云身份；
• gcp：通过CloudId().generateGcp(gcp_audience)生成，gcp_audience即 Akeyless 侧配置的 audience；
• azure_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：

1. 添加依赖：在上传到 S3 的requirements.txt中加入：

   apache-airflow-providers-akeyless[cloud_id]

2. 配置 Airflow configuration options：在 MWAA 控制台添加：

   配置键	值
   secrets.backend	airflow.providers.akeyless.secrets.akeyless.AkeylessBackend
   secrets.backend_kwargs	{"api_url": "https://api.akeyless.io", "access_id": "p-xxxxxxxxx", "access_type": "aws_iam"}

3. 网络打通：确保 MWAA 所在 VPC 具备到 Akeyless API 端点（api.akeyless.io或你的 Akeyless Gateway）的出站 HTTPS 访问能力。

4. 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中定义：

此外，源码还额外支持两个多团队相关参数（详见下节）：

注意：三个*_path参数传入时会被自动去除末尾/（rstrip("/")），避免路径拼接时出现双斜杠。

八、源码级深入：多团队模式与值解析

8.1 多团队（multi-team）部署的路径策略

当 Airflow 以多团队模式运行（core.multi_team = True）时，AkeylessBackend会按以下顺序解析密钥（见_get_team_or_global_secret，secrets/akeyless.py）：

1. 团队路径：{base_path}/{team_name}/{key}，命中即返回；
2. 全局回退路径：未命中且设置了global_secrets_path时，尝试{base_path}/{global_secrets_path}/{key}；
3. 默认回退：否则回退到{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_refusal，secrets/akeyless.py）。原因在于：这类密钥在团队路径未命中后会通过全局回退解析到{base}/{key}，而该前缀正是其他团队密钥所在命名空间，存在越权读取风险。对应测试test_get_variable_cannot_reach_another_teams_namespace（test_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 还提供AkeylessHook（hooks/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。

十、接入流程小结

1. 安装 Provider（云厂商认证追加[cloud_id]extra）；
2. 在 Akeyless 中按{base_path}/{key}约定创建静态密钥（Connection 建议使用字段展开的 JSON 格式）；
3. 通过airflow.cfg或环境变量配置secrets.backend与secrets.backend_kwargs；
4. 选择认证方式：静态环境用api_key，云托管环境优先aws_iam/gcp/azure_ad免静态凭据；
5. 重启 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