使用Azure AD JWT认证Python Azure Function HTTP请求的最佳实践是什么

前置依赖安装 #

首先安装需要的第三方库，在项目的requirements.txt中添加以下内容：

azure-functions
PyJWT>=2.8.0
cryptography>=41.0.0
requests>=2.31.0
cachetools>=5.3.0

核心JWT校验逻辑封装 #

封装通用校验工具，内置JWKS公钥缓存、签名校验、核心声明校验、访问范围校验能力，可在所有函数中复用：

import jwt
import requests
from cachetools import TTLCache
from typing import Optional, List
import os

# 缓存JWKS公钥，缓存时长24小时，避免每次请求重复拉取公钥
jwks_cache = TTLCache(maxsize=1, ttl=86400)

# 从函数环境变量读取配置，无需修改获取逻辑，直接在Azure门户配置对应值即可
TENANT_ID = os.environ["AZURE_AD_TENANT_ID"]
CLIENT_ID = os.environ["AZURE_AD_CLIENT_ID"]
AZURE_AD_JWKS_URI = f"https://login.microsoftonline.com/{TENANT_ID}/discovery/v2.0/keys"
EXPECTED_ISSUER = f"https://login.microsoftonline.com/{TENANT_ID}/v2.0"

def validate_jwt_token(auth_header: Optional[str], required_scopes: List[str]) -> dict:
    """
    校验JWT令牌合法性和访问范围
    :param auth_header: 请求头中的Authorization字段值
    :param required_scopes: 当前接口要求的访问范围列表
    :return: 解析后的令牌声明，校验失败直接抛出权限异常
    """
    # 校验Authorization头格式
    if not auth_header or not auth_header.startswith("Bearer "):
        raise PermissionError("无效的认证头")
    token = auth_header.split(" ")[1]

    # 拉取并缓存JWKS公钥集
    if "jwks" not in jwks_cache:
        resp = requests.get(AZURE_AD_JWKS_URI)
        resp.raise_for_status()
        jwks_cache["jwks"] = resp.json()
    jwks = jwks_cache["jwks"]

    # 解析未验证的令牌头获取匹配的签名密钥
    try:
        unverified_header = jwt.get_unverified_header(token)
    except jwt.InvalidTokenError:
        raise PermissionError("无效的令牌格式")
    
    rsa_key = {}
    for key in jwks["keys"]:
        if key["kid"] == unverified_header["kid"]:
            rsa_key = {
                "kty": key["kty"],
                "kid": key["kid"],
                "use": key["use"],
                "n": key["n"],
                "e": key["e"]
            }
            break
    if not rsa_key:
        raise PermissionError("未找到匹配的签名公钥")

    # 校验令牌签名、过期时间、签发人、受众等核心属性
    try:
        payload = jwt.decode(
            token,
            key=rsa_key,
            algorithms=["RS256"],
            issuer=EXPECTED_ISSUER,
            audience=CLIENT_ID,
            options={"verify_exp": True, "verify_nbf": True}
        )
    except jwt.InvalidTokenError as e:
        raise PermissionError(f"令牌校验失败: {str(e)}")

    # 校验要求的访问范围
    token_scopes = payload.get("scp", "").split()
    for scope in required_scopes:
        if scope not in token_scopes:
            raise PermissionError(f"缺少必要的访问范围: {scope}")
    
    return payload

函数入口调用示例 #

以HTTP触发的Azure Function为例，调用上述校验逻辑完成认证：

import azure.functions as func
from .jwt_validator import validate_jwt_token # 引入上述封装的校验函数

app = func.FunctionApp()

@app.function_name(name="DemoApi")
@app.route(route="demo", auth_level=func.AuthLevel.ANONYMOUS) # 函数层级认证设为匿名，使用自定义逻辑做Azure AD认证
def demo_api(req: func.HttpRequest) -> func.HttpResponse:
    try:
        # 校验令牌，示例要求必须有access_as_user范围
        token_payload = validate_jwt_token(
            auth_header=req.headers.get("Authorization"),
            required_scopes=["access_as_user"]
        )
        # 校验通过后可从payload中获取用户信息，比如用户oid、名称等
        user_id = token_payload.get("oid")
    except PermissionError as e:
        return func.HttpResponse(str(e), status_code=401)
    
    # 执行业务逻辑
    return func.HttpResponse(f"Hello, user {user_id}", status_code=200)

关键注意事项 #

• 函数层级的auth_level必须设为ANONYMOUS，才能让请求进入自定义认证逻辑，否则会被Azure Function自带的基础认证拦截
• 上述逻辑默认包含令牌过期、签名、签发人、受众的强制校验，无需额外开发
• 如果需要适配其他Azure AD环境，仅需修改JWKS地址和签发人前缀即可，核心校验逻辑无需调整

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