🐛(backend) fix wrong permission checj on sub page duplicate action

A user with only read access to a document and it sub documents should
not be able to duplicate a sub document in the document tree. To avoid
this we have to compute the direct parent abilities to determine if it
can create a new children. Doing so computes the abilities in cascase
from the document to the root tree. To asave some quesries and not
compute again and again the same ability, we create a abilities cache.
The hard part is to invalidate this cache. The cache is invalidated if a
related DocumentAccess instance is updated or deleted and also if the
document link_reach or link_role is updated. To introspect the
modification made on the model it self when the user save it, we decided
to use the library django-dirtyfields

src/backend/core/models.py

@@ -28,6 +28,7 @@ from django.utils.translation import get_language, override
 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 @@ class DocumentManager(MP_NodeManager.from_queryset(DocumentQuerySet)):
 
 
 # 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 @@ class Document(MP_Node, BaseModel):
         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 @@ class Document(MP_Node, BaseModel):
         """
         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 @@ class Document(MP_Node, BaseModel):
         """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 @@ class Document(MP_Node, BaseModel):
             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 @@ class Document(MP_Node, BaseModel):
             ]
         )
 
-        return {
+        abilities = {
             "accesses_manage": is_owner_or_admin,
             "accesses_view": has_access_role,
             "ai_transform": ai_access,
@@ -787,7 +872,7 @@ class Document(MP_Node, BaseModel):
             "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 @@ class Document(MP_Node, BaseModel):
             "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 @@ class Document(MP_Node, BaseModel):
         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 @@ class DocumentAccess(BaseAccess):
         """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 @@ class DocumentAccess(BaseAccess):
         """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):

src/backend/core/tests/test_models_documents.py

@@ -318,6 +318,10 @@ def test_models_documents_get_abilities_editor(
     nb_queries = 1 if is_authenticated else 0
     with django_assert_num_queries(nb_queries):
         assert document.get_abilities(user) == expected_abilities
+
+    with django_assert_num_queries(0):
+        assert document.get_abilities(user) == expected_abilities
+
     document.soft_delete()
     document.refresh_from_db()
     assert all(
@@ -681,6 +685,98 @@ def test_models_documents_get_abilities_children_destroy(  # noqa: PLR0913
     assert abilities["destroy"] is can_destroy
 
 
+def test_models_documents_get_abilities_cache(django_assert_num_queries):
+    """Abilities should be computed once and cached."""
+
+    user = factories.UserFactory()
+    parent = factories.DocumentFactory(link_reach="restricted", link_role="editor")
+    document = factories.DocumentFactory(
+        link_reach="restricted",
+        link_role="editor",
+        parent=parent,
+    )
+
+    access = factories.UserDocumentAccessFactory(
+        document=parent, user=user, role="editor"
+    )
+
+    # 1 for fetching the parent, 2 for computed the roles (parent and document)
+    with django_assert_num_queries(
+        3
+    ):  # 1 for fetching the parent, 2 for computed the roles (parent and document)
+        document.get_abilities(user)
+
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+    # modifying the access should invalidate the cache
+    access.role = "reader"
+    access.save()
+    # 2 for computed the roles (parent and document), the parent is still in document's cache
+    with django_assert_num_queries(2):
+        document.get_abilities(user)
+
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+    # because the parent abilities have alreadby been comptued when
+    # document abilities were computed, no query is done
+    with django_assert_num_queries(0):
+        parent.get_abilities(user)
+
+    # deleting the access should invalidate the cache
+    access.delete()
+    with django_assert_num_queries(2):
+        document.get_abilities(user)
+
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+    # Modifying the document link_reach should invalidate the cache
+    document.link_reach = "public"
+    document.save()
+    # 1 for computed the roles (document), parent abilities are still in cache
+    with django_assert_num_queries(1):
+        document.get_abilities(user)
+
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+    parent.link_reach = "public"
+    parent.save()
+    # 2 for computed the roles (parent and document)
+    with django_assert_num_queries(2):
+        document.get_abilities(user)
+
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+    # Modifying the parent link_role should invalidate the cache
+    document.link_role = "reader"
+    document.save()
+    # 1 for computed the roles (document), parent abilities are still in cache
+    with django_assert_num_queries(1):
+        document.get_abilities(user)
+
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+    parent.link_role = "reader"
+    parent.save()
+    # 2 for computed the roles (parent and document)
+    with django_assert_num_queries(2):
+        document.get_abilities(user)
+
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+    # Modifying any other property should not invalidate the cache
+    document.title = "new title"
+    document.save()
+    with django_assert_num_queries(0):
+        document.get_abilities(user)
+
+
 @override_settings(AI_ALLOW_REACH_FROM="public")
 @pytest.mark.parametrize(
     "is_authenticated,reach",

src/backend/impress/settings.py

@@ -808,6 +808,12 @@ class Base(Configuration):
         ),
     }
 
+    DOCUMENT_ABILITIES_CACHE_TIMEOUT = values.IntegerValue(
+        default=60 * 60,  # 1 hour
+        environ_name="ABILITIES_CACHE_TIMEOUT",
+        environ_prefix=None,
+    )
+
     # pylint: disable=invalid-name
     @property
     def ENVIRONMENT(self):

src/backend/pyproject.toml

@@ -33,6 +33,7 @@ dependencies = [
     "django-cors-headers==4.7.0",
     "django-countries==7.6.1",
     "django-csp==4.0",
+    "django-dirtyfields==1.9.7",
     "django-filter==25.1",
     "django-lasuite[all]==0.0.14",
     "django-parler==2.3",