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

* Updated regex pattern for multiline

* Add line split unit test

detection_rules/schemas/definitions.py

@@ -76,7 +76,9 @@ _version = r"\d+\.\d+(\.\d+[\w-]*)*"
 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 @@ class TestRemoteRules(BaseRuleTest):
         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"])