From 07203d091de93f47cdb975a0787548731abf67c3 Mon Sep 17 00:00:00 2001 From: eric-forte-elastic Date: Mon, 17 Aug 2026 15:08:10 -0400 Subject: [PATCH 1/3] support ES|QL subqueries --- detection_rules/esql.py | 75 +++++++++++++++++++++++ detection_rules/index_mappings.py | 9 ++- detection_rules/rule_validators.py | 74 +++++++++++++---------- detection_rules/schemas/definitions.py | 12 +++- pyproject.toml | 2 +- tests/test_esql.py | 84 ++++++++++++++++++++++++++ 6 files changed, 218 insertions(+), 38 deletions(-) create mode 100644 tests/test_esql.py diff --git a/detection_rules/esql.py b/detection_rules/esql.py index 12515ff6e85..b28758027db 100644 --- a/detection_rules/esql.py +++ b/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 diff --git a/detection_rules/index_mappings.py b/detection_rules/index_mappings.py index e2d8e6a17f0..81e7ad3bf7f 100644 --- a/detection_rules/index_mappings.py +++ b/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}") diff --git a/detection_rules/rule_validators.py b/detection_rules/rule_validators.py index ecaf8d9e857..d257e9d7ae3 100644 --- a/detection_rules/rule_validators.py +++ b/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,8 +891,9 @@ 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( @@ -909,20 +901,35 @@ def remote_validate_rule( # noqa: PLR0913, PLR0917 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") diff --git a/detection_rules/schemas/definitions.py b/detection_rules/schemas/definitions.py index 295abf43adc..e0c008fa716 100644 --- a/detection_rules/schemas/definitions.py +++ b/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(?:.+?(?:,\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 = { diff --git a/pyproject.toml b/pyproject.toml index 0787efb9385..39412e4dc75 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "detection_rules" -version = "2.1.2" +version = "2.1.3" 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" diff --git a/tests/test_esql.py b/tests/test_esql.py new file mode 100644 index 00000000000..d9492c7b7be --- /dev/null +++ b/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"), []) From 0bf8b4946e673eb8506bd3db1f9856e61f03cdbc Mon Sep 17 00:00:00 2001 From: eric-forte-elastic Date: Tue, 25 Aug 2026 08:50:08 -0400 Subject: [PATCH 2/3] add test for both index types --- tests/test_esql.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/test_esql.py b/tests/test_esql.py index d9492c7b7be..965771f4107 100644 --- a/tests/test_esql.py +++ b/tests/test_esql.py @@ -35,6 +35,11 @@ def test_sources_without_metadata_or_pipe(self): 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_sources_with_and_without_trailing_dash(self): + """Test that an index pattern is extracted whether or not a dash precedes its wildcard.""" + query = "FROM logs-azure.signinlogs*, logs-azure.auditlogs-* METADATA _id\n| WHERE x == 1" + self.assertListEqual(get_esql_query_indices(query), ["logs-azure.signinlogs*", "logs-azure.auditlogs-*"]) + 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" From b3319020fedf35138ef7974808101c987e9b0909 Mon Sep 17 00:00:00 2001 From: eric-forte-elastic Date: Tue, 25 Aug 2026 08:51:31 -0400 Subject: [PATCH 3/3] patch bump --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 7b60f537c76..f5ec8036982 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "detection_rules" -version = "2.1.8" +version = "2.1.9" 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"