Chore: consolidate direct grant lookups into get_permission_grantees

Six places listed who holds a permission on an object, each with its own
query and result shape, and one had already drifted into expanding group
members.

src/documents/data_models.py

@@ -5,8 +5,6 @@ from pathlib import Path
 from typing import TypedDict
 
 import magic
-from guardian.shortcuts import get_groups_with_perms
-from guardian.shortcuts import get_users_with_perms
 
 
 class StoredBarcode(TypedDict):
@@ -117,6 +115,9 @@ class DocumentMetadataOverrides:
         """
         Fills in the overrides from a document object
         """
+        # documents.models imports this module, so this can't be a top-level import
+        from documents.permissions import get_permission_grantees
+
         overrides = DocumentMetadataOverrides()
         overrides.title = doc.title
         overrides.correspondent_id = doc.correspondent.id if doc.correspondent else None
@@ -126,40 +127,16 @@ class DocumentMetadataOverrides:
         overrides.tag_ids = list(doc.tags.values_list("id", flat=True))
         overrides.created = doc.created
 
-        overrides.view_users = list(
-            get_users_with_perms(
-                doc,
-                only_with_perms_in=["view_document"],
-                with_group_users=False,
-            ).values_list("id", flat=True),
-        )
-        overrides.change_users = list(
-            get_users_with_perms(
-                doc,
-                only_with_perms_in=["change_document"],
-                with_group_users=False,
-            ).values_list("id", flat=True),
-        )
+        grantees = get_permission_grantees(type(doc), [doc.pk])[doc.pk]
+        overrides.view_users = grantees["view"]["users"]
+        overrides.view_groups = grantees["view"]["groups"]
+        overrides.change_users = grantees["change"]["users"]
+        overrides.change_groups = grantees["change"]["groups"]
         overrides.custom_fields = {
             custom_field.field.id: custom_field.value
             for custom_field in doc.custom_fields.all()
         }
 
-        groups_with_perms = get_groups_with_perms(
-            doc,
-            attach_perms=True,
-        )
-        overrides.view_groups = [
-            group.id
-            for group in groups_with_perms
-            if "view_document" in groups_with_perms[group]
-        ]
-        overrides.change_groups = [
-            group.id
-            for group in groups_with_perms
-            if "change_document" in groups_with_perms[group]
-        ]
-
         return overrides
 
 

src/documents/permissions.py

@@ -37,7 +37,17 @@ PermissionAction = Literal["view", "change"]
 PERMISSION_ACTIONS: Final[tuple[PermissionAction, ...]] = ("view", "change")
 
 
-class PermissionGrantees(TypedDict, total=False):
+class PermissionGrantees(TypedDict):
+    """
+    The user and group ids holding one permission action directly on an
+    object. Users who only hold it through a group are not listed.
+    """
+
+    users: list[int]
+    groups: list[int]
+
+
+class PermissionGranteesUpdate(TypedDict, total=False):
     """
     The user and group ids to grant one permission action to. A missing key
     leaves that kind of grant untouched.
@@ -47,7 +57,7 @@ class PermissionGrantees(TypedDict, total=False):
     groups: Iterable[int]
 
 
-PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGrantees]
+PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGranteesUpdate]
 
 
 class PaperlessObjectPermissions(DjangoObjectPermissions):
@@ -249,6 +259,56 @@ def set_permissions_for_objects(
             )
 
 
+def get_permission_grantees(
+    model: type[Model],
+    pks: Iterable[int],
+    actions: Iterable[PermissionAction] = PERMISSION_ACTIONS,
+) -> dict[int, dict[PermissionAction, PermissionGrantees]]:
+    """
+    Return the users and groups holding each of ``actions`` directly on each
+    object of ``model`` identified by ``pks``, in two queries however many
+    objects are asked about.
+
+    Every requested pk and action has an entry, with empty lists where nothing
+    is granted. Ids are sorted. Group grants stay group ids rather than being
+    expanded to their members, so a caller that copies them keeps access tied
+    to current group membership.
+    """
+    pks = list(pks)
+    actions = list(actions)
+    model_name = model.__name__.lower()
+    action_by_codename: dict[str, PermissionAction] = {
+        f"{action}_{model_name}": action for action in actions
+    }
+    result: dict[int, dict[PermissionAction, PermissionGrantees]] = {
+        pk: {action: {"users": [], "groups": []} for action in actions} for pk in pks
+    }
+    if not pks or not actions:
+        return result
+
+    ctype = ContentType.objects.get_for_model(model)
+    grant_filter = {
+        "content_type": ctype,
+        "permission__content_type": ctype,
+        "permission__codename__in": list(action_by_codename),
+        "object_pk__in": [str(pk) for pk in pks],
+    }
+    for perm_model, identity_field, key in (
+        (UserObjectPermission, "user_id", "users"),
+        (GroupObjectPermission, "group_id", "groups"),
+    ):
+        rows = (
+            perm_model.objects.filter(**grant_filter)
+            .values_list("object_pk", "permission__codename", identity_field)
+            .order_by(identity_field)
+        )
+        for object_pk, codename, identity_id in rows:
+            result[int(object_pk)][action_by_codename[codename]][key].append(
+                identity_id,
+            )
+    return result
+
+
 def permitted_object_ids(
     user: User | None,
     model: type[Model],

src/documents/search/_backend.py

@@ -48,8 +48,6 @@ if TYPE_CHECKING:
     from pathlib import Path
 
     from django.contrib.auth.models import AbstractUser
-    from django.contrib.auth.models import Group
-    from django.contrib.auth.models import User
     from django.db.models import QuerySet
 
     from documents.models import Document
@@ -528,8 +526,8 @@ class TantivyBackend:
         Annotate the queryset with ``annotate_effective_content`` when indexing
         more than a couple of documents, to resolve that without a query each.
         """
-        from guardian.shortcuts import get_groups_with_perms
-        from guardian.shortcuts import get_users_with_perms
+        from documents.models import Document
+        from documents.permissions import get_permission_grantees
 
         # Every searchable string is normalized on the way in, and every
         # query string on the way out (_parse_query), so the two agree on
@@ -674,28 +672,18 @@ class TantivyBackend:
             doc.add_unsigned("owner_id", document.owner_id)
 
         # Viewers with permission
-        if viewer_ids is None:
-            users_with_perms = get_users_with_perms(
-                document,
-                only_with_perms_in=["view_document"],
-                with_group_users=False,
-            )
-            viewer_ids = list(
-                cast("QuerySet[User]", users_with_perms).values_list("id", flat=True),
-            )
+        if viewer_ids is None or viewer_group_ids is None:
+            view = get_permission_grantees(
+                Document,
+                [document.pk],
+                actions=("view",),
+            )[document.pk]["view"]
+            if viewer_ids is None:
+                viewer_ids = view["users"]
+            if viewer_group_ids is None:
+                viewer_group_ids = view["groups"]
         for viewer_id in viewer_ids:
             doc.add_unsigned("viewer_id", viewer_id)
-        if viewer_group_ids is None:
-            groups_with_perms = get_groups_with_perms(
-                document,
-                only_with_perms_in=["view_document"],
-            )
-            viewer_group_ids = list(
-                cast("QuerySet[Group]", groups_with_perms).values_list(
-                    "id",
-                    flat=True,
-                ),
-            )
         for viewer_group_id in viewer_group_ids:
             doc.add_unsigned("viewer_group_id", viewer_group_id)
 
@@ -1243,48 +1231,19 @@ def _bulk_get_viewer_permissions(
     requesting user's current memberships. Expanding groups to user IDs here
     would leave stale access behind after a user is removed from a group.
     """
-    from collections import defaultdict
-
-    from django.contrib.contenttypes.models import ContentType
-    from guardian.models import GroupObjectPermission
-    from guardian.models import UserObjectPermission
-
     from documents.models import Document
-
-    # get_for_model is cached by Django, so this costs at most one query total.
-    ct = ContentType.objects.get_for_model(Document)
-    str_pks = [str(pk) for pk in doc_pks]
-
-    viewer_map: dict[int, set[int]] = defaultdict(set)
-    viewer_group_map: dict[int, set[int]] = defaultdict(set)
-
-    # Fold the permission lookup into the query via a join on codename instead
-    # of a separate Permission.objects.get(), which would otherwise run once per
-    # chunk during a full reindex.
-    user_qs = UserObjectPermission.objects.filter(
-        content_type=ct,
-        permission__content_type=ct,
-        permission__codename="view_document",
-        object_pk__in=str_pks,
-    ).values_list("object_pk", "user_id")
-    for object_pk, user_id in user_qs:
-        viewer_map[int(object_pk)].add(user_id)
-
-    group_qs = GroupObjectPermission.objects.filter(
-        content_type=ct,
-        permission__content_type=ct,
-        permission__codename="view_document",
-        object_pk__in=str_pks,
-    ).values_list("object_pk", "group_id")
-    for object_pk, group_id in group_qs:
-        viewer_group_map[int(object_pk)].add(group_id)
+    from documents.permissions import get_permission_grantees
 
     return {
-        object_pk: ViewerGrant(
-            viewer_ids=list(viewer_map.get(object_pk, ())),
-            viewer_group_ids=list(viewer_group_map.get(object_pk, ())),
+        pk: ViewerGrant(
+            viewer_ids=grantees["view"]["users"],
+            viewer_group_ids=grantees["view"]["groups"],
         )
-        for object_pk in viewer_map.keys() | viewer_group_map.keys()
+        for pk, grantees in get_permission_grantees(
+            Document,
+            doc_pks,
+            actions=("view",),
+        ).items()
     }
 
 

src/documents/serialisers.py

@@ -8,7 +8,6 @@ from datetime import timedelta
 from decimal import Decimal
 from typing import TYPE_CHECKING
 from typing import Any
-from typing import Literal
 from typing import TypedDict
 
 import magic
@@ -39,8 +38,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_groups_with_perms
-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
 from rest_framework import fields
@@ -82,8 +79,10 @@ from documents.parsers import is_mime_type_supported
 from documents.permissions import PERMISSION_ACTIONS
 from documents.permissions import PermissionAction
 from documents.permissions import PermissionGrantees
+from documents.permissions import PermissionGranteesUpdate
 from documents.permissions import PermissionsSpec
 from documents.permissions import get_document_count_filter_for_user
+from documents.permissions import get_permission_grantees
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import permitted_document_ids
 from documents.permissions import restrict_queryset_to_visible
@@ -237,7 +236,7 @@ class SetPermissionsMixin:
         return groups
 
     def validate_set_permissions(self, set_permissions=None) -> PermissionsSpec:
-        permissions_dict: dict[PermissionAction, PermissionGrantees] = {
+        permissions_dict: dict[PermissionAction, PermissionGranteesUpdate] = {
             action: {} for action in PERMISSION_ACTIONS
         }
         if set_permissions is not None:
@@ -314,35 +313,6 @@ class OwnedObjectSerializer(
             except KeyError:
                 pass
 
-    def _get_perms(self, obj, codename: str, target: Literal["users", "groups"]):
-        """
-        Get the given permissions from context or from django-guardian.
-
-        :param codename: The permission codename, e.g. 'view' or 'change'
-        :param target: 'users' or 'groups'
-        """
-        key = f"{target}_{codename}_perms"
-        cached = self.context.get(key, {}).get(obj.pk)
-        if cached is not None:
-            return list(cached)
-
-        # Permission not found in the context, get it from guardian
-        if target == "users":
-            return list(
-                get_users_with_perms(
-                    obj,
-                    only_with_perms_in=[f"{codename}_{obj.__class__.__name__.lower()}"],
-                    with_group_users=False,
-                ).values_list("id", flat=True),
-            )
-        else:  # groups
-            return list(
-                get_groups_with_perms(
-                    obj,
-                    only_with_perms_in=[f"{codename}_{obj.__class__.__name__.lower()}"],
-                ).values_list("id", flat=True),
-            )
-
     @extend_schema_field(
         field={
             "type": "object",
@@ -376,17 +346,12 @@ class OwnedObjectSerializer(
             },
         },
     )
-    def get_permissions(self, obj) -> dict:
-        return {
-            "view": {
-                "users": self._get_perms(obj, "view", "users"),
-                "groups": self._get_perms(obj, "view", "groups"),
-            },
-            "change": {
-                "users": self._get_perms(obj, "change", "users"),
-                "groups": self._get_perms(obj, "change", "groups"),
-            },
-        }
+    def get_permissions(self, obj) -> dict[PermissionAction, PermissionGrantees]:
+        # Prefer the page-level batch computed by BulkPermissionMixin.
+        grantees = self.context.get("permission_grantees", {}).get(obj.pk)
+        if grantees is not None:
+            return grantees
+        return get_permission_grantees(type(obj), [obj.pk])[obj.pk]
 
     def get_user_can_change(self, obj) -> bool:
         if obj.owner is None or obj.owner == self.user:
@@ -402,18 +367,18 @@ class OwnedObjectSerializer(
         # Prefer the page-level batch computed by BulkPermissionMixin
         # (get_serializer_context) over a fresh per-object guardian check,
         # which would otherwise query the permission tables once per row.
-        users_change_perms = self.context.get("users_change_perms")
-        groups_change_perms = self.context.get("groups_change_perms")
-        if users_change_perms is not None and groups_change_perms is not None:
-            if self.user.pk in users_change_perms.get(obj.pk, []):
+        grantees_by_pk = self.context.get("permission_grantees")
+        if grantees_by_pk is not None:
+            change = grantees_by_pk.get(obj.pk, {}).get("change")
+            if change is None:
+                return False
+            if self.user.pk in change["users"]:
                 return True
             user_group_ids = getattr(self, "_user_group_ids", None)
             if user_group_ids is None:
                 user_group_ids = set(self.user.groups.values_list("id", flat=True))
                 self._user_group_ids = user_group_ids
-            return bool(
-                user_group_ids.intersection(groups_change_perms.get(obj.pk, [])),
-            )
+            return bool(user_group_ids.intersection(change["groups"]))
 
         checker = ObjectPermissionChecker(self.user)
         return checker.has_perm(f"change_{obj.__class__.__name__.lower()}", obj)

src/documents/tests/test_permission_filtering_security.py

@@ -15,9 +15,11 @@ from documents.matching import match_document_types
 from documents.matching import match_storage_paths
 from documents.matching import match_tags
 from documents.models import Correspondent
+from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
+from documents.permissions import get_permission_grantees
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import permitted_document_ids
 from documents.permissions import permitted_object_ids
@@ -1372,3 +1374,101 @@ class TestRestrictQuerysetToVisible:
 
         assert own.pk in visible_ids
         assert hidden.pk not in visible_ids
+
+
+@pytest.mark.django_db
+class TestGetPermissionGrantees:
+    def test_lists_direct_grants_per_object_and_action(self) -> None:
+        """
+        GIVEN:
+            - Two documents, the first with view and change granted to a user
+              and a group, the second with only view granted to another user
+            - A user who is a member of the granted group
+        WHEN:
+            - get_permission_grantees() is called for both documents
+        THEN:
+            - Each document and action lists exactly its direct grants
+            - The group member is not listed as a user
+        """
+        direct = UserFactory()
+        other = UserFactory()
+        member = UserFactory()
+        group = Group.objects.create(name="grantees")
+        member.groups.add(group)
+        first = DocumentFactory()
+        second = DocumentFactory()
+        grant_object(direct, first, "view_document", "change_document")
+        grant_object(group, first, "view_document", "change_document")
+        grant_object(other, second, "view_document")
+
+        grantees = get_permission_grantees(Document, [first.pk, second.pk])
+
+        assert grantees == {
+            first.pk: {
+                "view": {"users": [direct.pk], "groups": [group.pk]},
+                "change": {"users": [direct.pk], "groups": [group.pk]},
+            },
+            second.pk: {
+                "view": {"users": [other.pk], "groups": []},
+                "change": {"users": [], "groups": []},
+            },
+        }
+
+    def test_ids_are_sorted(self) -> None:
+        """
+        GIVEN:
+            - A tag with view granted to three users in reverse creation order
+        WHEN:
+            - get_permission_grantees() is called for the tag
+        THEN:
+            - The user ids come back in ascending order
+        """
+        users = [UserFactory() for _ in range(3)]
+        tag = TagFactory()
+        for user in reversed(users):
+            grant_object(user, tag, "view_tag")
+
+        grantees = get_permission_grantees(Tag, [tag.pk], actions=("view",))
+
+        assert grantees[tag.pk]["view"]["users"] == sorted(user.pk for user in users)
+
+    def test_only_requested_actions_are_returned(self) -> None:
+        """
+        GIVEN:
+            - A tag with view and change granted to a user
+        WHEN:
+            - get_permission_grantees() is called for the view action only
+        THEN:
+            - Only the view action is returned
+        """
+        user = UserFactory()
+        tag = TagFactory()
+        grant_object(user, tag, "view_tag", "change_tag")
+
+        grantees = get_permission_grantees(Tag, [tag.pk], actions=("view",))
+
+        assert grantees == {tag.pk: {"view": {"users": [user.pk], "groups": []}}}
+
+    def test_grants_on_another_model_with_the_same_pk_are_ignored(self) -> None:
+        """
+        GIVEN:
+            - A tag and a correspondent sharing a primary key, with view
+              granted to a user on the tag only
+        WHEN:
+            - get_permission_grantees() is called for the correspondent
+        THEN:
+            - No grants are returned
+        """
+        user = UserFactory()
+        tag = TagFactory()
+        correspondent = CorrespondentFactory(pk=tag.pk)
+        grant_object(user, tag, "view_tag")
+
+        grantees = get_permission_grantees(Correspondent, [correspondent.pk])
+
+        assert grantees == {
+            correspondent.pk: {
+                "view": {"users": [], "groups": []},
+                "change": {"users": [], "groups": []},
+            },
+        }

src/documents/views.py

@@ -5,7 +5,6 @@ import platform
 import re
 import tempfile
 import zipfile
-from collections import defaultdict
 from collections import deque
 from collections.abc import Callable
 from datetime import datetime
@@ -17,7 +16,6 @@ from time import sleep
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import Final
-from typing import Literal
 from typing import NamedTuple
 from unicodedata import normalize
 from urllib.parse import quote
@@ -29,7 +27,6 @@ import pathvalidate
 from django.conf import settings
 from django.contrib.auth.models import Group
 from django.contrib.auth.models import User
-from django.contrib.contenttypes.models import ContentType
 from django.core.cache import cache
 from django.db import connections
 from django.db.migrations.loader import MigrationLoader
@@ -77,8 +74,6 @@ from drf_spectacular.utils import extend_schema
 from drf_spectacular.utils import extend_schema_serializer
 from drf_spectacular.utils import extend_schema_view
 from drf_spectacular.utils import inline_serializer
-from guardian.utils import get_group_obj_perms_model
-from guardian.utils import get_user_obj_perms_model
 from langdetect import detect
 from packaging import version as packaging_version
 from redis import Redis
@@ -176,6 +171,7 @@ 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 get_permission_grantees
 from documents.permissions import has_global_statistics_permission
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import has_system_status_permission
@@ -455,42 +451,6 @@ class BulkPermissionMixin:
     Prefetch Django-Guardian permissions for a list before serialization, to avoid N+1 queries.
     """
 
-    def _get_object_perms(
-        self,
-        objects: list,
-        perm_codenames: list[str],
-        actor: Literal["users", "groups"],
-    ) -> dict[int, dict[str, list[int]]]:
-        """
-        Collect object-level permissions for either users or groups.
-        """
-        model = self.queryset.model
-        obj_perm_model = (
-            get_user_obj_perms_model(model)
-            if actor == "users"
-            else get_group_obj_perms_model(model)
-        )
-        id_field = "user_id" if actor == "users" else "group_id"
-        ctype = ContentType.objects.get_for_model(model)
-        object_pks = [obj.pk for obj in objects]
-
-        perms_qs = obj_perm_model.objects.filter(
-            content_type=ctype,
-            object_pk__in=object_pks,
-            permission__codename__in=perm_codenames,
-        ).values_list("object_pk", id_field, "permission__codename")
-
-        perms: dict[int, dict[str, list[int]]] = defaultdict(lambda: defaultdict(list))
-        for object_pk, actor_id, codename in perms_qs:
-            perms[int(object_pk)][codename].append(actor_id)
-
-        # Ensure that all objects have all codenames, even if empty
-        for pk in object_pks:
-            for codename in perm_codenames:
-                perms[pk][codename]
-
-        return perms
-
     def get_serializer_context(self):
         """
         Get all permissions of the current list of objects at once and pass them to the serializer.
@@ -501,7 +461,7 @@ class BulkPermissionMixin:
         if getattr(self, "action", None) != "list":
             # Batching only pays off across a page of objects; for single-object
             # actions (retrieve, update, ...) the per-object fallback in
-            # get_user_can_change()/_get_perms() is cheap and avoids scanning
+            # get_user_can_change()/get_permissions() is cheap and avoids scanning
             # the whole queryset here.
             return context
 
@@ -514,33 +474,10 @@ class BulkPermissionMixin:
         else:
             queryset = self.filter_queryset(self.get_queryset())
 
-        model_name = self.queryset.model.__name__.lower()
-        permission_name_view = f"view_{model_name}"
-        permission_name_change = f"change_{model_name}"
-
-        user_perms = self._get_object_perms(
-            objects=queryset,
-            perm_codenames=[permission_name_view, permission_name_change],
-            actor="users",
+        context["permission_grantees"] = get_permission_grantees(
+            self.queryset.model,
+            [obj.pk for obj in queryset],
         )
-        group_perms = self._get_object_perms(
-            objects=queryset,
-            perm_codenames=[permission_name_view, permission_name_change],
-            actor="groups",
-        )
-
-        context["users_view_perms"] = {
-            pk: user_perms[pk][permission_name_view] for pk in user_perms
-        }
-        context["users_change_perms"] = {
-            pk: user_perms[pk][permission_name_change] for pk in user_perms
-        }
-        context["groups_view_perms"] = {
-            pk: group_perms[pk][permission_name_view] for pk in group_perms
-        }
-        context["groups_change_perms"] = {
-            pk: group_perms[pk][permission_name_change] for pk in group_perms
-        }
 
         return context