[Python] JWT를 적용해보아요

Junkyu_Kang·2025년 1월 22일

이제는 필수가 되어버린 JWT!

예전에 한번 해봤지만 python에 새롭게 적용을 하게 되어 내용을 간략하게 적으려 한다!

LOGIN 기능은 쉽게 만들 수 있다. 사실상 USER TABLE에서 USER_ID와 PASSWORD를 찾아 비교만 하면 되니까
하지만 데이터 암호화를 위해 비밀번호는 해쉬로 저장되어있을 것이고 암호화와 복호화 과정이 필요할 것이다.
애초에 비밀번호는 단방향 암호화로 진행되기 때문에 DB에는 해쉬로 저장되어 있고 로직 내에서 확인하는 과정이 있을 것이다.

JWT를 왜 사용하는가?
우선 JWT에는 다양한 정보를 담을 수 있다.
물론 크리티컬한 정보를 담아서는 안된다. 해봐야 ID나 찜 목록 등 쿠키에 저장할 만한 데이터를 가지고 있어야 한다.
하지만 이 정보를 바탕으로 인증인가 기능을 수행하기 위해 JWT를 사용한다.

JWT는 로컬스토리지, 세션스토리지, 쿠키 등 다양한 위치에 저장할 수 있지만 나는 쿠키로 저장했다.

데이터 탈취 위험 및 보안을 생각하여 쿠키로 선정하였고 안에는 USER_ID와 권한 정보가 있는 SITE_INFO를 넣었다.

물론 이건 프로토타입이다.. 따라하지 말것..

우선 순서는 이렇다.
1. 회원가입을 한다.
2. 해당 내용을 암호화 한다.
3. 저장한다.
4. 로그인한다.
5. DB에 접근하여 복호화 과정을 거친 후 RETURN을 받는다.
6. RETURN이 TRUE라면 해당 유저의 일부 정보를 COOKIE에 저장한다.

1. 회원가입

@app.route('/api/register', methods=['POST'])
def register():
    if not request.is_json:
        return jsonify({"error": "Content-Type must be application/json"}), 400
    
    db = PostgresqlDS(**db_params)

    data = request.json

    if data is None:
        raise PermissionError
    userid = data['user_id']
    username = data['user_name']
    password = data['password']
    phonenumber = data['phone_number']
    # created_at = datetime.datetime.now()

    hash_pass = LoginUserClass.hash_password(password)

    user_df = pd.DataFrame([{"user_id" : userid, "user_name": username, "password": hash_pass,  "phone_number": phonenumber}])

    db.save("user_info", {}, user_df)
    return jsonify({"message": "Success signup"}), 201

회원가입 기능이다. 이건 단순하게 보면 알 수 있을 것이다.
나는 POSTGRES에 대한 기능을 따로 UTIL화 하여 사용하기 때문에 기본 문법과 다른 부분을 이해하길 바란다.

2. 암호화

    @staticmethod
   def hash_password(originPass):
       salt = gensalt()
       hashed = hashpw(originPass.encode('utf-8'), salt)
       return hashed.decode('utf-8')

파이썬에서 암호화를 쉽게 진행할 수 있다.

물론

from bcrypt import hashpw, gensalt, checkpw
from flask import Flask, request, jsonify

위 내용은 IMPORT를 하고 할 수 있다.

3. 복호화

    @staticmethod
    def check_password(originPass, hashPass):
        # Ensure hashPass is bytes, not str
        if isinstance(hashPass, str):
            hashPass = hashPass.encode('utf-8')
        return checkpw(originPass.encode('utf-8'), hashPass)

로그인을 할 때는 해쉬값을 다시 복호화하여 유저가 입력한 PASSWORD와 일치 여부를 확인해야한다.

4. 로그인

@app.route('/api/login', methods=['POST'])
def login():
   data = request.json
   if data is None:
       raise PermissionError
   
   userId = data['user_id']
   password = data['password']

   db = PostgresqlDS(**db_params)

   user_result = db.load("user_info", ["password"], conditions={"user_id" : userId})

   auth_result = db.load("user_site", ["auth_id", "site_id"], conditions={"user_id" : userId})
   if auth_result.empty:
       return jsonify({"error": "No authorized sites found for the user"}), 403
   
   site_ids = [str(site_id) for site_id in auth_result["site_id"].tolist()]

   print(f"site_ids = {site_ids}")

   site_results = []
   for site_id in site_ids:
       site_result = db.load("sites", ["site_id", "site_name"], conditions={"site_id": site_id})
       if not site_result.empty:
           site_results.append(site_result)

   if site_results:
       site_result = pd.concat(site_results, ignore_index=True)
   else:
       return jsonify({"error": "No sites found for the given site_ids"}), 404
   
   if site_result.empty:
       return jsonify({"error": "No sites found for the given site_ids"}), 404
   
   merged_result = pd.merge(auth_result, site_result, on="site_id")
   merged_result["site_id"] = merged_result["site_id"].astype(str)
   merged_result = merged_result.drop(columns=["site_id"])
   user_auth_info = merged_result.to_dict(orient="records")

   if not user_result.empty and LoginUserClass.check_password(password, user_result.iloc[0]["password"]):
       token = LoginUserClass.generate_jwt({"username": userId, "user_auth_info": user_auth_info})
       response = jsonify({"message": "login successful"})
       response.set_cookie("token", token, httponly=True,secure=True, samesite="Strict", max_age=3600)
       return response, 200
   return jsonify({"error": "Invalid username or password"}), 401
   

사실 내용은 별거 없다.

DB에 조회할 내용을 위한 값을 USER_ID를 바탕으로 파라미터로 사용해 다른 테이블의 정보를 넣고 이를 JWT로 만드는 거다.

5. JWT 발급


   @staticmethod
   def generate_jwt(payload, expr=3600):
       payload["exp"] = datetime.datetime.utcnow() + datetime.timedelta(seconds=expr)
       token = jwt.encode(payload, Secret_key, algorithm="HS256")
       return token

위 내용을 즉, 로그인을 할 때 넣었던 내용을 바탕으로 JWT를 발급하는 코드다 EXPR은 유효시간으로 나는 60분간 유효한 시간의 JWT를 발급받은 것이며 사용한 알고리즘은 HS256이다.

Secret_key의 경우 본인이 직접 어떤 값을 지정할 것
절대 PASSWORD 이런거로 하지말고.. 좀 길고 복잡하고 어려운걸로..

해당 발급 받은 쿠키는 SET_COOKIE를 사용하여 저장하며 해당 내용에 대한 설정 정보는 로그인 파트에서 확인 가능하다.

6. JWT 확인

    @staticmethod
    def decode_jwt(token):
        try:
            payload = jwt.decode(token, Secret_key, algorithms=["HS256"])
            return payload
        except jwt.ExpiredSignatureError:
            return {"error": "Token expired"}
        except jwt.InvalidTokenError:
            return {"error": "Invalid token"}

JWT 복호화 과정이다.
사실 코드보면 이해하기 쉬울거다. 기간이 지난 토큰은 에러를 보내고 유효하지 않은 토큰도 에러를 발생시킨다.
이에 해당 TOKEN에 대한 정보는 PAYLOAD에 저장되고 리턴하게 된다.

7. 인증인가

    @staticmethod
  def token_required(f):
      @wraps(f)
      def decorated(*args, **kwargs):
          token = request.cookies.get("token")
          if not token:
              return jsonify({"error": "Token is missing"}), 403
          
          # token = token.split(" ")[1]

          decoded = LoginUserClass.decode_jwt(token)
          if "error" in decoded:
              return jsonify(decoded), 401
          return f(*args, user=decoded, **kwargs)
      return decorated

다른 방법이 있는지 모르겠지만 나는 WRAPS로 감싸서 각 API에 인증인가 절차를 추가할 예정이다. 해당 내용을 확인하고 TOCKEN이 없을 경우 에러를 발생시키며 해당 값을 바탕으로 권한과 기능 사용 여부 등을 지정, 확인할 수 있다.

물론 더 딥하게 들어가서 관리하고 데이터를 확인하는 과정이 있지만

그거까지 올리면... 나 신고당한다.. 안된다..

아무튼 JWT 생성 및 확인 방법은 여기까지!
안녕!

profile
강준규

0개의 댓글