扫描二维码
关注或者微信搜一搜：编程智域前端至全栈交流与成长

发现1000+提升效率与开发的AI工具和实用程序：https://tools.cmdragon.cn/

1. FastAPI安全中间件与请求拦截处理详解

1.1 中间件基础原理

中间件（Middleware）是处理HTTP请求的管道机制，如同安检系统对旅客的逐层检查。在FastAPI中，中间件能对进入的请求和返回的响应进行拦截处理，典型应用场景包括：

身份认证鉴权
请求日志记录
流量控制
数据格式验证
异常统一处理

框架采用Starlette中间件系统，支持同步/异步处理模式。中间件执行顺序与注册顺序相反，响应阶段按注册逆序执行，形成”洋葱模型”。

1.2 核心安全中间件使用

1.2.1 基础中间件注册

from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware

app = FastAPI()


class AuditMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 请求前处理
        print(f"收到 {request.method} 请求: {request.url}")

        # 传递请求到下级处理
        response = await call_next(request)

        # 响应后处理
        response.headers["X-Audit"] = "processed"
        return response


app.add_middleware(AuditMiddleware)

1.2.2 安全头设置

使用安全头中间件增强防护：

from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware

# 强制HTTPS
app.add_middleware(HTTPSRedirectMiddleware)

# 域名白名单
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["*.example.com"])

1.3 请求拦截实战案例

1.3.1 IP黑名单拦截

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

BLACKLIST_IPS = {"192.168.1.100", "10.0.0.5"}


@app.middleware("http")
async def ip_filter(request: Request, call_next):
    client_ip = request.client.host
    if client_ip in BLACKLIST_IPS:
        return JSONResponse(
            status_code=403,
            content={"detail": "访问被拒绝"}
        )
    return await call_next(request)

1.3.2 敏感词过滤

使用Pydantic模型进行数据验证：

from pydantic import BaseModel
from typing import List


class ContentCheck(BaseModel):
    text: str
    banned_words: List[str] = ["暴力", "敏感词"]


@app.middleware("http")
async def content_filter(request: Request, call_next):
    # 拦截POST请求
    if request.method == "POST":
        body = await request.json()
        checker = ContentCheck(**body)

        # 检测敏感词
        for word in checker.banned_words:
            if word in checker.text:
                return JSONResponse(
                    status_code=400,
                    content={"error": "包含违禁内容"}
                )

    return await call_next(request)

1.4 认证系统集成

1.4.1 JWT认证实现

from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")


def create_jwt_token(data: dict):
    expire = datetime.utcnow() + timedelta(hours=1)
    return jwt.encode(
        {"exp": expire, **data},
        SECRET_KEY,
        algorithm=ALGORITHM
    )


async def validate_token(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError:
        raise HTTPException(
            status_code=401,
            detail="无效的认证凭证"
        )

1.4.2 角色权限验证

from enum import Enum


class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"


def check_permission(required_role: UserRole):
    def validator(user: dict = Depends(validate_token)):
        if user.get("role") != required_role:
            raise HTTPException(403, "权限不足")
        return user

    return validator


@app.get("/admin")
async def admin_dashboard(
        user: dict = Depends(check_permission(UserRole.ADMIN))
):
    return {"message": "管理员面板"}

1.5 课后Quiz

Q1：当收到403状态码时，可能是什么原因导致的？
A：访问被拒绝，常见于IP黑名单拦截、权限不足或资源不可访问的情况

Q2：如何防止中间件影响API性能？
A：通过异步处理、避免阻塞操作、设置合理的缓存机制和精简处理逻辑

Q3：JWT令牌应该存储在客户端的什么位置最安全？
A：推荐存储在HttpOnly的Cookie中，或使用安全的内存存储方式

1.6 常见错误处理

错误1：422 Unprocessable Entity
原因：请求体数据验证失败
解决方法：

检查请求数据是否符合Pydantic模型定义
查看返回的详细错误信息
使用try-except块捕获验证异常

错误2：401 Unauthorized
原因：认证信息缺失或无效
解决方法：

检查Authorization头是否正确携带
验证token是否过期或被篡改
确保认证依赖项正确注入

错误3：500 Internal Server Error
原因：未处理的服务器端异常
解决方法：

查看服务端日志定位错误堆栈
添加全局异常处理器
使用调试模式获取详细信息

# 全局异常处理示例
@app.exception_handler(HTTPException)
async def custom_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": exc.detail}
    )

1.7 环境配置说明

运行要求：

Python 3.7+
FastAPI 0.68+
Uvicorn 0.15+

安装命令：

pip install fastapi==0.68.0 
pip install uvicorn==0.15.0
pip install python-jose==3.3.0 
pip install passlib==1.7.4

最佳实践：

生产环境启用HTTPS
敏感配置使用环境变量
定期更新依赖库版本
实施请求频率限制
启用访问日志审计

通过本文的实践示例和原理分析，开发者可以掌握FastAPI的安全中间件使用技巧，构建具备企业级安全防护能力的API服务。建议结合具体业务需求，选择合适的中间件组合方案。

余下文章内容请点击跳转至个人博客页面或者扫码关注或者微信搜一搜：编程智域前端至全栈交流与成长，阅读完整的文章：

如何在FastAPI中打造坚不可摧的安全防线？