Apache Airflow AWS Auth Manager 实战:基于 IAM Identity Center 与 Amazon Verified Permissions 的认证授权方案
2026/9/13 9:10:51 网站建设 项目流程

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(AWS 认证管理器)是apache-airflow-providers-amazon提供的一种可插拔认证方案:它用AWS IAM Identity Center完成用户认证(登录/登出),用Amazon Verified Permissions(AVP)完成所有授权决策。本文以 AWS auth manager 官方文档 为主线,逐步讲解如何配置 Airflow 接入该认证管理器、如何在两个 AWS 服务中创建所需资源、如何用 Cedar 策略语言编写细粒度权限,以及如何生成 JWT 令牌调用 Airflow 公共 API,并结合仓库源码说明其底层实现。


一、架构总览:认证与授权两大职责的拆分

从 index.rst 可知,AWS Auth Manager 由两个 AWS 服务驱动:

  • AWS IAM Identity Center:负责认证(authentication),即登录与登出。它基于 SAML 2.0 协议与 Airflow 对接,Airflow 环境管理员可以在 Identity Center 中统一管理用户与用户组。
  • Amazon Verified Permissions:负责授权(authorization),即决定"某个用户对某个资源能否执行某个操作"。所有用户权限策略都定义在 AVP 的策略存储(policy store)中,由 Airflow 环境管理员维护。

一旦启用了 AWS Auth Manager,Airflow 环境中的所有用户及其权限不再由默认的 Flask Auth Manager(FAB)管理,而是完全交由上述两个 AWS 服务接管。这一点也体现在源码的类定义中:aws_auth_manager.py 中的AwsAuthManager继承自 Airflow 的BaseAuthManager,并在init()中校验 Airflow 版本必须大于等于 3.0.0,否则抛出AirflowOptionalProviderFeatureException(提示 "AWS auth manager is only compatible with Airflow versions >= 3.0.0"),并会调用_check_avp_schema_version()检查 AVP 策略存储的 schema 是否为最新版本。

注意(实验性):当前 AWS Auth Manager 仍处于alpha/experimental阶段,接口与行为可能在没有警告的情况下发生变化,生产环境接入前请评估风险。


二、配置 Airflow 启用 AWS Auth Manager

在 Airflow 配置中需要完成两件事:指定 auth manager 的实现类,以及指定两个 AWS 服务所在的AWS region

2.1 通过配置文件(airflow.cfg)启用

airflow.cfg中添加以下配置:

[core] auth_manager = airflow.providers.amazon.aws.auth_manager.aws_auth_manager.AwsAuthManager region_name = <region_name>

注意:region_name在这里隶属于[core]段,指 AWS IAM Identity Center 与 Amazon Verified Permissions 所在区域(例如us-east-1)。

2.2 通过环境变量启用

等效的环境变量写法为:

export AIRFLOW__CORE__AUTH_MANAGER='airflow.providers.amazon.aws.auth_manager.aws_auth_manager.AwsAuthManager' export AIRFLOW__AWS_AUTH_MANAGER__REGION_NAME='<region_name>'

2.3 一致性要求与源码佐证

  • 配置必须在所有运行 Airflow 组件的宿主机/环境间保持一致(Scheduler、Webserver、ECS Task 容器等),否则各组件认证行为会不一致。
  • 从 constants.py 可以看到 AWS Auth Manager 使用的配置键定义:配置段名为aws_auth_manager,键包括conn_idregion_namesaml_metadata_urlavp_policy_store_id
  • CLI 子命令在创建 AVP 客户端时正是读取aws_auth_manager段的region_name(见 avp_commands.py 中_get_client()的实现)。

依赖说明:源码 routes/login.py 中导入onelogin.saml2时做了兜底检查,若未安装会提示先执行pip install apache-airflow-providers-amazon[python3-saml]安装 SAML 库。因此启用 AWS Auth Manager 前请确保已安装该 extra。


三、配置 AWS IAM Identity Center(认证侧)

AWS IAM Identity Center 承担登录与登出职责,底层技术是SAML 2.0。配置完成后,Airflow 环境管理员即可通过 Identity Center 服务管理用户和用户组。

3.1 创建 Identity Center 实例

AWS Auth Manager 需要两类资源:一个instance和一个application,都需要手工创建。

  • 实例类型有若干种,但只有 Organization 级别的实例支持 SAML 2.0 应用,请务必按 AWS 官方指引创建组织级实例(步骤:在 IAM Identity Center 控制台完成启用与组织设置)。
  • 创建后,Airflow 侧的登录会重定向到该实例的登录页。

3.2 创建 Customer managed SAML 2.0 应用

在 IAM Identity Center 控制台按以下步骤创建应用:

  1. 打开IAM Identity Center控制台,选择Applications(应用)。
  2. 选择Customer managed(客户管理)选项卡。
  3. 选择Add application(添加应用)。
  4. Select application type页面,Setup preference下选择I have an application I want to set up
  5. Application type下选择SAML 2.0,然后选择Next
  6. Configure application页面,为应用填写Display name(显示名称,如Airflow)和 Description(描述)。
  7. IAM Identity Center metadata区域,复制 IAM Identity Center SAML metadata file 的地址——稍后需要把这个地址配置到 Airflow 中(saml_metadata_url)。
  8. Application metadata区域,选择Manually type your metadata values,然后按下面的要求填写Application ACS URLApplication SAML audience
配置项取值
Application ACS URL<base_url>/auth/login_callback
Application SAML audienceaws-auth-manager-saml-client

重要:将<base_url>替换为 Airflow UI 的 base URL,它应与AIRFLOW__API__BASE_URL中定义的值一致(本地运行例如localhost:8080,注意此处示例省略了http://前缀,实际部署请按 API base_url 格式填写)。

  1. 选择Submit提交,应用创建完成。

源码佐证:routes/login.py 中_init_saml_auth()构建 SP(Service Provider)配置时,entityId正是硬编码的aws-auth-manager-saml-clientassertionConsumerService的 URL 由{base_url}/auth/login_callback拼接而成——与上面填写的 ACS URL 严格对应。

3.3 配置属性映射(Attribute mappings)

创建应用后,还需配置属性映射,AWS Auth Manager 依赖两个属性:idgroups

  1. 进入刚创建应用的详情页,选择ActionsEdit attribute mappings
  2. Attribute mappings页面配置身份源与 IAM Identity Center 之间的属性映射。如果使用默认的 Identity Center 目录作为身份源,可直接使用以下配置:

id(用户唯一标识)

  • User attribute in the application:id
  • Maps to this string value or user attribute in IAM Identity Center:${user:AD_GUID}
  • Format:basic

groups(用户所属组)

  • User attribute in the application:groups
  • Maps to this string value or user attribute in IAM Identity Center:${user:groups}
  • Format:basic
  1. 两个属性(idgroups)都定义好后,选择Save changes保存。

这两处属性在认证回调中会被消费:源码 routes/login.py 在login_callback中从 SAML 断言属性里取attributes["id"][0]作为user_idattributes["groups"]作为用户组,并据此构造AwsAuthManagerUser;而 user.py 中的AwsAuthManagerUser模型包含user_idgroupsusernameemail四个字段。

3.4 在 Airflow 中补充 Identity Center 相关配置

需要将前面记录的两个值配置进 Airflow:

  • <base_url>:即配置 IAM Identity Center 应用时填写的 base URL;
  • IAM Identity Center SAML metadata file 的地址。
[api] base_url = <base_url> [aws_auth_manager] saml_metadata_url = <saml_metadata_file_url>

或使用环境变量:

export AIRFLOW__API__BASE_URL='<base_url>' export AIRFLOW__AWS_AUTH_MANAGER__SAML_METADATA_URL='<saml_metadata_file_url>'

源码层面,routes/login.py 的_get_idp_data()通过conf.get_mandatory_value("aws_auth_manager", "saml_metadata_url")读取 IdP 元数据并调用OneLogin_Saml2_IdPMetadataParser.parse_remote()拉取远程 SAML 元数据文件——因此该 URL 必须是可达的远程地址。


四、配置 Amazon Verified Permissions(授权侧)

Amazon Verified Permissions 负责所有用户授权决策,所有用户权限策略都需要由 Airflow 环境管理员在 AVP 中定义。AWS Auth Manager 需要 AVP 中的一个policy store(策略存储),可以通过随附的 CLI 命令创建,也可以手工创建。

4.1 创建 Policy Store

方式一:使用 CLI 命令

使用随 AWS Auth Manager 一起提供的 CLI 命令前,需要先将 AWS Auth Manager 设置为 Airflow 的 auth manager(见上文配置步骤),然后执行:

airflow aws-auth-manager init-avp

命令成功退出即完成创建。如果输出下面的提示,说明 AVP 中已经存在为 Airflow 创建的 policy store:

Since an existing policy store with description ... has been found in Amazon Verified Permissions, the CLI made no changes to this policy store for security reasons. Any modification to this policy store must be done manually.

出现该提示时,如需调整 schema,请参考下文"更新 Policy Store Schema"部分手工处理。

从 avp_commands.py 的init_avp实现可以看出完整流程:先通过 boto3 创建verifiedpermissions客户端(region 取自配置),尝试创建 policy store;若已存在则打印上述安全提示、不做任何改动;若为新创建则自动写入 schema,最后打印需要配置到 Airflow 中的AIRFLOW__AWS_AUTH_MANAGER__AVP_POLICY_STORE_ID值。

方式二:手工创建
  1. 打开Amazon Verified Permissions控制台。
  2. 选择Create policy store(创建策略存储)。
  3. Configuration method部分选择Empty policy store
  4. Details部分,将描述(Description)填写为Airflow
  5. 选择Create policy store,创建完成。

4.2 更新 Policy Store Schema

Schema 定义了 Airflow 环境中可用的 principal、action、resource 集合。只有以下特殊场景才需要更新 schema,否则可跳过本节:

  • 手工创建了 policy store,其中尚无 schema;
  • 你有一个用于 Airflow 的既有 policy store,曾修改过 schema 想还原;
  • 你想把既有 policy store 的 schema 升级到最新版本(当本地 schema 与 最新 schema 版本 不一致时,Airflow 启动会打印警告信息)。
方式一:CLI 命令
airflow aws-auth-manager update-avp-schema

对应源码为 avp_commands.py 中的update_schema函数,它要求--policy-store-id参数(否则抛ValueError),然后对指定 policy store 写入最新 schema。

方式二:手工更新
  1. 打开Amazon Verified Permissions控制台。
  2. 选择 Airflow 使用的 policy store(默认描述为Airflow)。
  3. 在左侧导航栏选择Schema
  4. 选择Edit schema,再选择JSON mode
  5. 将 最新 schema 版本 的内容粘贴到Contents输入框中。
  6. 选择Save changes保存。

4.3 在 Airflow 中配置 Policy Store ID

创建 policy store 后,把其 ID 配置到 Airflow:

[aws_auth_manager] avp_policy_store_id = <avp_policy_store_id>

或:

export AIRFLOW__AWS_AUTH_MANAGER__AVP_POLICY_STORE_ID='<avp_policy_store_id>'

完成以上步骤后,AWS Auth Manager 即配置就绪,可以开始管理用户与权限(详见管理章节)。


五、管理 Airflow 环境:用户与权限

启用 AWS Auth Manager 后,所有用户与权限不再由默认的 Flask Auth Manager 管理,而是通过两个 AWS 服务管理:AWS IAM Identity Center 管用户Amazon Verified Permissions 管权限

5.1 通过 IAM Identity Center 管理用户

所有能访问 Airflow 环境的用户都必须定义在 AWS IAM Identity Center 中。你可以直接使用 IAM Identity Center 作为身份源,也可以将其作为身份源(如 Active Directory)与 Airflow 环境之间的代理。

查看用户列表:打开 IAM Identity Center 控制台 → 选择Users

5.2 通过 IAM Identity Center 管理用户组

可以使用 IAM Identity Center 中的组(Groups)把用户按逻辑实体分组(如按团队、按部门)。这些组后续可以在 Amazon Verified Permissions 中被引用,从而把权限授予一组用户。

查看组列表:打开 IAM Identity Center 控制台 → 选择Groups

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

注意:IAM Identity Center 中定义的用户和组并不会自动获得 Airflow 环境的访问权限,需要手动分配哪些用户/组可以访问 Airflow。

分配步骤如下:

  1. 打开IAM Identity Center控制台。

    提示:如果用户托管在 AWS Managed Microsoft AD 中,请确保执行下一步前 IAM Identity Center 控制台使用的是 AWS Managed Microsoft AD 目录所在的 Region。

  2. 选择Applications
  3. 选择Customer managed选项卡。
  4. 在应用列表中,选择名为Airflow的应用。
  5. 在应用详情页的Assigned users and groups部分,选择Assign users and groups
  6. Assign users or groups页面,选择要分配给 Airflow 的不同用户和组(支持搜索,可通过搜索结果选择多个账户来指定多个用户或组)。
  7. 选择Assign users完成分配。

5.4 通过 Amazon Verified Permissions 管理权限

AWS Auth Manager 使用Cedar 语言在 AVP 中定义细粒度权限,所有与 Airflow 环境相关的策略存储在一个 policy store 中。

管理策略的入口:

  1. 打开Amazon Verified Permissions控制台。
  2. 选择 Airflow 使用的 policy store(默认描述为Airflow)。
  3. 在左侧导航栏选择Policies
  4. Policies页面可以查看策略列表;创建新策略:选择Create policy→ 在Create policy下选择Create static policy
5.4.1 策略格式(Cedar 语言)

Cedar 策略由三个核心要素构成:

  • Principal:谁在发起请求?
  • Action:请求方想执行什么操作?
  • Resource:请求方想对什么资源执行该操作?

在 Airflow 环境下,这三个要素各自只允许特定取值集合。可以这样查看完整的 principal、action、resource 列表:打开 AVP 控制台 → 选择 Airflow 的 policy store → 左侧导航栏选择Schema

从 schema.json 可以看到当前内置的实体类型(entityTypes)包括:AssetAssetAliasBackfillConfigurationConnectionCustomDagMenuPoolTeamGroupUserVariableView,且User通过memberOfTypes: ["Group"]声明了用户隶属于组的层级关系。Action 则按<实体类型>.<方法>命名,例如Dag.GETDag.POSTConnection.LISTVariable.DELETEMenu.MENUBackfill.POST等;DagConnectionPoolVariable等资源类型的 action 还带有可选 context 属性(如dag_entityteam_name),用于支持更精细的条件判断。

5.4.2 策略示例

以下示例来自管理文档,可以按需修改或组合,打造适合自己环境的策略。

① 授予特定用户全部权限

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

注意:必须使用用户的user ID而不是用户名或其他属性来引用用户。user ID 可以在 AWS IAM Identity Center 中查到。

② 授予一组用户全部权限(等价于 FAB 的 Admin 角色)

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

注意:必须使用组的group ID而不是组名称。group ID 可以在 AWS IAM Identity Center 中查到。

③ 授予一组用户只读权限(等价于 FAB 的 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 用户权限(等价于 FAB 的 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 );

⑤ 授予一组用户运维权限(等价于 FAB 的 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的全部相关权限:

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

给一组用户授予 DAGfinancial-1financial-2的全部相关权限:

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

给一组用户授予访问 DAGtest日志的权限(利用context.dag_entity == "TASK_LOGS"条件限定资源实体):

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" };

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

AVP 在做授权检查时会综合考量所有策略:例如同时存在一条permit和一条forbid策略命中请求时,访问将被拒绝。这非常适合"某个用户属于被授予全部权限的组,但需要单独收回其部分访问权"的场景。

下面的策略从特定用户身上移除对 DAGsecret-dag-1secret-dag-2的访问权:

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

5.5 授权决策的底层实现

从源码看,授权检查集中在 facade.py 的AwsAuthManagerAmazonVerifiedPermissionsFacade中,通过 boto3 的verifiedpermissions客户端调用 AVP 的is_authorized/batch_is_authorized接口完成。值得一提的实现细节:Amazon Verified Permissions 单次batch_is_authorized调用最多允许 30 个请求,因此 facade.py 中定义了NB_REQUESTS_PER_BATCH = 30并对批量请求做了分批处理。而 aws_auth_manager.py 中的_check_avp_schema_version()会在 Airflow 启动时比对策略存储 schema 与内置 schema.json 是否一致,不一致时打印警告日志。


六、生成 JWT 令牌以调用 Airflow 公共 API

要使用 Airflow 公共 API,需要 JWT 令牌进行认证。本小节仅适用于已配置 AWS Auth Manager 的环境。

  1. 在浏览器中访问${ENDPOINT_URL}/auth/login/token,页面会重定向到AWS IAM Identity Center 登录页
  2. 输入你的 AWS IAM Identity Center 用户凭据并提交。

流程结束后会返回一个令牌(token),在 Airflow 公共 API 请求中携带该令牌即可完成认证。

从源码看,该流程由 routes/login.py 中的/auth/login/token路由触发:login_token()构建 SAML 认证并携带login-token的 relay state 重定向到 IdP;用户在 IdP 完成登录后回跳到login_callback,此时 relay state 为login-token,服务端调用get_auth_manager().generate_jwt(user)生成 JWT,并通过LoginResponse(access_token=token)把令牌返回给调用方(见 routes/login.py)。而正常 UI 登录(relay state 为login-redirect)则会把 JWT 写入名为JWT_TOKEN的 HttpOnly Cookie 后重定向回 Airflow UI。


七、总结与延伸阅读

AWS Auth Manager 把 Airflow 环境的认证与授权完整托付给 AWS 生态:IAM Identity Center 提供 SAML 2.0 单点登录与用户/组管理,Amazon Verified Permissions 基于 Cedar 语言提供细粒度、可组合(permit/forbid)的权限策略。对于已经深度使用 AWS、希望在 Airflow 中复用企业身份体系与统一权限治理的团队,这是一条与默认 FAB 认证体系完全不同的集成路径。同时要注意:该功能当前处于实验阶段,且仅支持 Airflow >= 3.0.0,部署前需充分验证。

相关文档与源码入口:

  • AWS auth manager 总览
  • 配置 Airflow 使用 AWS auth manager
  • 配置 AWS IAM Identity Center
  • 配置 Amazon Verified Permissions
  • 管理 Airflow 环境(用户与权限)
  • 生成 JWT 令牌
  • AwsAuthManager 核心实现
  • AVP 策略存储 Schema
  • AVP 门面与批量授权实现
  • 登录路由与 SAML 对接
  • CLI 子命令定义(init-avp / update-avp-schema)

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

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询