How to Implement Refresh Token Rotation in Python

import time
import hashlib
import secrets
from typing import Dict, Optional, Tuple


class MockTokenService:
    """Simulates refresh token rotation for a simple auth system."""

    def __init__(self):
        # Token hash -> (user_id, rotation_count, expires_at)
        self._active_tokens: Dict[str, Tuple[str, int, int]] = {}
        # Token hash -> (user_id, revoked_at) for audit of rotated tokens
        self._rotated_tokens: Dict[str, Tuple[str, int]] = {}
        self._token_ttl_seconds = 3600  # 1 hour

    def _hash_token(self, token: str) -> str:
        return hashlib.sha256(token.encode()).hexdigest()

    def issue_refresh_token(self, user_id: str) -> str:
        """Issue a new refresh token for a user."""
        token = secrets.token_urlsafe(32)
        token_hash = self._hash_token(token)
        self._active_tokens[token_hash] = (user_id, 0, time.time() + self._token_ttl_seconds)
        return token

    def rotate_refresh_token(self, old_token: str) -> Optional[str]:
        """Rotate a refresh token: revoke old, issue new one."""
        old_hash = self._hash_token(old_token)
        token_data = self._active_tokens.pop(old_hash, None)
        if not token_data:
            return None  # Invalid or already revoked

        user_id, rotation_count, expires_at = token_data
        if time.time() > expires_at:
            return None  # Expired

        # Record the rotated token for audit
        self._rotated_tokens[old_hash] = (user_id, int(time.time()))
        return self.issue_refresh_token(user_id)

    def validate_token(self, token: str) -> Optional[Tuple[str, int]]:
        """Return (user_id, rotation_count) if token is valid, else None."""
        token_hash = self._hash_token(token)
        token_data = self._active_tokens.get(token_hash)
        if not token_data or time.time() > token_data[2]:
            return None
        return (token_data[0], token_data[1])


if __name__ == "__main__":
    service = MockTokenService()

    # Issue initial token
    token = service.issue_refresh_token("user-123")
    print(f"Issued token: {token[:20]}...")
    print(f"Valid initially: {service.validate_token(token) is not None}")

    # Rotate it
    rotated = service.rotate_refresh_token(token)
    print(f"Rotated token: {rotated[:20]}...")
    print(f"Old token now invalid: {service.validate_token(token) is None}")
    print(f"New token valid: {service.validate_token(rotated) is not None}")

    # Attempt reuse of old token (should fail)
    reuse_attempt = service.rotate_refresh_token(token)
    print(f"Reuse detected (returns None): {reuse_attempt is None}")