mirror of
https://github.com/paperless-ngx/paperless-ngx.git
synced 2026-10-09 09:37:12 +00:00
Chore: share the authorizing owner rule between the id sets and the trash filter
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.
This commit is contained in:
1 parent
642511c7b9
commit
3f258db0e2
3 files changed
+40
-27
No files matched your search
@@ -50,6 +50,7 @@ from documents.models import ShareLink
|
||||
from documents.models import ShareLinkBundle
|
||||
from documents.models import StoragePath
|
||||
from documents.models import Tag
|
||||
from documents.permissions import annotate_authorizing_fields
|
||||
from documents.permissions import permitted_document_ids
|
||||
from documents.permissions import permitted_object_ids
|
||||
from documents.versioning import annotate_effective_content
|
||||
@@ -1075,6 +1076,9 @@ class PermittedObjectsFilter(BaseFilterBackend):
|
||||
|
||||
include_granted: bool = True
|
||||
perm_codename: str | None = None
|
||||
# A self-referencing foreign key whose target authorizes a row, so a row is
|
||||
# judged by its parent's owner and grants (``Document.root_document``).
|
||||
parent_field: str | None = None
|
||||
|
||||
def filter_queryset(self, request, queryset, view):
|
||||
# Before the superuser and owner-only paths, neither of which consults
|
||||
@@ -1086,11 +1090,23 @@ class PermittedObjectsFilter(BaseFilterBackend):
|
||||
if request.user.is_superuser:
|
||||
return queryset
|
||||
if not self.include_granted:
|
||||
return queryset.filter(Q(owner=request.user) | Q(owner__isnull=True))
|
||||
owner_field = "owner"
|
||||
if self.parent_field is not None:
|
||||
queryset = annotate_authorizing_fields(queryset, self.parent_field)
|
||||
owner_field = "authorizing_owner"
|
||||
return queryset.filter(
|
||||
Q(**{owner_field: request.user.pk})
|
||||
| Q(**{f"{owner_field}__isnull": True}),
|
||||
)
|
||||
model = queryset.model
|
||||
perm = self.perm_codename or f"view_{model._meta.model_name}"
|
||||
return queryset.filter(
|
||||
id__in=permitted_object_ids(request.user, model, perm),
|
||||
id__in=permitted_object_ids(
|
||||
request.user,
|
||||
model,
|
||||
perm,
|
||||
parent_field=self.parent_field,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -387,14 +387,7 @@ def permitted_object_ids(
|
||||
owner_field, key_field = "owner", "pk"
|
||||
if parent_field is not None:
|
||||
owner_field, key_field = "authorizing_owner", "authorizing_id"
|
||||
base_qs = base_qs.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(),
|
||||
),
|
||||
)
|
||||
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):
|
||||
@@ -446,6 +439,26 @@ def permitted_object_ids(
|
||||
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
|
||||
|
||||
+1
-17
@@ -5602,23 +5602,7 @@ class TrashView(ListModelMixin, PassUserMixin):
|
||||
|
||||
class _TrashPermittedObjectsFilter(PermittedObjectsFilter):
|
||||
include_granted = False
|
||||
|
||||
def filter_queryset(self, request, queryset, view):
|
||||
if request.user.is_superuser or not request.user.is_active:
|
||||
return super().filter_queryset(request, queryset, view)
|
||||
|
||||
# A version belongs to whoever owns its root
|
||||
def owned_or_unowned(prefix: str) -> Q:
|
||||
return Q(**{f"{prefix}owner": request.user}) | Q(
|
||||
**{f"{prefix}owner__isnull": True},
|
||||
)
|
||||
|
||||
return queryset.filter(
|
||||
(Q(root_document__isnull=True) & owned_or_unowned(""))
|
||||
| (
|
||||
Q(root_document__isnull=False) & owned_or_unowned("root_document__")
|
||||
),
|
||||
)
|
||||
parent_field = "root_document"
|
||||
|
||||
filter_backends = (_TrashPermittedObjectsFilter,)
|
||||
pagination_class = StandardPagination
|
||||
|
||||
Reference in new issue
Block a user