Как отозвать access token?

Отзыв (инвалидация) Access Token

Это важный аспект безопасности. Access Token по природе — сложно отозвать, потому что он самодостаточен (содержит всю информацию). Есть несколько подходов.

Проблема JWT токенов

JWT — это stateless токены. Сервер не хранит их и не может мгновенно отозвать:

Сервер выдал: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4ifQ...

Этот токен будет работать 15 минут, даже если сервер захочет его отозвать.
Почему? Потому что сервер не хранит список активных токенов.

Способ 1: Blacklist (чёрный список) — простой способ

Храним список отозванных токенов:

from fastapi import FastAPI, Depends, HTTPException
from jose import jwt, JWTError
from datetime import datetime, timedelta, timezone
import redis

app = FastAPI()
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 15

def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode["exp"] = expire
    to_encode["jti"] = str(uuid4())  # Уникальный ID для токена
    
    encoded = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        jti = payload.get("jti")
        
        # Проверяем, в чёрном ли списке
        if redis_client.exists(f"blacklist:{jti}"):
            raise HTTPException(status_code=401, detail="Token revoked")
        
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/logout")
async def logout(token: str):
    """Отзыв токена — добавляем в чёрный список"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        jti = payload.get("jti")
        exp = payload.get("exp")
        
        # Вычисляем, на сколько секунд кешировать (до истечения токена)
        ttl = exp - datetime.now(timezone.utc).timestamp()
        
        # Добавляем в чёрный список с TTL
        redis_client.setex(f"blacklist:{jti}", int(ttl), "revoked")
        
        return {"message": "Token revoked successfully"}
    except JWTError:
        raise HTTPException(status_code=400, detail="Invalid token")

@app.get("/api/protected")
async def protected_route(current_user: dict = Depends(verify_token)):
    return {"message": f"Hello {current_user['sub']}"}

Преимущества:

• Простая реализация
• Мгновенный отзыв

Недостатки:

• Redis должен быть всегда доступен
• Требует дополнительное хранилище
• При большом количестве пользователей может быть медленным

Способ 2: Token Versioning — изменяем версию токена

Хранимв БД версию токена, при отзыве инкрементируем:

from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime, timezone

class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True)
    token_version = Column(Integer, default=1)  # Версия текущего токена
    created_at = Column(DateTime, default=datetime.now(timezone.utc))

def create_access_token(user_id: int) -> str:
    user = db.query(User).filter(User.id == user_id).first()
    
    to_encode = {
        "sub": user_id,
        "token_version": user.token_version  # Добавляем версию в токен
    }
    expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode["exp"] = expire
    
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def verify_token(token: str):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        token_version = payload.get("token_version")
        
        # Проверяем версию токена в БД
        user = db.query(User).filter(User.id == user_id).first()
        if user.token_version != token_version:
            # Версия не совпадает — токен отозван
            raise HTTPException(status_code=401, detail="Token revoked")
        
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/logout")
async def logout(current_user: dict = Depends(verify_token)):
    """Отзыв токена — инкрементируем версию"""
    user = db.query(User).filter(User.id == current_user["sub"]).first()
    user.token_version += 1
    db.commit()
    
    return {"message": "Token revoked successfully"}

Преимущества:

• Не требует Redis
• Эффективно
• Можно отозвать все токены пользователя разом (+1 версию)

Недостатки:

• Требует запрос к БД для каждой проверки

Способ 3: Короткий Access Token + Refresh Token

Вместо отзыва длинного токена, используем длинный Refresh Token:

from datetime import timedelta

ACCESS_TOKEN_EXPIRE_MINUTES = 15  # Короткий
REFRESH_TOKEN_EXPIRE_DAYS = 30    # Длинный

@app.post("/login")
async def login(username: str, password: str):
    user = authenticate_user(username, password)
    
    access_token = create_access_token(
        data={"sub": str(user.id)},
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    refresh_token = create_refresh_token(
        data={"sub": str(user.id)},
        expires_delta=timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    )
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "Bearer"
    }

@app.post("/logout")
async def logout(current_user: dict = Depends(verify_token)):
    """Инвалидируем refresh token пользователя"""
    user = db.query(User).filter(User.id == current_user["sub"]).first()
    
    # Сохраняем время логаута
    user.last_logout_time = datetime.now(timezone.utc)
    db.commit()
    
    return {"message": "Logged out successfully"}

@app.post("/refresh")
async def refresh(refresh_token: str):
    try:
        payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        
        user = db.query(User).filter(User.id == user_id).first()
        
        # Проверяем, был ли логаут после выдачи refresh token
        token_issued_at = payload.get("iat")
        if user.last_logout_time:
            if token_issued_at < user.last_logout_time.timestamp():
                raise HTTPException(status_code=401, detail="Token invalidated")
        
        # Выдаём новый access token
        new_access_token = create_access_token({"sub": user_id})
        
        return {"access_token": new_access_token, "token_type": "Bearer"}
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

Как это работает:

1. Access Token коротковечный (15 минут) — мало риска если украдён
2. При логауте инвалидируем Refresh Token
3. После истечения Access Token — нужен Refresh Token
4. Если Refresh Token инвалиден — пользователь должен залогиниться снова

Способ 4: Хибридный подход — Blacklist + Version

Для критичных операций используем оба способа:

def verify_token_strict(token: str):
    """Строгая проверка для чувствительных операций"""
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")
        token_version = payload.get("token_version")
        jti = payload.get("jti")
        
        # 1. Проверяем версию
        user = db.query(User).filter(User.id == user_id).first()
        if user.token_version != token_version:
            raise HTTPException(status_code=401, detail="Token revoked")
        
        # 2. Проверяем blacklist
        if redis_client.exists(f"blacklist:{jti}"):
            raise HTTPException(status_code=401, detail="Token revoked")
        
        return payload
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.post("/logout")
async def logout(current_user: dict = Depends(verify_token_strict)):
    # Инвалидируем refresh token
    user = db.query(User).filter(User.id == current_user["sub"]).first()
    user.token_version += 1
    db.commit()
    
    # Добавляем access token в blacklist
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    jti = payload.get("jti")
    ttl = payload.get("exp") - datetime.now(timezone.utc).timestamp()
    redis_client.setex(f"blacklist:{jti}", int(ttl), "revoked")
    
    return {"message": "Logged out successfully"}

Сравнение способов

Рекомендации

✓ Используй комбинацию:

• Короткий Access Token (15 мин) — минимизирует риск
• Refresh Token — инвалидируется при логауте
• Token Versioning — для быстрой инвалидации всех токенов
• Blacklist — только для критичных операций

✓ На продакшене:

# Лучший вариант
- Access Token: 15 минут
- Refresh Token: 30 дней (в httpOnly cookie)
- Version checking: при каждой проверке токена
- Blacklist: только для логаута (заносим в Redis на 15 минут)

Итог

Access Token нельзя полностью отозвать без хранилища, но можно:

1. Сделать его короткоживущим (15 минут)
2. Использовать Token Versioning
3. Добавить в Blacklist критичные случаи
4. Инвалидировать Refresh Token при логауте