JWT 토큰을 이용해서 액세스 토큰 구현해보기

개인 프로젝트를 진행하며 JWT 토큰을 활용해 로그인 기능을 구현해봤습니다.

FastAPI + Python으로 진행했습니다.

1. 로그인 라우터 만들기

OAuth2PasswordRequestForm에서 사용할 username과 password를 login서비스의 매개변수로 입력합니다.

from fastapi import APIRouter, Depends
from fastapi.security import OAuth2PasswordRequestForm
from db.post.login import login

router = APIRouter()


@router.post("/login", tags=["login"])
def login_jwt(form_data: OAuth2PasswordRequestForm = Depends()):
    result = login(form_data.username, form_data.password)
    return result

OAuth2PasswordRequestForm

2. 로그인  서비스 만들기

일반적으로 로그인 후에 액세스 토큰이 발급되는 방식이 일반적이라고 하는데,
어차피 로그인이 완료되었다는 정보를 받고 토큰이 발급되나, 로그인 함수 내에서 ID와 PW가 인증이 됨과 동시에 토큰을 발급하는게 어떤 차이가 있는지는 잘 모르겠습니다.

from sqlalchemy.orm import sessionmaker
from db.connection import engine
from db.models import Account
from fastapi import HTTPException, status
from app.password import verify_password
from datetime import timedelta
from tokens.create_token import create_access_token
from dotenv import load_dotenv

Session = sessionmaker(bind=engine)
session = Session()

ACCESS_TOKEN_EXPIRE_MINUTES = 30

load_dotenv()


def login(user_id, pwd):
    try:
        user = session.query(Account).filter_by(user_id=user_id, permission=True).first()
        if not user:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="아이디를 다시 확인하세요")

        if not verify_password(pwd, user.pwd):
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="비밀번호가 틀렸습니다.")

        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(
            data={"user_id": user.user_id},
            expires_delta=access_token_expires
        )

        return {
            "access_token": access_token,
            "token_type": "bearer",
            "user_id": user.user_id
        }

    except HTTPException as err:
        raise err

    except Exception as err:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(err))

    finally:
        session.close()

3.액세스 토큰 만들기

2번에서 받아온 data와 expires_delta를 매개변수로 갖는 create_access_token()함수를 만들었습니다.

payload 안에 {"user_id": user.user_id, "exp": expire}가 들어있게 되고 인코딩을 거쳐 토큰을 반환합니다.

from datetime import datetime, timedelta
from dotenv import load_dotenv
from jose import jwt
import os

load_dotenv()

SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = "HS256"


def create_access_token(data: dict, expires_delta: timedelta):
    payload = data.copy()
    expire = datetime.utcnow() + expires_delta
    payload.update({"exp": expire})
    encode_jwt = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return encode_jwt

4. 액세스 토큰을 사용하여 API 접근 권한 부여하기

종속성 부여를 통해 API에 접근할 수 있는지 체크하는 기능입니다

from fastapi import APIRouter, Depends
from db.post.route_list import add_route_list
from app.auth import get_admin

router = APIRouter(prefix="/route")


@router.post('/route-list', tags=["노선"], dependencies=[Depends(get_admin)])
def route_list():
    result = add_route_list()
    return result