Show hidden annotations to group moderators

Author: marcospri

h/search/core.py

@@ -50,9 +50,8 @@ def __init__(
             query.Limiter(),
             query.DeletedFilter(),
             query.AuthFilter(request),
-            query.GroupFilter(request),
+            query.GroupAndModerationFilter(request),
             query.UserFilter(),
-            query.HiddenFilter(request),
             query.AnyMatcher(),
             query.TagsMatcher(),
             query.UriCombinedWildcardFilter(

h/search/query.py

@@ -7,6 +7,7 @@
 
 from h import storage
 from h.search.util import add_default_scheme, wildcard_uri_is_valid
+from h.security.permissions import Permission
 from h.util import uri
 
 LIMIT_DEFAULT = 20
@@ -207,23 +208,62 @@ def __call__(self, search, params):  # noqa: ARG002
         return search.filter("term", shared=True)
 
 
-class GroupFilter:
-    """
-    Filter that limits which groups annotations are returned from.
-
-    This excludes annotations from groups that the user is not authorized to
-    read or which are explicitly excluded by the search query.
-    """
-
+class GroupAndModerationFilter:
     def __init__(self, request):
         self.user = request.user
         self.group_service = request.find_service(name="group")
 
     def __call__(self, search, params):
-        # Remove parameter if passed, preventing it being passed to default query
         group_ids = popall(params, "group") or None
-        groups = self.group_service.groupids_readable_by(self.user, group_ids)
-        return search.filter("terms", group=groups)
+        groups = self.group_service.groups_readable_by(self.user, group_ids)
+
+        not_nipsa = ~Q("term", nipsa=True)
+        not_hidden = ~Q("term", hidden=True)
+
+        if not self.user:
+            # If there not a logged in user
+            if groups:
+                # Apply the group filter as it is
+                search = search.filter("terms", group=[g.pubid for g in groups])
+            # And don't show any hidden or NIPSA'd annotations
+            return search.filter(not_nipsa & not_hidden)
+
+        if not groups:
+            # If  the user is logged in, hide hidden annos and NIPSA'd annos except the ones authored by the user
+            return search.filter(
+                (not_hidden & not_nipsa) | Q("term", user=self.user.userid.lower())
+            )
+
+        query_clauses = []
+
+        from h.security import Identity, identity_permits
+        from h.traversal import GroupContext
+
+        # If t he user is logged in and we are filtering by groups
+        # we'll check for each group if we are a moderator
+        for group in groups:
+            user_is_moderator = identity_permits(
+                identity=Identity.from_models(user=self.user),
+                context=GroupContext(group),
+                permission=Permission.Group.MODERATE,
+            )
+            if user_is_moderator:
+                # For modereators:
+                # - We show all their annos
+                # - We don't filter out hideden annos
+                # - We do hide NIPSA'd annos
+                query_clauses = Q("term", group=group.pubid) & not_nipsa
+
+            else:
+                # For non moderators:
+                # - We show all their annos
+                # - We hide hidden annos
+                # - We hide NIPSA'd annos
+                query_clauses = Q("term", group=group.pubid) & (
+                    not_hidden & not_nipsa
+                ) | Q("term", user=self.user.userid.lower())
+
+        return search.filter(Q("bool", should=query_clauses))
 
 
 class UriCombinedWildcardFilter:
@@ -347,29 +387,6 @@ def __call__(self, search, _):
         return search.exclude("exists", field="deleted")
 
 
-class HiddenFilter:
-    """Return an Elasticsearch filter for filtering out moderated or NIPSA'd annotations."""
-
-    def __init__(self, request):
-        self.group_service = request.find_service(name="group")
-        self.user = request.user
-
-    def __call__(self, search, _):
-        """Filter out all hidden and NIPSA'd annotations except the current user's."""
-        # If any one of these "should" clauses is true then the annotation will
-        # get through the filter.
-        should_clauses = [
-            Q("bool", must_not=[Q("term", nipsa=True), Q("term", hidden=True)])
-        ]
-
-        if self.user is not None:
-            # Always show the logged-in user's annotations even if they have
-            # been hidden or the user has been NIPSA'd
-            should_clauses.append(Q("term", user=self.user.userid.lower()))
-
-        return search.filter(Q("bool", should=should_clauses))
-
-
 class AnyMatcher:
     """Match the contents of a selection of fields against the `any` parameter."""
 

h/services/group.py

@@ -1,7 +1,8 @@
 import sqlalchemy as sa
 from sqlalchemy import literal_column, select, union
+from sqlalchemy.orm import aliased
 
-from h.models import Group, GroupMembership
+from h.models import Group, GroupMembership, User
 from h.models.group import ReadableBy
 from h.util import group as group_util
 
@@ -76,28 +77,26 @@ def filter_by_name(self, name=None):
             .order_by(Group.created.desc())
         )
 
-    def groupids_readable_by(self, user, group_ids=None):
+    def groups_readable_by(self, user: User | None, group_ids=None) -> list[Group]:
         """
-        Return a list of pubids for which the user has read access.
+        Return a list of groups for which the user has read access.
 
         If the passed-in user is ``None``, this returns the list of
         world-readable groups.
 
         If `group_ids` is specified, only the subset of groups from that list is
         returned. This is more efficient if the caller wants to know which
         groups from a specific list are readable by the user.
-
-        :type user: `h.models.user.User`
         """
         # Very few groups are readable by world, query those separately
-        readable_by_world_query = select(Group.pubid).where(
+        readable_by_world_query = select(Group).where(
             Group.readable_by == ReadableBy.world
         )
         query = readable_by_world_query
 
         if user is not None:
             user_is_member_query = (
-                select(Group.pubid)
+                select(Group)
                 .join(GroupMembership)
                 .where(
                     GroupMembership.user_id == user.id,
@@ -106,14 +105,22 @@ def groupids_readable_by(self, user, group_ids=None):
             )
             # Union these with the world readable ones
             # We wrap a subquery around in case we need to apply the group_ids filter below
-            query = select(
-                union(readable_by_world_query, user_is_member_query).subquery()
-            )
+            union_groups = union(
+                readable_by_world_query, user_is_member_query
+            ).subquery()
+            # We explicitly alias the union to hint SQLAlchemy to fetch Group objects
+            group_alias = aliased(Group, union_groups)
+            query = select(group_alias)
 
         if group_ids:
             query = query.where(literal_column("pubid").in_(group_ids))
 
-        return self.session.scalars(query).all()
+        return self.session.scalars(query)
+
+    def groupids_readable_by(self, user, group_ids=None):
+        """Return a list of groupids for which the user has read access."""
+
+        return [g.pubid for g in self.groups_readable_by(user, group_ids)]
 
     def groupids_created_by(self, user):
         """

tests/unit/h/search/core_test.py

@@ -306,3 +306,9 @@ def test_only_200_replies_are_included(self, pyramid_request, Annotation):
 
         assert len(result.reply_ids) == 3
         assert oldest_reply.id not in result.reply_ids
+
+
+@pytest.fixture
+def group_service(group_service):
+    group_service.groups_readable_by.return_value = []
+    return group_service