diff --git a/sigma/backends/splunk/splunk.py b/sigma/backends/splunk/splunk.py index 4e4810b..ea454ed 100644 --- a/sigma/backends/splunk/splunk.py +++ b/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 diff --git a/tests/test_backend_splunk.py b/tests/test_backend_splunk.py index ac5bbc7..4604d48 100644 --- a/tests/test_backend_splunk.py +++ b/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 "(?suspicious_command)"\n| eval CommandLineCondition=if(isnotnull(CommandLineMatch), "true", "false")\n| rex field=Image "(?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 "(?suspicious_command)"\n| eval CommandLineCondition=if(isnotnull(CommandLineMatch), "true", "false")\n| rex field=Image "(?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 "(?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" diff --git a/tests/test_backend_splunk_correlations.py b/tests/test_backend_splunk_correlations.py index 0d5e0b2..ad93428 100644 --- a/tests/test_backend_splunk_correlations.py +++ b/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"' + )