perf: resolve permitted_document_ids once before loop-based permission checks (#13509)

* perf: resolve permitted_document_ids once before email/share loops

Replaces per-document has_perms_owner_aware calls in the email-document
action and bulk share-link-bundle creation with a single
permitted_document_ids(request.user) resolution before the loop,
reducing DB round-trips while preserving identical permission
semantics (including the per-document error message on the bundle
endpoint).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GRp4kf1mdn9ruv81zWAmh2

* perf: resolve permitted_document_ids(perm=change_document) once before bulk-edit loops

Migrates the bulk document edit permission check in views.py and the
custom-field DOCUMENTLINK validator in serialisers.py off of
has_perms_owner_aware-per-document loops, resolving
permitted_document_ids(user, perm="change_document") once instead.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GRp4kf1mdn9ruv81zWAmh2

* perf: resolve permitted_document_ids once for root-document version-listing loop

BulkDownloadView.post() previously called has_perms_owner_aware() per row
inside the loop that resolves each document's root and latest version.
Resolve permitted_document_ids(request.user) once before the loop and check
membership by root_doc.pk instead, consistent with the other consolidated
permission-filtering sites.

* test: use HTTPStatus enum instead of bare integers in security test assertions

* perf: resolve permitted_document_ids(perm=delete_document, include_deleted=True) once for trash loop

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GRp4kf1mdn9ruv81zWAmh2

* perf: drop unused select_related("owner") from email_documents

The permission check loop that used to call has_perms_owner_aware()
(which read .owner) was already replaced with permitted_document_ids()
resolved once into a set. No other code in email_documents touches
.owner, so the select_related is dead weight.

* test: repurpose inert grant into mixed-batch bulk-edit rejection case

The unrelated view_document grant in
test_bulk_edit_rejects_document_without_change_permission created a
document that was never referenced in the request payload. Turn it
into a genuinely useful case instead: a mixed batch containing one
document the requester is fully permitted to change alongside one
they are not, proving bulk_edit rejects the whole batch when any
document lacks change permission (not just checking the first/last
document in the list).

* test: add version-only-grant case discriminating root-vs-version permission check

The former "stranger" sub-case in
test_permission_checked_on_root_not_on_version had zero grants on either
root or version, so it passed under any implementation, correct or
buggy. Replace it with a user granted view_document on the version
itself (not the root): this only passes if bulk_download truly checks
root-only, catching a regression to "root OR version" that the old
case could never detect.

* perf: check permitted document IDs via DB-side exclude/exists instead of materializing the full set

email_documents, _has_document_permissions, TrashView.post, and
validate_documentlink_targets each resolved permitted_document_ids() into
a full Python set just to check membership for a small, bounded batch of
request document IDs. For a user with broad permitted access that pulls
their entire visible/editable document count into memory and across the
wire regardless of how many documents the request actually touches.
Pushing the membership check into the DB via exclude(...).exists() scales
with the request's batch size instead, without reintroducing the
per-row guardian join pathology from #13276 (confirmed via EXPLAIN
ANALYZE: the permission subplans are hashed once, not re-executed per
outer row).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

src/documents/serialisers.py

@@ -864,10 +864,10 @@ def validate_documentlink_targets(user, doc_ids):
     if user is None:
         return
 
-    target_documents = Document.objects.filter(id__in=doc_ids).select_related("owner")
-    if not all(
-        has_perms_owner_aware(user, "change_document", document)
-        for document in target_documents
+    if (
+        Document.objects.filter(id__in=doc_ids)
+        .exclude(id__in=permitted_document_ids(user, perm="change_document"))
+        .exists()
     ):
         raise PermissionDenied(
             _("Insufficient permissions."),

src/documents/tests/test_permission_filtering_security.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from http import HTTPStatus
 from unittest.mock import patch
 
 import pytest
@@ -193,7 +194,7 @@ class TestAiChatAllDocumentsPermissionBoundary:
             format="json",
         )
 
-        assert response.status_code == 200
+        assert response.status_code == HTTPStatus.OK
         mock_stream_chat.assert_called_once()
         _, kwargs = mock_stream_chat.call_args
         visible_ids = {doc.pk for doc in kwargs["documents"]}
@@ -284,3 +285,149 @@ class TestPermittedDocumentIdsArbitraryPermission:
             expected_visible=[],
             expected_hidden=[doc.pk],
         )
+
+
+@pytest.mark.django_db
+class TestEmailDocumentPermissionBoundary:
+    def test_email_action_rejects_document_without_view_permission(
+        self,
+        rest_api_client,
+    ):
+        owner = User.objects.create_user(username="owner")
+        requester = User.objects.create_user(username="requester")
+        rest_api_client.force_authenticate(user=requester)
+        hidden = DocumentFactory(owner=owner)
+
+        response = rest_api_client.post(
+            "/api/documents/email/",
+            {
+                "documents": [hidden.pk],
+                "addresses": "someone@example.com",
+                "subject": "test",
+                "message": "test",
+            },
+            format="json",
+        )
+        assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.django_db
+class TestBulkEditChangePermissionBoundary:
+    def test_bulk_edit_rejects_mixed_batch_when_any_document_lacks_change_permission(
+        self,
+        rest_api_client,
+    ):
+        # A bulk-edit request containing both a document the requester CAN
+        # change and one they CANNOT should be rejected as a whole: the
+        # permitted document must not be partially applied just because it
+        # was bundled with a forbidden one, proving the endpoint checks
+        # every document in the batch rather than only the first/last.
+        owner = User.objects.create_user(username="owner")
+        requester = User.objects.create_user(username="requester")
+        # grant the global change_document permission so the object-level
+        # check (not the global has_perm check) is what's under test
+        requester.user_permissions.add(
+            Permission.objects.get(codename="change_document"),
+        )
+        rest_api_client.force_authenticate(user=requester)
+        changeable = DocumentFactory(owner=owner)
+        assign_perm("view_document", requester, changeable)
+        assign_perm("change_document", requester, changeable)  # fully permitted
+        target = DocumentFactory(owner=owner)
+        assign_perm("view_document", requester, target)  # view only, NOT change
+
+        response = rest_api_client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "documents": [changeable.pk, target.pk],
+                "method": "modify_tags",
+                "parameters": {"add_tags": [], "remove_tags": []},
+            },
+            format="json",
+        )
+        assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.django_db
+class TestBulkDownloadPermissionChecksRootDocument:
+    def test_permission_checked_on_root_not_on_version(
+        self,
+        rest_api_client,
+        paperless_dirs,
+        _media_settings,
+    ):
+        owner = User.objects.create_user(username="owner")
+        requester = User.objects.create_user(username="requester")
+        rest_api_client.force_authenticate(user=requester)
+        root = DocumentFactory(owner=owner)
+        # a version of root that the requester has NOT been individually granted
+        version = DocumentFactory(owner=owner, root_document=root, version_index=1)
+        version.source_path.write_bytes(b"%PDF-1.4 test")
+        assign_perm("view_document", requester, root)  # granted on ROOT only
+
+        response = rest_api_client.post(
+            "/api/documents/bulk_download/",
+            {"documents": [version.pk]},
+            format="json",
+        )
+        assert (
+            response.status_code == HTTPStatus.OK
+        )  # visible because root is permitted
+
+        # Granted on the VERSION itself, but NOT on the root. If the endpoint
+        # ever regressed to checking "root OR version" instead of root-only,
+        # this grant would incorrectly unlock access. This is the case that
+        # actually discriminates correct (root-only) enforcement from a
+        # root-or-version bug; a user with no grant at all (the old
+        # `stranger` case) can't tell the two apart, since they're denied
+        # either way.
+        version_only_grantee = User.objects.create_user(username="version_only_grantee")
+        assign_perm("view_document", version_only_grantee, version)
+        rest_api_client.force_authenticate(user=version_only_grantee)
+        response = rest_api_client.post(
+            "/api/documents/bulk_download/",
+            {"documents": [version.pk]},
+            format="json",
+        )
+        assert (
+            response.status_code == HTTPStatus.FORBIDDEN
+        )  # version-only grant must not substitute for root permission
+
+
+@pytest.mark.django_db
+class TestTrashRestorePermissionBoundary:
+    def test_restore_rejects_document_without_delete_permission(
+        self,
+        rest_api_client,
+    ):
+        owner = User.objects.create_user(username="owner")
+        requester = User.objects.create_user(username="requester")
+        rest_api_client.force_authenticate(user=requester)
+        doc = DocumentFactory(owner=owner)
+        assign_perm("view_document", requester, doc)  # view only, NOT delete
+        doc.delete()
+
+        response = rest_api_client.post(
+            "/api/trash/",
+            {"documents": [doc.pk], "action": "restore"},
+            format="json",
+        )
+        assert response.status_code == HTTPStatus.FORBIDDEN
+
+    def test_restore_allows_document_with_explicit_delete_permission(
+        self,
+        rest_api_client,
+    ):
+        owner = User.objects.create_user(username="owner")
+        requester = User.objects.create_user(username="requester")
+        rest_api_client.force_authenticate(user=requester)
+        doc = DocumentFactory(owner=owner)
+        assign_perm("delete_document", requester, doc)
+        doc.delete()
+
+        response = rest_api_client.post(
+            "/api/trash/",
+            {"documents": [doc.pk], "action": "restore"},
+            format="json",
+        )
+        assert response.status_code == HTTPStatus.OK

src/documents/tests/test_share_link_bundles.py

@@ -60,7 +60,7 @@ class ShareLinkBundleAPITests(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
         self.assertIn("document_ids", response.data)
 
-    @mock.patch("documents.views.has_perms_owner_aware", return_value=False)
+    @mock.patch("documents.views.permitted_document_ids", return_value=set())
     def test_create_bundle_rejects_insufficient_permissions(self, perms_mock) -> None:
         payload = {
             "document_ids": [self.document.pk],

src/documents/views.py

@@ -1929,14 +1929,14 @@ class DocumentViewSet(
         message = validated_data.get("message")
         use_archive_version = validated_data.get("use_archive_version", True)
 
-        documents = Document.objects.select_related("owner").filter(pk__in=document_ids)
-        for document in documents:
-            if request.user is not None and not has_perms_owner_aware(
-                request.user,
-                "view_document",
-                document,
-            ):
-                return HttpResponseForbidden("Insufficient permissions")
+        documents = Document.objects.filter(pk__in=document_ids)
+        if (
+            request.user is not None
+            and documents.exclude(
+                pk__in=permitted_document_ids(request.user),
+            ).exists()
+        ):
+            return HttpResponseForbidden("Insufficient permissions")
 
         try:
             attachments: list[EmailAttachment] = []
@@ -2789,8 +2789,13 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
         )
 
         # check global and object permissions for all documents
-        has_perms = user.has_perm("documents.change_document") and all(
-            has_perms_owner_aware(user, "change_document", doc) for doc in document_objs
+        has_perms = (
+            user.has_perm(
+                "documents.change_document",
+            )
+            and not document_objs.exclude(
+                pk__in=permitted_document_ids(user, perm="change_document"),
+            ).exists()
         )
 
         # check ownership for methods that change original document
@@ -3843,9 +3848,10 @@ class BulkDownloadView(DocumentSelectionMixin, GenericAPIView[Any]):
         content = serializer.validated_data.get("content")
         follow_filename_format = serializer.validated_data.get("follow_formatting")
 
+        permitted_ids = set(permitted_document_ids(request.user))
         for document in documents:
             root_doc = get_root_document(document)
-            if not has_perms_owner_aware(request.user, "view_document", root_doc):
+            if root_doc.pk not in permitted_ids:
                 return HttpResponseForbidden("Insufficient permissions")
             versioned_documents.append(
                 get_latest_version_for_root(
@@ -4509,8 +4515,9 @@ class ShareLinkBundleViewSet(PassUserMixin, ModelViewSet[ShareLinkBundle]):
             )
 
         documents = list(documents_qs)
+        permitted_ids = set(permitted_document_ids(request.user))
         for document in documents:
-            if not has_perms_owner_aware(request.user, "view_document", document):
+            if document.pk not in permitted_ids:
                 raise ValidationError(
                     {
                         "document_ids": _(
@@ -5314,9 +5321,14 @@ class TrashView(ListModelMixin, PassUserMixin):
             if doc_ids is not None
             else self.filter_queryset(self.get_queryset()).all()
         )
-        for doc in docs:
-            if not has_perms_owner_aware(request.user, "delete_document", doc):
-                return HttpResponseForbidden("Insufficient permissions")
+        if docs.exclude(
+            pk__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":
             for doc in Document.deleted_objects.filter(id__in=doc_ids).all():