Merge branch 'main' of https://github.com/SigmaHQ/pySigma-backend-splunk

Author: thomaspatzke

sigma/backends/splunk/splunk.py

@@ -235,6 +235,9 @@ class SplunkBackend(TextQueryBackend):
     temporal_aggregation_expression: ClassVar[Dict[str, str]] = {
         "stats": "| bin _time span={timespan}\n| stats dc(event_type) as event_type_count by _time{groupby}",
     }
+    temporal_extended_aggregation_expression: ClassVar[Dict[str, str]] = {
+        "stats": "| bin _time span={timespan}\n| stats values(event_type) as event_types by _time{groupby}",
+    }
 
     timespan_mapping: ClassVar[Dict[str, str]] = {
         "M": "mon",
@@ -253,6 +256,13 @@ class SplunkBackend(TextQueryBackend):
     temporal_condition_expression: ClassVar[Dict[str, str]] = {
         "stats": "| search event_type_count {op} {count}"
     }
+    temporal_extended_condition_expression: ClassVar[dict[str, str]] = {
+        "stats": "| search {extended_condition}"
+    }
+    
+    extended_correlation_condition_rule_reference_expression: ClassVar[dict[str, str]] = {
+        "stats": 'event_types="{ruleid}"'
+    }
 
     def __init__(
         self,

tests/test_backend_splunk_correlations.py

@@ -126,6 +126,67 @@ def test_temporal_correlation_rule_stats_query(splunk_backend):
 
 | search event_type_count >= 2"""]
 
+def test_temporal_extended_correlation_rule_stats_query(splunk_backend):
+    correlation_rule = SigmaCollection.from_yaml(
+        """
+title: Base rule 1
+name: base_rule_1
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value1
+        fieldB: value2
+    condition: selection
+---
+title: Base rule 2
+name: base_rule_2
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value3
+        fieldB: value4
+    condition: selection
+---
+title: Base rule 3
+name: base_rule_3
+status: test
+logsource:
+    category: test
+detection:
+    selection:
+        fieldA: value5
+        fieldB: value6
+    condition: selection
+---
+title: Temporal correlation rule
+status: test
+correlation:
+    type: temporal
+    aliases:
+        field:
+            base_rule_1: fieldC
+            base_rule_2: fieldD
+    group-by:
+        - fieldC
+    condition: base_rule_1 and base_rule_2 and not base_rule_3
+    timespan: 15m
+"""
+    )
+    assert splunk_backend.convert(correlation_rule) == [
+        """| multisearch
+[ search fieldA="value1" fieldB="value2" | eval event_type="base_rule_1" | rename fieldC as field ]
+[ search fieldA="value3" fieldB="value4" | eval event_type="base_rule_2" | rename fieldD as field ]
+[ search fieldA="value5" fieldB="value6" | eval event_type="base_rule_3" ]
+
+| bin _time span=15m
+| stats values(event_type) as event_types by _time fieldC
+
+| search event_types="base_rule_1"   event_types="base_rule_2"   NOT event_types="base_rule_3\""""]
+
 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(