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

src/documents/permissions.py

@@ -163,14 +163,23 @@ def set_permissions_for_object(
                         )
 
 
-def permitted_document_ids(user):
+def permitted_document_ids(
+    user,
+    *,
+    perm: str = "view_document",
+    include_deleted: bool = False,
+):
     """
-    Return a queryset of document IDs the user may view, limited to non-deleted
-    documents. This intentionally avoids ``get_objects_for_user`` to keep the
-    subquery small and index-friendly.
+    Return a queryset of document IDs the user has ``perm`` on (default
+    ``"view_document"``). By default limited to non-deleted documents; pass
+    ``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.
     """
 
-    base_docs = Document.objects.filter(deleted_at__isnull=True).only("id", "owner")
+    manager = Document.global_objects if include_deleted else Document.objects
+    base_docs = manager.all()
+    base_docs = base_docs.only("id", "owner")
 
     if user is None or not getattr(user, "is_authenticated", False):
         # Just Anonymous user e.g. for drf-spectacular
@@ -181,7 +190,7 @@ def permitted_document_ids(user):
 
     document_ct = ContentType.objects.get_for_model(Document)
     perm_filter = {
-        "permission__codename": "view_document",
+        "permission__codename": perm,
         "permission__content_type": document_ct,
     }
 

src/documents/serialisers.py

@@ -39,7 +39,6 @@ from drf_spectacular.utils import extend_schema_field
 from drf_spectacular.utils import extend_schema_serializer
 from drf_writable_nested.serializers import NestedUpdateMixin
 from guardian.core import ObjectPermissionChecker
-from guardian.shortcuts import get_objects_for_user
 from guardian.shortcuts import get_users_with_perms
 from guardian.utils import get_group_obj_perms_model
 from guardian.utils import get_user_obj_perms_model
@@ -80,8 +79,8 @@ from documents.models import WorkflowTrigger
 from documents.parsers import is_mime_type_supported
 from documents.permissions import get_document_count_filter_for_user
 from documents.permissions import get_groups_with_only_permission
-from documents.permissions import get_objects_for_user_owner_aware
 from documents.permissions import has_perms_owner_aware
+from documents.permissions import permitted_document_ids
 from documents.permissions import set_permissions_for_object
 from documents.regex import validate_regex_pattern
 from documents.templating.filepath import validate_filepath_template_and_render
@@ -865,11 +864,8 @@ 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
-    ):
+    permitted_change_ids = set(permitted_document_ids(user, perm="change_document"))
+    if not set(doc_ids) <= permitted_change_ids:
         raise PermissionDenied(
             _("Insufficient permissions."),
         )
@@ -1011,13 +1007,8 @@ def _get_viewable_duplicates(
     ).exclude(pk=document.pk)
     duplicates = duplicates.filter(root_document__isnull=True)
     duplicates = duplicates.order_by("-created")
-    allowed = get_objects_for_user_owner_aware(
-        user,
-        "documents.view_document",
-        Document,
-        include_deleted=True,
-    )
-    return duplicates.filter(id__in=allowed)
+    allowed_ids = permitted_document_ids(user, include_deleted=True)
+    return duplicates.filter(id__in=allowed_ids)
 
 
 class DuplicateDocumentSummarySerializer(serializers.Serializer[dict[str, Any]]):
@@ -2659,13 +2650,8 @@ class TaskSerializerV9(serializers.ModelSerializer[PaperlessTask]):
         user = request.user
         qs = Document.global_objects.filter(pk=dup_of)
         if not user.is_staff:
-            with_perms = get_objects_for_user(
-                user,
-                "documents.view_document",
-                qs,
-                accept_global_perms=False,
-            )
-            qs = with_perms | qs.filter(owner=user) | qs.filter(owner__isnull=True)
+            allowed_ids = permitted_document_ids(user, include_deleted=True)
+            qs = qs.filter(pk__in=allowed_ids)
         return list(qs.values("id", "title", "deleted_at"))
 
 
@@ -3515,8 +3501,6 @@ class StoragePathTestSerializer(SerializerWithPerms):
             document_field = self.fields.get("document")
             if not isinstance(document_field, serializers.PrimaryKeyRelatedField):
                 return
-            document_field.queryset = get_objects_for_user_owner_aware(
-                user,
-                "documents.view_document",
-                Document,
+            document_field.queryset = Document.objects.filter(
+                id__in=permitted_document_ids(user),
             )

src/documents/tests/test_permission_filtering_security.py

@@ -0,0 +1,392 @@
+from __future__ import annotations
+
+from http import HTTPStatus
+from unittest.mock import patch
+
+import pytest
+from django.contrib.auth.models import AnonymousUser
+from django.contrib.auth.models import Group
+from django.contrib.auth.models import Permission
+from django.contrib.auth.models import User
+from django.test import override_settings
+from guardian.shortcuts import assign_perm
+from rest_framework.test import APIClient
+
+from documents.permissions import permitted_document_ids
+from documents.serialisers import _get_viewable_duplicates
+from documents.tests.factories import DocumentFactory
+
+
+def assert_visible_document_ids(actual_ids, *, expected_visible, expected_hidden):
+    actual_ids = set(actual_ids)
+    for doc_id in expected_visible:
+        assert doc_id in actual_ids, (
+            f"document {doc_id} should be visible but was hidden"
+        )
+    for doc_id in expected_hidden:
+        assert doc_id not in actual_ids, (
+            f"document {doc_id} leaked but should be hidden"
+        )
+
+
+@pytest.mark.django_db
+class TestPermittedDocumentIdsSecurity:
+    def test_owner_sees_own_document(self):
+        user = User.objects.create_user(username="alice")
+        stranger = User.objects.create_user(username="mallory")
+        owned = DocumentFactory(owner=user)
+        strangers_doc = DocumentFactory(owner=stranger)
+
+        visible = permitted_document_ids(user)
+
+        assert_visible_document_ids(
+            visible,
+            expected_visible=[owned.pk],
+            expected_hidden=[strangers_doc.pk],
+        )
+
+    def test_unowned_document_visible_to_everyone(self):
+        user = User.objects.create_user(username="alice")
+        unowned = DocumentFactory(owner=None)
+
+        assert_visible_document_ids(
+            permitted_document_ids(user),
+            expected_visible=[unowned.pk],
+            expected_hidden=[],
+        )
+
+    def test_explicit_user_permission_grants_visibility(self):
+        grantee = User.objects.create_user(username="alice")
+        stranger = User.objects.create_user(username="mallory")
+        owner = User.objects.create_user(username="owner")
+        shared = DocumentFactory(owner=owner)
+        not_shared = DocumentFactory(owner=owner)
+        assign_perm("view_document", grantee, shared)
+
+        assert_visible_document_ids(
+            permitted_document_ids(grantee),
+            expected_visible=[shared.pk],
+            expected_hidden=[not_shared.pk],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(stranger),
+            expected_visible=[],
+            expected_hidden=[shared.pk, not_shared.pk],
+        )
+
+    def test_explicit_group_permission_grants_visibility_to_members_only(self):
+        owner = User.objects.create_user(username="owner")
+        member = User.objects.create_user(username="member")
+        non_member = User.objects.create_user(username="non_member")
+        group = Group.objects.create(name="finance")
+        member.groups.add(group)
+        shared = DocumentFactory(owner=owner)
+        assign_perm("view_document", group, shared)
+
+        assert_visible_document_ids(
+            permitted_document_ids(member),
+            expected_visible=[shared.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(non_member),
+            expected_visible=[],
+            expected_hidden=[shared.pk],
+        )
+
+    def test_soft_deleted_document_excluded_by_default(self):
+        owner = User.objects.create_user(username="owner")
+        doc = DocumentFactory(owner=owner)
+        doc.delete()  # soft delete
+
+        assert_visible_document_ids(
+            permitted_document_ids(owner),
+            expected_visible=[],
+            expected_hidden=[doc.pk],
+        )
+
+    def test_superuser_sees_everything_including_no_perm_documents(self):
+        superuser = User.objects.create_superuser(username="root")
+        owner = User.objects.create_user(username="owner")
+        doc = DocumentFactory(owner=owner)
+
+        assert_visible_document_ids(
+            permitted_document_ids(superuser),
+            expected_visible=[doc.pk],
+            expected_hidden=[],
+        )
+
+    def test_anonymous_user_sees_only_unowned_documents(self):
+        owner = User.objects.create_user(username="owner")
+        owned = DocumentFactory(owner=owner)
+        unowned = DocumentFactory(owner=None)
+
+        assert_visible_document_ids(
+            permitted_document_ids(AnonymousUser()),
+            expected_visible=[unowned.pk],
+            expected_hidden=[owned.pk],
+        )
+
+
+@pytest.mark.django_db
+class TestPermittedDocumentIdsIncludeDeleted:
+    def test_include_deleted_true_reveals_soft_deleted_owned_document(self):
+        owner = User.objects.create_user(username="owner")
+        doc = DocumentFactory(owner=owner)
+        doc.delete()
+
+        assert_visible_document_ids(
+            permitted_document_ids(owner, include_deleted=True),
+            expected_visible=[doc.pk],
+            expected_hidden=[],
+        )
+
+    def test_include_deleted_true_still_respects_permission_boundary(self):
+        owner = User.objects.create_user(username="owner")
+        stranger = User.objects.create_user(username="mallory")
+        doc = DocumentFactory(owner=owner)
+        doc.delete()
+
+        assert_visible_document_ids(
+            permitted_document_ids(stranger, include_deleted=True),
+            expected_visible=[],
+            expected_hidden=[doc.pk],
+        )
+
+
+@pytest.mark.django_db
+class TestAiChatAllDocumentsPermissionBoundary:
+    """
+    Regression test pinning the "ask across all documents" AI chat behavior
+    (ChatStreamingView.post, no document_id) to the same owner/permission
+    boundary enforced by permitted_document_ids(). This call site was
+    migrated from get_objects_for_user_owner_aware() to
+    permitted_document_ids() in Task 5; this test must stay green across
+    that swap.
+    """
+
+    ENDPOINT = "/api/documents/chat/"
+
+    @override_settings(AI_ENABLED=True)
+    @patch("documents.views.stream_chat_with_documents")
+    def test_chat_all_documents_excludes_unshared_document(self, mock_stream_chat):
+        mock_stream_chat.return_value = iter([b"data"])
+
+        owner = User.objects.create_user(username="owner")
+        asker = User.objects.create_user(username="asker")
+        asker.user_permissions.add(
+            *Permission.objects.filter(codename="view_document"),
+        )
+        shared = DocumentFactory(owner=owner)
+        not_shared = DocumentFactory(owner=owner)
+        assign_perm("view_document", asker, shared)
+
+        client = APIClient()
+        client.force_authenticate(user=asker)
+        response = client.post(
+            self.ENDPOINT,
+            data={"q": "question"},
+            format="json",
+        )
+
+        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"]}
+        assert shared.pk in visible_ids
+        assert not_shared.pk not in visible_ids
+
+
+@pytest.mark.django_db
+class TestDuplicateDocumentsPermissionBoundary:
+    def test_get_viewable_duplicates_includes_soft_deleted_but_respects_perms(self):
+        owner = User.objects.create_user(username="owner")
+        stranger = User.objects.create_user(username="mallory")
+        original = DocumentFactory(owner=owner, checksum="dupe-checksum")
+        dup_visible = DocumentFactory(owner=owner, checksum="dupe-checksum")
+        dup_hidden = DocumentFactory(owner=owner, checksum="dupe-checksum")
+        dup_hidden.delete()  # soft delete, should still be found (include_deleted=True)
+        assign_perm("view_document", stranger, dup_visible)
+
+        result_owner = _get_viewable_duplicates(original, owner)
+        assert {d.pk for d in result_owner} == {dup_visible.pk, dup_hidden.pk}
+
+        result_stranger = _get_viewable_duplicates(original, stranger)
+        assert {d.pk for d in result_stranger} == {dup_visible.pk}
+
+
+@pytest.mark.django_db
+class TestPermittedDocumentIdsArbitraryPermission:
+    def test_change_document_permission_is_distinct_from_view(self):
+        owner = User.objects.create_user(username="owner")
+        viewer_only = User.objects.create_user(username="viewer")
+        editor = User.objects.create_user(username="editor")
+        doc = DocumentFactory(owner=owner)
+        assign_perm("view_document", viewer_only, doc)
+        assign_perm("change_document", editor, doc)
+        assign_perm("view_document", editor, doc)
+
+        assert_visible_document_ids(
+            permitted_document_ids(editor, perm="change_document"),
+            expected_visible=[doc.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(viewer_only, perm="change_document"),
+            expected_visible=[],
+            expected_hidden=[doc.pk],
+        )
+
+    def test_delete_permission_with_include_deleted_for_trash_restore(self):
+        owner = User.objects.create_user(username="owner")
+        stranger = User.objects.create_user(username="mallory")
+        doc = DocumentFactory(owner=owner)
+        doc.delete()
+
+        assert_visible_document_ids(
+            permitted_document_ids(owner, perm="delete_document", include_deleted=True),
+            expected_visible=[doc.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_document_ids(
+                stranger,
+                perm="delete_document",
+                include_deleted=True,
+            ),
+            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_document_without_change_permission(
+        self,
+        rest_api_client,
+    ):
+        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)
+        assign_perm(
+            "view_document",
+            requester,
+            DocumentFactory(owner=owner),
+        )  # unrelated grant
+        target = DocumentFactory(owner=owner)
+        assign_perm("view_document", requester, target)  # view only, NOT change
+
+        response = rest_api_client.post(
+            "/api/documents/bulk_edit/",
+            {
+                "documents": [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
+
+        stranger = User.objects.create_user(username="mallory")
+        rest_api_client.force_authenticate(user=stranger)
+        response = rest_api_client.post(
+            "/api/documents/bulk_download/",
+            {"documents": [version.pk]},
+            format="json",
+        )
+        assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@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/tests/test_views.py

@@ -612,11 +612,11 @@ class TestAIChatStreamingView(DirectoriesMixin, TestCase):
         self.assertIn(b"AI is required for this feature", response.content)
 
     @patch("documents.views.stream_chat_with_documents")
-    @patch("documents.views.get_objects_for_user_owner_aware")
+    @patch("documents.views.permitted_document_ids")
     @override_settings(AI_ENABLED=True)
-    def test_post_no_document_id(self, mock_get_objects, mock_stream_chat) -> None:
+    def test_post_no_document_id(self, mock_permitted_ids, mock_stream_chat) -> None:
         self.grant_view_document_permission()
-        mock_get_objects.return_value = [self.document]
+        mock_permitted_ids.return_value = [self.document.pk]
         mock_stream_chat.return_value = iter([b"data"])
         response = self.client.post(
             self.ENDPOINT,
@@ -625,23 +625,23 @@ class TestAIChatStreamingView(DirectoriesMixin, TestCase):
         )
         self.assertEqual(response.status_code, 200)
         self.assertEqual(response["Content-Type"], "text/event-stream")
-        mock_stream_chat.assert_called_once_with(
-            query_str="question",
-            documents=[self.document],
-            output_language=None,
-        )
+        mock_stream_chat.assert_called_once()
+        call_kwargs = mock_stream_chat.call_args.kwargs
+        self.assertEqual(call_kwargs["query_str"], "question")
+        self.assertEqual(list(call_kwargs["documents"]), [self.document])
+        self.assertIsNone(call_kwargs["output_language"])
 
     @patch("documents.views.stream_chat_with_documents")
-    @patch("documents.views.get_objects_for_user_owner_aware")
+    @patch("documents.views.permitted_document_ids")
     @override_settings(AI_ENABLED=True)
     def test_post_uses_user_display_language(
         self,
-        mock_get_objects,
+        mock_permitted_ids,
         mock_stream_chat,
     ) -> None:
         UiSettings.objects.create(user=self.user, settings={"language": "de-de"})
         self.grant_view_document_permission()
-        mock_get_objects.return_value = [self.document]
+        mock_permitted_ids.return_value = [self.document.pk]
         mock_stream_chat.return_value = iter([b"data"])
 
         response = self.client.post(
@@ -651,11 +651,11 @@ class TestAIChatStreamingView(DirectoriesMixin, TestCase):
         )
 
         self.assertEqual(response.status_code, 200)
-        mock_stream_chat.assert_called_once_with(
-            query_str="question",
-            documents=[self.document],
-            output_language="de-de",
-        )
+        mock_stream_chat.assert_called_once()
+        call_kwargs = mock_stream_chat.call_args.kwargs
+        self.assertEqual(call_kwargs["query_str"], "question")
+        self.assertEqual(list(call_kwargs["documents"]), [self.document])
+        self.assertEqual(call_kwargs["output_language"], "de-de")
 
     @patch("documents.views.stream_chat_with_documents")
     @override_settings(AI_ENABLED=True)

src/documents/views.py

@@ -177,6 +177,7 @@ from documents.permissions import get_objects_for_user_owner_aware
 from documents.permissions import has_global_statistics_permission
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
+from documents.permissions import permitted_document_ids
 from documents.permissions import set_permissions_for_object
 from documents.plugins.date_parsing import get_date_parser
 from documents.schema import generate_object_with_permissions_schema
@@ -1924,12 +1925,9 @@ class DocumentViewSet(
         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,
-            ):
+        if request.user is not None:
+            permitted_ids = set(permitted_document_ids(request.user))
+            if not all(document.pk in permitted_ids for document in documents):
                 return HttpResponseForbidden("Insufficient permissions")
 
         try:
@@ -2265,10 +2263,8 @@ class ChatStreamingView(GenericAPIView[Any]):
 
             documents = [document]
         else:
-            documents = get_objects_for_user_owner_aware(
-                request.user,
-                "view_document",
-                Document,
+            documents = Document.objects.filter(
+                id__in=permitted_document_ids(request.user),
             )
 
         output_language = _get_llm_output_language(ai_config=ai_config, request=request)
@@ -2736,10 +2732,8 @@ class DocumentSelectionMixin:
             for key, value in filters.items()
             if key not in _TANTIVY_SEARCH_PARAM_NAMES
         }
-        permitted_documents = get_objects_for_user_owner_aware(
-            user,
-            permission_codename,
-            Document,
+        permitted_documents = Document.objects.filter(
+            id__in=permitted_document_ids(user),
         )
         # orm-filtered docs
         filtered_documents = DocumentFilterSet(
@@ -2788,8 +2782,9 @@ class DocumentOperationPermissionMixin(PassUserMixin, DocumentSelectionMixin):
         )
 
         # check global and object permissions for all documents
+        permitted_change_ids = set(permitted_document_ids(user, perm="change_document"))
         has_perms = user.has_perm("documents.change_document") and all(
-            has_perms_owner_aware(user, "change_document", doc) for doc in document_objs
+            doc.pk in permitted_change_ids for doc in document_objs
         )
 
         # check ownership for methods that change original document
@@ -3347,10 +3342,8 @@ class SelectionDataView(GenericAPIView[Any]):
         serializer.is_valid(raise_exception=True)
 
         ids = serializer.validated_data.get("documents")
-        permitted_documents = get_objects_for_user_owner_aware(
-            request.user,
-            "documents.view_document",
-            Document,
+        permitted_documents = Document.objects.filter(
+            id__in=permitted_document_ids(request.user),
         )
         if permitted_documents.filter(pk__in=ids).count() != len(ids):
             return HttpResponseForbidden("Insufficient permissions")
@@ -3522,10 +3515,8 @@ class GlobalSearchView(PassUserMixin):
         OBJECT_LIMIT = 3
         docs = []
         if request.user.has_perm("documents.view_document"):
-            all_docs = get_objects_for_user_owner_aware(
-                request.user,
-                "view_document",
-                Document,
+            all_docs = Document.objects.filter(
+                id__in=permitted_document_ids(request.user),
             )
             if db_only:
                 docs = all_docs.filter(title__icontains=query)[:OBJECT_LIMIT]
@@ -3729,11 +3720,7 @@ class StatisticsView(GenericAPIView[Any]):
         documents = (
             Document.objects.all()
             if can_view_global_stats
-            else get_objects_for_user_owner_aware(
-                user,
-                "documents.view_document",
-                Document,
-            )
+            else Document.objects.filter(id__in=permitted_document_ids(user))
         ).filter(root_document__isnull=True)
         tags = (
             Tag.objects.all()
@@ -3850,9 +3837,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(
@@ -4516,8 +4504,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": _(
@@ -5319,9 +5308,15 @@ 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")
+        permitted_ids = set(
+            permitted_document_ids(
+                request.user,
+                perm="delete_document",
+                include_deleted=True,
+            ),
+        )
+        if not all(doc.pk in permitted_ids for doc in docs):
+            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():