Airflow 2.3.3配置Azure AD OAuth后，如何通过OAuth认证调用Rest API？

核心原理 #

Airflow Web GUI通过Azure AD OAuth2认证后会生成session cookie，API默认校验该cookie的有效性。只需先完成OAuth2授权流程获取Airflow的session cookie，即可用此cookie调用所有API接口，无需使用Basic Auth。

步骤1：获取Azure AD授权码 #

需用到Azure AD配置的客户端ID、租户ID、回调URL（需与Azure AD中配置完全一致，例如http://<your-airflow-host>/oauth-authorized/azure）。

cURL示例（手动获取授权码） #

# 替换占位符为你的实际配置
AZURE_TENANT_ID="your-tenant-id"
CLIENT_ID="your-client-id"
AIRFLOW_HOST="http://localhost:8080"
REDIRECT_URI="${AIRFLOW_HOST}/oauth-authorized/azure"

# 生成授权URL，复制到浏览器打开，登录后从跳转地址栏提取code参数
echo "在浏览器打开此URL：https://login.microsoftonline.com/${AZURE_TENANT_ID}/oauth2/v2.0/authorize?client_id=${CLIENT_ID}&response_type=code&redirect_uri=${REDIRECT_URI}&response_mode=query&scope=openid%20email%20profile&state=random-state-string"

步骤2：用授权码交换Airflow会话cookie #

拿到授权码后，调用Airflow的OAuth回调接口，将session cookie保存到本地文件。

cURL示例 #

# 替换占位符
AIRFLOW_HOST="http://localhost:8080"
AUTH_CODE="your-authorized-code"
CLIENT_ID="your-client-id"
CLIENT_SECRET="your-client-secret"
REDIRECT_URI="${AIRFLOW_HOST}/oauth-authorized/azure"

# 发送请求并保存session cookie到文件
curl -v "${AIRFLOW_HOST}/oauth-authorized/azure" \
  -d "code=${AUTH_CODE}" \
  -d "client_id=${CLIENT_ID}" \
  -d "client_secret=${CLIENT_SECRET}" \
  -d "redirect_uri=${REDIRECT_URI}" \
  -d "grant_type=authorization_code" \
  -c airflow_session_cookies.txt

步骤3：用会话cookie调用Airflow API #

使用保存的cookie调用API接口，例如触发DAG运行。

cURL示例（触发DAG） #

AIRFLOW_HOST="http://localhost:8080"
DAG_ID="your-dag-id"

# 携带cookie调用API
curl "${AIRFLOW_HOST}/api/v1/dags/${DAG_ID}/dagRuns" \
  -b airflow_session_cookies.txt \
  -X POST \
  -H "Content-Type: application/json" \
  -d '{"conf": {}}'

Python示例（完整流程） #

import requests
from urllib.parse import urlencode

# 配置参数，替换为你的实际值
AZURE_TENANT_ID = "your-tenant-id"
CLIENT_ID = "your-client-id"
CLIENT_SECRET = "your-client-secret"
AIRFLOW_HOST = "http://localhost:8080"
REDIRECT_URI = f"{AIRFLOW_HOST}/oauth-authorized/azure"
DAG_ID = "your-dag-id"

# 1. 生成授权URL，手动打开获取code
auth_params = {
    "client_id": CLIENT_ID,
    "response_type": "code",
    "redirect_uri": REDIRECT_URI,
    "response_mode": "query",
    "scope": "openid email profile",
    "state": "random-state"
}
auth_url = f"https://login.microsoftonline.com/{AZURE_TENANT_ID}/oauth2/v2.0/authorize?{urlencode(auth_params)}"
print(f"打开以下URL获取授权码：\n{auth_url}")
auth_code = input("输入地址栏中的code参数：")

# 2. 交换Airflow会话cookie
session = requests.Session()
callback_response = session.post(
    f"{AIRFLOW_HOST}/oauth-authorized/azure",
    data={
        "code": auth_code,
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "redirect_uri": REDIRECT_URI,
        "grant_type": "authorization_code"
    }
)
if callback_response.status_code not in [200, 302]:
    print(f"会话获取失败：{callback_response.text}")
    exit(1)

# 3. 调用API触发DAG
api_response = session.post(
    f"{AIRFLOW_HOST}/api/v1/dags/{DAG_ID}/dagRuns",
    json={"conf": {}}
)
print(f"API响应状态码：{api_response.status_code}")
print(f"API响应内容：{api_response.json()}")

注意事项 #

• 确保Azure AD应用的重定向URI与代码配置完全一致，包括HTTP/HTTPS协议
• CLIENT_SECRET需妥善保管，建议通过环境变量传递，避免硬编码
• 会话cookie存在过期时间，过期后需重新执行授权流程
• 自动化场景可考虑Azure AD客户端凭证模式，需Airflow配置支持或结合服务账号

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