test: add matching.py permission coverage for correspondents, document types, storage paths

Completes the parametrized coverage started for tags -- proves all 4
matching.py lookups migrated to permitted_object_ids respect
per-object view permissions, not just the tag case.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

src/documents/matching.py

@@ -19,7 +19,7 @@ from documents.models import StoragePath
 from documents.models import Tag
 from documents.models import Workflow
 from documents.models import WorkflowTrigger
-from documents.permissions import get_objects_for_user_owner_aware
+from documents.permissions import permitted_object_ids
 from documents.regex import safe_regex_search
 
 if TYPE_CHECKING:
@@ -55,10 +55,8 @@ def match_correspondents(document: Document, classifier: DocumentClassifier, use
         user = document.owner
 
     if user is not None:
-        correspondents = get_objects_for_user_owner_aware(
-            user,
-            "documents.view_correspondent",
-            Correspondent,
+        correspondents = Correspondent.objects.filter(
+            id__in=permitted_object_ids(user, Correspondent, "view_correspondent"),
         )
     else:
         correspondents = Correspondent.objects.all()
@@ -86,10 +84,8 @@ def match_document_types(document: Document, classifier: DocumentClassifier, use
         user = document.owner
 
     if user is not None:
-        document_types = get_objects_for_user_owner_aware(
-            user,
-            "documents.view_documenttype",
-            DocumentType,
+        document_types = DocumentType.objects.filter(
+            id__in=permitted_object_ids(user, DocumentType, "view_documenttype"),
         )
     else:
         document_types = DocumentType.objects.all()
@@ -116,7 +112,9 @@ def match_tags(document: Document, classifier: DocumentClassifier, user=None):
         user = document.owner
 
     if user is not None:
-        tags = get_objects_for_user_owner_aware(user, "documents.view_tag", Tag)
+        tags = Tag.objects.filter(
+            id__in=permitted_object_ids(user, Tag, "view_tag"),
+        )
     else:
         tags = Tag.objects.all()
 
@@ -145,10 +143,8 @@ def match_storage_paths(document: Document, classifier: DocumentClassifier, user
         user = document.owner
 
     if user is not None:
-        storage_paths = get_objects_for_user_owner_aware(
-            user,
-            "documents.view_storagepath",
-            StoragePath,
+        storage_paths = StoragePath.objects.filter(
+            id__in=permitted_object_ids(user, StoragePath, "view_storagepath"),
         )
     else:
         storage_paths = StoragePath.objects.all()

src/documents/permissions.py

@@ -7,6 +7,7 @@ from django.contrib.contenttypes.models import ContentType
 from django.db.models import Case
 from django.db.models import Count
 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
@@ -163,30 +164,32 @@ def set_permissions_for_object(
                         )
 
 
-def permitted_document_ids(
-    user,
+def permitted_object_ids(
+    user: User | None,
+    model: type[Model],
+    perm: str,
     *,
-    perm: str = "view_document",
     include_deleted: bool = False,
-):
+) -> QuerySet[int]:
     """
-    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.
+    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.
     """
-
-    manager = Document.global_objects if include_deleted else Document.objects
-    base_docs = manager.all()
-    base_docs = base_docs.only("id", "owner")
+    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")
 
     if user is None or not getattr(user, "is_authenticated", False):
-        # Just Anonymous user e.g. for drf-spectacular
-        return base_docs.filter(owner__isnull=True).values_list("id", flat=True)
+        return base_qs.filter(owner__isnull=True).values_list("id", flat=True)
 
     if getattr(user, "is_superuser", False):
-        return base_docs.values_list("id", flat=True)
+        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
@@ -194,31 +197,46 @@ def permitted_document_ids(
     # codename, so just drop any prefix rather than silently under-permitting.
     perm = perm.rsplit(".", 1)[-1]
 
-    document_ct = ContentType.objects.get_for_model(Document)
+    content_type = ContentType.objects.get_for_model(model)
     perm_filter = {
         "permission__codename": perm,
-        "permission__content_type": document_ct,
+        "permission__content_type": content_type,
     }
 
-    user_perm_docs = (
+    user_perm_ids = (
         UserObjectPermission.objects.filter(user=user, **perm_filter)
         .annotate(object_pk_int=Cast("object_pk", IntegerField()))
         .values_list("object_pk_int", flat=True)
     )
-
-    group_perm_docs = (
+    group_perm_ids = (
         GroupObjectPermission.objects.filter(group__user=user, **perm_filter)
         .annotate(object_pk_int=Cast("object_pk", IntegerField()))
         .values_list("object_pk_int", flat=True)
     )
+    permitted_ids = user_perm_ids.union(group_perm_ids)
 
-    permitted_documents = user_perm_docs.union(group_perm_docs)
-
-    return base_docs.filter(
-        Q(owner=user) | Q(owner__isnull=True) | Q(id__in=permitted_documents),
+    return base_qs.filter(
+        Q(owner=user) | Q(owner__isnull=True) | Q(id__in=permitted_ids),
     ).values_list("id", flat=True)
 
 
+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.
+    """
+    return permitted_object_ids(user, Document, perm, include_deleted=include_deleted)
+
+
 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.

src/documents/tests/test_permission_filtering_security.py

@@ -12,9 +12,22 @@ from django.test import override_settings
 from guardian.shortcuts import assign_perm
 from rest_framework.test import APIClient
 
+from documents.matching import match_correspondents
+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 DocumentType
+from documents.models import StoragePath
+from documents.models import Tag
 from documents.permissions import permitted_document_ids
+from documents.permissions import permitted_object_ids
 from documents.serialisers import _get_viewable_duplicates
+from documents.tests.factories import CorrespondentFactory
 from documents.tests.factories import DocumentFactory
+from documents.tests.factories import DocumentTypeFactory
+from documents.tests.factories import StoragePathFactory
+from documents.tests.factories import TagFactory
 
 
 def assert_visible_document_ids(actual_ids, *, expected_visible, expected_hidden):
@@ -431,3 +444,179 @@ class TestTrashRestorePermissionBoundary:
             format="json",
         )
         assert response.status_code == HTTPStatus.OK
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    ("model", "factory", "perm"),
+    [
+        (Tag, TagFactory, "view_tag"),
+        (Correspondent, CorrespondentFactory, "view_correspondent"),
+        (DocumentType, DocumentTypeFactory, "view_documenttype"),
+        (StoragePath, StoragePathFactory, "view_storagepath"),
+    ],
+)
+class TestPermittedObjectIdsGenericModels:
+    def test_owner_sees_own_object(self, model, factory, perm):
+        owner = User.objects.create_user(username=f"owner_{model.__name__}")
+        stranger = User.objects.create_user(username=f"stranger_{model.__name__}")
+        owned = factory(owner=owner)
+        strangers = factory(owner=stranger)
+
+        assert_visible_document_ids(
+            permitted_object_ids(owner, model, perm),
+            expected_visible=[owned.pk],
+            expected_hidden=[strangers.pk],
+        )
+
+    def test_unowned_object_visible_to_everyone(self, model, factory, perm):
+        user = User.objects.create_user(username=f"user_{model.__name__}")
+        unowned = factory(owner=None)
+
+        assert_visible_document_ids(
+            permitted_object_ids(user, model, perm),
+            expected_visible=[unowned.pk],
+            expected_hidden=[],
+        )
+
+    def test_explicit_permission_grants_visibility(self, model, factory, perm):
+        owner = User.objects.create_user(username=f"owner2_{model.__name__}")
+        grantee = User.objects.create_user(username=f"grantee_{model.__name__}")
+        stranger = User.objects.create_user(username=f"stranger2_{model.__name__}")
+        shared = factory(owner=owner)
+        not_shared = factory(owner=owner)
+        assign_perm(perm, grantee, shared)
+
+        assert_visible_document_ids(
+            permitted_object_ids(grantee, model, perm),
+            expected_visible=[shared.pk],
+            expected_hidden=[not_shared.pk],
+        )
+        assert_visible_document_ids(
+            permitted_object_ids(stranger, model, perm),
+            expected_visible=[],
+            expected_hidden=[shared.pk, not_shared.pk],
+        )
+
+    def test_group_permission_grants_visibility_to_members_only(
+        self,
+        model,
+        factory,
+        perm,
+    ):
+        owner = User.objects.create_user(username=f"owner3_{model.__name__}")
+        member = User.objects.create_user(username=f"member_{model.__name__}")
+        non_member = User.objects.create_user(username=f"nonmember_{model.__name__}")
+        group = Group.objects.create(name=f"group_{model.__name__}")
+        member.groups.add(group)
+        shared = factory(owner=owner)
+        assign_perm(perm, group, shared)
+
+        assert_visible_document_ids(
+            permitted_object_ids(member, model, perm),
+            expected_visible=[shared.pk],
+            expected_hidden=[],
+        )
+        assert_visible_document_ids(
+            permitted_object_ids(non_member, model, perm),
+            expected_visible=[],
+            expected_hidden=[shared.pk],
+        )
+
+    def test_superuser_sees_everything(self, model, factory, perm):
+        superuser = User.objects.create_superuser(username=f"root_{model.__name__}")
+        owner = User.objects.create_user(username=f"owner4_{model.__name__}")
+        obj = factory(owner=owner)
+
+        assert_visible_document_ids(
+            permitted_object_ids(superuser, model, perm),
+            expected_visible=[obj.pk],
+            expected_hidden=[],
+        )
+
+
+@pytest.mark.django_db
+class TestMatchingRespectsObjectPermissions:
+    def test_match_tags_only_considers_tags_visible_to_user(self):
+        owner = User.objects.create_user(username="tag_owner")
+        classifying_user = User.objects.create_user(username="classifier_user")
+        visible_tag = TagFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=Tag.MATCH_LITERAL,
+        )
+        hidden_tag = TagFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=Tag.MATCH_LITERAL,
+        )
+        assign_perm("view_tag", classifying_user, visible_tag)
+        doc = DocumentFactory(owner=classifying_user, content="an invoice document")
+
+        matched = match_tags(doc, classifier=None, user=classifying_user)
+        matched_ids = {t.pk for t in matched}
+        assert visible_tag.pk in matched_ids
+        assert hidden_tag.pk not in matched_ids
+
+    def test_match_correspondents_only_considers_correspondents_visible_to_user(self):
+        owner = User.objects.create_user(username="correspondent_owner")
+        classifying_user = User.objects.create_user(username="classifier_user2")
+        visible_correspondent = CorrespondentFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=Correspondent.MATCH_LITERAL,
+        )
+        hidden_correspondent = CorrespondentFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=Correspondent.MATCH_LITERAL,
+        )
+        assign_perm("view_correspondent", classifying_user, visible_correspondent)
+        doc = DocumentFactory(owner=classifying_user, content="an invoice document")
+
+        matched = match_correspondents(doc, classifier=None, user=classifying_user)
+        matched_ids = {c.pk for c in matched}
+        assert visible_correspondent.pk in matched_ids
+        assert hidden_correspondent.pk not in matched_ids
+
+    def test_match_document_types_only_considers_document_types_visible_to_user(self):
+        owner = User.objects.create_user(username="document_type_owner")
+        classifying_user = User.objects.create_user(username="classifier_user3")
+        visible_document_type = DocumentTypeFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=DocumentType.MATCH_LITERAL,
+        )
+        hidden_document_type = DocumentTypeFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=DocumentType.MATCH_LITERAL,
+        )
+        assign_perm("view_documenttype", classifying_user, visible_document_type)
+        doc = DocumentFactory(owner=classifying_user, content="an invoice document")
+
+        matched = match_document_types(doc, classifier=None, user=classifying_user)
+        matched_ids = {dt.pk for dt in matched}
+        assert visible_document_type.pk in matched_ids
+        assert hidden_document_type.pk not in matched_ids
+
+    def test_match_storage_paths_only_considers_storage_paths_visible_to_user(self):
+        owner = User.objects.create_user(username="storage_path_owner")
+        classifying_user = User.objects.create_user(username="classifier_user4")
+        visible_storage_path = StoragePathFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=StoragePath.MATCH_LITERAL,
+        )
+        hidden_storage_path = StoragePathFactory(
+            owner=owner,
+            match="invoice",
+            matching_algorithm=StoragePath.MATCH_LITERAL,
+        )
+        assign_perm("view_storagepath", classifying_user, visible_storage_path)
+        doc = DocumentFactory(owner=classifying_user, content="an invoice document")
+
+        matched = match_storage_paths(doc, classifier=None, user=classifying_user)
+        matched_ids = {sp.pk for sp in matched}
+        assert visible_storage_path.pk in matched_ids
+        assert hidden_storage_path.pk not in matched_ids