DEV-15453: Updated ES sanitizing functions to handle sanitization byp…

usaspending_api/common/query_with_filters.py

@@ -19,15 +19,13 @@
 from usaspending_api.search.filters.elasticsearch.naics import NaicsCodes
 from usaspending_api.search.filters.elasticsearch.psc import PSCCodes
 from usaspending_api.search.filters.elasticsearch.tas import TasCodes, TreasuryAccounts
-from usaspending_api.search.filters.time_period.decorators import (
-     NewAwardsOnlyTimePeriod,
-)
+from usaspending_api.search.filters.time_period.decorators import NewAwardsOnlyTimePeriod
 from usaspending_api.search.filters.time_period.query_types import (
     AwardSearchTimePeriod,
     SubawardSearchTimePeriod,
     TransactionSearchTimePeriod,
 )
-from usaspending_api.search.v2.es_sanitization import es_sanitize
+from usaspending_api.search.v2.es_sanitization import es_minimal_sanitize, es_sanitize
 
 logger = logging.getLogger(__name__)
 
@@ -195,9 +193,10 @@ def generate_elasticsearch_query(cls, filter_values: List[str], query_type: Quer
             "sub_ultimate_parent_uei",
         ]
         for filter_value in filter_values:
-            keyword_queries.append(ES_Q("multi_match", query=filter_value, fields=text_fields, type="phrase_prefix"))
+            sanitized_value = es_minimal_sanitize(filter_value)
+            keyword_queries.append(ES_Q("multi_match", query=sanitized_value, fields=text_fields, type="phrase_prefix"))
             keyword_queries.append(
-                ES_Q("query_string", query=filter_value, default_operator="OR", fields=keyword_fields)
+                ES_Q("query_string", query=sanitized_value, default_operator="OR", fields=keyword_fields)
             )
 
         return ES_Q("dis_max", queries=keyword_queries)

usaspending_api/search/tests/unit/test_elasticsearch_helpers.py

@@ -1,5 +1,4 @@
-from usaspending_api.search.v2.elasticsearch_helper import es_minimal_sanitize
-from usaspending_api.search.v2.es_sanitization import es_sanitize
+from usaspending_api.search.v2.es_sanitization import es_minimal_sanitize, es_sanitize
 
 
 def test_es_sanitize():
@@ -18,3 +17,62 @@ def test_es_minimal_sanitize():
     test_string = "!-^~/"
     processed_string = es_minimal_sanitize(test_string)
     assert processed_string == r"\!\-\^\~\/"
+
+
+def test_es_minimal_sanitize_length_equality_bypass():
+    """Test that the length-equality bypass vulnerability is fixed.
+
+    Previously, when the number of removed characters equaled the number of
+    escaped characters, the original unsanitized input was returned.
+    This test verifies that sanitization always occurs regardless of length equality.
+    """
+    # Test the exact exploit payload from the vulnerability report.
+    # "[* TO *]!!" has 10 chars, removes 2 brackets (8 chars), escapes 2 chars (10 chars).
+    # This previously bypassed sanitization due to length equality.
+    exploit_payload = "[* TO *]!!"
+    processed = es_minimal_sanitize(exploit_payload)
+    # Should remove brackets and escape special chars.
+    assert "[" not in processed, "Brackets should be removed"
+    assert "]" not in processed, "Brackets should be removed"
+    assert r"\!" in processed, "Exclamation marks should be escaped"
+    assert processed != exploit_payload, "Payload must be sanitized, not returned as-is"
+
+    # Test other balanced payloads that could bypass length check.
+    test_cases = [
+        # (input, should_not_contain, should_contain).
+        ("{test}!!", ["{", "}"], [r"\!"]),
+        ("[range]:", ["[", "]"], [r"\:"]),
+        ("\\test", ["\\"], []),  # Backslash removed, no escaping needed for 'test'.
+        ("{[test]}!&", ["{", "}", "[", "]"], [r"\!", r"\&"]),
+    ]
+
+    for test_input, forbidden_chars, required_escapes in test_cases:
+        result = es_minimal_sanitize(test_input)
+        for forbidden in forbidden_chars:
+            assert forbidden not in result, f"Character '{forbidden}' should be removed from '{test_input}'"
+        for required in required_escapes:
+            assert required in result, f"Escape sequence '{required}' should be present in result of '{test_input}'"
+        assert result != test_input, f"Input '{test_input}' must be sanitized, not returned unchanged"
+
+
+def test_es_minimal_sanitize_removes_lucene_operators():
+    """Test that dangerous Lucene query operators are properly sanitized."""
+    dangerous_inputs = [
+        "[* TO *]",         # Range query.
+        "{a TO z}",         # Range query with braces.
+        "test\\",           # Escape character.
+        "field:[value]",    # Field query with brackets.
+        "test~0.5",         # Fuzzy search (~ should be escaped).
+        "test^2",           # Boost operator (^ should be escaped).
+    ]
+
+    for dangerous_input in dangerous_inputs:
+        result = es_minimal_sanitize(dangerous_input)
+        # Verify brackets and backslashes are removed.
+        assert "[" not in result
+        assert "]" not in result
+        assert "{" not in result
+        assert "}" not in result
+        assert "\\" not in result or result.count("\\") > dangerous_input.count("\\")
+        # Result should differ from input (sanitized).
+        assert result != dangerous_input

usaspending_api/search/v2/es_sanitization.py

@@ -1,10 +1,11 @@
 import logging
 import re
+from typing import Any
 
 logger = logging.getLogger("console")
 
 
-def concat_if_array(data):
+def concat_if_array(data: Any) -> str:
     if isinstance(data, str):
         return data
     else:
@@ -17,7 +18,7 @@ def concat_if_array(data):
             return ""
 
 
-def es_sanitize(input_string):
+def es_sanitize(input_string: str) -> str:
     """Escapes reserved elasticsearch characters and removes when necessary"""
     processed_string = re.sub(r'([|{}()?\\"+\[\]<>])', "", input_string)
     processed_string = re.sub(r"[\-]", r"\-", processed_string)
@@ -35,7 +36,7 @@ def es_sanitize(input_string):
     return processed_string
 
 
-def es_minimal_sanitize(keyword):
+def es_minimal_sanitize(keyword: Any) -> str:
     keyword = concat_if_array(keyword)
     """Remove Lucene special characters and escapes when needed"""
     processed_string = re.sub(r"[{}\[\]\\]", "", keyword)
@@ -50,5 +51,4 @@ def es_minimal_sanitize(keyword):
     if len(processed_string) != len(keyword):
         msg = "Stripped characters from ES keyword search string New: '{}' Original: '{}'"
         logger.info(msg.format(processed_string, keyword))
-        keyword = processed_string
-    return keyword
+    return processed_string