JWT 인증 FastAPI: 코드 및 스키마

FastAPI에서 JWT 인증 구현하기: 코드 포함 완전 가이드

JWT(JSON 웹 토큰)는 API에서 상태 없는 인증을 가능하게 합니다. 토큰은 헤더, 페이로드, 서명으로 구성되며, base64로 인코딩되어 점(.)으로 구분됩니다.

헤더는 서명 알고리즘을 지정합니다:

{
  "alg": "HS256",
  "typ": "JWT"
}

페이로드는 사용자 ID(sub), 발급 시간(iat), 만료 시간(exp) 등의 클레임을 담습니다.

{
  "sub": "user_id_123",
  "exp": 1735689600,
  "iat": 1735603200
}

서명은 비밀 키를 사용해 HMAC-SHA256으로 생성되어 데이터 무결성을 보장합니다.

실제 서비스에서는 이중 토큰 시스템을 권장합니다:

• 액세스 토큰(15~30분) — API 요청에 사용
• 리프레시 토큰(7~30일) — 액세스 토큰 갱신용

프로젝트 아키텍처

계층 구조는 테스트와 확장성을 쉽게 만들어줍니다:

project/
├── app/
│   ├── core/          # config.py, security.py, dependencies.py
│   ├── models/        # user.py (SQLAlchemy)
│   ├── schemas/       # user.py, token.py (Pydantic)
│   ├── routers/       # auth.py, users.py
│   ├── services/      # user.py
│   └── main.py
├── requirements.txt
└── .env

라우터는 HTTP 로직을 처리하고, 스키마는 입력값을 검증하며, 모델은 데이터베이스 구조를 정의하고, 서비스에는 비즈니스 로직이 포함됩니다.

설정 및 보안 유틸리티

core/config.py 파일:

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    SECRET_KEY: str = "your-secret-key-change-in-production"
    ALGORITHM: str = "HS256"
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    REFRESH_TOKEN_EXPIRE_DAYS: int = 7
    
    class Config:
        env_file = ".env"

settings = Settings()

core/security.py에 핵심 함수들:

from datetime import datetime, timedelta
from jose import JWTError, jwt
from passlib.context import CryptContext
from .config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None) -> str:
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def create_refresh_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(days=settings.REFRESH_TOKEN_EXPIRE_DAYS)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
    return encoded_jwt

def decode_token(token: str) -> dict:
    try:
        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
        return payload
    except JWTError:
        return None

Pydantic 스키마를 통한 검증

schemas/user.py:

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str
    full_name: str

class UserResponse(BaseModel):
    id: int
    email: EmailStr
    full_name: str
    
    class Config:
        from_attributes = True

schemas/token.py:

from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    refresh_token: str
    token_type: str = "bearer"

class TokenData(BaseModel):
    user_id: int

현재 사용자 의존성 설정

core/dependencies.py는 OAuth2PasswordBearer를 활용합니다:

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: Session = Depends(get_db)
) -> User:
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    payload = decode_token(token)
    if payload is None:
        raise credentials_exception
    
    user_id = payload.get("sub")
    if user_id is None:
        raise credentials_exception
    
    token_data = TokenData(user_id=int(user_id))
    user = db.query(User).filter(User.id == token_data.user_id).first()
    if user is None:
        raise credentials_exception
    
    return user

인증 엔드포인트

routers/auth.py 파일:

• POST /api/auth/register — bcrypt로 해시된 사용자 생성
• POST /api/auth/login — 액세스 토큰과 리프레시 토큰 반환
• POST /api/auth/refresh — 액세스 토큰 재발급

완전한 로그인 구현 예시:

@router.post("/login", response_model=Token)
def login(
    form_data: OAuth2PasswordRequestForm = Depends(),
    db: Session = Depends(get_db)
):
    user = db.query(User).filter(User.email == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token = create_access_token(data={"sub": str(user.id)})
    refresh_token = create_refresh_token(data={"sub": str(user.id)})
    
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer"
    }

보호된 리소스 접근

routers/users.py:

@router.get("/me", response_model=UserResponse)
def get_current_user_info(current_user: User = Depends(get_current_user)):
    return current_user

이 엔드포인트는 Authorization 헤더에 Bearer 토큰이 필요합니다.

메인 애플리케이션 설정

main.py:

from fastapi import FastAPI
from .routers import auth, users

app = FastAPI(title="FastAPI JWT Auth", version="1.0.0")

app.include_router(auth.router)
app.include_router(users.router)

@app.get("/")
def root():
    return {"message": "Welcome to FastAPI JWT Auth API"}

핵심 요약

• HS256 알고리즘과 강력한 비밀키(최소 32자 이상) 사용
• XSS 공격 방지를 위해 리프레시 토큰은 HttpOnly 쿠키에 저장
• 매번 리프레시 시 토큰 회전 기능 구현
• 실패한 인증 시도는 반드시 로그 기록
• 만료된 토큰 테스트(예: exp 클레임 확인)

— Editorial Team