From efd22ad05beb1ed6a3d2205f7fa2ad284c3cb0a3 Mon Sep 17 00:00:00 2001 From: Greg Holden Date: Mon, 27 Jul 2026 11:27:02 -0400 Subject: [PATCH] DEV-15453: Updated ES sanitizing functions to handle sanitization bypass vulnerability --- usaspending_api/common/query_with_filters.py | 11 ++-- .../tests/unit/test_elasticsearch_helpers.py | 62 ++++++++++++++++++- usaspending_api/search/v2/es_sanitization.py | 10 +-- 3 files changed, 70 insertions(+), 13 deletions(-) diff --git a/usaspending_api/common/query_with_filters.py b/usaspending_api/common/query_with_filters.py index 8a60677ac8..e31f456313 100644 --- a/usaspending_api/common/query_with_filters.py +++ b/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) diff --git a/usaspending_api/search/tests/unit/test_elasticsearch_helpers.py b/usaspending_api/search/tests/unit/test_elasticsearch_helpers.py index 68d41c426d..71b55b4c39 100644 --- a/usaspending_api/search/tests/unit/test_elasticsearch_helpers.py +++ b/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 diff --git a/usaspending_api/search/v2/es_sanitization.py b/usaspending_api/search/v2/es_sanitization.py index 9344ae5b0f..9dfc6384ab 100644 --- a/usaspending_api/search/v2/es_sanitization.py +++ b/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