4、自定义鉴权中间件：Flask/FastAPI中间件原理、编写自定义鉴权函数、注册中间件到vLLM服务、请求拦截与验证流程

4.1 中间件到底是个啥？

说实话，很多刚接触vLLM的同学，一听到「中间件」三个字就头大。我当年也是。

其实没那么玄乎。中间件就是一道关卡。每个请求进来，都得先过我这关。过了，放行；不过，直接打回去。

在Flask里，中间件叫before_request钩子。在FastAPI里，叫Middleware。本质都一样——在路由处理函数执行前，插一脚。

我个人习惯把中间件想象成安检员。你进地铁站，安检员先扫你包，没问题才让你刷卡进站。中间件就是那个安检员，只不过它检查的是HTTP请求里的Token。

4.2 Flask中间件原理

Flask的中间件机制，说白了就是装饰器模式。你定义一个函数，用@app.before_request装饰一下，这个函数就会在每个请求处理前自动执行。

为什么会这样？因为Flask内部维护了一个请求处理管道。每个请求进来，先走一遍所有before_request函数，再走路由匹配，最后执行视图函数。

我记得有一次排查线上问题，发现某个接口响应特别慢。查了半天，原来是中间件里做了个数据库查询，每次请求都查一次。嗯，这里要注意——中间件里别做耗时操作，否则所有接口都跟着遭殃。

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.before_request
def check_auth():
    # 每个请求进来，先执行这里
    token = request.headers.get('Authorization')
    if not token:
        return jsonify({'error': 'Missing token'}), 401
    # 继续往下走

4.3 FastAPI中间件原理

FastAPI的中间件稍微有点不同。它用的是ASGI规范，中间件是一个可调用对象，接收request和call_next两个参数。

你想想看，call_next是什么？它就是「下一个中间件」或者「最终的路由处理函数」。你可以在调用call_next之前做拦截，也可以在它之后做后处理。

我在项目中遇到过一个问题：FastAPI中间件里如果抛异常，整个请求链会断掉。所以一定要用try-except包住call_next，否则客户端收到的错误信息会很奇怪。

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

app = FastAPI()

class AuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        token = request.headers.get('Authorization')
        if not token:
            return JSONResponse(
            status_code=401,
            content={'error': 'Missing token'}
        )
        # 验证通过，继续
        response = await call_next(request)
        return response

app.add_middleware(AuthMiddleware)

4.4 编写自定义鉴权函数

鉴权函数的核心就三件事：取Token、验Token、返回结果。

我曾经踩过一个坑——直接拿请求头里的Token去查数据库。每次请求都查一次，QPS一上来数据库直接被打满。后来改成用JWT，本地验证，不查库，性能提升了几十倍。

下面是我常用的鉴权函数模板：

import jwt
from datetime import datetime, timedelta

SECRET_KEY = "your-secret-key-here"

def verify_token(token: str) -> dict:
    """
    验证JWT Token
    返回payload字典，验证失败返回None
    """
    try:
        payload = jwt.decode(
            token,
            SECRET_KEY,
            algorithms=['HS256']
        )
        # 检查是否过期
        exp = payload.get('exp')
        if exp and datetime.utcnow() > datetime.fromtimestamp(exp):
            return None
        return payload
    except jwt.ExpiredSignatureError:
        # Token过期
        return None
    except jwt.InvalidTokenError:
        # Token无效
        return None

4.5 注册中间件到vLLM服务

vLLM本身用的是FastAPI。所以注册中间件的方式和FastAPI一模一样。

但有个细节要注意——vLLM启动时，会自己创建FastAPI实例。你不能直接改它的代码。那怎么办？

我建议的做法是：写一个启动脚本，在vLLM的app对象创建之后，手动添加中间件。

# custom_auth.py
from vllm.entrypoints.openai.api_server import app
from starlette.middleware.base import BaseHTTPMiddleware

class CustomAuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # 你的鉴权逻辑
        token = request.headers.get('Authorization')
        if not token:
            from starlette.responses import JSONResponse
            return JSONResponse(
            status_code=401,
            content={'error': 'Unauthorized'}
        )
        # 验证通过
        response = await call_next(request)
        return response

# 注册中间件
app.add_middleware(CustomAuthMiddleware)

# 启动vLLM时，导入这个模块即可
# python -c "import custom_auth; from vllm.entrypoints.openai.api_server import run_server; run_server()"

4.6 请求拦截与验证流程

整个流程其实就三步走：

1. 拦截：中间件捕获所有进入的HTTP请求
2. 验证：从请求头提取Token，调用鉴权函数
3. 放行/拒绝：验证通过则调用call_next，否则直接返回401

下面这张图，是我自己画的请求拦截流程图，你看一眼就明白了：

你看，流程其实很清晰。客户端发请求，中间件拦截，验Token，有效就放行到vLLM，无效直接打回。

4.7 避坑指南

我做了这么多年的鉴权中间件，踩过的坑能写一本书。这里挑几个最常见的：

• Token放在哪里？——我建议放请求头Authorization: Bearer <token>。别放URL参数，日志里会泄露。
• 中间件异常处理——一定要用try-except包住核心逻辑。否则中间件崩了，整个服务都挂。
• 性能问题——鉴权函数里别做IO操作。查数据库、调外部API，这些都会拖慢每个请求。
• 白名单路径——有些路径不需要鉴权，比如健康检查/health。记得在中间件里加白名单判断。

嗯，关于自定义鉴权中间件，我就讲这么多。下一节我们会聊更细粒度的权限控制——基于角色的访问控制（RBAC）。到时候你会看到，中间件只是第一步，真正的安全体系要层层递进。

---

公众号:蓝海资料掘金营，微信deep3321