Chore: type GlobalPermissionsByMethod.perms_map keys as HTTP methods

perms_map accepted any string key, so a misspelled method or a HEAD entry
would never match. Keys are now a Literal of the mappable methods, without
HEAD, and the map is typed read-only.

src/documents/bulk_edit.py

@@ -34,7 +34,6 @@ from documents.models import DocumentType
 from documents.models import PaperlessTask
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.permissions import PermissionsSpec
 from documents.permissions import set_permissions_for_objects
 from documents.plugins.helpers import DocumentsStatusManager
 from documents.tasks import bulk_update_documents
@@ -446,7 +445,7 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
 
 def set_permissions(
     doc_ids: list[int],
-    set_permissions: PermissionsSpec,
+    set_permissions: dict,
     *,
     owner: User | None = None,
     merge: bool = False,

src/documents/consumer.py

@@ -38,8 +38,7 @@ from documents.models import StoragePath
 from documents.models import Tag
 from documents.models import WorkflowTrigger
 from documents.parsers import ParseError
-from documents.permissions import PermissionsSpec
-from documents.permissions import set_permissions_for_objects
+from documents.permissions import set_permissions_for_object
 from documents.plugins.base import AlwaysRunPluginMixin
 from documents.plugins.base import ConsumeTaskPlugin
 from documents.plugins.base import NoCleanupPluginMixin
@@ -949,7 +948,7 @@ class ConsumerPlugin(
             or self.metadata.change_users is not None
             or self.metadata.change_groups is not None
         ):
-            permissions: PermissionsSpec = {
+            permissions = {
                 "view": {
                     "users": self.metadata.view_users or [],
                     "groups": self.metadata.view_groups or [],
@@ -959,7 +958,7 @@ class ConsumerPlugin(
                     "groups": self.metadata.change_groups or [],
                 },
             }
-            set_permissions_for_objects(permissions, Document, [document.pk])
+            set_permissions_for_object(permissions=permissions, object=document)
 
         if self.metadata.custom_fields:
             for field in CustomField.objects.filter(

src/documents/permissions.py

@@ -1,11 +1,11 @@
-from collections.abc import Iterable
 from collections.abc import Mapping
+from collections.abc import Sequence
 from typing import Any
-from typing import Final
+from typing import ClassVar
 from typing import Literal
-from typing import TypeAlias
-from typing import TypedDict
+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
@@ -26,29 +26,16 @@ from django.db.models.functions import Coalesce
 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
 from rest_framework.permissions import DjangoObjectPermissions
 
 from documents.models import Document
 from documents.versioning import get_root_document
 
-PermissionAction = Literal["view", "change"]
-PERMISSION_ACTIONS: Final[tuple[PermissionAction, ...]] = ("view", "change")
-
-
-class PermissionGrantees(TypedDict, total=False):
-    """
-    The user and group ids to grant one permission action to. A missing key
-    leaves that kind of grant untouched.
-    """
-
-    users: Iterable[int]
-    groups: Iterable[int]
-
-
-PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGrantees]
-
 
 class PaperlessObjectPermissions(DjangoObjectPermissions):
     """
@@ -81,28 +68,23 @@ class PaperlessAdminPermissions(BasePermission):
         return request.user.is_active and request.user.is_staff
 
 
-def has_global_statistics_permission(user: User | None) -> bool:
-    if (
-        user is None
-        or not getattr(user, "is_active", False)
-        or not getattr(user, "is_authenticated", False)
-    ):
-        return False
+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)
+    )
 
-    return getattr(user, "is_superuser", False) or user.has_perm(
-        "paperless.view_global_statistics",
+
+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")
     )
 
 
 def has_system_status_permission(user: User | None) -> bool:
-    if (
-        user is None
-        or not getattr(user, "is_active", False)
-        or not getattr(user, "is_authenticated", False)
-    ):
-        return False
-
-    return (
+    return _is_active_and_authenticated(user) and (
         getattr(user, "is_superuser", False)
         or getattr(user, "is_staff", False)
         or user.has_perm("paperless.view_system_monitoring")
@@ -123,10 +105,87 @@ def get_groups_with_only_permission(obj, codename):
     return Group.objects.filter(id__in=group_object_perm_group_ids).distinct()
 
 
+def set_permissions_for_object(
+    permissions: dict,
+    object,
+    *,
+    merge: bool = False,
+) -> None:
+    """
+    Set permissions for an object. The permissions are given as a mapping of actions
+    to a dict of user / group id lists, e.g.
+    {"view": {"users": [1], "groups": [2]}, "change": {"users": [], "groups": []}}.
+
+    If merge is True, the permissions are merged with the existing permissions and
+    no users or groups are removed. If False, the permissions are set to exactly
+    the given list of users and groups.
+    """
+
+    for action, entry in permissions.items():
+        permission = f"{action}_{object.__class__.__name__.lower()}"
+        if "users" in entry:
+            # users
+            users_to_add = User.objects.filter(id__in=entry["users"])
+            users_to_remove = (
+                get_users_with_perms(
+                    object,
+                    only_with_perms_in=[permission],
+                    with_group_users=False,
+                )
+                if not merge
+                else User.objects.none()
+            )
+            if users_to_add.exists() and users_to_remove.exists():
+                users_to_remove = users_to_remove.exclude(id__in=users_to_add)
+            if users_to_remove.exists():
+                for user in users_to_remove:
+                    remove_perm(permission, user, object)
+            if users_to_add.exists():
+                for user in users_to_add:
+                    assign_perm(permission, user, object)
+                    if action == "change":
+                        # change gives view too
+                        assign_perm(
+                            f"view_{object.__class__.__name__.lower()}",
+                            user,
+                            object,
+                        )
+        if "groups" in entry:
+            # groups
+            groups_to_add = Group.objects.filter(id__in=entry["groups"])
+            groups_to_remove = (
+                get_groups_with_only_permission(
+                    object,
+                    permission,
+                )
+                if not merge
+                else Group.objects.none()
+            )
+            if groups_to_add.exists() and groups_to_remove.exists():
+                groups_to_remove = groups_to_remove.exclude(id__in=groups_to_add)
+            if groups_to_remove.exists():
+                for group in groups_to_remove:
+                    remove_perm(permission, group, object)
+            if groups_to_add.exists():
+                for group in groups_to_add:
+                    assign_perm(permission, group, object)
+                    if action == "change":
+                        # change gives view too
+                        assign_perm(
+                            f"view_{object.__class__.__name__.lower()}",
+                            group,
+                            object,
+                        )
+
+
 def _resolve_permissions(codenames: set[str], ctype: ContentType) -> list[Permission]:
     """
-    Resolve ``codenames`` to Permission rows, raising for any that don't exist
-    so a bad action name fails loudly instead of granting nothing.
+    Resolves `codenames` to Permission rows, raising like the single-object
+    assign_perm() this bulk path replaces does (via a `.get()` internally)
+    if any codename doesn't exist. SetPermissionsSerializer rejects unknown
+    action names at the API, but a caller passing one directly would
+    otherwise get a plain `.filter()` that silently builds zero rows and
+    no-ops instead of reporting the bad input.
     """
     permission_objs = list(
         Permission.objects.filter(content_type=ctype, codename__in=codenames),
@@ -145,13 +204,16 @@ def _apply_bulk_permission_entry(
     perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
     identity_model: type[User] | type[Group],
     identity_field: str,
-    ids: Iterable[int],
+    ids: list[int],
     codename: str,
     permission_objs: list[Permission],
     ctype: ContentType,
     object_pks: list[str],
     merge: bool,
 ) -> None:
+    # Only the ids are needed to build permission rows (via `<field>_id=`),
+    # so avoid fetching full User/Group rows for identities that may not
+    # even end up being granted anything new.
     add_ids = set(
         identity_model.objects.filter(id__in=ids).values_list("id", flat=True),
     )
@@ -189,26 +251,45 @@ def _apply_bulk_permission_entry(
         for pk in object_pks
         for identity_id in add_ids
     ]
-    # ignore_conflicts skips grants that already exist (unique constraint).
+    # ignore_conflicts skips only rows that already exist as an exact
+    # (identity, permission, object) match -- the same de-dup the
+    # underlying (user|group, permission, object_pk) unique constraint
+    # already enforces for the single-object assign_perm() this replaces,
+    # so it doesn't change what counts as "already granted". batch_size
+    # caps how many rows go into a single INSERT statement.
     perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000)
 
 
 def set_permissions_for_objects(
-    permissions: PermissionsSpec,
+    permissions: dict,
     model: type[Model],
     pks: QuerySet | list,
     *,
     merge: bool = False,
 ) -> None:
     """
-    Set the view/change grants for every object in ``pks``. With ``merge``,
-    the given users and groups are added; without it, each given action's
-    direct grants are replaced. Granting change also grants view, and an
-    action entry with neither key is left untouched.
+    Bulk equivalent of set_permissions_for_object: applies the same
+    permission changes to every object identified by `pks` at once.
 
-    guardian's bulk assign_perm is avoided because it skips the direct row
-    for anyone who already has access through a group, so revoking the group
-    later would silently remove access that was granted explicitly.
+    Takes a model + pks (rather than model instances) deliberately -- the
+    permission rows built below only ever need `pk`, `content_type`, and
+    identity ids, so callers shouldn't have to fetch full rows (with every
+    other field) just to hand them to this function.
+
+    Deliberately does not use guardian's queryset/list-aware assign_perm:
+    passing a list as the object routes to bulk_assign_perm, which skips
+    creating a direct permission row for anyone who already has the
+    permission via ANY group membership (it checks
+    ObjectPermissionChecker.has_perm, which is group-inheritance-aware) --
+    unlike the single-object assign_perm this replaces, which always
+    ensures a direct row via get_or_create regardless of group-derived
+    access. Losing that guarantee would mean a later revocation of the
+    group's grant silently strips access an admin explicitly asked to be
+    direct. Bulk-creating rows straight against the permission models
+    instead (see _apply_bulk_permission_entry) preserves the original
+    always-create-a-direct-row semantics while still batching every object
+    and every identity into one query per action, rather than one query per
+    (object, user) pair.
     """
     object_pks = [str(pk) for pk in pks]
     if not object_pks:  # pragma: no cover
@@ -217,9 +298,12 @@ def set_permissions_for_objects(
     model_name = model.__name__.lower()
     ctype = ContentType.objects.get_for_model(model)
 
-    # Resolve every action before writing, so a bad one can't leave the
-    # actions ahead of it applied.
-    permissions_by_action: dict[PermissionAction, list[Permission]] = {}
+    # Every action is resolved up front, before anything is written, so an
+    # unrecognized action name (see _resolve_permissions) aborts the whole
+    # call instead of leaving the actions ahead of it already applied.
+    # SetPermissionsSerializer rejects unknown actions at the API, so this
+    # guards any other caller.
+    permissions_by_action: dict[str, list[Permission]] = {}
     for action, entry in permissions.items():
         if "users" not in entry and "groups" not in entry:
             continue
@@ -622,7 +706,32 @@ def has_perms_owner_aware(user, perms, obj):
     return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
 
 
-class ViewDocumentsPermissions(BasePermission):
+# 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):
     """
     Permissions class that checks for model permissions for only viewing Documents.
     """
@@ -633,31 +742,18 @@ class ViewDocumentsPermissions(BasePermission):
         "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
 
-        return request.user.has_perms(self.perms_map.get(request.method, []))
-
-
-class TrashPermissions(BasePermission):
+class TrashPermissions(GlobalPermissionsByMethod):
     """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
 
-        return request.user.has_perms(self.perms_map.get(request.method, []))
-
-
-class PaperlessNotePermissions(BasePermission):
+class PaperlessNotePermissions(GlobalPermissionsByMethod):
     """
     Permissions class that checks for model permissions for Notes.
     """
@@ -677,28 +773,13 @@ class PaperlessNotePermissions(BasePermission):
         ],
     }
 
-    def has_permission(self, request, view):
-        if not request.user or (not request.user.is_authenticated):  # pragma: no cover
-            return False
 
-        perms = self.perms_map[request.method]
-
-        return request.user.has_perms(perms)
-
-
-class AcknowledgeTasksPermissions(BasePermission):
+class AcknowledgeTasksPermissions(GlobalPermissionsByMethod):
     """
     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/serialisers.py

@@ -78,16 +78,12 @@ from documents.models import WorkflowActionEmail
 from documents.models import WorkflowActionWebhook
 from documents.models import WorkflowTrigger
 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 PermissionsSpec
 from documents.permissions import get_document_count_filter_for_user
 from documents.permissions import get_groups_with_only_permission
 from documents.permissions import has_perms_owner_aware
 from documents.permissions import permitted_document_ids
 from documents.permissions import restrict_queryset_to_visible
-from documents.permissions import set_permissions_for_objects
+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
 from documents.templating.utils import convert_format_str_to_template_format
@@ -215,6 +211,9 @@ class MatchingModelSerializer(serializers.ModelSerializer[Any]):
         return match
 
 
+PERMISSION_ACTIONS = ("view", "change")
+
+
 class SetPermissionsMixin:
     def _validate_user_ids(self, user_ids):
         users = User.objects.none()
@@ -236,10 +235,8 @@ class SetPermissionsMixin:
                 )
         return groups
 
-    def validate_set_permissions(self, set_permissions=None) -> PermissionsSpec:
-        permissions_dict: dict[PermissionAction, PermissionGrantees] = {
-            action: {} for action in PERMISSION_ACTIONS
-        }
+    def validate_set_permissions(self, set_permissions=None):
+        permissions_dict = {action: {} for action in PERMISSION_ACTIONS}
         if set_permissions is not None:
             for action in PERMISSION_ACTIONS:
                 if action in set_permissions:
@@ -247,18 +244,18 @@ class SetPermissionsMixin:
                         users = set_permissions[action]["users"]
                         permissions_dict[action]["users"] = self._validate_user_ids(
                             users,
-                        ).values_list("id", flat=True)
+                        )
                     if "groups" in set_permissions[action]:
                         groups = set_permissions[action]["groups"]
                         permissions_dict[action]["groups"] = self._validate_group_ids(
                             groups,
-                        ).values_list("id", flat=True)
+                        )
                 else:
                     del permissions_dict[action]
         return permissions_dict
 
-    def _set_permissions(self, permissions: PermissionsSpec, object) -> None:
-        set_permissions_for_objects(permissions, type(object), [object.pk])
+    def _set_permissions(self, permissions, object) -> None:
+        set_permissions_for_object(permissions, object)
 
 
 class SerializerWithPerms(serializers.Serializer[dict[str, Any]]):
@@ -3089,7 +3086,7 @@ class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
             )
         return objects
 
-    def _validate_permissions(self, permissions) -> PermissionsSpec:
+    def _validate_permissions(self, permissions) -> dict:
         return self.validate_set_permissions(
             permissions,
         )

src/documents/tests/test_api_documents.py

@@ -3653,6 +3653,34 @@ 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,6 +1001,21 @@ 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_bulk_edit.py

@@ -23,7 +23,6 @@ from documents.models import Document
 from documents.models import DocumentType
 from documents.models import StoragePath
 from documents.models import Tag
-from documents.permissions import PermissionsSpec
 from documents.permissions import set_permissions_for_objects
 from paperless_testing.dirs import DirectoriesMixin
 from paperless_testing.factories import DocumentFactory
@@ -446,7 +445,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
 
         grant_object(self.group1, self.doc1, "view_document")
 
-        permissions: PermissionsSpec = {
+        permissions = {
             "view": {
                 "users": [self.user1.id, self.user2.id],
                 "groups": [self.group2.id],
@@ -489,7 +488,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         grant_object(self.user1, self.doc1, "view_document")
         grant_object(self.group1, self.doc1, "view_document")
 
-        permissions: PermissionsSpec = {
+        permissions = {
             "view": {
                 "users": [self.user2.id],
                 "groups": [self.group2.id],
@@ -540,7 +539,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
               batched call rather than one call per (document, identity)
               pair
         """
-        permissions: PermissionsSpec = {
+        permissions = {
             "view": {
                 "users": [self.user1.id, self.user2.id],
                 "groups": [self.group2.id],
@@ -647,7 +646,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         """
         with self.assertRaises(Permission.DoesNotExist):
             set_permissions_for_objects(
-                {"not_a_real_action": {"users": [self.user1.id], "groups": []}},  # type: ignore[bad-assignment]
+                {"not_a_real_action": {"users": [self.user1.id], "groups": []}},
                 Document,
                 [self.doc1.pk],
             )
@@ -675,7 +674,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
             set_permissions_for_objects(
                 {
                     "view": {"users": [self.user1.id], "groups": []},
-                    "not_a_real_action": {"users": [self.user1.id], "groups": []},  # type: ignore[bad-assignment]
+                    "not_a_real_action": {"users": [self.user1.id], "groups": []},
                 },
                 Document,
                 [self.doc1.pk],

src/documents/workflows/mutations.py

@@ -7,8 +7,7 @@ from documents.data_models import DocumentMetadataOverrides
 from documents.models import CustomFieldInstance
 from documents.models import Document
 from documents.models import WorkflowAction
-from documents.permissions import PermissionsSpec
-from documents.permissions import set_permissions_for_objects
+from documents.permissions import set_permissions_for_object
 from documents.templating.workflows import parse_w_workflow_placeholders
 
 logger = logging.getLogger("paperless.workflows.mutations")
@@ -74,7 +73,7 @@ def apply_assignment_to_document(
             action.has_assign_change_groups,
         ],
     ):
-        permissions: PermissionsSpec = {
+        permissions = {
             "view": {
                 "users": action.assign_view_users.values_list("id", flat=True),
                 "groups": action.assign_view_groups.values_list("id", flat=True),
@@ -84,10 +83,9 @@ def apply_assignment_to_document(
                 "groups": action.assign_change_groups.values_list("id", flat=True),
             },
         }
-        set_permissions_for_objects(
-            permissions,
-            Document,
-            [document.pk],
+        set_permissions_for_object(
+            permissions=permissions,
+            object=document,
             merge=True,
         )
 
@@ -245,14 +243,13 @@ def apply_removal_to_document(
         document.owner = None
 
     if action.remove_all_permissions:
-        permissions: PermissionsSpec = {
+        permissions = {
             "view": {"users": [], "groups": []},
             "change": {"users": [], "groups": []},
         }
-        set_permissions_for_objects(
-            permissions,
-            Document,
-            [document.pk],
+        set_permissions_for_object(
+            permissions=permissions,
+            object=document,
             merge=False,
         )