FastAPI와 JWT를 활용한 로그인 시스템 구현

 

이번 글에서는 Python으로 웹 어플리케이션 개발을 쉽게 만들어주는 FastAPI와 JWT(Json Web Tokens)를 이용하여 로그인 시스템을 구현하는 방법을 알아보겠습니다.

 

FastAPI란?

FastAPI는 Python으로 작성된 웹 어플리케이션을 빠르고 쉽게 만들어주는 웹 프레임워크입니다. 기존에 Flask와 Django가 있었는데, FastAPI는 이들의 장점을 모아 놓은 것으로 자동 문서화, 빠른 속도, 강력한 타입 힌팅 등이 있습니다. 또한, 비동기 처리를 지원해주어 I/O 바운드 작업에서 높은 성능을 발휘합니다.

 

JWT란?

JWT는 JSON Web Tokens의 약자로, 웹 어플리케이션에서 사용자 인증을 위해 많이 사용되는 토큰 기반 인증 방식입니다. JWT는 클라이언트가 서버에게 로그인 요청을 보내면 서버는 해당 사용자를 인증하고, 인증된 사용자에게 JWT 토큰을 발급합니다. 이후, 해당 토큰을 사용자가 갖고 있으면 인증된 상태로 서버에게 요청을 보낼 수 있게 됩니다.

 

FastAPI와 JWT를 이용한 로그인 시스템 구현

1. FastAPI 설치하기

가상환경을 만들고, 다음 명령어를 통해 FastAPI를 설치합니다.

pip install fastapi

2. JWT 설치하기
다음 명령어를 통해 JWT를 설치합니다.

pip install pyjwt

 

3. JWT Secret Key 설정하기
JWT를 사용하기 위해서는 secret key가 필요합니다. 이 secret key는 서버에서만 알고 있어야 하므로, 환경 변수에 등록하는 것이 좋습니다. 다음과 같이 JWT_SECRET_KEY를 환경 변수에 등록해줍니다.

import os

os.environ["JWT_SECRET_KEY"] = "mysecretkey"

 

 

4. 사용자 인증 API 만들기
다음은 사용자 인증 API를 만들어 보겠습니다.

from fastapi import FastAPI, HTTPException
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from passlib.context import CryptContext
from jose import jwt
from datetime import datetime, timedelta
import os

app = FastAPI()
security = HTTPBasic()
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = "HS256"
JWT_EXP_DELTA_SECONDS = 20

users = {
    "user1": {
        "username": "user1",
        "full_name": "User One",
        "email": "user1@example.com",
        "hashed_password": "$2b$12$nnbNsw/kSWoAXJgYIvMG8u5O7jP6/n5B6m5B.8cW6j5U5MokU9d5m"
    },

    "user2": {
        "username": "user2",
        "full_name": "User Two",
        "email": "user2@example.com",
        "hashed_password": "$2b$12$D4KptNl9tT.y84vT8aznS.rSf.TZB/mbxsLDpV8I.N0fZiYro9XoO"
    }
}

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def authenticate_user(username: str, password: str):
    if username not in users:
        return False
    user = users[username]
    if not verify_password(password, user["hashed_password"]):
        return False
    return user

def create_access_token(data: dict, expires_delta: timedelta):
    to_encode = data.copy()
    expire = datetime.utcnow() + expires_delta
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
    return encoded_jwt

@app.post("/token")
async def login(form_data: HTTPBasicCredentials = Depends(security)):
    user = authenticate_user(form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=400, detail="Incorrect username or password"
        )
    access_token_expires = timedelta(seconds=JWT_EXP_DELTA_SECONDS)
    access_token = create_access_token(
        data={"sub": user["username"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

 

여기서 사용되는 함수들을 간단히 살펴보겠습니다.

• verify_password: 입력받은 plain password와 해시된 password를 비교하여 인증을 수행합니다.
• get_password_hash: password를 해시하여 반환합니다.
• authenticate_user: 입력받은 username과 password가 일치하는지 확인하여, 일치하면 해당 사용자 정보를 반환합니다.
• create_access_token: JWT access token을 생성합니다.
• login: HTTPBasicCredentials를 이용하여 사용자 인증을 수행하고, 인증이 성공하면 create_access_token 함수를 이용하여 access token을 발급합니다.

 

5. access token 사용하기
이제 access token을 발급하고, 해당 token을 이용하여 보호된 엔드포인트에 요청하는 방법을 알아보겠습니다.

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from datetime import datetime, timedelta
import os

app = FastAPI()
security = HTTPBearer()

JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = "HS256"
JWT_EXP_DELTA_SECONDS = 20

def verify_token(token: str):
    try:
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid authentication token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid authentication token")
    return username

@app.get("/protected")
async def protected_route(username: str = Depends(security)):
    user = verify_token(username)
    return {"user": user, "message": "This is a protected route!"}

 

5. 여기서 사용되는 함수들을 간단히 살펴보겠습니다.

verify_token: access token을 검증하고, 검증이 성공하면 해당 token에 포함된 사용자 정보를 반환합니다.

protected_route: HTTPBearer를 이용하여 access token을 검증하고, 검증이 성공하면 해당 사용자 정보를 반환합니다.

위와 같이 작성된 코드는 "/token" 엔드포인트를 통해 access token을 발급하고, "/protected" 엔드포인트를 통해 해당 token을 이용하여 인증된 사용자만 접근할 수 있는 보호된 라우트를 구현할 수 있습니다.

 

 

6. 로그아웃 API 만들기
로그아웃 기능은 access token을 무효화하면 구현할 수 있습니다. 이를 위해 사용자 정보에 access token의 만료 시간을 추가하고, 로그아웃 API에서 해당 정보를 삭제하면 됩니다.

from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from datetime import datetime, timedelta
import os

app = FastAPI()
security = HTTPBearer()

JWT_SECRET_KEY = os.environ["JWT_SECRET_KEY"]
JWT_ALGORITHM = "HS256"
JWT_EXP_DELTA_SECONDS = 20

user_db = {
    "user1": {
        "password": "password1",
        "access_token": "",
        "exp": datetime.utcnow() - timedelta(minutes=30)
    },
    "user2": {
        "password": "password2",
        "access_token": "",
        "exp": datetime.utcnow() - timedelta(minutes=30)
    }
}

def verify_token(token: str):
    try:
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[JWT_ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise HTTPException(status_code=401, detail="Invalid authentication token")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid authentication token")
    return username

def create_token(username: str, exp: int) -> str:
    payload = {
        "sub": username,
        "exp": datetime.utcnow() + timedelta(seconds=exp)
    }
    token = jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
    return token

@app.post("/token")
async def login(username: str, password: str):
    if username not in user_db:
        raise HTTPException(status_code=401, detail="Invalid username")
    if user_db[username]["password"] != password:
        raise HTTPException(status_code=401, detail="Invalid password")
    exp = JWT_EXP_DELTA_SECONDS
    access_token = create_token(username, exp)
    user_db[username]["access_token"] = access_token
    user_db[username]["exp"] = datetime.utcnow() + timedelta(seconds=exp)
    return {"access_token": access_token, "token_type": "bearer"}

@app.delete("/logout")
async def logout(token: str = Depends(security)):
    username = verify_token(token)
    user_db[username]["access_token"] = ""
    user_db[username]["exp"] = datetime.utcnow() - timedelta(minutes=30)
    return {"message": "Successfully logged out"}

 

위 코드에서는 사용자 정보에 access_token과 exp 필드를 추가하고, logout 함수에서 해당 필드를 삭제합니다. exp 필드는 access token의 만료 시간을 나타내며, 삭제 후 exp 필드가 현재 시간보다 이전인 경우 해당 사용자는 더 이상 인증되지 않습니다.

이제 "/token" 엔드포인트를 통해 access token을 발급하고, "/protected" 엔드포인트를 통해 해당 token을 이용하여 보호된 라우트에 접근할 수 있습니다. 또한 "/logout" 엔드포인트를 통해 로그아웃을 할 수 있습니다.

결론

FastAPI와 JWT를 이용하여 간단한 로그인 시스템을 구현해보았습니다. JWT를 이용한 인증 시스템은 쉽게 구현할 수 있고, FastAPI와 함께 사용하면 더욱 빠르고 안전한 웹 애플리케이션을 구축할 수 있습니다.

하지만 보안 취약점이 존재할 수 있으므로, 실제 프로젝트에서는 보안 전문가의 지도를 받아보는 것이 좋습니다. 또한, 비밀번호를 안전하게 저장하기 위해 해싱 알고리즘을 사용하거나, SSL 인증서를 적용하여 통신을 암호화하는 등의 보안 조치를 추가해야 합니다.

이번 블로그 글에서는 간단한 예제를 통해 FastAPI와 JWT를 이용한 로그인 시스템을 구현해보았습니다. 이를 기반으로 더욱 다양한 기능을 추가하고, 안전한 웹 애플리케이션을 구축해보시기 바랍니다.