Feat/improve data model conversion

README.md

@@ -15,6 +15,35 @@ It supports the following output formats:
 * default: plain Splunk queries
 * savedsearches: Splunk savedsearches.conf format.
 
+## Data model (tstats) query settings
+
+The `data_model` output format generates accelerated `tstats` queries against a Splunk data
+model. The following settings let you tune the generated query without changing the Sigma
+rules themselves. They are read from the pipeline processing state (set with a
+`SetStateTransformation` / the `set_state` pipeline item), so detection logic stays separate
+from environment-specific query configuration. Defaults keep the generated query unchanged.
+
+* `summariesonly` (also available as the `summariesonly` backend option, e.g.
+  `-O summariesonly=true` with sigma-cli): controls the `tstats summariesonly=...` flag.
+  Accepts a boolean or a string (`true`/`false`). When both are set, the pipeline
+  processing state takes precedence over the backend option. Default: `false`.
+
+The following are Splunk-specific extensions of the data model output and have no direct
+representation in the Sigma standard:
+
+* `tstats_span`: adds `_time` to the `by` clause and a `span=<value>` bucket, e.g.
+  `tstats_span: 1h`. The value is a number followed by an optional Splunk time unit
+  (`s`, `sec`, `secs`, `second`, `seconds`, `m`, `min`, `mins`, `minute`, `minutes`,
+  `h`, `hr`, `hrs`, `hour`, `hours`, `d`, `day`, `days`, `mon`, `mont`, `months`); a bare
+  number (e.g. `30`) is treated as seconds.
+* `tstats_aggregations`: a list of additional aggregation functions appended after the
+  default `count`. Each entry is a mapping with `func` (one of the supported Splunk stats
+  functions, e.g. `values`, `sum`, `dc`), `field`, and an optional `as`/`alias`, e.g.
+  `[{func: values, field: Processes.process_name, as: process_names}]`. `count` is always
+  present and cannot be redefined here.
+* `tstats_aggregation`: a raw aggregation string appended verbatim after `count` (advanced
+  use). When set, it takes precedence over `tstats_aggregations`.
+
 This backend is currently maintained by:
 
 * [Thomas Patzke](https://github.com/thomaspatzke/)

sigma/backends/splunk/splunk.py

@@ -23,7 +23,7 @@
     splunk_dns_cim_mapping,
 )
 import sigma
-from typing import Any, Callable, ClassVar, Dict, List, Optional, Pattern, Tuple, Union
+from typing import Any, Callable, ClassVar, Dict, List, Optional, Pattern, Set, Tuple, Union
 
 
 class SplunkDeferredRegularExpression(DeferredTextQueryExpression):
@@ -272,6 +272,17 @@ class SplunkBackend(TextQueryBackend):
         "stats": 'event_types="{ruleid}"'
     }
 
+    # Splunk stats/tstats functions allowed in the `tstats_aggregations` data model setting.
+    data_model_aggregation_functions: ClassVar[Set[str]] = {
+        "avg", "dc", "distinct_count", "earliest", "estdc", "first", "last",
+        "latest", "list", "max", "mean", "median", "min", "mode", "range",
+        "stdev", "stdevp", "sum", "sumsq", "values", "var", "varp",
+    }
+    _data_model_identifier_re: ClassVar[Pattern] = re.compile(r"^[\w.]+$")
+    _data_model_span_re: ClassVar[Pattern] = re.compile(
+        r"^\d+(?:s|sec|secs|second|seconds|m|min|mins|minute|minutes|h|hr|hrs|hour|hours|d|day|days|mon|mont|months)?$"
+    )
+
     def __init__(
         self,
         processing_pipeline: Optional[
@@ -280,11 +291,13 @@ def __init__(
         collect_errors: bool = False,
         min_time: str = "-30d",
         max_time: str = "now",
+        summariesonly: bool = False,
         query_settings: Callable[[SigmaRule], Dict[str, str]] = lambda x: {},
         output_settings: Dict = {},
         **kwargs,
     ):
         super().__init__(processing_pipeline, collect_errors, **kwargs)
+        self.summariesonly = summariesonly
         self.query_settings = query_settings
         self.output_settings = {
             "dispatch.earliest_time": min_time,
@@ -557,13 +570,87 @@ def finalize_query_data_model(
         if has_or_regex_deferred:
             deferred_part += search_marker + where_query
 
-        return f"""| tstats summariesonly=false allow_old_summaries=true fillnull_value="null" count min(_time) as firstTime max(_time) as lastTime from datamodel={data_model_set} where {where_query} by {fields}{deferred_part}
+        summariesonly = self._data_model_summariesonly(state)
+        aggregation = self._data_model_aggregation(state)
+        time_prefix, span_suffix = self._data_model_span(state)
+
+        return f"""| tstats summariesonly={summariesonly} allow_old_summaries=true fillnull_value="null" count min(_time) as firstTime max(_time) as lastTime{aggregation} from datamodel={data_model_set} where {where_query} by {time_prefix}{fields}{span_suffix}{deferred_part}
 | `drop_dm_object_name({data_set})`
 | convert timeformat="%Y-%m-%dT%H:%M:%S" ctime(firstTime)
 | convert timeformat="%Y-%m-%dT%H:%M:%S" ctime(lastTime)
 """.replace(
             "\n", " "
         )
 
+    def _data_model_summariesonly(self, state: ConversionState) -> str:
+        """Resolve tstats `summariesonly` from pipeline state, falling back to the backend option."""
+        value = state.processing_state.get("summariesonly", self.summariesonly)
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("true", "t", "yes", "1"):
+                return "true"
+            if normalized in ("false", "f", "no", "0"):
+                return "false"
+            raise SigmaFeatureNotSupportedByBackendError(
+                f"Invalid 'summariesonly' value for tstats data model query: {value!r}"
+            )
+        return self.bool_values[bool(value)]
+
+    def _data_model_aggregation(self, state: ConversionState) -> str:
+        """Render extra tstats aggregation functions from pipeline state, appended to the default count."""
+        raw = state.processing_state.get("tstats_aggregation")
+        if raw is not None:
+            raw = str(raw).strip()
+            return f" {raw}" if raw else ""
+        specs = state.processing_state.get("tstats_aggregations")
+        if not specs:
+            return ""
+        return "".join(
+            f" {self._render_data_model_aggregation(spec)}" for spec in specs
+        )
+
+    def _render_data_model_aggregation(self, spec: Dict[str, str]) -> str:
+        if not isinstance(spec, dict):
+            raise SigmaFeatureNotSupportedByBackendError(
+                f"Each 'tstats_aggregations' entry must be a mapping, got: {spec!r}"
+            )
+        func = str(spec.get("func", "")).strip().lower()
+        if func not in self.data_model_aggregation_functions:
+            raise SigmaFeatureNotSupportedByBackendError(
+                f"Unsupported tstats aggregation function: {spec.get('func')!r}"
+            )
+        field = spec.get("field")
+        if not field:
+            raise SigmaFeatureNotSupportedByBackendError(
+                f"tstats aggregation function '{func}' requires a 'field'"
+            )
+        field = str(field).strip()
+        if not self._data_model_identifier_re.match(field):
+            raise SigmaFeatureNotSupportedByBackendError(
+                f"Invalid field name in 'tstats_aggregations': {field!r}"
+            )
+        token = f"{func}({field})"
+        alias = spec.get("as") or spec.get("alias")
+        if alias:
+            alias = str(alias).strip()
+            if not self._data_model_identifier_re.match(alias):
+                raise SigmaFeatureNotSupportedByBackendError(
+                    f"Invalid alias in 'tstats_aggregations': {alias!r}"
+                )
+            token += f" as {alias}"
+        return token
+
+    def _data_model_span(self, state: ConversionState) -> Tuple[str, str]:
+        """Return the `_time`/`span=` fragments for tstats time bucketing from pipeline state."""
+        span = state.processing_state.get("tstats_span")
+        if span is None:
+            return "", ""
+        span = str(span).strip()
+        if not self._data_model_span_re.match(span):
+            raise SigmaFeatureNotSupportedByBackendError(
+                f"Invalid 'tstats_span' value for tstats data model query: {span!r}"
+            )
+        return "_time ", f" span={span}"
+
     def finalize_output_data_model(self, queries: List[str]) -> List[str]:
         return queries