feat: add new feature for setting refresh timeout for domain admin user

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

src/spaceone/identity/conf/global_conf.py

@@ -54,6 +54,7 @@
         "token_timeout": 1800,  # 30 minutes
         "token_max_timeout": 604800,  # 7 days
         "refresh_timeout": 10800,  # 3 hours
+        "admin_refresh_timeout": 2592000,  # 30days
     },
     "mfa": {"verify_code_timeout": 300},
 }

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

@@ -1,6 +1,5 @@
 from google.protobuf.json_format import ParseDict
-from spaceone.api.identity.v2 import (user_pb2, user_profile_pb2,
-                                      user_profile_pb2_grpc)
+from spaceone.api.identity.v2 import user_pb2, user_profile_pb2, user_profile_pb2_grpc
 from spaceone.core.pygrpc import BaseAPI
 
 from spaceone.identity.service.user_profile_service import UserProfileService
@@ -16,6 +15,12 @@ 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()
@@ -80,7 +80,7 @@ def issue_token(
         )
 
         refresh_token = key_gen.generate_token(
-            "REFRESH_TOKEN", timeout=self.CONST_REFRESH_TIMEOUT
+            "REFRESH_TOKEN", timeout=self._get_refresh_token_timeout()
         )
         if self.owner_type != "SYSTEM":
             # todo: remove
@@ -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, timeout: int
+            self, user_id: str, domain_id: str, private_jwk: dict, timeout: int
     ) -> dict:
         permissions = [
             "identity:UserProfile",
@@ -127,6 +127,17 @@ def set_timeout(self, timeout: Union[int, None]) -> int:
             timeout = self.CONST_TOKEN_TIMEOUT
         return timeout
 
+    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
+        ):
+            refresh_timeout = max(self.user.refresh_timeout, refresh_timeout)
+
+        return refresh_timeout
+
     @staticmethod
     def check_verify_code(user_id, domain_id, verify_code):
         if cache.is_set():

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

@@ -36,7 +36,8 @@ class User(MongoModel):
     required_actions = ListField(StringField(choices=("UPDATE_PASSWORD",)), default=[])
     language = StringField(max_length=7, default="en")
     timezone = StringField(max_length=50, default="UTC")
-    tags = DictField(Default=None)
+    refresh_timeout = IntField(default=None, min_value=1800, max_value=2592000)
+    tags = DictField(default=None)
     domain_id = StringField(max_length=40)
     created_at = DateTimeField(auto_now_add=True)
     last_accessed_at = DateTimeField(default=None, null=True)
@@ -54,6 +55,7 @@ class User(MongoModel):
             "language",
             "timezone",
             "required_actions",
+            "refresh_timeout",
             "tags",
             "last_accessed_at",
         ],

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

@@ -27,6 +27,7 @@ class UserResponse(BaseModel):
     language: Union[str, None] = None
     timezone: Union[str, None] = None
     required_actions: Union[List[str], None] = None
+    refresh_timeout: Union[int, None] = None
     tags: Union[dict, None] = None
     domain_id: Union[str, None] = None
     created_at: Union[datetime, None] = None

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

@@ -4,6 +4,7 @@
 
 __all__ = [
     "UserProfileUpdateRequest",
+    "UserProfileSetRefreshTokenTimeout",
     "UserProfileVerifyEmailRequest",
     "UserProfileConfirmEmailRequest",
     "UserProfileResetPasswordRequest",
@@ -27,6 +28,12 @@ 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,6 +82,38 @@ 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:
@@ -620,3 +652,20 @@ 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

@@ -254,7 +254,9 @@ def disable_mfa(self, params: UserDisableMFARequest) -> Union[UserResponse, dict
         if mfa_type == "OTP":
             user_secret_id = user_mfa.get("options", {}).get("user_secret_id")
             secret_manager: SecretManager = self.locator.get_manager(SecretManager)
-            secret_manager.delete_user_secret_with_system_token(domain_id, user_secret_id)
+            secret_manager.delete_user_secret_with_system_token(
+                domain_id, user_secret_id
+            )
 
         user_mfa = {"state": "DISABLED"}
         self.user_mgr.update_user_by_vo({"mfa": user_mfa}, user_vo)