Fix: authorize document versions by their root and speed up permission id sets

permitted_document_ids judged a version by its own owner and grants, so a
version whose owner had drifted from its root's was visible to the wrong
people and hidden from the right ones. Callers patched this individually by
mapping each document to its root first. The query itself was also slow on
MariaDB: the guardian grants were a UNION cast to integers and tested with
IN inside an OR with the owner checks, which MariaDB cannot materialize, so it
re-scans the user's grants for every document. At 20k documents that took
seconds for a user with a couple of hundred grants.

permitted_object_ids now looks grants up as an EXISTS keyed on the row id cast
to a string, which uses guardian's unique index, and matches the user's groups
with an IN subquery. It takes an optional parent_field naming a self-referencing
foreign key whose target authorizes the row, and permitted_document_ids passes
root_document, so a version is visible exactly when its root is. The helper
that mapped documents to their roots at the call sites is no longer needed, so
the email, selection data, share link bundle, trash, bulk download and bulk
edit checks use the id set directly.

src/documents/permissions.py

@@ -6,9 +6,13 @@ from django.contrib.auth.models import Permission
 from django.contrib.auth.models import User
 from django.contrib.contenttypes.models import ContentType
 from django.db.models import Case
+from django.db.models import CharField
 from django.db.models import Count
+from django.db.models import Exists
+from django.db.models import F
 from django.db.models import IntegerField
 from django.db.models import Model
+from django.db.models import OuterRef
 from django.db.models import Q
 from django.db.models import QuerySet
 from django.db.models import Value
@@ -349,6 +353,7 @@ def permitted_object_ids(
     perm: str,
     *,
     include_deleted: bool = False,
+    parent_field: str | None = None,
 ) -> QuerySet[int]:
     """
     Generic version of ``permitted_document_ids`` for any model with an
@@ -357,6 +362,20 @@ def permitted_object_ids(
     soft-delete pattern (currently only ``Document``); for every other model
     it is accepted but has no effect, since those models have no soft-delete
     concept.
+
+    ``parent_field`` names a self-referencing foreign key whose target
+    authorizes the row (``Document.root_document``). A row with a parent is
+    visible exactly when its parent is, judged by the parent's owner and
+    grants, so the row's own owner and grants are ignored.
+
+    Guardian stores ``object_pk`` as a string, so each grant is an ``EXISTS``
+    keyed on the row's id cast to a string, which can use guardian's
+    (user, permission, object_pk) unique index. Casting every ``object_pk`` to
+    an integer for an ``id IN (...)`` is not indexable, and MariaDB cannot
+    materialize it inside the owner ``OR``, so it re-scans the user's grants
+    for every row. The user's groups are matched with an ``IN`` subquery
+    rather than a join through the membership table, which SQLite plans badly
+    once the grant tables grow.
     """
     has_soft_delete = hasattr(model, "global_objects")
     manager = (
@@ -364,8 +383,21 @@ def permitted_object_ids(
     )
     base_qs = manager.all().only("id", "owner")
 
+    owner_field, key_field = "owner", "pk"
+    if parent_field is not None:
+        owner_field, key_field = "authorizing_owner", "authorizing_id"
+        base_qs = base_qs.annotate(
+            authorizing_id=Coalesce(f"{parent_field}_id", "id"),
+            authorizing_owner=Case(
+                When(**{f"{parent_field}_id__isnull": True}, then=F("owner_id")),
+                default=F(f"{parent_field}__owner_id"),
+                output_field=IntegerField(),
+            ),
+        )
+    unowned = Q(**{f"{owner_field}__isnull": True})
+
     if user is None or not getattr(user, "is_authenticated", False):
-        return base_qs.filter(owner__isnull=True).values_list("id", flat=True)
+        return base_qs.filter(unowned).values_list("id", flat=True)
 
     # Deactivated users get nothing, deactivated superusers included, so this
     # has to come before the superuser shortcut. guardian's
@@ -389,20 +421,24 @@ def permitted_object_ids(
         "permission__content_type": content_type,
     }
 
-    user_perm_ids = (
+    key_as_text = Cast(OuterRef(key_field), CharField(max_length=64))
-        UserObjectPermission.objects.filter(user=user, **perm_filter)
+    granted_to_user = Exists(
-        .annotate(object_pk_int=Cast("object_pk", IntegerField()))
+        UserObjectPermission.objects.filter(
-        .values_list("object_pk_int", flat=True)
+            user=user,
+            object_pk=key_as_text,
+            **perm_filter,
+        ),
     )
-    group_perm_ids = (
+    granted_to_group = Exists(
-        GroupObjectPermission.objects.filter(group__user=user, **perm_filter)
+        GroupObjectPermission.objects.filter(
-        .annotate(object_pk_int=Cast("object_pk", IntegerField()))
+            group_id__in=user.groups.values("id"),
-        .values_list("object_pk_int", flat=True)
+            object_pk=key_as_text,
+            **perm_filter,
+        ),
     )
-    permitted_ids = user_perm_ids.union(group_perm_ids)
 
     return base_qs.filter(
-        Q(owner=user) | Q(owner__isnull=True) | Q(id__in=permitted_ids),
+        Q(**{owner_field: user.pk}) | unowned | granted_to_user | granted_to_group,
     ).values_list("id", flat=True)
 
 
@@ -471,30 +507,16 @@ def permitted_document_ids(
     ``include_deleted=True`` for callers that need to check permission on
     soft-deleted documents (e.g. trash restore). This intentionally avoids
     ``get_objects_for_user`` to keep the subquery small and index-friendly.
-    """
-    return permitted_object_ids(user, Document, perm, include_deleted=include_deleted)
 
-
+    A version is authorized by its root document, so a version's own owner and
-def documents_without_permitted_root(
+    grants never matter.
-    documents: QuerySet[Document],
-    user: User | None,
-    *,
-    perm: str = "view_document",
-    include_deleted: bool = False,
-) -> QuerySet[Document]:
     """
-    The documents the user lacks ``perm`` on. Versions are authorized by their
+    return permitted_object_ids(
-    root document, so a version's own owner is ignored. A single query, without
-    loading the documents or joining the root.
-    """
-    return documents.annotate(
-        root_id=Coalesce("root_document_id", "id"),
-    ).exclude(
-        root_id__in=permitted_document_ids(
         user,
-            perm=perm,
+        Document,
+        perm,
         include_deleted=include_deleted,
-        ),
+        parent_field="root_document",
     )
 
 

src/documents/tests/test_permission_filtering_security.py

@@ -32,6 +32,8 @@ from paperless_testing.permissions import grant_global
 from paperless_testing.permissions import grant_object
 
 if TYPE_CHECKING:
+    from django.contrib.auth.models import User
+
     from paperless_testing.dirs import PaperlessDirs
 
 
@@ -178,6 +180,296 @@ class TestPermittedDocumentIdsIncludeDeleted:
         )
 
 
+@pytest.mark.django_db
+class TestPermittedDocumentIdsVersions:
+    """
+    A version is authorized by its root document: the version's own owner and
+    grants never matter.
+    """
+
+    @pytest.mark.parametrize(
+        ("root_owner", "version_owner", "expected_visible"),
+        [
+            pytest.param(
+                "other",
+                "nobody",
+                False,
+                id="unowned-version-of-private-root",
+            ),
+            pytest.param("other", "user", False, id="own-version-of-private-root"),
+            pytest.param("user", "other", True, id="foreign-version-of-own-root"),
+            pytest.param("user", "nobody", True, id="unowned-version-of-own-root"),
+            pytest.param("nobody", "other", True, id="private-version-of-unowned-root"),
+        ],
+    )
+    def test_version_follows_root_owner(
+        self,
+        root_owner: str,
+        version_owner: str,
+        *,
+        expected_visible: bool,
+    ) -> None:
+        """
+        GIVEN:
+            - A root document and a version with differing owners
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - The version is visible exactly when its root is
+        """
+        user = UserFactory()
+        owners = {"user": user, "other": UserFactory(), "nobody": None}
+        root = DocumentFactory(owner=owners[root_owner])
+        version = DocumentFactory(root_document=root, owner=owners[version_owner])
+
+        visible = set(permitted_document_ids(user))
+
+        assert (version.pk in visible) is expected_visible
+        assert (root.pk in visible) is expected_visible
+
+    @staticmethod
+    def grantee(user: User, kind: str) -> User | Group:
+        """The user itself, or a new group the user belongs to."""
+        if kind == "user":
+            return user
+        group = Group.objects.create(name="shared")
+        user.groups.add(group)
+        return group
+
+    @pytest.mark.parametrize(
+        "grantee_kind",
+        [pytest.param("user", id="user"), pytest.param("group", id="group")],
+    )
+    def test_grant_on_root_applies_to_version(self, grantee_kind: str) -> None:
+        """
+        GIVEN:
+            - A private root document shared with a user or one of their groups
+            - A version of it owned by someone else
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - Both the root and the version are visible
+            - A user without the grant sees neither
+        """
+        user = UserFactory()
+        stranger = UserFactory()
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+        grant_object(self.grantee(user, grantee_kind), root, "view_document")
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[root.pk, version.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(stranger),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    @pytest.mark.parametrize(
+        "grantee_kind",
+        [pytest.param("user", id="user"), pytest.param("group", id="group")],
+    )
+    def test_grant_on_version_is_ignored(self, grantee_kind: str) -> None:
+        """
+        GIVEN:
+            - A private root document
+            - A version with an explicit grant for the user or one of their groups
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - Neither the root nor the version is visible
+        """
+        user = UserFactory()
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+        grant_object(self.grantee(user, grantee_kind), version, "view_document")
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    def test_grant_on_one_root_does_not_reach_another_roots_version(self) -> None:
+        """
+        GIVEN:
+            - Two private roots, each with a version
+            - The user may view only the first root
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - Only the first root and its version are visible
+        """
+        user = UserFactory()
+        first = DocumentFactory(owner=UserFactory())
+        first_version = DocumentFactory(root_document=first, owner=UserFactory())
+        second = DocumentFactory(owner=UserFactory())
+        second_version = DocumentFactory(root_document=second, owner=user)
+        grant_object(user, first, "view_document")
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[first.pk, first_version.pk],
+            expected_hidden=[second.pk, second_version.pk],
+        )
+
+    def test_user_in_several_groups(self) -> None:
+        """
+        GIVEN:
+            - A user in two groups
+            - Two private roots shared with one group each, and a third shared with nobody
+            - A version of each root
+        WHEN:
+            - The permitted document ids are resolved for the user
+        THEN:
+            - The two shared roots and their versions are visible
+            - The third root and its version are not
+        """
+        user = UserFactory()
+        groups = [Group.objects.create(name=f"group{i}") for i in range(2)]
+        user.groups.add(*groups)
+        shared = [DocumentFactory(owner=UserFactory()) for _ in groups]
+        for root, group in zip(shared, groups, strict=True):
+            grant_object(group, root, "view_document")
+        unshared = DocumentFactory(owner=UserFactory())
+        shared_versions = [
+            DocumentFactory(root_document=root, owner=UserFactory()) for root in shared
+        ]
+        unshared_version = DocumentFactory(root_document=unshared, owner=None)
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[
+                *(root.pk for root in shared),
+                *(version.pk for version in shared_versions),
+            ],
+            expected_hidden=[unshared.pk, unshared_version.pk],
+        )
+
+    def test_permission_is_resolved_through_the_root(self) -> None:
+        """
+        GIVEN:
+            - A private root document where the user may view and change
+        WHEN:
+            - The permitted ids are resolved for view, change and delete
+        THEN:
+            - The version is visible for view and change only
+        """
+        user = UserFactory()
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+        grant_object(user, root, "view_document", "change_document")
+
+        assert version.pk in set(permitted_document_ids(user))
+        assert version.pk in set(permitted_document_ids(user, perm="change_document"))
+        assert version.pk in set(
+            permitted_document_ids(user, perm="documents.change_document"),
+        )
+        assert version.pk not in set(
+            permitted_document_ids(user, perm="delete_document"),
+        )
+
+    def test_anonymous_sees_versions_of_unowned_roots_only(self) -> None:
+        """
+        GIVEN:
+            - A version owned by nobody under a private root
+            - A version owned by someone under an unowned root
+        WHEN:
+            - The permitted document ids are resolved for an anonymous user
+        THEN:
+            - Only the version of the unowned root is visible
+        """
+        private_root = DocumentFactory(owner=UserFactory())
+        private_version = DocumentFactory(root_document=private_root, owner=None)
+        open_root = DocumentFactory(owner=None)
+        open_version = DocumentFactory(root_document=open_root, owner=UserFactory())
+
+        assert_visible_document_ids(
+            permitted_document_ids(AnonymousUser()),
+            expected_visible=[open_root.pk, open_version.pk],
+            expected_hidden=[private_root.pk, private_version.pk],
+        )
+
+    def test_deleted_versions_follow_their_deleted_root(self) -> None:
+        """
+        GIVEN:
+            - A soft-deleted root document and its version, which deleting the
+              root soft-deletes too; the version is owned by someone else
+        WHEN:
+            - The permitted document ids are resolved with and without deleted
+              documents
+        THEN:
+            - Nothing is visible by default
+            - With deleted documents included, the version is visible to the
+              root's owner and not to the version's own owner
+        """
+        owner = UserFactory()
+        version_owner = UserFactory()
+        root = DocumentFactory(owner=owner)
+        version = DocumentFactory(root_document=root, owner=version_owner)
+        root.delete()
+
+        assert not {root.pk, version.pk} & set(permitted_document_ids(owner))
+        assert_visible_document_ids(
+            permitted_document_ids(owner, include_deleted=True),
+            expected_visible=[root.pk, version.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(version_owner, include_deleted=True),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    @pytest.mark.parametrize(
+        "is_superuser",
+        [
+            pytest.param(False, id="regular-user"),
+            pytest.param(True, id="superuser"),
+        ],
+    )
+    def test_inactive_user_sees_no_versions(self, *, is_superuser: bool) -> None:
+        """
+        GIVEN:
+            - An inactive user, possibly a superuser, who owns a root and its version
+        WHEN:
+            - The permitted document ids are resolved for them
+        THEN:
+            - Nothing is visible
+        """
+        user = UserFactory(is_active=False, is_superuser=is_superuser)
+        root = DocumentFactory(owner=user)
+        version = DocumentFactory(root_document=root, owner=user)
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[],
+            expected_hidden=[root.pk, version.pk],
+        )
+
+    def test_superuser_sees_all_versions(self) -> None:
+        """
+        GIVEN:
+            - A private root owned by someone else, with a version
+        WHEN:
+            - The permitted document ids are resolved for a superuser
+        THEN:
+            - Both the root and the version are visible
+        """
+        superuser = UserFactory(superuser=True)
+        root = DocumentFactory(owner=UserFactory())
+        version = DocumentFactory(root_document=root, owner=UserFactory())
+
+        assert_visible_document_ids(
+            permitted_document_ids(superuser),
+            expected_visible=[root.pk, version.pk],
+            expected_hidden=[],
+        )
+
+
 @pytest.mark.django_db
 class TestAiChatAllDocumentsPermissionBoundary:
     """

src/documents/views.py

@@ -174,7 +174,6 @@ from documents.permissions import TrashPermissions
 from documents.permissions import ViewDocumentsPermissions
 from documents.permissions import annotate_document_count_by_ids
 from documents.permissions import annotate_document_count_for_related_queryset
-from documents.permissions import documents_without_permitted_root
 from documents.permissions import get_document_count_filter_for_user
 from documents.permissions import get_objects_for_user_owner_aware
 from documents.permissions import has_global_statistics_permission
@@ -2117,7 +2116,7 @@ class DocumentViewSet(
         documents = Document.objects.filter(pk__in=document_ids)
         if (
             request.user is not None
-            and documents_without_permitted_root(documents, request.user).exists()
+            and documents.exclude(id__in=permitted_document_ids(request.user)).exists()
         ):
             return HttpResponseForbidden("Insufficient permissions")
 
@@ -3009,12 +3008,8 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
             user.has_perm(
                 "documents.change_document",
             )
-            and not Document.global_objects.filter(
+            and not Document.global_objects.filter(pk__in=documents)
-                pk__in=[doc.pk for doc in root_docs],
+            .exclude(pk__in=permitted_document_ids(user, perm="change_document"))
-            )
-            .exclude(
-                pk__in=permitted_document_ids(user, perm="change_document"),
-            )
             .exists()
         )
 
@@ -3627,7 +3622,7 @@ class SelectionDataView(DocumentSelectionMixin, GenericAPIView[Any]):
         documents = Document.objects.filter(pk__in=ids)
         if (
             documents.count() != len(ids)
-            or documents_without_permitted_root(documents, request.user).exists()
+            or documents.exclude(id__in=permitted_document_ids(request.user)).exists()
         ):
             return HttpResponseForbidden("Insufficient permissions")
 
@@ -4124,21 +4119,16 @@ class BulkDownloadView(DocumentSelectionMixin, GenericAPIView[Any]):
             validated_data=serializer.validated_data,
         )
         documents = Document.objects.filter(pk__in=ids)
-        versioned_documents = []
         compression = serializer.validated_data.get("compression")
         content = serializer.validated_data.get("content")
         follow_filename_format = serializer.validated_data.get("follow_formatting")
 
-        permitted_ids = set(permitted_document_ids(request.user))
+        if documents.exclude(id__in=permitted_document_ids(request.user)).exists():
-        for document in documents:
-            root_doc = get_root_document(document)
-            if root_doc.pk not in permitted_ids:
             return HttpResponseForbidden("Insufficient permissions")
-            versioned_documents.append(
+        versioned_documents = [
-                get_latest_version_for_root(
+            get_latest_version_for_root(get_root_document(document))
-                    root_doc,
+            for document in documents
-                ),
+        ]
-            )
 
         if content == "both":
             strategy_class = OriginalAndArchiveStrategy
@@ -4811,7 +4801,7 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]):
             )
 
         denied_id = (
-            documents_without_permitted_root(documents_qs, request.user)
+            documents_qs.exclude(id__in=permitted_document_ids(request.user))
             .order_by("pk")
             .values_list("pk", flat=True)
             .first()
@@ -5657,11 +5647,12 @@ class TrashView(ListModelMixin, PassUserMixin):
             if doc_ids is not None
             else self.filter_queryset(self.get_queryset()).all()
         )
-        if documents_without_permitted_root(
+        if docs.exclude(
-            docs,
+            id__in=permitted_document_ids(
                 request.user,
                 perm="delete_document",
                 include_deleted=True,
+            ),
         ).exists():
             return HttpResponseForbidden("Insufficient permissions")
         action = serializer.validated_data.get("action")