Refresh Token Rotation — Automatically Rotating Refresh Tokens for Enhanced Security
In this tutorial, you will learn about Refresh Token Rotation. We cover key concepts, practical examples, and best practices to help you master this topic.
Refresh token rotation invalidates the previous refresh token each time a new one is issued, creating a chain of single-use tokens that limits the damage from token theft.
What You'll Learn
Refresh token rotation implementation, token family chains, concurrent refresh race conditions, theft detection via duplicate use, and storage considerations for rotated tokens.
Why It Matters
A static refresh token is a long-lived credential. If stolen, the attacker has permanent access. Rotation makes each refresh token single-use — the attacker must steal the most recent token and use it before the legitimate client.
Real-World Use
Auth0 uses rotating refresh tokens by default for SPAs. Google's OAuth2 implementation rotates refresh tokens. Durga Antivirus Pro rotates partner refresh tokens on every use with automatic theft alerts.
sequenceDiagram
participant Client as Client
participant Server as Auth Server
Client->>Server: POST /login
Server->>Client: { access_token, refresh_token: R1 }
Note over Client: Uses R1
Client->>Server: POST /refresh (R1)
Server->>Server: Invalidate R1, issue R2
Server->>Client: { access_token, refresh_token: R2 }
Note over Client: Uses R2
Client->>Server: POST /refresh (R2)
Server->>Server: Invalidate R2, issue R3
Server->>Client: { access_token, refresh_token: R3 }
Note over Client,Server: Theft scenario
Attacker->>Server: POST /refresh (R2) — but R2 is already used!
Server->>Server: Theft detected! Revoke entire chain
Server->>Attacker: 401 + "Token family revoked"
Code Example: Refresh Token Rotation Implementation
import secrets, hashlib, datetime, jwt
from flask import Flask, request, jsonify
from collections import defaultdict
app = Flask(__name__)
SECRET = "rotation-secret"
# token_families[family_id] = { active_hash, previous_hashes, user, revoked }
token_families = {}
def create_token_family(user_id):
family_id = secrets.token_urlsafe(16)
refresh_token = secrets.token_urlsafe(32)
token_hash = hashlib.sha256(refresh_token.encode()).hexdigest()
token_families[family_id] = {
"user_id": user_id,
"active_hash": token_hash,
"previous_hashes": [],
"revoked": False,
"created_at": datetime.datetime.utcnow().isoformat(),
"max_chain_length": 100
}
return family_id, refresh_token
def rotate_refresh_token(family_id, current_token_hash):
"""Rotate refresh token: invalidate old, issue new."""
family = token_families.get(family_id)
if not family or family["revoked"]:
return None, None
# Check if current token matches active hash (normal rotation)
if family["active_hash"] == current_token_hash:
# Move active hash to previous, generate new
family["previous_hashes"].append(current_token_hash)
new_token = secrets.token_urlsafe(32)
new_hash = hashlib.sha256(new_token.encode()).hexdigest()
family["active_hash"] = new_hash
return new_token, None
# Check if current token is a previous hash (theft detection)
if current_token_hash in family["previous_hashes"]:
family["revoked"] = True
return None, "Token family revoked — possible theft detected"
return None, "Invalid refresh token"
@app.route("/api/auth/login", methods=["POST"])
def login():
user = request.json.get("username", "analyst")
family_id, refresh_token = create_token_family(user)
access_token = jwt.encode({
"sub": user,
"iat": datetime.datetime.utcnow(),
"exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=15),
"jti": secrets.token_hex(16)
}, SECRET, algorithm="HS256")
return jsonify({
"access_token": access_token,
"refresh_token": refresh_token,
"token_family": family_id,
"expires_in": 900
})
@app.route("/api/auth/refresh", methods=["POST"])
def refresh():
old_token = request.json.get("refresh_token")
family_id = request.json.get("token_family")
if not old_token or not family_id:
return jsonify({"error": "Missing refresh token or family ID"}), 401
token_hash = hashlib.sha256(old_token.encode()).hexdigest()
new_token, error = rotate_refresh_token(family_id, token_hash)
if error:
return jsonify({"error": error}), 401
if not new_token:
return jsonify({"error": "Invalid refresh token"}), 401
new_access = jwt.encode({
"sub": token_families[family_id]["user_id"],
"iat": datetime.datetime.utcnow(),
"exp": datetime.datetime.utcnow() + datetime.timedelta(minutes=15)
}, SECRET, algorithm="HS256")
return jsonify({
"access_token": new_access,
"refresh_token": new_token,
"expires_in": 900
})
Code Example: Concurrent Refresh Handling
import threading
class SafeRotator:
"""Thread-safe refresh token rotation with locking."""
def __init__(self):
self.locks = defaultdict(threading.Lock)
def rotate(self, family_id, current_hash):
"""Thread-safe token rotation."""
with self.locks[family_id]:
return self._do_rotate(family_id, current_hash)
def _do_rotate(self, family_id, current_hash):
family = token_families.get(family_id)
if not family or family["revoked"]:
return None, "Invalid family"
if family["active_hash"] != current_hash:
return None, "Token already rotated"
family["previous_hashes"].append(current_hash)
new_token = secrets.token_urlsafe(32)
family["active_hash"] = hashlib.sha256(new_token.encode()).hexdigest()
return new_token, None
# Usage — two concurrent requests
rotator = SafeRotator()
@app.route("/api/auth/refresh", methods=["POST"])
def refresh_safe():
old_token = request.json.get("refresh_token")
family_id = request.json.get("token_family")
token_hash = hashlib.sha256(old_token.encode()).hexdigest()
new_token, error = rotator.rotate(family_id, token_hash)
if error:
return jsonify({"error": error}), 401
# Issue new access token...
return jsonify({"access_token": "...", "refresh_token": new_token})
Code Example: Client-Side Refresh Token Management
class RefreshTokenManager {
constructor(storageKey = 'auth_refresh') {
this.storageKey = storageKey;
this.currentToken = null;
this.currentFamily = null;
this.refreshPromise = null;
}
setTokens(accessToken, refreshToken, familyId) {
this.accessToken = accessToken;
this.currentToken = refreshToken;
this.currentFamily = familyId;
this.persistRefreshToken(refreshToken, familyId);
}
async refreshAccessToken() {
if (this.refreshPromise) return this.refreshPromise;
this.refreshPromise = (async () => {
const resp = await fetch('/api/auth/refresh', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
refresh_token: this.currentToken,
token_family: this.currentFamily
})
});
if (!resp.ok) {
// Token family may be revoked — force re-login
this.clearTokens();
throw new Error('Session expired');
}
const data = await resp.json();
// Old refresh token is now invalid — update to new one
this.currentToken = data.refresh_token;
this.accessToken = data.access_token;
this.persistRefreshToken(data.refresh_token, this.currentFamily);
return data.access_token;
})();
try {
return await this.refreshPromise;
} finally {
this.refreshPromise = null;
}
}
persistRefreshToken(token, family) {
try {
sessionStorage.setItem(this.storageKey, JSON.stringify({
token, family, updated: Date.now()
}));
} catch {
// Storage full or unavailable — token is in memory
}
}
clearTokens() {
this.accessToken = null;
this.currentToken = null;
this.currentFamily = null;
sessionStorage.removeItem(this.storageKey);
}
}
Common Mistakes
1. Not Detecting Token Theft
When a rotated token is presented after the new one has been issued, the server should revoke the entire token family. This detects and contains theft.
2. No Maximum Chain Length
Rotation creates an ever-growing chain of previous hashes. Set a maximum chain length (e.g., 100 rotations) after which the user must re-authenticate.
3. Race Conditions on Refresh
Two simultaneous refresh requests both see the same active token. Without locking, both can succeed, creating two valid refresh tokens. Use a mutex per family.
4. Storing Refresh Tokens in Plaintext
Hash refresh tokens with SHA-256 before storing. If the database is breached, hashed tokens prevent direct token theft.
5. Not Rotating on Refresh Failure
If refresh fails due to network error, do not discard the current token. Keep it and retry. Only rotate on a successful server response.
Practice Questions
- What is the primary security benefit of refresh token rotation?
- How does the server detect token theft using rotation?
- Why is thread-safe locking needed for rotation?
- What happens to the previous refresh token when a new one is issued?
- How does the maximum chain length prevent unbounded storage growth?
Answers:
- Each refresh token is single-use. If stolen, the token can only be used once before it becomes invalid. The legitimate client will detect the theft on its next refresh.
- When the server receives a previously valid (but not current) refresh token, it knows it was stolen. It revokes the entire family and alerts the security team.
- Without locking, two concurrent refresh requests could both pass validation, creating two valid refresh tokens. The attacker could use one and the client the other.
- The previous token is marked as used (moved to previous_hashes). If presented again, the server detects replay and revokes the family.
- After N rotations, force the user to re-authenticate. This limits the stored hash chain and ensures periodic credential verification.
Challenge: Build a complete refresh token rotation system with concurrent-safe rotation, theft detection, maximum chain length enforcement, and automated security alerts.
FAQ
Mini Project
Build a Flask API with refresh token rotation, concurrent-safe rotation with thread locking, theft detection with automatic family revocation, and a test suite that simulates token theft and verifies the family is blocked.
What's Next
Now learn about Refresh Token Expiry Strategies for configuring absolute and sliding expiry Windows.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro