[FR] Support ES|QL sub queries

detection_rules/esql.py

@@ -8,6 +8,13 @@
 import re
 from dataclasses import dataclass
 
+from .schemas.definitions import (
+    ESQL_COMMENTS_AND_LITERALS_REGEX,
+    ESQL_FROM_KEYWORD_REGEX,
+    ESQL_FROM_SOURCES_TERMINATOR_REGEX,
+    ESQL_INDEX_PATTERN_REGEX,
+)
+
 
 @dataclass
 class EventDataset:
@@ -20,6 +27,14 @@ def __str__(self) -> str:
         return f"{self.package}.{self.integration}"
 
 
+@dataclass
+class EsqlSourceGroup:
+    """Dataclass for the FROM clauses of a query that read the same index patterns."""
+
+    indices: list[str]
+    spans: list[tuple[int, int]]
+
+
 def get_esql_query_event_dataset_integrations(query: str) -> list[EventDataset]:
     """Extract event.dataset and data_stream.dataset integrations from an ES|QL query."""
     number_of_parts = 2
@@ -58,3 +73,63 @@ def get_esql_query_event_dataset_integrations(query: str) -> list[EventDataset]:
             event_datasets.append(EventDataset(package=parts[0], integration=parts[1]))
 
     return event_datasets
+
+
+def split_esql_source_list(sources: str) -> list[str]:
+    """Split a FROM clause source list into its local index patterns."""
+    indices: list[str] = []
+    for source in sources.split(","):
+        # Truncate cross cluster search indices to local indices
+        index = source.split(":", 1)[-1].strip()
+        if ESQL_INDEX_PATTERN_REGEX.match(index):
+            indices.append(index)
+    return indices
+
+
+def get_esql_query_source_groups(query: str) -> list[EsqlSourceGroup]:
+    """Group the FROM clauses of an ES|QL query by the index patterns they read."""
+
+    def blank(match: re.Match[str]) -> str:
+        return "".join("\n" if char == "\n" else " " for char in match.group(0))
+
+    # Blanked in place, preserving offsets, so that the FROM keyword or something shaped like an
+    # index pattern is never read out of a comment or a query value
+    scannable = ESQL_COMMENTS_AND_LITERALS_REGEX.sub(blank, query)
+
+    groups: dict[tuple[str, ...], EsqlSourceGroup] = {}
+    for match in ESQL_FROM_KEYWORD_REGEX.finditer(scannable):
+        start = match.end()
+        # The outer FROM of a subquery union takes subqueries rather than index patterns,
+        # so it has no source list of its own and only each subquery's FROM clause is grouped
+        if scannable[start:].lstrip().startswith("("):
+            continue
+        terminator = ESQL_FROM_SOURCES_TERMINATOR_REGEX.search(scannable, start)
+        end = terminator.start() if terminator else len(scannable)
+        sources = scannable[start:end]
+        indices = split_esql_source_list(sources)
+        # Guards against a FROM keyword that is part of an expression rather than a source clause
+        if not indices:
+            continue
+        # Clauses reading the same sources share a group, so they also share prepared test indices
+        group = groups.setdefault(tuple(indices), EsqlSourceGroup(indices=indices, spans=[]))
+        group.spans.append((start, start + len(sources.rstrip())))
+
+    return list(groups.values())
+
+
+def get_esql_query_indices(query: str) -> list[str]:
+    """Extract the unique index patterns from every FROM clause in an ES|QL query."""
+    indices: list[str] = []
+    for group in get_esql_query_source_groups(query):
+        for index in group.indices:
+            if index not in indices:
+                indices.append(index)
+    return indices
+
+
+def replace_esql_query_sources(query: str, replacements: dict[tuple[int, int], str]) -> str:
+    """Replace each FROM clause source list with the index string mapped to its span."""
+    # Applied back to front so that earlier spans keep their offsets
+    for (start, end), replacement in sorted(replacements.items(), reverse=True):
+        query = query[:start] + replacement + query[end:]
+    return query

detection_rules/index_mappings.py

@@ -281,7 +281,9 @@ def get_filtered_index_schema(  # noqa: PLR0913, PLR0917
 
     matches: list[str] = []
     for index in indices:
-        pattern = re.compile(index.replace(".", r"\.").replace("*", ".*").rstrip("-"))
+        # Escaped rather than substituted so that an unexpected source, e.g. a parenthesis picked up
+        # from a subquery, raises EsqlUnknownIndexError below instead of an invalid pattern error
+        pattern = re.compile(re.escape(index.rstrip("-")).replace(r"\*", ".*"))
         matches.extend([key for key in filtered_keys if pattern.fullmatch(key)])
 
     if not matches:
@@ -334,10 +336,13 @@ def create_remote_indices(
     existing_mappings: dict[str, Any],
     index_lookup: dict[str, Any],
     log: Callable[[str], None],
+    name_suffix: str = "",
 ) -> str:
     """Create remote indices for validation and return the index string."""
 
-    suffix = str(int(time.time() * 1000))
+    # A rule prepares one set of indices per FROM clause, and those sets can be created within the
+    # same millisecond, so the caller passes a suffix to keep the index names distinct
+    suffix = f"{int(time.time() * 1000)}{name_suffix}"
     test_index = f"rule-test-index-{suffix}"
     response = create_index_with_index_mapping(elastic_client, test_index, existing_mappings)
     log(f"Index `{test_index}` created: {response}")

detection_rules/rule_validators.py

@@ -32,7 +32,11 @@
 from .beats import get_datasets_and_modules, parse_beats_from_index
 from .config import CUSTOM_RULES_DIR, load_current_package_version, parse_rules_config
 from .custom_schemas import update_auto_generated_schema
-from .esql import get_esql_query_event_dataset_integrations
+from .esql import (
+    get_esql_query_event_dataset_integrations,
+    get_esql_query_source_groups,
+    replace_esql_query_sources,
+)
 from .esql_errors import EsqlTypeMismatchError
 from .index_mappings import (
     create_remote_indices,
@@ -48,7 +52,7 @@
 )
 from .rule import EQLRuleData, QueryRuleData, QueryValidator, RuleMeta, TOMLRuleContents, set_eql_config
 from .schemas import get_latest_stack_version, get_stack_schemas, get_stack_versions
-from .schemas.definitions import ESQL_DYNAMIC_FIELD_PREFIXES, FROM_SOURCES_REGEX
+from .schemas.definitions import ESQL_DYNAMIC_FIELD_PREFIXES
 
 EQL_ERROR_TYPES = (
     eql.EqlCompileError
@@ -763,19 +767,6 @@ def unique_fields(self) -> list[str]:  # type: ignore[reportIncompatibleMethodOv
             return [field["name"] for field in self.esql_unique_fields]
         return []
 
-    def get_esql_query_indices(self, query: str) -> tuple[str, list[str]]:
-        """Extract indices from an ES|QL query."""
-        match = FROM_SOURCES_REGEX.search(query)
-        if not match:
-            return "", []
-
-        sources_str = match.group("sources")
-        # Truncate cross cluster search indices to local indices
-        sources_list: list[str] = [
-            source.split(":", 1)[-1].strip() if ":" in source else source.strip() for source in sources_str.split(",")
-        ]
-        return sources_str, sources_list
-
     def get_unique_field_type(self, field_name: str) -> str | None:  # type: ignore[reportIncompatibleMethodOverride]
         """Get the type of the unique field. Requires remote validation to have occurred."""
         esql_unique_fields = getattr(self, "esql_unique_fields", [])
@@ -900,29 +891,45 @@ def remote_validate_rule(  # noqa: PLR0913, PLR0917
         stack_version = get_latest_stack_version()
 
         self.log(f"Validating against {stack_version} stack")
-        indices_str, indices = self.get_esql_query_indices(query)  # type: ignore[reportUnknownVariableType]
-        self.log(f"Extracted indices from query: {', '.join(indices)}")
+        source_groups = get_esql_query_source_groups(query)
+        if not source_groups:
+            raise ValueError("Failed to extract any index pattern from the query's FROM clause(s).")
 
         event_dataset_integrations = get_esql_query_event_dataset_integrations(query)
         self.log(
             "Extracted Event Dataset integrations from query: "
             f"{', '.join(str(integration) for integration in event_dataset_integrations)}"
         )
 
-        # Get mappings for all matching existing index templates
-        existing_mappings, index_lookup, combined_mappings = prepare_mappings(
-            elastic_client, indices, event_dataset_integrations, metadata, stack_version, self.log
-        )
-        self.log(f"Collected mappings: {len(existing_mappings)}")
-        self.log(f"Combined mappings prepared: {len(combined_mappings)}")
+        # Each FROM clause is prepared against only the indices it reads, so a subquery cannot
+        # validate a field that exists solely in the index of one of its siblings
+        combined_mappings: dict[str, Any] = {}
+        source_replacements: dict[tuple[int, int], str] = {}
+        test_indices: list[str] = []
+        for position, group in enumerate(source_groups):
+            self.log(f"Extracted indices from query: {', '.join(group.indices)}")
+
+            # Get mappings for all matching existing index templates
+            existing_mappings, index_lookup, group_mappings = prepare_mappings(
+                elastic_client, group.indices, event_dataset_integrations, metadata, stack_version, self.log
+            )
+            self.log(f"Collected mappings: {len(existing_mappings)}")
+            self.log(f"Combined mappings prepared: {len(group_mappings)}")
+            utils.combine_dicts(combined_mappings, group_mappings)
 
-        # Create remote indices
-        full_index_str = create_remote_indices(elastic_client, existing_mappings, index_lookup, self.log)
+            # Create remote indices
+            full_index_str = create_remote_indices(
+                elastic_client, existing_mappings, index_lookup, self.log, name_suffix=f"-{position}"
+            )
+            source_replacements.update(dict.fromkeys(group.spans, full_index_str))
+            test_indices.extend(index.strip() for index in full_index_str.split(","))
 
-        # Replace all sources with the test indices
-        query = query.replace(indices_str, full_index_str)  # type: ignore[reportUnknownVariableType]
+        # Replace the sources of every FROM clause with the test indices prepared for it
+        query = replace_esql_query_sources(query, source_replacements)
 
-        query_columns, response = execute_query_against_indices(elastic_client, query, full_index_str, self.log)  # type: ignore[reportUnknownVariableType]
+        # Deduplicated because the test indices are also the set the execution cleans up afterwards
+        all_index_str = ", ".join(dict.fromkeys(test_indices))
+        query_columns, response = execute_query_against_indices(elastic_client, query, all_index_str, self.log)  # type: ignore[reportUnknownVariableType]
         self.esql_unique_fields = query_columns
 
         # Build a mapping lookup for all stack versions to validate against.
@@ -944,10 +951,13 @@ def remote_validate_rule(  # noqa: PLR0913, PLR0917
             version = str(parsed.replace(patch=max(parsed.patch, inferred_patch)))  # noqa: PLW2901
             if version in mappings_lookup:
                 continue
-            _, _, combined_mappings = prepare_mappings(
-                elastic_client, indices, event_dataset_integrations, metadata, version, self.log
-            )
-            mappings_lookup[version] = combined_mappings
+            version_mappings: dict[str, Any] = {}
+            for group in source_groups:
+                _, _, group_mappings = prepare_mappings(
+                    elastic_client, group.indices, event_dataset_integrations, metadata, version, self.log
+                )
+                utils.combine_dicts(version_mappings, group_mappings)
+            mappings_lookup[version] = version_mappings
 
         for version, mapping in mappings_lookup.items():
             self.log(f"Validating {rule_id} against {version} stack")

detection_rules/schemas/definitions.py

@@ -76,9 +76,15 @@ def validator_wrapper(value: Any) -> Any:
 CONDITION_VERSION_PATTERN = re.compile(rf"^\^{_version}$")
 VERSION_PATTERN = f"^{_version}$"
 MINOR_SEMVER = re.compile(r"^\d+\.\d+$")
-FROM_SOURCES_REGEX = re.compile(
-    r"^\s*FROM\s+(?P<sources>(?:.+?(?:,\s*)?\n?)+?)\s*(?:\||\bmetadata\b|//|$)", re.IGNORECASE | re.MULTILINE
-)
+# ES|QL comments and string literals are blanked before a query is scanned, so that the FROM
+# keyword, or something shaped like an index pattern, is never read out of prose or a query value
+ESQL_COMMENTS_AND_LITERALS_REGEX = re.compile(r'"""(?:.|\n)*?"""|"(?:[^"\\\n]|\\.)*"|//[^\n]*|/\*(?:.|\n)*?\*/')
+# An ES|QL query has one FROM clause per source, and subqueries nest them,
+# e.g. `FROM (FROM logs-a-* | ...), (FROM logs-b-* | ...)`
+ESQL_FROM_KEYWORD_REGEX = re.compile(r"\bFROM\b\s+", re.IGNORECASE)
+# An ES|QL source list runs until the first pipe, the METADATA directive, or the end of the subquery
+ESQL_FROM_SOURCES_TERMINATOR_REGEX = re.compile(r"\||\)|\bMETADATA\b", re.IGNORECASE)
+ESQL_INDEX_PATTERN_REGEX = re.compile(r"^[\w.*\-]+$")
 ESQL_DYNAMIC_FIELD_PREFIXES = ("Esql.", "Esql_priv.")
 BRANCH_PATTERN = f"{VERSION_PATTERN}|^master$"
 ELASTICSEARCH_EQL_FEATURES = {

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "detection_rules"
-version = "2.1.6"
+version = "2.1.8"
 description = "Detection Rules is the home for rules used by Elastic Security. This repository is used for the development, maintenance, testing, validation, and release of rules for Elastic Security’s Detection Engine."
 readme = "README.md"
 requires-python = ">=3.12"

tests/test_esql.py

@@ -0,0 +1,84 @@
+# Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+# or more contributor license agreements. Licensed under the Elastic License
+# 2.0; you may not use this file except in compliance with the Elastic License
+# 2.0.
+
+"""Test ES|QL query parsing."""
+
+import unittest
+
+from detection_rules.esql import (
+    get_esql_query_indices,
+    get_esql_query_source_groups,
+    replace_esql_query_sources,
+)
+
+
+def replace_with_group_position(query: str) -> str:
+    """Replace each FROM clause source list with a marker for the group it belongs to."""
+    groups = get_esql_query_source_groups(query)
+    replacements = {span: f"test-index-{position}" for position, group in enumerate(groups) for span in group.spans}
+    return replace_esql_query_sources(query, replacements)
+
+
+class TestESQLQuerySources(unittest.TestCase):
+    """Test extraction and replacement of the sources of an ES|QL query."""
+
+    def test_flat_sources(self):
+        """Test a query with a single FROM clause."""
+        query = "FROM logs-a-*, logs-b-* METADATA _id\n| WHERE x == 1"
+        self.assertListEqual(get_esql_query_indices(query), ["logs-a-*", "logs-b-*"])
+        self.assertEqual(replace_with_group_position(query), "FROM test-index-0 METADATA _id\n| WHERE x == 1")
+
+    def test_sources_without_metadata_or_pipe(self):
+        """Test that a source list terminated by the end of the query is extracted."""
+        self.assertListEqual(get_esql_query_indices("FROM logs-a-*"), ["logs-a-*"])
+        self.assertListEqual(get_esql_query_indices("FROM logs-a-*\n| WHERE x"), ["logs-a-*"])
+
+    def test_cross_cluster_sources(self):
+        """Test that cross cluster sources are truncated to local indices."""
+        query = "FROM cluster_one:logs-a-*, logs-b-* METADATA _id\n| WHERE x"
+        self.assertListEqual(get_esql_query_indices(query), ["logs-a-*", "logs-b-*"])
+
+    def test_subqueries_are_grouped_by_their_own_sources(self):
+        """Test that subqueries reading different indices are grouped and replaced separately."""
+        query = "FROM\n(\n  FROM logs-a-* METADATA _id\n  | WHERE x\n),\n(\n  FROM logs-b-* METADATA _id\n)\n| WHERE y"
+        groups = get_esql_query_source_groups(query)
+        self.assertListEqual([group.indices for group in groups], [["logs-a-*"], ["logs-b-*"]])
+        self.assertListEqual([len(group.spans) for group in groups], [1, 1])
+        self.assertEqual(
+            replace_with_group_position(query),
+            "FROM\n(\n  FROM test-index-0 METADATA _id\n  | WHERE x\n),\n(\n  FROM test-index-1 METADATA _id\n)\n"
+            "| WHERE y",
+        )
+
+    def test_subqueries_reading_the_same_sources_share_a_group(self):
+        """Test that subqueries reading the same indices share one group, and so one set of indices."""
+        query = "FROM (FROM logs-a-* METADATA _id | WHERE x), (FROM logs-a-* METADATA _id | WHERE y) | WHERE z"
+        groups = get_esql_query_source_groups(query)
+        self.assertListEqual([group.indices for group in groups], [["logs-a-*"]])
+        self.assertEqual(len(groups[0].spans), 2)
+        self.assertListEqual(get_esql_query_indices(query), ["logs-a-*"])
+        self.assertEqual(
+            replace_with_group_position(query),
+            "FROM (FROM test-index-0 METADATA _id | WHERE x), (FROM test-index-0 METADATA _id | WHERE y) | WHERE z",
+        )
+
+    def test_sources_are_not_read_from_comments(self):
+        """Test that a FROM keyword or index pattern within a comment is ignored."""
+        line_comment = "FROM logs-a-*\n// downloads from logs-evil-* are excluded\n| WHERE x"
+        block_comment = "/*\nSelects rows from logs-evil-* only\n*/\nFROM logs-a-* METADATA _id\n| WHERE x"
+        self.assertListEqual(get_esql_query_indices(line_comment), ["logs-a-*"])
+        self.assertListEqual(get_esql_query_indices(block_comment), ["logs-a-*"])
+
+    def test_sources_are_not_read_from_literals(self):
+        """Test that a FROM keyword or index pattern within a string literal is ignored."""
+        literal = 'FROM logs-a-*\n| WHERE msg LIKE "*copied from logs-evil-**"\n| WHERE y'
+        raw_literal = 'FROM logs-a-*\n| EVAL x = REPLACE(y, """from logs-evil-*""", "")\n| WHERE z'
+        self.assertListEqual(get_esql_query_indices(literal), ["logs-a-*"])
+        self.assertListEqual(get_esql_query_indices(raw_literal), ["logs-a-*"])
+
+    def test_query_without_sources(self):
+        """Test that a query with no FROM clause yields no groups."""
+        self.assertListEqual(get_esql_query_source_groups("| WHERE x == 1"), [])
+        self.assertListEqual(get_esql_query_indices("| WHERE x == 1"), [])