diff --git a/sigma/backends/splunk/splunk.py b/sigma/backends/splunk/splunk.py index 4e4810b..0fc2cb5 100644 --- a/sigma/backends/splunk/splunk.py +++ b/sigma/backends/splunk/splunk.py @@ -573,12 +573,21 @@ def finalize_query_data_model( # Re-add the search command after deferred expressions for OR regex filtering if has_or_regex_deferred: deferred_part += search_marker + where_query + # The Condition variables are only created by the eval commands + # after tstats, so they must not be referenced inside the tstats WHERE + # clause. Keep only the conjuncts that do not depend on them there; the + # full expression is applied by the trailing search command. + where_query = self._data_model_where_without_fields( + where_query, + state.processing_state.get("deferred_or_condition_fields", set()), + ) + where_clause = f" where {where_query}" if where_query else "" 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} + 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_clause} 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) @@ -586,6 +595,74 @@ def finalize_query_data_model( "\n", " " ) + @staticmethod + def _data_model_where_without_fields(query: str, fields: Set[str]) -> str: + """ + Return the top-level conjuncts of a search expression that do not reference + any of the given fields, joined with a space. If the expression has an OR at + top level, it is a single conjunct. The result is implied by the original + expression, so it can be used as a pre-filter. + """ + # Split into atoms at whitespace outside quotes and parentheses. + atoms = [] + current = "" + depth = 0 + in_quote = False + escaped = False + for c in query: + if in_quote: + current += c + if escaped: + escaped = False + elif c == "\\": + escaped = True + elif c == '"': + in_quote = False + continue + if c.isspace() and depth == 0: + if current: + atoms.append(current) + current = "" + continue + if c == '"': + in_quote = True + elif c == "(": + depth += 1 + elif c == ")": + depth -= 1 + current += c + if current: + atoms.append(current) + + # Group atoms into conjuncts: NOT binds to the following atom and + # "field IN (...)" is a single conjunct. + if "OR" in atoms: + conjuncts = [query] + else: + conjuncts = [] + i = 0 + while i < len(atoms): + if atoms[i] == "NOT" and i + 1 < len(atoms): + conjunct = [atoms[i], atoms[i + 1]] + i += 2 + else: + conjunct = [atoms[i]] + i += 1 + while i + 1 < len(atoms) and atoms[i] == "IN": + conjunct += [atoms[i], atoms[i + 1]] + i += 2 + conjuncts.append(" ".join(conjunct)) + + field_refs = [ + re.compile(r"(?]") + for field in fields + ] + return " ".join( + conjunct + for conjunct in conjuncts + if not any(field_ref.search(conjunct) for field_ref in field_refs) + ) + 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) diff --git a/tests/test_backend_splunk.py b/tests/test_backend_splunk.py index ac5bbc7..b9e7548 100644 --- a/tests/test_backend_splunk.py +++ b/tests/test_backend_splunk.py @@ -980,7 +980,7 @@ def test_splunk_data_model_process_creation_with_or_regex(): result = splunk_backend.convert(SigmaCollection.from_yaml(rule), "data_model") assert result == [ """| tstats summariesonly=false allow_old_summaries=true fillnull_value="null" count min(_time) as firstTime max(_time) as lastTime from datamodel=Endpoint.Processes where -Processes.parent_process_path="*explorer.exe" (Processes.process="*test_value*" OR processCondition="true") by Processes.process Processes.dest Processes.process_current_directory Processes.process_path Processes.process_integrity_level Processes.original_file_name Processes.parent_process +Processes.parent_process_path="*explorer.exe" by Processes.process Processes.dest Processes.process_current_directory Processes.process_path Processes.process_integrity_level Processes.original_file_name Processes.parent_process Processes.parent_process_path Processes.parent_process_guid Processes.parent_process_id Processes.process_guid Processes.process_id Processes.user | rex field=Processes.process "(?foo.*bar)" | eval processCondition=if(isnotnull(processMatch), "true", "false") @@ -994,6 +994,91 @@ def test_splunk_data_model_process_creation_with_or_regex(): ] +def test_splunk_data_model_process_creation_with_top_level_or_regex(): + """The eval-created processCondition variable does not exist inside tstats, so + a disjunction that references it must not be used as tstats WHERE filter.""" + splunk_backend = SplunkBackend(processing_pipeline=splunk_cim_data_model()) + rule = """ +title: Test +status: test +logsource: + category: process_creation + product: windows +detection: + sel_cmd: + CommandLine|re: foo.*bar + sel_img: + Image: x.exe + condition: 1 of sel_* + """ + result = splunk_backend.convert(SigmaCollection.from_yaml(rule), "data_model") + assert result == [ + """| tstats summariesonly=false allow_old_summaries=true fillnull_value="null" count min(_time) as firstTime max(_time) as lastTime from datamodel=Endpoint.Processes +by Processes.process Processes.dest Processes.process_current_directory Processes.process_path Processes.process_integrity_level Processes.original_file_name Processes.parent_process +Processes.parent_process_path Processes.parent_process_guid Processes.parent_process_id Processes.process_guid Processes.process_id Processes.user +| rex field=Processes.process "(?foo.*bar)" +| eval processCondition=if(isnotnull(processMatch), "true", "false") +| search processCondition="true" OR Processes.process_path="x.exe" +| `drop_dm_object_name(Processes)` +| convert timeformat="%Y-%m-%dT%H:%M:%S" ctime(firstTime) +| convert timeformat="%Y-%m-%dT%H:%M:%S" ctime(lastTime) +""".replace( + "\n", " " + ) + ] + + +def test_splunk_data_model_or_regex_keeps_independent_conjuncts(): + splunk_backend = SplunkBackend(processing_pipeline=splunk_cim_data_model()) + rule = """ +title: Test +status: test +logsource: + category: process_creation + product: windows +detection: + sel_cmd: + - CommandLine|re: foo.*bar + - Image: x.exe + sel_parent: + ParentImage: + - p1.exe + - p2.exe + filter: + User: admin + condition: sel_cmd and sel_parent and not filter + """ + result = splunk_backend.convert(SigmaCollection.from_yaml(rule), "data_model") + tstats_where = result[0].split(" where ", 1)[1].split(" by ", 1)[0] + assert ( + tstats_where + == '(Processes.parent_process_path IN ("p1.exe", "p2.exe")) NOT Processes.user="admin"' + ) + assert "processCondition" not in tstats_where + + +@pytest.mark.parametrize( + "query,expected", + [ + ('a="1" xCondition="true"', 'a="1"'), + ('a="1" OR xCondition="true"', ""), + ('a="1" (b="2" OR xCondition="true")', 'a="1"'), + ('NOT xCondition="true" b IN ("1", "2")', 'b IN ("1", "2")'), + ('NOT b IN ("1", "2") xCondition2="true"', 'NOT b IN ("1", "2")'), + # conservative: a quoted value that looks like a reference is also dropped + ('a="1" b="xCondition=\\"true\\" c"', 'a="1"'), + ('a="1" b="2"', 'a="1" b="2"'), + ], +) +def test_splunk_data_model_where_without_fields(query, expected): + assert ( + SplunkBackend._data_model_where_without_fields( + query, {"xCondition", "xCondition2"} + ) + == expected + ) + + def test_splunk_data_model_no_data_model_specified(): splunk_backend = SplunkBackend() rule = """