[FR] ES|QL remote validation support newline split indices (#5356)

* Updated regex pattern for multiline

* Add line split unit test

Author: eric-forte-elastic

detection_rules/schemas/definitions.py

@@ -76,7 +76,9 @@ 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*(?:\||\bmetadata\b|//|$)", re.IGNORECASE | re.MULTILINE)
+FROM_SOURCES_REGEX = re.compile(
+    r"^\s*FROM\s+(?P<sources>(?:.+?(?:,\s*)?\n?)+?)\s*(?:\||\bmetadata\b|//|$)", re.IGNORECASE | re.MULTILINE
+)
 BRANCH_PATTERN = f"{VERSION_PATTERN}|^master$"
 ELASTICSEARCH_EQL_FEATURES = {
     "allow_negation": (Version.parse("8.9.0"), None),

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "detection_rules"
-version = "1.5.17"
+version = "1.5.18"
 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_rules_remote.py

@@ -155,6 +155,22 @@ def test_esql_filtered_index_error(self):
         with pytest.raises(EsqlSchemaError):
             _ = RuleCollection().load_dict(production_rule)
 
+    def test_new_line_split_index(self):
+        """Test an ESQL rule's index validation to ensure that it can handle new line split indices."""
+        file_path = get_path(["tests", "data", "command_control_dummy_production_rule.toml"])
+        original_production_rule = load_rule_contents(file_path)
+        production_rule = deepcopy(original_production_rule)[0]
+        production_rule["metadata"]["integration"] = ["aws"]
+        production_rule["rule"]["query"] = """
+        from logs-aws.cloud*, logs-network_traffic.http-*,
+        logs-nginx.access-* metadata _id, _version, _index
+        | where @timestamp > now() - 30 minutes
+        and aws.cloudtrail.user_identity.type == "IAMUser"
+        | keep
+        aws.*
+        """
+        _ = RuleCollection().load_dict(production_rule)
+
     def test_esql_endpoint_alerts_index(self):
         """Test an ESQL rule's schema validation using ecs fields in the alerts index."""
         file_path = get_path(["tests", "data", "command_control_dummy_production_rule.toml"])