Fix structural annotation document visibility

config/graphql/annotation_types.py

@@ -657,24 +657,53 @@ def resolve_document(self, info) -> Any:
         method ever ran.
         """
         if self.document_id:
-            return self.document
+            from opencontractserver.documents.models import Document
+
+            return (
+                BaseService.filter_visible_qs(
+                    Document.objects.filter(pk=self.document_id),
+                    info.context.user,
+                    request=info.context,
+                )
+                .select_related("creator")
+                .first()
+            )
         # Structural annotations have document=NULL; resolve via structural_set
         if self.structural_set_id:
+            from opencontractserver.documents.models import Document
+
             structural_set = self.structural_set
             if structural_set is not None:
-                # Use prefetched documents if available (evaluates prefetch cache)
+                # Use prefetched documents if available (evaluates prefetch cache),
+                # but do not trust the prefetch alone as a permission gate: the
+                # unscoped annotations query may prefetch every document sharing
+                # this structural set. Intersect candidates with visible_to_user
+                # before returning a DocumentType while preserving prefetch order.
                 prefetched = list(structural_set.documents.all())
                 if prefetched:
-                    return prefetched[0]
+                    prefetched_ids = [document.id for document in prefetched]
+                    visible_ids = set(
+                        BaseService.filter_visible_qs(
+                            Document.objects.filter(pk__in=prefetched_ids),
+                            info.context.user,
+                            request=info.context,
+                        ).values_list("pk", flat=True)
+                    )
+                    return next(
+                        (
+                            document
+                            for document in prefetched
+                            if document.id in visible_ids
+                        ),
+                        None,
+                    )
             # Fallback when the caller did not apply
             # ``AnnotationService.structural_document_prefetch`` (deferred import
             # avoids a module-level cycle with documents.models). Scope to this
             # annotation's own corpus and order deterministically so we never
             # reintroduce the original arbitrary ``.documents.first()`` bug;
             # query-context scoping (which corpus is being viewed) only happens
             # via the prefetch above, so this is a best-effort degraded path.
-            from opencontractserver.documents.models import Document
-
             documents = Document.objects.filter(
                 structural_annotation_set_id=self.structural_set_id
             )
@@ -684,7 +713,15 @@ def resolve_document(self, info) -> Any:
                     path_records__is_current=True,
                     path_records__is_deleted=False,
                 )
-            return documents.order_by("slug").first()
+            return (
+                BaseService.filter_visible_qs(
+                    documents,
+                    info.context.user,
+                    request=info.context,
+                )
+                .order_by("slug")
+                .first()
+            )
         return None
 
     def resolve_annotation_type(self, info) -> Any:

opencontractserver/tests/test_corpus_cards_structural_document_resolution.py

@@ -59,6 +59,11 @@ def setUp(self):
             password="testpass123",
             email="cards_struct_doc@test.com",
         )
+        self.other_user = User.objects.create_user(
+            username="cards_struct_doc_other_user",
+            password="testpass123",
+            email="cards_struct_doc_other@test.com",
+        )
 
         # A single content hash → a single StructuralAnnotationSet shared by
         # the source document and every corpus copy of it.
@@ -84,6 +89,18 @@ def setUp(self):
         set_permissions_for_obj_to_user(
             self.user, self.source_doc, [PermissionTypes.READ]
         )
+        # A private document owned by another user that shares the structural
+        # set and sorts first in the unscoped structural document prefetch.
+        # Resolving through this row would leak another user's DocumentType.
+        self.private_doc = Document.objects.create(
+            title="Private shared S-1",
+            slug="aaa-private-shared-s-1",
+            creator=self.other_user,
+            pdf_file_hash=content_hash,
+            structural_annotation_set=self.structural_set,
+            page_count=3,
+            processing_started=timezone.now(),
+        )
 
         # Two corpuses, each receiving an isolated copy that SHARES the set.
         self.corpus_a = Corpus.objects.create(
@@ -153,6 +170,20 @@ def _client(self):
         }
     """
 
+    _UNSCOPED_QUERY = """
+        query Cards {
+            annotations(structural: true, first: 100) {
+                edges {
+                    node {
+                        id
+                        structural
+                        document { id slug title }
+                    }
+                }
+            }
+        }
+    """
+
     def _nodes_by_annotation_id(self, corpus):
         """Run the corpus cards query and return ``{annotation_gid: node}``.
 
@@ -256,3 +287,25 @@ def test_prefetch_document_id_takes_precedence_over_corpus_id(self):
             [self.doc_b.id],
             "document_id must take precedence over corpus_id in the prefetch",
         )
+
+    def test_unscoped_structural_resolution_skips_private_shared_documents(self):
+        """Unscoped annotation browsing must not leak a private shared doc."""
+        annotations = self._make_structural_annotations(self.corpus_a, "A")
+        result = self._client().execute(self._UNSCOPED_QUERY)
+        self.assertIsNone(
+            result.get("errors"), f"GraphQL errors: {result.get('errors')}"
+        )
+        nodes = {
+            edge["node"]["id"]: edge["node"]
+            for edge in result["data"]["annotations"]["edges"]
+        }
+        expected_annotation_gid = to_global_id("AnnotationType", annotations[0].id)
+        self.assertIn(expected_annotation_gid, nodes)
+
+        resolved_doc = nodes[expected_annotation_gid]["document"]
+        self.assertIsNotNone(resolved_doc)
+        self.assertNotEqual(
+            resolved_doc["id"],
+            to_global_id("DocumentType", self.private_doc.id),
+            "resolve_document returned a private document from the shared set",
+        )