基于Python的Azure Function自定义JWT认证授权中间件实现问题

问题本质 #

你当前的实现存在两个关键问题：

1. pre_invocation_app_level方法的返回值类型固定为None，直接返回func.HttpResponse不会生效，无法中断请求并返回未授权响应
2. 没有有效机制将JWT验证后的用户信息传递到后续的端点函数中

可行解决方案 #

1. 验证失败时中断请求并返回响应 #

通过抛出自定义异常，结合Azure Functions的全局异常处理逻辑，实现验证失败时直接返回指定的HttpResponse。

2. 传递JWT验证结果到端点 #

修改func_args字典，将验证后的用户信息注入其中，端点函数可直接通过同名参数接收该信息。

完整实现代码 #

第一步：定义自定义异常与认证中间件

from azure.functions import AppExtensionBase, Context
import typing
from logging import Logger
import azure.functions as func
import jwt  # 需先安装：pip install pyjwt

# 自定义未授权异常，携带响应信息
class UnauthorizedException(Exception):
    def __init__(self, message: str = "Unauthorized", headers: dict = None):
        self.message = message
        self.headers = headers or {"WWW-Authenticate": "Custom"}

class AuthMiddleware(AppExtensionBase):
    @classmethod
    def pre_invocation_app_level(
        cls,
        logger: Logger,
        context: Context,
        func_args: typing.Dict[str, object] = ...,
        *args,
        **kwargs,
    ) -> None:
        req = func_args.get("req")
        if not req:
            raise UnauthorizedException("Invalid request")
        
        auth_header = req.headers.get("Authorization")
        if not auth_header or not auth_header.startswith("Bearer "):
            raise UnauthorizedException()
        
        # 提取并验证JWT Token
        token = auth_header.split(" ")[1]
        try:
            # 替换为你的实际JWT密钥、算法及验证规则（如校验iss/aud）
            payload = jwt.decode(token, "your-jwt-secret", algorithms=["HS256"])
            # 将验证后的用户信息注入func_args，端点函数可通过user参数接收
            func_args["user"] = payload
        except jwt.InvalidTokenError:
            raise UnauthorizedException("Invalid or expired token")

第二步：注册中间件与全局异常处理

在函数应用入口文件（如__init__.py）中完成中间件注册和异常处理配置：

import azure.functions as func
from .auth_middleware import AuthMiddleware, UnauthorizedException

# 初始化函数应用并注册认证中间件
app = func.FunctionApp()
app.register_extension(AuthMiddleware)

# 全局捕获未授权异常并返回响应
@app.exception_handler(UnauthorizedException)
def handle_unauthorized(req: func.HttpRequest, exc: UnauthorizedException) -> func.HttpResponse:
    return func.HttpResponse(
        exc.message,
        status_code=401,
        headers=exc.headers
    )

第三步：端点函数接收验证后的用户信息

@app.route(route="protected", methods=["GET"])
def protected_endpoint(req: func.HttpRequest, user: dict) -> func.HttpResponse:
    # 直接使用验证后的用户数据
    return func.HttpResponse(f"欢迎访问受保护接口，当前用户：{user.get('username')}")

关键说明 #

• pre_invocation_app_level中修改func_args后，端点函数只需声明同名参数即可自动获取注入的用户信息
• 自定义异常+全局异常处理是目前Python Worker Extension中实现请求中断并返回自定义响应的可靠方式
• JWT验证逻辑需根据你的业务场景调整，比如从环境变量读取密钥、校验签发者/受众等

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