jongkwan.dev
Development · Essay №054

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

Jongkwan Lee2026년 2월 28일17 min read
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)

http
POST /api/auth/login { username, password }
→ 200 { requires_2fa: true, temp_token: "..." }

Step 2: 2FA verification

http
POST /api/auth/verify-2fa { temp_token, otp_code }
→ 200 { access_token, refresh_token }

Step 3: API request

http
GET /api/account/balance
Authorization: Bearer <access_token>
→ 200 { ... }

Step 4: Token refresh

http
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.

TokenExpiryStoragePurpose
Access Token30 minutesMemory (JS variable)Authenticating API requests
Refresh Token7 daysHttpOnly CookieRenewing the access token
Temp Token5 minutesMemoryTemporary 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=Strict cookie so JavaScript cannot reach it.

Installing dependencies

toml
# 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

python
# 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.

python
# 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.

python
# 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 payload

FastAPI dependency injection

Apply authentication to protected routes with Depends.

python
# 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

python
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

python
# 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:

  1. Call /2fa/setup and receive otp_secret and otp_uri.
  2. Turn otp_uri into a QR code, or fetch the image from the /2fa/qrcode endpoint.
  3. Scan the QR code in Google Authenticator or a similar app.
  4. Verify with the code the app displays, which activates 2FA.

Login flow (password plus 2FA)

python
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=1 accepts 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.

python
@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,
    )

Delivering the refresh token in an HttpOnly cookie keeps it out of reach of JavaScript, which makes it safe against XSS.

python
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 response

Cookie security options:

OptionValueEffect
httponlyTrueNot reachable through document.cookie (XSS defense)
secureTrueCookie sent only over HTTPS
samesitestrictCookie not sent on requests from other sites (defense against Cross-Site Request Forgery, CSRF)
path/api/auth/refreshCookie 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.

bash
pnpm add jose

Middleware implementation

typescript
// 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.

typescript
// 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.

typescript
// 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:

HeaderEffect
Strict-Transport-SecurityForces HTTPS (two years, subdomains included)
X-Frame-OptionsPrevents clickjacking (blocks iframe embedding)
X-Content-Type-OptionsPrevents MIME sniffing
Referrer-PolicySends only the origin instead of the full URL on cross-site navigation
Permissions-PolicyDisables 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.

typescript
// 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.

python
# 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=True cannot be combined with a wildcard (*) in allow_origins. Concrete origins have to be listed.


Rate limiting

To block brute-force attacks, the login endpoint gets a rate limit.

python
# 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:

ItemValue
Maximum attempts5 per 15 minutes
Lockout duration15 minutes
Limit keyClient IP

Handling CVE-2025-29927

The vulnerability

A critical vulnerability in Next.js allows complete middleware bypass through the x-middleware-subrequest header.

text
Affected versions: Next.js < 15.2.3, < 14.2.25, < 13.5.9
Severity: Critical

Exploiting it lets an attacker skip the authentication middleware and reach protected routes directly.

The fix

First priority: update Next.js

bash
pnpm add next@latest  # 15.2.3 or newer

Temporary mitigation (before updating)

typescript
// 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.

ItemDescriptionPriority
JWT secretAt least 256 bits (openssl rand -hex 32)Required
HTTPS onlyGuaranteed by Cloudflare TunnelRequired
HttpOnly cookieXSS protection for the refresh tokenRequired
SameSite=StrictCSRF protectionRequired
Rate limiting5 logins per 15 minutesRequired
HSTSmax-age two years, preloadRequired
X-Frame-OptionsSAMEORIGIN (clickjacking protection)Required
Token rotationNew token issued on every refreshRecommended
TOTP valid_window=1Accepts only 30 seconds either sideRecommended
CSPNonce-based script-srcRecommended
CVE-2025-29927Next.js 15.2.3 or newerRequired

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.

python
# 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

env
# 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.