chore(iast): query string redaction synchronization (#15366)

Implements query string and vulnerability evidence redaction
synchronization to address inconsistencies where query strings could be
visible in span tags but redacted in IAST evidence (or vice
versa). This PR adds pattern harmonization by making the IAST evidence
redactor aware of the query string obfuscation pattern
(`DD_TRACE_OBFUSCATION_QUERY_STRING_REGEXP`). When sources originate
from query strings (`OriginType.QUERY`), they are now checked against
the same pattern used at the span level, ensuring consistent redaction
across both systems.


APPSEC-52879

Author: avara1986

ddtrace/appsec/_iast/_evidence_redaction/_sensitive_handler.py

@@ -2,8 +2,10 @@
 import string
 
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.settings._config import config
 from ddtrace.internal.settings.asm import config as asm_config
 
+from .._taint_tracking import OriginType
 from .._utils import _get_source_index
 from ..constants import VULN_CMDI
 from ..constants import VULN_CODE_INJECTION
@@ -41,6 +43,8 @@ class SensitiveHandler:
     def __init__(self):
         self._name_pattern = re.compile(asm_config._iast_redaction_name_pattern, re.IGNORECASE | re.MULTILINE)
         self._value_pattern = re.compile(asm_config._iast_redaction_value_pattern, re.IGNORECASE | re.MULTILINE)
+        # Query string obfuscation pattern for synchronization with span-level redaction
+        self._query_string_pattern = config._obfuscation_query_string_pattern
 
         self._sensitive_analyzers = {
             VULN_CMDI: command_injection_sensitive_analyzer,
@@ -131,6 +135,21 @@ def is_sensible_value(self, value):
         """
         return bool(self._value_pattern.search(value))
 
+    def is_query_string_source(self, source):
+        """
+        Checks if a source originates from a query string.
+
+        Args:
+        - source: The source to check.
+
+        Returns:
+        - bool: True if the source is from a query string, False otherwise.
+        """
+        try:
+            return source is not None and hasattr(source, "origin") and source.origin == OriginType.QUERY
+        except Exception:
+            return False
+
     def is_sensible_source(self, source):
         """
         Checks if a source is sensible.
@@ -141,11 +160,22 @@ def is_sensible_source(self, source):
         Returns:
         - bool: True if the source is sensible, False otherwise.
         """
-        return (
-            source is not None
-            and source.value is not None
-            and (self.is_sensible_name(source.name) or self.is_sensible_value(source.value))
-        )
+        if source is None or source.value is None:
+            return False
+
+        # For query string sources, check against the query string obfuscation pattern
+        # to maintain synchronization with span-level redaction
+        if self.is_query_string_source(source) and self._query_string_pattern is not None:
+            try:
+                # Convert pattern to string for matching (pattern is in bytes, source value is string)
+                value_bytes = source.value if isinstance(source.value, bytes) else source.value.encode("utf-8")
+                if self._query_string_pattern.search(value_bytes):
+                    return True
+            except Exception:
+                log.debug("Error checking query string pattern against source", exc_info=True)
+
+        # Standard IAST redaction patterns
+        return self.is_sensible_name(source.name) or self.is_sensible_value(source.value)
 
     def scrub_evidence(self, vulnerability_type, evidence, tainted_ranges, sources):
         """
@@ -166,7 +196,10 @@ def scrub_evidence(self, vulnerability_type, evidence, tainted_ranges, sources):
                 if not evidence.value:
                     log.debug("No evidence value found in evidence %s", evidence)
                     return None
-                sensitive_ranges = sensitive_analyzer(evidence, self._name_pattern, self._value_pattern)
+                # Pass query string pattern for synchronization with span-level redaction
+                sensitive_ranges = sensitive_analyzer(
+                    evidence, self._name_pattern, self._value_pattern, self._query_string_pattern
+                )
                 return self.to_redacted_json(evidence.value, sensitive_ranges, tainted_ranges, sources)
         return None
 

ddtrace/appsec/_iast/_evidence_redaction/command_injection_sensitive_analyzer.py

@@ -10,7 +10,19 @@
 pattern = re.compile(COMMAND_PATTERN, re.IGNORECASE | re.MULTILINE)
 
 
-def command_injection_sensitive_analyzer(evidence, name_pattern=None, value_pattern=None):
+def command_injection_sensitive_analyzer(evidence, name_pattern=None, value_pattern=None, query_string_pattern=None):
+    """
+    Command injection sensitive analyzer for evidence redaction.
+
+    Args:
+    - evidence: The evidence to analyze
+    - name_pattern: Pattern for matching sensitive names (unused in command injection analyzer)
+    - value_pattern: Pattern for matching sensitive values (unused in command injection analyzer)
+    - query_string_pattern: Query string obfuscation pattern (unused in command injection analyzer)
+
+    Returns:
+    - list: List of sensitive ranges to redact
+    """
     regex_result = pattern.search(evidence.value)
     if regex_result and len(regex_result.groups()) > 0:
         start = regex_result.start(1)

ddtrace/appsec/_iast/_evidence_redaction/default_sensitive_analyzer.py

@@ -4,7 +4,19 @@
 log = get_logger(__name__)
 
 
-def default_sensitive_analyzer(evidence, name_pattern, value_pattern):
+def default_sensitive_analyzer(evidence, name_pattern, value_pattern, query_string_pattern=None):
+    """
+    Default sensitive analyzer for evidence redaction.
+
+    Args:
+    - evidence: The evidence to analyze
+    - name_pattern: Pattern for matching sensitive names
+    - value_pattern: Pattern for matching sensitive values
+    - query_string_pattern: Query string obfuscation pattern (unused in default analyzer)
+
+    Returns:
+    - list: List of sensitive ranges to redact
+    """
     if name_pattern.search(evidence.value) or value_pattern.search(evidence.value):
         return [{"start": 0, "end": len(evidence.value)}]
 

ddtrace/appsec/_iast/_evidence_redaction/header_injection_sensitive_analyzer.py

@@ -5,7 +5,19 @@
 log = get_logger(__name__)
 
 
-def header_injection_sensitive_analyzer(evidence, name_pattern, value_pattern):
+def header_injection_sensitive_analyzer(evidence, name_pattern, value_pattern, query_string_pattern=None):
+    """
+    Header injection sensitive analyzer for evidence redaction.
+
+    Args:
+    - evidence: The evidence to analyze
+    - name_pattern: Pattern for matching sensitive names
+    - value_pattern: Pattern for matching sensitive values
+    - query_string_pattern: Query string obfuscation pattern (unused in header injection analyzer)
+
+    Returns:
+    - list: List of sensitive ranges to redact
+    """
     evidence_value = evidence.value
     sections = evidence_value.split(HEADER_NAME_VALUE_SEPARATOR)
     header_name = sections[0]

ddtrace/appsec/_iast/_evidence_redaction/sql_sensitive_analyzer.py

@@ -41,7 +41,19 @@
 patterns[DBAPI_MYSQLDB] = patterns[DBAPI_MYSQL]
 
 
-def sql_sensitive_analyzer(evidence, name_pattern, value_pattern):
+def sql_sensitive_analyzer(evidence, name_pattern, value_pattern, query_string_pattern=None):
+    """
+    SQL sensitive analyzer for evidence redaction.
+
+    Args:
+    - evidence: The evidence to analyze
+    - name_pattern: Pattern for matching sensitive names
+    - value_pattern: Pattern for matching sensitive values
+    - query_string_pattern: Query string obfuscation pattern (unused in SQL analyzer)
+
+    Returns:
+    - list: List of sensitive ranges to redact
+    """
     pattern = patterns.get(evidence.dialect, patterns[DBAPI_MYSQL])
     tokens = []
 

ddtrace/appsec/_iast/_evidence_redaction/url_sensitive_analyzer.py

@@ -29,8 +29,52 @@ def find_query_fragment(ranges, evidence):
         regex_result = QUERY_FRAGMENT_PATTERN.search(evidence.value, regex_result.end())
 
 
-def url_sensitive_analyzer(evidence, name_pattern=None, value_pattern=None):
+def find_query_string_matches(ranges, evidence, query_string_pattern):
+    """
+    Find sensitive data in query string using the query string obfuscation pattern.
+    This ensures synchronization with span-level query string redaction.
+    """
+    if query_string_pattern is None:
+        return
+
+    try:
+        # Extract query string portion from URL
+        if "?" not in evidence.value:
+            return
+
+        # Find the query string part
+        query_start = evidence.value.find("?")
+        query_end = evidence.value.find("#") if "#" in evidence.value else len(evidence.value)
+        query_string = evidence.value[query_start:query_end]
+
+        # Convert to bytes for pattern matching (query string pattern is in bytes)
+        query_bytes = query_string if isinstance(query_string, bytes) else query_string.encode("utf-8")
+
+        # Find all matches
+        for match in query_string_pattern.finditer(query_bytes):
+            start = query_start + match.start()
+            end = query_start + match.end()
+            ranges.append({"start": start, "end": end})
+    except Exception:
+        log.debug("Error applying query string pattern to URL evidence", exc_info=True)
+
+
+def url_sensitive_analyzer(evidence, name_pattern=None, value_pattern=None, query_string_pattern=None):
+    """
+    Analyzes URL evidence for sensitive information.
+
+    Args:
+    - evidence: The evidence to analyze
+    - name_pattern: Pattern for matching sensitive names
+    - value_pattern: Pattern for matching sensitive values
+    - query_string_pattern: Pattern for matching sensitive query strings (for synchronization)
+
+    Returns:
+    - list: List of sensitive ranges to redact
+    """
     ranges = []
     find_authority(ranges, evidence)
     find_query_fragment(ranges, evidence)
+    # Apply query string pattern for synchronization with span-level redaction
+    find_query_string_matches(ranges, evidence, query_string_pattern)
     return ranges