Merge pull request #40 from ImMin5/master

Modify init identity manager at secret and trusted_secret service

src/spaceone/secret/manager/identity_manager.py

@@ -1,11 +1,15 @@
 from spaceone.core import config
+from spaceone.core.auth.jwt import JWTUtil
 from spaceone.core.manager import BaseManager
 from spaceone.core.connector.space_connector import SpaceConnector
 
 
 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_conn: SpaceConnector = self.locator.get_connector(
             "SpaceConnector", service="identity"
         )
@@ -27,11 +31,18 @@ def get_trusted_account(self, trusted_account_id):
     def list_trusted_accounts(self, query):
         return self.identity_conn.dispatch("TrustedAccount.list", {"query": query})
 
-    def get_service_account(self, service_account_id):
-        return self.identity_conn.dispatch(
-            "ServiceAccount.get",
-            {"service_account_id": service_account_id},
-        )
+    def get_service_account(self, service_account_id: str, domain_id: str):
+        if self.token_type == "SYSTEM_TOKEN":
+            return self.identity_conn.dispatch(
+                "ServiceAccount.get",
+                {"service_account_id": service_account_id},
+                x_domain_id=domain_id,
+            )
+        else:
+            return self.identity_conn.dispatch(
+                "ServiceAccount.get",
+                {"service_account_id": service_account_id},
+            )
 
     def list_service_accounts(self, query):
         return self.identity_conn.dispatch("ServiceAccount.list", {"query": query})

src/spaceone/secret/service/secret_service.py

@@ -23,7 +23,6 @@ class SecretService(BaseService):
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.secret_mgr: SecretManager = self.locator.get_manager("SecretManager")
-        self.identity_mgr: IdentityManager = self.locator.get_manager("IdentityManager")
 
     @transaction(
         permission="secret:Secret.write",
@@ -56,27 +55,28 @@ def create(self, params):
         resource_group = params["resource_group"]
         domain_id = params["domain_id"]
         workspace_id = params.get("workspace_id")
+        identity_mgr: IdentityManager = self.locator.get_manager("IdentityManager")
 
         # Check permission by resource group
         if resource_group == "PROJECT":
             if "service_account_id" in params:
-                service_account_info = self.identity_mgr.get_service_account(
-                    params["service_account_id"]
+                service_account_info = identity_mgr.get_service_account(
+                    params["service_account_id"], domain_id
                 )
 
                 params["provider"] = service_account_info["provider"]
                 params["project_id"] = service_account_info["project_id"]
                 params["workspace_id"] = service_account_info["workspace_id"]
             elif "project_id" in params:
-                project_info = self.identity_mgr.get_project(params["project_id"])
+                project_info = identity_mgr.get_project(params["project_id"])
                 params["workspace_id"] = project_info["workspace_id"]
             else:
                 raise ERROR_REQUIRED_PARAMETER(key="project_id")
         elif resource_group == "WORKSPACE":
             if workspace_id is None:
                 raise ERROR_REQUIRED_PARAMETER(key="workspace_id")
 
-            self.identity_mgr.check_workspace(workspace_id, domain_id)
+            identity_mgr.check_workspace(workspace_id, domain_id)
             params["project_id"] = "*"
         else:
             params["workspace_id"] = "*"
@@ -134,7 +134,10 @@ def update(self, params):
 
         if secret_vo.resource_group == "PROJECT":
             if project_id := params.get("project_id"):
-                self.identity_mgr.get_project(project_id)
+                identity_mgr: IdentityManager = self.locator.get_manager(
+                    "IdentityManager"
+                )
+                identity_mgr.get_project(project_id)
             else:
                 raise ERROR_PERMISSION_DENIED()
 

src/spaceone/secret/service/trusted_secret_service.py

@@ -213,11 +213,7 @@ def get_data(self, params):
             trusted_secret_id, domain_id, workspace_id
         )
 
-        secret_conn_mgr: SecretConnectorManager = self.locator.get_manager(
-            "SecretConnectorManager"
-        )
-
-        trusted_secret_data = secret_conn_mgr.get_secret(trusted_secret_id)
+        trusted_secret_data = self._get_trusted_secret_data(trusted_secret_id)
 
         return {
             "encrypted": trusted_secret_vo.encrypted,
@@ -326,3 +322,11 @@ def _check_related_secret(self, trusted_secret_id, domain_id):
         )
         if secret_vos.count() > 0:
             raise ERROR_EXIST_RELATED_SECRET(secret_id=secret_vos[0].secret_id)
+
+    def _get_trusted_secret_data(self, trusted_secret_id):
+
+        secret_conn_mgr: SecretConnectorManager = self.locator.get_manager(
+            "SecretConnectorManager"
+        )
+
+        return secret_conn_mgr.get_secret(trusted_secret_id)