JWT Authentication and TOTP 2FA

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

POST /api/auth/verify-2fa { temp_token, otp_code }
→ 200 + Set-Cookie: access_token, refresh_token

GET /api/account/balance
Cookie: access_token=<JWT>
→ 200 { ... }

POST /api/auth/refresh (refresh_token HttpOnly cookie)
→ 200 + Set-Cookie: access_token, rotated refresh_token

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

# 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_"

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

# app/auth/jwt.py
from datetime import datetime, timedelta, timezone
from uuid import uuid4
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, family: str | None = None) -> 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,
        "jti": str(uuid4()),
        "family": family or str(uuid4()),
        "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],
        options={"require_exp": True, "require_iat": True, "require_sub": True},
    )
 
    if payload.get("type") != expected_type:
        raise ValueError(
            f"Expected '{expected_type}', got '{payload.get('type')}'"
        )
 
    if not isinstance(payload.get("sub"), str) or not payload["sub"]:
        raise ValueError("Token requires a non-empty sub")
 
    if expected_type == "refresh" and not all(
        isinstance(payload.get(claim), str) and payload[claim]
        for claim in ("jti", "family")
    ):
        raise ValueError("Refresh token requires jti and family")
 
    return payload

# app/auth/deps.py
from typing import Annotated
from fastapi import Cookie, Depends, HTTPException, status
from jose import JWTError
from app.auth.jwt import decode_token
 
async def get_current_user(
    access_token: Annotated[str | None, Cookie()] = None,
) -> dict:
    """Return the currently authenticated user."""
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
 
    if access_token is None:
        raise credentials_exception
 
    try:
        payload = decode_token(access_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)]

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'
 
# repeated calls in one time step can both return True
totp.verify(code)
totp.verify(code)
 
# 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'

# 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")

class LoginRequest(BaseModel):
    username: str
    password: str
 
 
class LoginResponse(BaseModel):
    requires_2fa: bool = False
    temp_token: str | None = None
    token_type: str = "bearer"
 
 
def set_session_cookies(response: Response, subject: str) -> None:
    access = create_access_token(subject)
    refresh = create_refresh_token(subject)
    auth_state.register_refresh(refresh)
    response.set_cookie(
        "access_token", access,
        httponly=True, secure=True, samesite="strict",
        max_age=30 * 60, path="/",
    )
    response.set_cookie(
        "refresh_token", refresh,
        httponly=True, secure=True, samesite="strict",
        max_age=7 * 24 * 3600, path="/api/auth/refresh",
    )
 
 
@router.post("/login", response_model=LoginResponse)
async def login(req: LoginRequest, response: Response):
    """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)
 
    set_session_cookies(response, req.username)
    return LoginResponse()
 
 
class Verify2FARequest(BaseModel):
    temp_token: str
    otp_code: str
 
 
@router.post("/verify-2fa", response_model=LoginResponse)
async def verify_2fa_login(req: Verify2FARequest, response: Response):
    """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"]
 
    user = auth_state.load_user(username)
    if user is None or not verify_totp_once(
        username, user.otp_secret, req.otp_code
    ):
        raise HTTPException(400, "Invalid OTP code")
 
    set_session_cookies(response, username)
    return LoginResponse()

import hmac
import sqlite3
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Annotated
 
import pyotp
from fastapi import Cookie, HTTPException, Response
from fastapi.responses import JSONResponse
from jose import JWTError
 
 
class AuthState:
    def __init__(self, path: Path):
        self.path = path
        with self.connect() as db:
            db.executescript("""
                CREATE TABLE IF NOT EXISTS refresh_tokens (
                    jti TEXT PRIMARY KEY,
                    family TEXT NOT NULL,
                    subject TEXT NOT NULL,
                    expires_at INTEGER NOT NULL,
                    used_at INTEGER,
                    revoked_at INTEGER
                );
                CREATE TABLE IF NOT EXISTS totp_steps (
                    subject TEXT PRIMARY KEY,
                    last_step INTEGER NOT NULL
                );
            """)
 
    def connect(self):
        return sqlite3.connect(self.path, isolation_level=None)
 
    def register_refresh(self, token: str) -> None:
        payload = decode_token(token, expected_type="refresh")
        with self.connect() as db:
            db.execute(
                "INSERT INTO refresh_tokens VALUES (?, ?, ?, ?, NULL, NULL)",
                (payload["jti"], payload["family"], payload["sub"], payload["exp"]),
            )
 
    def rotate_refresh(self, token: str) -> str:
        payload = decode_token(token, expected_type="refresh")
        now = int(time.time())
        with self.connect() as db:
            db.execute("BEGIN IMMEDIATE")
            row = db.execute(
                "SELECT used_at, revoked_at, expires_at FROM refresh_tokens WHERE jti = ?",
                (payload["jti"],),
            ).fetchone()
            if row is None or row[0] is not None or row[1] is not None or row[2] <= now:
                db.execute(
                    "UPDATE refresh_tokens SET revoked_at = ? WHERE family = ?",
                    (now, payload["family"]),
                )
                db.execute("COMMIT")
                raise ValueError("refresh token reuse or invalid state")
 
            db.execute(
                "UPDATE refresh_tokens SET used_at = ? WHERE jti = ?",
                (now, payload["jti"]),
            )
            new_token = create_refresh_token(payload["sub"], payload["family"])
            new_payload = decode_token(new_token, expected_type="refresh")
            db.execute(
                "INSERT INTO refresh_tokens VALUES (?, ?, ?, ?, NULL, NULL)",
                (new_payload["jti"], new_payload["family"], new_payload["sub"], new_payload["exp"]),
            )
            return new_token
 
    def claim_totp_step(self, subject: str, step: int) -> bool:
        with self.connect() as db:
            result = db.execute("""
                INSERT INTO totp_steps(subject, last_step) VALUES (?, ?)
                ON CONFLICT(subject) DO UPDATE SET last_step = excluded.last_step
                WHERE excluded.last_step > totp_steps.last_step
            """, (subject, step))
            return result.rowcount == 1
 
 
auth_state = AuthState(Path("auth-state.sqlite3"))
 
 
def verify_totp_once(subject: str, secret: str, code: str) -> bool:
    totp = pyotp.TOTP(secret)
    current_step = totp.timecode(datetime.now(timezone.utc))
    for step in range(current_step - 1, current_step + 2):
        if hmac.compare_digest(totp.generate_otp(step), code):
            return auth_state.claim_totp_step(subject, step)
    return False
 
 
@router.post("/refresh")
async def rotate_refresh(
    response: Response,
    refresh_token: Annotated[str | None, Cookie()] = None,
):
    if refresh_token is None:
        raise HTTPException(401, "Missing refresh token")
 
    try:
        payload = decode_token(refresh_token, expected_type="refresh")
        new_refresh_token = auth_state.rotate_refresh(refresh_token)
    except (JWTError, ValueError):
        error = JSONResponse(
            status_code=401,
            content={"detail": "Invalid or reused refresh token"},
        )
        error.delete_cookie("refresh_token", path="/api/auth/refresh")
        error.delete_cookie("access_token", path="/")
        return error
 
    access_token = create_access_token(payload["sub"])
    response.set_cookie(
        "refresh_token", new_refresh_token,
        httponly=True, secure=True, samesite="strict",
        max_age=7 * 24 * 3600, path="/api/auth/refresh",
    )
    response.set_cookie(
        "access_token", access_token,
        httponly=True, secure=True, samesite="strict",
        max_age=30 * 60, path="/",
    )
    return {"refreshed": True}

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)
    auth_state.register_refresh(refresh_token)
 
    response = JSONResponse(content={"authenticated": True})
    response.set_cookie(
        key="access_token", value=access_token,
        httponly=True, secure=True, samesite="strict",
        max_age=30 * 60, path="/",
    )
    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

pnpm add jose

// 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"];
 
function isPublicPath(pathname: string): boolean {
  return PUBLIC_PATHS.some(
    (path) => pathname === path || pathname.startsWith(`${path}/`),
  );
}
 
function unauthorized(request: NextRequest) {
  const { pathname } = request.nextUrl;
  if (pathname === "/api" || pathname.startsWith("/api/")) {
    return NextResponse.json({ detail: "Unauthorized" }, { status: 401 });
  }
  return NextResponse.redirect(new URL("/login", request.url));
}
 
export async function middleware(request: NextRequest) {
  const { pathname } = request.nextUrl;
 
  // let public paths through
  if (isPublicPath(pathname)) {
    return NextResponse.next();
  }
 
  // read the access token from the same-origin HttpOnly cookie
  const token = request.cookies.get("access_token")?.value;
 
  if (!token) {
    return unauthorized(request);
  }
 
  try {
    // verify the JWT (jose - Edge Runtime compatible)
    const { payload } = await jwtVerify(token, JWT_SECRET, {
      algorithms: ["HS256"],
      requiredClaims: ["sub", "exp", "iat", "type"],
    });
 
    if (payload.type !== "access" || typeof payload.sub !== "string" || !payload.sub) {
      throw new Error("Invalid access token claims");
    }
 
    // 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 {
    return unauthorized(request);
  }
}
 
export const config = {
  matcher: [
    "/((?!_next/static|_next/image|favicon.ico).*)",
  ],
};

// lib/auth.ts
export async function fetchWithAuth(
  url: string,
  options: RequestInit = {},
): Promise<Response> {
  let response = await fetch(url, { ...options, credentials: "include" });
 
  // 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) {
      response = await fetch(url, { ...options, credentials: "include" });
    } else {
      // refresh failed too, go to the login page
      window.location.href = "/login";
    }
  }
 
  return response;
}

// 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;

// 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;
}

# 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"],
)

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

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

pnpm add next@latest  # 15.2.3 or newer

// add at the very top of middleware.ts
if (request.headers.get("x-middleware-subrequest")) {
  return new NextResponse(null, { status: 403 });
}

# 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())

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