Apache Airflow AWS Auth Manager 实战指南：用 IAM Identity Center 管用户、用 Amazon Verified Permissions 管权限-育师

Apache Airflow AWS Auth Manager 实战指南：用 IAM Identity Center 管用户、用 Amazon Verified Permissions 管权限

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

Apache Airflow 的 AWS auth manager 将默认的 Flask auth manager（Flask AppBuilder）替换为基于 AWS 的授权集成，用户与权限分别交由两个服务管理：AWS IAM Identity Center（负责用户身份）与Amazon Verified Permissions（负责细粒度权限）。本文以 manage/index.rst 为骨架，完整讲解用户、用户组的定义与分配流程，以及基于 Cedar 语言的策略编写与权限管理实战，并结合仓库源码说明策略模型与鉴权调用链的底层实现。

注意：AWS auth manager 目前处于 alpha/experimental 阶段，接口与行为可能在没有预告的情况下发生变化（参见 auth-manager 总览文档）。

一、架构总览：身份与授权分离

当启用 AWS auth manager 后，Airflow 环境中所有用户及其权限不再由 Airflow 默认的 Flask auth manager 管理，而是拆分为两个 AWS 服务：

AWS IAM Identity Center：负责认证（authentication），即"用户是谁"；
Amazon Verified Permissions（AVP）：负责授权（authorization），即"用户能做什么"。

AWS auth manager 的全部鉴权决策都依赖 Amazon Verified Permissions 完成，所有权限策略都需要由 Airflow 环境管理员定义在 AVP 中。整体架构可参考 diagram_auth_manager_architecture.png，它展示了身份源、IAM Identity Center、AVP 与 Airflow 环境之间的协作关系。

从源码看，鉴权请求由 avp/facade.py 中的AwsAuthManagerAmazonVerifiedPermissionsFacade封装：它读取 Airflow 配置中的aws_auth_manager段（连接 ID、区域、policy store ID），构建 AVP 客户端，并在每次is_authorized调用时把「主体（principal）+ 动作（action）+ 资源（resource）+ 上下文（context）」组装为 AVP API 请求。配置键定义见 constants.py：

CONF_SECTION_NAME = "aws_auth_manager" CONF_CONN_ID_KEY = "conn_id" CONF_REGION_NAME_KEY = "region_name" CONF_SAML_METADATA_URL_KEY = "saml_metadata_url" CONF_AVP_POLICY_STORE_ID_KEY = "avp_policy_store_id"

二、通过 AWS IAM Identity Center 管理用户

所有能访问 Airflow 环境的用户都必须定义在 AWS IAM Identity Center 中。Identity Center 既可以作为身份源，也可以作为身份源（例如 Active Directory）与 Airflow 环境之间的代理。

2.1 查看用户列表

打开 IAM Identity Center 控制台；
选择Users（用户），即可看到当前身份源中定义的全部用户。

2.2 使用用户组

可以使用 IAM Identity Center 中的用户组（Groups）把用户组织为逻辑实体（例如按团队、按部门划分）。这些用户组随后可以在 Amazon Verified Permissions 中被引用，从而把权限一次性授予一组用户，而无需逐个用户重复配置策略。

查看用户组列表：

打开 IAM Identity Center 控制台；
选择Groups（用户组）。

2.3 将用户和用户组分配到 Airflow 环境

需要特别注意的是：定义在 IAM Identity Center 中的用户和用户组并不会自动获得 Airflow 环境的访问权限，管理员必须手动分配哪些用户能够访问 Airflow。

分配步骤如下：

打开 IAM Identity Center 控制台；
如果你使用 AWS Managed Microsoft AD 管理用户，请确保 IAM Identity Center 控制台使用的是 AWS Managed Microsoft AD 目录所在的区域，再执行下一步。
选择Applications（应用程序）；
选择Customer managed（客户托管）标签页；
在应用程序列表中，选择名为Airflow的应用程序；
在应用程序详情页的Assigned users and groups（已分配的用户和用户组）区域，选择Assign users and groups（分配用户和用户组）；
在Assign users or groups页面中，勾选要分配给 Airflow 的用户和用户组。可以搜索用户和用户组，也可以通过搜索结果的勾选同时指定多个用户或用户组；
选择Assign users（分配用户）。

完成分配后，这些用户才能通过 Identity Center 完成认证并进入 Airflow 环境；而他们能执行什么操作，则由下一节的 Amazon Verified Permissions 策略决定。

三、通过 Amazon Verified Permissions 管理用户权限

AWS auth manager 使用 Amazon Verified Permissions 定义并校验用户权限，权限模型基于Cedar 语言实现细粒度授权。整个 Airflow 环境对应一个policy store（策略库），所有相关策略都集中存放在其中。

3.1 策略管理的操作路径

打开 Amazon Verified Permissions 控制台；
选择 Airflow 使用的 policy store（默认其描述为Airflow）；
在左侧导航窗格中选择Policies（策略）；
在Policies页面可以看到策略列表。创建新策略时： a. 选择Create policy（创建策略）； b. 在Create policy下选择Create static policy（创建静态策略）。

3.2 Cedar 策略格式与三类核心要素

策略使用 Cedar 语言定义。Cedar 中一条策略由三个要素组成：

要素	含义	在 Airflow 场景中的取值
Principal	谁在发起请求？	`Airflow::User::"<user_id>"`或`Airflow::Group::"<group_id>"`
Action	主体想执行什么操作？	`Airflow::Action::"<Resource>.<METHOD>"`，例如`Dag.GET`、`Connection.POST`
Resource	主体想对什么资源执行操作？	`Airflow::Dag::"<dag_id>"`等资源实体，或用通配符匹配全部

在 Airflow 环境的语境下，这三个要素各自只允许特定的一组取值。查看 policy store schema 中支持的 principal、action 和 resource 列表：

打开 Amazon Verified Permissions 控制台；
选择 Airflow 使用的 policy store（默认描述为Airflow）；
在左侧导航窗格中选择Schema（模式）。

从源码可以印证上述命名约定：avp/entities.py 定义了实体枚举与 ID 生成规则：

所有实体统一加Airflow::前缀（AVP_PREFIX_ENTITIES = "Airflow::"），因此策略中写的是Airflow::User、Airflow::Group、Airflow::Dag等；
Action ID 的约定为<资源类型>.<方法>，例如Variable.GET；
一个细节：当请求方法是GET且未指定具体实体 ID 时，get_action_id会自动将其转换为LIST——这解释了为什么策略清单里同时存在Dag.GET与Dag.LIST、Configuration.GET与Configuration.LIST等成对动作。

schema 中完整的实体类型包括Asset、AssetAlias、Backfill、Configuration、Connection、Custom、Dag、Menu、Pool、Team、Variable、View，以及作为主体出现的User（声明了memberOfTypes: ["Group"]，即用户隶属于用户组）和Group。完整定义见 avp/schema.json。

3.3 策略示例：从全量授权到细粒度控制

以下示例均可在 Amazon Verified Permissions 中直接定义，可自由修改或组合，构造出贴合自身需求的定制策略。

给指定用户全部权限

permit( principal == Airflow::User::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action, resource );

注意：必须用用户 ID而非用户名或其他属性来引用用户。用户 ID 可以在 AWS IAM Identity Center 中查到。

给一组用户全部权限

等价于 Flask AppBuilder 中的Admin 角色（参见 apache-airflow-providers-fab 的访问控制文档）：

permit( principal in Airflow::Group::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action, resource );

注意：必须用组 ID而非组名称引用用户组。组 ID 可以在 AWS IAM Identity Center 中查到。

给一组用户只读权限

等价于 Flask AppBuilder 中的Viewer 角色：

permit( principal in Airflow::Group::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action in [ Airflow::Action::"Configuration.GET", Airflow::Action::"Configuration.LIST", Airflow::Action::"Connection.GET", Airflow::Action::"Connection.LIST", Airflow::Action::"Custom.GET", Airflow::Action::"Custom.LIST", Airflow::Action::"Dag.GET", Airflow::Action::"Dag.LIST", Airflow::Action::"Menu.MENU", Airflow::Action::"Pool.GET", Airflow::Action::"Pool.LIST", Airflow::Action::"Variable.GET", Airflow::Action::"Variable.LIST", Airflow::Action::"Asset.GET", Airflow::Action::"Asset.LIST", Airflow::Action::"AssetAlias.GET", Airflow::Action::"AssetAlias.LIST", Airflow::Action::"Backfill.GET", Airflow::Action::"Backfill.LIST", Airflow::Action::"View.GET" ], resource );

给一组用户标准 Airflow 用户权限

等价于 Flask AppBuilder 中的User 角色：

permit( principal in Airflow::Group::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action in [ Airflow::Action::"Configuration.GET", Airflow::Action::"Configuration.LIST", Airflow::Action::"Connection.GET", Airflow::Action::"Connection.LIST", Airflow::Action::"Custom.GET", Airflow::Action::"Custom.LIST", Airflow::Action::"Dag.GET", Airflow::Action::"Dag.LIST", Airflow::Action::"Menu.MENU", Airflow::Action::"Pool.GET", Airflow::Action::"Pool.LIST", Airflow::Action::"Variable.GET", Airflow::Action::"Variable.LIST", Airflow::Action::"Asset.GET", Airflow::Action::"Asset.LIST", Airflow::Action::"View.GET", Airflow::Action::"Dag.POST", Airflow::Action::"Dag.PUT", Airflow::Action::"Dag.DELETE" ], resource );

给一组用户运维权限

等价于 Flask AppBuilder 中的Op 角色：

permit( principal in Airflow::Group::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action in [ Airflow::Action::"Configuration.GET", Airflow::Action::"Configuration.LIST", Airflow::Action::"Connection.GET", Airflow::Action::"Connection.LIST", Airflow::Action::"Custom.GET", Airflow::Action::"Custom.LIST", Airflow::Action::"Dag.GET", Airflow::Action::"Dag.LIST", Airflow::Action::"Menu.MENU", Airflow::Action::"Pool.GET", Airflow::Action::"Pool.LIST", Airflow::Action::"Variable.GET", Airflow::Action::"Variable.LIST", Airflow::Action::"Asset.GET", Airflow::Action::"Asset.LIST", Airflow::Action::"View.GET", Airflow::Action::"Dag.POST", Airflow::Action::"Dag.PUT", Airflow::Action::"Dag.DELETE", Airflow::Action::"Connection.POST", Airflow::Action::"Connection.PUT", Airflow::Action::"Connection.DELETE", Airflow::Action::"Pool.POST", Airflow::Action::"Pool.PUT", Airflow::Action::"Pool.DELETE", Airflow::Action::"Variable.POST", Airflow::Action::"Variable.PUT", Airflow::Action::"Variable.DELETE", Airflow::Action::"Asset.POST", Airflow::Action::"Asset.DELETE", Airflow::Action::"Backfill.POST", Airflow::Action::"Backfill.PUT" ], resource );

给一组用户指定 DAG 的权限

下面的策略把 DAGtest的全部 DAG 相关权限授予一组用户：

permit( principal in Airflow::Group::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action, resource == Airflow::Dag::"test" );

下面的策略把 DAGfinancial-1和financial-2的全部 DAG 相关权限授予一组用户：

permit( principal in Airflow::Group::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action, resource in [Airflow::Dag::"financial-1", Airflow::Dag::"financial-2"] );

下面的策略借助context（上下文）只授予一组用户访问 DAGtest日志的权限：

permit( principal in Airflow::Group::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action, resource == Airflow::Dag::"test" ) when { context has dag_entity && context.dag_entity == "TASK_LOGS" };

其中dag_entity是 schema 中Dag类动作声明的可选 context 属性（见 avp/schema.json），用于把授权判断细化到 DAG 的某个具体子实体（例如任务日志）。在源码侧，context 由 avp/facade.py 的_build_context封装为 AVP API 要求的contextMap结构随请求发送。

禁止特定用户执行特定操作

所有定义在 Amazon Verified Permissions 中的策略在做鉴权判断时都会被考虑。例如，如果同时存在一条permit（允许）策略和一条forbid（禁止）策略匹配了同一个请求，那么访问将被拒绝。这个特性很实用：比如某个用户属于"拥有全部权限"的用户组，但你想单独限制该用户。

下面的策略把 DAGsecret-dag-1和secret-dag-2从某个特定用户的访问范围中移除：

forbid( principal == Airflow::User::"aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", action, resource in [Airflow::Dag::"secret-dag-1", Airflow::Dag::"secret-dag-2"] );

四、鉴权调用链：策略如何被执行

理解策略如何生效有助于排查授权问题。从源码看，AWS auth manager 的鉴权核心在 avp/facade.py：

is_authorized：单次鉴权，组装 principal（Airflow::User::<user_id>）、action（Airflow::Action::<Resource>.<METHOD>）、resource（未指定 entity_id 时使用*通配全部）及 entities 列表，调用 AVP 的is_authorizedAPI，返回decision == "ALLOW"即为放行；
_get_user_group_entities：把当前用户及其所属用户组构造为 AVP 的实体列表——用户实体声明其 parents 为各用户组实体。这正是策略中principal in Airflow::Group::"..."能够匹配到组内用户的底层机制；
batch_is_authorized/get_batch_is_authorized_results：批量鉴权。由于 AVP 单次batch_is_authorized调用最多支持 30 个请求（源码中以NB_REQUESTS_PER_BATCH = 30常量注释），facade 会自动把请求按 30 个一批分块发送并汇总结果；batch_is_authorized要求所有请求都返回ALLOW才算通过；
is_policy_store_schema_up_to_date：对比当前 policy store 的 schema 与仓库内置的 avp/schema.json 是否一致，若不匹配，Airflow 启动时会出现警告提示。

配套的单元测试位于 tests/unit/amazon/aws/auth_manager/，例如 test_facade.py 覆盖了鉴权请求构造与批量分块逻辑，test_entities.py 验证了实体类型与 Action ID 的生成约定，可作深入阅读参考。

五、配套资源与前置准备

policy store 的创建与 schema 更新：AWS auth manager 需要一个 policy store 存放所有策略。可以使用随附的 CLI 命令airflow aws-auth-manager init-avp一键创建（若已存在同名 policy store，出于安全原因 CLI 不会做任何修改，并提示必须手动更新 schema），或通过控制台手动创建空 policy store 后导入最新 schema；schema 升级命令为airflow aws-auth-manager update-avp-schema。CLI 的具体实现（init_avp、update_schema、dry-run 支持等）见 cli/avp_commands.py。完整步骤见 setup/amazon-verified-permissions.rst。
Airflow 配置：在[aws_auth_manager]段中设置avp_policy_store_id（policy store ID），示例见 setup/config.rst。
身份源与 Identity Center 配置：包括 SAML 元数据 URL、连接 ID 等配置项，见 setup/identity-center.rst。
JWT Token：如需通过 Airflow 公共 API 访问环境，可参考 token.rst 生成 JWT token。

六、小结与最佳实践

身份与授权分离：用户身份统一收敛在 IAM Identity Center，权限策略统一收敛在 Amazon Verified Permissions，Airflow 侧不再维护用户账号与角色映射；
按组授权：优先以用户组为单位编写principal in Airflow::Group::"..."策略，便于按团队/部门横向扩展；
细粒度控制：利用Airflow::Dag::"<dag_id>"资源限定和context（如dag_entity == "TASK_LOGS"）实现 DAG 级、乃至日志级的授权；
显式禁止优先：利用forbid策略对"全量授权组内"的个别用户做例外限制，permit 与 forbid 冲突时以拒绝为准；
引用 ID 而非名称：编写策略时，用户、用户组必须使用其在 IAM Identity Center 中的 ID；
关注 schema 一致性：升级 Airflow/Provider 后留意启动告警，必要时通过airflow aws-auth-manager update-avp-schema同步最新 schema，避免新增资源类型因 schema 缺失而无法授权。

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

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考