Update Validation Errors for Index handling

Author: eric-forte-elastic

CLI.md

@@ -51,6 +51,8 @@ For instance, some users may want to increase the default value in cases where h
 
 Using the environment variable `DR_REMOTE_ESQL_VALIDATION` will enable remote ESQL validation for rules that use ESQL queries. This validation will be performed whenever the rule is loaded including for example the view-rule command. This requires the appropriate kibana_url or cloud_id, api_key, and es_url to be set in the config file or as environment variables.
 
+Using the environment variable `DR_SKIP_EMPTY_INDEX_CLEANUP` will disable the cleanup of remote testing indexes that are created as part of the remote ESQL validation. By default, these indexes are deleted after the validation is complete, or upon validation error.
+
 ## Importing rules into the repo
 
 You can import rules into the repo using the `create-rule` or `import-rules-to-repo` commands. Both of these commands will

detection_rules/esql_errors.py

@@ -1,5 +1,9 @@
 """ESQL exceptions."""
 
+from elasticsearch import Elasticsearch  # type: ignore[reportMissingTypeStubs]
+
+from .misc import getdefault
+
 __all__ = (
     "EsqlSchemaError",
     "EsqlSemanticError",
@@ -8,29 +12,45 @@
 )
 
 
+def cleanup_empty_indices(
+    elastic_client: Elasticsearch, index_patterns: tuple[str, ...] = ("rule-test-*", "test-*")
+) -> None:
+    """Delete empty indices matching the given patterns."""
+    if getdefault("skip_empty_index_cleanup")():
+        return
+    for pattern in index_patterns:
+        indices = elastic_client.cat.indices(index=pattern, format="json")
+        empty_indices = [index["index"] for index in indices if index["docs.count"] == "0"]  # type: ignore[reportMissingTypeStubs]
+        for empty_index in empty_indices:
+            _ = elastic_client.indices.delete(index=empty_index)
+
+
 class EsqlSchemaError(Exception):
     """Error in ESQL schema. Validated via Kibana until AST is available."""
 
-    def __init__(self, message: str):
+    def __init__(self, message: str, elastic_client: Elasticsearch) -> None:
+        cleanup_empty_indices(elastic_client)
         super().__init__(message)
 
 
 class EsqlSyntaxError(Exception):
     """Error with ESQL syntax. Validated via Kibana until AST is available."""
 
-    def __init__(self, message: str):
+    def __init__(self, message: str, elastic_client: Elasticsearch) -> None:
+        cleanup_empty_indices(elastic_client)
         super().__init__(message)
 
 
 class EsqlSemanticError(Exception):
     """Error with ESQL semantics. Validated via Kibana until AST is available."""
 
-    def __init__(self, message: str):
+    def __init__(self, message: str, elastic_client: Elasticsearch) -> None:
+        cleanup_empty_indices(elastic_client)
         super().__init__(message)
 
 
 class EsqlTypeMismatchError(Exception):
     """Error when validating types in ESQL."""
 
-    def __init__(self, message: str):
+    def __init__(self, message: str) -> None:
         super().__init__(message)

detection_rules/index_mappings.py

@@ -11,30 +11,95 @@
 
 from elastic_transport import ObjectApiResponse
 from elasticsearch import Elasticsearch  # type: ignore[reportMissingTypeStubs]
+from elasticsearch.exceptions import BadRequestError
 from semver import Version
 
 from . import ecs, integrations, misc, utils
 from .config import load_current_package_version
+from .esql_errors import EsqlSchemaError, EsqlSemanticError, EsqlSyntaxError, cleanup_empty_indices
 from .integrations import (
     load_integrations_manifests,
     load_integrations_schemas,
 )
 from .rule import RuleMeta
 from .schemas import get_stack_schemas
+from .schemas.definitions import HTTP_STATUS_BAD_REQUEST
+from .utils import combine_dicts
 
 
 def get_rule_integrations(metadata: RuleMeta) -> list[str]:
     """Retrieve rule integrations from metadata."""
-    rule_integrations: list[str] = []
     if metadata.integration:
-        if isinstance(metadata.integration, list):
-            rule_integrations = metadata.integration
-        else:
-            rule_integrations = [metadata.integration]
+        rule_integrations: list[str] = (
+            metadata.integration if isinstance(metadata.integration, list) else [metadata.integration]
+        )
+    else:
+        rule_integrations: list[str] = []
     return rule_integrations
 
 
-def prepare_integration_mappings(
+def create_index_with_index_mapping(
+    elastic_client: Elasticsearch, index_name: str, mappings: dict[str, Any]
+) -> ObjectApiResponse[Any] | None:
+    """Create an index with the specified mappings and settings to support large number of fields and nested objects."""
+    try:
+        return elastic_client.indices.create(
+            index=index_name,
+            mappings={"properties": mappings},
+            settings={
+                "index.mapping.total_fields.limit": 10000,
+                "index.mapping.nested_fields.limit": 500,
+                "index.mapping.nested_objects.limit": 10000,
+            },
+        )
+    except BadRequestError as e:
+        error_message = str(e)
+        if (
+            e.status_code == HTTP_STATUS_BAD_REQUEST
+            and "validation_exception" in error_message
+            and "Validation Failed: 1: this action would add [2] shards" in error_message
+        ):
+            cleanup_empty_indices(elastic_client)
+            try:
+                return elastic_client.indices.create(
+                    index=index_name,
+                    mappings={"properties": mappings},
+                    settings={
+                        "index.mapping.total_fields.limit": 10000,
+                        "index.mapping.nested_fields.limit": 500,
+                        "index.mapping.nested_objects.limit": 10000,
+                    },
+                )
+            except BadRequestError as retry_error:
+                raise EsqlSchemaError(str(retry_error), elastic_client) from retry_error
+        raise EsqlSchemaError(error_message, elastic_client) from e
+
+
+def get_existing_mappings(elastic_client: Elasticsearch, indices: list[str]) -> tuple[dict[str, Any], dict[str, Any]]:
+    """Retrieve mappings for all matching existing index templates."""
+    existing_mappings: dict[str, Any] = {}
+    index_lookup: dict[str, Any] = {}
+    for index in indices:
+        index_tmpl_mappings = get_simulated_index_template_mappings(elastic_client, index)
+        index_lookup[index] = index_tmpl_mappings
+        combine_dicts(existing_mappings, index_tmpl_mappings)
+    return existing_mappings, index_lookup
+
+
+def get_simulated_index_template_mappings(elastic_client: Elasticsearch, name: str) -> dict[str, Any]:
+    """
+    Return the mappings from the index configuration that would be applied
+    to the specified index from an existing index template
+
+    https://elasticsearch-py.readthedocs.io/en/stable/api/indices.html#elasticsearch.client.IndicesClient.simulate_index_template
+    """
+    template = elastic_client.indices.simulate_index_template(name=name)
+    if not template:
+        return {}
+    return template["template"]["mappings"]["properties"]
+
+
+def prepare_integration_mappings(  # noqa: PLR0913
     rule_integrations: list[str],
     event_dataset_integrations: list[utils.EventDataset],
     package_manifests: Any,
@@ -97,14 +162,14 @@ def create_remote_indices(
     """Create remote indices for validation and return the index string."""
     suffix = str(int(time.time() * 1000))
     test_index = f"rule-test-index-{suffix}"
-    response = misc.create_index_with_index_mapping(elastic_client, test_index, existing_mappings)
+    response = create_index_with_index_mapping(elastic_client, test_index, existing_mappings)
     log(f"Index `{test_index}` created: {response}")
     full_index_str = test_index
 
     # create all integration indices
     for index, properties in index_lookup.items():
         ind_index_str = f"test-{index.rstrip('*')}{suffix}"
-        response = misc.create_index_with_index_mapping(elastic_client, ind_index_str, properties)
+        response = create_index_with_index_mapping(elastic_client, ind_index_str, properties)
         log(f"Index `{ind_index_str}` created: {response}")
         full_index_str = f"{full_index_str}, {ind_index_str}"
 
@@ -124,8 +189,13 @@ def execute_query_against_indices(
         response = elastic_client.esql.query(query=query)
         log(f"Got query response: {response}")
         query_columns = response.get("columns", [])
+    except BadRequestError as e:
+        error_msg = str(e)
+        if "parsing_exception" in error_msg:
+            raise EsqlSyntaxError(str(e), elastic_client) from e
+        raise EsqlSemanticError(str(e), elastic_client) from e
     finally:
-        if delete_indices:
+        if delete_indices or misc.getdefault("skip_empty_index_cleanup")():
             for index_str in test_index_str.split(","):
                 response = elastic_client.indices.delete(index=index_str.strip())
                 log(f"Test index `{index_str}` deleted: {response}")
@@ -182,7 +252,7 @@ def get_ecs_schema_mappings(current_version: Version) -> dict[str, Any]:
     return ecs_schema
 
 
-def prepare_mappings(
+def prepare_mappings(  # noqa: PLR0913
     elastic_client: Elasticsearch,
     indices: list[str],
     event_dataset_integrations: list[utils.EventDataset],
@@ -191,7 +261,7 @@ def prepare_mappings(
     log: Callable[[str], None],
 ) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
     """Prepare index mappings for the given indices and rule integrations."""
-    existing_mappings, index_lookup = misc.get_existing_mappings(elastic_client, indices)
+    existing_mappings, index_lookup = get_existing_mappings(elastic_client, indices)
 
     # Collect mappings for the integrations
     rule_integrations = get_rule_integrations(metadata)

detection_rules/misc.py

@@ -14,13 +14,10 @@
 
 import click
 import requests
-from elastic_transport import ObjectApiResponse
 from elasticsearch import AuthenticationException, Elasticsearch
-from elasticsearch.exceptions import BadRequestError
 from kibana import Kibana  # type: ignore[reportMissingTypeStubs]
 
-from .esql_errors import EsqlSchemaError
-from .utils import add_params, cached, combine_dicts, load_etc_dump
+from .utils import add_params, cached, load_etc_dump
 
 LICENSE_HEADER = """
 Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
@@ -412,46 +409,3 @@ def _wrapped(*args: Any, **kwargs: Any) -> Any:  # noqa: PLR0912
         return _wrapped
 
     return _wrapper
-
-
-def get_simulated_index_template_mappings(elastic_client: Elasticsearch, name: str) -> dict[str, Any]:
-    """
-    Return the mappings from the index configuration that would be applied
-    to the specified index from an existing index template
-
-    https://elasticsearch-py.readthedocs.io/en/stable/api/indices.html#elasticsearch.client.IndicesClient.simulate_index_template
-    """
-    template = elastic_client.indices.simulate_index_template(name=name)
-    if not template:
-        return {}
-    return template["template"]["mappings"]["properties"]
-
-
-def create_index_with_index_mapping(
-    elastic_client: Elasticsearch, index_name: str, mappings: dict[str, Any]
-) -> ObjectApiResponse[Any] | None:
-    """Create an index with the specified mappings and settings to support large number of fields and nested objects."""
-    try:
-        return elastic_client.indices.create(
-            index=index_name,
-            mappings={"properties": mappings},
-            settings={
-                "index.mapping.total_fields.limit": 10000,
-                "index.mapping.nested_fields.limit": 500,
-                "index.mapping.nested_objects.limit": 10000,
-            },
-        )
-    except BadRequestError as e:
-        if e.status_code == 400 and "validation_exception" in str(e):
-            raise EsqlSchemaError(str(e)) from e
-
-
-def get_existing_mappings(elastic_client: Elasticsearch, indices: list[str]) -> tuple[dict[str, Any], dict[str, Any]]:
-    """Retrieve mappings for all matching existing index templates."""
-    existing_mappings: dict[str, Any] = {}
-    index_lookup: dict[str, Any] = {}
-    for index in indices:
-        index_tmpl_mappings = get_simulated_index_template_mappings(elastic_client, index)
-        index_lookup[index] = index_tmpl_mappings
-        combine_dicts(existing_mappings, index_tmpl_mappings)
-    return existing_mappings, index_lookup

detection_rules/rule_validators.py

@@ -736,6 +736,12 @@ def log(self, val: str) -> None:
         if self.verbosity >= unit_test_verbose_level:
             print(f"{self.rule_id}:", val)
 
+    @property
+    def ast(self) -> Any:
+        """Return the AST of the ESQL query. Dependant in ESQL parser which is not implemented"""
+        # Needs to return none to prevent not implemented error
+        return None
+
     @cached_property
     def unique_fields(self) -> list[str]:  # type: ignore[reportIncompatibleMethodOverride]
         """Return a list of unique fields in the query. Requires remote validation to have occurred."""
@@ -791,20 +797,23 @@ def validate(self, data: "QueryRuleData", rule_meta: RuleMeta, force_remote_vali
                 if option.name is not None
             }
 
-            kibana_client = misc.get_kibana_client(**resolved_kibana_options)
             resolved_elastic_options = {
                 option.name: option.default() if callable(option.default) else option.default
                 for option in misc.elasticsearch_options
                 if option.name is not None
             }
-            elastic_client = misc.get_elasticsearch_client(**resolved_elastic_options)
-            _ = self.remote_validate_rule(
-                kibana_client,
-                elastic_client,
-                data.query,
-                rule_meta,
-                data.rule_id,
-            )
+
+            with (
+                misc.get_kibana_client(**resolved_kibana_options) as kibana_client,  # type: ignore[reportUnknownVariableType]
+                misc.get_elasticsearch_client(**resolved_elastic_options) as elastic_client,  # type: ignore[reportUnknownVariableType]
+            ):
+                _ = self.remote_validate_rule(
+                    kibana_client,
+                    elastic_client,
+                    data.query,
+                    rule_meta,
+                    data.rule_id,
+                )
 
     def remote_validate_rule_contents(
         self, kibana_client: Kibana, elastic_client: Elasticsearch, contents: TOMLRuleContents, verbosity: int = 0

detection_rules/schemas/definitions.py

@@ -56,6 +56,7 @@ def validator_wrapper(value: Any) -> Any:
     return validator_wrapper
 
 
+HTTP_STATUS_BAD_REQUEST = 400
 ASSET_TYPE = "security_rule"
 SAVED_OBJECT_TYPE = "security-rule"