refactor: jwt code refactor

Author: raccoon-mh

pkg/pip_requirements.txt

@@ -17,7 +17,6 @@ redis
 cachetools
 pycryptodome
 jwcrypto
-python-jose
 python-dateutil
 python-consul
 dnspython

src/setup.py

@@ -15,7 +15,8 @@
 
 
 import os
-from setuptools import setup, find_packages
+
+from setuptools import find_packages, setup
 
 setup(
     name="spaceone_core",
@@ -55,7 +56,6 @@
         # crypto(jwt) packages
         "pycryptodome",
         "jwcrypto",
-        "python-jose",
         # utils packages
         "python-dateutil",
         "python-consul",

src/spaceone/core/auth/jwt/jwt_util.py

@@ -1,6 +1,8 @@
 import json
+
 from jwcrypto import jwk
-from jose import jwt
+from jwcrypto import jwt as jwcrypto_jwt
+from jwcrypto.jws import JWS
 
 
 class JWTUtil:
@@ -13,24 +15,59 @@ def generate_jwk(key_type="RSA", size=2048):
 
     @staticmethod
     def encode(payload: dict, private_jwk: dict, algorithm="RS256") -> str:
-        return jwt.encode(payload, key=private_jwk, algorithm=algorithm)
+        # Convert dict to JWK object
+        key = jwk.JWK(**private_jwk)
+
+        # Create JWT object with claims and header
+        jwt_obj = jwcrypto_jwt.JWT(claims=payload, header={"alg": algorithm})
+
+        # Sign the token
+        jwt_obj.make_signed_token(key)
+
+        # Serialize to compact format
+        return jwt_obj.serialize()
 
     @staticmethod
     def decode(token: str, public_jwk: dict, algorithm="RS256", options=None) -> dict:
         if options is None:
             options = {}
 
-        options["verify_aud"] = options.get("verify_aud", False)
+        # Convert dict to JWK object
+        key = jwk.JWK(**public_jwk)
+
+        # Create JWT object and deserialize
+        jwt_obj = jwcrypto_jwt.JWT(jwt=token, key=key, algs=[algorithm])
 
-        return jwt.decode(token, key=public_jwk, algorithms=algorithm, options=options)
+        # Validate the token
+        verify_aud = options.get("verify_aud", False)
+        check_claims = None
+        if verify_aud and "aud" in options:
+            check_claims = {"aud": options["aud"]}
+
+        if check_claims:
+            jwt_obj._check_claims = check_claims
+
+        jwt_obj.validate(key)
+
+        # Parse claims from JSON string
+        return json.loads(jwt_obj.claims)
 
     @staticmethod
     def unverified_decode(token: str) -> dict:
-        return jwt.get_unverified_claims(token)
+        # Deserialize JWS without verification
+        jws = JWS()
+        jws.deserialize(token, None)
+
+        # Parse payload from JSON string
+        payload = jws.payload
+        if isinstance(payload, bytes):
+            payload = payload.decode("utf-8")
+
+        return json.loads(payload)
 
     @staticmethod
     def get_value_from_token(token: str, key: str, default: any = None) -> any:
         try:
             return JWTUtil.unverified_decode(token).get(key, default)
-        except Exception as e:
+        except Exception:
             return default