Show hidden annotations to group moderators

Author: marcospri

h/search/core.py

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

h/search/query.py

@@ -1,19 +1,22 @@
-from datetime import datetime as dt
+from datetime import UTC, datetime
 
-from dateutil import tz
 from dateutil.parser import parse
 from elasticsearch_dsl import Q
 from elasticsearch_dsl.query import SimpleQueryString
 
 from h import storage
+from h.models import Group, User
 from h.search.util import add_default_scheme, wildcard_uri_is_valid
+from h.security.identity import Identity
+from h.security.permissions import Permission
+from h.security.permits import identity_permits
 from h.util import uri
 
 LIMIT_DEFAULT = 20
 # Elasticsearch requires offset + limit must be <= 10,000.
 LIMIT_MAX = 200
 OFFSET_MAX = 9800
-DEFAULT_DATE = dt(1970, 1, 1, 0, 0, 0, 0).replace(tzinfo=tz.tzutc())  # noqa: DTZ001
+DEFAULT_DATE = datetime(1970, 1, 1, 0, 0, 0, 0, tzinfo=UTC)
 
 
 def popall(multidict, key):
@@ -142,7 +145,7 @@ def _parse_date(str_value):
         except ValueError:
             try:
                 date = parse(str_value, default=DEFAULT_DATE)
-                return dt.timestamp(date) * 1000
+                return datetime.timestamp(date) * 1000
 
             except ValueError:
                 pass
@@ -207,23 +210,57 @@ 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")
 
+    @staticmethod
+    def _is_moderator(user: User, group: Group) -> bool:
+        from h.traversal import GroupContext
+
+        return identity_permits(
+            identity=Identity.from_models(user=user),
+            context=GroupContext(group),
+            permission=Permission.Group.MODERATE,
+        )
+
     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)
+
+        if not groups:
+            # We are assuming that there's always some group readable by the user (eg. __world__)
+            # if for some reason that was not the case, we would return all annotations
+            # let's explicitly raise an error in that case
+            msg = "No groups readable by the user"
+            raise ValueError(msg)
+
+        not_hidden = ~Q("term", hidden=True)
+
+        if not self.user:
+            return search.filter(
+                not_hidden & Q("terms", group=[g.pubid for g in groups])
+            )
+
+        user_annotations = Q("term", user=self.user.userid.lower())
+        query_clauses = []
+        # 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:
+            group_annotations = Q("term", group=group.pubid)
+            if self._is_moderator(self.user, group):
+                # We don't filter out hidden annotations for moderators
+                query_clauses.append(group_annotations)
+
+            else:
+                # For non moderators we hide moderated annotations except the ones authored
+                # by the user
+                query_clauses.append(
+                    group_annotations & (not_hidden | user_annotations)
+                )
+
+        return search.filter(Q("bool", should=query_clauses))
 
 
 class UriCombinedWildcardFilter:

h/security/permits.py

@@ -1,7 +1,6 @@
 from pyramid.security import Allowed, Denied
 
 from h.security.identity import Identity
-from h.security.permission_map import PERMISSION_MAP
 
 
 def identity_permits(
@@ -17,6 +16,8 @@ def identity_permits(
     :param context: Context object representing the objects acted upon
     :param permission: Permission requested
     """
+    from h.security.permission_map import PERMISSION_MAP
+
     if clauses := PERMISSION_MAP.get(permission):  # noqa: SIM102
         # Grant the permissions if for *any* single clause...
         if any(

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/conftest.py

@@ -2,14 +2,21 @@
 
 import pytest
 
+from h.models import Group
 from h.services.group import GroupService
 from h.services.search_index import SearchIndexService
 
 
 @pytest.fixture
-def group_service(pyramid_config):
+def world_group(db_session):
+    return db_session.query(Group).filter_by(pubid="__world__").one()
+
+
+@pytest.fixture
+def group_service(pyramid_config, world_group):
     group_service = mock.create_autospec(GroupService, instance=True, spec_set=True)
     group_service.groupids_readable_by.return_value = ["__world__"]
+    group_service.groups_readable_by.return_value = [world_group]
     pyramid_config.register_service(group_service, name="group")
     return group_service