Cleanup Esql Error Types

Author: eric-forte-elastic

detection_rules/esql_errors.py

@@ -43,16 +43,21 @@ class EsqlSchemaError(EsqlKibanaBaseError):
     """Error in ESQL schema. Validated via Kibana until AST is available."""
 
 
-class EsqlSemanticError(EsqlKibanaBaseError):
-    """Error with ESQL semantics. Validated via Kibana until AST is available."""
-
-
 class EsqlSyntaxError(EsqlKibanaBaseError):
     """Error with ESQL syntax. Validated via Kibana until AST is available."""
 
 
 class EsqlTypeMismatchError(Exception):
     """Error when validating types in ESQL."""
 
+    def __init__(self, message: str, elastic_client: Elasticsearch | None = None) -> None:
+        if elastic_client:
+            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) -> None:
         super().__init__(message)

detection_rules/index_mappings.py

@@ -17,7 +17,13 @@
 from . import ecs, integrations, misc, utils
 from .config import load_current_package_version
 from .esql import EventDataset
-from .esql_errors import EsqlSchemaError, EsqlSemanticError, EsqlSyntaxError, cleanup_empty_indices
+from .esql_errors import (
+    EsqlKibanaBaseError,
+    EsqlSchemaError,
+    EsqlSyntaxError,
+    EsqlTypeMismatchError,
+    cleanup_empty_indices,
+)
 from .integrations import (
     load_integrations_manifests,
     load_integrations_schemas,
@@ -251,12 +257,15 @@ def execute_query_against_indices(
         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 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}")
+        if "Unknown column" in error_msg:
+            raise EsqlSchemaError(str(e), elastic_client) from e
+        if "verification_exception" in error_msg:
+            raise EsqlTypeMismatchError(str(e), elastic_client) from e
+        raise EsqlKibanaBaseError(str(e), elastic_client) from e
+    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}")
 
     query_column_names = [c["name"] for c in query_columns]
     log(f"Got query columns: {', '.join(query_column_names)}")
@@ -367,14 +376,16 @@ def prepare_mappings(  # noqa: PLR0913
         non_ecs_mapping.update(non_ecs.get(index, {}))
     non_ecs_mapping = ecs.flatten(non_ecs_mapping)
     non_ecs_mapping = utils.convert_to_nested_schema(non_ecs_mapping)
-    if not combined_mappings and not non_ecs_mapping:
-        raise ValueError("No mappings found")
-    index_lookup.update({"rule-non-ecs-index": non_ecs_mapping})
 
     # Load ECS in an index mapping format (nested schema)
     current_version = Version.parse(load_current_package_version(), optional_minor_and_patch=True)
     ecs_schema = get_ecs_schema_mappings(current_version)
 
     index_lookup.update({"rule-ecs-index": ecs_schema})
+    utils.combine_dicts(combined_mappings, ecs_schema)
+
+    if not combined_mappings and not non_ecs_mapping and not ecs_schema:
+        raise ValueError("No mappings found")
+    index_lookup.update({"rule-non-ecs-index": non_ecs_mapping})
 
     return existing_mappings, index_lookup, combined_mappings

detection_rules/rule.py

@@ -30,6 +30,7 @@
 from . import beats, ecs, endgame, utils
 from .config import load_current_package_version, parse_rules_config
 from .esql import get_esql_query_event_dataset_integrations
+from .esql_errors import EsqlSemanticError
 from .integrations import (
     find_least_compatible_version,
     get_integration_schema_fields,
@@ -963,7 +964,7 @@ class ESQLRuleData(QueryRuleData):
     def validates_esql_data(self, data: dict[str, Any], **_: Any) -> None:
         """Custom validation for query rule type and subclasses."""
         if data.get("index"):
-            raise ValidationError("Index is not a valid field for ES|QL rule type.")
+            raise EsqlSemanticError("Index is not a valid field for ES|QL rule type.")
 
         # Convert the query string to lowercase to handle case insensitivity
         query_lower = data["query"].lower()
@@ -981,7 +982,7 @@ def validates_esql_data(self, data: dict[str, Any], **_: Any) -> None:
 
         # Ensure that non-aggregate queries have metadata
         if not combined_pattern.search(query_lower):
-            raise ValidationError(
+            raise EsqlSemanticError(
                 f"Rule: {data['name']} contains a non-aggregate query without"
                 f" metadata fields '_id', '_version', and '_index' ->"
                 f" Add 'metadata _id, _version, _index' to the from command or add an aggregate function."
@@ -991,7 +992,7 @@ def validates_esql_data(self, data: dict[str, Any], **_: Any) -> None:
         # Match | followed by optional whitespace/newlines and then 'keep'
         keep_pattern = re.compile(r"\|\s*keep\b", re.IGNORECASE | re.DOTALL)
         if not keep_pattern.search(query_lower):
-            raise ValidationError(
+            raise EsqlSemanticError(
                 f"Rule: {data['name']} does not contain a 'keep' command -> Add a 'keep' command to the query."
             )
 

detection_rules/rule_validators.py

@@ -789,12 +789,13 @@ def validate_columns_index_mapping(
             # Check if the column exists in combined_mappings or a valid field generated from a function or operator
             keys = column_name.split(".")
             schema_type = utils.get_column_from_index_mapping_schema(keys, combined_mappings)
+            schema_type = kql.parser.elasticsearch_type_family(schema_type) if schema_type else None
 
             # Validate the type
             if not schema_type or column_type != schema_type:
                 mismatched_columns.append(
                     f"Dynamic field `{column_name}` is not correctly mapped. "
-                    f"If not dynamic: expected `{schema_type}`, got `{column_type}`."
+                    f"If not dynamic: expected from schema: `{schema_type}`, got from Kibana: `{column_type}`."
                 )
 
         if mismatched_columns: