mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-13 03:27:13 +00:00
Trenton H
4171ecf166
The trash list needs to show a version to whoever owns its root document, but it deliberately shows only owned and unowned documents and ignores explicit shares, so it cannot use permitted_object_ids. #14384 gave it its own filter_queryset with a hand-built condition through the root, which describes the same rule as the annotations in permitted_object_ids a second time. Move those annotations into annotate_authorizing_fields. PermittedObjectsFilter gains a parent_field attribute that its owner-only and granted paths both use, and the trash filter shrinks to setting include_granted and parent_field.
788 lines
29 KiB
Python
788 lines
29 KiB
Python
from typing import Any
|
|
from typing import TypeVar
|
|
|
|
from django.contrib.auth.models import Group
|
|
from django.contrib.auth.models import Permission
|
|
from django.contrib.auth.models import User
|
|
from django.contrib.contenttypes.models import ContentType
|
|
from django.db.models import Case
|
|
from django.db.models import CharField
|
|
from django.db.models import Count
|
|
from django.db.models import F
|
|
from django.db.models import IntegerField
|
|
from django.db.models import Model
|
|
from django.db.models import Q
|
|
from django.db.models import QuerySet
|
|
from django.db.models import Value
|
|
from django.db.models import When
|
|
from django.db.models.functions import Cast
|
|
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
|
|
|
|
|
|
class PaperlessObjectPermissions(DjangoObjectPermissions):
|
|
"""
|
|
A permissions backend that checks for object-level permissions
|
|
or for ownership.
|
|
"""
|
|
|
|
perms_map = {
|
|
"GET": ["%(app_label)s.view_%(model_name)s"],
|
|
"OPTIONS": ["%(app_label)s.view_%(model_name)s"],
|
|
"HEAD": ["%(app_label)s.view_%(model_name)s"],
|
|
"POST": ["%(app_label)s.add_%(model_name)s"],
|
|
"PUT": ["%(app_label)s.change_%(model_name)s"],
|
|
"PATCH": ["%(app_label)s.change_%(model_name)s"],
|
|
"DELETE": ["%(app_label)s.delete_%(model_name)s"],
|
|
}
|
|
|
|
def has_object_permission(self, request, view, obj):
|
|
if hasattr(obj, "owner") and obj.owner is not None:
|
|
if request.user == obj.owner:
|
|
return True
|
|
else:
|
|
return super().has_object_permission(request, view, obj)
|
|
else:
|
|
return True # no owner
|
|
|
|
|
|
class PaperlessAdminPermissions(BasePermission):
|
|
def has_permission(self, request, view):
|
|
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
|
|
|
|
return 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 (
|
|
getattr(user, "is_superuser", False)
|
|
or getattr(user, "is_staff", False)
|
|
or user.has_perm("paperless.view_system_monitoring")
|
|
)
|
|
|
|
|
|
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.
|
|
"""
|
|
permission_objs = list(
|
|
Permission.objects.filter(content_type=ctype, codename__in=codenames),
|
|
)
|
|
missing = codenames - {p.codename for p in permission_objs}
|
|
if missing:
|
|
raise Permission.DoesNotExist(
|
|
f"Permission matching query does not exist for codename(s): "
|
|
f"{', '.join(sorted(missing))}",
|
|
)
|
|
return permission_objs
|
|
|
|
|
|
def _apply_bulk_permission_entry(
|
|
*,
|
|
perm_model: type[UserObjectPermission] | type[GroupObjectPermission],
|
|
identity_model: type[User] | type[Group],
|
|
identity_field: str,
|
|
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),
|
|
)
|
|
|
|
if not merge:
|
|
existing_ids = set(
|
|
perm_model.objects.filter(
|
|
content_type=ctype,
|
|
object_pk__in=object_pks,
|
|
permission__codename=codename,
|
|
)
|
|
.values_list(f"{identity_field}_id", flat=True)
|
|
.distinct(),
|
|
)
|
|
remove_ids = existing_ids - add_ids
|
|
if remove_ids:
|
|
perm_model.objects.filter(
|
|
content_type=ctype,
|
|
object_pk__in=object_pks,
|
|
permission__codename=codename,
|
|
**{f"{identity_field}_id__in": remove_ids},
|
|
).delete()
|
|
|
|
if not add_ids:
|
|
return
|
|
|
|
rows = [
|
|
perm_model(
|
|
content_type=ctype,
|
|
object_pk=pk,
|
|
permission=permission_obj,
|
|
**{f"{identity_field}_id": identity_id},
|
|
)
|
|
for permission_obj in permission_objs
|
|
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.
|
|
perm_model.objects.bulk_create(rows, ignore_conflicts=True, batch_size=1000)
|
|
|
|
|
|
def set_permissions_for_objects(
|
|
permissions: dict,
|
|
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.
|
|
|
|
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
|
|
return
|
|
|
|
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]] = {}
|
|
for action, entry in permissions.items():
|
|
if "users" not in entry and "groups" not in entry:
|
|
continue
|
|
implied_codenames = {f"{action}_{model_name}"}
|
|
if action == "change":
|
|
# change gives view too
|
|
implied_codenames.add(f"view_{model_name}")
|
|
permissions_by_action[action] = _resolve_permissions(
|
|
implied_codenames,
|
|
ctype,
|
|
)
|
|
|
|
for action, entry in permissions.items():
|
|
codename = f"{action}_{model_name}"
|
|
permission_objs = permissions_by_action.get(action, [])
|
|
|
|
if "users" in entry:
|
|
_apply_bulk_permission_entry(
|
|
perm_model=UserObjectPermission,
|
|
identity_model=User,
|
|
identity_field="user",
|
|
ids=entry["users"],
|
|
codename=codename,
|
|
permission_objs=permission_objs,
|
|
ctype=ctype,
|
|
object_pks=object_pks,
|
|
merge=merge,
|
|
)
|
|
|
|
if "groups" in entry:
|
|
_apply_bulk_permission_entry(
|
|
perm_model=GroupObjectPermission,
|
|
identity_model=Group,
|
|
identity_field="group",
|
|
ids=entry["groups"],
|
|
codename=codename,
|
|
permission_objs=permission_objs,
|
|
ctype=ctype,
|
|
object_pks=object_pks,
|
|
merge=merge,
|
|
)
|
|
|
|
|
|
def permitted_object_ids(
|
|
user: User | None,
|
|
model: type[Model],
|
|
perm: str,
|
|
*,
|
|
include_deleted: bool = False,
|
|
parent_field: str | None = None,
|
|
) -> QuerySet[int]:
|
|
"""
|
|
Generic version of ``permitted_document_ids`` for any model with an
|
|
``owner`` field and guardian object-level permissions. ``include_deleted``
|
|
only has an effect for models exposing a ``global_objects``/``deleted_at``
|
|
soft-delete pattern (currently only ``Document``); for every other model
|
|
it is accepted but has no effect, since those models have no soft-delete
|
|
concept.
|
|
|
|
``parent_field`` names a self-referencing foreign key whose target
|
|
authorizes the row (``Document.root_document``). A row with a parent is
|
|
visible exactly when its parent is, judged by the parent's owner and
|
|
grants, so the row's own owner and grants are ignored.
|
|
|
|
Guardian stores ``object_pk`` as a string, so the row key is cast to a
|
|
string and tested against the user's and groups' grants with a single
|
|
uncorrelated ``IN``. Postgres and SQLite build that set once. MariaDB
|
|
evaluates it as an index probe per row, which is cheap because the
|
|
lookups use guardian's unique indexes. Casting every ``object_pk`` to an
|
|
integer instead cannot use an index, and MariaDB cannot materialize it
|
|
inside the owner ``OR``, so it re-scans the user's grants for every row.
|
|
A correlated ``EXISTS`` per grant fixes MariaDB too, but Postgres and
|
|
SQLite re-run it for every row and end up slower than the original. The
|
|
user's groups are matched with an ``IN`` subquery rather than a join
|
|
through the membership table, which SQLite plans badly once the grant
|
|
tables grow.
|
|
"""
|
|
has_soft_delete = hasattr(model, "global_objects")
|
|
manager = (
|
|
model.global_objects if include_deleted and has_soft_delete else model.objects
|
|
)
|
|
base_qs = manager.all().only("id", "owner")
|
|
|
|
owner_field, key_field = "owner", "pk"
|
|
if parent_field is not None:
|
|
owner_field, key_field = "authorizing_owner", "authorizing_id"
|
|
base_qs = annotate_authorizing_fields(base_qs, parent_field)
|
|
unowned = Q(**{f"{owner_field}__isnull": True})
|
|
|
|
if user is None or not getattr(user, "is_authenticated", False):
|
|
return base_qs.filter(unowned).values_list("id", flat=True)
|
|
|
|
# Deactivated users get nothing, deactivated superusers included, so this
|
|
# has to come before the superuser shortcut. guardian's
|
|
# ObjectPermissionChecker denies inactive users, but get_objects_for_user
|
|
# (the pattern this replaces) does not, so it would not be inherited.
|
|
if not getattr(user, "is_active", False):
|
|
return base_qs.none().values_list("id", flat=True)
|
|
|
|
if getattr(user, "is_superuser", False):
|
|
return base_qs.values_list("id", flat=True)
|
|
|
|
# Guardian's UserObjectPermission/GroupObjectPermission always store a bare
|
|
# codename, but has_perm()-style callers commonly pass the qualified
|
|
# "app_label.codename" form. content_type already disambiguates the
|
|
# codename, so just drop any prefix rather than silently under-permitting.
|
|
perm = perm.rsplit(".", 1)[-1]
|
|
|
|
content_type = ContentType.objects.get_for_model(model)
|
|
perm_filter = {
|
|
"permission__codename": perm,
|
|
"permission__content_type": content_type,
|
|
}
|
|
|
|
# Both grant sets are compared to the row key as strings, exactly as
|
|
# guardian stores them, and are uncorrelated, so each engine can build the
|
|
# set once instead of probing per row.
|
|
user_keys = UserObjectPermission.objects.filter(
|
|
user=user,
|
|
**perm_filter,
|
|
).values_list("object_pk", flat=True)
|
|
group_keys = GroupObjectPermission.objects.filter(
|
|
group_id__in=user.groups.values("id"),
|
|
**perm_filter,
|
|
).values_list("object_pk", flat=True)
|
|
permitted_keys = user_keys.union(group_keys, all=True)
|
|
|
|
return (
|
|
base_qs.annotate(permitted_key=Cast(key_field, CharField(max_length=64)))
|
|
.filter(
|
|
Q(**{owner_field: user.pk}) | unowned | Q(permitted_key__in=permitted_keys),
|
|
)
|
|
.values_list("id", flat=True)
|
|
)
|
|
|
|
|
|
ModelT = TypeVar("ModelT", bound=Model)
|
|
|
|
|
|
def annotate_authorizing_fields(
|
|
queryset: QuerySet[ModelT],
|
|
parent_field: str,
|
|
) -> QuerySet[ModelT]:
|
|
"""
|
|
Annotate each row with ``authorizing_id`` and ``authorizing_owner``: the id
|
|
and owner of the row that authorizes it. A row with a parent (the
|
|
self-referencing foreign key ``parent_field``) is authorized by its parent,
|
|
any other row by itself.
|
|
"""
|
|
return queryset.annotate(
|
|
authorizing_id=Coalesce(f"{parent_field}_id", "id"),
|
|
authorizing_owner=Case(
|
|
When(**{f"{parent_field}_id__isnull": True}, then=F("owner_id")),
|
|
default=F(f"{parent_field}__owner_id"),
|
|
output_field=IntegerField(),
|
|
),
|
|
)
|
|
|
|
|
|
def user_is_unrestricted(user: User | None) -> bool:
|
|
"""
|
|
True when ``user`` means "no restriction at all" (an absent user, or an
|
|
*active* superuser) without needing a database check to know it.
|
|
|
|
``permitted_object_ids(None, ...)`` itself means the much narrower "only
|
|
unowned rows", which is NOT the same thing as "no user filtering
|
|
requested", so callers must special-case this before ever calling it.
|
|
A deactivated superuser is deliberately NOT unrestricted here, matching
|
|
permitted_object_ids's own is_active-before-is_superuser ordering.
|
|
|
|
Callers that can avoid a database round trip entirely when this is true
|
|
(e.g. checking a single already-loaded object's visibility rather than
|
|
filtering a queryset) should do so via this function directly, rather
|
|
than through restrict_queryset_to_visible() below.
|
|
"""
|
|
if user is None:
|
|
return True
|
|
return (
|
|
getattr(user, "is_authenticated", False)
|
|
and getattr(user, "is_active", False)
|
|
and getattr(user, "is_superuser", False)
|
|
)
|
|
|
|
|
|
def restrict_queryset_to_visible(
|
|
queryset: QuerySet[ModelT],
|
|
user: User | None,
|
|
perm: str,
|
|
) -> QuerySet[ModelT]:
|
|
"""
|
|
Restrict ``queryset`` to the rows ``user`` may see with ``perm``.
|
|
|
|
Delegates the visibility check to the database as a
|
|
``WHERE id IN (subquery)`` rather than materializing the full
|
|
permitted-id set into a Python collection first: a caller that only
|
|
needs to check a small handful of rows (a resolved-id list, a few
|
|
RAG-neighbour candidate ids) never pays for scanning or holding the
|
|
installation's entire taxonomy in memory to do it.
|
|
|
|
Returns ``queryset`` unchanged for user_is_unrestricted(user); every
|
|
other case is delegated to ``permitted_object_ids`` rather than
|
|
re-deciding the ordering here.
|
|
"""
|
|
if user_is_unrestricted(user):
|
|
return queryset
|
|
return queryset.filter(pk__in=permitted_object_ids(user, queryset.model, perm))
|
|
|
|
|
|
def permitted_document_ids(
|
|
user: User | None,
|
|
*,
|
|
perm: str = "view_document",
|
|
include_deleted: bool = False,
|
|
) -> QuerySet[int]:
|
|
"""
|
|
Document-specific convenience wrapper around ``permitted_object_ids``.
|
|
Return a queryset of document IDs the user has ``perm`` on (default
|
|
``"view_document"``). By default limited to non-deleted documents; pass
|
|
``include_deleted=True`` for callers that need to check permission on
|
|
soft-deleted documents (e.g. trash restore). This intentionally avoids
|
|
``get_objects_for_user`` to keep the subquery small and index-friendly.
|
|
|
|
A version is authorized by its root document, so a version's own owner and
|
|
grants never matter.
|
|
"""
|
|
return permitted_object_ids(
|
|
user,
|
|
Document,
|
|
perm,
|
|
include_deleted=include_deleted,
|
|
parent_field="root_document",
|
|
)
|
|
|
|
|
|
def get_document_count_filter_for_user(user, related_name: str = "documents"):
|
|
"""
|
|
Return the Q object used to filter document counts for the given user.
|
|
|
|
The filter is expressed as an ``id__in`` against a small subquery of permitted
|
|
document IDs to keep the generated SQL simple and avoid large OR clauses.
|
|
|
|
``related_name`` is the ORM path from the annotated model to Document (e.g.
|
|
``"documents"`` for Tag's direct M2M, or ``"fields__document"`` for CustomField,
|
|
which only reaches Document via the CustomFieldInstance through-model).
|
|
"""
|
|
|
|
if getattr(user, "is_superuser", False):
|
|
# Superuser: no permission filtering needed
|
|
return Q(**{f"{related_name}__deleted_at__isnull": True})
|
|
|
|
permitted_ids = permitted_document_ids(user)
|
|
return Q(**{f"{related_name}__id__in": permitted_ids})
|
|
|
|
|
|
def annotate_document_count_by_ids(
|
|
queryset: QuerySet[Any],
|
|
through_model: Any,
|
|
related_object_field: str,
|
|
document_ids: Any,
|
|
target_field: str = "document_id",
|
|
) -> QuerySet[Any]:
|
|
"""
|
|
Annotate a queryset with a document count for a relation to Document that
|
|
goes through an M2M/through-model table (e.g. Tag via
|
|
``Document.tags.through``, or CustomField via ``CustomFieldInstance``),
|
|
for an explicit, already-resolved set of document ids.
|
|
|
|
Counts are computed via a single, independent GROUP BY over the relation
|
|
table -- with the id filter expressed as a plain ``WHERE`` rather than an
|
|
aggregate ``FILTER`` -- then injected via ``Case``/``When``. This
|
|
deliberately avoids two slower alternatives found while building this:
|
|
|
|
- A per-outer-row correlated subquery (one execution per row of the
|
|
annotated queryset): fine at a handful of rows, catastrophic once the
|
|
queryset has hundreds/thousands of rows.
|
|
- ``Count(..., filter=Q(id__in=document_ids), distinct=True)`` applied
|
|
directly to the M2M relation: Postgres can fail to plan the ``id__in``
|
|
check as a semi-join and instead re-checks subquery membership once per
|
|
row of the (much larger) M2M join -- worse than the correlated subquery.
|
|
|
|
Aggregation is restricted to rows whose ``related_object_field`` is one of
|
|
``queryset``'s pks, so passing a subset (e.g. a handful of tag descendants)
|
|
doesn't pay the cost of counting for every row matching ``document_ids``.
|
|
|
|
Args:
|
|
queryset: base queryset to annotate (must contain pk)
|
|
through_model: model representing the relation (e.g., Document.tags.through
|
|
or CustomFieldInstance)
|
|
related_object_field: field on the relation pointing back to queryset pk
|
|
document_ids: the document ids to count against -- a concrete list/set,
|
|
or a simple (already resolved) queryset of ids. Callers
|
|
that need this filtered by a complex condition (e.g. a
|
|
permission check) should resolve it to a concrete list
|
|
first if the same ids will be reused across multiple
|
|
calls, rather than passing the complex queryset itself
|
|
into each -- see ``_get_selection_data_for_queryset``.
|
|
target_field: field on the relation pointing to Document id
|
|
"""
|
|
|
|
counts = (
|
|
through_model.objects.filter(
|
|
**{
|
|
f"{related_object_field}__in": queryset.values("pk"),
|
|
f"{target_field}__in": document_ids,
|
|
},
|
|
)
|
|
.values(related_object_field)
|
|
.annotate(c=Count(target_field, distinct=True))
|
|
)
|
|
counts_by_pk = {row[related_object_field]: row["c"] for row in counts}
|
|
|
|
if not counts_by_pk:
|
|
return queryset.annotate(
|
|
document_count=Value(0, output_field=IntegerField()),
|
|
)
|
|
|
|
return queryset.annotate(
|
|
document_count=Case(
|
|
*(When(pk=pk, then=Value(count)) for pk, count in counts_by_pk.items()),
|
|
default=Value(0),
|
|
output_field=IntegerField(),
|
|
),
|
|
)
|
|
|
|
|
|
def annotate_document_count_for_related_queryset(
|
|
queryset: QuerySet[Any],
|
|
through_model: Any,
|
|
related_object_field: str,
|
|
target_field: str = "document_id",
|
|
user: User | None = None,
|
|
) -> QuerySet[Any]:
|
|
"""
|
|
Same as ``annotate_document_count_by_ids``, but resolves the document ids
|
|
from the given user's view permissions rather than taking them directly.
|
|
"""
|
|
|
|
return annotate_document_count_by_ids(
|
|
queryset,
|
|
through_model=through_model,
|
|
related_object_field=related_object_field,
|
|
document_ids=permitted_document_ids(user),
|
|
target_field=target_field,
|
|
)
|
|
|
|
|
|
def get_objects_for_user_owner_aware(
|
|
user: User | None,
|
|
perms: str | list[str],
|
|
Model: Any,
|
|
*,
|
|
include_deleted: bool = False,
|
|
) -> QuerySet[Any]:
|
|
"""
|
|
Returns objects the user owns, are unowned, or has explicit perms.
|
|
When include_deleted is True, soft-deleted items are also included.
|
|
|
|
Legacy slow path (guardian-backed, O(n) style permission resolution).
|
|
Most queryset-filtering call sites have migrated onto
|
|
``PermittedObjectsFilter``/``permitted_object_ids()``, but this function
|
|
is kept because production callers still remain. Several callers remain
|
|
across ``documents/``, ``paperless_mail/``, and ``paperless_ai/`` --
|
|
grep for this function name before removing it.
|
|
"""
|
|
manager = (
|
|
Model.global_objects
|
|
if include_deleted and hasattr(Model, "global_objects")
|
|
else Model.objects
|
|
)
|
|
|
|
objects_owned = manager.filter(owner=user)
|
|
objects_unowned = manager.filter(owner__isnull=True)
|
|
objects_with_perms = get_objects_for_user(
|
|
user=user,
|
|
perms=perms,
|
|
klass=manager.all(),
|
|
accept_global_perms=False,
|
|
)
|
|
return objects_owned | objects_unowned | objects_with_perms
|
|
|
|
|
|
def has_perms_owner_aware(user, perms, obj):
|
|
"""
|
|
Legacy slow path (guardian-backed) single-object permission check.
|
|
|
|
The queryset-filtering side of this migrated onto
|
|
``PermittedObjectsFilter``/``permitted_object_ids()``, but this
|
|
single-object check still has many production callers. Several callers
|
|
remain across ``documents/``, ``paperless_mail/``, and ``paperless_ai/``
|
|
-- grep for this function name before removing it.
|
|
|
|
A document version is authorized by its root document, like in
|
|
``permitted_document_ids``, so a version's own owner and grants never
|
|
matter. Fetch the root with ``select_related("root_document__owner")`` to
|
|
avoid extra queries.
|
|
"""
|
|
if isinstance(obj, Document):
|
|
obj = get_root_document(obj)
|
|
checker = ObjectPermissionChecker(user)
|
|
return obj.owner is None or obj.owner == user or checker.has_perm(perms, obj)
|
|
|
|
|
|
class ViewDocumentsPermissions(BasePermission):
|
|
"""
|
|
Permissions class that checks for model permissions for only viewing Documents.
|
|
"""
|
|
|
|
perms_map = {
|
|
"OPTIONS": ["documents.view_document"],
|
|
"GET": ["documents.view_document"],
|
|
"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):
|
|
"""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):
|
|
"""
|
|
Permissions class that checks for model permissions for Notes.
|
|
"""
|
|
|
|
perms_map = {
|
|
"OPTIONS": ["documents.view_note", "documents.view_document"],
|
|
"GET": ["documents.view_note", "documents.view_document"],
|
|
"POST": [
|
|
"documents.add_note",
|
|
"documents.view_document",
|
|
"documents.change_document",
|
|
],
|
|
"DELETE": [
|
|
"documents.delete_note",
|
|
"documents.view_document",
|
|
"documents.change_document",
|
|
],
|
|
}
|
|
|
|
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):
|
|
"""
|
|
Permissions class that checks for model permissions for acknowledging tasks.
|
|
"""
|
|
|
|
perms_map = {
|
|
"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)
|