🐛(backend) fix wrong permission check on sub page duplicate action [DO NOT MERGE]

src/backend/core/models.py

@@ -28,6 +28,7 @@
 from django.utils.translation import gettext_lazy as _
 
 from botocore.exceptions import ClientError
+from dirtyfields import DirtyFieldsMixin
 from rest_framework.exceptions import ValidationError
 from timezone_field import TimeZoneField
 from treebeard.mp_tree import MP_Node, MP_NodeManager, MP_NodeQuerySet
@@ -353,7 +354,7 @@ def get_queryset(self):
 
 
 # pylint: disable=too-many-public-methods
-class Document(MP_Node, BaseModel):
+class Document(DirtyFieldsMixin, MP_Node, BaseModel):
     """Pad document carrying the content."""
 
     title = models.CharField(_("title"), max_length=255, null=True, blank=True)
@@ -426,11 +427,21 @@ def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self._ancestors_link_definition = None
         self._computed_link_definition = None
+        self._cached_parent_obj = None
 
     def save(self, *args, **kwargs):
         """Write content to object storage only if _content has changed."""
+        invalidate_abilities_cache = False
+        if not self._state.adding:
+            changed_fields = self.get_dirty_fields()
+            if "link_reach" in changed_fields or "link_role" in changed_fields:
+                invalidate_abilities_cache = True
+
         super().save(*args, **kwargs)
 
+        if invalidate_abilities_cache:
+            self.invalidate_abilities_cache()
+
         if self._content:
             file_key = self.file_key
             bytes_content = self._content.encode("utf-8")
@@ -463,6 +474,42 @@ def is_leaf(self):
         """
         return not self.has_deleted_children and self.numchild == 0
 
+    def get_ancestors(self):
+        """
+        :returns: A queryset containing the current node object's ancestors,
+            starting by the root node and descending to the parent.
+        """
+        if self.is_root():
+            return Document.objects.none()
+
+        paths = [self.path[0:pos] for pos in range(0, len(self.path), self.steplen)[1:]]
+        return (
+            Document.objects.select_related("creator")
+            .filter(path__in=paths)
+            .order_by("depth")
+        )
+
+    def get_parent(self, update=False):
+        """
+        :returns: the parent node of the current node object.
+            Caches the result in the object itself to help in loops.
+        """
+        depth = int(len(self.path) / self.steplen)
+        if depth <= 1:
+            return None
+
+        if update:
+            self._cached_parent_obj = None
+
+        if self._cached_parent_obj is not None:
+            return self._cached_parent_obj
+
+        parentpath = self._get_basepath(self.path, depth - 1)
+        self._cached_parent_obj = Document.objects.select_related("creator").get(
+            path=parentpath
+        )
+        return self._cached_parent_obj
+
     @property
     def key_base(self):
         """Key base of the location where the document is stored in object storage."""
@@ -712,10 +759,44 @@ def computed_link_role(self):
         """Actual link role on the document."""
         return self.computed_link_definition["link_role"]
 
+    def _compute_abilities_cache_key(self):
+        """Generate a unique cache key for each document."""
+        return f"document:abilities:{self.path!s}"
+
+    def _get_abilities_cache_for_user(self, user):
+        """Return the abilities cache for the document and user."""
+        key = self._compute_abilities_cache_key()
+        document_abilities = cache.get(key, {})
+
+        user_id = user.id if user.is_authenticated else "anonymous"
+
+        return document_abilities.get(user_id)
+
+    def _set_abilities_cache_for_user(self, user, abilities):
+        """Set the abilities cache for the document and user."""
+        key = self._compute_abilities_cache_key()
+        document_abilities = cache.get(key, {})
+        user_id = user.id if user.is_authenticated else "anonymous"
+        document_abilities[user_id] = abilities
+        cache.set(key, document_abilities, settings.DOCUMENT_ABILITIES_CACHE_TIMEOUT)
+
+    def invalidate_abilities_cache(self):
+        """Invalidate the abilities cache for the document."""
+        key = self._compute_abilities_cache_key()
+        cache.delete(key)
+
+        # Invalidate in cascade the abilities for all children
+        for child in self.get_children():
+            child.invalidate_abilities_cache()
+
     def get_abilities(self, user):
         """
         Compute and return abilities for a given user on the document.
         """
+        abilities = self._get_abilities_cache_for_user(user)
+        if abilities is not None:
+            return abilities
+
         # First get the role based on specific access
         role = self.get_role(user)
 
@@ -759,6 +840,10 @@ def get_abilities(self, user):
             if self.is_root()
             else (is_owner_or_admin or (user.is_authenticated and self.creator == user))
         )
+        can_duplicate = can_get and user.is_authenticated
+        if not self.is_root() and user.is_authenticated:
+            parent_ability = self.get_parent().get_abilities(user)
+            can_duplicate = parent_ability["children_create"]
 
         ai_allow_reach_from = settings.AI_ALLOW_REACH_FROM
         ai_access = any(
@@ -772,7 +857,7 @@ def get_abilities(self, user):
             ]
         )
 
-        return {
+        abilities = {
             "accesses_manage": is_owner_or_admin,
             "accesses_view": has_access_role,
             "ai_transform": ai_access,
@@ -787,7 +872,7 @@ def get_abilities(self, user):
             "cors_proxy": can_get,
             "descendants": can_get,
             "destroy": can_destroy,
-            "duplicate": can_get and user.is_authenticated,
+            "duplicate": can_duplicate,
             "favorite": can_get and user.is_authenticated,
             "link_configuration": is_owner_or_admin,
             "invite_owner": is_owner,
@@ -804,6 +889,8 @@ def get_abilities(self, user):
             "versions_list": has_access_role,
             "versions_retrieve": has_access_role,
         }
+        self._set_abilities_cache_for_user(user, abilities)
+        return abilities
 
     def send_email(self, subject, emails, context=None, language=None):
         """Generate and send email from a template."""
@@ -889,6 +976,7 @@ def soft_delete(self):
         self.ancestors_deleted_at = self.deleted_at = timezone.now()
         self.save()
         self.invalidate_nb_accesses_cache()
+        self.invalidate_abilities_cache()
 
         if self.depth > 1:
             self._meta.model.objects.filter(pk=self.get_parent().pk).update(
@@ -1050,6 +1138,7 @@ def save(self, *args, **kwargs):
         """Override save to clear the document's cache for number of accesses."""
         super().save(*args, **kwargs)
         self.document.invalidate_nb_accesses_cache()
+        self.document.invalidate_abilities_cache()
 
     @property
     def target_key(self):
@@ -1060,6 +1149,7 @@ def delete(self, *args, **kwargs):
         """Override delete to clear the document's cache for number of accesses."""
         super().delete(*args, **kwargs)
         self.document.invalidate_nb_accesses_cache()
+        self.document.invalidate_abilities_cache()
 
     def set_user_roles_tuple(self, ancestors_role, current_role):
         """

src/backend/core/tests/documents/test_api_documents_children_list.py

@@ -98,7 +98,7 @@ def test_api_documents_children_list_anonymous_public_parent(django_assert_num_q
 
     with django_assert_num_queries(9):
         APIClient().get(f"/api/v1.0/documents/{document.id!s}/children/")
-    with django_assert_num_queries(5):
+    with django_assert_num_queries(4):
         response = APIClient().get(f"/api/v1.0/documents/{document.id!s}/children/")
 
     assert response.status_code == 200
@@ -187,7 +187,7 @@ def test_api_documents_children_list_authenticated_unrelated_public_or_authentic
     child1, child2 = factories.DocumentFactory.create_batch(2, parent=document)
     factories.UserDocumentAccessFactory(document=child1)
 
-    with django_assert_num_queries(9):
+    with django_assert_num_queries(11):
         client.get(f"/api/v1.0/documents/{document.id!s}/children/")
     with django_assert_num_queries(5):
         response = client.get(
@@ -267,10 +267,10 @@ def test_api_documents_children_list_authenticated_public_or_authenticated_paren
     child1, child2 = factories.DocumentFactory.create_batch(2, parent=document)
     factories.UserDocumentAccessFactory(document=child1)
 
-    with django_assert_num_queries(10):
+    with django_assert_num_queries(17):
         client.get(f"/api/v1.0/documents/{document.id!s}/children/")
 
-    with django_assert_num_queries(6):
+    with django_assert_num_queries(5):
         response = client.get(f"/api/v1.0/documents/{document.id!s}/children/")
 
     assert response.status_code == 200
@@ -373,7 +373,7 @@ def test_api_documents_children_list_authenticated_related_direct(
     child1, child2 = factories.DocumentFactory.create_batch(2, parent=document)
     factories.UserDocumentAccessFactory(document=child1)
 
-    with django_assert_num_queries(9):
+    with django_assert_num_queries(11):
         response = client.get(
             f"/api/v1.0/documents/{document.id!s}/children/",
         )
@@ -456,7 +456,7 @@ def test_api_documents_children_list_authenticated_related_parent(
         document=grand_parent, user=user
     )
 
-    with django_assert_num_queries(10):
+    with django_assert_num_queries(17):
         response = client.get(
             f"/api/v1.0/documents/{document.id!s}/children/",
         )
@@ -591,7 +591,7 @@ def test_api_documents_children_list_authenticated_related_team_members(
 
     access = factories.TeamDocumentAccessFactory(document=document, team="myteam")
 
-    with django_assert_num_queries(9):
+    with django_assert_num_queries(11):
         response = client.get(f"/api/v1.0/documents/{document.id!s}/children/")
 
     # pylint: disable=R0801

src/backend/core/tests/documents/test_api_documents_list.py

@@ -152,7 +152,7 @@ def test_api_documents_list_authenticated_direct(django_assert_num_queries):
         str(child4_with_access.id),
     }
 
-    with django_assert_num_queries(14):
+    with django_assert_num_queries(17):
         response = client.get("/api/v1.0/documents/")
 
     # nb_accesses should now be cached
@@ -272,7 +272,7 @@ def test_api_documents_list_authenticated_link_reach_public_or_authenticated(
 
     expected_ids = {str(document1.id), str(document2.id), str(visible_child.id)}
 
-    with django_assert_num_queries(11):
+    with django_assert_num_queries(13):
         response = client.get("/api/v1.0/documents/")
 
     # nb_accesses should now be cached

src/backend/core/tests/documents/test_api_documents_retrieve.py

@@ -305,7 +305,7 @@ def test_api_documents_retrieve_authenticated_public_or_authenticated_parent(rea
             "cors_proxy": True,
             "content": True,
             "destroy": False,
-            "duplicate": True,
+            "duplicate": grand_parent.link_role == "editor",
             "favorite": True,
             "invite_owner": False,
             "link_configuration": False,
@@ -500,7 +500,7 @@ def test_api_documents_retrieve_authenticated_related_parent():
             "cors_proxy": True,
             "content": True,
             "destroy": access.role in ["administrator", "owner"],
-            "duplicate": True,
+            "duplicate": access.role != "reader",
             "favorite": True,
             "invite_owner": access.role == "owner",
             "link_configuration": access.role in ["administrator", "owner"],
@@ -855,7 +855,7 @@ def test_api_documents_retrieve_user_role(django_assert_max_num_queries):
     )
     expected_role = choices.RoleChoices.max(*[access.role for access in accesses])
 
-    with django_assert_max_num_queries(14):
+    with django_assert_max_num_queries(16):
         response = client.get(f"/api/v1.0/documents/{document.id!s}/")
 
     assert response.status_code == 200

src/backend/core/tests/documents/test_api_documents_trashbin.py

@@ -165,7 +165,7 @@ def test_api_documents_trashbin_authenticated_direct(django_assert_num_queries):
 
     expected_ids = {str(document1.id), str(document2.id), str(document3.id)}
 
-    with django_assert_num_queries(10):
+    with django_assert_num_queries(14):
         response = client.get("/api/v1.0/documents/trashbin/")
 
     with django_assert_num_queries(4):

src/backend/core/tests/documents/test_api_documents_tree.py

@@ -377,7 +377,7 @@ def test_api_documents_tree_list_authenticated_unrelated_public_or_authenticated
     document, sibling = factories.DocumentFactory.create_batch(2, parent=parent)
     child = factories.DocumentFactory(link_reach="public", parent=document)
 
-    with django_assert_num_queries(13):
+    with django_assert_num_queries(16):
         client.get(f"/api/v1.0/documents/{document.id!s}/tree/")
 
     with django_assert_num_queries(5):