AWS Lambda部署的FastAPI全局JWT校验遇服务器错误问题排查

问题背景 #

在AWS Lambda上部署FastAPI服务，通过API Gateway对外提供接口，要求所有请求必须携带有效JWT令牌。使用FastAPI中间件实现全局校验时，令牌缺失或无效时抛出HTTPException会引发服务器错误，无法正常返回403状态码并拦截请求。

解决方案 #

FastAPI的HTTPException是由路由层的异常处理器处理的，底层中间件中抛出该异常无法被正确捕获，需直接返回Starlette的响应对象替代抛出异常。

修改后的main.py代码 #

from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, status
from starlette.responses import JSONResponse  # 新增导入
from mangum import Mangum
from pymongo import MongoClient
import config
from routes import project_router, forms_models_router, inspections_router, structures_router
import auth


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("Starting connection to database...")
    app.mongodb_client = MongoClient(config.mongodb_url)
    app.database = app.mongodb_client[config.mongo_database]
    yield
    app.mongodb_client.close()
    print("Database connection closed.")

app = FastAPI(lifespan=lifespan)

@app.middleware("http")
async def check_token(request: Request, call_next):
    headers = request.headers
    try:
        auth_header = headers["Authorization"]
        # 可选：处理Bearer前缀（如果令牌格式是Bearer <token>）
        # if not auth_header.startswith("Bearer "):
        #     return JSONResponse(
        #         status_code=status.HTTP_403_FORBIDDEN,
        #         content={"detail": "Invalid authentication scheme."}
        #     )
        # encoded_jwt = auth_header.split(" ")[1]
        
        encoded_jwt = auth_header
        if not auth.check_jwt(encoded_jwt, "mekas", config.jwt_secret):
            return JSONResponse(
                status_code=status.HTTP_403_FORBIDDEN,
                content={"detail": "Invalid authentication."}
            )
    except KeyError:
        return JSONResponse(
            status_code=status.HTTP_403_FORBIDDEN,
            content={"detail": "Invalid authentication."}
        )
    # 校验通过，继续处理请求
    response = await call_next(request)
    return response

app.include_router(project_router, tags=["projects"], prefix="/projects")
app.include_router(forms_models_router, tags=["forms_models"], prefix="/forms_models")
app.include_router(inspections_router, tags=["inspections"], prefix="/inspections")
app.include_router(structures_router, tags=["structures"])

handler = Mangum(app)

关键修改点 #

• 导入starlette.responses.JSONResponse，用于直接构造错误响应
• 将原代码中抛出HTTPException的逻辑替换为返回JSONResponse，指定403状态码和错误详情
• 可选：添加对Authorization头格式的校验（比如是否以Bearer 开头），进一步增强安全性

原理说明 #

Starlette中间件处于请求处理的底层链路，抛出FastAPI的HTTPException不会被上层的异常处理器捕获，导致未处理异常引发服务器错误。直接返回JSONResponse可以绕过这个问题，直接向客户端返回标准的403错误响应，同时终止后续路由处理流程。

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