Merge pull request #14 from stat-kwon/master

fix: apply get token type method by using JWTUtil

src/spaceone/notification/connector/notification_plugin_connector.py

@@ -2,6 +2,7 @@
 
 from spaceone.core.connector import BaseConnector
 from spaceone.core.connector.space_connector import SpaceConnector
+from spaceone.core.auth.jwt.jwt_util import JWTUtil
 
 __all__ = ["NotificationPluginConnector"]
 _LOGGER = logging.getLogger(__name__)
@@ -10,8 +11,9 @@
 class NotificationPluginConnector(BaseConnector):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        token = self.transaction.get_meta("token")
+        self.token_type = JWTUtil.get_value_from_token(token, "typ")
         self.noti_plugin_connector = None
-        self.token_type = self.transaction.get_meta("authorization.token_type")
 
     def initialize(self, endpoint: str):
         static_endpoint = self.config.get("endpoint")

src/spaceone/notification/manager/identity_manager.py

@@ -1,8 +1,8 @@
 import logging
 
-from spaceone.core import config
 from spaceone.core.connector.space_connector import SpaceConnector
 from spaceone.core.manager import BaseManager
+from spaceone.core.auth.jwt.jwt_util import JWTUtil
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -20,10 +20,11 @@
 class IdentityManager(BaseManager):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        token = self.transaction.get_meta("token")
+        self.token_type = JWTUtil.get_value_from_token(token, "typ")
         self.identity_connector: SpaceConnector = self.locator.get_connector(
             "SpaceConnector", service="identity"
         )
-        self.token_type = self.transaction.get_meta("authorization.token_type")
 
     def get_resource(self, resource_id: str, resource_type: str, domain_id: str):
         get_method = _GET_RESOURCE_METHODS[resource_type]

src/spaceone/notification/manager/plugin_manager.py

@@ -2,7 +2,7 @@
 
 from spaceone.core.manager import BaseManager
 from spaceone.core.connector.space_connector import SpaceConnector
-from spaceone.core import config
+from spaceone.core.auth.jwt.jwt_util import JWTUtil
 
 from spaceone.notification.connector.notification_plugin_connector import (
     NotificationPluginConnector,
@@ -14,13 +14,14 @@
 class PluginManager(BaseManager):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        token = self.transaction.get_meta("token")
+        self.token_type = JWTUtil.get_value_from_token(token, "typ")
         self.plugin_connector: SpaceConnector = self.locator.get_connector(
             "SpaceConnector", service="plugin"
         )
         self.noti_plugin_connector: NotificationPluginConnector = (
             self.locator.get_connector(NotificationPluginConnector)
         )
-        self.token_type = self.transaction.get_meta("authorization.token_type")
 
     def initialize(self, plugin_info: dict, domain_id: str) -> dict:
         _LOGGER.debug(f"[initialize] plugin_info: {plugin_info}")

src/spaceone/notification/manager/user_secret_manager.py

@@ -2,18 +2,19 @@
 
 from spaceone.core.manager import BaseManager
 from spaceone.core.connector.space_connector import SpaceConnector
-from spaceone.notification.error import *
+from spaceone.core.auth.jwt.jwt_util import JWTUtil
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class UserSecretManager(BaseManager):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
+        token = self.transaction.get_meta("token")
+        self.token_type = JWTUtil.get_value_from_token(token, "typ")
         self.secret_connector: SpaceConnector = self.locator.get_connector(
             "SpaceConnector", service="secret"
         )
-        self.token_type = self.transaction.get_meta("authorization.token_type")
 
     def create_user_secret(self, params):
         return self.secret_connector.dispatch("UserSecret.create", params)