Validate search query values before sending to solr

geotags/tests.py

@@ -110,3 +110,10 @@ def test_browse_geotags_for_query(self):
         resp = self.client.get(reverse("geotags-query") + "?q=barcelona")
         check_values = {"query_description": '"barcelona"'}
         self.check_context(resp.context, check_values)
+
+    def test_geotags_for_query_barray_invalid_filter_returns_empty(self):
+        # A corrupted/invalid filter sets sqp.errors, which must short-circuit before
+        # Solr; the endpoint returns an empty bytearray rather than crashing.
+        resp = self.client.get(reverse("geotags-for-query-barray") + "?f=samplerate%3Aabc")
+        self.assertEqual(resp.status_code, 200)
+        self.assertEqual(len(resp.content), 0)

geotags/views.py

@@ -169,12 +169,16 @@ def geotags_for_query_barray(request):
     else:
         # Otherwise, perform a search query to get the results
         sqp = SearchQueryProcessor(request)
-        query_params = sqp.as_query_params()
-        if "facets" in query_params:
-            # No need to compute facets for bytearray query
-            del query_params["facets"]
-        results, _ = perform_search_engine_query(query_params)
-        results_docs = results.docs
+        if sqp.errors:
+            # invalid filter: don't search on solr
+            results_docs = []
+        else:
+            query_params = sqp.as_query_params()
+            if "facets" in query_params:
+                # No need to compute facets for bytearray query
+                del query_params["facets"]
+            results, _ = perform_search_engine_query(query_params)
+            results_docs = results.docs
 
     if results_docs is None:
         results_docs = []

search/tests/test_views.py

@@ -301,6 +301,29 @@ def test_failed_search_result_clustering_view(self, get_clusters_for_query, get_
         self.assertTemplateUsed(resp, "search/clustering_results.html")
         self.assertEqual(resp.context["clusters_data"], None)
 
+    def test_clusters_section_invalid_filter_returns_no_clusters(self):
+        # A corrupted/invalid filter sets sqp.errors, which must short-circuit
+        # and not send a request to solr and return an empty page.
+        resp = self.client.get(reverse("clusters-section") + "?f=samplerate%3Aabc")
+        self.assertEqual(resp.status_code, 200)
+        self.assertTemplateUsed(resp, "search/clustering_results.html")
+        self.assertEqual(resp.context["clusters_data"], None)
+
+    def test_clustered_graph_invalid_filter_returns_error(self):
+        # A corrupted/invalid filter sets sqp.errors, which must short-circuit
+        # and not send a request to solr and return an empty response.
+        resp = self.client.get(reverse("clustered-graph-json") + "?f=samplerate%3Aabc")
+        self.assertEqual(resp.status_code, 200)
+        self.assertIn(b"error", resp.content)
+
+    @mock.patch("search.views.get_clusters_for_query")
+    def test_clustered_graph_clustering_timeout_returns_error(self, get_clusters_for_query):
+        # On clustering timeout/failure get_clusters_for_query returns {"clusters": None}
+        get_clusters_for_query.return_value = {"clusters": None}
+        resp = self.client.get(reverse("clustered-graph-json"))
+        self.assertEqual(resp.status_code, 200)
+        self.assertIn(b"error", resp.content)
+
 
 @pytest.mark.django_db
 class TestSearchDeepPagination:

search/views.py

@@ -328,7 +328,7 @@ def _get_clusters_data_helper(sqp):
 
 def clusters_section(request):
     sqp = search_query_processor.SearchQueryProcessor(request)
-    clusters_data = _get_clusters_data_helper(sqp)
+    clusters_data = None if sqp.errors else _get_clusters_data_helper(sqp)
     if clusters_data is None:
         return render(request, "search/clustering_results.html", {"clusters_data": None})
     return render(request, "search/clustering_results.html", {"sqp": sqp, "clusters_data": clusters_data})
@@ -339,9 +339,13 @@ def clustered_graph(request):
     # TODO: this view is currently not used in the new UI, but we could add a modal in the
     # clustering section to show results in a graph.
     sqp = search_query_processor.SearchQueryProcessor(request)
+    if sqp.errors:
+        return JsonResponse(json.dumps({"error": True}), safe=False)
+
     results = get_clusters_for_query(sqp)
-    if results is None:
-        JsonResponse(json.dumps({"error": True}), safe=False)
+    if results is None or results.get("clusters") is None:
+        return JsonResponse(json.dumps({"error": True}), safe=False)
+
     graph = get_clustering_data_for_graph_display(sqp, results["graph"])
     return JsonResponse(json.dumps(graph), safe=False)
 

utils/search/filter_validation.py

@@ -0,0 +1,89 @@
+import luqum.tree
+from luqum.parser import parser
+
+FIELD_TYPES_MAP = {
+    "samplerate": int,
+    "bitrate": int,
+    "bitdepth": int,
+    "channels": int,
+    "duration": float,
+}
+
+COMPLEX_EXPR_TYPES = (
+    luqum.tree.Range,
+    luqum.tree.OpenRange,
+    luqum.tree.From,
+    luqum.tree.To,
+    luqum.tree.FieldGroup,
+    luqum.tree.Boost,
+)
+
+
+def parse_filter(filter_string):
+    """Parse a Lucene-style filter string into its list of top-level nodes.
+    Empty string -> []. Raises luqum.exceptions.ParseError on malformed input.
+    """
+    if not filter_string:
+        return []
+    tree = parser.parse(filter_string)
+    return [tree] if type(tree) == luqum.tree.SearchField else tree.children
+
+
+def validate_filter_types(nodes) -> str | None:
+    """Validate that in the given top-level filter nodes, the values of fields
+    with a known type (FIELD_TYPES_MAP) can be cast to that type. Return a
+    human-readable error string on first failure, or None on success.
+    """
+    for node in nodes:
+        if type(node) != luqum.tree.SearchField:
+            continue
+        if node.name not in FIELD_TYPES_MAP:
+            continue
+        expr = node.expr
+        if isinstance(expr, COMPLEX_EXPR_TYPES):
+            # Ranges/inequalities/groups/boosts etc. e.g. samplerate:[a TO b]
+            # validate directly in solr
+            continue
+        if isinstance(expr, luqum.tree.Phrase):
+            # String in "quotes"
+            value = expr.value[1:-1]
+        else:
+            value = str(expr)
+        if value == "*":
+            # `field:*` is a valid solr query, continue
+            continue
+        expected_type = FIELD_TYPES_MAP[node.name]
+        # This does let through some values that are valid in python but not solr
+        # (e.g. int(1_000)), but we accept these few cases because they are rare
+        try:
+            expected_type(value)
+        except (ValueError, TypeError):
+            return f"Filter parsing error: '{node.name}' value must be {expected_type.__name__}, got {value!r}"
+    return None
+
+
+def find_dropped_filters(nodes) -> list[str]:
+    """Return a list of human-readable representations of filters that
+    SearchQueryProcessor would silently drop.
+
+    Hand-made URLs sometimes use `%2B` instead of a space between filters, which
+    decodes as a `+` instead of a space. Lucene reads that `+` as a MUST prefix,
+    so the following filter parses as a top-level Plus-wrapped node:
+
+        sent:     f=tag%3A%22tap%22%2Busername%3A%22ascap%22
+        decoded:  tag:"tap"+username:"ascap"
+        expected: tag:"tap" username:"ascap"
+        parsed:   [SearchField('tag', '"tap"'), Plus(SearchField('username', '"ascap"'))]
+        expected  [SearchField('tag', '"tap"'), SearchField('username', '"ascap"')]
+
+    SearchQueryProcessor currently drops the second part of the search query, so
+    this function returns ['username:"ascap"'].
+
+    Empty list means nothing was removed.
+    """
+    dropped = []
+    for node in nodes:
+        if isinstance(node, luqum.tree.Plus) and isinstance(node.a, luqum.tree.SearchField):
+            # str() of a luqum node can keep trailing whitespace from the source string
+            dropped.append(str(node.a).strip())
+    return dropped

utils/search/search_query_processor.py

@@ -22,12 +22,12 @@
 import json
 import urllib.parse
 
+import luqum.exceptions
 import luqum.tree
 from django.conf import settings
 from django.core.exceptions import ValidationError
 from django.urls import reverse
 from django.utils.http import urlencode
-from luqum.parser import parser
 from luqum.pretty import prettify
 
 from sounds.models import Sound
@@ -36,6 +36,7 @@
 from utils.clustering_utilities import get_clusters_for_query, get_ids_in_cluster
 from utils.encryption import create_hash
 from utils.search import SearchEngineException
+from utils.search.filter_validation import parse_filter, validate_filter_types
 from utils.search.search_sounds import allow_beta_search_features
 
 from .search_query_processor_options import (
@@ -66,6 +67,9 @@ class SearchQueryProcessor:
     """The SearchQueryProcessor class is used to parse and process search query information from a request object and
     compute a number of useful items for displaying search information in templates, constructing search URLs, and
     preparing search options to be passed to the backend search engine.
+
+    You must check `self.errors` after constructing this object. If it's not empty then it's not guaranteed
+    to be completely initialised.
     """
 
     request = None
@@ -337,14 +341,10 @@ def __init__(self, request, facets=None):
             self.options[option_name] = option
 
         # Get filter and parse it. Make sure it is iterable (even if it only has one element)
-        self.f = urllib.parse.unquote(request.GET.get("f", "")).strip().lstrip()
+        self.f = urllib.parse.unquote(request.GET.get("f", "")).strip()
         if self.f:
             try:
-                f_parsed = parser.parse(self.f)
-                if type(f_parsed) == luqum.tree.SearchField:
-                    self.f_parsed = [f_parsed]
-                else:
-                    self.f_parsed = f_parsed.children
+                self.f_parsed = parse_filter(self.f)
             except luqum.exceptions.ParseError as e:
                 self.errors = f"Filter parsing error: {e}"
                 self.f_parsed = []
@@ -364,6 +364,14 @@ def __init__(self, request, facets=None):
                     self.errors = f"Filter parsing error: invalid tag value '{tag_value}'"
                     return
 
+        # Validate that values on filter fields are the correct type. Catches issues where incorrectly formatted
+        # queries cause a solr exception (e.g. `samplerate:22050+tag:"x"` parses as "22050+tag" instead of 22050)
+        if not self.errors:
+            type_error = validate_filter_types(self.f_parsed)
+            if type_error is not None:
+                self.errors = type_error
+                return
+
         # Remove duplicate filters if any
         nodes_in_filter = []
         f_parsed_no_duplicates = []
@@ -686,7 +694,12 @@ def as_query_params(self, exclude_facet_filters=False):
 
         Returns:
             dict: Dictionary with the query parameters to be used by the SearchEngine.search_sounds method.
+
+        Raises:
+            SearchEngineException: if self.errors is set (see class docstring).
         """
+        if self.errors:
+            raise SearchEngineException(f"Cannot build query params, filter has errors: {self.errors}")
 
         # Filter field weights by "search in" options
         field_weights = self.get_option_value_to_apply("field_weights")