JWT Refresh Token Rotation in FastAPI — The Right Way

# ⚠️ This is incomplete — don't ship this
@app.post("/login")
def login(credentials: LoginSchema, db: Session = Depends(get_db)):
    user = authenticate_user(db, credentials.username, credentials.password)
    access_token = create_access_token({"sub": user.id})
    refresh_token = create_refresh_token({"sub": user.id})
    return {"access_token": access_token, "refresh_token": refresh_token}

# models.py
class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
    token_hash: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
    family: Mapped[uuid.UUID] = mapped_column(nullable=False)  # rotation chain
    used: Mapped[bool] = mapped_column(default=False)
    expires_at: Mapped[datetime] = mapped_column(nullable=False)
    created_at: Mapped[datetime] = mapped_column(default=datetime.utcnow)

# auth/routes.py
@router.post("/refresh")
async def refresh_tokens(
    request: Request,
    payload: RefreshRequest,
    db: AsyncSession = Depends(get_async_db),
):
    token_hash = hash_token(payload.refresh_token)

    # Load the token record
    result = await db.execute(
        select(RefreshToken)
        .where(RefreshToken.token_hash == token_hash)
        .with_for_update()  # pessimistic lock — prevents race conditions
    )
    record = result.scalar_one_or_none()

    if not record:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    # Reuse detection — token was already rotated
    if record.used:
        # Revoke entire rotation family
        await db.execute(
            update(RefreshToken)
            .where(RefreshToken.family == record.family)
            .values(used=True)
        )
        await db.commit()
        raise HTTPException(
            status_code=401,
            detail="Refresh token reuse detected — all sessions revoked"
        )

    if record.expires_at < datetime.utcnow():
        raise HTTPException(status_code=401, detail="Refresh token expired")

    # Mark old token as used (rotated)
    record.used = True

    # Issue new token in the same family
    new_raw = secrets.token_urlsafe(48)
    new_record = RefreshToken(
        user_id=record.user_id,
        token_hash=hash_token(new_raw),
        family=record.family,
        expires_at=datetime.utcnow() + timedelta(days=30),
    )
    db.add(new_record)
    await db.commit()

    # Issue new access token
    access_token = create_access_token({"sub": str(record.user_id)})

    return {
        "access_token": access_token,
        "refresh_token": new_raw,
    }

import hashlib
import secrets

def hash_token(raw: str) -> str:
    """SHA-256 hash — fast, one-way, collision-resistant."""
    return hashlib.sha256(raw.encode()).hexdigest()

def generate_refresh_token() -> tuple[str, str]:
    """Returns (raw_token, hashed_token)."""
    raw = secrets.token_urlsafe(48)
    return raw, hash_token(raw)

@router.post("/login")
async def login(response: Response, ...):
    ...
    raw_refresh, _ = generate_refresh_token()
    response.set_cookie(
        key="refresh_token",
        value=raw_refresh,
        httponly=True,
        secure=True,
        samesite="lax",
        max_age=60 * 60 * 24 * 30,  # 30 days
        path="/auth/refresh",         # scoped — not sent on every request
    )
    return {"access_token": access_token}

# tests/test_auth.py
async def test_refresh_token_rotation(client, user_factory):
    user = await user_factory()
    login_resp = await client.post("/auth/login", json={"username": user.email, "password": "testpass"})
    original_refresh = login_resp.cookies["refresh_token"]

    # First refresh — should succeed
    r1 = await client.post("/auth/refresh", cookies={"refresh_token": original_refresh})
    assert r1.status_code == 200
    new_refresh = r1.cookies["refresh_token"]
    assert new_refresh != original_refresh

    # Reuse old token — should trigger reuse detection
    r2 = await client.post("/auth/refresh", cookies={"refresh_token": original_refresh})
    assert r2.status_code == 401
    assert "reuse detected" in r2.json()["detail"]

    # New token should also be revoked after reuse detection
    r3 = await client.post("/auth/refresh", cookies={"refresh_token": new_refresh})
    assert r3.status_code == 401