diff --git a/README.md b/README.md index 28e3a6c..aba9f27 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,24 @@ representation in the Sigma standard: * `tstats_aggregation`: a raw aggregation string appended verbatim after `count` (advanced use). When set, it takes precedence over `tstats_aggregations`. +## Correlation methods + +Correlation rules (`event_count`, `value_count`, `temporal`) can be converted with two +methods, selected with the `correlation_method` argument of `convert()` (sigma-cli: +`--correlation-method`): + +* `stats` (default): `| bin _time span= | stats ... by _time ...`. The timespan is a + fixed bucket, so events that belong together but fall into two adjacent buckets (e.g. at + 10:04:59 and 10:05:01 with `timespan: 5m`) are not correlated. +* `streamstats`: `| sort 0 _time | streamstats time_window= ...`. This is the sliding + window of the Sigma correlation specification: every event is annotated with the aggregate of + the events of its group in the preceding timespan, and every event whose window matches the + condition is returned. It needs a time-ordered result set (hence the `sort`) and is subject to + the `max_stream_window` limit in `limits.conf`. Lower-bound conditions (`gt`, `gte`) and + positive rule references are evaluated exactly; upper-bound conditions and negated rule + references in extended temporal conditions are evaluated on the window that ends at each + event. + This backend is currently maintained by: * [Thomas Patzke](https://github.com/thomaspatzke/) \ No newline at end of file diff --git a/sigma/backends/splunk/splunk.py b/sigma/backends/splunk/splunk.py index ea454ed..310502a 100644 --- a/sigma/backends/splunk/splunk.py +++ b/sigma/backends/splunk/splunk.py @@ -211,11 +211,15 @@ class SplunkBackend(TextQueryBackend): # Correlations correlation_methods: ClassVar[Dict[str, str]] = { "stats": "Correlation using stats command (more efficient, static time window)", - # "transaction": "Correlation using transaction command (less efficient, sliding time window", + "streamstats": ( + "Correlation using streamstats command (sliding time window, " + "emits every event whose trailing window matches the condition)" + ), } default_correlation_method: ClassVar[str] = "stats" default_correlation_query: ClassVar[str] = { - "stats": "{search}\n\n{aggregate}\n\n{condition}" + "stats": "{search}\n\n{aggregate}\n\n{condition}", + "streamstats": "{search}\n\n{aggregate}\n\n{condition}", } correlation_search_single_rule_expression: ClassVar[str] = "{query}" @@ -230,50 +234,79 @@ class SplunkBackend(TextQueryBackend): ) correlation_search_field_normalization_expression_joiner: ClassVar[str] = "" + # The "stats" method buckets events into fixed, non-overlapping windows. The + # "streamstats" method implements the sliding window of the Sigma correlation + # specification: for every event, the aggregate covers the events of the same + # group within the preceding timespan. streamstats time_window requires the + # events to be ordered by time, so they are sorted explicitly (multisearch does + # not guarantee a time-ordered output). event_count_aggregation_expression: ClassVar[Dict[str, str]] = { "stats": "| bin _time span={timespan}\n| stats count as event_count{fields} by _time{groupby}", + "streamstats": "| sort 0 _time\n| streamstats time_window={timespan} count as event_count{fields}{groupby}", } value_count_aggregation_expression: ClassVar[Dict[str, str]] = { "stats": "| bin _time span={timespan}\n| stats dc({field}) as value_count{fields} by _time{groupby}", + "streamstats": "| sort 0 _time\n| streamstats time_window={timespan} dc({field}) as value_count{fields}{groupby}", } temporal_aggregation_expression: ClassVar[Dict[str, str]] = { "stats": "| bin _time span={timespan}\n| stats dc(event_type) as event_type_count{fields} by _time{groupby}", + "streamstats": "| sort 0 _time\n| streamstats time_window={timespan} dc(event_type) as event_type_count{fields}{groupby}", } temporal_extended_aggregation_expression: ClassVar[Dict[str, str]] = { "stats": "| bin _time span={timespan}\n| stats values(event_type) as event_types{fields} by _time{groupby}", + "streamstats": "| sort 0 _time\n| streamstats time_window={timespan} values(event_type) as event_types{fields}{groupby}", } - correlation_fields_expression: ClassVar[Dict[str, str]] = {"stats": "{fields}"} + correlation_fields_expression: ClassVar[Dict[str, str]] = { + "stats": "{fields}", + "streamstats": "{fields}", + } correlation_fields_field_expression: ClassVar[Dict[str, str]] = { - "stats": " values({field}) as {field}" + "stats": " values({field}) as {field}", + "streamstats": " values({field}) as {field}", } correlation_fields_field_expression_joiner: ClassVar[Dict[str, str]] = { - "stats": "" + "stats": "", + "streamstats": "", } timespan_mapping: ClassVar[Dict[str, str]] = { "M": "mon", } - groupby_expression: ClassVar[Dict[str, str]] = {"stats": " {fields}"} - groupby_field_expression: ClassVar[Dict[str, str]] = {"stats": "{field}"} - groupby_field_expression_joiner: ClassVar[Dict[str, str]] = {"stats": " "} + groupby_expression: ClassVar[Dict[str, str]] = { + "stats": " {fields}", + "streamstats": " by {fields}", + } + groupby_field_expression: ClassVar[Dict[str, str]] = { + "stats": "{field}", + "streamstats": "{field}", + } + groupby_field_expression_joiner: ClassVar[Dict[str, str]] = { + "stats": " ", + "streamstats": " ", + } event_count_condition_expression: ClassVar[Dict[str, str]] = { - "stats": "| search event_count {op} {count}" + "stats": "| search event_count {op} {count}", + "streamstats": "| search event_count {op} {count}", } value_count_condition_expression: ClassVar[Dict[str, str]] = { - "stats": "| search value_count {op} {count}" + "stats": "| search value_count {op} {count}", + "streamstats": "| search value_count {op} {count}", } temporal_condition_expression: ClassVar[Dict[str, str]] = { - "stats": "| search event_type_count {op} {count}" + "stats": "| search event_type_count {op} {count}", + "streamstats": "| search event_type_count {op} {count}", } temporal_extended_condition_expression: ClassVar[dict[str, str]] = { - "stats": "| search {extended_condition}" + "stats": "| search {extended_condition}", + "streamstats": "| search {extended_condition}", } extended_correlation_condition_rule_reference_expression: ClassVar[dict[str, str]] = { - "stats": 'event_types="{ruleid}"' + "stats": 'event_types="{ruleid}"', + "streamstats": 'event_types="{ruleid}"', } # Splunk stats/tstats functions allowed in the `tstats_aggregations` data model setting. diff --git a/tests/test_backend_splunk_correlations.py b/tests/test_backend_splunk_correlations.py index ad93428..a4398ae 100644 --- a/tests/test_backend_splunk_correlations.py +++ b/tests/test_backend_splunk_correlations.py @@ -425,3 +425,272 @@ def test_temporal_extended_correlation_and_nested_in_or(splunk_backend): assert " ".join(query.split()).endswith( '| search (event_types="base_rule_1" event_types="base_rule_2") OR event_types="base_rule_3"' ) + + +streamstats_base_rule = """ +title: Base rule 1 +name: base_rule_1 +status: test +logsource: + category: test +detection: + selection: + fieldA: value1 + fieldB: value2 + condition: selection +--- +""" + +streamstats_base_rules = ( + streamstats_base_rule + + """ +title: Base rule 2 +name: base_rule_2 +status: test +logsource: + category: test +detection: + selection: + fieldA: value3 + fieldB: value4 + condition: selection +--- +""" +) + + +def test_correlation_methods_default_is_stats(splunk_backend): + assert splunk_backend.default_correlation_method == "stats" + assert set(splunk_backend.correlation_methods) == {"stats", "streamstats"} + + +def test_event_count_correlation_rule_streamstats_query(splunk_backend): + correlation_rule = SigmaCollection.from_yaml( + streamstats_base_rule + + """ +title: Multiple occurrences of base event +status: test +correlation: + type: event_count + rules: + - base_rule_1 + group-by: + - fieldC + - fieldD + timespan: 15m + condition: + gte: 10 +""" + ) + assert splunk_backend.convert( + correlation_rule, correlation_method="streamstats" + ) == [ + """fieldA="value1" fieldB="value2" + +| sort 0 _time +| streamstats time_window=15m count as event_count by fieldC fieldD + +| search event_count >= 10""" + ] + + +def test_event_count_correlation_rule_streamstats_query_without_group_by( + splunk_backend, +): + correlation_rule = SigmaCollection.from_yaml( + streamstats_base_rule + + """ +title: Multiple occurrences of base event +status: test +correlation: + type: event_count + rules: + - base_rule_1 + timespan: 1h + condition: + gte: 10 +""" + ) + assert splunk_backend.convert( + correlation_rule, correlation_method="streamstats" + ) == [ + """fieldA="value1" fieldB="value2" + +| sort 0 _time +| streamstats time_window=1h count as event_count + +| search event_count >= 10""" + ] + + +def test_value_count_correlation_rule_streamstats_query(splunk_backend): + correlation_rule = SigmaCollection.from_yaml( + streamstats_base_rule + + """ +title: Many distinct values +status: test +correlation: + type: value_count + rules: + - base_rule_1 + group-by: + - fieldC + timespan: 15m + condition: + gte: 10 + field: fieldD +""" + ) + assert splunk_backend.convert( + correlation_rule, correlation_method="streamstats" + ) == [ + """fieldA="value1" fieldB="value2" + +| sort 0 _time +| streamstats time_window=15m dc(fieldD) as value_count by fieldC + +| search value_count >= 10""" + ] + + +def test_temporal_correlation_rule_streamstats_query(splunk_backend): + correlation_rule = SigmaCollection.from_yaml( + streamstats_base_rules + + """ +title: Temporal correlation rule +status: test +correlation: + type: temporal + rules: + - base_rule_1 + - base_rule_2 + aliases: + field: + base_rule_1: fieldC + base_rule_2: fieldD + group-by: + - fieldC + timespan: 15m +""" + ) + assert splunk_backend.convert( + correlation_rule, correlation_method="streamstats" + ) == [ + """| 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 ] + +| sort 0 _time +| streamstats time_window=15m dc(event_type) as event_type_count by fieldC + +| search event_type_count >= 2""" + ] + + +def test_temporal_extended_correlation_rule_streamstats_query(splunk_backend): + correlation_rule = SigmaCollection.from_yaml( + streamstats_base_rules + + """ +title: Temporal correlation rule +status: test +correlation: + type: temporal + group-by: + - fieldC + condition: base_rule_1 and base_rule_2 + timespan: 15m +""" + ) + assert splunk_backend.convert( + correlation_rule, correlation_method="streamstats" + ) == [ + """| multisearch +[ search fieldA="value1" fieldB="value2" | eval event_type="base_rule_1" ] +[ search fieldA="value3" fieldB="value4" | eval event_type="base_rule_2" ] + +| sort 0 _time +| streamstats time_window=15m values(event_type) as event_types by fieldC + +| search event_types="base_rule_1" event_types="base_rule_2\"""" + ] + + +def test_correlation_rule_subrule_fields_in_streamstats_output(splunk_backend): + 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: 1M + condition: + gte: 10 +""" + ) + assert splunk_backend.convert( + correlation_rule, correlation_method="streamstats" + ) == [ + """fieldA="value1" + +| sort 0 _time +| streamstats time_window=1mon count as event_count values(fieldA) as fieldA values(fieldB) as fieldB by fieldC + +| search event_count >= 10""" + ] + + +def test_event_count_correlation_rule_with_regex_deferred_streamstats(splunk_backend): + 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: Multiple occurrences of base event +status: test +correlation: + type: event_count + rules: + - base_rule + group-by: + - fieldC + timespan: 15m + condition: + gte: 10 +""" + ) + assert splunk_backend.convert( + correlation_rule, correlation_method="streamstats" + ) == [ + """fieldA="value1" +| regex fieldB="value2" + +| sort 0 _time +| streamstats time_window=15m count as event_count by fieldC + +| search event_count >= 10""" + ]