FastAPI로 완벽한 블로그 만들기: 사용자 시스템 추가

이전 글에서 FastAPI를 사용하여 기본적인 개인 블로그를 만들고 성공적으로 배포했습니다.

하지만 이 블로그에는 심각한 보안 문제가 있습니다. 누구나 마음대로 글을 생성할 수 있다는 것이죠.

다음 튜토리얼에서는 이 블로그를 더 안전하게 만들기 위해 사용자 및 인증 시스템을 추가할 것입니다.

서론이 길었습니다. 시작해 봅시다.

인증 방식 소개

웹 개발에서 가장 일반적인 두 가지 인증 방식은 토큰 기반(예: JWT) 과 세션 기반(쿠키) 입니다.

• JWT (JSON Web Tokens): 현재 가장 인기 있는 인증 방식입니다. 사용자가 로그인하면 서버는 토큰을 생성하여 클라이언트로 반환합니다. 클라이언트는 후속 요청에 이 토큰을 포함하고, 서버는 토큰을 검증하여 사용자 신원을 확인하기만 하면 됩니다. 서버는 사용자 상태를 저장할 필요가 없으므로, 분산되고 수평적으로 확장 가능한 대규모 애플리케이션에 매우 적합합니다.
• 세션-쿠키: 사용자가 로그인하면 서버는 세션을 생성하고 세션 ID를 쿠키를 통해 브라우저로 반환합니다. 브라우저는 후속 요청에 이 쿠키를 자동으로 포함합니다. 서버는 세션 ID를 기반으로 해당 세션 정보를 찾아 사용자를 식별합니다.

이 튜토리얼에서는 전통적인 세션-쿠키 방식을 선택할 것입니다. 블로그가 단순한 아키텍처를 가진 모놀리스이므로, 인증에 세션-쿠키를 사용하는 것이 가장 직접적이고 고전적이며 충분히 안전한 접근 방식입니다.

1단계: 사용자 모듈 생성

인증을 처리하기 전에 사용자 시스템이 필요합니다.

1. 사용자 데이터 모델 생성

models.py 파일을 열고 Post 클래스 위에 User 모델을 추가합니다.

# models.py
import uuid
from datetime import datetime
from typing import Optional
from sqlmodel import Field, SQLModel

class User(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    username: str = Field(unique=True, index=True)
    password: str # 저장되는 비밀번호는 암호화된 해시입니다.

class Post(SQLModel, table=True):
    id: Optional[uuid.UUID] = Field(default_factory=uuid.uuid4, primary_key=True)
    title: str
    content: str
    createdAt: datetime = Field(default_factory=datetime.utcnow, nullable=False)

첫 번째 글에서 main.py의 create_db_and_tables 함수를 설정했으므로, 애플리케이션 시작 시 모든 SQLModel 모델을 자동으로 감지하고 해당 데이터베이스 테이블을 생성합니다. 따라서 SQL 문을 수동으로 실행할 필요가 없습니다.

SQL을 수동으로 실행해야 하고 데이터베이스가 Leapcell에서 생성되었다면,

그래픽 인터페이스를 사용하여 SQL 문을 쉽게 실행할 수 있습니다. 웹사이트의 데이터베이스 관리 페이지로 이동하여 위 문장들을 SQL 인터페이스에 붙여넣고 실행하기만 하면 됩니다.

2. 비밀번호 암호화 라이브러리 설치

보안을 위해 사용자 비밀번호는 절대로 일반 텍스트로 데이터베이스에 저장해서는 안 됩니다. 비밀번호를 해싱하기 위해 bcrypt 라이브러리를 사용할 것입니다.

먼저 requirements.txt 파일에 bcrypt를 추가합니다:

# requirements.txt
fastapi
uvicorn[standard]
sqlmodel
psycopg2-binary
jinja2
python-dotenv
python-multipart
bcrypt

그런 다음 설치 명령을 실행합니다:

pip install -r requirements.txt

2단계: 사용자 등록 및 유효성 검사 로직 구현

다음으로 사용자 데이터를 처리하고 비밀번호를 유효성 검사하는 함수를 만들 것입니다.

프로젝트의 루트 디렉토리에 users_service.py라는 새 파일을 만들고 사용자 관련 비즈니스 로직을 저장합니다.

# users_service.py
import bcrypt
from sqlmodel import Session, select
from models import User

def get_user_by_username(username: str, session: Session) -> User | None:
    """사용자 이름으로 사용자 찾기"""
    statement = select(User).where(User.username == username)
    return session.exec(statement).first()

def create_user(user_data: dict, session: Session) -> User:
    """새 사용자를 생성하고 비밀번호를 해싱합니다."""
    # 일반 텍스트 비밀번호를 바이트로 변환
    password_bytes = user_data["password"].encode('utf-8')
    # 솔트(salt)를 생성하고 비밀번호를 해싱
    salt = bcrypt.gensalt()
    hashed_password = bcrypt.hashpw(password_bytes, salt)
    
    new_user = User(
        username=user_data["username"],
        # 데이터베이스에 저장하기 위해 해시된 비밀번호(바이트)를 문자열로 디코딩
        password=hashed_password.decode('utf-8')
    )
    
    session.add(new_user)
    session.commit()
    session.refresh(new_user)
    return new_user

그런 다음 사용자 인증을 처리하기 위해 auth_service.py 파일을 만듭니다.

# auth_service.py
import bcrypt
from sqlmodel import Session
from models import User
from users_service import get_user_by_username

def validate_user(username: str, plain_password: str, session: Session) -> User | None:
    """사용자 이름과 비밀번호가 일치하는지 확인합니다."""
    user = get_user_by_username(username, session)
    if not user:
        return None
        
    # 입력된 일반 텍스트 비밀번호와 저장된 해시 비밀번호를 모두 바이트로 인코딩
    plain_password_bytes = plain_password.encode('utf-8')
    hashed_password_bytes = user.password.encode('utf-8')
    
    # 비교를 위해 bcrypt.checkpw 사용
    if bcrypt.checkpw(plain_password_bytes, hashed_password_bytes):
        return user # 유효성 검사 성공, 사용자 정보 반환
        
    return None # 유효성 검사 실패

3단계: 로그인 및 등록 페이지 생성

사용자가 등록하고 로그인할 수 있는 인터페이스를 제공해야 합니다. templates 폴더에 login.html과 register.html 파일을 만듭니다.

• register.html

  {% include "_header.html" %}
  <form action="/users/register" method="POST" class="post-form">
    <h2>Register</h2>
    <div class="form-group">
      <label for="username">Username</label>
      <input type="text" id="username" name="username" required />
    </div>
    <div class="form-group">
      <label for="password">Password</label>
      <input type="password" id="password" name="password" required />
    </div>
    <button type="submit">Register</button>
  </form>
  <p style="text-align: center; margin-top: 1rem;">
    이미 계정이 있으신가요? <a href="/auth/login">여기를 클릭하여 로그인</a>.
  </p>
  {% include "_footer.html" %}

• login.html

  {% include "_header.html" %}
  <form action="/auth/login" method="POST" class="post-form">
    <h2>Login</h2>
    <div class="form-group">
      <label for="username">Username</label>
      <input type="text" id="username" name="username" required />
    </div>
    <div class="form-group">
      <label for="password">Password</label>
      <input type="password" id="password" name="password" required />
    </div>
    <button type="submit">Login</button>
  </form>
  <p style="text-align: center; margin-top: 1rem;">
    계정이 없으신가요? <a href="/users/register">여기를 클릭하여 등록</a>.
  </p>
  {% include "_footer.html" %}

동시에, 상단 오른쪽 코너에 등록 및 로그인 링크를 추가하기 위해 _header.html을 업데이트합니다.

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8" />
  <meta name="viewport" content="width=device-width, initial-scale=1.0" />
  <title>{{ title }}</title>
  <link rel="stylesheet" href="/static/css/style.css" />
</head>
<body>
  <header>
    <h1><a href="/">My Blog</a></h1>
    <nav>
      <a href="/posts/new" class="new-post-btn">New Post</a>
      <a href="/users/register" class="nav-link">Register</a>
      <a href="/auth/login" class="nav-link">Login</a>
    </nav>
  </header>
  <main>

4단계: 라우팅 및 컨트롤러 로직 구현

이제 프로젝트 구조를 더 명확하게 하기 위해 라우팅 로직을 다른 파일로 분할할 것입니다.

1. 프로젝트 루트 디렉토리에 routers 폴더를 생성합니다.

2. Post 관련 라우트(@app.get("/posts", ...) 등)를 모두 main.py에서 잘라내어 routers/posts.py 파일에 붙여넣습니다.

   # routers/posts.py
   import uuid
   from fastapi import APIRouter, Request, Depends, Form
   from fastapi.responses import HTMLResponse, RedirectResponse
   from fastapi.templating import Jinja2Templates
   from sqlmodel import Session, select
   
   from database import get_session
   from models import Post
   
   router = APIRouter()
   templates = Jinja2Templates(directory="templates")
   
   @router.get("/", response_class=HTMLResponse)
   def root():
       return RedirectResponse(url="/posts", status_code=302)
   
   @router.get("/posts", response_class=HTMLResponse)
   def get_all_posts(request: Request, session: Session = Depends(get_session)):
       statement = select(Post).order_by(Post.createdAt.desc())
       posts = session.exec(statement).all()
       return templates.TemplateResponse("index.html", {"request": request, "posts": posts, "title": "Home"})
   
   @router.get("/posts/new", response_class=HTMLResponse)
   def new_post_form(request: Request):
       return templates.TemplateResponse("new-post.html", {"request": request, "title": "New Post"})
   
   @router.post("/posts", response_class=HTMLResponse)
   def create_post(
       title: str = Form(...), 
       content: str = Form(...),
       session: Session = Depends(get_session)
   ):
       new_post = Post(title=title, content=content)
       session.add(new_post)
       session.commit()
       return RedirectResponse(url="/posts", status_code=302)
   
   @router.get("/posts/{post_id}", response_class=HTMLResponse)
   def get_post_by_id(request: Request, post_id: uuid.UUID, session: Session = Depends(get_session)):
       post = session.get(Post, post_id)
       return templates.TemplateResponse("post.html", {"request": request, "post": post, "title": post.title})

   참고: @app 데코레이터 대신 @router를 사용합니다.

3. routers 폴더에 사용자 등록을 처리할 users.py를 생성합니다.

   # routers/users.py
   from fastapi import APIRouter, Request, Depends, Form
   from fastapi.responses import HTMLResponse, RedirectResponse
   from fastapi.templating import Jinja2Templates
   from sqlmodel import Session
   
   from database import get_session
   import users_service
   
   router = APIRouter()
   templates = Jinja2Templates(directory="templates")
   
   @router.get("/users/register", response_class=HTMLResponse)
   def show_register_form(request: Request):
       return templates.TemplateResponse("register.html", {"request": request, "title": "Register"})
   
   @router.post("/users/register")
   def register_user(
       username: str = Form(...),
       password: str = Form(...),
       session: Session = Depends(get_session)
   ):
       # 편의상 복잡한 유효성 검사는 없음
       users_service.create_user({"username": username, "password": password}, session)
       return RedirectResponse(url="/auth/login", status_code=302)

4. routers 폴더에 사용자 로그인을 처리할 auth.py를 생성합니다.

   # routers/auth.py
   from fastapi import APIRouter, Request, Depends, Form, HTTPException
   from fastapi.responses import HTMLResponse, RedirectResponse
   from fastapi.templating import Jinja2Templates
   from sqlmodel import Session
   
   from database import get_session
   import auth_service
   
   router = APIRouter()
   templates = Jinja2Templates(directory="templates")
   
   @router.get("/auth/login", response_class=HTMLResponse)
   def show_login_form(request: Request):
       return templates.TemplateResponse("login.html", {"request": request, "title": "Login"})
   
   @router.post("/auth/login")
   def login(
       username: str = Form(...),
       password: str = Form(...),
       session: Session = Depends(get_session)
   ):
       user = auth_service.validate_user(username, password, session)
       if not user:
           raise HTTPException(status_code=401, detail="Incorrect username or password")
       
       # 유효성 검사 성공
       return RedirectResponse(url="/posts", status_code=302)

5. 마지막으로 main.py를 업데이트하여 이전 라우트를 제거하고 새 라우터 파일을 포함합니다.

   # main.py
   from contextlib import asynccontextmanager
   from fastapi import FastAPI
   from fastapi.staticfiles import StaticFiles
   
   from database import create_db_and_tables
   from routers import posts, users, auth
   
   @asynccontextmanager
   async def lifespan(app: FastAPI):
       print("Creating tables..")
       create_db_and_tables()
       yield
   
   app = FastAPI(lifespan=lifespan)
   
   # 정적 파일 디렉토리 마운트
   app.mount("/static", StaticFiles(directory="public"), name="static")
   
   # 라우터 포함
   app.include_router(posts.router)
   app.include_router(users.router)
   app.include_router(auth.router)

5단계: 테스트

이 시점에서 기본 사용자 등록 및 로그인 유효성 검사 로직을 완성했습니다.

프로젝트를 다시 시작합니다:

uvicorn main:app --reload

http://localhost:3000/users/register로 이동하여 등록합니다.

성공적으로 등록한 후에는 자동으로 http://localhost:3000/auth/login으로 리디렉션되어 로그인하게 됩니다.

올바른 계정과 잘못된 계정 정보를 입력했을 때의 결과를 테스트할 수 있습니다. 예를 들어, 잘못된 정보를 입력하면 401 Unauthorized 오류 페이지가 표시됩니다.

하지만 현재 로그인은 일회성 유효성 검사 과정일 뿐입니다. 서버가 사용자의 로그인 상태를 "기억"하지 않습니다. 브라우저를 닫거나 다른 페이지를 방문하면 여전히 인증되지 않은 상태로 남게 됩니다.

다음 글에서는 실제 사용자 로그인 상태 지속성을 달성하고 권한에 따라 페이지 액세스 및 작업을 제한하기 위해 세션 관리를 소개할 것입니다.