Apache Airflow Simple Auth Manager Token API 参考指南：登录、鉴权与 JWT Token 获取全解析

Apache Airflow Simple Auth Manager Token API 参考指南：登录、鉴权与 JWT Token 获取全解析

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

本文以 Apache Airflow（Airflow 3）仓库中的 Simple auth manager Token API 参考文档为骨架，完整讲解由该文档渲染出的 OpenAPI 规格（v2-simple-auth-manager-generated.yaml）所定义的 4 个认证端点：POST /auth/token、GET /auth/token、GET /auth/token/login与POST /auth/token/cli。文章同时深入路由（routes/login.py）、登录服务（services/login.py）与数据模型（datamodels/login.py）的实现，帮助读者掌握在 simple auth manager 环境下如何完成用户登录、获取 JWT Token、驱动 Airflow 公共 API 以及 CLI 认证的完整链路。

Simple auth manager 在 Airflow 3 中的定位

在开始阅读 Token API 之前，需要先理解它在整体架构中的位置。Apache Airflow 将"用户认证 + 用户授权"封装为可插拔的Auth manager组件：所有 auth manager 实现共同的公开接口 BaseAuthManager，并可通过配置自由切换。Airflow 在同一时刻只允许启用一个 auth manager，由[core]配置段的auth_manager选项决定，详见 auth-manager 概念文档。

Simple auth manager 是 Airflow 3 中随发行版内置且默认启用的 auth manager，其实现位于 simple_auth_manager.py。官方文档明确提示：

The Simple auth manager is intended for development and testing. If you're using it in production, ensure that access is controlled through other means.（Simple auth manager 面向开发与测试场景；若在生产中使用，必须通过其他手段控制访问。）

可以随时用下面的命令确认当前环境启用的是哪个 auth manager：

$ airflow config get-value core auth_manager airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager

按文档目录中的导航结构，本页（sam-token-api-ref.rst）是与 simple auth manager 使用指南 和 生成 JWT Token 操作指南 并列的三个组成部分之一：其中index.rst负责讲解用户/角色/团队的管理配置，token.rst给出调用姿势，而本页则以Swagger-UI 插件形式嵌入完整的 OpenAPI 3.1 规格文件，即 v2-simple-auth-manager-generated.yaml。

该 YAML 规格的 info 段说明如下：

info: title: Simple auth manager sub application description: > This is the simple auth manager fastapi sub application. This API is only available if the auth manager used in the Airflow environment is simple auth manager. This sub application provides the login form for users to log in. version: 0.1.0

关键约束有二：

1. 仅当环境配置为 simple auth manager 时，下列端点才可用。若切换为其他 auth manager（如 FAB），对应路径与语义会完全不同。
2. 该子应用由 auth manager 通过get_fastapi_app方法挂载到/auth路径下（参见 auth-manager 概念文档 中 "Extending API server application" 一节），因此最终端点地址统一带有/auth前缀。

Token API 端点总览

规格文件 v2-simple-auth-manager-generated.yaml 在paths下定义了 4 个端点，全部带有SimpleAuthManagerLogin标签：

这 4 个端点分别对应 routes/login.py 中通过AirflowRouter（tags 为SimpleAuthManagerLogin）注册的 4 个函数，规格文件正是从这些路由自动生成的。下面逐一解析。

POST /auth/token —— 用户名密码换取 JWT Token

这是最核心、最常用的登录端点，对应路由实现：

@login_router.post( "/token", status_code=status.HTTP_201_CREATED, ... ) def create_token(body: LoginBody = Depends(parse_login_body)) -> LoginResponse: """Authenticate the user.""" return LoginResponse(access_token=SimpleAuthManagerLogin.create_token(body=body))

源码出处：routes/login.py 中的create_token。

请求

根据 OpenAPI 规格，请求体（requestBody）为必填，且支持两种 Content-Type：

• application/json：{"username": "<username>", "password": "<password>"}
• application/x-www-form-urlencoded：同样按LoginBody结构提交表单

请求体模型LoginBody定义（datamodels/login.py 中的StrictBaseModel）：

LoginBody: type: object additionalProperties: false # 不接受额外字段 required: [username, password] # 两者都必填 properties: username: { type: string } password: { type: string }

路由通过Depends(parse_login_body)解析请求体，parse_login_body位于 utils.py：它根据请求的Content-Type头决定走 JSON 解析还是application/x-www-form-urlencoded解析。由于规格允许的Content-Type只有上述两种，其他媒体类型（如text/plain、multipart/form-data）会命中415 Unsupported Media Type响应。

认证逻辑与状态码语义

认证的核心逻辑在登录服务 services/login.py 的SimpleAuthManagerLogin.create_token静态方法中，整个流程为：

1. 读取 all-admins 开关：若配置[core] simple_auth_manager_all_admins = True，则跳过全部密码校验，直接签发匿名管理员 Token（返回Anonymous/ADMIN用户），此时传入的 body 即使为空字符串也能成功。
2. 校验凭据完整性：若开关关闭且username或password为空，抛出400 Bad Request，错误信息为Username and password must be provided。
3. 校验用户与密码：通过SimpleAuthManager.get_users()与SimpleAuthManager.get_passwords()读取配置定义的用户及密码文件；匹配时使用hmac.compare_digest做常量时间比较（源码注释明确标注针对 CWE-208，避免时序侧信道）。当用户不存在或密码不匹配时抛出401 Unauthorized，错误信息为Invalid credentials。
4. 签发 Token：认证通过后构造SimpleAuthManagerUser，调用get_auth_manager().generate_jwt(...)生成 JWT，Token 有效期由api_auth.jwt_expiration_time配置决定（默认 86400 秒，见下文配置小节）。

因此本端点的响应码完整集合是：

注意成功码是201 Created（路由显式声明了status.HTTP_201_CREATED），而非常见的200 OK。

响应

成功后返回模型LoginResponse（仅一个字段）：

LoginResponse: type: object required: [access_token] properties: access_token: { type: string }

实际 JSON 形如：

{ "access_token": "<jwt-token-string>" }

GET /auth/token —— 全管理员模式的免凭据取 Token

当配置[core] simple_auth_manager_all_admins = True（关闭认证、人人都是管理员）时，可通过GET /auth/token免凭据直接换取 Token：

@login_router.get( "/token", status_code=status.HTTP_201_CREATED, responses=create_openapi_http_exception_doc([status.HTTP_403_FORBIDDEN]), ) def create_token_all_admins() -> LoginResponse: """Create a token with no credentials only if ``simple_auth_manager_all_admins`` is True.""" return LoginResponse(access_token=SimpleAuthManagerLogin.create_token_all_admins())

源码出处：routes/login.py 中的create_token_all_admins。

底层调用 services/login.py 的create_token_all_admins：

is_simple_auth_manager_all_admins = conf.getboolean("core", "simple_auth_manager_all_admins") if not is_simple_auth_manager_all_admins: raise HTTPException( status.HTTP_403_FORBIDDEN, "This method is only allowed if ``[core] simple_auth_manager_all_admins`` is True", ) return SimpleAuthManagerLogin._create_anonymous_admin_user(...)

若开关关闭，则返回403 Forbidden。该端点成功时同样是201，body 与POST /auth/token一致（LoginResponse）。被签发的匿名管理员用户名为Anonymous、角色为ADMIN，拥有全部权限。

GET /auth/token/login —— 免凭据登录并种 Cookie

第三个端点是用于UI 浏览器登录流程的免凭据端点（同样仅在 all-admins 模式下可用）：

@login_router.get( "/token/login", status_code=status.HTTP_307_TEMPORARY_REDIRECT, ... ) def login_all_admins(request: Request) -> RedirectResponse: """Login the user with no credentials.""" fallback_url = conf.get("api", "base_url", fallback="/") next_url = request.query_params.get("next") redirect_url = next_url if next_url and is_safe_url(next_url, request=request) else fallback_url response = RedirectResponse(url=redirect_url) # The default config has this as an empty string, so we can't use `has_option`. # And look at the request info (needs `--proxy-headers` flag to api-server) secure = request.base_url.scheme == "https" or bool(conf.get("api", "ssl_cert", fallback="")) response.set_cookie( COOKIE_NAME_JWT_TOKEN, SimpleAuthManagerLogin.create_token_all_admins(), path=get_cookie_path(), secure=secure, httponly=True, samesite="lax", ) return response

源码出处：routes/login.py 中的login_all_admins。

行为要点：

• 校验逻辑与GET /auth/token相同，all-admins 开关关闭时返回403 Forbidden。
• 成功时不再返回 JSON Token，而是把 JWT 写入名为_token（COOKIE_NAME_JWT_TOKEN，定义于 base_auth_manager.py）的HttpOnly Cookie，然后307 临时重定向。这与 auth-manager 概念文档 中约定的协议一致：auth manager 需先把 JWT 存进_tokenCookie 再跳转到 Airflow UI，UI 读取后即保存并删除该 Cookie。注意 Cookie 参数httponly=True必须保留，UI 自身不管理 Token。
• 重定向目标：优先使用查询参数next指定的地址，但必须通过is_safe_url校验（防止开放重定向）；未提供或不安全时回退到api.base_url（默认/）。
• Cookie 的secure标志按当前请求协议动态判定：当请求为https://或配置了api.ssl_cert时置为 True。

POST /auth/token/cli —— CLI 专属的 Token 端点

Airflow CLI 在调用 API 时使用的是生命周期更短的 JWT，对应端点：

@login_router.post( "/token/cli", status_code=status.HTTP_201_CREATED, responses=create_openapi_http_exception_doc([status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED]), ) def create_token_cli(body: LoginBody) -> LoginResponse: """Authenticate the user for the CLI.""" return LoginResponse( access_token=SimpleAuthManagerLogin.create_token( body=body, expiration_time_in_seconds=conf.getint("api_auth", "jwt_cli_expiration_time") ) )

源码出处：routes/login.py 中的create_token_cli。

它与POST /auth/token的差异仅在于 Token 有效期：显式使用api_auth.jwt_cli_expiration_time（默认 3600 秒）而不是通用的api_auth.jwt_expiration_time（默认 86400 秒）。请求体同样为LoginBody（仅 JSON，无需表单分支），错误码集合为400（凭据缺失）、401（凭据错误）、422（校验失败），成功返回201与LoginResponse。

错误响应与校验模型

规格文件components.schemas还统一定义了供各端点复用的错误与校验模型：

• HTTPExceptionResponse：仅一个必填字段detail，其值可以是字符串，也可以是任意对象（additionalProperties: true），对应 FastAPI 抛出HTTPException时的响应体。例如上面提到的{"detail": "Invalid credentials"}。
• HTTPValidationError：FastAPI/Pydantic 在请求体校验失败（422）时的标准结构，其detail是ValidationError数组。
• ValidationError：包含loc（出错位置，元素为字符串或整数）、msg、type、input、ctx（可选上下文）等字段，遵循 Pydantic v2 校验错误约定。

实战：完整调用流程

从登录到调用 Airflow 公共 API

在 simple auth manager 环境下使用 Airflow 公共 API 的标准流程在 token.rst 中有完整示例：先用POST /auth/token拿 Token，再把 Token 放入后续 API 请求。

第一步，携带用户名密码换取 JWT：

ENDPOINT_URL="http://localhost:8080" curl -X 'POST' \ "${ENDPOINT_URL}/auth/token" \ -H 'Content-Type: application/json' \ -d '{ "username": "<username>", "password": "<password>" }'

响应中的access_token即所需 JWT，可放入 Airflow 公共 API 请求的Authorization头使用（按公共 API 约定为Bearer前缀）：

curl -X 'GET' "${ENDPOINT_URL}/api/v1/dags" \ -H "Authorization: Bearer <access_token>"

第二步，若环境开启了[core] simple_auth_manager_all_admins，则可以不提供任何凭据直接获取：

ENDPOINT_URL="http://localhost:8080" curl -X 'GET' "${ENDPOINT_URL}/auth/token"

同样返回LoginResponse结构的access_token。

提示：simple auth manager 环境下，用户由配置定义，密码由系统自动生成并打印在 webserver 日志中，同时持久化到core.simple_auth_manager_passwords_file指定的 JSON 文件（默认$AIRFLOW_HOME/simple_auth_manager_passwords.json.generated，可直接读取或更新）。在 Breeze 开发环境里预置了用户admin与viewer（密码与用户名相同），分别拥有全部权限与只读权限。关于用户、角色（viewer / user / op / admin）与多团队配置，请参阅 simple auth manager 使用指南。

以 form-urlencoded 方式登录

由于POST /auth/token同时接受表单编码，浏览器原生表单或习惯使用-d键值对的场景可以这样请求：

curl -X 'POST' "${ENDPOINT_URL}/auth/token" \ -H 'Content-Type: application/x-www-form-urlencoded' \ -d 'username=<username>&password=<password>'

开启 all-admins 模式的 UI 免密登录

若以浏览器访问 UI 并开启了 all-admins 模式，可直接打开：

http://localhost:8080/auth/token/login

服务端会校验next查询参数（存在且通过安全校验则跳转到该地址），把 JWT 种入_tokenHttpOnly Cookie 后完成登录。

关键配置速查

综合规格说明、路由源码与 config.yml 中的默认值，与该 Token API 直接相关的配置如下（配置归属段均为 Airflow 3 的 FastAPI 配置体系，修改后通常需重启 api-server 生效）：

版本相关配置提示：config.yml标注jwt_expiration_time、jwt_cli_expiration_time等配置在版本 3.0.0 加入，且指出集群中所有运行 Airflow 组件的机器时间必须同步（建议使用 ntpd），否则可能因时钟偏差出现 "forbidden" 错误。

源码与测试验证指引

若要进一步在仓库中验证上述行为，可关注以下文件：

• 规格定义：v2-simple-auth-manager-generated.yaml —— 本文解析的完整 OpenAPI 3.1 文档。
• 路由实现：routes/login.py —— 4 个端点的 FastAPI 路由声明、状态码与响应文档。
• 认证服务：services/login.py —— 常量时间密码比对、all-admins 逻辑与 JWT 签发。
• 数据模型：datamodels/login.py ——LoginBody与LoginResponse的 Pydantic 定义。
• 请求体解析：utils.py ——parse_login_body对 JSON / form-urlencoded 的分发。
• 单元测试：test_login.py 与 test_login.py —— 覆盖各端点正常路径、错误码以及 all-admins 开关行为。

简单总结：Simple auth manager 的 Token API 用最小的表面积覆盖了"认证 → 取 Token → 调用 API/CLI"的全部需求。普通用户使用POST /auth/token凭据换 Token；开发/测试环境可开启[core] simple_auth_manager_all_admins = True后，通过GET /auth/token免凭据取 Token、通过GET /auth/token/login完成浏览器 Cookie 登录；CLI 场景则统一走生命周期更短的POST /auth/token/cli。四个端点组合在一起，正好对应 Airflow 3 在 simple auth manager 环境下完整的 JWT 认证闭环。

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