Improve algolia.py input data validation

backend/apps/core/api/algolia.py

@@ -6,11 +6,13 @@
 from algoliasearch.http.exceptions import AlgoliaException
 from django.conf import settings
 from django.core.cache import cache
+from django.core.exceptions import ValidationError
 from django.http import JsonResponse
 
 from apps.common.index import IndexBase
 from apps.common.utils import get_user_ip_address
 from apps.core.utils.params_mapping import get_params_for_index
+from apps.core.validators import validate_search_params
 
 CACHE_PREFIX = "algolia_proxy"
 CACHE_TTL_IN_SECONDS = 3600  # 1 hour
@@ -50,11 +52,16 @@ def algolia_search(request):
     try:
         data = json.loads(request.body)
 
+        try:
+            validate_search_params(data)
+        except ValidationError as error:
+            return JsonResponse({"error": error.message}, status=400)
+
         facet_filters = data.get("facetFilters", [])
         index_name = data.get("indexName")
         ip_address = get_user_ip_address(request)
-        limit = int(data.get("hitsPerPage", 25))
-        page = int(data.get("page", 1))
+        limit = data.get("hitsPerPage", 25)
+        page = data.get("page", 1)
         query = data.get("query", "")
 
         cache_key = f"{CACHE_PREFIX}:{index_name}:{query}:{page}:{limit}"

backend/apps/core/validators.py

@@ -0,0 +1,79 @@
+"""Validators for the search parameters of the Algolia endpoint."""
+
+import re
+
+from django.core.exceptions import ValidationError
+from django.core.validators import validate_slug
+
+
+def validate_index_name(index_name):
+    """Validate index name."""
+    if not index_name or not isinstance(index_name, str):
+        message = "indexName is required and must be a string."
+        raise ValidationError(message)
+
+    try:
+        validate_slug(index_name)
+    except ValidationError:
+        message = (
+            "Invalid indexName value provided. "
+            "Only alphanumeric characters hyphens and underscores are allowed."
+        )
+        raise ValidationError(message) from None
+
+
+def validate_limit(limit):
+    """Validate limit."""
+    if not isinstance(limit, int):
+        message = "hitsPerPage must be an integer."
+        raise ValidationError(message)
+
+    min_limit = 1
+    max_limit = 1000
+    if limit < min_limit or limit > max_limit:
+        message = "hitsPerPage value must be between 1 and 1000."
+        raise ValidationError(message)
+
+
+def validate_page(page):
+    """Validate page."""
+    if not isinstance(page, int):
+        message = "page value must be an integer."
+        raise ValidationError(message)
+
+    if page <= 0:
+        message = "page value must be a positive integer."
+        raise ValidationError(message)
+
+
+def validate_query(query):
+    """Validate query."""
+    if not query:
+        return
+
+    if not isinstance(query, str):
+        message = "query must be a string."
+        raise ValidationError(message)
+
+    if not re.match(r"^[a-zA-Z0-9-_ ]*$", query):
+        message = (
+            "Invalid query value provided. "
+            "Only alphanumeric characters, hyphens, spaces and underscores are allowed."
+        )
+        raise ValidationError(message)
+
+
+def validate_facet_filters(facet_filters):
+    """Validate facet filters."""
+    if not isinstance(facet_filters, list):
+        message = "facetFilters must be a list."
+        raise ValidationError(message)
+
+
+def validate_search_params(data):
+    """Validate search parameters."""
+    validate_facet_filters(data.get("facetFilters", []))
+    validate_index_name(data.get("indexName"))
+    validate_limit(data.get("hitsPerPage", 25))
+    validate_page(data.get("page", 1))
+    validate_query(data.get("query", ""))

backend/tests/core/api/algolia_test.py

@@ -88,3 +88,53 @@ def test_algolia_search_invalid_method(self):
 
         assert response.status_code == requests.codes.method_not_allowed
         assert response_data["error"] == "Method GET is not allowed"
+
+    @pytest.mark.parametrize(
+        ("index_name", "query", "page", "hits_per_page", "facet_filters", "error_message"),
+        [
+            # Index name tests
+            (
+                5,
+                "owasp",
+                2,
+                20,
+                ["idx_is_active:true"],
+                "indexName is required and must be a string.",
+            ),
+            # Query tests
+            ("chapters", 5, 2, 20, ["idx_is_active:true"], "query must be a string."),
+            # Page tests
+            ("committees", "review", "0", 5, [], "page value must be an integer."),
+            # hitsPerPage tests
+            ("committees", "review", 1, "1001", [], "hitsPerPage must be an integer."),
+            # Facet filters tests
+            ("issues", "bug", 1, 10, "idx_is_active:true", "facetFilters must be a list."),
+        ],
+    )
+    def test_algolia_search_invalid_request(
+        self,
+        index_name,
+        query,
+        page,
+        hits_per_page,
+        facet_filters,
+        error_message,
+    ):
+        """Test invalid requests for the algolia_search."""
+        mock_request = Mock()
+        mock_request.method = "POST"
+        mock_request.body = json.dumps(
+            {
+                "facetFilters": facet_filters,
+                "hitsPerPage": hits_per_page,
+                "indexName": index_name,
+                "page": page,
+                "query": query,
+            }
+        )
+
+        response = algolia_search(mock_request)
+        response_data = json.loads(response.content)
+
+        assert response.status_code == requests.codes.bad_request
+        assert response_data["error"] == error_message

backend/tests/core/validators_test.py

@@ -0,0 +1,113 @@
+import pytest
+from django.core.exceptions import ValidationError
+
+from apps.core.validators import (
+    validate_facet_filters,
+    validate_index_name,
+    validate_limit,
+    validate_page,
+    validate_query,
+)
+
+
+class TestAlgoliaValidators:
+    # Index name tests
+    @pytest.mark.parametrize(
+        ("index_name", "error_message"),
+        [
+            (5, "indexName is required and must be a string."),
+            ("", "indexName is required and must be a string."),
+            (
+                "index!name",
+                (
+                    "Invalid indexName value provided. "
+                    "Only alphanumeric characters hyphens and underscores are allowed."
+                ),
+            ),
+            (
+                "index name",
+                (
+                    "Invalid indexName value provided. "
+                    "Only alphanumeric characters hyphens and underscores are allowed."
+                ),
+            ),
+        ],
+    )
+    def test_invalid_index_name(self, index_name, error_message):
+        with pytest.raises(ValidationError) as exc_info:
+            validate_index_name(index_name)
+        assert str(exc_info.value.messages[0]) == error_message
+
+    def test_valid_index_name(self):
+        validate_index_name("index_name")
+
+    # hitsPerPage tests
+    @pytest.mark.parametrize(
+        ("limit", "error_message"),
+        [
+            (0, "hitsPerPage value must be between 1 and 1000."),
+            (1001, "hitsPerPage value must be between 1 and 1000."),
+            ("5", "hitsPerPage must be an integer."),
+        ],
+    )
+    def test_invalid_limit(self, limit, error_message):
+        with pytest.raises(ValidationError) as exc_info:
+            validate_limit(limit)
+        assert str(exc_info.value.messages[0]) == error_message
+
+    def test_valid_limit(self):
+        validate_limit(5)
+
+    # Page tests
+    @pytest.mark.parametrize(
+        ("page", "error_message"),
+        [
+            (0, "page value must be a positive integer."),
+            ("5", "page value must be an integer."),
+        ],
+    )
+    def test_invalid_page(self, page, error_message):
+        with pytest.raises(ValidationError) as exc_info:
+            validate_page(page)
+        assert str(exc_info.value.messages[0]) == error_message
+
+    def test_valid_page(self):
+        validate_page(5)
+
+    # Query tests
+    @pytest.mark.parametrize(
+        ("query", "error_message"),
+        [
+            (5, "query must be a string."),
+            (
+                "query!name",
+                (
+                    "Invalid query value provided. "
+                    "Only alphanumeric characters, hyphens, spaces and underscores are allowed."
+                ),
+            ),
+        ],
+    )
+    def test_invalid_query(self, query, error_message):
+        with pytest.raises(ValidationError) as exc_info:
+            validate_query(query)
+        assert str(exc_info.value.messages[0]) == error_message
+
+    @pytest.mark.parametrize(("query"), [("query_name"), ("query-name"), ("query name")])
+    def test_valid_query(self, query):
+        validate_query(query)
+
+    # Facet filters tests
+    @pytest.mark.parametrize(
+        ("facet_filters", "error_message"),
+        [
+            (5, "facetFilters must be a list."),
+        ],
+    )
+    def test_invalid_facet_filters(self, facet_filters, error_message):
+        with pytest.raises(ValidationError) as exc_info:
+            validate_facet_filters(facet_filters)
+        assert str(exc_info.value.messages[0]) == error_message
+
+    def test_valid_facet_filters(self):
+        validate_facet_filters([])