[관리자 페이지] Admin Page - 회원가입 구현하기 #6 JWT 토큰 발행

https://jwt.io/

[JWT.IO

JSON Web Tokens are an open, industry standard RFC 7519 method for representing claims securely between two parties.

jwt.io](https://jwt.io/)

아이디 / 패스워드로 사용자 로그인을 성공하면 jwt 를 발급합니다.

accessToken, refreshToken 2개를 발급합니다.

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

load_dotenv(os.path.join(os.path.dirname(__file__), "..", "..", ".env"))


class CommonUtils:
    ACCESS_TOKEN_EXPIRE = os.getenv("ACCESS_TOKEN_EXPIRE")
    REFRESH_TOKEN_EXPIRE = os.getenv("REFRESH_TOKEN_EXPIRE")
    SECRET_KEY = os.getenv("SECRET_KEY")
    @staticmethod
    def create_token(user_info):
        today = datetime.now()

        access_token_expire_time = today + timedelta(seconds=float(CommonUtils.ACCESS_TOKEN_EXPIRE))
        refresh_token_expire_time = today + timedelta(seconds=float(CommonUtils.REFRESH_TOKEN_EXPIRE))

        access_token_payload = {"iat": today, "iss": "hiio420.com", "exp": access_token_expire_time,
                                "sub": "Access Token",
                                "userId": user_info.id}
        access_token = jwt.encode(access_token_payload, "secret", algorithm="HS256")
        refresh_token_payload = {"iat": today, "iss": "hiio420.com", "exp": refresh_token_expire_time,
                                 "sub": "Refresh Token",
                                 "userId": user_info.id}
        refresh_token = jwt.encode(refresh_token_payload, CommonUtils.SECRET_KEY, algorithm="HS256")
        return {"access_token": access_token, "refresh_token": refresh_token}

.env 파일을 사용합니다.

.env 파일에는 token의 만료 시간과 토큰 발행에 사용할 시크릿 문자열이 있습니다.

유틸을 만든 이유는 나중에 이부분만 다른 프로젝트에도 사용될 수 있지 않을까하는 생각에 만들었습니다.

## UserService class
...
 
    def get_token(self, user_info: UserModel) -> TokenModel:
        token = CommonUtils.create_token(user_info)
        return TokenModel(accessToken=token["access_token"], refreshToken=token["refresh_token"])

UserService에 get_token을 만들고 user_info를 파라미터로 전달합니다.
get_token은 TokenModel을 반환합니다.

from pydantic import BaseModel, Field


class TokenModel(BaseModel):
    accessToken: str = Field("", title="Access Token")
    refreshToken: str = Field("", title="Refresh Token")

@api_main.post("/signin",response_model=TokenModel)
def signin(user: UserWithPasswordModel):
    user: UserModel = userService.sign_in(user)
    return userService.get_token(user)

[관리자 페이지] Admin Page - 회원가입 구현하기 #5 비밀번호 체크

유저의 id와 암호환 된 password를 DB에 저장했습니다.
이제 유저가 로그인하기 위해 id와 password를 post로 넘기면 해당 id와 password를 가진 user를 찾아 반환합니다.

db model 인 User classdp validate_password를 정의해 줍니다.

class User(Base):
    __tablename__ = "USER"
    __table_args__ = {
        'comment': 'USER TABLE'
    }
    id = Column("USER_ID", String(20), primary_key=True, comment="USER ID")
    password = Column("PASSWORD", String(120), nullable=False, comment="USER PASSWORD")

    def validate_password(self, password):
        return bcrypt.checkpw(password.encode("utf-8"), self.password.encode("utf-8"))

bcrypt의 checkpw를 통해서 encode된 password와 db에서 가져온 password를 비교해 True / False를 반환합니다.

UserService에 sign_in이라는 메소드를 만들어 줍니다.

    def sign_in(self, user: UserWithPasswordModel) -> UserModel:

        # select user by id
        selected_user: Optional[User] = self.select_user_by_id(user.id)
        if selected_user is None:
            raise HTTPException(status_code=404, detail="User does not exist")

        # valid passwrod
        is_password_ok = selected_user.validate_password(user.password)
        if not is_password_ok:
            raise HTTPException(status_code=404, detail="Password incorrect")

        return UserModel(**selected_user.__dict__)

User가 없을 경우 404 / password가 맞지않을 경우 404를 에러가 발생하고,

모두 통과하면 password 속성이 없는 UserModel을 반환합니다.

[관리자 페이지] Admin Page - 회원가입 구현하기 #4 비밀번호 암호화

사용자의 정보는 id 와 password입니다. 이 password를 DB에 입력하기 전에 암호화 해서 넣어 줍니다.

python에서 암호화 모듈은 bcrypt를 사용합니다.

pip install bcrypt

암호화 하기 이전에 UserMoodel을 2개의 class로 나누어 줍니다.
기존의 UserModel에 id만을 남기고 password는 UserWithPasswordModel class에 옮겨주고 이 class는 UserModel을 상속받습니다.

# api/models/user.py
from typing import Optional

import bcrypt
from pydantic import BaseModel, Field


class UserModel(BaseModel):
    id: Optional[str] = Field(None, description="user's id")

    class Config:
        from_attritues = True


class UserWithPasswordModel(UserModel):
    password: Optional[str] = Field(None, description="user's password")

    class Config:
        from_attritues = True

암호화를 실행하는 method를 UserWithPasswordModel에 정의합니다.

class UserWithPasswordModel(UserModel):
    password: Optional[str] = Field(None, description="user's password")

    class Config:
        from_attritues = True

    def hash_password(self):
        self.password = bcrypt.hashpw(self.password.encode("utf-8"), bcrypt.gensalt()).decode("utf-8")

hash password 메소드를 실행하면 UserWithPasswordModel에서 받은 password는 암호화 되어 집니다.

UserService insert_user 에 hash_password 메소드를 실행하는 코드를 추가합니다.

    def insert_user(self, user: UserWithPasswordModel) -> bool:
        try:
            user.hash_password()
            self.db.add(User(**user.dict()))
            self.db.commit()
            return True
        except Exception as e:
            self.db.rollback()
            return False

암호화된 비밀번호가 insert 된것을 볼 수 있습니다.

[관리자 페이지] Admin Page - 회원가입 구현하기 #3 Insert User Data

현재 user가 id와 password만 api에 제공하면 데이터가 insert 되야 합니다.

이 과정을 진행 하기 위해서는 sqlAlchemy에서 session을 얻어와 FastAPI DI를 이용해 실행되는 api method에 주입시켜 줘야합니다.

먼저 session을 얻어오는 함수를 core.py에 작성해 줍니다.

# database/core.py
from typing import Generator

from sqlalchemy import URL, create_engine
from sqlalchemy.orm import sessionmaker, Session

url = URL.create(drivername="mariadb+mariadbconnector", username="hiio420", password="...", host="...",
                 port="...", database="HIIO_ADMIN")

engine = create_engine(url)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

def get_db() -> Generator[Session, None, None]:
    with SessionLocal() as session:
        yield session

sessionmaker를 통해 SessionLocal을 정의하고 get_db 함수를 만들어 with 과 yield 를 이용해 session을 이용합니다.

https://docs.python.org/ko/dev/reference/expressions.html#yield-expressions

이제 signup과 signin 함수에 di를 하기 위해 파라미터 값으로 Depends를 이용해 넘겨 줍니다.

# api/api_main.py
from typing import Optional

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

from api.models import UserModel
from database import get_db, User

api_main = APIRouter()


@api_main.post("/signin")
def signin(user:UserModel,session:Session = Depends(get_db)):
    return {"accessToken": "accessToken", "refreshToken": "refreshToken"}


@api_main.post("/signup")
def signup(user:UserModel,session:Session = Depends(get_db)):
    return {"message": "success"}

db에 user 정보를 저장하는 service 를 하나 만들어 줍니다.

.
┣━api
┣━database
┣━service
┃  ┣━__init__.py
┃  ┣━user.py

root 디렉토리에 service directory를 만들고 그안이 init 파일과 user 파일을 만들어 줍니다.

UserService class를 만들어 주는데 생성자로 db Genertor 객체를 받는(get_db)서비스를 만들어 주고
method로 insert_user를 정의해 줍니다.

# service/user.py
from typing import Generator

from sqlalchemy import insert
from sqlalchemy.orm import Session

from api.models import UserModel
from database import User


class UserService:

    def __init__(self, db: Generator[Session, None, None]):
        self.i = 0
        self.db = next(db)

    def insert_user(self, user: UserModel) -> bool:
        try:
            self.db.add(User(**user.dict()))
            self.db.commit()
            return True
        except Exception as e:
            self.db.rollback()
            return False

이제 __init__.py 에 UserService를 초기화 하여 userService 변수에 할당하고 all을 이용해서 다른 파일에서도 사용할 수 있게끔 해줍니다.

# service/__init__.py
from database import get_db
from .user import UserService

userService = UserService(get_db())

__all__ = ['userService']

api_main.py 파일의 signup 함수에 userSerivce의 insert_user를 호출합니다.

# api/api_main.py

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

from api.models import UserModel
from database import get_db
from service import userService
api_main = APIRouter()


@api_main.post("/signin")
def signin(user: UserModel, session: Session = Depends(get_db)):
    return {"accessToken": "accessToken", "refreshToken": "refreshToken"}


@api_main.post("/signup")
def signup(user: UserModel):
    status = userService.insert_user(user)
    return {"message": f"Insert {status}"}

서버를 실행하고 docs 다시 접속 해봅니다.

데이터가 잘 된것을 볼 수 있습니다.

[관리자 페이지] Admin Page - 회원가입 구현하기 #2 API Router

[관리자 페이지] Admin Page - 회원가입 구현하기 #1 User Table 생성

Fast API는 include_router 를 통해서 api router 를 등록하여 prefix, global DI 등을 사용할 수 있습니다.

pip install fastapi

fastapi를 설치해 주고

root 디렉토리의 main.py 의 코드를 아래와 같이 수정합니다.

# main.py
from fastapi import FastAPI

app = FastAPI()

CLI 또는 IDE Run 으로 실행하고 localhost:4883/docs 로 접속해보면

화면을 만날 수 있습니다.

1. API 목록

현재 필요한 api 목록은 최소 2개입니다.

api 디렉토리를 만들어 주고 api_main 이라는 파일을 만들어 router를 정의해 줍니다.

.
┣━api
┃  ┣━__init__.py
┃  ┣━api_main.py
┣━database

# api/api_main.py
from typing import Optional

from fastapi import APIRouter

api_main = APIRouter()


@api_main.post("/signin")
def signin(userId: Optional[str], userPw: Optional[str]):
    return {"accessToken": "accessToken", "refreshToken": "refreshToken"}


@api_main.post("signup")
def signup(userId: Optional[str], userPw: Optional[str]):
    return {"message": "success"}

main.py 에는 api_main을 import 시키고 router를 include 합니다.

# main.py
from fastapi import FastAPI

from api import api_main

app = FastAPI(title="hiioAdminServer")

app.include_router(api_main)

다시 localhost:4883/docs 에 접속합니다.

API 가 생긴걸 볼 수 있습니다.

3. Post Method Request Model

pydentic 패키지를 이용해 Request로 들어오는 파라미터를 정의해서 사용해 줍니다.

.
┣━api
┃  ┣━models
┃  ┃ ┣━__init__.py
┃  ┃ ┣━user.py
┃  ┃
┃  ┣━__init__.py
┃  ┣━api_main.py
┣━database

api 디렉토리 안에 user 파일을 만들고

pydentic 의 BaseModel을 상속받는 UserModel class를 만들어 줍니다. 속성은 database에 정의된 id, password로 동일하게 만들어 줍니다.

# api/models/user.py
from typing import Optional

from pydantic import BaseModel, Field


class UserModel(BaseModel):
    id: Optional[str] = Field(None, description="user's id")
    password: Optional[str] = Field(None, description="user's password")

    class Config:
        from_attritues = True

class Config:
from_attributes 를 사용하면 기존의 orm_mode=True를 대체해서 사용할 수 있습니다.

아까 작성했던 signin 과 signup 함수의 파라미터를 user:User로 변경해 줍니다.

# api/api_main.py
from typing import Optional

from fastapi import APIRouter

from api.models import User

api_main = APIRouter()


@api_main.post("/signin")
def signin(user:UserModel):
    return {"accessToken": "accessToken", "refreshToken": "refreshToken"}


@api_main.post("signup")
def signup(user:UserModel):
    return {"message": "success"}

변경 된 Request body로 나오는 것을 볼 수 있습니다.

[관리자 페이지] Admin Page - 회원가입 구현하기 #1

아주 간단한 회원 가입을 위해 먼저 DB -> Server -> Front 순으로 구현해봅니다.

DB 는 MariaDB 11

Server는 FastAPI

Front는 NextJS

를 사용합니다.

Python ORM 라이브러리인 sqlAlchemy 로 데이터를 조회/등록/수정/삭제합니다.

1. User Table 만들기

회원 정보는 USER_ID 와 PASSWORD 만을 가지고 있습니다.

MariaDB에 어드민 페이지를 위한 데이터베이스와 User를 만들고 권한을 부여합니다.

프로젝트 이름은 HIIO_ADMIN 으로 정했습니다.(지금)

DB NAME 은 HIIO_ADMIN
USER 는 hiio420으로 생성합니다.

CREATE DATABASE HIIO_ADMIN;
CREATE USER 'hiio420'@'%' INDENTIFIED BY '<password>';
GRANT ALL PRIVILEGES ON HIIO_ADMIN.* TO 'hiio420'@'%' IDENTIFIED BY '<password>';

sql 쿼리로 테이블을 만들수도 있지만
sqlAlchemy를 이용해 봅니다.

디렉토리 구조는 아래와 같습니다.

.
┣━database

┃  ┣━entity
┃     ┣━__init__.py
┃     ┣━base.py
┃     ┣━user.py
┃  ┣━__init__.py
┃  ┣━core.py
┃
┣━README.md
┣━.gitignore

pip 로 mariadb와 sqlAlchemy를 설치합니다.

pip install mariadb
pip install sqlAlchemy

mariadb는 DB 연결에 쓰입니다.

core.py에 DB 연결을 위한 URL을 생성하고 create_engine method로 engine을 만들어 줍니다.

entity 디렉토리에는 DB entity 관련 class 들을 위치 시킵니다.base.py 에는 DeclarativeBase 상속받는 Base class를 만들고 user.py에는 Base class를 상속받는 User class를 만들어서 create_all()로 테이블을 생성해 줍니다.

# database/core.py
from sqlalchemy import URL, create_engine

url = URL.create(drivername="mariadb+mariadbconnector", username="hiio420", password="...", host="...",
                 port="...", database="HIIO_ADMIN")

engine = create_engine(url)

# database/entity/base.py
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass

# database/entity/user.py
from sqlalchemy import Column, String

from database.entity.base import Base


class User(Base):
    __tablename__ = "USER"
    __table_args__ = {
        'comment': 'USER TABLE'
    }
    id = Column("USER_ID", String(20), primary_key=True, comment="USER ID")
    password = Column("PASSWORD", String(120), nullable=False, comment="USER PASSWORD")

root 디렉토리에 main.py 를 만들어서 실행 시켜 보면 User 테이블이 만들어 진 것을 볼 수 있습니다.

# main.py
from database import engine, Base

if __name__ == '__main__':
    Base.metadata.drop_all(engine)
    Base.metadata.create_all(engine)

[관리자 페이지] Admin Page - 로그인 프로세스 생각해보기 #2

User는 먼저 로그인 화면으로 진입합니다. 로그인 할 수 있는 아이디나 패스워드가 없는 경우 회원가입을 진행하고 회원가입이 정상적으로 완료되면 다시 로그인 화면으로 돌아가서 로그인을 진행 성공하면 메인 페이지로 이동하는 프로세스를 생각해 봅니다.

flowchart LR
    User-->SignIn

        SignIn-->|Don't Have credential|SignUp

        subgraph SignUp Process
        SignUp --->|Fail| Alert1[Alert SignIn fail]
        end

    subgraph SignIn Process
         SignIn ===>|Success| Main
        SignIn--->|Fail| Alert2[Alert SignIn fail]
        end

        SignUp ===>|success| SignIn

Admin Page - 로그인 프로세스 생각해보기

User가 Admin Page에 접근했을 경우

1. User의 Token 이 존재 하는지 확인
2. 존재 한다면 Server에서 Token 발송 검증 요청
3. User 정보 return
4. Token 이 없다면 Sign In Page로 이동

위 순서로 접근한다고 하고 아래와 같은 Sequence Diagram을 생각해 봤다.

Diagram

sequenceDiagram
        autonumber
        actor User
    participant Admin Main
    participant Admin Server
    participant Admin DB

    User ->>+ Admin Main: Access Admin Main Page
        alt Token exists === true
            Admin Main ->>+ Admin Server: Request User Info WIth Token


            rect rgba(0, 23, 255, .1)
                alt Token valid Ok 

                    Admin Server ->>+ Admin DB : Select User by UserId
                    Activate Admin DB

                    rect rgba(0, 56, 255, .1)
                        alt User exists Ok
                            Admin DB ->>- Admin Server:Return User Info
                            Activate Admin Server
                            Admin Server ->>- Admin Main:Return User Info
                            Activate Admin Main
                            Admin Main ->>- User : Data
                        else User not exists
                            Admin DB ->>- Admin Server:Return None
                            Activate Admin Server
                            Admin Server ->>- Admin Main : Return 404 & Data is None
                            Activate Admin Main 
                            Admin Main ->>- User : Alert Msg
                        end
                    end
                else Token Valid False
                    Admin Server ->>- Admin Main : return 401
                    Activate Admin Main
                    Admin Main ->>- User : Token Valid Fail Redirect Admin Sign In
                end
            end
            else Token not Exists
                Admin Main ->>- User : Token Not Exists & Redirect Admin Sign In
        end

위 프로세스를 토대로 13 개의 요구사항을 정의 해 보았습니다.

이를 토대로 개발에 들어가면 좋을지 고민해 봅니다.

아직은 어설프지만 다이어그램이나 요구사항을 만들어 봤다는 것의 의의를 둡니다.

User 접근에 대한 개발을 하기에 앞서 아무런 User 정보가 없기 때문에 User 정보를 입력받는 Sign Up 과 로그인하는 Sign In 프로세스에 대한 고민을 좀더 해봐야 할 거 같습니다.