Merge branch 'dev' into feature/ocr-templates

SECURITY.md

@@ -61,6 +61,7 @@ The following are not generally considered vulnerabilities unless accompanied by
 - claims based solely on the presence of a library, framework feature or code pattern without a working exploit
 - pickle deserialization of internal data from trusted components such as the Redis-compatible broker or Paperless-ngx data directory
 - users with permission to edit users granting themselves additional privileges; this is expected behavior for that trusted permission
+- users with change permission on a mail account redirecting stored credentials to another mail server, including through connectivity tests. Mail account editing is a trusted permission, and UI password masking does not imply credential isolation from account editors
 - reports that rely on admin-level access, application-configuration access, workflow-editing privileges, shell access, or other high-trust roles unless they demonstrate an unintended privilege boundary bypass
 - optional webhook, mail, AI, OCR, or integration behavior described without a product-level vulnerability
 - missing limits or hardening settings presented without concrete impact

docs/configuration.md

@@ -803,6 +803,17 @@ system. See the corresponding
 
     Defaults to None
 
+#### [`PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR=<comma-separated-list>`](#PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR) {#PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR}
+
+: A list of authentication method values which, if present in the `amr` claim of the ID token sent by an OpenID Connect provider, cause Paperless-ngx to skip its own two-factor authentication prompt for that login. This avoids users having to enter a second factor twice when the identity provider already performed multi-factor authentication. Logins with a username and password, or via a provider which does not send a matching `amr` value, still require the Paperless-ngx code.
+
+    Which values to trust depends entirely on your identity provider, e.g. Authelia sends `mfa` and Pocket ID sends `phr` for passkey logins (but `otp` for single-factor email codes, which should _not_ be trusted). Some providers, such as Keycloak, do not send an `amr` claim by default and need to be configured to do so. You can see what your provider sends from the Django admin as a superuser by checking `id_token` → `amr` in the "Extra data" of the social account.
+
+    !!! warning
+        Only list values that mean a second factor was actually used for the login.
+
+    Defaults to `mfa`, the standard value for multi-factor authentication ([RFC 8176](https://www.rfc-editor.org/rfc/rfc8176.html)).
+
 #### [`PAPERLESS_SOCIAL_ACCOUNT_DEFAULT_GROUPS=<comma-separated-list>`](#PAPERLESS_SOCIAL_ACCOUNT_DEFAULT_GROUPS) {#PAPERLESS_SOCIAL_ACCOUNT_DEFAULT_GROUPS}
 
 : A list of group names that users who signup via social accounts will be added to upon signup. Groups listed here must already exist.

docs/usage.md

@@ -472,6 +472,8 @@ Users can enable two-factor authentication (2FA) for their accounts from the 'My
 
 Should a user lose access to their 2FA device and all recovery codes, a superuser can disable 2FA for the user from the 'Users & Groups' management screen.
 
+If users log in via an OpenID Connect provider which already enforces multi-factor authentication, see [`PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR`](configuration.md#PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR) to skip the Paperless-ngx prompt for those logins.
+
 ## Workflows
 
 !!! note

src/documents/bulk_edit.py

@@ -37,6 +37,7 @@ from documents.models import OcrTemplate
 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
@@ -454,7 +455,7 @@ def reprocess(doc_ids: list[int], *, remote_ocr: bool = False) -> Literal["OK"]:
 
 def set_permissions(
     doc_ids: list[int],
-    set_permissions: dict,
+    set_permissions: PermissionsSpec,
     *,
     owner: User | None = None,
     merge: bool = False,

src/documents/consumer.py

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

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

@@ -1,8 +1,12 @@
+from collections.abc import Iterable
 from collections.abc import Mapping
 from collections.abc import Sequence
 from typing import Any
 from typing import ClassVar
+from typing import Final
 from typing import Literal
+from typing import TypeAlias
+from typing import TypedDict
 from typing import TypeGuard
 from typing import TypeVar
 from typing import cast
@@ -26,15 +30,38 @@ 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_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):
+    """
+    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.
+    """
+
+    users: Iterable[int]
+    groups: Iterable[int]
+
+
+PermissionsSpec: TypeAlias = Mapping[PermissionAction, PermissionGranteesUpdate]
+
 
 class PaperlessObjectPermissions(DjangoObjectPermissions):
     """
@@ -90,101 +117,10 @@ def has_system_status_permission(user: User | None) -> bool:
     )
 
 
-def get_groups_with_only_permission(obj, codename):
-    ctype = ContentType.objects.get_for_model(obj)
-    permission = Permission.objects.get(content_type=ctype, codename=codename)
-    group_object_perm_group_ids = (
-        GroupObjectPermission.objects.filter(
-            object_pk=obj.pk,
-            content_type=ctype,
-        )
-        .filter(permission=permission)
-        .values_list("group_id")
-    )
-    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]:
     """
-    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.
+    Resolve ``codenames`` to Permission rows, raising for any that don't exist
+    so a bad action name fails loudly instead of granting nothing.
     """
     permission_objs = list(
         Permission.objects.filter(content_type=ctype, codename__in=codenames),
@@ -203,16 +139,13 @@ def _apply_bulk_permission_entry(
     perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
     identity_model: type[User] | type[Group],
     identity_field: str,
-    ids: list[int],
+    ids: Iterable[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),
     )
@@ -250,45 +183,26 @@ def _apply_bulk_permission_entry(
         for pk in object_pks
         for identity_id in add_ids
     ]
-    # 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.
+    # ignore_conflicts skips grants that already exist (unique constraint).
     perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000)
 
 
 def set_permissions_for_objects(
-    permissions: dict,
+    permissions: PermissionsSpec,
     model: type[Model],
     pks: QuerySet | list,
     *,
     merge: bool = False,
 ) -> None:
     """
-    Bulk equivalent of set_permissions_for_object: applies the same
-    permission changes to every object identified by `pks` at once.
+    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.
 
-    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.
+    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.
     """
     object_pks = [str(pk) for pk in pks]
     if not object_pks:  # pragma: no cover
@@ -297,12 +211,9 @@ def set_permissions_for_objects(
     model_name = model.__name__.lower()
     ctype = ContentType.objects.get_for_model(model)
 
-    # 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]] = {}
+    # 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]] = {}
     for action, entry in permissions.items():
         if "users" not in entry and "groups" not in entry:
             continue
@@ -346,6 +257,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)
 
@@ -1200,12 +1188,6 @@ def chunked(iterable, size):
         yield chunk
 
 
-_EMPTY_VIEWER_GRANT: Final[ViewerGrant] = ViewerGrant(
-    viewer_ids=[],
-    viewer_group_ids=[],
-)
-
-
 class _DocumentViewerStream(QuerySetStream["Document"]):
     """Yield document permission data while batch-loading grants.
 
@@ -1231,7 +1213,7 @@ class _DocumentViewerStream(QuerySetStream["Document"]):
         for chunk in chunked(documents, self._chunk_size):
             grants_by_pk = _bulk_get_viewer_permissions([doc.pk for doc in chunk])
             for doc in chunk:
-                yield doc, grants_by_pk.get(doc.pk, _EMPTY_VIEWER_GRANT)
+                yield doc, grants_by_pk[doc.pk]
 
 
 def _bulk_get_viewer_permissions(
@@ -1243,48 +1225,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
@@ -40,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_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,12 +80,17 @@ 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 PermissionGranteesUpdate
+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 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
-from documents.permissions import set_permissions_for_object
+from documents.permissions import set_permissions_for_objects
 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,9 +218,6 @@ class MatchingModelSerializer(serializers.ModelSerializer[Any]):
         return match
 
 
-PERMISSION_ACTIONS = ("view", "change")
-
-
 class SetPermissionsMixin:
     def _validate_user_ids(self, user_ids):
         users = User.objects.none()
@@ -239,8 +239,10 @@ class SetPermissionsMixin:
                 )
         return groups
 
-    def validate_set_permissions(self, set_permissions=None):
-        permissions_dict = {action: {} for action in PERMISSION_ACTIONS}
+    def validate_set_permissions(self, set_permissions=None) -> PermissionsSpec:
+        permissions_dict: dict[PermissionAction, PermissionGranteesUpdate] = {
+            action: {} for action in PERMISSION_ACTIONS
+        }
         if set_permissions is not None:
             for action in PERMISSION_ACTIONS:
                 if action in set_permissions:
@@ -248,18 +250,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, object) -> None:
-        set_permissions_for_object(permissions, object)
+    def _set_permissions(self, permissions: PermissionsSpec, object) -> None:
+        set_permissions_for_objects(permissions, type(object), [object.pk])
 
 
 class SerializerWithPerms(serializers.Serializer[dict[str, Any]]):
@@ -315,35 +317,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_only_permission(
-                    obj,
-                    codename=f"{codename}_{obj.__class__.__name__.lower()}",
-                ).values_list("id", flat=True),
-            )
-
     @extend_schema_field(
         field={
             "type": "object",
@@ -377,17 +350,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:
@@ -403,18 +371,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)
@@ -3104,7 +3072,7 @@ class BulkEditObjectsSerializer(SerializerWithPerms, SetPermissionsMixin):
             )
         return objects
 
-    def _validate_permissions(self, permissions) -> dict:
+    def _validate_permissions(self, permissions) -> PermissionsSpec:
         return self.validate_set_permissions(
             permissions,
         )

src/documents/tests/test_bulk_edit.py

@@ -25,6 +25,7 @@ from documents.models import OcrTemplate
 from documents.models import StoragePath
 from documents.models import Tag
 from documents.pdf_ops import PageSpec
+from documents.permissions import PermissionsSpec
 from documents.permissions import set_permissions_for_objects
 from documents.tasks import extract_zones
 from paperless_testing.dirs import DirectoriesMixin
@@ -448,7 +449,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
 
         grant_object(self.group1, self.doc1, "view_document")
 
-        permissions = {
+        permissions: PermissionsSpec = {
             "view": {
                 "users": [self.user1.id, self.user2.id],
                 "groups": [self.group2.id],
@@ -491,7 +492,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         grant_object(self.user1, self.doc1, "view_document")
         grant_object(self.group1, self.doc1, "view_document")
 
-        permissions = {
+        permissions: PermissionsSpec = {
             "view": {
                 "users": [self.user2.id],
                 "groups": [self.group2.id],
@@ -542,7 +543,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
               batched call rather than one call per (document, identity)
               pair
         """
-        permissions = {
+        permissions: PermissionsSpec = {
             "view": {
                 "users": [self.user1.id, self.user2.id],
                 "groups": [self.group2.id],
@@ -649,7 +650,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
         """
         with self.assertRaises(Permission.DoesNotExist):
             set_permissions_for_objects(
-                {"not_a_real_action": {"users": [self.user1.id], "groups": []}},
+                {"not_a_real_action": {"users": [self.user1.id], "groups": []}},  # type: ignore[bad-assignment]
                 Document,
                 [self.doc1.pk],
             )
@@ -677,7 +678,7 @@ class TestBulkEdit(DirectoriesMixin, TestCase):
             set_permissions_for_objects(
                 {
                     "view": {"users": [self.user1.id], "groups": []},
-                    "not_a_real_action": {"users": [self.user1.id], "groups": []},
+                    "not_a_real_action": {"users": [self.user1.id], "groups": []},  # type: ignore[bad-assignment]
                 },
                 Document,
                 [self.doc1.pk],

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/tests/test_views.py

@@ -292,6 +292,33 @@ class TestViews(DirectoriesMixin, TestCase):
             else:
                 assert False, f"Unexpected tag found: {tag['name']}"
 
+    def test_retrieve_with_full_permissions(self) -> None:
+        """
+        GIVEN:
+            - A tag with view granted to two groups and change granted to only
+              one of them
+        WHEN:
+            - The single tag is requested with full permissions
+        THEN:
+            - Each action lists only the groups granted that action
+        """
+        view_only = Group.objects.create(name="view_only")
+        view_and_change = Group.objects.create(name="view_and_change")
+        tag = Tag.objects.create(name="invoice")
+        grant_object(view_only, tag, "view_tag")
+        grant_object(view_and_change, tag, "view_tag", "change_tag")
+        grant_global(self.user, "view_tag")
+
+        self.client.force_login(self.user)
+        response = self.client.get(f"/api/tags/{tag.pk}/?full_perms=true")
+
+        assert response.status_code == 200
+        permissions = response.json()["permissions"]
+        assert sorted(permissions["view"]["groups"]) == sorted(
+            [view_only.pk, view_and_change.pk],
+        )
+        assert permissions["change"]["groups"] == [view_and_change.pk]
+
     def test_list_no_n_plus_1_queries(self) -> None:
         """
         GIVEN:

src/documents/views.py

@@ -6,7 +6,6 @@ import re
 import subprocess
 import tempfile
 import zipfile
-from collections import defaultdict
 from collections import deque
 from collections.abc import Callable
 from dataclasses import asdict
@@ -19,7 +18,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
@@ -31,7 +29,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
@@ -79,8 +76,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
@@ -180,6 +175,7 @@ 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_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
@@ -465,42 +461,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.
@@ -516,7 +476,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
 
@@ -529,33 +489,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
 

src/documents/workflows/mutations.py

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

src/locale/en_US/LC_MESSAGES/django.po

@@ -2,7 +2,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: paperless-ngx\n"
 "Report-Msgid-Bugs-To: \n"
-"POT-Creation-Date: 2026-10-22 19:29+0000\n"
+"POT-Creation-Date: 2026-10-24 21:11+0000\n"
 "PO-Revision-Date: 2022-02-17 04:17\n"
 "Last-Translator: \n"
 "Language-Team: English\n"
@@ -1652,49 +1652,49 @@ msgstr ""
 msgid "workflow runs"
 msgstr ""
 
-#: documents/serialisers.py:515 documents/serialisers.py:872
-#: documents/serialisers.py:2904 documents/views.py:364 documents/views.py:2787
+#: documents/serialisers.py:483 documents/serialisers.py:840
+#: documents/serialisers.py:2872 documents/views.py:360 documents/views.py:2724
 #: paperless_mail/serialisers.py:156
 msgid "Insufficient permissions."
 msgstr ""
 
-#: documents/serialisers.py:708
+#: documents/serialisers.py:676
 msgid "Invalid color."
 msgstr ""
 
-#: documents/serialisers.py:2371
+#: documents/serialisers.py:2339
 #, python-format
 msgid "File type %(type)s not supported"
 msgstr ""
 
-#: documents/serialisers.py:2415
+#: documents/serialisers.py:2383
 #, python-format
 msgid "Custom field id must be an integer: %(id)s"
 msgstr ""
 
-#: documents/serialisers.py:2422
+#: documents/serialisers.py:2390
 #, python-format
 msgid "Custom field with id %(id)s does not exist"
 msgstr ""
 
-#: documents/serialisers.py:2439 documents/serialisers.py:2449
+#: documents/serialisers.py:2407 documents/serialisers.py:2417
 msgid ""
 "Custom fields must be a list of integers or an object mapping ids to values."
 msgstr ""
 
-#: documents/serialisers.py:2444
+#: documents/serialisers.py:2412
 msgid "Some custom fields don't exist or were specified twice."
 msgstr ""
 
-#: documents/serialisers.py:2591
+#: documents/serialisers.py:2559
 msgid "Invalid variable detected."
 msgstr ""
 
-#: documents/serialisers.py:2960
+#: documents/serialisers.py:2928
 msgid "Duplicate document identifiers are not allowed."
 msgstr ""
 
-#: documents/serialisers.py:2990 documents/views.py:4837
+#: documents/serialisers.py:2958 documents/views.py:4774
 #, python-format
 msgid "Documents not found: %(ids)s"
 msgstr ""
@@ -1945,40 +1945,40 @@ msgstr ""
 msgid ", "
 msgstr ""
 
-#: documents/views.py:319
+#: documents/views.py:315
 msgid "Specify only one of text, title_search, query, or more_like_id."
 msgstr ""
 
-#: documents/views.py:357 documents/views.py:2784
+#: documents/views.py:353 documents/views.py:2721
 msgid "Invalid more_like_id"
 msgstr ""
 
-#: documents/views.py:1722
+#: documents/views.py:1659
 msgid "Invalid AI configuration."
 msgstr ""
 
-#: documents/views.py:1733
+#: documents/views.py:1670
 msgid "AI backend request timed out."
 msgstr ""
 
-#: documents/views.py:1745
+#: documents/views.py:1682
 msgid "AI backend rejected the request. Check logs for details."
 msgstr ""
 
-#: documents/views.py:4853
+#: documents/views.py:4790
 #, python-format
 msgid "Insufficient permissions to share document %(id)s."
 msgstr ""
 
-#: documents/views.py:4900
+#: documents/views.py:4837
 msgid "Bundle is already being processed."
 msgstr ""
 
-#: documents/views.py:4966
+#: documents/views.py:4903
 msgid "The share link bundle is still being prepared. Please try again later."
 msgstr ""
 
-#: documents/views.py:4980
+#: documents/views.py:4917
 msgid "The share link bundle is unavailable."
 msgstr ""
 
@@ -2270,151 +2270,151 @@ msgstr ""
 msgid "paperless application settings"
 msgstr ""
 
-#: paperless/settings/__init__.py:562
+#: paperless/settings/__init__.py:566
 msgid "English (US)"
 msgstr ""
 
-#: paperless/settings/__init__.py:563
+#: paperless/settings/__init__.py:567
 msgid "Arabic"
 msgstr ""
 
-#: paperless/settings/__init__.py:564
+#: paperless/settings/__init__.py:568
 msgid "Afrikaans"
 msgstr ""
 
-#: paperless/settings/__init__.py:565
+#: paperless/settings/__init__.py:569
 msgid "Belarusian"
 msgstr ""
 
-#: paperless/settings/__init__.py:566
+#: paperless/settings/__init__.py:570
 msgid "Bulgarian"
 msgstr ""
 
-#: paperless/settings/__init__.py:567
+#: paperless/settings/__init__.py:571
 msgid "Catalan"
 msgstr ""
 
-#: paperless/settings/__init__.py:568
+#: paperless/settings/__init__.py:572
 msgid "Czech"
 msgstr ""
 
-#: paperless/settings/__init__.py:569
+#: paperless/settings/__init__.py:573
 msgid "Danish"
 msgstr ""
 
-#: paperless/settings/__init__.py:570
+#: paperless/settings/__init__.py:574
 msgid "German"
 msgstr ""
 
-#: paperless/settings/__init__.py:571
+#: paperless/settings/__init__.py:575
 msgid "Greek"
 msgstr ""
 
-#: paperless/settings/__init__.py:572
+#: paperless/settings/__init__.py:576
 msgid "English (GB)"
 msgstr ""
 
-#: paperless/settings/__init__.py:573
+#: paperless/settings/__init__.py:577
 msgid "Spanish"
 msgstr ""
 
-#: paperless/settings/__init__.py:574
+#: paperless/settings/__init__.py:578
 msgid "Persian"
 msgstr ""
 
-#: paperless/settings/__init__.py:575
+#: paperless/settings/__init__.py:579
 msgid "Finnish"
 msgstr ""
 
-#: paperless/settings/__init__.py:576
+#: paperless/settings/__init__.py:580
 msgid "French"
 msgstr ""
 
-#: paperless/settings/__init__.py:577
+#: paperless/settings/__init__.py:581
 msgid "Hungarian"
 msgstr ""
 
-#: paperless/settings/__init__.py:578
+#: paperless/settings/__init__.py:582
 msgid "Indonesian"
 msgstr ""
 
-#: paperless/settings/__init__.py:579
+#: paperless/settings/__init__.py:583
 msgid "Italian"
 msgstr ""
 
-#: paperless/settings/__init__.py:580
+#: paperless/settings/__init__.py:584
 msgid "Japanese"
 msgstr ""
 
-#: paperless/settings/__init__.py:581
+#: paperless/settings/__init__.py:585
 msgid "Korean"
 msgstr ""
 
-#: paperless/settings/__init__.py:582
+#: paperless/settings/__init__.py:586
 msgid "Luxembourgish"
 msgstr ""
 
-#: paperless/settings/__init__.py:583
+#: paperless/settings/__init__.py:587
 msgid "Norwegian"
 msgstr ""
 
-#: paperless/settings/__init__.py:584
+#: paperless/settings/__init__.py:588
 msgid "Dutch"
 msgstr ""
 
-#: paperless/settings/__init__.py:585
+#: paperless/settings/__init__.py:589
 msgid "Polish"
 msgstr ""
 
-#: paperless/settings/__init__.py:586
+#: paperless/settings/__init__.py:590
 msgid "Portuguese (Brazil)"
 msgstr ""
 
-#: paperless/settings/__init__.py:587
+#: paperless/settings/__init__.py:591
 msgid "Portuguese"
 msgstr ""
 
-#: paperless/settings/__init__.py:588
+#: paperless/settings/__init__.py:592
 msgid "Romanian"
 msgstr ""
 
-#: paperless/settings/__init__.py:589
+#: paperless/settings/__init__.py:593
 msgid "Russian"
 msgstr ""
 
-#: paperless/settings/__init__.py:590
+#: paperless/settings/__init__.py:594
 msgid "Slovak"
 msgstr ""
 
-#: paperless/settings/__init__.py:591
+#: paperless/settings/__init__.py:595
 msgid "Slovenian"
 msgstr ""
 
-#: paperless/settings/__init__.py:592
+#: paperless/settings/__init__.py:596
 msgid "Serbian"
 msgstr ""
 
-#: paperless/settings/__init__.py:593
+#: paperless/settings/__init__.py:597
 msgid "Swedish"
 msgstr ""
 
-#: paperless/settings/__init__.py:594
+#: paperless/settings/__init__.py:598
 msgid "Turkish"
 msgstr ""
 
-#: paperless/settings/__init__.py:595
+#: paperless/settings/__init__.py:599
 msgid "Ukrainian"
 msgstr ""
 
-#: paperless/settings/__init__.py:596
+#: paperless/settings/__init__.py:600
 msgid "Vietnamese"
 msgstr ""
 
-#: paperless/settings/__init__.py:597
+#: paperless/settings/__init__.py:601
 msgid "Chinese Simplified"
 msgstr ""
 
-#: paperless/settings/__init__.py:598
+#: paperless/settings/__init__.py:602
 msgid "Chinese Traditional"
 msgstr ""
 

src/paperless/adapter.py

@@ -4,6 +4,7 @@ from urllib.parse import quote
 from allauth.account.adapter import DefaultAccountAdapter
 from allauth.core import context
 from allauth.headless.tokens.strategies.sessions import SessionTokenStrategy
+from allauth.mfa.stages import AuthenticateStage
 from allauth.socialaccount.adapter import DefaultSocialAccountAdapter
 from django.conf import settings
 from django.contrib.auth.models import Group
@@ -19,7 +20,38 @@ from paperless.signals import handle_social_account_updated
 logger = logging.getLogger("paperless.auth")
 
 
+class SocialAccountAuthenticateStage(AuthenticateStage):
+    """
+    Skips the 2FA prompt for a social login if the ID token's `amr` claim shows the
+    identity provider already used one of the SOCIAL_ACCOUNT_MFA_TRUSTED_AMR methods.
+    """
+
+    def handle(self) -> tuple:
+        sociallogin = (self.login.signal_kwargs or {}).get("sociallogin")
+        if sociallogin is not None and settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR:
+            id_token = (sociallogin.account.extra_data or {}).get("id_token")
+            amr = id_token.get("amr") if isinstance(id_token, dict) else None
+            if (
+                isinstance(amr, list)
+                and all(isinstance(value, str) for value in amr)
+                and set(amr) & set(settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR)
+            ):
+                logger.debug(
+                    f"Skipping 2FA for `{self.login.user}`, provider `{sociallogin.account.provider}` reported amr {amr}",
+                )
+                return None, True
+        return super().handle()
+
+
 class CustomAccountAdapter(DefaultAccountAdapter):
+    def get_login_stages(self) -> list[str]:
+        return [
+            "paperless.adapter.SocialAccountAuthenticateStage"
+            if stage == "allauth.mfa.stages.AuthenticateStage"
+            else stage
+            for stage in super().get_login_stages()
+        ]
+
     def is_open_for_signup(self, request):
         """
         Check whether the site is open for signups, which can be

src/paperless/settings/__init__.py

@@ -349,6 +349,10 @@ SOCIAL_ACCOUNT_SYNC_SUPERUSER_GROUP: Final[str | None] = os.getenv(
 SOCIAL_ACCOUNT_SYNC_STAFF_GROUP: Final[str | None] = os.getenv(
     "PAPERLESS_SOCIAL_ACCOUNT_SYNC_STAFF_GROUP",
 )
+SOCIAL_ACCOUNT_MFA_TRUSTED_AMR = get_list_from_env(
+    "PAPERLESS_SOCIAL_ACCOUNT_MFA_TRUSTED_AMR",
+    "mfa",
+)
 
 HEADLESS_TOKEN_STRATEGY = "paperless.adapter.DrfTokenStrategy"
 

src/paperless/tests/test_adapter.py

@@ -2,19 +2,26 @@ import logging
 
 import pytest
 from allauth.account.adapter import get_adapter
+from allauth.account.models import Login
 from allauth.core import context
+from allauth.mfa.totp.internal.auth import TOTP
+from allauth.mfa.totp.internal.auth import generate_totp_secret
 from allauth.socialaccount.adapter import get_adapter as get_social_adapter
+from allauth.socialaccount.models import SocialAccount
+from allauth.socialaccount.models import SocialLogin
 from django.contrib.auth.models import AnonymousUser
 from django.contrib.auth.models import Group
 from django.contrib.auth.models import User
 from django.forms import ValidationError
 from django.http import HttpRequest
+from django.test import RequestFactory
 from django.urls import reverse
 from pytest_django.fixtures import Settings
 from pytest_mock import MockerFixture
 from rest_framework.authtoken.models import Token
 
 from paperless.adapter import DrfTokenStrategy
+from paperless.adapter import SocialAccountAuthenticateStage
 from paperless_testing.factories import UserFactory
 
 
@@ -129,6 +136,120 @@ class TestCustomAccountAdapter:
         assert not user2.is_superuser
 
 
+@pytest.mark.django_db
+class TestSocialAccountAuthenticateStage:
+    @staticmethod
+    def _requires_mfa(user: User, extra_data: dict | None) -> bool:
+        signal_kwargs = None
+        if extra_data is not None:
+            account = SocialAccount(
+                provider="openid_connect",
+                uid="uid",
+                extra_data=extra_data,
+            )
+            signal_kwargs = {"sociallogin": SocialLogin(user=user, account=account)}
+        request = RequestFactory().get("/")
+        request.session = {}
+        stage = SocialAccountAuthenticateStage(
+            None,
+            request,
+            Login(user=user, signal_kwargs=signal_kwargs),
+        )
+        response, cont = stage.handle()
+        assert cont is True
+        requires_mfa = response is not None
+        assert stage.state.get("authentication_required", False) is requires_mfa
+        if requires_mfa:
+            assert response.status_code == 302
+            assert response.url == reverse("mfa_authenticate")
+        return requires_mfa
+
+    def test_stage_replaces_allauth_stage(self) -> None:
+        stages = get_adapter().get_login_stages()
+
+        assert "paperless.adapter.SocialAccountAuthenticateStage" in stages
+        assert "allauth.mfa.stages.AuthenticateStage" not in stages
+
+    @pytest.mark.parametrize(
+        ("trusted_amr", "extra_data", "expected"),
+        [
+            pytest.param(
+                ["mfa"],
+                {"id_token": {"amr": ["pwd", "mfa"]}},
+                False,
+                id="trusted-amr",
+            ),
+            pytest.param(
+                ["phr"],
+                {"id_token": {"amr": ["otp"]}},
+                True,
+                id="untrusted-amr",
+            ),
+            pytest.param([], {"id_token": {"amr": ["mfa"]}}, True, id="setting-unset"),
+            pytest.param(["mfa"], {"id_token": {}}, True, id="no-amr-claim"),
+            pytest.param(
+                ["mfa"],
+                {"userinfo": {"amr": ["mfa"]}},
+                True,
+                id="userinfo-only",
+            ),
+            pytest.param(["m"], {"id_token": {"amr": "mfa"}}, True, id="string-amr"),
+            pytest.param(
+                ["mfa"],
+                {"id_token": {"amr": {"mfa": 1}}},
+                True,
+                id="dict-amr",
+            ),
+            pytest.param(
+                ["mfa"],
+                {"id_token": {"amr": [["mfa"]]}},
+                True,
+                id="nested-amr",
+            ),
+            pytest.param(
+                ["mfa"],
+                {"id_token": {"amr": ["mfa", 1]}},
+                True,
+                id="mixed-amr",
+            ),
+            pytest.param(["mfa"], {"id_token": {"amr": None}}, True, id="null-amr"),
+            pytest.param(["mfa"], {"id_token": {"amr": []}}, True, id="empty-amr"),
+            pytest.param(["mfa"], {"id_token": "mfa"}, True, id="string-id-token"),
+            pytest.param(["mfa"], {"id_token": ["mfa"]}, True, id="list-id-token"),
+            pytest.param(["mfa"], {"id_token": None}, True, id="null-id-token"),
+            pytest.param(["mfa"], None, True, id="regular-login"),
+        ],
+    )
+    def test_mfa_skipped_only_for_trusted_amr(
+        self,
+        settings: Settings,
+        trusted_amr: list[str],
+        extra_data: dict | None,
+        *,
+        expected: bool,
+    ) -> None:
+        """
+        GIVEN:
+            - A user with TOTP enabled
+        WHEN:
+            - The user logs in, via a social account or not
+        THEN:
+            - The 2FA prompt is only skipped if the ID token's amr claim contains
+              one of SOCIAL_ACCOUNT_MFA_TRUSTED_AMR
+        """
+        settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR = trusted_amr
+        user = UserFactory(username="testuser")
+        TOTP.activate(user, generate_totp_secret())
+
+        assert self._requires_mfa(user, extra_data) is expected
+
+    def test_no_mfa_for_user_without_authenticator(self, settings: Settings) -> None:
+        settings.SOCIAL_ACCOUNT_MFA_TRUSTED_AMR = ["mfa"]
+        user = UserFactory(username="testuser")
+
+        assert not self._requires_mfa(user, {"id_token": {"amr": ["pwd"]}})
+
+
 class TestCustomSocialAccountAdapter:
     @pytest.mark.django_db
     def test_is_open_for_signup(self, settings: Settings) -> None: