JWT 인증과 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",            # 비밀번호 해싱
    "pyotp>=2.9",                      # TOTP 2FA
    "qrcode[pil]>=7.4",               # QR 코드 생성
]

# 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분
    refresh_token_expire_days: int = 7       # 7일
    temp_token_expire_minutes: int = 5       # 2FA 대기 토큰 5분
 
    # TOTP
    totp_issuer: str = "PersonalTrader"
    totp_interval: int = 30                  # 30초 주기
 
    # 보안
    max_login_attempts: int = 5              # 5회 실패 시 잠금
    lockout_minutes: int = 15               # 15분 잠금
 
    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:
    """Access Token 생성 (기본 30분)."""
    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:
    """Refresh Token 생성 (7일)."""
    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:
    """2FA 대기용 임시 토큰 (5분)."""
    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:
    """JWT 토큰 디코딩 및 type 클레임 검증."""
    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:
    """현재 인증된 사용자 반환."""
    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}
 
 
# 보호 라우트에서 사용
CurrentUser = Annotated[dict, Depends(get_current_user)]

import pyotp
 
# 비밀키 생성 (사용자별 1회)
secret = pyotp.random_base32()  # 'JBSWY3DPEHPK3PXP' 형태
 
# TOTP 객체 생성
totp = pyotp.TOTP(secret)
 
# 현재 OTP 코드 (6자리)
code = totp.now()  # '492039'
 
# 같은 시간 구간의 반복 호출은 둘 다 True일 수 있음
totp.verify(code)
totp.verify(code)
 
# QR 코드용 프로비저닝 URI
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 설정 - OTP 비밀키 및 QR URI 반환."""
    secret = pyotp.random_base32()
    totp = pyotp.TOTP(secret)
    uri = totp.provisioning_uri(
        name=current_user["username"],
        issuer_name="PersonalTrader",
    )
 
    # DB에 otp_secret 임시 저장 (활성화 전)
    # 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):
    """2FA QR 코드 이미지 반환."""
    # DB에서 pending otp_secret 조회
    secret = "PENDING_SECRET"  # 실제 구현 시 DB 조회
 
    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):
    """1단계: 비밀번호 검증."""
    # 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")
 
    # 로그인 시도 횟수 확인 (브루트포스 방지)
    # 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):
    """2단계: TOTP 코드 검증 후 실제 토큰 발급."""
    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):
    """Refresh Token을 HttpOnly Cookie로 설정."""
    # ... 인증 로직 ...
 
    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,           # JS 접근 차단
        secure=True,             # HTTPS only
        samesite="strict",       # CSRF 방지
        max_age=7 * 24 * 3600,   # 7일
        path="/api/auth/refresh", # 갱신 엔드포인트에서만 전송
    )
    return response

pnpm add jose

// middleware.ts (프로젝트 루트)
import { NextRequest, NextResponse } from "next/server";
import { jwtVerify } from "jose";
 
const JWT_SECRET = new TextEncoder().encode(
  process.env.JWT_SECRET!
);
 
// 인증 불필요한 경로
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;
 
  // 공개 경로는 통과
  if (isPublicPath(pathname)) {
    return NextResponse.next();
  }
 
  // 같은 출처의 HttpOnly Cookie에서 Access Token 확인
  const token = request.cookies.get("access_token")?.value;
 
  if (!token) {
    return unauthorized(request);
  }
 
  try {
    // JWT 검증 (jose - Edge Runtime 호환)
    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");
    }
 
    // 사용자 정보를 요청 헤더에 실어 다운스트림으로 전달
    // (response.headers에 넣으면 브라우저로 노출되고 서버 컴포넌트에는 닿지 않는다)
    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" });
 
  // 401이면 Refresh Token으로 갱신 시도
  if (response.status === 401) {
    const refreshResponse = await fetch("/api/auth/refresh", {
      method: "POST",
      credentials: "include", // HttpOnly Cookie 전송
    });
 
    if (refreshResponse.ok) {
      response = await fetch(url, { ...options, credentials: "include" });
    } else {
      // Refresh도 실패 → 로그인 페이지로
      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 부분 — JWT 검증 로직과 한 함수로 합쳐 사용)
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",  # 개발 환경
    ],
    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분
RATE_LIMIT_MAX = 5         # 5회
 
 
async def check_login_rate_limit(request: Request):
    """로그인 시도 횟수 제한 (IP 기반)."""
    client_ip = request.client.host if request.client else "unknown"
    now = datetime.now(timezone.utc)
 
    # 윈도우 밖의 기록 정리
    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)

영향 버전: Next.js < 15.2.3, < 14.2.25, < 13.5.9
심각도: Critical

pnpm add next@latest  # 15.2.3 이상

// 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)
 
    # 보안
    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=<AUTH_JWT_SECRET과 동일한 값>