[high] Parenthesise AND groups under OR (Splunk search evaluates OR before AND)

sigma/backends/splunk/splunk.py

@@ -2,7 +2,11 @@
 import re
 from sigma.conversion.state import ConversionState
 from sigma.modifiers import SigmaRegularExpression
-from sigma.correlations import SigmaCorrelationRule
+from sigma.correlations import (
+    SigmaCorrelationRule,
+    CorrelationConditionAND,
+    CorrelationConditionOR,
+)
 from sigma.rule import SigmaRule, SigmaDetection
 from sigma.conversion.base import TextQueryBackend, DeferredQueryExpression
 from sigma.conversion.deferred import DeferredTextQueryExpression
@@ -315,6 +319,22 @@ def _generate_settings(settings):
             )  # cannot use \ in f-strings
         return output
 
+    def compare_precedence(self, outer, inner) -> bool:
+        """
+        Always group an AND nested inside an OR (and, via the precedence tuple, an OR nested
+        inside an AND).
+
+        The Splunk `search` command documents its Boolean evaluation order as parentheses,
+        NOT, OR, AND, i.e. OR binds tighter than the implicit AND, which is the opposite of
+        the order used by `eval`/`where`. Grouping every mixed AND/OR nesting makes the
+        emitted query independent of which of the two orders applies.
+        """
+        if isinstance(outer, (ConditionOR, CorrelationConditionOR)) and isinstance(
+            inner, (ConditionAND, CorrelationConditionAND)
+        ):
+            return False
+        return super().compare_precedence(outer, inner)
+
     def convert_condition_field_eq_val_re(
         self,
         cond: ConditionFieldEqualsValueExpression,
@@ -394,6 +414,39 @@ def finish_query(
 
         return super().finish_query(rule, query, state)
 
+    @staticmethod
+    def _has_top_level_or(query: str) -> bool:
+        """Return True if query contains an OR token outside of quotes and parentheses."""
+        depth = 0
+        in_quote = False
+        escaped = False
+        token = ""
+        for c in query + " ":
+            if in_quote:
+                if escaped:
+                    escaped = False
+                elif c == "\\":
+                    escaped = True
+                elif c == '"':
+                    in_quote = False
+                continue
+            if c == '"':
+                in_quote = True
+                token = ""
+            elif c == "(":
+                depth += 1
+                token = ""
+            elif c == ")":
+                depth -= 1
+                token = ""
+            elif c.isspace():
+                if token == "OR" and depth == 0:
+                    return True
+                token = ""
+            else:
+                token += c
+        return False
+
     def finalize_query_default(
         self,
         rule: Union[SigmaRule, SigmaCorrelationRule],
@@ -426,11 +479,12 @@ def finalize_query_default(
                     break
 
             remaining_query = search_query[pos:]
-            # If the remaining query starts with OR we would split a disjunction
-            # across the pipeline boundary, making the query semantically wrong.
-            # In that case, skip hoisting so the full disjunction stays intact in
-            # the trailing | search stage.
-            if prefix_parts and not remaining_query.lstrip().startswith("OR"):
+            # Hoisting a term in front of the pipeline makes it a conjunct of the
+            # whole search. That is only equivalent if the search expression is a
+            # top-level conjunction; if it contains an OR outside of parentheses,
+            # hoisting would move a term out of a disjunction branch, so the full
+            # expression stays intact in the trailing | search stage.
+            if prefix_parts and not self._has_top_level_or(search_query):
                 prefix = " ".join(prefix_parts)
                 query = (
                     prefix

tests/test_backend_splunk.py

@@ -101,13 +101,13 @@ def test_splunk_or_and_expression(splunk_backend: SplunkBackend):
         """
     )
     assert splunk_backend.convert(rule) == [
-        'fieldA="valueA1" fieldB="valueB1" OR fieldA="valueA2" fieldB="valueB2"'
+        '(fieldA="valueA1" fieldB="valueB1") OR (fieldA="valueA2" fieldB="valueB2")'
     ]
 
 
 def test_splunk_or_nested_in_and_expression(splunk_backend: SplunkBackend):
-    """An OR nested inside an AND must be parenthesized: implicit AND (juxtaposition)
-    binds tighter than OR in SPL, so an ungrouped OR would silently widen the query."""
+    """An OR nested inside an AND must be parenthesized, otherwise the ungrouped OR
+    changes the meaning of the query. See also the AND-nested-in-OR test below."""
     rule = SigmaCollection.from_yaml(
         """
             title: Test
@@ -129,6 +129,58 @@ def test_splunk_or_nested_in_and_expression(splunk_backend: SplunkBackend):
     ]
 
 
+def test_splunk_and_nested_in_or_expression(splunk_backend: SplunkBackend):
+    """An AND nested inside an OR must be parenthesized as well. The Splunk search
+    command documents its evaluation order as parentheses, NOT, OR, AND, so without
+    grouping `A B OR C D` would be read as `A AND (B OR C) AND D`."""
+    rule = SigmaCollection.from_yaml(
+        """
+            title: Test
+            status: test
+            logsource:
+                product: azure
+                service: signinlogs
+            detection:
+                selection_50074:
+                    ResultType: 50074
+                    ResultDescription|contains: 'Strong Auth required'
+                selection_500121:
+                    ResultType: 500121
+                    ResultDescription|contains: 'Authentication failed during strong authentication request'
+                condition: 1 of selection_*
+        """
+    )
+    assert splunk_backend.convert(rule) == [
+        '(ResultType=50074 ResultDescription="*Strong Auth required*") OR '
+        '(ResultType=500121 ResultDescription="*Authentication failed during strong authentication request*")'
+    ]
+
+
+def test_splunk_mixed_and_or_nesting_always_grouped(splunk_backend: SplunkBackend):
+    rule = SigmaCollection.from_yaml(
+        """
+            title: Test
+            status: test
+            logsource:
+                category: test_category
+                product: test_product
+            detection:
+                a:
+                    fieldA: valueA
+                b:
+                    fieldB: valueB
+                c:
+                    fieldC: valueC
+                d:
+                    fieldD: valueD
+                condition: (a and (b or c)) or d
+        """
+    )
+    assert splunk_backend.convert(rule) == [
+        '(fieldA="valueA" (fieldB="valueB" OR fieldC="valueC")) OR fieldD="valueD"'
+    ]
+
+
 def test_splunk_in_expression(splunk_backend: SplunkBackend):
     assert (
         splunk_backend.convert(
@@ -326,7 +378,7 @@ def test_splunk_regex_query_explicit_or_with_add_condition():
     )
 
     assert splunk_backend.convert(collection) == [
-        'index="test" source="test"\n| rex field=CommandLine "(?<CommandLineMatch>suspicious_command)"\n| eval CommandLineCondition=if(isnotnull(CommandLineMatch), "true", "false")\n| rex field=Image "(?<ImageMatch>suspicious_command)"\n| eval ImageCondition=if(isnotnull(ImageMatch), "true", "false")\n| search (EventID=4688 CommandLineCondition="true" OR ImageCondition="true")'
+        'index="test" source="test"\n| rex field=CommandLine "(?<CommandLineMatch>suspicious_command)"\n| eval CommandLineCondition=if(isnotnull(CommandLineMatch), "true", "false")\n| rex field=Image "(?<ImageMatch>suspicious_command)"\n| eval ImageCondition=if(isnotnull(ImageMatch), "true", "false")\n| search ((EventID=4688 CommandLineCondition="true") OR ImageCondition="true")'
     ]
 
 
@@ -363,6 +415,66 @@ def test_splunk_disjunction_with_deferred_regex_no_search_or():
     assert 'NOT fieldBCondition="true"' in query
 
 
+_AND_REGEX_OR_RULE = """
+    title: Test
+    status: test
+    logsource:
+        category: test_category
+        product: test_product
+    detection:
+        a:
+            fieldA: valueA
+        b:
+            fieldB|re: 'foo.*bar'
+        c:
+            fieldC: valueC
+        condition: (a and b) or c
+"""
+
+
+def test_splunk_no_hoisting_out_of_disjunction(splunk_backend: SplunkBackend):
+    """A term of one OR branch must not be moved in front of the rex/eval pipeline,
+    where it would become a conjunct of the whole disjunction."""
+    assert splunk_backend.convert(SigmaCollection.from_yaml(_AND_REGEX_OR_RULE)) == [
+        '\n| rex field=fieldB "(?<fieldBMatch>foo.*bar)"'
+        '\n| eval fieldBCondition=if(isnotnull(fieldBMatch), "true", "false")'
+        '\n| search (fieldA="valueA" fieldBCondition="true") OR fieldC="valueC"'
+    ]
+
+
+def test_splunk_no_hoisting_out_of_ungrouped_disjunction():
+    """Hoisting must also be skipped if the disjunction is not parenthesized. The
+    subclass disables the AND-in-OR grouping to exercise the hoisting guard alone."""
+    from sigma.conversion.base import TextQueryBackend
+
+    class UngroupedSplunkBackend(SplunkBackend):
+        compare_precedence = TextQueryBackend.compare_precedence
+
+    query = UngroupedSplunkBackend().convert(
+        SigmaCollection.from_yaml(_AND_REGEX_OR_RULE)
+    )[0]
+    assert query.startswith("\n| rex field=fieldB ")
+    assert query.endswith(
+        '\n| search fieldA="valueA" fieldBCondition="true" OR fieldC="valueC"'
+    )
+
+
+@pytest.mark.parametrize(
+    "query,expected",
+    [
+        ('a="1" b="2"', False),
+        ('a="1" OR b="2"', True),
+        ('a="1" (b="2" OR c="3")', False),
+        ('(a="1" b="2") OR c="3"', True),
+        ('a="x OR y" b="2"', False),
+        ('a="x \\" OR y" b="2"', False),
+        ('a IN ("1", "2") NOT b="3"', False),
+    ],
+)
+def test_splunk_has_top_level_or(query, expected):
+    assert SplunkBackend._has_top_level_or(query) is expected
+
+
 def test_splunk_regex_group_name_is_capped_for_long_fields():
     SplunkDeferredORRegularExpression.reset()
     field = "msg_normalized_header_subject"

tests/test_backend_splunk_correlations.py

@@ -374,3 +374,54 @@ def test_correlation_rule_subrule_fields_in_stats_output(splunk_backend):
 
 | search event_count >= 10"""
     ]
+
+
+def test_temporal_extended_correlation_and_nested_in_or(splunk_backend):
+    """The extended condition is evaluated by the search command, so an AND nested in
+    an OR must be grouped there too."""
+    correlation_rule = SigmaCollection.from_yaml(
+        """
+title: Base rule 1
+name: base_rule_1
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value1
+    condition: selection
+---
+title: Base rule 2
+name: base_rule_2
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value2
+    condition: selection
+---
+title: Base rule 3
+name: base_rule_3
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value3
+    condition: selection
+---
+title: Temporal correlation rule
+status: test
+correlation:
+    type: temporal
+    group-by:
+        - fieldC
+    condition: (base_rule_1 and base_rule_2) or base_rule_3
+    timespan: 15m
+"""
+    )
+    query = splunk_backend.convert(correlation_rule)[0]
+    assert " ".join(query.split()).endswith(
+        '| search (event_types="base_rule_1" event_types="base_rule_2") OR event_types="base_rule_3"'
+    )