diff --git a/src/documents/filters.py b/src/documents/filters.py index 5702a966d..a95e3b288 100644 --- a/src/documents/filters.py +++ b/src/documents/filters.py @@ -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, + ), ) diff --git a/src/documents/permissions.py b/src/documents/permissions.py index 13d9425e9..4480d2584 100644 --- a/src/documents/permissions.py +++ b/src/documents/permissions.py @@ -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 diff --git a/src/documents/views.py b/src/documents/views.py index 7e5ab806f..06b63b663 100644 --- a/src/documents/views.py +++ b/src/documents/views.py @@ -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