JWT Authentication and TOTP 2FA
A working security setup built from a three-token JWT strategy, pyotp TOTP 2FA, and Next.js Edge middleware verification
Contents
Split tokens into three kinds by purpose and separate issuance (FastAPI) from verification (Next.js Edge), and the two layers only need to share a secret for authentication to work.
End-to-end authentication flow
The FastAPI backend issues tokens after checking the password and a TOTP-based second factor (2FA, Two-Factor Authentication). The Next.js frontend middleware verifies that token on every request. Both layers share the same secret for the JSON Web Token (JWT), so issuance and verification happen independently in each layer.
System components
The four authentication steps
The whole authentication process runs in four steps.
Step 1: Login (password check)
POST /api/auth/login { username, password }
→ 200 { requires_2fa: true, temp_token: "..." }Step 2: 2FA verification
POST /api/auth/verify-2fa { temp_token, otp_code }
→ 200 { access_token, refresh_token }Step 3: API request
GET /api/account/balance
Authorization: Bearer <access_token>
→ 200 { ... }Step 4: Token refresh
POST /api/auth/refresh { refresh_token }
→ 200 { access_token, new_refresh_token }The three-token JWT strategy
Instead of a single token, this setup uses three tokens split by purpose. Each is distinguished by a type claim and differs in expiry and storage location.
| Token | Expiry | Storage | Purpose |
|---|---|---|---|
| Access Token | 30 minutes | Memory (JS variable) | Authenticating API requests |
| Refresh Token | 7 days | HttpOnly Cookie | Renewing the access token |
| Temp Token | 5 minutes | Memory | Temporary auth while waiting for 2FA |
Keeping the access token in memory only reduces the risk of it being stolen through Cross-Site Scripting (XSS). The refresh token goes into a
HttpOnly+Secure+SameSite=Strictcookie so JavaScript cannot reach it.
Installing dependencies
# pyproject.toml
[project]
dependencies = [
"fastapi>=0.115",
"python-jose[cryptography]>=3.3", # JWT
"passlib[bcrypt]>=1.7", # password hashing
"pyotp>=2.9", # TOTP 2FA
"qrcode[pil]>=7.4", # QR code generation
]python-jose also covers JWK and JWE, but it is not actively maintained and has a CVE history. If all you need is JWT signing and verification, PyJWT is a well-maintained alternative. The jwt.encode and jwt.decode calls in this post port to either library with almost no change.
Auth settings
# app/auth/config.py
from pydantic_settings import BaseSettings
class AuthSettings(BaseSettings):
# JWT
jwt_secret: str # openssl rand -hex 32
jwt_algorithm: str = "HS256"
access_token_expire_minutes: int = 30 # 30 minutes
refresh_token_expire_days: int = 7 # 7 days
temp_token_expire_minutes: int = 5 # 5 minutes for the 2FA wait token
# TOTP
totp_issuer: str = "PersonalTrader"
totp_interval: int = 30 # 30-second period
# Security
max_login_attempts: int = 5 # lock after 5 failures
lockout_minutes: int = 15 # 15-minute lockout
class Config:
env_prefix = "AUTH_"Password hashing
Never store plaintext passwords. Hash them one-way with bcrypt.
# app/auth/password.py
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
return pwd_context.verify(plain, hashed)Creating and verifying JWT tokens
All three tokens are signed with the same secret key and distinguished by the type claim.
# app/auth/jwt.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from app.auth.config import AuthSettings
settings = AuthSettings()
def create_access_token(
sub: str,
extra: dict | None = None,
expires_delta: timedelta | None = None,
) -> str:
"""Create an access token (30 minutes by default)."""
now = datetime.now(timezone.utc)
expire = now + (expires_delta or timedelta(
minutes=settings.access_token_expire_minutes
))
payload = {
"sub": sub,
"exp": expire,
"iat": now,
"type": "access",
}
if extra:
payload.update(extra)
return jwt.encode(
payload, settings.jwt_secret, algorithm=settings.jwt_algorithm
)
def create_refresh_token(sub: str) -> str:
"""Create a refresh token (7 days)."""
now = datetime.now(timezone.utc)
expire = now + timedelta(days=settings.refresh_token_expire_days)
payload = {
"sub": sub,
"exp": expire,
"iat": now,
"type": "refresh",
}
return jwt.encode(
payload, settings.jwt_secret, algorithm=settings.jwt_algorithm
)
def create_temp_token(sub: str) -> str:
"""Create a temporary token for the 2FA wait (5 minutes)."""
now = datetime.now(timezone.utc)
expire = now + timedelta(
minutes=settings.temp_token_expire_minutes
)
payload = {
"sub": sub,
"exp": expire,
"iat": now,
"type": "temp_2fa",
}
return jwt.encode(
payload, settings.jwt_secret, algorithm=settings.jwt_algorithm
)
def decode_token(token: str, expected_type: str = "access") -> dict:
"""Decode a JWT token and check the type claim."""
payload = jwt.decode(
token, settings.jwt_secret,
algorithms=[settings.jwt_algorithm]
)
if payload.get("type") != expected_type:
raise ValueError(
f"Expected '{expected_type}', got '{payload.get('type')}'"
)
return payloadFastAPI dependency injection
Apply authentication to protected routes with Depends.
# app/auth/deps.py
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from app.auth.jwt import decode_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
async def get_current_user(
token: Annotated[str, Depends(oauth2_scheme)],
) -> dict:
"""Return the currently authenticated user."""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = decode_token(token, expected_type="access")
username: str | None = payload.get("sub")
if username is None:
raise credentials_exception
except (JWTError, ValueError):
raise credentials_exception
return {"username": username}
# used on protected routes
CurrentUser = Annotated[dict, Depends(get_current_user)]TOTP with pyotp and QR codes
How TOTP works
A Time-based One-Time Password (TOTP) is generated from a shared secret key and the current time. The server and the client (the authenticator app) hold the same secret and produce the same six-digit code on a 30-second interval.
Basic pyotp usage
import pyotp
# generate the secret key (once per user)
secret = pyotp.random_base32() # looks like 'JBSWY3DPEHPK3PXP'
# create a TOTP object
totp = pyotp.TOTP(secret)
# current OTP code (6 digits)
code = totp.now() # '492039'
# verify a code
totp.verify("492039") # True (within 30 seconds)
totp.verify("492039") # False (after 30 seconds)
# provisioning URI for the QR code
uri = totp.provisioning_uri(
name="user@example.com",
issuer_name="PersonalTrader",
)
# → 'otpauth://totp/PersonalTrader:user@example.com?secret=...&issuer=PersonalTrader'2FA setup endpoints
# app/auth/routes.py
import io
import pyotp
import qrcode
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
router = APIRouter(prefix="/api/auth", tags=["auth"])
class OTPSetupResponse(BaseModel):
otp_secret: str
otp_uri: str
@router.post("/2fa/setup", response_model=OTPSetupResponse)
async def setup_2fa(current_user: CurrentUser):
"""2FA setup - return the OTP secret key and QR URI."""
secret = pyotp.random_base32()
totp = pyotp.TOTP(secret)
uri = totp.provisioning_uri(
name=current_user["username"],
issuer_name="PersonalTrader",
)
# store otp_secret in the DB provisionally (before activation)
# await user_repo.set_pending_otp_secret(username, secret)
return OTPSetupResponse(otp_secret=secret, otp_uri=uri)
@router.get("/2fa/qrcode")
async def get_2fa_qrcode(current_user: CurrentUser):
"""Return the 2FA QR code image."""
# look up the pending otp_secret in the DB
secret = "PENDING_SECRET" # replace with a real DB lookup
totp = pyotp.TOTP(secret)
uri = totp.provisioning_uri(
name=current_user["username"],
issuer_name="PersonalTrader",
)
img = qrcode.make(uri)
buf = io.BytesIO()
img.save(buf, format="PNG")
buf.seek(0)
return StreamingResponse(buf, media_type="image/png")The user flow:
- Call
/2fa/setupand receiveotp_secretandotp_uri. - Turn
otp_uriinto a QR code, or fetch the image from the/2fa/qrcodeendpoint. - Scan the QR code in Google Authenticator or a similar app.
- Verify with the code the app displays, which activates 2FA.
Login flow (password plus 2FA)
class LoginRequest(BaseModel):
username: str
password: str
class LoginResponse(BaseModel):
requires_2fa: bool = False
temp_token: str | None = None
access_token: str | None = None
refresh_token: str | None = None
token_type: str = "bearer"
@router.post("/login", response_model=LoginResponse)
async def login(req: LoginRequest):
"""Step 1: verify the password."""
# user = await user_repo.get_by_username(req.username)
# if not user or not verify_password(req.password, user.hashed_password):
# raise HTTPException(401, "Incorrect username or password")
# check the login attempt count (brute-force protection)
# if user.login_attempts >= settings.max_login_attempts:
# raise HTTPException(429, "Account locked. Try again later.")
has_2fa = True # user.otp_enabled
if has_2fa:
temp_token = create_temp_token(sub=req.username)
return LoginResponse(requires_2fa=True, temp_token=temp_token)
access_token = create_access_token(sub=req.username)
refresh_token = create_refresh_token(sub=req.username)
return LoginResponse(
access_token=access_token,
refresh_token=refresh_token,
)
class Verify2FARequest(BaseModel):
temp_token: str
otp_code: str
@router.post("/verify-2fa", response_model=LoginResponse)
async def verify_2fa_login(req: Verify2FARequest):
"""Step 2: verify the TOTP code, then issue the real tokens."""
try:
payload = decode_token(req.temp_token, expected_type="temp_2fa")
except Exception:
raise HTTPException(401, "Invalid or expired temp token")
username = payload["sub"]
# look up otp_secret in the DB
otp_secret = "USER_OTP_SECRET"
totp = pyotp.TOTP(otp_secret)
if not totp.verify(req.otp_code, valid_window=1): # +-30 seconds allowed
raise HTTPException(400, "Invalid OTP code")
access_token = create_access_token(sub=username)
refresh_token = create_refresh_token(sub=username)
return LoginResponse(
access_token=access_token,
refresh_token=refresh_token,
)
valid_window=1accepts codes from one interval before and after the current one, 90 seconds in total. That absorbs clock drift between server and client so authentication does not fail for a harmless reason.
Refresh token rotation and HttpOnly cookies
Refresh token rotation
Every time a refresh token is used, issue a new refresh token and blacklist the old one. Reuse of a stolen token is then detected immediately.
@router.post("/refresh")
async def refresh_token(req: RefreshRequest):
"""Refresh token rotation - issue a new refresh token on every renewal."""
payload = decode_token(req.refresh_token, expected_type="refresh")
# invalidate the old refresh token (DB blacklist)
# await token_repo.blacklist(req.refresh_token)
access_token = create_access_token(sub=payload["sub"])
new_refresh_token = create_refresh_token(sub=payload["sub"])
return LoginResponse(
access_token=access_token,
refresh_token=new_refresh_token,
)Setting the HttpOnly cookie
Delivering the refresh token in an HttpOnly cookie keeps it out of reach of JavaScript, which makes it safe against XSS.
from fastapi.responses import JSONResponse
@router.post("/login-cookie")
async def login_with_cookie(req: LoginRequest):
"""Set the refresh token as an HttpOnly cookie."""
# ... authentication logic ...
access_token = create_access_token(sub=req.username)
refresh_token = create_refresh_token(sub=req.username)
response = JSONResponse(content={
"access_token": access_token,
"token_type": "bearer",
})
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True, # block JS access
secure=True, # HTTPS only
samesite="strict", # CSRF protection
max_age=7 * 24 * 3600, # 7 days
path="/api/auth/refresh", # sent only to the refresh endpoint
)
return responseCookie security options:
| Option | Value | Effect |
|---|---|---|
httponly | True | Not reachable through document.cookie (XSS defense) |
secure | True | Cookie sent only over HTTPS |
samesite | strict | Cookie not sent on requests from other sites (defense against Cross-Site Request Forgery, CSRF) |
path | /api/auth/refresh | Cookie scope limited to the refresh endpoint |
Next.js Edge Runtime middleware
Verifying JWTs in the Edge Runtime
Next.js middleware runs in the Edge Runtime. The jsonwebtoken library is Node.js-only and cannot be used there, so verification goes through jose, which is Edge-compatible.
pnpm add joseMiddleware implementation
// middleware.ts (project root)
import { NextRequest, NextResponse } from "next/server";
import { jwtVerify } from "jose";
const JWT_SECRET = new TextEncoder().encode(
process.env.JWT_SECRET!
);
// paths that do not require authentication
const PUBLIC_PATHS = ["/login", "/api/auth"];
export async function middleware(request: NextRequest) {
const { pathname } = request.nextUrl;
// let public paths through
if (PUBLIC_PATHS.some((path) => pathname.startsWith(path))) {
return NextResponse.next();
}
// look for the access token (cookie or Authorization header)
const token =
request.cookies.get("access_token")?.value ||
request.headers
.get("authorization")
?.replace("Bearer ", "");
if (!token) {
return NextResponse.redirect(new URL("/login", request.url));
}
try {
// verify the JWT (jose - Edge Runtime compatible)
const { payload } = await jwtVerify(token, JWT_SECRET, {
algorithms: ["HS256"],
});
// pass the user info downstream on the request headers
// (putting it on response.headers exposes it to the browser and never reaches server components)
const requestHeaders = new Headers(request.headers);
requestHeaders.set("x-user", payload.sub as string);
return NextResponse.next({ request: { headers: requestHeaders } });
} catch {
// expired or invalid token, redirect to login
return NextResponse.redirect(new URL("/login", request.url));
}
}
export const config = {
matcher: [
"/((?!_next/static|_next/image|favicon.ico).*)",
],
};Automatic token refresh on the client
This utility function transparently uses the refresh token when the access token has expired.
// lib/auth.ts
let accessToken: string | null = null;
export async function fetchWithAuth(
url: string,
options: RequestInit = {},
): Promise<Response> {
const headers = new Headers(options.headers);
if (accessToken) {
headers.set("Authorization", `Bearer ${accessToken}`);
}
let response = await fetch(url, { ...options, headers });
// on 401, try to renew with the refresh token
if (response.status === 401) {
const refreshResponse = await fetch("/api/auth/refresh", {
method: "POST",
credentials: "include", // send the HttpOnly cookie
});
if (refreshResponse.ok) {
const data = await refreshResponse.json();
accessToken = data.access_token;
// retry the original request
headers.set("Authorization", `Bearer ${accessToken}`);
response = await fetch(url, { ...options, headers });
} else {
// refresh failed too, go to the login page
window.location.href = "/login";
}
}
return response;
}Security headers
Next.js security headers
Add security headers to every response in next.config.ts.
// next.config.ts
const securityHeaders = [
{
key: "X-DNS-Prefetch-Control",
value: "on",
},
{
key: "Strict-Transport-Security",
value: "max-age=63072000; includeSubDomains; preload",
},
{
key: "X-Frame-Options",
value: "SAMEORIGIN",
},
{
key: "X-Content-Type-Options",
value: "nosniff",
},
{
key: "Referrer-Policy",
value: "origin-when-cross-origin",
},
{
key: "Permissions-Policy",
value: "camera=(), microphone=(), geolocation=()",
},
];
const nextConfig = {
async headers() {
return [
{
source: "/(.*)",
headers: securityHeaders,
},
];
},
};
export default nextConfig;What each header does:
| Header | Effect |
|---|---|
Strict-Transport-Security | Forces HTTPS (two years, subdomains included) |
X-Frame-Options | Prevents clickjacking (blocks iframe embedding) |
X-Content-Type-Options | Prevents MIME sniffing |
Referrer-Policy | Sends only the origin instead of the full URL on cross-site navigation |
Permissions-Policy | Disables the camera, microphone, and geolocation APIs |
Content Security Policy (CSP)
CSP narrows the blast radius of XSS by restricting which script sources may execute. With a nonce-based policy, even inline scripts run only when they carry the signed value.
Next.js allows exactly one middleware per project. The CSP logic below is not a separate file, it must be merged into the same middleware() function as the JWT verification above. The order is: generate the nonce and attach it to the request headers, verify the token, then set the CSP header on the final NextResponse.
// middleware.ts (CSP part - merge into one function with the JWT verification logic)
export function middleware(request: NextRequest) {
const nonce = Buffer.from(crypto.randomUUID())
.toString("base64");
const cspHeader = `
default-src 'self';
script-src 'self' 'nonce-${nonce}' 'strict-dynamic';
style-src 'self' 'unsafe-inline';
img-src 'self' blob: data:;
font-src 'self';
connect-src 'self' wss://jongkwan.dev ws://localhost:*;
frame-ancestors 'none';
base-uri 'self';
form-action 'self';
`.replace(/\s{2,}/g, " ").trim();
const response = NextResponse.next();
response.headers.set("Content-Security-Policy", cspHeader);
response.headers.set("x-nonce", nonce);
return response;
}FastAPI CORS configuration
The frontend and backend run on different domains (or ports), so CORS has to be configured.
# app/main.py
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=[
"https://jongkwan.dev",
"http://localhost:3000", # development
],
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE"],
allow_headers=["Authorization", "Content-Type"],
)
allow_credentials=Truecannot be combined with a wildcard (*) inallow_origins. Concrete origins have to be listed.
Rate limiting
To block brute-force attacks, the login endpoint gets a rate limit.
# app/auth/middleware.py
from collections import defaultdict
from datetime import datetime, timezone
from fastapi import Request, HTTPException
login_attempts: dict[str, list[datetime]] = defaultdict(list)
RATE_LIMIT_WINDOW = 900 # 15 minutes
RATE_LIMIT_MAX = 5 # 5 attempts
async def check_login_rate_limit(request: Request):
"""Limit login attempts (per IP)."""
client_ip = request.client.host if request.client else "unknown"
now = datetime.now(timezone.utc)
# drop records outside the window
attempts = login_attempts[client_ip]
cutoff = now.timestamp() - RATE_LIMIT_WINDOW
attempts[:] = [a for a in attempts if a.timestamp() > cutoff]
if len(attempts) >= RATE_LIMIT_MAX:
raise HTTPException(
status_code=429,
detail=f"Too many login attempts. "
f"Try again in {RATE_LIMIT_WINDOW // 60} minutes.",
)
attempts.append(now)This counter lives in process memory, so every worker or instance counts separately. In a multi-instance deployment it has to move to shared storage such as Redis before the limit means anything.
Configuration summary:
| Item | Value |
|---|---|
| Maximum attempts | 5 per 15 minutes |
| Lockout duration | 15 minutes |
| Limit key | Client IP |
Handling CVE-2025-29927
The vulnerability
A critical vulnerability in Next.js allows complete middleware bypass through the x-middleware-subrequest header.
Affected versions: Next.js < 15.2.3, < 14.2.25, < 13.5.9
Severity: CriticalExploiting it lets an attacker skip the authentication middleware and reach protected routes directly.
The fix
First priority: update Next.js
pnpm add next@latest # 15.2.3 or newerTemporary mitigation (before updating)
// add at the very top of middleware.ts
if (request.headers.get("x-middleware-subrequest")) {
return new NextResponse(null, { status: 403 });
}This mitigation is an emergency measure only, meant to hold until the update lands. Next.js still has to be updated to a current version.
Security checklist
The priority levels below assume a service exposed to the internet. Every required item has to be met before going live.
| Item | Description | Priority |
|---|---|---|
| JWT secret | At least 256 bits (openssl rand -hex 32) | Required |
| HTTPS only | Guaranteed by Cloudflare Tunnel | Required |
| HttpOnly cookie | XSS protection for the refresh token | Required |
| SameSite=Strict | CSRF protection | Required |
| Rate limiting | 5 logins per 15 minutes | Required |
| HSTS | max-age two years, preload | Required |
| X-Frame-Options | SAMEORIGIN (clickjacking protection) | Required |
| Token rotation | New token issued on every refresh | Recommended |
| TOTP valid_window=1 | Accepts only 30 seconds either side | Recommended |
| CSP | Nonce-based script-src | Recommended |
| CVE-2025-29927 | Next.js 15.2.3 or newer | Required |
Data model
otp_secret holds the shared TOTP key, and otp_enabled and otp_verified distinguish the stages of 2FA activation.
login_attempts and locked_until record the lockout state used for brute-force defense.
# app/models/user.py
from sqlalchemy import Boolean, Column, DateTime, Integer, String
from sqlalchemy.sql import func
from app.db.base import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
username = Column(String(50), unique=True, nullable=False, index=True)
email = Column(String(255), unique=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
# 2FA
otp_enabled = Column(Boolean, default=False)
otp_secret = Column(String(32), nullable=True)
otp_verified = Column(Boolean, default=False)
# security
login_attempts = Column(Integer, default=0)
locked_until = Column(DateTime(timezone=True), nullable=True)
last_login = Column(DateTime(timezone=True), nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), onupdate=func.now())Environment variables
# JWT
AUTH_JWT_SECRET=<openssl rand -hex 32>
AUTH_JWT_ALGORITHM=HS256
AUTH_ACCESS_TOKEN_EXPIRE_MINUTES=30
AUTH_REFRESH_TOKEN_EXPIRE_DAYS=7
# TOTP
AUTH_TOTP_ISSUER=PersonalTrader
# Next.js
JWT_SECRET=<same value as AUTH_JWT_SECRET>The JWT secret must be a random value of at least 256 bits, generated with openssl rand -hex 32. FastAPI and Next.js have to share the same secret for both sides to verify tokens.
Summary
The core of this design is splitting tokens by purpose and separating where each one is stored. The access token lives in memory and the refresh token in an HttpOnly cookie, which shrinks the surface for XSS and token theft. TOTP 2FA holds the line as a second gate even if the password leaks, and refresh token rotation detects reuse of a stolen token.
Verification is separated from issuance and handled by the Next.js Edge middleware on every request, so the two layers only need to share a secret. Rate limiting, security headers, and the CVE-2025-29927 fix have to be in place before this counts as a meaningful defense on an internet-facing service. Pieces such as the in-memory rate limit and the python-jose dependency need replacing as the deployment grows, and the criteria for that are marked in the text above.