使用Cloud Function触发Airflow2 DAG报401未授权错误如何解决

问题表现 #

通过Cloud Function触发Cloud Composer 2托管的Airflow 2 DAG时，抛出如下错误：

requests.exceptions.HTTPError: 401 Client Error: Unauthorized for url: https://<your-composer-airflow-domain>/api/v1/dags/<your-dag-id>/dagRuns

初始使用Cloud Function默认App Engine服务账号运行触发逻辑时报错，后续执行gcloud composer environments run <your-composer-env-name> --location=<your-region> users -- list列出Airflow内已有授权用户，将选中用户配置为Cloud Function运行身份后，仍然返回相同401错误。

根因 #

该报错是因为触发Airflow Stable API需要同时通过两层认证：GCP侧IAM权限校验、Airflow侧用户权限校验，仅配置其中一层就会返回401。

修复步骤 #

1. 配置GCP侧IAM权限 #

给Cloud Function实际使用的运行服务账号（不管是默认App Engine服务账号还是自定义服务账号），绑定Composer User角色（最小权限包含composer.environments.execute权限），绑定命令参考：

gcloud projects add-iam-policy-binding <your-gcp-project-id> \
  --member=serviceAccount:<your-cf-service-account-full-email> \
  --role=roles/composer.user

不建议直接给Editor/Owner等宽权限，遵循最小权限原则即可

2. 配置Airflow侧API访问权限 #

Airflow不会自动继承GCP IAM的身份权限，必须将调用方服务账号注册为Airflow合法用户，并授予DAG触发相关权限：

• 如果服务账号不在之前查询的Airflow用户列表里，执行命令创建对应用户：

gcloud composer environments run <your-composer-env-name> \
  --location=<your-region> users -- create \
  --username <your-cf-service-account-full-email> \
  --firstname Service --lastname Account \
  --role User \
  --email <your-cf-service-account-full-email>

• 如果服务账号已经存在，直接给它授予包含DAG运行权限的角色（比如Op角色，或者自定义包含dag_runs:create、dags:get权限的角色）：

gcloud composer environments run <your-composer-env-name> \
  --location=<your-region> users -- add-role \
  --username <your-cf-service-account-full-email> \
  --role Op

3. 修正Cloud Function内的认证逻辑 #

不需要在代码里硬编码任何账号密码、密钥文件，直接用Google官方Auth库自动获取当前运行环境的身份凭据，构造合法请求即可，Python运行时参考代码：

import google.auth
from google.auth.transport.requests import AuthorizedSession

# 替换为自己环境的实际配置
COMPOSER_AIRFLOW_WEB_URL = "https://<your-composer-airflow-domain>"
TARGET_DAG_ID = "<your-dag-id>"

def trigger_dag_entry(event, context):
    # 自动加载当前Cloud Function身份凭据，指定正确的权限范围
    credentials, _ = google.auth.default(
        scopes=["https://www.googleapis.com/auth/cloud-platform"]
    )
    session = AuthorizedSession(credentials)
    
    # 构造DAG触发请求
    response = session.post(
        f"{COMPOSER_AIRFLOW_WEB_URL}/api/v1/dags/{TARGET_DAG_ID}/dagRuns",
        json={
            "conf": {} # 这里传入需要传给DAG的自定义参数
        }
    )
    response.raise_for_status()
    return response.json()

不要手动拼接Authorization请求头，用官方AuthorizedSession会自动处理令牌签名、刷新逻辑，避免令牌格式错误导致的401

4. 异常排查 #

如果以上步骤配置完成后仍然报错，逐一检查以下配置项：

• 确认Composer环境Airflow配置中[api] auth_backends项包含airflow.composer.api.backend.ComposerAuthBackend，Composer 2默认开启该配置，手动修改后会导致认证链路失效
• 确认代码中填写的Airflow Web UI地址和当前环境匹配，地址可在Composer环境控制台详情页直接复制
• 确认目标DAG ID存在且未被暂停，DAG不存在一般返回404，可作为辅助排查项

内容的提问来源于stack exchange，提问作者Douglas Correa