Fix deferred expressions in correlation rules

sigma/backends/splunk/splunk.py

@@ -2,6 +2,7 @@
 import re
 from sigma.conversion.state import ConversionState
 from sigma.modifiers import SigmaRegularExpression
+from sigma.correlations import SigmaCorrelationRule
 from sigma.rule import SigmaRule, SigmaDetection
 from sigma.conversion.base import TextQueryBackend, DeferredQueryExpression
 from sigma.conversion.deferred import DeferredTextQueryExpression
@@ -309,7 +310,7 @@ def convert_condition_field_eq_field(
 
     def finish_query(
         self,
-        rule: SigmaRule,
+        rule: Union[SigmaRule, SigmaCorrelationRule],
         query: Union[str, DeferredQueryExpression],
         state: ConversionState,
     ) -> Union[str, DeferredQueryExpression]:
@@ -341,13 +342,22 @@ def finish_query(
         return super().finish_query(rule, query, state)
 
     def finalize_query_default(
-        self, rule: SigmaRule, query: str, index: int, state: ConversionState
+        self,
+        rule: Union[SigmaRule, SigmaCorrelationRule],
+        query: str,
+        index: int,
+        state: ConversionState,
     ) -> str:
-        table_fields = " | table " + ",".join(rule.fields) if rule.fields else ""
-        return query + table_fields
+        if isinstance(rule, SigmaRule) and rule.fields:
+            return query + " | table " + ",".join(rule.fields)
+        return query
 
     def finalize_query_savedsearches(
-        self, rule: SigmaRule, query: str, index: int, state: ConversionState
+        self,
+        rule: Union[SigmaRule, SigmaCorrelationRule],
+        query: str,
+        index: int,
+        state: ConversionState,
     ) -> str:
         clean_title = rule.title.translate(
             {ord(c): None for c in "[]"}
@@ -357,7 +367,9 @@ def finalize_query_savedsearches(
             rule.description.strip() if rule.description else ""
         )
         query_settings["search"] = query + (
-            "\n| table " + ",".join(rule.fields) if rule.fields else ""
+            "\n| table " + ",".join(rule.fields)
+            if isinstance(rule, SigmaRule) and rule.fields
+            else ""
         )
 
         return f"\n[{clean_title}]" + self._generate_settings(query_settings)

tests/test_backend_splunk_correlations.py

@@ -124,4 +124,192 @@ def test_temporal_correlation_rule_stats_query(splunk_backend):
 | bin _time span=15m
 | stats dc(event_type) as event_type_count by _time fieldC
 
-| search event_type_count >= 2"""]
+| search event_type_count >= 2"""]
+
+def test_event_count_correlation_rule_with_regex_deferred(splunk_backend):
+    """Test that deferred regex expressions are included in correlation sub-queries."""
+    correlation_rule = SigmaCollection.from_yaml(
+        """
+title: Base rule
+name: base_rule
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value1
+        fieldB: value2
+    condition: selection
+---
+title: Base rule 2
+name: base_rule2
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value1
+        fieldB|re: value2
+    condition: selection
+---
+title: Multiple occurrences of base event
+status: test
+correlation:
+    type: event_count
+    rules:
+        - base_rule
+        - base_rule2
+    group-by:
+        - fieldC
+        - fieldD
+    timespan: 15m
+    condition:
+        gte: 10
+            """
+    )
+    assert splunk_backend.convert(correlation_rule) == [
+        """| multisearch
+[ search fieldA="value1" fieldB="value2" | eval event_type="base_rule" ]
+[ search fieldA="value1"
+| regex fieldB="value2" | eval event_type="base_rule2" ]
+
+| bin _time span=15m
+| stats count as event_count by _time fieldC fieldD
+
+| search event_count >= 10"""
+    ]
+
+def test_single_rule_correlation_with_regex_deferred(splunk_backend):
+    """Test that deferred regex expressions are included in single-rule correlation queries."""
+    correlation_rule = SigmaCollection.from_yaml(
+        """
+title: Base rule
+name: base_rule
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value1
+        fieldB|re: value2
+    condition: selection
+---
+title: Single rule correlation
+status: test
+correlation:
+    type: event_count
+    rules:
+        - base_rule
+    group-by:
+        - fieldC
+    timespan: 15m
+    condition:
+        gte: 10
+            """
+    )
+    assert splunk_backend.convert(correlation_rule) == [
+        """fieldA="value1"
+| regex fieldB="value2"
+
+| bin _time span=15m
+| stats count as event_count by _time fieldC
+
+| search event_count >= 10"""
+    ]
+
+def test_event_count_correlation_rule_with_regex_or_deferred(splunk_backend):
+    """Test that deferred OR regex expressions (rex) are included in correlation sub-queries."""
+    correlation_rule = SigmaCollection.from_yaml(
+        """
+title: Base rule
+name: base_rule
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value1
+        fieldB: value2
+    condition: selection
+---
+title: Base rule 2
+name: base_rule2
+status: test
+logsource:
+    category: test
+detection:
+    sel1:
+        fieldA|re: value1
+    sel2:
+        fieldB|re: value2
+    condition: sel1 or sel2
+---
+title: Correlation
+status: test
+correlation:
+    type: event_count
+    rules:
+        - base_rule
+        - base_rule2
+    group-by:
+        - fieldC
+    timespan: 15m
+    condition:
+        gte: 10
+            """
+    )
+    assert splunk_backend.convert(correlation_rule) == [
+        """| multisearch
+[ search fieldA="value1" fieldB="value2" | eval event_type="base_rule" ]
+[ search \n| rex field=fieldA "(?<fieldAMatch>value1)"
+| eval fieldACondition=if(isnotnull(fieldAMatch), "true", "false")
+| rex field=fieldB "(?<fieldBMatch>value2)"
+| eval fieldBCondition=if(isnotnull(fieldBMatch), "true", "false")
+| search fieldACondition="true" OR fieldBCondition="true" | eval event_type="base_rule2" ]
+
+| bin _time span=15m
+| stats count as event_count by _time fieldC
+
+| search event_count >= 10"""
+    ]
+
+def test_correlation_rule_subrule_fields_not_in_output(splunk_backend):
+    """Test that sub-rule fields (| table) are not included in correlation queries."""
+    correlation_rule = SigmaCollection.from_yaml(
+        """
+title: Base rule
+name: base_rule
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value1
+    condition: selection
+fields:
+    - fieldA
+    - fieldB
+---
+title: Correlation
+status: test
+correlation:
+    type: event_count
+    rules:
+        - base_rule
+    group-by:
+        - fieldC
+    timespan: 15m
+    condition:
+        gte: 10
+            """
+    )
+    result = splunk_backend.convert(correlation_rule)
+    assert "table" not in result[0]
+    assert result == [
+        """fieldA="value1"
+
+| bin _time span=15m
+| stats count as event_count by _time fieldC
+
+| search event_count >= 10"""
+    ]