FastAPI随机生成Secret Key后通过中间件自动获取的问题求助

你的核心问题是错误地将中间件类直接作为依赖注入的参数，FastAPI无法直接解析中间件类来获取secret_key。下面是具体的修复方案：

1. 修正中间件的实现 #

原中间件的__call__方法参数和调用逻辑不符合ASGI规范，正确写法如下：

class SecretKeyMiddleware:
    def __init__(self, app, secret_key: str):
        self.app = app
        self.secret_key = secret_key

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            from fastapi import Request
            request = Request(scope, receive)
            request.state.secret_key = self.secret_key
        # 继续处理请求链路
        await self.app(scope, receive, send)

同时修改中间件注册方式：

app.add_middleware(SecretKeyMiddleware, secret_key=secret_key)

2. 创建依赖函数提取secret_key #

写一个专门的依赖函数，从请求状态中读取secret_key，替代直接依赖中间件类：

from fastapi import Request, HTTPException, status

def get_secret_key(request: Request):
    if not hasattr(request.state, "secret_key"):
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="请求状态中未找到secret_key"
        )
    return request.state.secret_key

3. 修改verify_token和check_active的依赖 #

把原来依赖SecretKeyMiddleware的地方，替换成上面的get_secret_key：

def verify_token(token: str, secret_key: str = Depends(get_secret_key)):
    try:
        payload = jwt.decode(token, key=secret_key, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的token"
        )

def check_active(token: str = Depends(oauth2_scheme), secret_key: str = Depends(get_secret_key)):
    try:
        payload = verify_token(token, secret_key)
        exp_timestamp = payload.get('exp')
        if exp_timestamp is None or datetime.now(timezone.utc) > datetime.fromtimestamp(exp_timestamp, timezone.utc):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="token已过期，请重新登录"
            )
        return payload
    except jwt.DecodeError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的token"
        )

4. 其他关键优化 #

• 移除create_access_token里的TokenData.secret_key = secret_key，这行代码无实际作用，直接使用全局生成的secret_key即可
• 删除JWT claims中的password字段：JWT是明文可解码的，存入密码会导致严重安全泄露
• 修正get_current_user里的重复判断逻辑，原代码中elif email:是无效判断，若需支持ID查询，应从token中提取id字段再处理

修改后的完整crud.py核心片段 #

from fastapi import FastAPI, Request, HTTPException, status, Depends
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from passlib.context import CryptContext
from datetime import datetime, timedelta, timezone
from sqlalchemy.orm import Session
import secrets

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

class SecretKeyMiddleware:
    def __init__(self, app, secret_key: str):
        self.app = app
        self.secret_key = secret_key

    async def __call__(self, scope, receive, send):
        if scope["type"] == "http":
            request = Request(scope, receive)
            request.state.secret_key = self.secret_key
        await self.app(scope, receive, send)

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
secret_key = secrets.token_hex(32)
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

app.add_middleware(SecretKeyMiddleware, secret_key=secret_key)

def get_secret_key(request: Request):
    if not hasattr(request.state, "secret_key"):
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail="请求状态中未找到secret_key"
        )
    return request.state.secret_key

def create_access_token(user):
    print("User: ", user.username)
    try:
        claims = {
            "sub": user.username,
            "exp": datetime.now(timezone.utc) + timedelta(days=30),
            "is_active": True
        }
        return jwt.encode(claims, key=secret_key, algorithm="HS256")
    except Exception as ex:
        print(str(ex))
        raise ex

def verify_token(token: str, secret_key: str = Depends(get_secret_key)):
    try:
        payload = jwt.decode(token, key=secret_key, algorithms=["HS256"])
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的token"
        )

def check_active(token: str = Depends(oauth2_scheme), secret_key: str = Depends(get_secret_key)):
    try:
        payload = verify_token(token, secret_key)
        exp_timestamp = payload.get('exp')
        if exp_timestamp is None or datetime.now(timezone.utc) > datetime.fromtimestamp(exp_timestamp, timezone.utc):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="token已过期，请重新登录"
            )
        return payload
    except jwt.DecodeError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的token"
        )

def get_current_user(token: dict = Depends(check_active), db: Session = Depends(get_db)):
    try:
        email = token.get("sub")
        if not email:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="token中未包含有效用户信息",
            )

        user = get_user_by_email(db, email=email)
        # 如需支持ID查询，可添加以下逻辑
        # user_id = token.get("id")
        # elif user_id:
        #     user = get_user_by_id(db, id=user_id)

        if user is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="用户不存在",
            )

        return user
    except jwt.ExpiredSignatureError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="token已过期",
        )
    except jwt.JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的token",
        )

生产环境注意事项 #

• 启动时随机生成secret_key会导致服务重启后所有旧token失效，生产环境建议将secret_key存入环境变量
• 确保JWT的过期时间设置合理，避免过长有效期带来的安全风险

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