Chore: drop get_objects_for_user_owner_aware

Drops get_objects_for_user_owner_aware. Callers now use restrict_queryset_to_visible.

The same question was answered two ways, and the guardian path did not deny
deactivated users the way permitted_object_ids does.

src/documents/permissions.py

@@ -1,11 +1,5 @@
-from collections.abc import Mapping
-from collections.abc import Sequence
 from typing import Any
-from typing import ClassVar
-from typing import Literal
-from typing import TypeGuard
 from typing import TypeVar
-from typing import cast
 
 from django.contrib.auth.models import Group
 from django.contrib.auth.models import Permission
@@ -27,7 +21,6 @@ from guardian.core import ObjectPermissionChecker
 from guardian.models import GroupObjectPermission
 from guardian.models import UserObjectPermission
 from guardian.shortcuts import assign_perm
-from guardian.shortcuts import get_objects_for_user
 from guardian.shortcuts import get_users_with_perms
 from guardian.shortcuts import remove_perm
 from rest_framework.permissions import BasePermission
@@ -68,23 +61,28 @@ class PaperlessAdminPermissions(BasePermission):
         return request.user.is_active and request.user.is_staff
 
 
-def _is_active_and_authenticated(user: User | None) -> TypeGuard[User]:
-    return (
-        user is not None
-        and getattr(user, "is_active", False)
-        and getattr(user, "is_authenticated", False)
-    )
-
-
 def has_global_statistics_permission(user: User | None) -> bool:
-    return _is_active_and_authenticated(user) and (
-        getattr(user, "is_superuser", False)
-        or user.has_perm("paperless.view_global_statistics")
+    if (
+        user is None
+        or not getattr(user, "is_active", False)
+        or not getattr(user, "is_authenticated", False)
+    ):
+        return False
+
+    return getattr(user, "is_superuser", False) or user.has_perm(
+        "paperless.view_global_statistics",
     )
 
 
 def has_system_status_permission(user: User | None) -> bool:
-    return _is_active_and_authenticated(user) and (
+    if (
+        user is None
+        or not getattr(user, "is_active", False)
+        or not getattr(user, "is_authenticated", False)
+    ):
+        return False
+
+    return (
         getattr(user, "is_superuser", False)
         or getattr(user, "is_staff", False)
         or user.has_perm("paperless.view_system_monitoring")
@@ -650,41 +648,6 @@ def annotate_document_count_for_related_queryset(
     )
 
 
-def get_objects_for_user_owner_aware(
-    user: User | None,
-    perms: str | list[str],
-    Model: Any,
-    *,
-    include_deleted: bool = False,
-) -> QuerySet[Any]:
-    """
-    Returns objects the user owns, are unowned, or has explicit perms.
-    When include_deleted is True, soft-deleted items are also included.
-
-    Legacy slow path (guardian-backed, O(n) style permission resolution).
-    Most queryset-filtering call sites have migrated onto
-    ``PermittedObjectsFilter``/``permitted_object_ids()``, but this function
-    is kept because production callers still remain. Several callers remain
-    across ``documents/``, ``paperless_mail/``, and ``paperless_ai/`` --
-    grep for this function name before removing it.
-    """
-    manager = (
-        Model.global_objects
-        if include_deleted and hasattr(Model, "global_objects")
-        else Model.objects
-    )
-
-    objects_owned = manager.filter(owner=user)
-    objects_unowned = manager.filter(owner__isnull=True)
-    objects_with_perms = get_objects_for_user(
-        user=user,
-        perms=perms,
-        klass=manager.all(),
-        accept_global_perms=False,
-    )
-    return objects_owned | objects_unowned | objects_with_perms
-
-
 def has_perms_owner_aware(user, perms, obj):
     """
     Legacy slow path (guardian-backed) single-object permission check.
@@ -706,32 +669,7 @@ def has_perms_owner_aware(user, perms, obj):
     return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
 
 
-# HEAD is deliberately absent: GlobalPermissionsByMethod checks it against GET.
-MappedMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
-
-
-class GlobalPermissionsByMethod(BasePermission):
-    """
-    Requires the global (model-level) permissions listed in ``perms_map`` for
-    the request's HTTP method. HEAD is checked against the ``"GET"`` entry,
-    since DRF runs the GET handler for it. A method missing from
-    ``perms_map`` is denied.
-    """
-
-    perms_map: ClassVar[Mapping[MappedMethod, Sequence[str]]]
-
-    def has_permission(self, request: Any, view: Any) -> bool:
-        if not request.user or not request.user.is_authenticated:
-            return False
-
-        method = "GET" if request.method == "HEAD" else request.method
-        # An unknown method is not a MappedMethod, but .get() returns None for
-        # it, so it is still denied.
-        perms = self.perms_map.get(cast("MappedMethod", method))
-        return perms is not None and request.user.has_perms(perms)
-
-
-class ViewDocumentsPermissions(GlobalPermissionsByMethod):
+class ViewDocumentsPermissions(BasePermission):
     """
     Permissions class that checks for model permissions for only viewing Documents.
     """
@@ -742,18 +680,31 @@ class ViewDocumentsPermissions(GlobalPermissionsByMethod):
         "POST": ["documents.view_document"],
     }
 
+    def has_permission(self, request, view):
+        if not request.user or (not request.user.is_authenticated):  # pragma: no cover
+            return False
 
-class TrashPermissions(GlobalPermissionsByMethod):
+        return request.user.has_perms(self.perms_map.get(request.method, []))
+
+
+class TrashPermissions(BasePermission):
     """Check the global document permission for each trash operation."""
 
     perms_map = {
         "OPTIONS": ["documents.view_document"],
+        "HEAD": ["documents.view_document"],
         "GET": ["documents.view_document"],
         "POST": ["documents.delete_document"],
     }
 
+    def has_permission(self, request, view):
+        if not request.user or not request.user.is_authenticated:  # pragma: no cover
+            return False
 
-class PaperlessNotePermissions(GlobalPermissionsByMethod):
+        return request.user.has_perms(self.perms_map.get(request.method, []))
+
+
+class PaperlessNotePermissions(BasePermission):
     """
     Permissions class that checks for model permissions for Notes.
     """
@@ -773,13 +724,28 @@ class PaperlessNotePermissions(GlobalPermissionsByMethod):
         ],
     }
 
+    def has_permission(self, request, view):
+        if not request.user or (not request.user.is_authenticated):  # pragma: no cover
+            return False
 
-class AcknowledgeTasksPermissions(GlobalPermissionsByMethod):
+        perms = self.perms_map[request.method]
+
+        return request.user.has_perms(perms)
+
+
+class AcknowledgeTasksPermissions(BasePermission):
     """
     Permissions class that checks for model permissions for acknowledging tasks.
     """
 
     perms_map = {
-        "OPTIONS": ["documents.change_paperlesstask"],
         "POST": ["documents.change_paperlesstask"],
     }
+
+    def has_permission(self, request: Any, view: Any) -> bool:
+        if not request.user or not request.user.is_authenticated:  # pragma: no cover
+            return False
+
+        perms = self.perms_map.get(request.method, [])
+
+        return request.user.has_perms(perms)

src/documents/signals/handlers.py

@@ -52,7 +52,7 @@ from documents.models import Workflow
 from documents.models import WorkflowAction
 from documents.models import WorkflowRun
 from documents.models import WorkflowTrigger
-from documents.permissions import get_objects_for_user_owner_aware
+from documents.permissions import restrict_queryset_to_visible
 from documents.plugins.helpers import DocumentsStatusManager
 from documents.templating.utils import convert_format_str_to_template_format
 from documents.utils import compute_checksum
@@ -81,15 +81,11 @@ DRF_DATETIME_FIELD = serializers.DateTimeField()
 
 
 def add_inbox_tags(sender, document: Document, logging_group=None, **kwargs) -> None:
-    if document.owner is not None:
-        tags = get_objects_for_user_owner_aware(
-            document.owner,
-            "documents.view_tag",
-            Tag,
-        )
-    else:
-        tags = Tag.objects.all()
-    inbox_tags = tags.filter(is_inbox_tag=True)
+    inbox_tags = restrict_queryset_to_visible(
+        Tag.objects.filter(is_inbox_tag=True),
+        document.owner,
+        "view_tag",
+    )
     document.add_nested_tags(inbox_tags)
 
 

src/documents/tests/test_api_documents.py

@@ -3653,34 +3653,6 @@ class TestDocumentApi(DirectoriesMixin, ConsumeTaskMixin, APITestCase):
         )
         self.assertEqual(response.status_code, status.HTTP_200_OK)
 
-    def test_notes_head_and_unsupported_methods(self) -> None:
-        """
-        GIVEN:
-            - A user allowed to view notes on their own document
-        WHEN:
-            - The notes endpoint is requested with HEAD, PUT and PATCH
-        THEN:
-            - HEAD is allowed like GET
-            - PUT and PATCH are rejected with 403 rather than a server error
-        """
-        user = UserFactory()
-        grant_global(user, "view_note", "view_document")
-        doc = DocumentFactory(owner=user)
-        self.client.force_authenticate(user)
-
-        self.assertEqual(
-            self.client.head(f"/api/documents/{doc.pk}/notes/").status_code,
-            status.HTTP_200_OK,
-        )
-        self.assertEqual(
-            self.client.put(f"/api/documents/{doc.pk}/notes/").status_code,
-            status.HTTP_403_FORBIDDEN,
-        )
-        self.assertEqual(
-            self.client.patch(f"/api/documents/{doc.pk}/notes/").status_code,
-            status.HTTP_403_FORBIDDEN,
-        )
-
     def test_delete_note(self) -> None:
         """
         GIVEN:

src/documents/tests/test_api_search.py

@@ -1001,21 +1001,6 @@ class TestDocumentSearchApi(DirectoriesMixin, APITestCase):
         self.assertEqual(response.status_code, status.HTTP_200_OK)
         self.assertEqual(len(response.data), 10)
 
-    def test_search_autocomplete_head_requires_view_document(self) -> None:
-        """
-        GIVEN:
-            - A user without the global view document permission
-        WHEN:
-            - A HEAD request is made for autocomplete
-        THEN:
-            - It is rejected like the equivalent GET request
-        """
-        self.client.force_authenticate(user=UserFactory())
-
-        response = self.client.head("/api/search/autocomplete/?term=test")
-
-        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
-
     def test_search_autocomplete_respect_permissions(self) -> None:
         """
         GIVEN:

src/documents/tests/test_consumer.py

@@ -34,8 +34,10 @@ from documents.tests.utils import GetConsumerMixin
 from paperless_mail.models import MailRule
 from paperless_testing.assertions import FileSystemAssertsMixin
 from paperless_testing.dirs import DirectoriesMixin
+from paperless_testing.factories import TagFactory
 from paperless_testing.factories import UserFactory
 from paperless_testing.fakes.progress import FakeProgressManager
+from paperless_testing.permissions import grant_object
 
 
 class _BaseNewStyleParser:
@@ -514,6 +516,65 @@ class TestConsumer(
         self.assertTrue(group_checker.has_perm("view_document", document))
         self._assert_first_last_send_progress()
 
+    def testInboxTagsScopedToOwner(self) -> None:
+        """
+        GIVEN:
+            - Inbox tags that are owned by the document owner, unowned, owned by
+              another user but shared with the document owner, and owned by
+              another user and not shared
+            - A non-inbox tag owned by the document owner
+        WHEN:
+            - A document is consumed with that owner
+        THEN:
+            - The document gets the owned, unowned and shared inbox tags only
+        """
+        owner = UserFactory()
+        other = UserFactory()
+        owned = TagFactory(is_inbox_tag=True, owner=owner)
+        unowned = TagFactory(is_inbox_tag=True, owner=None)
+        shared = TagFactory(is_inbox_tag=True, owner=other)
+        grant_object(owner, shared, "view_tag")
+        TagFactory(is_inbox_tag=True, owner=other)
+        TagFactory(is_inbox_tag=False, owner=owner)
+
+        with self.get_consumer(
+            self.get_test_file(),
+            DocumentMetadataOverrides(owner_id=owner.pk),
+        ) as consumer:
+            consumer.run()
+
+            document = Document.objects.first()
+            assert document is not None
+
+        self.assertCountEqual(document.tags.all(), [owned, unowned, shared])
+        self._assert_first_last_send_progress()
+
+    def testInboxTagsUnownedDocument(self) -> None:
+        """
+        GIVEN:
+            - Inbox tags owned by two different users, an unowned inbox tag and
+              a non-inbox tag
+        WHEN:
+            - A document is consumed with no owner
+        THEN:
+            - The document gets every inbox tag
+        """
+        inbox_tags = [
+            TagFactory(is_inbox_tag=True, owner=UserFactory()),
+            TagFactory(is_inbox_tag=True, owner=UserFactory()),
+            TagFactory(is_inbox_tag=True, owner=None),
+        ]
+        TagFactory(is_inbox_tag=False)
+
+        with self.get_consumer(self.get_test_file()) as consumer:
+            consumer.run()
+
+            document = Document.objects.first()
+            assert document is not None
+
+        self.assertCountEqual(document.tags.all(), inbox_tags)
+        self._assert_first_last_send_progress()
+
     def testNotAFile(self) -> None:
         with self.assertRaisesMessage(ConsumerError, "File not found"):
             with self.get_consumer(Path("non-existing-file")) as consumer:

src/documents/views.py

@@ -175,12 +175,12 @@ 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 get_document_count_filter_for_user
-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 permitted_object_ids
+from documents.permissions import restrict_queryset_to_visible
 from documents.permissions import set_permissions_for_objects
 from documents.permissions import user_is_unrestricted
 from documents.plugins.date_parsing import get_date_parser
@@ -3836,49 +3836,51 @@ class GlobalSearchView(PassUserMixin):
                     if doc_id in docs_by_id
                 ][:OBJECT_LIMIT]
         saved_views = (
-            get_objects_for_user_owner_aware(
+            restrict_queryset_to_visible(
+                SavedView.objects.filter(name__icontains=query),
                 request.user,
                 "view_savedview",
-                SavedView,
-            ).filter(name__icontains=query)
+            )
             if request.user.has_perm("documents.view_savedview")
             else []
         )
         saved_views = saved_views[:OBJECT_LIMIT]
         tags = (
-            get_objects_for_user_owner_aware(request.user, "view_tag", Tag).filter(
-                name__icontains=query,
+            restrict_queryset_to_visible(
+                Tag.objects.filter(name__icontains=query),
+                request.user,
+                "view_tag",
             )
             if request.user.has_perm("documents.view_tag")
             else []
         )
         tags = tags[:OBJECT_LIMIT]
         correspondents = (
-            get_objects_for_user_owner_aware(
+            restrict_queryset_to_visible(
+                Correspondent.objects.filter(name__icontains=query),
                 request.user,
                 "view_correspondent",
-                Correspondent,
-            ).filter(name__icontains=query)
+            )
             if request.user.has_perm("documents.view_correspondent")
             else []
         )
         correspondents = correspondents[:OBJECT_LIMIT]
         document_types = (
-            get_objects_for_user_owner_aware(
+            restrict_queryset_to_visible(
+                DocumentType.objects.filter(name__icontains=query),
                 request.user,
                 "view_documenttype",
-                DocumentType,
-            ).filter(name__icontains=query)
+            )
             if request.user.has_perm("documents.view_documenttype")
             else []
         )
         document_types = document_types[:OBJECT_LIMIT]
         storage_paths = (
-            get_objects_for_user_owner_aware(
+            restrict_queryset_to_visible(
+                StoragePath.objects.filter(name__icontains=query),
                 request.user,
                 "view_storagepath",
-                StoragePath,
-            ).filter(name__icontains=query)
+            )
             if request.user.has_perm("documents.view_storagepath")
             else []
         )
@@ -3896,21 +3898,21 @@ class GlobalSearchView(PassUserMixin):
         )
         groups = groups[:OBJECT_LIMIT]
         mail_rules = (
-            get_objects_for_user_owner_aware(
+            restrict_queryset_to_visible(
+                MailRule.objects.filter(name__icontains=query),
                 request.user,
                 "view_mailrule",
-                MailRule,
-            ).filter(name__icontains=query)
+            )
             if request.user.has_perm("paperless_mail.view_mailrule")
             else []
         )
         mail_rules = mail_rules[:OBJECT_LIMIT]
         mail_accounts = (
-            get_objects_for_user_owner_aware(
+            restrict_queryset_to_visible(
+                MailAccount.objects.filter(name__icontains=query),
                 request.user,
                 "view_mailaccount",
-                MailAccount,
-            ).filter(name__icontains=query)
+            )
             if request.user.has_perm("paperless_mail.view_mailaccount")
             else []
         )
@@ -4026,33 +4028,33 @@ class StatisticsView(GenericAPIView[Any]):
         tags = (
             Tag.objects.all()
             if can_view_global_stats
-            else get_objects_for_user_owner_aware(user, "documents.view_tag", Tag)
+            else restrict_queryset_to_visible(Tag.objects.all(), user, "view_tag")
         ).only("id", "is_inbox_tag")
         correspondent_count = (
             Correspondent.objects.count()
             if can_view_global_stats
-            else get_objects_for_user_owner_aware(
+            else restrict_queryset_to_visible(
+                Correspondent.objects.all(),
                 user,
-                "documents.view_correspondent",
-                Correspondent,
+                "view_correspondent",
             ).count()
         )
         document_type_count = (
             DocumentType.objects.count()
             if can_view_global_stats
-            else get_objects_for_user_owner_aware(
+            else restrict_queryset_to_visible(
+                DocumentType.objects.all(),
                 user,
-                "documents.view_documenttype",
-                DocumentType,
+                "view_documenttype",
             ).count()
         )
         storage_path_count = (
             StoragePath.objects.count()
             if can_view_global_stats
-            else get_objects_for_user_owner_aware(
+            else restrict_queryset_to_visible(
+                StoragePath.objects.all(),
                 user,
-                "documents.view_storagepath",
-                StoragePath,
+                "view_storagepath",
             ).count()
         )
 

src/paperless_mail/serialisers.py

@@ -2,8 +2,8 @@ from django.utils.translation import gettext as _
 from rest_framework import serializers
 from rest_framework.exceptions import PermissionDenied
 
-from documents.permissions import get_objects_for_user_owner_aware
 from documents.permissions import has_perms_owner_aware
+from documents.permissions import restrict_queryset_to_visible
 from documents.serialisers import CorrespondentField
 from documents.serialisers import DocumentTypeField
 from documents.serialisers import OwnedObjectSerializer
@@ -68,10 +68,10 @@ class AccountField(serializers.PrimaryKeyRelatedField[MailAccount]):
         if user is None:
             return MailAccount.objects.none()
 
-        return get_objects_for_user_owner_aware(
+        return restrict_queryset_to_visible(
+            MailAccount.objects.all(),
             user,
             "change_mailaccount",
-            MailAccount,
         ).order_by("-id")