diff --git a/src/documents/filters.py b/src/documents/filters.py index b511d56de..e4f5f7b01 100644 --- a/src/documents/filters.py +++ b/src/documents/filters.py @@ -39,7 +39,6 @@ from guardian.utils import get_user_obj_perms_model from rest_framework import serializers from rest_framework.filters import BaseFilterBackend from rest_framework.filters import OrderingFilter -from rest_framework_guardian.filters import ObjectPermissionsFilter from documents.models import Correspondent from documents.models import CustomField @@ -51,7 +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 permitted_document_ids +from documents.permissions import permitted_object_ids if TYPE_CHECKING: from collections.abc import Callable @@ -1028,59 +1027,56 @@ class PaperlessTaskFilterSet(FilterSet): return queryset.exclude(status__in=PaperlessTask.COMPLETE_STATUSES) -class ObjectOwnedOrGrantedPermissionsFilter(ObjectPermissionsFilter): +class PermittedObjectsFilter(BaseFilterBackend): """ - A filter backend that limits results to those where the requesting user - has read object level permissions, owns the objects, or objects without - an owner (for backwards compat) + Filters a queryset down to objects the requesting user owns, are + unowned, or (when ``include_granted`` is True) has an explicit + user/group guardian permission on. Backed by ``permitted_object_ids`` + -- a single ``id__in`` subquery, not a join -- so it can't produce + duplicate rows even when the base queryset already carries independent + joins (e.g. multi-value ``tags__id__all`` filtering), and stays + index-friendly at scale instead of falling back to guardian's + varchar-cast join. + + Set ``include_granted = False`` on a subclass for endpoints that + intentionally only show owned/unowned objects regardless of explicit + shares (e.g. ``TrashView``). """ + include_granted: bool = True + perm_codename: str | None = None + def filter_queryset(self, request, queryset, view): if request.user.is_superuser: return queryset - objects_with_perms = super().filter_queryset(request, queryset, view) - objects_owned = queryset.filter(owner=request.user) - objects_unowned = queryset.filter(owner__isnull=True) - return objects_with_perms | objects_owned | objects_unowned + model = queryset.model + if not self.include_granted: + return queryset.filter(Q(owner=request.user) | Q(owner__isnull=True)) + perm = self.perm_codename or f"view_{model._meta.model_name}" + return queryset.filter( + id__in=permitted_object_ids(request.user, model, perm), + ) -class DocumentPermissionsFilter(BaseFilterBackend): +# Deprecated aliases for backwards compatibility +class ObjectOwnedOrGrantedPermissionsFilter(PermittedObjectsFilter): """ - A filter backend limiting Document results to those the requesting user - owns, are unowned, or has explicit (user- or group-level) view - permission on. - - Unlike ``ObjectOwnedOrGrantedPermissionsFilter``, this does not build an - ``objects_with_perms | objects_owned | objects_unowned`` union of - querysets derived from the same base queryset. When that base queryset - already carries independent joins on a multi-valued relation (e.g. two - separate joins from ``tags__id__all`` filtering on two tags), each - OR-ed branch can end up pairing those joins' aliases differently, - letting more than one row out of the join's cross product satisfy the - combined WHERE -- returning the same document more than once. Filtering - via a single ``id__in`` against ``permitted_document_ids`` (a plain - subquery, not a join) sidesteps that entirely and is also cheaper than - guardian's join-based permission check. + Deprecated: Use PermittedObjectsFilter instead. """ - def filter_queryset(self, request, queryset, view): - if request.user.is_superuser: - return queryset - return queryset.filter(id__in=permitted_document_ids(request.user)) - -class ObjectOwnedPermissionsFilter(ObjectPermissionsFilter): +class DocumentPermissionsFilter(PermittedObjectsFilter): """ - A filter backend that limits results to those where the requesting user - owns the objects or objects without an owner (for backwards compat) + Deprecated: Use PermittedObjectsFilter instead. """ - def filter_queryset(self, request, queryset, view): - if request.user.is_superuser: - return queryset - objects_owned = queryset.filter(owner=request.user) - objects_unowned = queryset.filter(owner__isnull=True) - return objects_owned | objects_unowned + +class ObjectOwnedPermissionsFilter(PermittedObjectsFilter): + """ + Deprecated: Use PermittedObjectsFilter instead. + """ + + include_granted = False class DocumentsOrderingFilter(OrderingFilter): diff --git a/src/documents/tests/test_permitted_objects_filter.py b/src/documents/tests/test_permitted_objects_filter.py new file mode 100644 index 000000000..d3df22805 --- /dev/null +++ b/src/documents/tests/test_permitted_objects_filter.py @@ -0,0 +1,70 @@ +import pytest +from django.contrib.auth.models import User +from guardian.shortcuts import assign_perm +from rest_framework.test import APIRequestFactory + +from documents.filters import PermittedObjectsFilter +from documents.models import Tag +from documents.tests.factories import TagFactory + + +class _DummyView: + queryset = Tag.objects.all() + + +@pytest.mark.django_db +class TestPermittedObjectsFilter: + def test_superuser_bypasses_filtering_entirely(self): + superuser = User.objects.create_superuser(username="root") + owner = User.objects.create_user(username="owner") + TagFactory(owner=owner) + request = APIRequestFactory().get("/") + request.user = superuser + + result = PermittedObjectsFilter().filter_queryset( + request, + Tag.objects.all(), + _DummyView(), + ) + assert result.count() == Tag.objects.count() + + def test_non_superuser_sees_only_owned_unowned_and_granted(self): + owner = User.objects.create_user(username="owner") + grantee = User.objects.create_user(username="grantee") + owned = TagFactory(owner=grantee) + unowned = TagFactory(owner=None) + granted = TagFactory(owner=owner) + hidden = TagFactory(owner=owner) + assign_perm("view_tag", grantee, granted) + request = APIRequestFactory().get("/") + request.user = grantee + + result = PermittedObjectsFilter().filter_queryset( + request, + Tag.objects.all(), + _DummyView(), + ) + visible_ids = set(result.values_list("id", flat=True)) + assert visible_ids == {owned.pk, unowned.pk, granted.pk} + assert hidden.pk not in visible_ids + + def test_include_granted_false_excludes_explicitly_shared_objects(self): + owner = User.objects.create_user(username="owner2") + grantee = User.objects.create_user(username="grantee2") + owned = TagFactory(owner=grantee) + granted = TagFactory(owner=owner) + assign_perm("view_tag", grantee, granted) + request = APIRequestFactory().get("/") + request.user = grantee + + class _OwnerOnlyFilter(PermittedObjectsFilter): + include_granted = False + + result = _OwnerOnlyFilter().filter_queryset( + request, + Tag.objects.all(), + _DummyView(), + ) + visible_ids = set(result.values_list("id", flat=True)) + assert visible_ids == {owned.pk} + assert granted.pk not in visible_ids