queryparser: add RestrictedTerm to exclude allowed phrases by permission

invenio_records_resources/services/records/queryparser/transformer.py

@@ -33,15 +33,55 @@ def term_name(self):
         """Get the term name."""
         return self._term_name
 
-    def map_word(self, node):
+    def map_word(self, node, **kwargs):
         """Modify a word node."""
         return self._word_fun(node) if self._word_fun else node
 
-    def map_phrase(self, node):
+    def map_phrase(self, node, **kwargs):
         """Modify a phrase node."""
         return self._phrase_fun(node) if self._phrase_fun else node
 
 
+class RestrictedTerm:
+    """Class used to apply specific permissions to search."""
+
+    def __init__(self, permission, term_name=None, word=None, phrase=None):
+        self.permission = permission
+        self._word_fun = word
+        self._phrase_fun = phrase
+        self._term_name = term_name
+
+    def allows(self, identity):
+        if not (self._word_fun or self._phrase_fun):
+            return self.permission.allows(identity)
+        else:
+            # if we specified the rewrite phrase of word function,
+            # let them take over the permission check
+            return True
+
+    @property
+    def term_name(self):
+        """Get the term name."""
+        return self._term_name
+
+    def map_word(self, node, context, **kwargs):
+        """Modify a word node."""
+        is_restricted = not self.permission.allows(context["identity"])
+        if self._word_fun and is_restricted:
+            return self._word_fun(node)
+        else:
+            return node
+
+    #
+    def map_phrase(self, node, context, **kwargs):
+        """Modify a phrase node."""
+        is_restricted = not self.permission.allows(context["identity"])
+        if self._phrase_fun and is_restricted:
+            return self._phrase_fun(node)
+        else:
+            return node
+
+
 class SearchFieldTransformer(TreeTransformer):
     """Transform from user-friendly field names to internal field names."""
 
@@ -56,11 +96,19 @@ def visit_search_field(self, node, context):
         # Use the node name if not mapped for transformation.
         term_name = self._mapping.get(node.name, node.name)
         field_value_mapper = None
-
         if isinstance(term_name, FieldValueMapper):
             field_value_mapper = term_name
             term_name = field_value_mapper.term_name
-
+        if isinstance(term_name, RestrictedTerm):
+            field_value_mapper = term_name
+            allows = term_name.allows(context["identity"])
+            term_name = node.name
+            if not allows:
+                raise QuerystringValidationError(
+                    _("Invalid search field: {field_name}.").format(
+                        field_name=node.name
+                    )
+                )
         # If a allow list exists, the term must be allowed to be queried.
         if self._allow_list and not term_name in self._allow_list:
             raise QuerystringValidationError(
@@ -79,9 +127,9 @@ def visit_search_field(self, node, context):
     def visit_word(self, node, context):
         """Visit a word term."""
         mapper = context.get("field_value_mapper")
-        yield node if mapper is None else mapper.map_word(node)
+        yield node if mapper is None else mapper.map_word(node, context=context)
 
     def visit_phrase(self, node, context):
         """Visit a phrase term."""
         mapper = context.get("field_value_mapper")
-        yield node if mapper is None else mapper.map_phrase(node)
+        yield node if mapper is None else mapper.map_phrase(node, context=context)

tests/services/test_service_queryparser.py

@@ -9,14 +9,18 @@
 """Query parser tests."""
 
 import pytest
-from invenio_access.permissions import system_identity
-from luqum.tree import Phrase
+from flask_principal import ActionNeed
+from invenio_access.permissions import Permission, SystemRoleNeed, system_identity
+from luqum.tree import Phrase, Word
 
 from invenio_records_resources.services.records.queryparser import (
     FieldValueMapper,
     QueryParser,
     SearchFieldTransformer,
 )
+from invenio_records_resources.services.records.queryparser.transformer import (
+    RestrictedTerm,
+)
 
 
 @pytest.fixture()
@@ -194,3 +198,77 @@ def lol(node):
     assert p(system_identity).parse(query).to_dict() == {
         "query_string": {"query": transformed_query}
     }
+
+
+@pytest.mark.parametrize(
+    "query,transformed_query",
+    [
+        # Search field transformer does nor rewrite the first query, raises exception handled in
+        # https://github.com/inveniosoftware/invenio-records-resources/commit/f795bdb0594c7d2fbe03a8d22e7c2fa344f25c35
+        ("internal_notes.note:abc", "internal_notes.note:abc"),
+    ],
+)
+def test_querystring_restricted_term(query, transformed_query, identity_simple, app):
+    """Invalid syntax falls back to multi match query."""
+
+    sysadmin_permission = Permission(SystemRoleNeed("system_process"))
+
+    def word_internal_notes(node):
+        """Quote DOIs."""
+        if not node.value.startswith("internal_notes"):
+            return node
+        return Word("")
+
+    p = QueryParser.factory(
+        mapping={
+            "internal_notes.note": RestrictedTerm(sysadmin_permission),
+            "_exists_": RestrictedTerm(sysadmin_permission, word=word_internal_notes),
+        },
+        tree_transformer_cls=SearchFieldTransformer,
+    )
+
+    parser = p(system_identity)
+
+    assert parser.parse(query).to_dict() == {"query_string": {"query": query}}
+
+    parser = p(identity_simple)
+
+    assert parser.parse(query).to_dict() == {
+        "multi_match": {"query": transformed_query}
+    }
+
+
+@pytest.mark.parametrize(
+    "query,transformed_query",
+    [
+        ("_exists_:internal_notes", "_exists_:"),
+    ],
+)
+def test_querystring_restricted_term(query, transformed_query, identity_simple, app):
+    """Invalid syntax falls back to multi match query."""
+
+    sysadmin_permission = Permission(SystemRoleNeed("system_process"))
+
+    def word_internal_notes(node):
+        """Quote DOIs."""
+        if not node.value.startswith("internal_notes"):
+            return node
+        return Word("")
+
+    p = QueryParser.factory(
+        mapping={
+            "internal_notes.note": RestrictedTerm(sysadmin_permission),
+            "_exists_": RestrictedTerm(sysadmin_permission, word=word_internal_notes),
+        },
+        tree_transformer_cls=SearchFieldTransformer,
+    )
+
+    parser = p(system_identity)
+
+    assert parser.parse(query).to_dict() == {"query_string": {"query": query}}
+
+    parser = p(identity_simple)
+
+    assert parser.parse(query).to_dict() == {
+        "query_string": {"query": transformed_query}
+    }