Merge branch 'develop' into 'main'

Develop

See merge request locker/api-core!466

locker_server/api/permissions/locker_permissions/backup_credential_pwd_permission.py

@@ -3,7 +3,7 @@
 
 class BackupCredentialPwdPermission(APIPermission):
     def has_permission(self, request, view):
-        return self.is_auth(request) and request.user.activated
+        return self.is_auth(request)
 
     def has_object_permission(self, request, view, obj):
         user = request.user

locker_server/api_orm/model_parsers/cipher_parsers.py

@@ -11,7 +11,8 @@
 
 class CipherParser:
     @classmethod
-    def parse_cipher(cls, cipher_orm: CipherORM, parse_collection_ids=False, parse_histories=False) -> Cipher:
+    def parse_cipher(cls, cipher_orm: CipherORM, parse_collection_ids=False, parse_histories=False,
+                     limit_history: int = None) -> Cipher:
         user_parser = get_specific_model_parser("UserParser")
         team_parser = get_specific_model_parser("TeamParser")
         try:
@@ -46,11 +47,18 @@ def parse_cipher(cls, cipher_orm: CipherORM, parse_collection_ids=False, parse_h
             cipher.collection_ids = collection_ids
 
         if parse_histories is True:
-            cipher.history = cls.parse_password_history(cipher_orm=cipher_orm)
+            try:
+                show_history = getattr(cipher_orm, "show_history")
+            except AttributeError:
+                show_history = True
+            if show_history is False:
+                cipher.history = []
+            else:
+                cipher.history = cls.parse_password_history(cipher_orm=cipher_orm, limit_history=limit_history)
         return cipher
 
     @classmethod
-    def parse_password_history(cls, cipher_orm: CipherORM) -> List:
+    def parse_password_history(cls, cipher_orm: CipherORM, limit_history: int = None) -> List:
         history = []
         histories_orm = cipher_orm.cipher_histories.order_by('creation_date').values('last_use_date', 'data')
         for history_orm in histories_orm:
@@ -60,6 +68,9 @@ def parse_password_history(cls, cipher_orm: CipherORM) -> List:
                 "last_used_date": convert_readable_date(history_orm.get("last_use_date")),
                 "password": data.get("password")
             })
+        if limit_history is not None:
+            history = history[:limit_history]
+            # history = history[-limit_history:]
         return history
 
     @classmethod

locker_server/api_orm/repositories/cipher_repository.py

@@ -127,8 +127,20 @@ def _get_multiple_ciphers_orm_by_user(user_id: int, only_personal=False, only_ma
                 default=True,
                 output_field=BooleanField()
             )
+        ).annotate(
+            show_history=Case(
+                When(
+                    Q(
+                        team__team_members__role_id__in=[MEMBER_ROLE_ADMIN, MEMBER_ROLE_MANAGER, MEMBER_ROLE_MEMBER],
+                        team__team_members__user_id=user_id,
+                    ), then=False
+                ),
+                default=True,
+                output_field=BooleanField()
+            )
         )
         hide_password_cipher_ids = team_ciphers_orm.filter(view_password=False).values_list('id', flat=True)
+        hide_history_cipher_ids = team_ciphers_orm.filter(show_history=False).values_list('id', flat=True)
         if only_edited:
             team_ciphers_orm = team_ciphers_orm.filter(
                 team__team_members__role_id__in=[MEMBER_ROLE_OWNER, MEMBER_ROLE_ADMIN, MEMBER_ROLE_MANAGER],
@@ -149,6 +161,12 @@ def _get_multiple_ciphers_orm_by_user(user_id: int, only_personal=False, only_ma
                 default=True,
                 output_field=BooleanField()
             )
+        ).annotate(
+            show_history=Case(
+                When(id__in=hide_history_cipher_ids, then=False),
+                default=True,
+                output_field=BooleanField()
+            )
         ).order_by('-revision_date')  # .prefetch_related('collections_ciphers')
 
         collection_id_param = filters.get("collection_id")
@@ -164,7 +182,8 @@ def list_cipher_collection_ids(self, cipher_id: str) -> List[str]:
 
     def get_multiple_by_user(self, user_id: int, only_personal=False, only_managed_team=False,
                              only_edited=False, only_deleted=False,
-                             exclude_team_ids=None, filter_ids=None, exclude_types=None) -> List[Cipher]:
+                             exclude_team_ids=None, filter_ids=None, exclude_types=None,
+                             limit_history: int = None) -> List[Cipher]:
         """
         Get list ciphers of user
         :param user_id: (int) The user id
@@ -175,6 +194,7 @@ def get_multiple_by_user(self, user_id: int, only_personal=False, only_managed_t
         :param exclude_team_ids: (list) Excluding all ciphers have team_id in this list
         :param filter_ids: (list) List filtered cipher ids
         :param exclude_types: (list) Excluding all ciphers have type in this list
+        :param limit_history: (int) Limit the number of item histories
         :return:
         """
 
@@ -184,7 +204,7 @@ def get_multiple_by_user(self, user_id: int, only_personal=False, only_managed_t
             exclude_team_ids=exclude_team_ids, filter_ids=filter_ids, exclude_types=exclude_types
         ).prefetch_related('collections_ciphers').prefetch_related('cipher_histories')
         return [ModelParser.cipher_parser().parse_cipher(
-            cipher_orm=c, parse_collection_ids=True, parse_histories=True
+            cipher_orm=c, parse_collection_ids=True, parse_histories=True, limit_history=limit_history
         ) for c in ciphers_orm]
 
     def get_ciphers_created_by_user(self, user_id: int) -> List[Cipher]:
@@ -247,7 +267,7 @@ def check_member_belongs_cipher_collections(self, cipher: Cipher, member: TeamMe
 
     def sync_and_statistic_ciphers(self, user_id: int, only_personal=False, only_managed_team=False,
                                    only_edited=False, only_deleted=False,
-                                   exclude_team_ids=None, filter_ids=None, exclude_types=None,
+                                   exclude_team_ids=None, filter_ids=None, exclude_types=None, limit_history=None,
                                    **ciphers_filter) -> Dict:
         ciphers_orm = self._get_multiple_ciphers_orm_by_user(
             user_id=user_id, only_personal=only_personal, only_managed_team=only_managed_team,
@@ -273,7 +293,7 @@ def sync_and_statistic_ciphers(self, user_id: int, only_personal=False, only_man
             },
             "ciphers": [
                 ModelParser.cipher_parser().parse_cipher(
-                    cipher_orm=c, parse_collection_ids=True, parse_histories=True
+                    cipher_orm=c, parse_collection_ids=True, parse_histories=True, limit_history=limit_history
                 ) for c in ciphers_orm
             ]
         }
@@ -521,6 +541,7 @@ def update_cipher(self, cipher_id: str, cipher_data: Dict) -> Cipher:
         user_cipher_id = cipher_data.get("user_id")
         team_id = cipher_data.get("team_id")
         collection_ids = cipher_data.get("collection_ids", [])
+        limit_history = cipher_data.get("limit_history")
 
         # If team_id is not null => This cipher belongs to team
         if team_id:
@@ -529,9 +550,9 @@ def update_cipher(self, cipher_id: str, cipher_data: Dict) -> Cipher:
         # Create new cipher history
         if cipher_orm.type in SAVE_HISTORY_CIPHER_TYPES:
             password_history = cipher_data.get("password_history") or []
-            limit_history = cipher_data.get("limit_history")
-            if limit_history and len(password_history) > limit_history:
-                password_history = password_history[-limit_history:]
+            # limit_history = cipher_data.get("limit_history")
+            # if limit_history and len(password_history) > limit_history:
+            #     password_history = password_history[-limit_history:]
             if cipher_orm.type == CIPHER_TYPE_LOGIN and len(password_history) > cipher_orm.cipher_histories.count():
                 num = len(password_history)-cipher_orm.cipher_histories.count()
                 password_histories_data = password_history[:num]
@@ -583,7 +604,9 @@ def update_cipher(self, cipher_id: str, cipher_data: Dict) -> Cipher:
         })
         bump_account_revision_date(user=cipher_orm.user)
 
-        return ModelParser.cipher_parser().parse_cipher(cipher_orm=cipher_orm, parse_histories=True)
+        return ModelParser.cipher_parser().parse_cipher(
+            cipher_orm=cipher_orm, parse_histories=True, limit_history=limit_history
+        )
 
     def update_folders(self, cipher_id: str, new_folders_data) -> Cipher:
         cipher_orm = self._get_cipher_orm(cipher_id=cipher_id)

locker_server/api_orm/repositories/user_plan_repository.py

@@ -1,6 +1,6 @@
 import ast
 import math
-from typing import Dict, Optional, List, Tuple
+from typing import Dict, Optional, List, Tuple, Union
 
 from django.conf import settings
 from django.core.exceptions import MultipleObjectsReturned
@@ -226,8 +226,9 @@ def get_default_enterprise(self, user_id: int, enterprise_name: str = None,
             multiple_default_enterprises_orm.exclude(enterprise_id=default_enterprise_orm.id).delete()
         return ModelParser.enterprise_parser().parse_enterprise(enterprise_orm=default_enterprise_orm)
 
-    def get_max_allow_cipher_type(self, user: User) -> Dict:
-        user_orm = self._get_user_orm(user_id=user.user_id)
+    def get_max_allow_cipher_type(self, user: Union[User, int]) -> Dict:
+        user_id = user if isinstance(user, int) else user.user_id
+        user_orm = self._get_user_orm(user_id=user_id)
         user_enterprise_ids = user_orm.enterprise_members.filter(
             status=E_MEMBER_STATUS_CONFIRMED, is_activated=True,
             enterprise__locked=False
@@ -236,7 +237,7 @@ def get_max_allow_cipher_type(self, user: User) -> Dict:
             role_id=E_MEMBER_ROLE_PRIMARY_ADMIN
         ).values_list('user_id', flat=True)
         personal_plans_orm = PMUserPlanORM.objects.filter(
-            user_id__in=list(primary_admins) + [user.user_id]
+            user_id__in=list(primary_admins) + [user_id]
         ).select_related('pm_plan')
         cipher_limits = PMPlanORM.objects.filter(id__in=personal_plans_orm.values_list('pm_plan_id')).values(
             'limit_password', 'limit_secure_note', 'limit_identity', 'limit_payment_card', 'limit_crypto_asset',

locker_server/core/repositories/cipher_repository.py

@@ -17,7 +17,8 @@ def list_cipher_collection_ids(self, cipher_id: str) -> List[str]:
     @abstractmethod
     def get_multiple_by_user(self, user_id: int, only_personal=False, only_managed_team=False,
                              only_edited=False, only_deleted=False,
-                             exclude_team_ids=None, filter_ids=None, exclude_types=None) -> List[Cipher]:
+                             exclude_team_ids=None, filter_ids=None, exclude_types=None,
+                             limit_history=None) -> List[Cipher]:
         pass
 
     @abstractmethod
@@ -68,7 +69,7 @@ def check_member_belongs_cipher_collections(self, cipher: Cipher, member: TeamMe
     @abstractmethod
     def sync_and_statistic_ciphers(self, user_id: int, only_personal=False, only_managed_team=False,
                                    only_edited=False, only_deleted=False,
-                                   exclude_team_ids=None, filter_ids=None, exclude_types=None,
+                                   exclude_team_ids=None, filter_ids=None, exclude_types=None, limit_history=None,
                                    **ciphers_filter) -> Dict:
         pass
 

locker_server/core/repositories/user_plan_repository.py

@@ -42,7 +42,7 @@ def get_default_enterprise(self, user_id: int, enterprise_name: str = None,
         pass
 
     @abstractmethod
-    def get_max_allow_cipher_type(self, user: User) -> Dict:
+    def get_max_allow_cipher_type(self, user: Union[User, int]) -> Dict:
         pass
 
     @abstractmethod

locker_server/core/services/cipher_service.py

@@ -194,9 +194,7 @@ def update_cipher(self, cipher: Cipher, user: User, cipher_data: Dict, view_acti
                         raise CollectionCannotAddException(collection_id=member_collection_id)
 
         # Validate plan
-        allow_cipher_type = self.user_plan_repository.get_max_allow_cipher_type(user=user)
-        limit_history = allow_cipher_type.get("limit_history")
-        cipher_data.update({"limit_history": limit_history})
+        cipher_data.update({"limit_history": self.get_limit_history(user_id=user_id)})
         cipher = self.cipher_repository.update_cipher(cipher_id=cipher.cipher_id, cipher_data=cipher_data)
         return cipher
 
@@ -234,20 +232,23 @@ def get_multiple_by_ids(self, cipher_ids: List[str]) -> List[Cipher]:
     def get_multiple_by_user(self, user_id: int, only_personal=False, only_managed_team=False,
                              only_edited=False, only_deleted=False,
                              exclude_team_ids=None, filter_ids=None, exclude_types=None) -> List[Cipher]:
+        limit_history = self.get_limit_history(user_id=user_id)
         return self.cipher_repository.get_multiple_by_user(
             user_id=user_id, only_personal=only_personal, only_managed_team=only_managed_team,
             only_edited=only_edited, only_deleted=only_deleted, exclude_team_ids=exclude_team_ids,
-            filter_ids=filter_ids, exclude_types=exclude_types
+            filter_ids=filter_ids, exclude_types=exclude_types, limit_history=limit_history
         )
 
     def sync_and_statistic_ciphers(self, user_id: int, only_personal=False, only_managed_team=False,
                                    only_edited=False, only_deleted=False,
                                    exclude_team_ids=None, filter_ids=None, exclude_types=None,
                                    **ciphers_filter) -> Dict:
+        limit_history = self.get_limit_history(user_id=user_id)
         return self.cipher_repository.sync_and_statistic_ciphers(
             user_id=user_id, only_personal=only_personal, only_managed_team=only_managed_team,
             only_edited=only_edited, only_deleted=only_deleted,
             exclude_team_ids=exclude_team_ids, filter_ids=filter_ids, exclude_types=exclude_types,
+            limit_history=limit_history,
             **ciphers_filter
         )
 
@@ -293,3 +294,7 @@ def statistic_multiple_cipher_by_user_id(self, user_id: int, only_personal=False
             filter_ids=filter_ids,
             exclude_types=exclude_types
         )
+
+    def get_limit_history(self, user_id: int):
+        allow_cipher_type = self.user_plan_repository.get_max_allow_cipher_type(user_id)
+        return allow_cipher_type.get("limit_history")

locker_server/core/services/sharing_service.py

@@ -245,7 +245,7 @@ def _validate_cipher(self, user: User, cipher: Dict) -> Optional[Cipher]:
         if not cipher:
             return None
         cipher_obj = self.cipher_repository.get_by_id(cipher_id=cipher.get("id"))
-        if not cipher:
+        if not cipher_obj:
             raise CipherDoesNotExistException
         # If the cipher isn't shared?
         if cipher_obj.user and cipher_obj.user.user_id != user.user_id: