FastAPI用JWT做身份认证遇黑名单失效难题，Token注销即时生效方案比拼，Redis版本控制与双Token刷新策略实战

一、JWT黑名单失效的问题根源

在Web应用开发中，用户登录认证是最基础也最关键的一环。FastAPI作为当下最流行的Python异步Web框架，配合JWT（JSON Web Token）做身份认证是非常经典的组合。但很多开发者在实际项目中都会遇到一个令人头疼的问题：用户注销登录后，之前签发的Token仍然有效，别人拿着这个Token照样能访问受保护的资源。

这个问题本质上是因为JWT的设计哲学是无状态的。一旦Token签发，服务器端就不需要保存任何会话信息，验证Token只需要用签名密钥做一次验签就能判断真伪。这带来了高性能的好处，但也导致了Token一旦签发就无法撤销的麻烦。

想象一下这样的场景：用户在手机上登录了系统，然后去公司用电脑登录。用户想在手机上登出，但登出操作只是让前端把Token删除了，服务器端并不知情。如果有人偷走了这个Token，或者用户在手机上丢失了设备，那么这个Token就永远有效，直到它自然过期。对于涉及用户隐私和资金安全的应用来说，这是一个严重的安全隐患。

1.1 为什么JWT注销是个难题

JWT的核心优势在于它自带签名，服务器不需要查库就能验证身份。这意味着验证过程极快，对服务器压力小。但硬币的另一面是，既然服务器不保存任何Token信息，那它怎么知道某个Token已经被用户主动注销了呢？

传统的会话认证方式下，用户注销就是把服务器上的session记录删掉，下次请求因为没有有效session而被拒绝。但JWT不同，Token是客户端持有的，服务器根本不知道客户端做了什么操作。这就好比银行给你发了一张银行卡，你想注销这张卡，银行却说你卡里没余额就销不了号——这不是银行的锅，是这套体系本身的设计决定了它很难处理主动撤销的需求。

1.2 传统黑名单方案的性能瓶颈

解决JWT注销问题最直观的方案就是维护一个黑名单。用户注销时，把Token的标识（通常是jti字段）存入黑名单。每次请求认证时，除了验签之外，还要额外查一下这个Token是否在黑名单里。如果在，就直接拒绝请求。

这个方案听起来简单直接，但实际落地时有不少坑。如果用户量不大，用数据库存黑名单完全没问题。但当系统用户达到百万级别，每天产生数百万Token时，每次请求都要查数据库就显得非常笨重。数据库查询本身有几十毫秒的开销，在高并发场景下很容易成为瓶颈。

更有甚者，有些开发者直接把黑名单存内存里，比如用Python的set或dict来维护。这种做法在小项目中看似高效，但一旦服务重启，所有黑名单数据全部丢失。更糟糕的是，如果是多实例部署，A实例登出的Token在B实例的内存黑名单里根本没有记录，照样可以通过验证。

二、Redis黑名单方案实战

面对黑名单方案的性能挑战，Redis成了最合适的选择。Redis作为内存数据库，读写速度远超传统数据库，同时支持设置过期时间，正好契合Token黑名单的需求——我们只关心Token在有效期内是否在黑名单里，过期了就算登黑名单也没意义了。

2.1 环境搭建与依赖安装

下面我们通过一个完整的FastAPI项目来演示如何用Redis实现Token黑名单。首先安装所需的依赖包。

pip install fastapi uvicorn redis python-jose[cryptography] pydantic python-multipart passlib[bcrypt]

项目目录结构建议如下，清晰的目录结构有助于后续维护。

fastapi_jwt_blacklist/
├── main.py          # 应用入口
├── auth.py          # JWT认证逻辑
├── redis_client.py  # Redis连接管理
└── requirements.txt # 依赖文件

2.2 黑名单实现代码

技术栈：Python + FastAPI + Redis + python-jose

首先是Redis客户端的单例封装，确保整个应用中只维护一个Redis连接池。

# redis_client.py
import redis

class RedisClient:
    """Redis连接池管理，避免频繁创建连接造成资源浪费"""
    _instance = None
    _pool = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            # 创建连接池，max_connections限制最大连接数
            cls._pool = redis.ConnectionPool(
                host='localhost',       # Redis服务器地址
                port=6379,             # Redis默认端口
                decode_responses=True  # 自动解码为字符串
            )
        return cls._instance

    def get_client(self):
        """返回Redis客户端实例"""
        return redis.Redis(connection_pool=self._pool)

redis_client = RedisClient()

接下来是核心的JWT认证模块，包含Token签发、验证和黑名单检查的完整逻辑。

# auth.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials

# JWT配置
SECRET_KEY = "your-super-secret-key-change-in-production"  # 生产环境必须更换
ALGORITHM = "HS256"  # 签名算法
ACCESS_TOKEN_EXPIRE_HOURS = 1  # 访问Token有效期1小时
REFRESH_TOKEN_EXPIRE_DAYS = 30  # 刷新Token有效期30天

# 密码哈希上下文
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# HTTP Bearer认证方案
bearer = HTTPBearer(auto_error=False)

# 初始化Redis客户端
from redis_client import redis_client
redis_conn = redis_client.get_client()

def verify_password(plain_password: str, hashed_password: str) -> bool:
    """验证密码是否正确"""
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    """对密码进行哈希加密"""
    return pwd_context.hash(password)

def create_access_token(user_id: str, expires_delta: Optional[timedelta] = None):
    """
    签发访问Token
    jti字段作为Token的唯一标识，用于黑名单管理
    """
    expire = datetime.utcnow() + (expires_delta or timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS))
    payload = {
        "sub": user_id,           # 用户ID
        "type": "access",          # Token类型标识
        "exp": expire,             # 过期时间
        "jti": f"access-{user_id}-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"  # 唯一标识
    }
    encoded_jwt = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def create_refresh_token(user_id: str):
    """签发刷新Token，有效期更长"""
    expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    payload = {
        "sub": user_id,
        "type": "refresh",
        "exp": expire,
        "jti": f"refresh-{user_id}-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
    }
    encoded_jwt = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

def add_token_to_blacklist(token_jti: str, expires_delta: timedelta):
    """
    将Token加入黑名单
    使用Redis的SETEX命令，设置值的同时设置过期时间
    过期时间等于Token剩余有效期，避免无效数据堆积
    """
    redis_conn.setex(f"blacklist:{token_jti}", int(expires_delta.total_seconds()), "revoked")

def is_token_blacklisted(token_jti: str) -> bool:
    """
    检查Token是否在黑名单中
    Redis的EXISTS命令时间复杂度O(1)，性能极优
    """
    return redis_conn.exists(f"blacklist:{token_jti}")

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(bearer)) -> str:
    """
    依赖函数：从请求头中提取并验证Token
    这是FastAPI的依赖注入机制，在路由中通过Depends自动调用
    """
    if credentials is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="未提供认证Token"
        )

    token = credentials.credentials
    payload = None

    # 解析Token
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Token无效或已过期：{str(e)}"
        )

    # 检查黑名单
    token_jti = payload.get("jti")
    if token_jti and is_token_blacklisted(token_jti):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="该Token已注销，请重新登录"
        )

    user_id = payload.get("sub")
    if user_id is None:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token格式错误"
        )

    return user_id

然后是应用主文件，包含登录、登出等接口。

# main.py
from fastapi import FastAPI, Depends
from pydantic import BaseModel
from auth import (
    verify_password, get_password_hash,
    create_access_token, create_refresh_token,
    add_token_to_blacklist, get_current_user
)
from redis_client import redis_client
from datetime import timedelta
import jwt
from auth import SECRET_KEY, ALGORITHM

app = FastAPI(title="FastAPI JWT黑名单认证系统")

# 模拟用户数据存储，生产环境应使用数据库
fake_users_db = {
    "admin": {
        "username": "admin",
        "hashed_password": get_password_hash("admin123"),
        "user_id": "user_001"
    },
    "zhangsan": {
        "username": "zhangsan",
        "hashed_password": get_password_hash("password456"),
        "user_id": "user_002"
    }
}

# 请求和响应模型
class LoginRequest(BaseModel):
    username: str
    password: str

class TokenResponse(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class RefreshRequest(BaseModel):
    refresh_token: str

@app.post("/login", response_model=TokenResponse)
def login(request: LoginRequest):
    """
    用户登录接口
    验证密码正确后签发访问Token和刷新Token
    """
    user = fake_users_db.get(request.username)
    if not user:
        raise HTTPException(status_code=400, detail="用户名或密码错误")

    if not verify_password(request.password, user["hashed_password"]):
        raise HTTPException(status_code=400, detail="用户名或密码错误")

    # 签发Token
    access_token = create_access_token(user["user_id"])
    refresh_token = create_refresh_token(user["user_id"])

    return TokenResponse(
        access_token=access_token,
        refresh_token=refresh_token
    )

@app.post("/logout")
def logout(user_id: str = Depends(get_current_user)):
    """
    用户登出接口
    将当前使用的Token加入黑名单，立即失效
    同时将该用户的所有刷新Token也加入黑名单
    """
    # 登出时，将该用户所有刷新Token失效
    # 使用Redis的SCAN命令模糊查询该用户的所有刷新Token
    pattern = f"refresh:{user_id}:*"
    for token_key in redis_conn.scan_iter(pattern):
        redis_conn.delete(token_key)

    return {"message": "登出成功，Token已失效"}

@app.post("/refresh")
def refresh_token(request: RefreshRequest):
    """
    刷新Token接口
    使用有效的刷新Token换取新的访问Token
    旧的访问Token加入黑名单
    """
    try:
        payload = jwt.decode(request.refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
    except Exception:
        raise HTTPException(status_code=401, detail="刷新Token无效")

    if payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="这不是一个刷新Token")

    # 检查刷新Token是否在黑名单中
    token_jti = payload.get("jti")
    if redis_conn.exists(f"blacklist:{token_jti}"):
        raise HTTPException(status_code=401, detail="刷新Token已失效")

    # 签发新的访问Token
    new_access_token = create_access_token(payload["sub"])
    return {"access_token": new_access_token, "token_type": "bearer"}

@app.get("/protected/resource")
def get_protected_resource(user_id: str = Depends(get_current_user)):
    """受保护的接口，需要有效Token才能访问"""
    return {"message": f"欢迎用户 {user_id}，这是受保护的资源"}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

2.3 性能优化策略

Redis黑名单方案虽然有效，但如果每个Token都单独存一个key，用户量大了之后key的数量会爆炸式增长。这里介绍几个实用的优化技巧。

# 优化版本：使用Set集合存储同一用户的所有Token标识
# 每个用户对应一个Set，Set中存储该用户所有有效的Token的jti
import time

def store_user_token(user_id: str, token_jti: str, expires_delta: timedelta):
    """将Token关联到用户，方便批量撤销"""
    redis_conn.sadd(f"user_tokens:{user_id}", token_jti)
    redis_conn.expire(f"user_tokens:{user_id}", int(expires_delta.total_seconds()) + 3600)

def revoke_all_user_tokens(user_id: str):
    """
    一次性撤销某个用户的所有Token
    比如用户修改密码后，应该让所有旧Token立即失效
    """
    token_set_key = f"user_tokens:{user_id}"
    # 获取该用户所有Token的jti
    token_jtis = redis_conn.smembers(token_set_key)
    # 批量加入黑名单
    pipeline = redis_conn.pipeline()  # 使用Pipeline批量操作，减少网络往返
    for jti in token_jtis:
        pipeline.set(f"blacklist:{jti}", "revoked", ex=86400)
    pipeline.delete(token_set_key)  # 删除用户Token集合
    pipeline.execute()

这里用到了Redis的Pipeline特性。正常情况下，每次操作Redis都是一次网络往返，如果有100个Token要操作就是100次网络请求。Pipeline把这些操作打包一次发送，大大提升了效率。

三、Token版本控制方案

黑名单方案虽然成熟，但每次请求都要查Redis，在高并发场景下仍然有一定的性能开销。Token版本控制方案提供了一种不同的思路，它不需要为每个Token建黑名单，而是通过版本号来判断Token是否有效。

3.1 版本控制的原理

版本控制的核心思想很简单：每个用户有一个当前版本号，初始值为1。用户登录时，系统返回Token，同时记录版本号。用户登出时，版本号加1。这样，之前签发的Token因为携带的是旧版本号，在下次验证时就会被拒绝。

这个方案的精妙之处在于，服务器只需要为每个用户维护一个版本号（一个整数），而不是维护成千上万个Token的黑名单。无论用户登录多少次、持有多少个Token，服务器只需要存储一个数字。

验证逻辑是这样的：请求中的Token携带版本号V，服务器存储的版本号也是V。如果两者相等，说明Token有效。用户登出后版本号变成V+1，再验证时发现Token里的V和服务器存的V+1不匹配，就拒绝请求。

3.2 完整实现代码

技术栈：Python + FastAPI + Redis

# token_version_control.py
import jwt
from datetime import datetime, timedelta
from typing import Optional
from fastapi import HTTPException, Depends
from fastapi.security import HTTPBearer
from redis_client import redis_client

SECRET_KEY = "your-super-secret-key-change-in-production"
ALGORITHM = "HS256"
redis_conn = redis_client.get_client()

bearer = HTTPBearer(auto_error=False)

def create_token(user_id: str, version: int, token_type: str = "access"):
    """
    创建携带版本号的Token
    version字段放在payload中，验证时比对
    """
    expire = datetime.utcnow() + timedelta(hours=1 if token_type == "access" else 720)
    payload = {
        "sub": user_id,
        "type": token_type,
        "ver": version,       # 版本号
        "exp": expire
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def increment_user_version(user_id: str):
    """
    用户登出时，版本号加1
    使用Redis的INCR原子操作，保证并发安全
    """
    version = redis_conn.incr(f"token_version:{user_id}")
    return version

def get_user_current_version(user_id: str) -> int:
    """获取用户当前版本号，不存在则返回0"""
    version = redis_conn.get(f"token_version:{user_id}")
    return int(version) if version else 0

def get_current_user_version_control(
    credentials: HTTPBearer = Depends(bearer)
) -> str:
    """
    带版本控制的Token验证依赖函数
    比对Token中的版本号和Redis中存储的版本号
    """
    if credentials is None:
        raise HTTPException(status_code=401, detail="未提供认证信息")

    try:
        payload = jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
    except Exception:
        raise HTTPException(status_code=401, detail="Token无效或已过期")

    user_id = payload.get("sub")
    token_version = payload.get("ver", 0)
    current_version = get_user_current_version(user_id)

    # 核心验证逻辑：Token中的版本必须等于或大于服务器当前版本
    # 但实际应该是完全相等，因为登出后版本会递增
    if token_version < current_version:
        raise HTTPException(
            status_code=401,
            detail="Token已失效，请重新登录"
        )

    return user_id

# 示例接口
from fastapi import FastAPI
app = FastAPI()

@app.post("/logout_version")
def logout_version_control(user_id: str = Depends(get_current_user_version_control)):
    """
    使用版本控制进行登出
    只需要将版本号加1，所有旧Token立即失效
    """
    new_version = increment_user_version(user_id)
    return {"message": "登出成功", "new_version": new_version}

版本控制方案还有一个重要场景：密码修改。当用户修改密码后，应该让所有之前的Token立即失效，防止他人使用旧凭证登录。

@app.post("/change-password")
def change_password(user_id: str = Depends(get_current_user_version_control)):
    """
    修改密码后，强制所有Token失效
    通过递增版本号实现
    """
    # 此处省略密码修改的数据库操作
    # 核心是递增版本号
    new_version = increment_user_version(user_id)
    return {"message": "密码修改成功，所有会话已刷新", "new_version": new_version}

四、双Token刷新策略实战

在实际项目中，我们通常不会只依赖单一Token来兼顾安全和用户体验。双Token策略是目前业界最主流的做法，它结合了一个短时效的访问Token和一个长时效的刷新Token，既保证了安全性，又避免了用户频繁登录的困扰。

4.1 双Token的工作原理

双Token体系中，访问Token（Access Token）的有效期很短，通常只有15分钟到1小时。它用于日常的业务请求，被窃取的风险窗口较小。刷新Token（Refresh Token）的有效期较长，可以是一周到一个月，但只能用于向服务器换取新的访问Token，不能直接访问业务接口。

用户的操作流程是这样的：登录后拿到两个Token。访问业务接口时只带访问Token。当访问Token过期后，前端用刷新Token去调用刷新接口，换取新的访问Token，全程用户无感知。当用户主动登出时，刷新Token也一起被撤销，这样就彻底切断了会话。

这个方案的巧妙之处在于，即使攻击者窃取了访问Token，也只能用很短的时间窗口来发起攻击。而如果攻击者连刷新Token都拿到了，那他其实就相当于拿到了用户的完整凭证，这种情况下系统应该有其他层面的防护机制。

4.2 完整实现代码

技术栈：Python + FastAPI + Redis + python-jose

下面是完整的双Token刷新系统实现。

# dual_token.py
import jwt
from datetime import datetime, timedelta
from typing import Optional
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from jose import JWTError
from passlib.context import CryptContext
from redis_client import redis_client

app = FastAPI(title="双Token刷新系统")

# ============ 配置常量 ============
SECRET_KEY = "change-me-in-production-use-longer-random-string"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15  # 访问Token仅15分钟
REFRESH_TOKEN_EXPIRE_DAYS = 7     # 刷新Token有效期7天
REUSE_DETECTION_EXPIRE_HOURS = 1  # 刷新Token复用检测的冷却时间

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
redis_conn = redis_client.get_client()
bearer = HTTPBearer(auto_error=False)

# ============ 用户数据模型 ============
class User(BaseModel):
    user_id: str
    username: str
    hashed_password: str

class LoginRequest(BaseModel):
    username: str
    password: str

class TokenPair(BaseModel):
    access_token: str
    refresh_token: str
    access_token_expires: datetime
    refresh_token_expires: datetime
    token_type: str = "bearer"

class RefreshRequest(BaseModel):
    refresh_token: str

# ============ 模拟用户数据库 ============
FAKE_USERS = {
    "admin": User(
        user_id="u001",
        username="admin",
        hashed_password=pwd_context.hash("secure_password_123")
    ),
    "user1": User(
        user_id="u002",
        username="user1",
        hashed_password=pwd_context.hash("my_password_456")
    )
}

# ============ Token签发函数 ============
def generate_access_token(user_id: str) -> str:
    """
    签发访问Token
    包含用户ID、Token类型、过期时间和唯一标识
    """
    now = datetime.utcnow()
    expire = now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    jti = f"at-{user_id}-{now.strftime('%Y%m%d%H%M%S%f')}"
    payload = {
        "sub": user_id,
        "type": "access",
        "exp": expire,
        "iat": now,
        "jti": jti
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def generate_refresh_token(user_id: str) -> str:
    """
    签发刷新Token
    有效期更长，但只用于换取访问Token
    """
    now = datetime.utcnow()
    expire = now + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    jti = f"rt-{user_id}-{now.strftime('%Y%m%d%H%M%S%f')}"
    payload = {
        "sub": user_id,
        "type": "refresh",
        "exp": expire,
        "iat": now,
        "jti": jti
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)

def store_refresh_token(user_id: str, token_jti: str):
    """
    在Redis中记录当前有效的刷新Token
    key格式：refresh_token:{user_id}
    同一个用户同一时间只允许一个有效的刷新Token（单端登录）
    如需支持多端登录，可改用Set存储多个jti
    """
    redis_conn.set(f"refresh_token:{user_id}", token_jti)
    redis_conn.expire(f"refresh_token:{user_id}", REFRESH_TOKEN_EXPIRE_DAYS * 86400)

def register_refresh_token_used(token_jti: str):
    """
    标记刷新Token已被使用过
    用于检测Token复用攻击
    """
    redis_conn.set(
        f"refresh_used:{token_jti}",
        "used",
        ex=REUSE_DETECTION_EXPIRE_HOURS * 3600
    )

def is_refresh_token_reused(token_jti: str) -> bool:
    """检查刷新Token是否被重复使用"""
    return redis_conn.exists(f"refresh_used:{token_jti}")

# ============ Token验证函数 ============
def decode_token(token: str) -> dict:
    """解析Token，不校验黑名单（黑名单检查由调用方处理）"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return payload
    except JWTError as e:
        raise HTTPException(status_code=401, detail=f"Token验证失败：{str(e)}")

def verify_access_token(credentials: HTTPAuthorizationCredentials) -> str:
    """
    验证访问Token
    在双Token体系中，访问Token不需要黑名单检查
    因为它的有效期极短，即使泄露，攻击窗口也很有限
    """
    if credentials is None:
        raise HTTPException(status_code=401, detail="请先登录")

    payload = decode_token(credentials.credentials)

    if payload.get("type") != "access":
        raise HTTPException(status_code=401, detail="这不是访问Token")

    user_id = payload.get("sub")
    if not user_id:
        raise HTTPException(status_code=401, detail="Token格式错误")

    return user_id

# ============ 路由接口 ============
@app.post("/login", response_model=TokenPair)
def login(request: LoginRequest):
    """
    用户登录
    验证成功后签发一对Token
    """
    user = FAKE_USERS.get(request.username)
    if not user:
        raise HTTPException(status_code=400, detail="用户名或密码错误")

    if not pwd_context.verify(request.password, user.hashed_password):
        raise HTTPException(status_code=400, detail="用户名或密码错误")

    access_token = generate_access_token(user.user_id)
    refresh_token = generate_refresh_token(user.user_id)

    # 从刷新Token中提取jti并存入Redis
    refresh_payload = decode_token(refresh_token)
    store_refresh_token(user.user_id, refresh_payload["jti"])

    now = datetime.utcnow()
    return TokenPair(
        access_token=access_token,
        refresh_token=refresh_token,
        access_token_expires=now + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
        refresh_token_expires=now + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    )

@app.post("/refresh")
def refresh_access_token(request: RefreshRequest):
    """
    刷新访问Token
    这是双Token体系的核心接口
    刷新Token只能用来换访问Token，不能访问业务资源
    """
    # 1. 解析刷新Token
    payload = decode_token(request.refresh_token)

    if payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="这不是刷新Token")

    user_id = payload.get("sub")
    token_jti = payload.get("jti")

    # 2. 检测刷新Token是否被重复使用（检测Token窃取）
    if is_refresh_token_reused(token_jti):
        # 发现Token被复用，说明可能有攻击者在尝试使用偷来的Token
        # 立即撤销该用户所有会话
        revoke_user_all_tokens(user_id)
        raise HTTPException(
            status_code=401,
            detail="检测到Token异常复用，所有会话已强制登出，请重新登录"
        )

    # 3. 验证刷新Token是否为该用户当前有效的刷新Token
    stored_jti = redis_conn.get(f"refresh_token:{user_id}")
    if stored_jti != token_jti:
        raise HTTPException(status_code=401, detail="刷新Token已失效")

    # 4. 标记旧刷新Token已使用
    register_refresh_token_used(token_jti)

    # 5. 签发全新的Token对
    new_access_token = generate_access_token(user_id)
    new_refresh_token = generate_refresh_token(user_id)
    new_refresh_jti = decode_token(new_refresh_token)["jti"]

    # 6. 更新Redis中的刷新Token记录
    store_refresh_token(user_id, new_refresh_jti)

    return {
        "access_token": new_access_token,
        "refresh_token": new_refresh_token,
        "token_type": "bearer"
    }

@app.post("/logout")
def logout(user_id: str = Depends(verify_access_token)):
    """
    用户登出
    删除Redis中该用户的刷新Token记录
    下次刷新时因为找不到有效刷新Token而失败
    """
    redis_conn.delete(f"refresh_token:{user_id}")
    return {"message": "登出成功，请保管好您的设备"}

def revoke_user_all_tokens(user_id: str):
    """
    管理员接口或安全事件时调用
    强制撤销某个用户的所有Token
    """
    redis_conn.delete(f"refresh_token:{user_id}")

@app.get("/protected/data")
def get_protected_data(user_id: str = Depends(verify_access_token)):
    """受保护的业务接口，使用访问Token验证"""
    return {
        "user_id": user_id,
        "data": "这是需要登录才能看到的敏感数据",
        "security_note": "访问Token有效期仅15分钟，过期后需自动刷新"
    }

4.3 安全性设计

双Token体系还有一个重要的安全考虑：刷新Token的存储方式。前端的处理非常关键，如果前端把刷新Token存在localStorage里，那么任何能访问页面的JavaScript代码（包括被注入的XSS攻击代码）都能读取到这个Token，后果不堪设想。

正确的做法是将刷新Token存在HttpOnly Cookie中，这样JavaScript无法读取，能有效防御XSS攻击窃取Token。

# 使用HttpOnly Cookie存储刷新Token的示例
from fastapi import FastAPI, Response, Request
from starlette.responses import JSONResponse

@app.post("/login")
def login_with_cookie(request: LoginRequest, response: Response):
    """
    登录时将刷新Token写入HttpOnly Cookie
    前端只能通过Cookie自动携带，JS代码无法读取
    """
    # ... 验证逻辑同上 ...

    access_token = generate_access_token(user.user_id)
    refresh_token = generate_refresh_token(user.user_id)

    # 设置HttpOnly Cookie
    # httponly=True 禁止JS读取
    # secure=True 仅HTTPS传输
    # samesite="strict" 防止CSRF攻击
    response.set_cookie(
        key="refresh_token",
        value=refresh_token,
        httponly=True,
        secure=True,
        samesite="strict",
        max_age=7 * 86400  # 7天有效期
    )

    return {
        "access_token": access_token,  # 访问Token返回给前端JS使用
        "token_type": "bearer"
    }

@app.post("/refresh_from_cookie")
def refresh_from_cookie(request: Request):
    """
    从Cookie中读取刷新Token来刷新访问Token
    Cookie由浏览器自动携带，无需前端手动处理
    """
    refresh_token = request.cookies.get("refresh_token")

    if not refresh_token:
        raise HTTPException(status_code=401, detail="未找到有效的刷新Token")

    # ... 后续验证和刷新逻辑同上 ...
    return {"access_token": new_access_token}

@app.post("/logout_with_cookie")
def logout_with_cookie(response: Response):
    """登出时清除刷新Token Cookie"""
    response.delete_cookie(key="refresh_token")
    return {"message": "登出成功"}

关于CSRF攻击的防护，当刷新Token存在Cookie中时，需要特别注意跨站请求伪造的风险。通过设置SameSite属性可以有效防御。SameSite=Strict表示Cookie不会在任何跨站请求中发送，最安全但可能影响某些第三方嵌入场景。SameSite=Lax是大多数场景下的合理折中，同站跳转仍然会携带Cookie，但跨站POST请求不会。

另外，更严格的方案是在请求头中额外加一个由前端生成的CSRF Token，服务器端比对，双重保险。

五、三种方案对比分析

5.1 应用场景分析

Redis黑名单方案适合的场景是对Token注销有即时性要求、且系统已有Redis基础设施的项目。它实现简单，逻辑清晰，对开发者来说最容易理解。如果你的项目已经有Redis用于缓存，那么引入黑名单几乎没有额外的基础设施成本。

Token版本控制方案适合的场景是单端登录或者需要严格会话管理的系统。比如银行类应用，同一用户同时只能在一个设备上登录。版本控制的实现非常轻量，每个用户只需要在Redis中存一个整数。但它有一个天然的局限：不支持同一用户多设备同时在线（或者说支持起来比较复杂）。

双Token刷新策略适合的场景是几乎所有中大型Web应用。它是目前业界的事实标准，从JWT本身的社区实践到各大厂商的身份认证服务，都采用类似的思路。它平衡了安全性和用户体验，短时效的访问Token降低了泄露风险，长时效的刷新Token让用户不需要频繁重新登录。

5.2 技术优缺点对比

Redis黑名单方案的最大优点是实现简单，逻辑直观。每次请求加一次Redis查询，在现代网络环境下开销可以忽略不计。缺点是Redis中存储的key数量会随着用户登录次数线性增长，虽然可以通过设置过期时间自动清理，但管理起来还是有一定复杂性。此外，它需要每次请求都做额外的网络调用，在极高并发场景下有一定的性能代价。

Token版本控制方案的最大优点是极其轻量，每个用户只需一个整数存储。登出操作是原子的，不存在并发问题。缺点是原生不支持多设备登录，如果需要支持，需要额外设计版本号的管理机制。另外，版本号只支持单调递增，无法实现部分Token撤销（比如只撤销某个特定设备的Token而不影响其他设备）。

双Token方案的最大优点是安全和体验兼顾，是目前最成熟的方案。缺点是实现相对复杂，需要考虑的问题较多，比如刷新Token的存储安全、CSRF防护、Token复用检测等。此外，刷新Token的生命周期管理也需要精心设计。

5.3 注意事项

无论选择哪种方案，有几个通用的注意事项必须牢记。

第一，永远不要把密钥硬编码在代码中。上面的示例中的SECRET_KEY只是为了演示，生产环境必须使用环境变量或者密钥管理服务。密钥一旦泄露，所有签发的Token都可以被伪造。

第二，JWT的过期时间必须设置合理。访问Token建议15分钟到1小时，刷新Token建议7天到30天。过长的有效期会增加安全风险，过短则影响用户体验。

第三，Token存储在前端也要谨慎。访问Token可以存内存中（页面刷新丢失也没关系，重新请求即可），刷新Token必须用HttpOnly Cookie存储。

第四，Redis的持久化配置要合理。虽然黑名单数据可以接受一定程度的丢失（Token最终会自然过期），但版本控制数据丢失会导致所有旧Token失效，用户需要重新登录，体验不好。

第五，所有敏感操作（如修改密码、重置密码、查看敏感信息）应该强制要求重新登录，而不是依赖现有的Token。这是纵深防御的重要一环。

六、文章总结

在FastAPI项目中使用JWT做身份认证时，Token注销的即时生效是一个绕不开的问题。我们讨论了三种主流方案，各有千秋。Redis黑名单方案直观易用，适合大多数项目快速落地。Token版本控制方案轻量高效，适合对会话管理要求严格的场景。双Token刷新策略是目前业界的事实标准，综合了安全性和用户体验的最佳平衡。

实际项目中，很多团队会组合使用这些方案。比如用双Token策略做主认证体系，同时用版本控制做辅助的快速登出机制。重要的是理解每种方案的原理和适用场景，根据自己的业务需求做出合适的选择。安全是一个持续的过程，没有一劳永逸的解决方案，我们需要不断地审视和改进自己的认证体系。