feat: modify get_url api

src/spaceone/cost_analysis/conf/global_conf.py

@@ -12,6 +12,16 @@
     }
 }
 
+# Cost Report Token Settings
+COST_REPORT_TOKEN_TIMEOUT = 259200  # 3 days
+COST_REPORT_DEFAULT_PERMISSIONS = [
+    "cost-analysis:CostReport.read",
+    "cost-analysis:CostReportData.read",
+    "cost-analysis:CostReportConfig.read",
+    "config:Domain.read",
+    "identity.Provider.read",
+]
+
 CACHES = {
     "default": {},
     "local": {

src/spaceone/cost_analysis/manager/identity_manager.py

@@ -18,6 +18,20 @@ def __init__(self, *args, **kwargs):
             SpaceConnector, service="identity"
         )
 
+    def get_user(self, domain_id: str, user_id: str) -> dict:
+        system_token = config.get_global("TOKEN")
+        response = self.identity_conn.dispatch(
+            "User.list",
+            {"user_id": user_id, "state": "ENABLED"},
+            x_domain_id=domain_id,
+            token=system_token,
+        )
+        users_info = response.get("results", [])
+        if users_info:
+            return users_info[0]
+        else:
+            return {}
+
     def get_domain_name(self, domain_id: str) -> str:
         system_token = config.get_global("TOKEN")
 
@@ -141,7 +155,5 @@ def grant_token(
         self,
         params: dict,
     ) -> str:
-        if self.token_type == "SYSTEM_TOKEN":
-            return "system_token"
-        else:
-            return self.transaction.get_meta("token")
+        token_info = self.identity_conn.dispatch("Token.grant", params)
+        return token_info["access_token"]

src/spaceone/cost_analysis/service/cost_report_data_service.py

@@ -52,7 +52,7 @@ def __init__(self, *args, **kwargs):
     @append_keyword_filter(["product", "cost_report_data_id"])
     @convert_model
     def list(
-            self, params: CostReportDataSearchQueryRequest
+        self, params: CostReportDataSearchQueryRequest
     ) -> Union[CostReportsDataResponse, dict]:
         """List cost report data"""
 
@@ -162,20 +162,20 @@ def create_cost_report_data(self, cost_report_vo: CostReport):
         )
 
     def _aggregate_monthly_cost_report_data(
-            self,
-            domain_id: str,
-            workspace_id: str,
-            cost_report_config_id,
-            cost_report_id: str,
-            currency: str,
-            workspace_name: str,
-            project_name_map: dict,
-            service_account_name_map: dict,
-            data_source_currency_map: dict,
-            data_source_ids: list,
-            report_month: str,
-            issue_date: str,
-            is_confirmed: bool = False,
+        self,
+        domain_id: str,
+        workspace_id: str,
+        cost_report_config_id,
+        cost_report_id: str,
+        currency: str,
+        workspace_name: str,
+        project_name_map: dict,
+        service_account_name_map: dict,
+        data_source_currency_map: dict,
+        data_source_ids: list,
+        report_month: str,
+        issue_date: str,
+        is_confirmed: bool = False,
     ):
         report_year = report_month.split("-")[0]
 
@@ -294,7 +294,7 @@ def _get_service_account_name_map(self, workspace_id: str, domain_id: str) -> di
 
     @staticmethod
     def _get_data_source_currency_map(
-            data_source_filter: dict, workspace_id: str, domain_id: str
+        data_source_filter: dict, workspace_id: str, domain_id: str
     ) -> Tuple[dict, list]:
         data_source_currency_map = {}
         data_source_mgr = DataSourceManager()

src/spaceone/cost_analysis/service/cost_report_serivce.py

@@ -90,17 +90,23 @@ def send(self, params: CostReportSendRequest) -> None:
     def get_url(self, params: CostReportGetUrlRequest) -> dict:
         """Get cost report url"""
 
+        identity_mgr: IdentityManager = self.locator.get_manager("IdentityManager")
+
+        user_id = self.transaction.get_meta("authorization.user_id")
         domain_id = params.domain_id
+        user_info = identity_mgr.get_user(user_id, domain_id)
+        language = user_info.get("language", "en")
         cost_report_id = params.cost_report_id
 
         # check cost report
         cost_report_vo = self.cost_report_mgr.get_cost_report(
             domain_id, cost_report_id, params.workspace_id
         )
+        workspace_id = cost_report_vo.workspace_id
 
-        # sso_access_token = self._get_temporary_sso_access_token(domain_id)
+        sso_access_token = self._get_temporary_sso_access_token(domain_id, workspace_id)
         cost_report_link = self._get_console_cost_report_url(
-            domain_id, cost_report_id, "token"
+            domain_id, cost_report_id, sso_access_token, language
         )
 
         return {"cost_report_link": cost_report_link}
@@ -350,47 +356,48 @@ def send_cost_report(self, cost_report_vo: CostReport) -> None:
         # list workspace owner role bindings
         identity_mgr: IdentityManager = self.locator.get_manager("IdentityManager")
 
-        workspace_ids = []
-        if workspace_id is not None:
-            rb_query = {
-                "filter": [
-                    {"k": "role_type", "v": role_types, "o": "in"},
-                    {"k": "workspace_id", "v": workspace_id, "o": "eq"},
-                ],
-            }
-            role_bindings_info = identity_mgr.list_role_bindings(
-                params={"query": rb_query}, domain_id=domain_id
-            )
+        rb_query = {
+            "filter": [
+                {"k": "role_type", "v": role_types, "o": "in"},
+                {"k": "workspace_id", "v": workspace_id, "o": "eq"},
+            ],
+        }
+        role_bindings_info = identity_mgr.list_role_bindings(
+            params={"query": rb_query}, domain_id=domain_id
+        )
 
-            workspace_ids = [
-                role_binding_info["workspace_id"]
-                for role_binding_info in role_bindings_info.get("results", [])
-            ]
-            workspace_ids = list(set(workspace_ids))
+        rb_users_ids = [
+            role_binding_info.get("user_id")
+            for role_binding_info in role_bindings_info.get("results", [])
+        ]
+
+        # list users in workspace
+        users_info = identity_mgr.list_workspace_users(
+            params={"workspace_id": workspace_id, "state": "ENABLED"},
+            domain_id=domain_id,
+        )
+
+        filtered_users_info = self.filtered_users_info(users_info, rb_users_ids)
 
-        # list workspace owner users
         email_mgr = EmailManager()
-        for workspace_id in workspace_ids:
-            users_info = identity_mgr.list_workspace_users(
-                params={"workspace_id": workspace_id, "state": "ENABLED"},
-                domain_id=domain_id,
-            )
-            sso_access_token = self._get_temporary_sso_access_token(domain_id)
+        sso_access_token = self._get_temporary_sso_access_token(domain_id, workspace_id)
+        for user_info in filtered_users_info:
+            user_id = user_info["user_id"]
+            email = user_info.get("email", user_id)
+            language = user_info.get("language", "en")
+
             cost_report_link = self._get_console_cost_report_url(
-                domain_id, cost_report_vo.cost_report_id, sso_access_token
+                domain_id, cost_report_vo.cost_report_id, sso_access_token, language
             )
-            for user_info in users_info.get("results", []):
-                user_id = user_info["user_id"]
-                email = user_info.get("email", user_id)
-                language = user_info.get("language", "en")
-                email_mgr.send_cost_report_email(
-                    user_id, email, cost_report_link, language, cost_report_vo
-                )
-            _LOGGER.debug(
-                f"[send_cost_report] send cost report ({workspace_id}/{cost_report_vo.cost_report_id}) to \
-                {users_info.get('total_count', 0)} users"
+
+            email_mgr.send_cost_report_email(
+                user_id, email, cost_report_link, language, cost_report_vo
             )
 
+        _LOGGER.debug(
+            f"[send_cost_report] send cost report ({workspace_id}/{cost_report_vo.cost_report_id}) to {users_info.get('total_count', 0)} users"
+        )
+
     def _get_workspace_name_map(self, domain_id: str) -> Tuple[dict, list]:
         identity_mgr: IdentityManager = self.locator.get_manager("IdentityManager")
         workspace_name_map = {}
@@ -405,31 +412,46 @@ def _get_workspace_name_map(self, domain_id: str) -> Tuple[dict, list]:
         return workspace_name_map, workspace_ids
 
     def _get_console_cost_report_url(
-        self, domain_id: str, cost_report_id: str, token: str
+        self, domain_id: str, cost_report_id: str, token: str, language: str
     ) -> str:
         domain_name = self._get_domain_name(domain_id)
 
         console_domain = config.get_global("EMAIL_CONSOLE_DOMAIN")
         console_domain = console_domain.format(domain_name=domain_name)
 
-        return f"{console_domain}/cost-report?sso_access_token={token}&cost_report_id={cost_report_id}"
+        return f"{console_domain}/cost-report?sso_access_token={token}&cost_report_id={cost_report_id}&language={language}"
 
     def _get_domain_name(self, domain_id: str) -> str:
         identity_mgr: IdentityManager = self.locator.get_manager("IdentityManager")
         domain_name = identity_mgr.get_domain_name(domain_id)
         return domain_name
 
-    def _get_temporary_sso_access_token(self, domain_id: str) -> str:
+    def _get_temporary_sso_access_token(self, domain_id: str, workspace_id: str) -> str:
         identity_mgr: IdentityManager = self.locator.get_manager("IdentityManager")
         system_token = config.get_global("TOKEN")
+        timeout = config.get_global("COST_REPORT_TOKEN_TIMEOUT", 259200)
+        permissions = config.get_global(
+            "COST_REPORT_DEFAULT_PERMISSIONS",
+            [
+                "cost-analysis:CostReport.read",
+                "cost-analysis:CostReportData.read",
+                "cost-analysis:CostReportConfig.read",
+                "config:Domain.read",
+                "identity.Provider.read",
+            ],
+        )
+
         params = {
             "grant_type": "SYSTEM_TOKEN",
-            "scope": "SYSTEM",
+            "scope": "WORKSPACE",
             "token": system_token,
+            "workspace_id": workspace_id,
+            "domain_id": domain_id,
+            "timeout": timeout,
+            "permissions": permissions,
         }
         # todo : make temporary token
-        token = identity_mgr.grant_token(params)
-        return token
+        return identity_mgr.grant_token(params)
 
     @staticmethod
     def _get_current_and_last_month() -> Tuple[str, str]:
@@ -502,3 +524,11 @@ def _aggregate_result_by_currency(results: list) -> list:
                 workspace_result_map[workspace_id] = result.copy()
 
         return [workspace_result for workspace_result in workspace_result_map.values()]
+
+    @staticmethod
+    def filtered_users_info(users_info, rb_users_ids) -> list:
+        filtered_users_info = []
+        for user_info in users_info:
+            if user_info["user_id"] in rb_users_ids:
+                filtered_users_info.append(user_info)
+        return filtered_users_info