Token Refresh Patterns — Complete Implementation Guide

const express = require("express");
const jwt = require("jsonwebtoken");
const crypto = require("crypto");

const app = express();
app.use(express.json());

const ACCESS_SECRET = crypto.randomBytes(64).toString("hex");
const REFRESH_SECRET = crypto.randomBytes(64).toString("hex");
const tokenFamily = new Map();

function generateTokens(user) {
  const accessToken = jwt.sign(
    { sub: user.id, role: user.role, type: "access" },
    ACCESS_SECRET,
    { expiresIn: "15m" }
  );

  const refreshId = crypto.randomBytes(16).toString("hex");
  const refreshToken = jwt.sign(
    { sub: user.id, type: "refresh", refreshId, family: user.id },
    REFRESH_SECRET,
    { expiresIn: "7d" }
  );

  tokenFamily.set(refreshId, {
    refreshId,
    userId: user.id,
    createdAt: Date.now(),
    isUsed: false,
  });

  return { accessToken, refreshToken, expiresIn: 900 };
}

app.post("/api/auth/login", (req, res) => {
  const { email, password } = req.body;
  if (email !== "alice@example.com" || password !== "correct-password") {
    return res.status(401).json({ error: "Invalid credentials" });
  }
  const tokens = generateTokens({ id: 1, role: "user" });
  res.json(tokens);
});

app.post("/api/auth/refresh", (req, res) => {
  const { refreshToken } = req.body;
  if (!refreshToken) {
    return res.status(401).json({ error: "Refresh token required" });
  }

  try {
    const decoded = jwt.verify(refreshToken, REFRESH_SECRET);
    const stored = tokenFamily.get(decoded.refreshId);

    if (!stored) {
      return res.status(401).json({ error: "Refresh token not found" });
    }

    if (stored.isUsed) {
      console.log(`[TOKEN THEFT] Refresh token ${decoded.refreshId} already used!`);
      const familyMembers = [...tokenFamily.values()]
        .filter(t => t.userId === decoded.sub && !t.isUsed);
      familyMembers.forEach(t => {
        t.isUsed = true;
        console.log(`[TOKEN THEFT] Invalidating sibling token: ${t.refreshId}`);
      });
      return res.status(401).json({ error: "Token reuse detected" });
    }

    stored.isUsed = true;
    const tokens = generateTokens({ id: decoded.sub, role: decoded.role || "user" });
    console.log(`[Refresh] Tokens rotated for user ${decoded.sub}`);
    res.json(tokens);

  } catch (err) {
    res.status(401).json({ error: "Invalid refresh token" });
  }
});

app.listen(3000);

// API client with automatic token refresh
class ApiClient {
  constructor(baseURL, authService) {
    this.baseURL = baseURL;
    this.authService = authService;
    this.accessToken = null;
    this.refreshPromise = null;
  }

  async request(endpoint, options = {}) {
    if (!this.accessToken) {
      throw new Error("Not authenticated");
    }

    const response = await fetch(`${this.baseURL}${endpoint}`, {
      ...options,
      headers: {
        ...options.headers,
        Authorization: `Bearer ${this.accessToken}`,
      },
    });

    if (response.status === 401) {
      const newToken = await this.refreshAccessToken();
      this.accessToken = newToken;
      return this.request(endpoint, options);
    }

    return response.json();
  }

  async refreshAccessToken() {
    if (this.refreshPromise) {
      return this.refreshPromise;
    }

    this.refreshPromise = this.authService.refresh()
      .then((token) => {
        this.refreshPromise = null;
        return token;
      })
      .catch((err) => {
        this.refreshPromise = null;
        this.accessToken = null;
        throw err;
      });

    return this.refreshPromise;
  }
}

import time
import jwt
import secrets
from datetime import datetime, timedelta

class TokenSimulator:
    def __init__(self):
        self.secret = secrets.token_hex(32)
        self.refresh_count = 0
        self.events = []

    def create_tokens(self):
        access = jwt.encode({
            "sub": "user_1", "type": "access",
            "exp": int(time.time()) + 15,
        }, self.secret, algorithm="HS256")
        refresh = jwt.encode({
            "sub": "user_1", "type": "refresh", "id": secrets.token_hex(8),
            "exp": int(time.time()) + 300,
        }, self.secret, algorithm="HS256")
        return access, refresh

    def simulate_session(self, duration_minutes=10):
        access, refresh = self.create_tokens()
        start = time.time()
        end = start + duration_minutes * 60
        now = start

        while now < end:
            try:
                jwt.decode(access, self.secret, algorithms=["HS256"])
                self.events.append({"time": datetime.utcnow(), "event": "API call OK"})
            except jwt.ExpiredSignatureError:
                try:
                    jwt.decode(refresh, self.secret, algorithms=["HS256"])
                    access, refresh = self.create_tokens()
                    self.refresh_count += 1
                    self.events.append({"time": datetime.utcnow(), "event": f"Token refreshed (#{self.refresh_count})"})
                except jwt.ExpiredSignatureError:
                    self.events.append({"time": datetime.utcnow(), "event": "Session expired"})
                    break
            now += 20

        print(f"=== Session Report ===")
        print(f"Duration: {duration_minutes} minutes simulated")
        print(f"API calls: {len(self.events)}")
        print(f"Refreshes: {self.refresh_count}")
        print(f"Active session: {'Yes' if self.events and self.events[-1]['event'] != 'Session expired' else 'No'}")

sim = TokenSimulator()
sim.simulate_session(5)