Trash fix

src/documents/tests/test_api_trash.py

@@ -279,3 +279,71 @@ class TestTrashAPI(DirectoriesMixin, APITestCase):
             Document.objects.filter(root_document=root).values_list("id", flat=True),
             [version.pk for version in versions],
         )
+
+    def test_api_trash_version_follows_root_owner(self) -> None:
+        """
+        GIVEN:
+            - A deleted version of user2's document, owned by nobody
+            - A deleted version of the user's document, owned by user2
+        WHEN:
+            - The user lists the trash and tries to restore or empty the versions
+        THEN:
+            - Only the version of the user's own document is listed
+            - The other version can't be restored or emptied
+            - The version of the user's own document can be restored
+        """
+        user2 = UserFactory(username="user2")
+        other_root = Document.objects.create(
+            title="other root",
+            checksum="other-root",
+            mime_type="application/pdf",
+            owner=user2,
+        )
+        other_version = Document.objects.create(
+            title="other version",
+            checksum="other-version",
+            mime_type="application/pdf",
+            root_document=other_root,
+            version_index=1,
+        )
+        other_version.delete()
+        own_root = Document.objects.create(
+            title="own root",
+            checksum="own-root",
+            mime_type="application/pdf",
+            owner=self.user,
+        )
+        own_version = Document.objects.create(
+            title="own version",
+            checksum="own-version",
+            mime_type="application/pdf",
+            owner=user2,
+            root_document=own_root,
+            version_index=1,
+        )
+        own_version.delete()
+
+        resp = self.client.get("/api/trash/")
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            [doc["id"] for doc in resp.data["results"]],
+            [own_version.pk],
+        )
+
+        for action in ("restore", "empty"):
+            with self.subTest(action=action):
+                resp = self.client.post(
+                    "/api/trash/",
+                    {"action": action, "documents": [other_version.pk]},
+                )
+                self.assertEqual(resp.status_code, status.HTTP_403_FORBIDDEN)
+                self.assertTrue(
+                    Document.deleted_objects.filter(pk=other_version.pk).exists(),
+                )
+
+        resp = self.client.post(
+            "/api/trash/",
+            {"action": "restore", "documents": [own_version.pk]},
+        )
+        self.assertEqual(resp.status_code, status.HTTP_200_OK)
+        self.assertTrue(Document.objects.filter(pk=own_version.pk).exists())

src/documents/views.py

@@ -5607,6 +5607,23 @@ class TrashView(ListModelMixin, PassUserMixin):
     class _TrashPermittedObjectsFilter(PermittedObjectsFilter):
         include_granted = False
 
+        def filter_queryset(self, request, queryset, view):
+            if request.user.is_superuser or not request.user.is_active:
+                return super().filter_queryset(request, queryset, view)
+
+            # A version belongs to whoever owns its root
+            def owned_or_unowned(prefix: str) -> Q:
+                return Q(**{f"{prefix}owner": request.user}) | Q(
+                    **{f"{prefix}owner__isnull": True},
+                )
+
+            return queryset.filter(
+                (Q(root_document__isnull=True) & owned_or_unowned(""))
+                | (
+                    Q(root_document__isnull=False) & owned_or_unowned("root_document__")
+                ),
+            )
+
     filter_backends = (_TrashPermittedObjectsFilter,)
     pagination_class = StandardPagination
 
@@ -5636,13 +5653,18 @@ class TrashView(ListModelMixin, PassUserMixin):
             if doc_ids is not None
             else self.filter_queryset(self.get_queryset()).all()
         )
-        if docs.exclude(
-            pk__in=permitted_document_ids(
-                request.user,
-                perm="delete_document",
-                include_deleted=True,
-            ),
-        ).exists():
+        # Versions are authorized by their root document
+        if (
+            docs.annotate(root_id=Coalesce("root_document_id", "id"))
+            .exclude(
+                root_id__in=permitted_document_ids(
+                    request.user,
+                    perm="delete_document",
+                    include_deleted=True,
+                ),
+            )
+            .exists()
+        ):
             return HttpResponseForbidden("Insufficient permissions")
         action = serializer.validated_data.get("action")
         if action == "restore":