Merge branch 'feat/item-history' into 'develop'

Feat/item history

See merge request locker/api-core!449

locker_server/api/v1_0/ciphers/views.py

@@ -168,7 +168,6 @@ def retrieve(self, request, *args, **kwargs):
         user = self.request.user
         self.check_pwd_session_auth(request=request)
         cipher = self.get_object()
-
         try:
             cipher_obj = self.cipher_service.get_multiple_by_user(
                 user_id=user.user_id, filter_ids=[cipher.cipher_id]

locker_server/api/v1_0/sync/serializers.py

@@ -133,7 +133,7 @@ def to_representation(self, instance):
             "notes": data.get("notes"),
             "organization_id": instance.team.team_id if instance.team else None,
             "organization_use_totp": True if login else False,
-            "password_history": None,
+            "password_history": instance.history,
             "reprompt": instance.reprompt,
             "revision_date": convert_readable_date(instance.revision_date),
             "creation_date": convert_readable_date(instance.creation_date),

locker_server/api_orm/fixtures/plan.json

@@ -67,40 +67,6 @@
       "plan_type": "Basic"
     }
   },
-  {
-    "model": "api_orm.PMPlan",
-    "fields": {
-      "name": "Lifetime Premium",
-      "alias": "pm_lifetime_premium",
-      "price_usd": 0,
-      "price_vnd": 0,
-      "yearly_price_usd": 0,
-      "yearly_price_vnd": 0,
-      "half_yearly_price_usd": 0,
-      "half_yearly_price_vnd": 0,
-      "sync_device": null,
-      "limit_password": null,
-      "limit_secure_note": null,
-      "limit_identity": null,
-      "limit_payment_card": null,
-      "limit_crypto_asset": null,
-      "limit_totp": null,
-      "tools_password_reuse": true,
-      "tools_master_password_check": true,
-      "tools_data_breach": true,
-      "emergency_access": true,
-      "personal_share": true,
-      "is_family_plan": false,
-      "is_team_plan": false,
-      "max_number": 1,
-      "team_dashboard": false,
-      "team_policy": false,
-      "team_prevent_password": false,
-      "team_activity_log": false,
-      "relay_premium": true,
-      "plan_type": "Basic"
-    }
-  },
   {
     "model": "api_orm.PMPlan",
     "fields": {

locker_server/api_orm/migrations/0001_initial.py

@@ -871,6 +871,7 @@ class Migration(migrations.Migration):
                 ('limit_identity', models.IntegerField(default=None, null=True)),
                 ('limit_payment_card', models.IntegerField(default=None, null=True)),
                 ('limit_crypto_asset', models.IntegerField(default=None, null=True)),
+                ('limit_totp', models.IntegerField(default=None, null=True)),
                 ('tools_password_reuse', models.BooleanField(default=False)),
                 ('tools_master_password_check', models.BooleanField(default=False)),
                 ('tools_data_breach', models.BooleanField(default=False)),

locker_server/api_orm/migrations/0021_auto_20240312_1502.py

@@ -40,10 +40,5 @@ class Migration(migrations.Migration):
     ]
 
     operations = [
-        migrations.AddField(
-            model_name='pmplanorm',
-            name='limit_totp',
-            field=models.IntegerField(default=None, null=True),
-        ),
         migrations.RunPython(update_records),
     ]

locker_server/api_orm/migrations/0025_auto_20240530_1032.py

@@ -0,0 +1,32 @@
+# Generated by Django 3.2.23 on 2024-05-30 03:32
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('api_orm', '0024_auto_20240528_1045'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='CipherHistoryORM',
+            fields=[
+                ('id', models.CharField(default=uuid.uuid4, max_length=128, primary_key=True, serialize=False)),
+                ('creation_date', models.FloatField()),
+                ('revision_date', models.FloatField()),
+                ('last_use_date', models.FloatField(null=True)),
+                ('reprompt', models.IntegerField(default=0)),
+                ('score', models.FloatField(default=0)),
+                ('data', models.TextField(blank=True, null=True)),
+                ('cipher', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='cipher_histories', to=settings.LS_CIPHER_MODEL)),
+            ],
+            options={
+                'db_table': 'cs_cipher_histories',
+            },
+        ),
+    ]

locker_server/api_orm/model_parsers/cipher_parsers.py

@@ -1,12 +1,16 @@
+import ast
+from typing import List
+
 from locker_server.api_orm.model_parsers.wrapper_specific_model_parser import get_specific_model_parser
 from locker_server.api_orm.models import *
 from locker_server.core.entities.cipher.cipher import Cipher
+from locker_server.core.entities.cipher.cipher_history import CipherHistory
 from locker_server.core.entities.cipher.folder import Folder
 
 
 class CipherParser:
     @classmethod
-    def parse_cipher(cls, cipher_orm: CipherORM, parse_collection_ids=False) -> Cipher:
+    def parse_cipher(cls, cipher_orm: CipherORM, parse_collection_ids=False, parse_histories=False) -> Cipher:
         user_parser = get_specific_model_parser("UserParser")
         team_parser = get_specific_model_parser("TeamParser")
         try:
@@ -39,8 +43,24 @@ def parse_cipher(cls, cipher_orm: CipherORM, parse_collection_ids=False) -> Ciph
             except AttributeError:
                 pass
             cipher.collection_ids = collection_ids
+
+        if parse_histories is True:
+            cipher.history = cls.parse_password_history(cipher_orm=cipher_orm)
         return cipher
 
+    @classmethod
+    def parse_password_history(cls, cipher_orm: CipherORM) -> List:
+        history = []
+        histories_orm = cipher_orm.cipher_histories.order_by('creation_date').values('last_use_date', 'data')
+        for history_orm in histories_orm:
+            data = history_orm.get("data")
+            data = {} if not data else ast.literal_eval(str(data))
+            history.append({
+                "last_use_date": history_orm.get("last_use_date"),
+                "password": data.get("password")
+            })
+        return history
+
     @classmethod
     def parse_folder(cls, folder_orm: FolderORM) -> Folder:
         user_parser = get_specific_model_parser("UserParser")
@@ -52,3 +72,18 @@ def parse_folder(cls, folder_orm: FolderORM) -> Folder:
             user=user_parser.parse_user(user_orm=folder_orm.user)
         )
 
+    @classmethod
+    def parse_cipher_history(cls, cipher_history_orm: CipherHistoryORM, parse_cipher=False) -> CipherHistory:
+        cipher = None
+        if parse_cipher is True:
+            cipher = cls.parse_cipher(cipher_orm=cipher_history_orm.cipher)
+        return CipherHistory(
+            cipher_history_id=cipher_history_orm.id,
+            creation_date=cipher_history_orm.creation_date,
+            revision_date=cipher_history_orm.revision_date,
+            last_use_date=cipher_history_orm.last_use_date,
+            reprompt=cipher_history_orm.reprompt,
+            score=cipher_history_orm.score,
+            data=cipher_history_orm.get_data(),
+            cipher=cipher,
+        )

locker_server/api_orm/models/__init__.py

@@ -42,6 +42,7 @@
 # ------------------------ Sharing Models ----------------------------- #
 from locker_server.api_orm.models.teams.teams import TeamORM
 from locker_server.api_orm.models.ciphers.ciphers import CipherORM
+from locker_server.api_orm.models.ciphers.cipher_histories import CipherHistoryORM
 from locker_server.api_orm.models.members.member_roles import MemberRoleORM
 from locker_server.api_orm.models.members.team_members import TeamMemberORM
 from locker_server.api_orm.models.teams.collections import CollectionORM

locker_server/api_orm/models/ciphers/cipher_histories.py

@@ -0,0 +1,42 @@
+import ast
+import uuid
+
+from django.db import models
+
+from locker_server.settings import locker_server_settings
+from locker_server.shared.utils.app import now
+
+
+class CipherHistoryORM(models.Model):
+    id = models.CharField(primary_key=True, max_length=128, default=uuid.uuid4)
+    creation_date = models.FloatField()
+    revision_date = models.FloatField()
+    last_use_date = models.FloatField(null=True)
+    reprompt = models.IntegerField(default=0)
+    score = models.FloatField(default=0)
+    data = models.TextField(blank=True, null=True)
+    cipher = models.ForeignKey(
+        locker_server_settings.LS_CIPHER_MODEL, on_delete=models.CASCADE, related_name="cipher_histories"
+    )
+
+    class Meta:
+        db_table = 'cs_cipher_histories'
+
+    @classmethod
+    def create(cls, cipher, **data):
+        cipher_history_orm = cls(
+            creation_date=data.get("creation_date", now()),
+            revision_date=data.get("revision_date", now()),
+            last_use_date=data.get("last_use_date"),
+            reprompt=data.get("reprompt", 0),
+            score=data.get("score", 0),
+            data=data.get("data"),
+            cipher=cipher
+        )
+        cipher_history_orm.save()
+        return cipher_history_orm
+
+    def get_data(self):
+        if not self.data:
+            return {}
+        return ast.literal_eval(str(self.data))

locker_server/api_orm/repositories/cipher_repository.py

@@ -182,8 +182,10 @@ def get_multiple_by_user(self, user_id: int, only_personal=False, only_managed_t
             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
-        ).prefetch_related('collections_ciphers')
-        return [ModelParser.cipher_parser().parse_cipher(cipher_orm=c, parse_collection_ids=True) for c in ciphers_orm]
+        ).prefetch_related('collections_ciphers').prefetch_related('cipher_histories')
+        return [ModelParser.cipher_parser().parse_cipher(
+            cipher_orm=c, parse_collection_ids=True, parse_histories=True
+        ) for c in ciphers_orm]
 
     def get_ciphers_created_by_user(self, user_id: int) -> List[Cipher]:
         ciphers_orm = CipherORM.objects.filter(created_by_id=user_id)
@@ -205,6 +207,12 @@ def list_cipher_ids_by_folder_id(self, user_id: int, folder_id: str) -> List[str
     def list_cipher_ids_by_collection_id(self, collection_id: str) -> List[str]:
         return list(CollectionCipherORM.objects.filter(collection_id=collection_id).values_list('cipher_id', flat=True))
 
+    def list_password_histories(self, cipher_id: str) -> List:
+        cipher_orm = self._get_cipher_orm(cipher_id=cipher_id)
+        if not cipher_orm:
+            return None
+        return ModelParser.cipher_parser().parse_password_history(cipher_orm=cipher_orm)
+
     # ------------------------ Get Cipher resource --------------------- #
     def get_by_id(self, cipher_id: str) -> Optional[Cipher]:
         cipher_orm = self._get_cipher_orm(cipher_id=cipher_id)
@@ -248,7 +256,7 @@ def sync_and_statistic_ciphers(self, user_id: int, only_personal=False, only_man
             **ciphers_filter
         ).select_related('user').select_related('created_by').select_related('team').prefetch_related(
             'collections_ciphers'
-        )
+        ).prefetch_related('cipher_histories')
         total_cipher = ciphers_orm.count()
         not_deleted_ciphers_orm = ciphers_orm.filter(deleted_date__isnull=True)
         not_deleted_ciphers_statistic = not_deleted_ciphers_orm.values('type').annotate(
@@ -264,7 +272,9 @@ 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) for c in ciphers_orm
+                ModelParser.cipher_parser().parse_cipher(
+                    cipher_orm=c, parse_collection_ids=True, parse_histories=True
+                ) for c in ciphers_orm
             ]
         }
 
@@ -515,6 +525,20 @@ def update_cipher(self, cipher_id: str, cipher_data: Dict) -> Cipher:
         # If team_id is not null => This cipher belongs to team
         if team_id:
             user_cipher_id = None
+
+        # Create new cipher history
+        if cipher_orm.type in SAVE_HISTORY_CIPHER_TYPES:
+            new_data = cipher_data.get("data", cipher_orm.get_data())
+            if cipher_orm.type == CIPHER_TYPE_LOGIN and \
+                    cipher_orm.get_data().get("password") != new_data.get("password"):
+                history_data = {
+                    "last_use_date": cipher_orm.last_use_date,
+                    "reprompt": cipher_orm.reprompt,
+                    "score": cipher_orm.score,
+                    "data": cipher_orm.data,
+                }
+                cipher_orm.cipher_histories.model.create(cipher_orm, **history_data)
+
         # Create new cipher object
         cipher_orm.revision_date = now()
         cipher_orm.reprompt = cipher_data.get("reprompt", cipher_orm.reprompt) or 0
@@ -550,7 +574,7 @@ 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)
+        return ModelParser.cipher_parser().parse_cipher(cipher_orm=cipher_orm, parse_histories=True)
 
     def update_folders(self, cipher_id: str, new_folders_data) -> Cipher:
         cipher_orm = self._get_cipher_orm(cipher_id=cipher_id)

locker_server/core/entities/cipher/cipher.py

@@ -1,11 +1,12 @@
 from locker_server.core.entities.team.team import Team
 from locker_server.core.entities.user.user import User
+from locker_server.shared.constants.ciphers import CIPHER_TYPE_LOGIN
 
 
 class Cipher(object):
     def __init__(self, cipher_id: str, creation_date: float = None, revision_date: float = None,
                  deleted_date: float = None, last_use_date: float = None, num_use: int = 0, reprompt: int = 0,
-                 score: float = 0,  cipher_type: int = None, data: str = None, favorites: str = "", folders: str = "",
+                 score: float = 0,  cipher_type: int = None, data=None, favorites: str = "", folders: str = "",
                  view_password: bool = True, user: User = None, created_by: User = None, team: Team = None):
         self._cipher_id = cipher_id
         self._creation_date = creation_date
@@ -20,6 +21,7 @@ def __init__(self, cipher_id: str, creation_date: float = None, revision_date: f
         self._favorites = favorites
         self._folders = folders
         self._view_password = view_password
+        self._history = []
         self._collection_ids = []
         self._user = user
         self._created_by = created_by
@@ -63,7 +65,10 @@ def cipher_type(self):
 
     @property
     def data(self):
-        return self._data
+        d = self._data
+        if self._cipher_type == CIPHER_TYPE_LOGIN:
+            d.update({"password_history": self.history})
+        return d
 
     @property
     def favorites(self):
@@ -100,3 +105,11 @@ def collection_ids(self):
     @collection_ids.setter
     def collection_ids(self, collection_ids_value):
         self._collection_ids = collection_ids_value
+
+    @property
+    def history(self):
+        return self._history
+
+    @history.setter
+    def history(self, history_value):
+        self._history = history_value

locker_server/core/entities/cipher/cipher_history.py

@@ -0,0 +1,47 @@
+from locker_server.core.entities.cipher.cipher import Cipher
+
+
+class CipherHistory(object):
+    def __init__(self, cipher_history_id: str, creation_date: float = None, revision_date: float = None,
+                 last_use_date: float = None, reprompt: int = 0,
+                 score: float = 0, data: str = None, cipher: Cipher = None):
+        self._cipher_history_id = cipher_history_id
+        self._creation_date = creation_date
+        self._revision_date = revision_date
+        self._last_use_date = last_use_date
+        self._reprompt = reprompt
+        self._score = score
+        self._data = data
+        self._cipher = cipher
+
+    @property
+    def cipher_history_id(self):
+        return self._cipher_history_id
+
+    @property
+    def creation_date(self):
+        return self._creation_date
+
+    @property
+    def revision_date(self):
+        return self._revision_date
+
+    @property
+    def last_use_date(self):
+        return self._last_use_date
+
+    @property
+    def reprompt(self):
+        return self._reprompt
+
+    @property
+    def score(self):
+        return self._score
+
+    @property
+    def data(self):
+        return self._data
+
+    @property
+    def cipher(self):
+        return self._cipher