如何使用Python/FastAPI更新Azure AD B2C中的用户自定义属性

前置准备 #

• 在Azure AD B2C租户中注册应用，授予Directory.ReadWrite.All应用权限（需为应用权限而非委托权限，适配后台操作场景），并完成管理员同意。
• 记录应用的客户端ID、客户端密钥、租户的租户ID。

Python基础脚本示例 #

先通过基础脚本验证核心调用逻辑：

import requests

# 配置信息，建议用环境变量替代硬编码
TENANT_ID = "你的租户ID"
CLIENT_ID = "你的客户端ID"
CLIENT_SECRET = "你的客户端密钥"
USER_OBJECT_ID = "目标用户的Object ID"
# 自定义属性命名规则：extension_<去掉连字符的应用ID>_<属性名>
CUSTOM_ATTR_KEY = "extension_abc123def456_favoriteColor"

def get_access_token():
    token_url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "https://graph.microsoft.com/.default"
    }
    response = requests.post(token_url, data=payload)
    response.raise_for_status()
    return response.json()["access_token"]

def update_user_attr(token):
    graph_url = f"https://graph.microsoft.com/v1.0/users/{USER_OBJECT_ID}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        CUSTOM_ATTR_KEY: "blue"  # 替换为目标属性值
    }
    response = requests.patch(graph_url, json=payload, headers=headers)
    response.raise_for_status()
    print("属性更新成功")

if __name__ == "__main__":
    token = get_access_token()
    update_user_attr(token)

FastAPI集成示例 #

将逻辑封装为接口，适配服务端场景：

from fastapi import FastAPI, HTTPException
import requests
import os

app = FastAPI()

# 从环境变量读取敏感配置
TENANT_ID = os.getenv("B2C_TENANT_ID")
CLIENT_ID = os.getenv("B2C_CLIENT_ID")
CLIENT_SECRET = os.getenv("B2C_CLIENT_SECRET")

def get_access_token():
    token_url = f"https://login.microsoftonline.com/{TENANT_ID}/oauth2/v2.0/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "https://graph.microsoft.com/.default"
    }
    try:
        response = requests.post(token_url, data=payload)
        response.raise_for_status()
        return response.json()["access_token"]
    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"获取令牌失败: {str(e)}")

@app.patch("/b2c-users/{user_id}/custom-attr")
async def update_b2c_user_attr(
    user_id: str,
    attr_key: str,
    attr_value: str
):
    token = get_access_token()
    graph_url = f"https://graph.microsoft.com/v1.0/users/{user_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {attr_key: attr_value}
    try:
        response = requests.patch(graph_url, json=payload, headers=headers)
        response.raise_for_status()
        return {"status": "success", "msg": "用户属性已更新"}
    except requests.exceptions.RequestException as e:
        raise HTTPException(status_code=500, detail=f"更新失败: {str(e)}")

关键注意事项 #

• 自定义属性命名：必须遵循extension_<无连字符应用ID>_<属性名>格式，可在Azure AD B2C的“用户属性”页面查看已创建属性的完整键名。
• 权限校验：确保应用已获得Directory.ReadWrite.All应用权限并完成管理员同意，否则会返回权限不足错误。
• 安全优化：敏感配置（如客户端密钥）不要硬编码，使用环境变量或Azure Key Vault存储。
• 错误处理：生产环境需补充令牌自动刷新、用户不存在等场景的异常捕获逻辑。

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