中文
devexamples

FastAPI JWT Refresh Token Rotation

Issue short-lived access tokens plus rotating refresh tokens in FastAPI, detecting reuse of revoked tokens on every exchange.

PythonFastapiAuth Jwt

Code

snippet
from datetime import datetime, timedelta, timezone

import jwt
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer
from pydantic import BaseModel

SECRET = 'rotate-me-in-production'
REFRESH_TTL = timedelta(days=14)
ACCESS_TTL = timedelta(minutes=15)

app = FastAPI()
oauth2 = OAuth2PasswordBearer(tokenUrl='/token')
revoked: set[str] = set()  # 生产环境应持久化并设置过期


class TokenPair(BaseModel):
    access: str
    refresh: str


def issue_tokens(user_id: str) -> TokenPair:
    now = datetime.now(timezone.utc)
    return TokenPair(
        access=jwt.encode(
            {'sub': user_id, 'exp': now + ACCESS_TTL, 'typ': 'access'},
            SECRET, algorithm='HS256'),
        refresh=jwt.encode(
            {'sub': user_id, 'exp': now + REFRESH_TTL, 'typ': 'refresh',
             'jti': f'{user_id}-{int(now.timestamp())}'},
            SECRET, algorithm='HS256'),
    )


@app.post('/token/refresh', response_model=TokenPair)
def refresh(pair: TokenPair):
    try:
        claims = jwt.decode(pair.refresh, SECRET, algorithms=['HS256'])
    except jwt.InvalidTokenError:
        raise HTTPException(401, 'invalid refresh token')

    if pair.refresh in revoked:
        raise HTTPException(401, 'token reuse detected')

    revoked.add(pair.refresh)
    return issue_tokens(claims['sub'])


@app.get('/me')
def me(token: str = Depends(oauth2)):
    claims = jwt.decode(token, SECRET, algorithms=['HS256'])
    if claims.get('typ') != 'access':
        raise HTTPException(401, 'wrong token type')
    return {'user': claims['sub']}

Related Examples