[medium] Keep eval-only regex condition variables out of tstats where

sigma/backends/splunk/splunk.py

@@ -573,19 +573,96 @@ 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 <field>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)
 """.replace(
             "\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"(?<![\w.])" + re.escape(field) + r"\s*[=!<>]")
+            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)

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 "(?<processMatch>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 "(?<processMatch>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 = """