如何在Python FastAPI中结合Oauth2用JWT token实现用户登录注销验证

JWT本身是无状态的，服务端不会存储已签发的有效token，所以无法直接清除已发布的JWT。要实现注销后token失效，行业内常用两种方案：

• 方案1：短有效期access token + 长有效期refresh token，注销时仅将refresh token加入黑名单，access token到期后用户无法刷新获取新token，自然失效
• 方案2：将所有注销的token直接加入黑名单，每次鉴权时先校验token是否在黑名单中，适合对安全性要求高的场景

下文用方案2做完整实现，逻辑更直观。 #

前置依赖安装 #

执行以下命令安装所需工具包：

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

完整代码实现 #

1. 基础配置与依赖导入 #

from datetime import datetime, timedelta
from typing import Optional
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

# 配置项，生产环境密钥请存到环境变量，禁止硬编码
SECRET_KEY = "your-custom-secret-key-keep-it-safe"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# Token黑名单，演示用内存存储，生产环境建议替换为Redis
token_blacklist = set()

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

# OAuth2 认证配置，token获取路径为/login
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

app = FastAPI()

2. 数据模型与模拟用户库 #

# 模拟用户数据库，存储加密后的密码
fake_users_db = {
    "testuser": {
        "username": "testuser",
        "full_name": "Test User",
        "email": "test@example.com",
        "hashed_password": pwd_context.hash("testpass123"),
        "disabled": False,
    }
}

# Pydantic 数据校验模型
class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: Optional[str] = None

class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

class UserInDB(User):
    hashed_password: str

3. 工具函数实现 #

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_user(db, username: str):
    if username in db:
        return UserInDB(**db[username])

def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta if expires_delta else timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    # 优先校验token是否已被注销
    if token in token_blacklist:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token has been revoked",
            headers={"WWW-Authenticate": "Bearer"},
        )
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_data = TokenData(username=username)
    except JWTError:
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: User = Depends(get_current_user)):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user

4. 接口实现 #

# 登录接口，返回有效access token
@app.post("/login", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 受保护接口，需要携带有效token才能访问
@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user

# 注销接口，将当前请求的token加入黑名单
@app.post("/logout")
async def logout(token: str = Depends(oauth2_scheme)):
    token_blacklist.add(token)
    return {"msg": "Successfully logged out"}

测试流程 #

1. 启动服务后调用/login接口，传入用户名testuser、密码testpass123，获取access_token
2. 调用/users/me接口，请求头携带Authorization: Bearer <获取到的token>，可正常返回用户信息
3. 调用/logout接口，同样携带上述token，返回注销成功
4. 再次调用/users/me接口，会返回401未授权，说明token已失效

注意事项 #

演示用的内存黑名单在服务重启后数据会丢失，生产环境建议使用带过期策略的Redis存储黑名单，过期时间和JWT的有效期保持一致即可，避免存储空间无限增长。如果用户量级很大，建议切换为短access token+refresh token的方案，仅拉黑refresh token可以大幅减少存储成本。

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