FastAPI로 JWT 인증 구현하기: 실무 경험 공유

on

현대 웹 애플리케이션 개발에서 사용자 인증은 핵심적인 요소입니다. 특히 마이크로서비스 아키텍처가 대세인 지금, 토큰 기반 인증 방식은 필수적입니다. 이 중에서도 JWT(JSON Web Token)는 간결하면서도 강력한 인증 메커니즘을 제공합니다. 오늘은 제가 최근 프로젝트에서 FastAPI와 JWT를 활용해 구현한 인증 시스템에 대한 경험과 노하우를 공유하고자 합니다.

FastAPI로 JWT 인증 구현하기: 실무 경험 공유

JWT란 무엇인가?

JWT는 JSON Web Token의 약자로, 당사자 간 정보를 안전하게 JSON 객체로 전송하기 위한 개방형 표준(RFC 7519)입니다. 디지털 서명이 적용되어 있어 정보의 신뢰성을 검증할 수 있습니다. JWT는 크게 세 부분으로 구성됩니다:

  • Header(헤더): 토큰 유형과 사용된 암호화 알고리즘 정보 포함
  • Payload(페이로드): 클레임(claim)이라 불리는 사용자 및 추가 데이터 포함
  • Signature(서명): 토큰이 변조되지 않았음을 확인하는 서명

JWT의 가장 큰 장점은 서버 측에서 별도의 세션 저장소가 필요 없다는 점입니다. 모든 필요한 정보가 토큰 자체에 포함되어 있고, 서명을 통해 무결성이 보장되기 때문입니다.


FastAPI와 JWT 인증의 결합

FastAPI는 Python의 최신 웹 프레임워크로, 높은 성능과 간결한 코드, 자동 문서화 기능이 특징입니다. 이번 글에서는 FastAPI에서 JWT 인증을 구현하는 방법을 단계별로 설명하겠습니다.

1. 필요한 라이브러리 설치

먼저, 필요한 패키지들을 설치합니다.

pip install fastapi uvicorn python-jose[cryptography] passlib[bcrypt] python-multipart

각 패키지의 역할은 다음과 같습니다:

  • fastapi: 웹 API 프레임워크
  • uvicorn: ASGI 서버 구현체
  • python-jose: JWT 토큰 생성 및 검증
  • passlib: 패스워드 해싱
  • python-multipart: 폼 데이터 처리

2. 프로젝트 구조 설정

실제 프로젝트에서는 코드의 모듈화가 중요합니다. 다음과 같은 구조로 프로젝트를 구성했습니다:

my_project/
├── app/
│   ├── __init__.py
│   ├── main.py            # FastAPI 애플리케이션 인스턴스
│   ├── models/            # 데이터베이스 모델
│   │   ├── __init__.py
│   │   └── user.py        # 사용자 모델
│   ├── schemas/           # Pydantic 모델(스키마)
│   │   ├── __init__.py
│   │   ├── token.py       # 토큰 관련 스키마
│   │   └── user.py        # 사용자 관련 스키마
│   ├── crud/              # CRUD 작업
│   │   ├── __init__.py
│   │   └── user.py        # 사용자 CRUD
│   ├── api/               # API 라우터
│   │   ├── __init__.py
│   │   ├── deps.py        # 종속성(의존성) 함수
│   │   └── endpoints/     # 엔드포인트별 라우터
│   │       ├── __init__.py
│   │       ├── auth.py    # 인증 관련 엔드포인트
│   │       └── users.py   # 사용자 관련 엔드포인트
│   ├── core/              # 핵심 설정
│   │   ├── __init__.py
│   │   ├── config.py      # 설정 관리
│   │   └── security.py    # 보안 유틸리티 함수
│   └── db/                # 데이터베이스 설정
│       ├── __init__.py
│       └── session.py     # DB 세션
└── requirements.txt       # 의존성 패키지 목록

3. 보안 유틸리티 함수 구현

JWT 인증을 위한 핵심 보안 기능을 구현합니다. core/security.py 파일에 다음과 같은 코드를 작성했습니다:

from datetime import datetime, timedelta
from typing import Any, Union, Optional

from jose import jwt
from passlib.context import CryptContext

# 비밀 키와 알고리즘 설정
SECRET_KEY = "생성한_복잡한_시크릿_키"  # 실제 프로덕션에서는 환경 변수로 관리
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30  # 토큰 만료 시간

# 패스워드 해싱을 위한 컨텍스트
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(
    subject: Union[str, Any], expires_delta: Optional[timedelta] = None
) -> str:
    """액세스 토큰 생성"""
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    
    # JWT 페이로드
    to_encode = {"exp": expire, "sub": str(subject)}
    # 토큰 생성
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

이 코드에서 중요한 점은 실제 프로덕션 환경에서는 SECRET_KEY를 환경 변수로 관리해야 한다는 것입니다. 소스 코드에 직접 비밀 키를 넣는 것은 보안상 위험합니다.

4. 사용자 인증 관련 스키마 정의

Pydantic을 사용하여 입출력 데이터의 유효성을 검사하는 스키마를 정의합니다. schemas/token.py와 schemas/user.py 파일을 다음과 같이 작성했습니다:

# schemas/token.py
from typing import Optional
from pydantic import BaseModel

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenPayload(BaseModel):
    sub: Optional[int] = None

# schemas/user.py
from typing import Optional
from pydantic import BaseModel, EmailStr

class UserBase(BaseModel):
    email: Optional[EmailStr] = None
    is_active: Optional[bool] = True

class UserCreate(UserBase):
    email: EmailStr
    password: str

class UserInDB(UserBase):
    id: Optional[int] = None

    class Config:
        orm_mode = True

class User(UserInDB):
    pass

5. 의존성 함수 구현

FastAPI에서는 의존성 주입을 사용하여 코드를 깔끔하게 유지할 수 있습니다. api/deps.py 파일에 토큰 검증 및 현재 사용자를 가져오는 의존성 함수를 구현했습니다:

from typing import Generator, Optional

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.security import ALGORITHM, SECRET_KEY
from app.db.session import SessionLocal
from app.models.user import User
from app.schemas.token import TokenPayload

# OAuth2 패스워드 흐름 설정
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")

def get_db() -> Generator:
    """
    데이터베이스 세션을 제공하는 의존성 함수
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(
    db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
    """
    JWT 토큰으로부터 현재 인증된 사용자를 가져오는 의존성 함수
    """
    # 토큰이 유효하지 않을 경우 예외 발생
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="토큰 인증에 실패했습니다",
        headers={"WWW-Authenticate": "Bearer"},
    )
    
    try:
        # 토큰 디코딩
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
            
        token_data = TokenPayload(sub=user_id)
    except JWTError:
        raise credentials_exception
        
    # 사용자 정보 조회
    user = db.query(User).filter(User.id == token_data.sub).first()
    if user is None:
        raise credentials_exception
    if not user.is_active:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="비활성화된 사용자입니다",
        )
    return user

6. 인증 엔드포인트 구현

api/endpoints/auth.py 파일에 로그인 엔드포인트를 구현했습니다:

from datetime import timedelta
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app.api.deps import get_db
from app.core.security import (
    create_access_token,
    verify_password,
    ACCESS_TOKEN_EXPIRE_MINUTES,
)
from app.models.user import User
from app.schemas.token import Token

router = APIRouter()

@router.post("/login", response_model=Token)
def login_access_token(
    db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()
) -> Any:
    """
    OAuth2 호환 토큰 로그인, JWT 액세스 토큰 발급
    """
    # 이메일로 사용자 조회
    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_400_BAD_REQUEST,
            detail="잘못된 이메일 또는 비밀번호입니다",
        )
    
    # 토큰 만료 시간 설정
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    
    # 액세스 토큰 생성
    return {
        "access_token": create_access_token(
            user.id, expires_delta=access_token_expires
        ),
        "token_type": "bearer",
    }

7. 보호된 사용자 엔드포인트 구현

api/endpoints/users.py 파일에 인증된 사용자만 접근할 수 있는 엔드포인트를 구현했습니다:

from typing import Any

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session

from app.api.deps import get_current_user, get_db
from app.models.user import User
from app.schemas.user import User as UserSchema

router = APIRouter()

@router.get("/me", response_model=UserSchema)
def read_users_me(
    current_user: User = Depends(get_current_user),
) -> Any:
    """
    현재 인증된 사용자 정보 조회
    """
    return current_user

8. 라우터 등록 및 애플리케이션 구성

마지막으로, main.py 파일에서 모든 라우터를 등록하고 FastAPI 애플리케이션을 구성합니다:

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.api.endpoints import auth, users

app = FastAPI(
    title="FastAPI JWT 인증 예제",
    description="FastAPI를 이용한 JWT 인증 구현",
    version="1.0.0",
)

# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 프로덕션에서는 특정 오리진만 허용
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 라우터 등록
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(users.router, prefix="/api/users", tags=["users"])

@app.get("/")
def read_root():
    return {"message": "FastAPI JWT 인증 예제에 오신 것을 환영합니다!"}

if __name__ == "__main__":
    import uvicorn
    
    uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

실무 적용 시 고려사항

위의 코드는 JWT 인증의 기본 구현 방법을 보여주지만, 실제 프로덕션 환경에서는 몇 가지 추가적인 고려사항이 있습니다:

토큰 갱신(Refresh Token) 메커니즘

액세스 토큰의 수명은 보안상 짧게 유지하는 것이 좋습니다. 하지만 사용자 경험을 위해 리프레시 토큰을 구현하여 액세스 토큰이 만료되었을 때 새로운 토큰을 자동으로 발급받을 수 있게 했습니다:

# schemas/token.py에 추가
class TokenRefresh(BaseModel):
    refresh_token: str

# core/security.py에 추가
REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_refresh_token(
    subject: Union[str, Any], expires_delta: Optional[timedelta] = None
) -> str:
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    
    to_encode = {"exp": expire, "sub": str(subject), "type": "refresh"}
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# api/endpoints/auth.py에 추가
@router.post("/refresh", response_model=Token)
def refresh_token(
    db: Session = Depends(get_db), refresh_token: TokenRefresh = Body(...)
) -> Any:
    try:
        payload = jwt.decode(
            refresh_token.refresh_token, SECRET_KEY, algorithms=[ALGORITHM]
        )
        
        # 리프레시 토큰이 아닌 경우 에러
        if payload.get("type") != "refresh":
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="유효하지 않은 토큰 유형입니다",
            )
            
        user_id: str = payload.get("sub")
        if user_id is None:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="유효하지 않은 인증 정보입니다",
            )
            
        user = db.query(User).filter(User.id == user_id).first()
        if user is None:
            raise HTTPException(
                status_code=status.HTTP_404_NOT_FOUND,
                detail="사용자를 찾을 수 없습니다",
            )
            
        # 새 액세스 토큰 생성
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        return {
            "access_token": create_access_token(
                user_id, expires_delta=access_token_expires
            ),
            "token_type": "bearer",
        }
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="유효하지 않은 토큰입니다",
            headers={"WWW-Authenticate": "Bearer"},
        )

토큰 폐기 및 블랙리스트

JWT는 상태가 없는(stateless) 인증 방식이기 때문에, 발급된 토큰을 무효화하기 어렵습니다. 이를 해결하기 위해 Redis를 사용한 토큰 블랙리스트를 구현했습니다:

# Redis 클라이언트 설정
import redis
from app.core.config import settings

redis_client = redis.Redis(
    host=settings.REDIS_HOST,
    port=settings.REDIS_PORT,
    db=settings.REDIS_DB,
    decode_responses=True,
)

# 로그아웃 시 토큰 블랙리스트에 추가
@router.post("/logout")
def logout(token: str = Depends(oauth2_scheme)) -> Any:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # 토큰 만료 시간 계산
        exp = payload.get("exp")
        jti = payload.get("jti", "")  # 토큰 식별자(JWT ID)
        
        # 토큰 블랙리스트에 추가 (만료 시간까지만 저장)
        token_key = f"blacklist:{jti}"
        redis_client.set(token_key, "1")
        redis_client.expireat(token_key, exp)
        
        return {"message": "로그아웃 성공"}
    except JWTError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="유효하지 않은 토큰입니다",
        )

# deps.py의 get_current_user 함수 수정
def get_current_user(
    db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
) -> User:
    # 기존 코드...
    
    try:
        # 토큰 디코딩
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        jti = payload.get("jti", "")
        
        # 블랙리스트 확인
        if redis_client.exists(f"blacklist:{jti}"):
            raise credentials_exception
            
        # 나머지 기존 코드...
    except JWTError:
        raise credentials_exception

추가 보안 조치

실제 프로덕션 환경에서는 다음과 같은 추가적인 보안 조치를 적용했습니다:

  • Rate Limiting: 로그인 시도 횟수 제한
  • HTTPS 강제: 모든 API 요청 암호화
  • CSRF 보호: 교차 사이트 요청 위조 방지
  • JTI(JWT ID): 토큰에 고유 식별자 추가
  • 토큰 도난 감지: IP 및 User-Agent 정보 확인

끝으로

이 글에서는 FastAPI에서 JWT를 이용한 인증 시스템을 구현하는 방법에 대해 살펴보았습니다. 단순한 구현을 넘어 실제 프로덕션 환경에서 필요한 추가적인 고려사항들도 함께 다루었습니다.

JWT는 사용자 인증에 널리 사용되는 강력한 도구이지만, 모든 상황에 적합한 것은 아닙니다. 프로젝트의 요구사항과 보안 정책에 맞게 적절한 인증 메커니즘을 선택하는 것이 중요합니다.

실무에서 FastAPI와 JWT를 결합한 인증 시스템은 빠른 개발 속도와 높은 성능을 제공하면서도, 안전하고 유지보수가 용이한 코드를 작성할 수 있게 해주었습니다. 이 글이 여러분의 프로젝트에 도움이 되길 바랍니다.

Comments

Post a Comment