feat: modify feature for setting refresh 'UserProfile' -> 'User'

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

src/spaceone/identity/interface/grpc/user.py

@@ -37,6 +37,12 @@ def set_required_actions(self, request, context):
         response: dict = user_svc.set_required_actions(params)
         return self.dict_to_message(response)
 
+    def set_refresh_timeout(self, request, context):
+        params, metadata = self.parse_request(request, context)
+        user_svc = UserService(metadata)
+        response: dict = user_svc.set_refresh_timeout(params)
+        return self.dict_to_message(response)
+
     def delete(self, request, context):
         params, metadata = self.parse_request(request, context)
         user_svc = UserService(metadata)

src/spaceone/identity/interface/grpc/user_profile.py

@@ -15,12 +15,6 @@ def update(self, request, context):
         response: dict = user_profile_svc.update(params)
         return ParseDict(response, user_pb2.UserInfo())
 
-    def set_refresh_timeout(self, request, context):
-        params, metadata = self.parse_request(request, context)
-        user_profile_svc = UserProfileService(metadata)
-        response: dict = user_profile_svc.set_refresh_timeout(params)
-        return ParseDict(response, user_pb2.UserInfo())
-
     def verify_email(self, request, context):
         params, metadata = self.parse_request(request, context)
         user_profile_svc = UserProfileService(metadata)

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

@@ -39,15 +39,15 @@ def get_token_manager_by_auth_type(cls, auth_type):
         raise ERROR_INVALID_AUTHENTICATION_TYPE(auth_type=auth_type)
 
     def issue_token(
-            self,
-            private_jwk,
-            refresh_private_jwk,
-            domain_id,
-            workspace_id=None,
-            timeout=None,
-            permissions=None,
-            projects=None,
-            app_id=None,
+        self,
+        private_jwk,
+        refresh_private_jwk,
+        domain_id,
+        workspace_id=None,
+        timeout=None,
+        permissions=None,
+        projects=None,
+        app_id=None,
     ):
         if self.is_authenticated is False:
             raise ERROR_NOT_AUTHENTICATED()
@@ -89,7 +89,7 @@ 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
     ) -> dict:
         permissions = [
             "identity:UserProfile",
@@ -130,9 +130,9 @@ def set_timeout(self, timeout: Union[int, None]) -> int:
     def _get_refresh_token_timeout(self) -> int:
         refresh_timeout = self.CONST_REFRESH_TIMEOUT
         if (
-                self.user
-                and self.user.role_type == "DOMAIN_ADMIN"
-                and self.user.refresh_timeout
+            self.user
+            and self.user.role_type == "DOMAIN_ADMIN"
+            and self.user.refresh_timeout
         ):
             refresh_timeout = max(self.user.refresh_timeout, refresh_timeout)
 

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

@@ -8,6 +8,7 @@
     "UserVerifyEmailRequest",
     "UserStatQueryRequest",
     "UserSetRequiredActionsRequest",
+    "UserSetRefreshTimeout",
     "UserDisableMFARequest",
     "UserDeleteRequest",
     "UserEnableRequest",
@@ -63,6 +64,12 @@ class UserSetRequiredActionsRequest(BaseModel):
     domain_id: str
 
 
+class UserSetRefreshTimeout(BaseModel):
+    user_id: str
+    refresh_timeout: int
+    domain_id: str
+
+
 class UserDeleteRequest(BaseModel):
     user_id: str
     domain_id: str

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

@@ -4,7 +4,6 @@
 
 __all__ = [
     "UserProfileUpdateRequest",
-    "UserProfileSetRefreshTokenTimeout",
     "UserProfileVerifyEmailRequest",
     "UserProfileConfirmEmailRequest",
     "UserProfileResetPasswordRequest",
@@ -28,12 +27,6 @@ class UserProfileUpdateRequest(BaseModel):
     domain_id: str
 
 
-class UserProfileSetRefreshTokenTimeout(BaseModel):
-    user_id: str
-    refresh_timeout: int
-    domain_id: str
-
-
 class UserProfileVerifyEmailRequest(BaseModel):
     user_id: str
     email: Union[str, None] = None

src/spaceone/identity/service/user_profile_service.py

@@ -82,38 +82,6 @@ def update(self, params: UserProfileUpdateRequest) -> Union[UserResponse, dict]:
 
         return UserResponse(**user_vo.to_dict())
 
-    @transaction(permission="identity:UserProfile.write", role_types=["USER"])
-    @convert_model
-    def set_refresh_timeout(
-        self, params: UserProfileSetRefreshTokenTimeout
-    ) -> Union[UserResponse, dict]:
-        """
-        Args:
-            params (UserProfileSetRefreshTokenTimeout): {
-                "refresh_timeout": "int",
-                "user_id": "str",           # inject from auth
-                "domain_id": "str"          # inject from auth
-            }
-        Returns:
-            UserResponse:
-        """
-
-        user_id = params.user_id
-        domain_id = params.domain_id
-        user_vo = self.user_mgr.get_user(user_id, domain_id)
-
-        if user_vo.role_type != "DOMAIN_ADMIN":
-            raise ERROR_PERMISSION_DENIED()
-
-        refresh_timeout = self._get_refresh_timeout_from_config(params.refresh_timeout)
-        print(refresh_timeout)
-        user_vo = self.user_mgr.update_user_by_vo(
-            {"refresh_timeout": refresh_timeout}, user_vo
-        )
-
-        print(user_vo.refresh_timeout)
-        return UserResponse(**user_vo.to_dict())
-
     @transaction(permission="identity:UserProfile.write", role_types=["USER"])
     @convert_model
     def verify_email(self, params: UserProfileVerifyEmailRequest) -> None:
@@ -652,20 +620,3 @@ def _get_my_workspace_groups_info(
     def _check_mfa_options(options, mfa_type):
         if mfa_type in ["EMAIL"] and not options:
             raise ERROR_REQUIRED_PARAMETER(key="options.email")
-
-    @staticmethod
-    def _get_refresh_timeout_from_config(refresh_timeout: int) -> int:
-        identity_conf = config.get_global("IDENTITY") or {}
-        token_conf = identity_conf.get("token", {})
-        config_refresh_timeout = token_conf.get("refresh_timeout")
-        if refresh_timeout < config_refresh_timeout:
-            raise ERROR_INVALID_PARAMETER(
-                key="refresh_timeout",
-                reason=f"Minimum value for refresh_timeout is {config_refresh_timeout}",
-            )
-        refresh_timeout = max(refresh_timeout, config_refresh_timeout)
-
-        config_admin_refresh_timeout = token_conf.get("admin_refresh_timeout", 2592000)
-        refresh_timeout = min(refresh_timeout, config_admin_refresh_timeout)
-
-        return refresh_timeout

src/spaceone/identity/service/user_service.py

@@ -293,6 +293,36 @@ def set_required_actions(
 
         return UserResponse(**user_vo.to_dict())
 
+    @transaction(permission="identity:User.write", role_types=["DOMAIN_ADMIN"])
+    @convert_model
+    def set_refresh_timeout(
+        self, params: UserSetRefreshTimeout
+    ) -> Union[UserResponse, dict]:
+        """
+        Args:
+            params (UserProfileSetRefreshTimeout): {
+                "user_id": "str",
+                "refresh_timeout": "int",
+                "domain_id": "str"          # inject from auth
+            }
+        Returns:
+            UserResponse:
+        """
+
+        user_id = params.user_id
+        domain_id = params.domain_id
+        user_vo = self.user_mgr.get_user(user_id, domain_id)
+
+        if user_vo.role_type != "DOMAIN_ADMIN":
+            raise ERROR_PERMISSION_DENIED()
+
+        refresh_timeout = self._get_refresh_timeout_from_config(params.refresh_timeout)
+        user_vo = self.user_mgr.update_user_by_vo(
+            {"refresh_timeout": refresh_timeout}, user_vo
+        )
+
+        return UserResponse(**user_vo.to_dict())
+
     @transaction(permission="identity:User.write", role_types=["DOMAIN_ADMIN"])
     @convert_model
     def delete(self, params: UserDeleteRequest) -> None:
@@ -527,3 +557,20 @@ def _get_domain_default_language(domain_id: str, language: str = None) -> str:
             else:
                 language = "en"
         return language
+
+    @staticmethod
+    def _get_refresh_timeout_from_config(refresh_timeout: int) -> int:
+        identity_conf = config.get_global("IDENTITY") or {}
+        token_conf = identity_conf.get("token", {})
+        config_refresh_timeout = token_conf.get("refresh_timeout")
+        if refresh_timeout < config_refresh_timeout:
+            raise ERROR_INVALID_PARAMETER(
+                key="refresh_timeout",
+                reason=f"Minimum value for refresh_timeout is {config_refresh_timeout}",
+            )
+        refresh_timeout = max(refresh_timeout, config_refresh_timeout)
+
+        config_admin_refresh_timeout = token_conf.get("admin_refresh_timeout", 2592000)
+        refresh_timeout = min(refresh_timeout, config_admin_refresh_timeout)
+
+        return refresh_timeout