Apache Airflow 接入 Akeyless：akeyless 连接类型的配置指南与底层实现解析

Apache Airflow 接入 Akeyless：akeyless 连接类型的配置指南与底层实现解析

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow

本篇技术指南围绕 Apache Airflow 中akeyless连接类型（Connection Type）展开，说明如何通过 Airflow 的 Connection 机制连接 Akeyless Vault Platform（Akeyless 官方云服务或自建 Gateway）。文章以providers/akeyless/docs/connections.rst为骨架，结合apache-airflow-providers-akeyless提供方包内的 Hook 源码、provider 元数据与单元测试，覆盖 Host / Login / Password / Extra 四个连接字段的含义、八种access_type认证方式的完整 Extra 参数表、JSON 配置示例，以及认证与令牌获取在底层AkeylessHook.authenticate()中的真实实现。读完本文，你将能够在 Airflow UI 或代码中完整配置akeyless连接，并理解每种认证方式需要哪些字段、为什么需要这些字段。

一、akeyless连接类型是什么

akeyless是 Apache Airflow 为 Akeyless Vault Platform 提供的官方连接类型，由apache-airflow-providers-akeyless提供方包注册。该提供方包当前版本为 0.3.1，其核心交付物包括（见 provider.yaml）：

• 一个Hook：airflow.providers.akeyless.hooks.akeyless.AkeylessHook，是对官方akeylessPython SDK 的轻量封装，提供认证、静态/动态/轮转密钥的读写操作；
• 一个Connection Type：akeyless，注册了自定义的连接表单字段与 UI 行为；
• 一个Secrets Backend：AkeylessBackend，可将 Connections、Variables 与配置项直接托管在 Akeyless 中（详见 secrets-backend.rst）。

安装提供方包即可同时获得连接类型能力：

pip install apache-airflow-providers-akeyless

包依赖（见 pyproject.toml）：

若使用 AWS IAM / GCP / Azure AD 云身份认证，还需安装可选依赖cloud_id（akeyless-cloud-id>=0.3.0）：

pip install apache-airflow-providers-akeyless[cloud_id]

二、连接字段总览

在 Airflow 的Admin → Connections页面新建连接时，选择类型Akeyless（Hook 内部注册的hook_name = "Akeyless"、conn_type = "akeyless"）。该连接类型使用四个标准字段，其含义与通用连接字段不同，见下表：

Host 字段的底层处理

在 AkeylessHook.client 的实现中，API 地址的解析逻辑为：

1. 优先取self._conn.host，若为空则回退到 Extra 中的api_url，再回退到默认值https://api.akeyless.io；
2. 若地址不以http开头，会自动补全为https://前缀；
3. 最终以该地址构造akeyless.Configuration(host=api_url)与akeyless.ApiClient，返回akeyless.V2Api客户端（该客户端以cached_property缓存，重复调用不会重复构造）。

因此填写 Host 时可以直接写域名（如api.akeyless.io），Hook 会自动补全协议；也可以显式写完整 URL，包括自建 Gateway 的非 443 端口。

三、Extra 字段：认证方式与专属参数

akeyless连接的核心是 Extra 中的access_type字段，它决定了认证方式，并进一步决定其他 Extra 字段的取值。源码 akeyless.py 中定义了合法的认证类型集合：

VALID_AUTH_TYPES = ("api_key", "aws_iam", "gcp", "azure_ad", "uid", "jwt", "k8s", "certificate")

全部字段说明如下：

命名提示：文档中以jwt描述该字段，但 Hook 源码（akeyless.py）读取的是 Extra 中的jwt_token键，连接表单中该控件的 key 同样是jwt_token（见 provider.yaml 与get_connection_form_widgets()）。实际配置时应使用"jwt_token": "..."。

若 Extra 中未指定access_type，authenticate()会默认按api_key处理；若指定了不在VALID_AUTH_TYPES中的值，会抛出ValueError并列出全部合法取值——单元测试test_invalid_access_type_raises对此有覆盖（见 test_akeyless.py）。

UI 表单行为

该连接类型在 Airflow UI 上做了一系列定制（见get_ui_field_behaviour()与 provider.yaml）：

• 隐藏字段：extra、schema、port三个通用字段被隐藏，避免与 Akeyless 语义冲突；
• 重命名标签：login显示为 "Access ID"，password显示为 "Access Key"，host显示为 "API URL"；
• 自定义控件：通过get_connection_form_widgets()为access_type、uid_token、gcp_audience、azure_object_id、jwt_token、k8s_auth_config_name、certificate_data、private_key_data各生成一个输入框，方便在表单中直接填写而无需手写 JSON。

四、认证方式示例

1. API Key 认证（默认）

Extra 为：

{ "access_type": "api_key" }

同时将Login填为 Access ID，Password填为 Access Key。这是最简单的认证方式，也是默认行为。源码对应逻辑（akeyless.py）：

if access_type == "api_key": body.access_key = self._conn.password

即构造akeyless.Auth(access_id=login)后，把连接密码写入body.access_key，再调用self.client.auth(body).token换取 API 令牌。

2. AWS IAM 认证

{ "access_type": "aws_iam" }

该方式利用运行主机（如 EC2、ECS、EKS、Amazon MWAA）的 AWS IAM 角色身份认证，无需在连接中保存任何静态密钥。前提是安装可选依赖akeyless_cloud_id包：

pip install apache-airflow-providers-akeyless[cloud_id]

源码通过akeyless_cloud_id.CloudId().generate()在运行时生成 cloud ID 并放入认证请求（见_get_cloud_id()，akeyless.py）。若未安装该包，authenticate()会抛出ImportError并提示安装命令。

3. Kubernetes 认证

{ "access_type": "k8s", "k8s_auth_config_name": "my-k8s-config" }

k8s_auth_config_name对应 Akeyless 中预先配置的 Kubernetes Auth Config 名称。源码将其透传给 SDK 的认证请求体（body.k8s_auth_config_name = self._extra.get("k8s_auth_config_name")），由 Akeyless 服务端与集群侧配置共同完成校验。

4. 其他认证方式（源码行为速览）

其中uid认证的特殊之处值得留意：Universal Identity 令牌本身就是可用的 API 令牌，因此 Hook 直接返回self._extra["uid_token"]，不再向 Akeyless 发起Auth请求，省去一次往返（见 akeyless.py）。

五、在代码中使用连接

连接配置完成后，即可在 DAG 中通过AkeylessHook使用。Hook 构造时接收连接 ID（默认akeyless_default）：

from airflow.providers.akeyless.hooks.akeyless import AkeylessHook hook = AkeylessHook(akeyless_conn_id="akeyless_default")

authenticate()会依据连接中的access_type完成认证并返回 API 令牌，后续所有密钥操作（get_secret_value、create_secret、list_items、get_dynamic_secret_value、get_rotated_secret_value等）都会先调用它换取令牌，再透传给akeyless.V2Api。系统测试 DAG example_dag_akeyless.py 给出了完整的落地样例，其 DAG 头部的注释明确列出了连接创建要求：

• Connection ID：akeyless_default
• Connection Type：akeyless
• Host：https://api.akeyless.io
• Login：你的 Akeyless Access ID
• Password：你的 Akeyless Access Key
• Extra：{"access_type": "api_key"}

示例 DAG 中三个任务的执行路径正好覆盖了三种典型用法：

def _get_static_secret(): hook = AkeylessHook(akeyless_conn_id=AKEYLESS_CONN_ID) value = hook.get_secret_value("/example/my-secret") return value is not None def _list_secrets(): hook = AkeylessHook(akeyless_conn_id=AKEYLESS_CONN_ID) items = hook.list_items("/example") return len(items) def _get_dynamic_secret(): hook = AkeylessHook(akeyless_conn_id=AKEYLESS_CONN_ID) creds = hook.get_dynamic_secret_value("/example/dynamic-db-producer") return creds is not None

任务依次串联：get_secret >> list_items >> get_dynamic。从这里可以看到，同一份连接配置同时服务静态密钥读取、路径列举与动态密钥生成三类场景。

六、连接验证与测试

在 Airflow UI 保存连接时，"Test" 按钮会调用 Hook 的test_connection()方法（akeyless.py）：

def test_connection(self) -> tuple[bool, str]: try: self.authenticate() return True, "Connection successfully tested" except Exception as e: return False, str(e)

即通过实际执行一次认证来验证连接是否可用：认证成功返回(True, "Connection successfully tested")，失败则返回(False, 异常信息)。单元测试test_test_connection_success与test_test_connection_failure分别模拟了成功与Auth failed两种路径（见 test_akeyless.py）。由于认证失败信息会原样返回，测试连接时出现的 Access ID / Access Key 错误、网络不可达、cloud_id包缺失等问题都能直接在 UI 中看到具体原因。

七、进阶阅读

• 若希望把 Airflow 的 Connections、Variables 与配置项整体托管到 Akeyless，由平台自动按/<base_path>/<key>规则解析，可参考 Akeyless Secrets Backend 指南，其中包含airflow.cfg、环境变量、Amazon MWAA 与 Google Managed Service 两种托管环境的完整配置示例；
• 连接类型与 Hook 的完整元数据声明见 provider.yaml，其中的connection-types与ui-field-behaviour段落与本文所述的表单行为一一对应；
• 提供方包的安装要求与变更历史见 README.rst 与 changelog.rst；
• 提供方包内全部源码入口位于 hooks/akeyless.py，单元测试覆盖了全部八种认证方式中的主要分支，可在 test_akeyless.py 中逐一核对各access_type的实际行为。

【免费下载链接】airflowApache Airflow - A platform to programmatically author, schedule, and monitor workflows项目地址: https://gitcode.com/GitHub_Trending/ai/airflow