feat: add injected params at token when user use reset_password

Signed-off-by: ImMin5 <[email protected]>

Author: ImMin5

src/spaceone/identity/manager/token_manager/base.py

@@ -93,7 +93,12 @@ def issue_token(
         return {"access_token": access_token, "refresh_token": refresh_token}
 
     def issue_temporary_token(
-        self, user_id: str, domain_id: str, private_jwk: dict, timeout: int
+        self,
+        user_id: str,
+        domain_id: str,
+        private_jwk: dict,
+        timeout: int,
+        injected_params: dict,
     ) -> dict:
         permissions = [
             "identity:UserProfile",
@@ -108,7 +113,11 @@ def issue_temporary_token(
 
         # Issue token
         access_token = key_gen.generate_token(
-            "ACCESS_TOKEN", timeout=timeout, role_type="USER", permissions=permissions
+            "ACCESS_TOKEN",
+            timeout=timeout,
+            role_type="USER",
+            permissions=permissions,
+            injected_params=injected_params,
         )
 
         return {"access_token": access_token}

src/spaceone/identity/model/user_profile/request.py

@@ -31,6 +31,7 @@ class UserProfileUpdatePasswordRequest(BaseModel):
     user_id: str
     current_password: Union[str, None] = None
     new_password: str
+    required_actions: Union[list, None] = None
     domain_id: str
 
 

src/spaceone/identity/service/user_profile_service.py

@@ -91,6 +91,7 @@ def update_password(
             params (UserProfileUpdatePasswordRequest): {
                 'current_password': 'str',
                 'new_password': 'str',      # required
+                'required_actions': 'list', # injected from auth (required)
                 'user_id': 'str',           # injected from auth (required)
                 'domain_id': 'str'          # injected from auth (required)
             }
@@ -101,6 +102,11 @@ def update_password(
         user_vo = self.user_mgr.get_user(params.user_id, params.domain_id)
 
         required_actions = list(user_vo.required_actions)
+
+        if params.required_actions:
+            required_actions.extend(params.required_actions)
+            user_vo.required_actions = list(set(required_actions))
+
         if "UPDATE_PASSWORD" not in required_actions and not params.current_password:
             raise ERROR_REQUIRED_PARAMETER(key="current_password")
 
@@ -209,7 +215,8 @@ def reset_password(self, params: UserProfileResetPasswordRequest) -> None:
         reset_password_type = config.get_global("RESET_PASSWORD_TYPE", "ACCESS_TOKEN")
         email_manager = EmailManager()
         if reset_password_type == "ACCESS_TOKEN":
-            token = self._issue_temporary_token(user_id, domain_id)
+            injected_params = {"required_actions": ["UPDATE_PASSWORD"]}
+            token = self._issue_temporary_token(user_id, domain_id, injected_params)
             reset_password_link = self._get_console_sso_url(
                 domain_id, token["access_token"]
             )
@@ -488,7 +495,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) -> dict:
+    def _issue_temporary_token(
+        self, user_id: str, domain_id: str, injected_params: dict
+    ) -> dict:
         identity_conf = config.get_global("IDENTITY") or {}
         token_conf = identity_conf.get("token", {})
         timeout = token_conf.get("temporary_token_timeout", 86400)
@@ -497,7 +506,11 @@ def _issue_temporary_token(self, user_id: str, domain_id: str) -> dict:
 
         local_token_manager = LocalTokenManager()
         return local_token_manager.issue_temporary_token(
-            user_id, domain_id, private_jwk=private_jwk, timeout=timeout
+            user_id,
+            domain_id,
+            private_jwk=private_jwk,
+            timeout=timeout,
+            injected_params=injected_params,
         )
 
     def _get_console_sso_url(self, domain_id: str, token: str) -> str:

src/spaceone/identity/service/user_service.py

@@ -114,7 +114,12 @@ def create_user(self, params: dict) -> User:
                 user_id = user_vo.user_id
 
                 email_manager.send_temporary_password_email_when_user_added(
-                    domain_display_name, user_id, email, console_link, temp_password, language
+                    domain_display_name,
+                    user_id,
+                    email,
+                    console_link,
+                    temp_password,
+                    language,
                 )
         else:
             user_vo = self.user_mgr.create_user(params)
@@ -127,13 +132,20 @@ def create_user(self, params: dict) -> User:
                 email_mgr = EmailManager()
 
                 domain_name = self._get_domain_name(domain_id)
-                domain_display_name = self._get_domain_display_name(domain_id, domain_name)
+                domain_display_name = self._get_domain_display_name(
+                    domain_id, domain_name
+                )
 
                 console_link = self._get_console_url(domain_name)
                 external_auth_provider = self._get_external_auth_provider(domain_id)
 
                 email_mgr.send_invite_email_when_external_user_added(
-                    domain_display_name, user_id, user_id, console_link, language, external_auth_provider
+                    domain_display_name,
+                    user_id,
+                    user_id,
+                    console_link,
+                    language,
+                    external_auth_provider,
                 )
 
         return user_vo
@@ -468,7 +480,11 @@ def _get_domain_name(self, domain_id: str) -> str:
         return domain_vo.name
 
     def _issue_temporary_token(
-        self, user_id: str, domain_id: str, timeout: int = None
+        self,
+        user_id: str,
+        domain_id: str,
+        timeout: int = None,
+        injected_params: dict = None,
     ) -> dict:
         if timeout is None:
             identity_conf = config.get_global("IDENTITY", {}) or {}
@@ -479,7 +495,11 @@ def _issue_temporary_token(
 
         local_token_manager = LocalTokenManager()
         return local_token_manager.issue_temporary_token(
-            user_id, domain_id, private_jwk, timeout=timeout
+            user_id,
+            domain_id,
+            private_jwk,
+            timeout=timeout,
+            injected_params=injected_params,
         )
 
     @staticmethod