Merge pull request #276 from ImMin5/feature-service-accont-auto-sync

Add feature for inviting external user

src/spaceone/identity/manager/email_manager.py

@@ -22,21 +22,25 @@
         "reset_password": "Reset your password",
         "temp_password": "Your password has been changed",
         "verify_email": "Verify your notification email",
+        "invite_external_user": "You've been invited to join.",
     },
     "ko": {
         "reset_password": "비밀번호 재설정 안내",
         "temp_password": "임시 비밀번호 발급 안내",
         "verify_email": "알림전용 이메일 계정 인증 안내",
+        "invite_external_user": "계정 초대 안내.",
     },
     "en": {
         "reset_password": "Reset your password",
         "temp_password": "Your password has been changed",
         "verify_email": "Verify your notification email",
+        "invite_external_user": "You've been invited to join.",
     },
     "ja": {
         "reset_password": "パスワードリセットのご案内",
         "temp_password": "仮パスワード発行のご案内",
         "verify_email": "通知メールアカウント認証のご案内",
+        "invite_external_user": "参加するように招待されました",
     },
 }
 
@@ -63,7 +67,7 @@ def send_reset_password_email(self, user_id, email, reset_password_link, languag
         self.smtp_connector.send_email(email, subject, email_contents)
 
     def send_temporary_password_email(
-            self, user_id, email, console_link, temp_password, language
+        self, user_id, email, console_link, temp_password, language
     ):
         service_name = self._get_service_name()
         language_map_info = LANGUAGE_MAPPER.get(language, "default")
@@ -80,7 +84,7 @@ def send_temporary_password_email(
         self.smtp_connector.send_email(email, subject, email_contents)
 
     def send_reset_password_email_when_user_added(
-            self, user_id, email, reset_password_link, language
+        self, user_id, email, reset_password_link, language
     ):
         service_name = self._get_service_name()
         language_map_info = LANGUAGE_MAPPER.get(language, "default")
@@ -98,7 +102,7 @@ def send_reset_password_email_when_user_added(
         self.smtp_connector.send_email(email, subject, email_contents)
 
     def send_temporary_password_email_when_user_added(
-            self, user_id, email, console_link, temp_password, language
+        self, user_id, email, console_link, temp_password, language
     ):
         service_name = self._get_service_name()
         language_map_info = LANGUAGE_MAPPER.get(language, "default")
@@ -114,6 +118,29 @@ def send_temporary_password_email_when_user_added(
 
         self.smtp_connector.send_email(email, subject, email_contents)
 
+    def send_invite_email_when_external_user_added(
+        self,
+        user_id: str,
+        email: str,
+        console_link: str,
+        language: str,
+        auth_type: str = "EXTERNAL",
+    ):
+        service_name = self._get_service_name()
+        language_map_info = LANGUAGE_MAPPER.get(language, "default")
+
+        template = JINJA_ENV.get_template(f"sso_invite_user_link_{language}.html")
+
+        email_contents = template.render(
+            user_name=user_id,
+            auth_type=auth_type,
+            service_name=service_name,
+            login_link=console_link,
+        )
+
+        subject = f'[{service_name}] {language_map_info["invite_external_user"]}'
+        self.smtp_connector.send_email(email, subject, email_contents)
+
     def send_verification_email(self, user_id, email, verification_code, language):
         service_name = self._get_service_name()
         language_map_info = LANGUAGE_MAPPER.get(language, "default")

src/spaceone/identity/service/job_service.py

@@ -65,7 +65,7 @@ def create_jobs_by_trusted_account(self, params: dict):
         current_hour = params.get("current_hour", datetime.utcnow().hour)
 
         for trusted_account_vo in self._get_all_schedule_enabled_trusted_accounts(
-                current_hour
+            current_hour
         ):
             try:
                 self.create_service_account_job(trusted_account_vo, {})
@@ -334,7 +334,7 @@ def sync_service_accounts(self, params: dict) -> None:
         )
 
     def create_service_account_job(
-            self, trusted_account_vo: TrustedAccount, job_options: dict
+        self, trusted_account_vo: TrustedAccount, job_options: dict
     ) -> Union[Job, dict]:
         resource_group = trusted_account_vo.resource_group
         provider = trusted_account_vo.provider
@@ -437,10 +437,10 @@ def _get_trusted_secret_data(self, trusted_secret_id: str, domain_id: str) -> di
         return secret_data
 
     def _check_duplicate_job(
-            self,
-            domain_id: str,
-            trusted_account_id: str,
-            this_job_vo: Job,
+        self,
+        domain_id: str,
+        trusted_account_id: str,
+        this_job_vo: Job,
     ) -> bool:
         query = {
             "filter": [
@@ -464,7 +464,7 @@ def _check_duplicate_job(
         return False
 
     def _is_job_failed(
-            self, job_id: str, domain_id: str, workspace_id: str = None
+        self, job_id: str, domain_id: str, workspace_id: str = None
     ) -> bool:
         job_vo: Job = self.job_mgr.get_job(domain_id, job_id, workspace_id)
 
@@ -474,10 +474,10 @@ def _is_job_failed(
             return False
 
     def _close_job(
-            self,
-            job_id: str,
-            domain_id: str,
-            workspace_id: str = None,
+        self,
+        job_id: str,
+        domain_id: str,
+        workspace_id: str = None,
     ):
         job_vo: Job = self.job_mgr.get_job(domain_id, job_id, workspace_id)
         if job_vo.status == "IN_PROGRESS":
@@ -486,7 +486,7 @@ def _close_job(
             self.job_mgr.update_job_by_vo({"finished_at": datetime.utcnow()}, job_vo)
 
     def _create_workspace(
-            self, domain_id: str, trusted_account_id: str, location_info: dict
+        self, domain_id: str, trusted_account_id: str, location_info: dict
     ) -> Workspace:
         name = location_info.get("name")
         reference_id = location_info.get("resource_id")
@@ -503,9 +503,6 @@ def _create_workspace(
         if workspace_vos:
             workspace_vo = workspace_vos[0]
 
-            if workspace_vo.name != name:
-                params["name"] = name
-
             if not workspace_vo.references:
                 params["references"] = [reference_id]
             elif reference_id not in workspace_vo.references:
@@ -535,12 +532,12 @@ def _create_workspace(
         return workspace_vo
 
     def _create_project_group(
-            self,
-            domain_id: str,
-            workspace_id: str,
-            trusted_account_id: str,
-            location_info: dict,
-            parent_group_id: str = None,
+        self,
+        domain_id: str,
+        workspace_id: str,
+        trusted_account_id: str,
+        location_info: dict,
+        parent_group_id: str = None,
     ) -> ProjectGroup:
         name = location_info["name"]
         reference_id = location_info["resource_id"]
@@ -590,14 +587,14 @@ def _create_project_group(
         return project_group_vo
 
     def _create_project(
-            self,
-            result: dict,
-            domain_id: str,
-            workspace_id: str,
-            trusted_account_id: str,
-            project_group_id: str = None,
-            sync_options: dict = None,
-            project_type: str = "PRIVATE",
+        self,
+        result: dict,
+        domain_id: str,
+        workspace_id: str,
+        trusted_account_id: str,
+        project_group_id: str = None,
+        sync_options: dict = None,
+        project_type: str = "PRIVATE",
     ) -> Project:
         name = result["name"]
         reference_id = result["resource_id"]
@@ -636,13 +633,13 @@ def _create_project(
         return project_vo
 
     def _create_service_account(
-            self,
-            result: dict,
-            project_vo: Project,
-            trusted_account_id: str,
-            trusted_secret_id: str,
-            provider: str,
-            sync_options: dict = None,
+        self,
+        result: dict,
+        project_vo: Project,
+        trusted_account_id: str,
+        trusted_secret_id: str,
+        provider: str,
+        sync_options: dict = None,
     ) -> Union[ServiceAccount, None]:
         domain_id = project_vo.domain_id
         workspace_id = project_vo.workspace_id
@@ -728,7 +725,7 @@ def _create_service_account(
         return service_account_vo
 
     def _remove_old_reference_id_from_workspace(
-            self, domain_id: str, workspace_id: str, reference_id: str
+        self, domain_id: str, workspace_id: str, reference_id: str
     ) -> None:
         query = {
             "filter": [

src/spaceone/identity/service/user_service.py

@@ -69,7 +69,7 @@ def create_user(self, params: dict) -> User:
         reset_password = params["reset_password"]
         domain_id = params["domain_id"]
         email = params.get("email")
-        language = params.get("language") or "en"
+        language = params.get("language", "en") or "en"
 
         if reset_password:
             self._check_reset_password_eligibility(user_id, auth_type, email)
@@ -86,7 +86,7 @@ def create_user(self, params: dict) -> User:
                 identity_conf = config.get_global("IDENTITY", {}) or {}
                 token_conf = identity_conf.get("token", {})
                 timeout = token_conf.get("invite_token_timeout", 604800)
-                
+
                 token = self._issue_temporary_token(user_id, domain_id, timeout)
                 reset_password_link = self._get_console_sso_url(
                     domain_id, token["access_token"]
@@ -107,6 +107,20 @@ def create_user(self, params: dict) -> User:
                 )
         else:
             user_vo = self.user_mgr.create_user(params)
+            if (
+                auth_type == "EXTERNAL"
+                and self._check_invite_external_user_eligibility(user_id, user_id)
+            ):
+                email_manager = EmailManager()
+                console_link = self._get_console_url(domain_id)
+
+                email_manager.send_invite_email_when_external_user_added(
+                    user_id,
+                    user_id,
+                    console_link,
+                    language,
+                    user_vo.auth_type,
+                )
 
         return user_vo
 
@@ -237,7 +251,7 @@ def disable_mfa(self, params: UserDisableMFARequest) -> Union[UserResponse, dict
     @transaction(permission="identity:User.write", role_types=["DOMAIN_ADMIN"])
     @convert_model
     def set_required_actions(
-            self, params: UserSetRequiredActionsRequest
+        self, params: UserSetRequiredActionsRequest
     ) -> Union[UserResponse, dict]:
         """Set required actions
 
@@ -408,7 +422,9 @@ def _get_domain_name(self, domain_id: str) -> str:
         domain_vo = self.domain_mgr.get_domain(domain_id)
         return domain_vo.name
 
-    def _issue_temporary_token(self, user_id: str, domain_id: str, timeout: int = None) -> dict:
+    def _issue_temporary_token(
+        self, user_id: str, domain_id: str, timeout: int = None
+    ) -> dict:
         if timeout is None:
             identity_conf = config.get_global("IDENTITY", {}) or {}
             token_conf = identity_conf.get("token", {})
@@ -444,7 +460,7 @@ def _get_console_url(self, domain_id):
         return console_domain.format(domain_name=domain_name)
 
     @staticmethod
-    def _check_reset_password_eligibility(user_id, auth_type, email):
+    def _check_reset_password_eligibility(user_id: str, auth_type: str, email: str):
         if auth_type == "EXTERNAL":
             raise ERROR_UNABLE_TO_RESET_PASSWORD_IN_EXTERNAL_AUTH(user_id=user_id)
         elif email is None:
@@ -460,8 +476,26 @@ def _generate_temporary_password():
                 for _ in range(12)
             )
             if (
-                    re.search("[a-z]", random_password)
-                    and re.search("[A-Z]", random_password)
-                    and re.search("[0-9]", random_password)
+                re.search("[a-z]", random_password)
+                and re.search("[A-Z]", random_password)
+                and re.search("[0-9]", random_password)
             ):
                 return random_password
+
+    @staticmethod
+    def _check_invite_external_user_eligibility(user_id: str, email: str) -> bool:
+        rule = r"[^@]+@[^@]+\.[^@]+"
+
+        if email is None:
+            _LOGGER.debug(
+                f"[_check_invite_external_user_eligibility] email is None (user_id={user_id})"
+            )
+            return False
+
+        if not re.match(rule, email):
+            _LOGGER.debug(
+                f"[_check_invite_external_user_eligibility] email format is incorrect (user_id={user_id}, email={email})"
+            )
+            return False
+
+        return True