leverage eql to validate subquery with synthetic sequence

Author: Mikaayenson

detection_rules/rule_validators.py

@@ -484,7 +484,6 @@ def _validate_against_packaged_integrations(
             query_text: str,
             packaged: list[dict[str, Any]],
             trailer_builder: Callable[[str, str | None, str, str, str], str],
-            join_values: list[Any] | None = None,
             *,
             accumulate_schemas: bool = True,
         ) -> EQL_ERROR_TYPES | ValueError | None:
@@ -510,7 +509,7 @@ def _validate_against_packaged_integrations(
                 if accumulate_schemas:
                     package_schemas.setdefault(package, {}).update(**integration_schema)
 
-                # Build trailer and validate the query text
+                # Build trailer and validate provided text (already synthetic if needed by caller)
                 err_trailer = trailer_builder(package, integration, package_version, stack_version, ecs_version)
                 exc = self.validate_query_text_with_schema(
                     query_text,
@@ -534,22 +533,6 @@ def _validate_against_packaged_integrations(
                 elif exc is not None:
                     return exc
 
-                # Validate join/group-by fields exist in this integration schema (if provided)
-                for jf in join_values or []:
-                    jf_str = str(jf)
-                    if jf_str not in integration_schema:
-                        trailer = (
-                            f"\n\tJoin field not found in schema.\n\t"
-                            f"package: {package}, integration: {integration}, package_version: {package_version}, "
-                            f"stack: {stack_version}, ecs: {ecs_version}"
-                        )
-                        error_fields[jf_str] = {
-                            "error": ValueError(f"Unknown field: {jf_str}"),
-                            "trailer": trailer,
-                            "package": package,
-                            "integration": integration,
-                        }
-
             return None
 
         # Function to extract the field name from an error message
@@ -568,6 +551,13 @@ def _full_query_trailer_builder(pkg: str, integ: str | None, pkg_ver: str, stk_v
                 f"package: {pkg}, integration: {integ}, package_version: {pkg_ver}, stack: {stk_ver}, ecs: {ecs_ver}"
             )
 
+        # Function to build a minimal synthetic sequence containing the subquery
+        def _build_synthetic_sequence_from_subquery(subquery: "ast.SubqueryBy") -> str:
+            subquery_text = str(subquery)
+            join_fields = [str(j) for j in (getattr(subquery, "join_values", []) or [])]
+            dummy_by = f" by {', '.join(join_fields)}" if join_fields else ""
+            return f"sequence\n  {subquery_text}\n  [any where true]{dummy_by}"
+
         # Determine if this is a sequence query via rule data flag
         if data.is_sequence:  # type: ignore[reportAttributeAccessIssue]
             sequence: ast.Sequence = self.ast.first  # type: ignore[reportAttributeAccessIssue]
@@ -586,18 +576,17 @@ def _full_query_trailer_builder(pkg: str, integ: str | None, pkg_ver: str, stk_v
                 # Build subquery-specific package_integrations
                 subquery_pkg_ints = parse_datasets(list(subquery_datasets), packages_manifest)
 
-                # Validate the subquery's event query (without the "by" fields)
-                subquery_query_str = subquery.query.render()  # type: ignore[reportUnknownVariableType]
+                # Validate the entire subquery by wrapping it in a minimal sequence so EQL validates any join fields
+                synthetic_sequence: str = _build_synthetic_sequence_from_subquery(subquery)  # type: ignore[reportUnknownVariableType]
 
                 # Only mark as validated if there are subquery-specific integrations to check
                 if subquery_pkg_ints:
                     did_subquery_validation = True
 
                 exc = _validate_against_packaged_integrations(
-                    subquery_query_str,  # type: ignore[reportUnknownVariableType]
+                    synthetic_sequence,  # validate as a minimal sequence to enforce join field checks
                     subquery_pkg_ints,
                     _subquery_trailer_builder,
-                    join_values=list(getattr(subquery, "join_values", []) or []),  # type: ignore[reportUnknownVariableType]
                     accumulate_schemas=False,
                 )
                 if exc is not None:
@@ -610,7 +599,6 @@ def _full_query_trailer_builder(pkg: str, integ: str | None, pkg_ver: str, stk_v
                     self.query,
                     package_integrations,
                     _full_query_trailer_builder,
-                    join_values=None,
                 )
                 if exc is not None:
                     return exc
@@ -637,7 +625,6 @@ def _full_query_trailer_builder(pkg: str, integ: str | None, pkg_ver: str, stk_v
             self.query,
             package_integrations,
             _full_query_trailer_builder,
-            join_values=None,
         )
         if exc is not None:
             return exc

tests/test_python_library.py

@@ -133,6 +133,47 @@ def test_sequence_invalid_join_field_wrong_package(self) -> None:
         with self.assertRaisesRegex(ValueError, r"Error in both stack and integrations checks"):
             rc.load_dict(bad_rule)
 
+    def test_sequence_top_level_by_and_runs_across_integrations_valid(self) -> None:
+        """Sequence-level by and per-subquery runs; subqueries use different integrations and validate correctly."""
+        rc = RuleCollection()
+        query = """
+        sequence by host.id, user.id with maxspan=1s
+          [any where event.dataset == "azure.auditlogs" and event.action == "Register device"] by azure.auditlogs.properties.initiated_by.user.userPrincipalName with runs=5
+          [authentication where event.dataset == "okta.system" and okta.event_type == "user.mfa.okta_verify.deny_push"] by okta.actor.id
+        """
+        rule = {
+            "metadata": mk_metadata(["azure", "okta"], comments="Top-level sequence by and runs"),
+            "rule": mk_rule(
+                name="EQL sequence with top-level by and runs",
+                rule_id="4e5f6a99-4567-4f8d-9f72-1d8e5f3e5f15",
+                description="Validate top-level sequence by and per-subquery runs across integrations.",
+                risk_score=42,
+                query=query,
+            ),
+        }
+        rc.load_dict(rule)
+
+    def test_sequence_top_level_by_and_runs_across_integrations_invalid_join(self) -> None:
+        """Sequence-level by with runs; okta subquery incorrectly uses an azure join field causing validation failure."""
+        rc = RuleCollection()
+        query = """
+        sequence by host.id, user.id with maxspan=1s
+          [any where event.dataset == "azure.auditlogs" and event.action == "Register device"] by azure.auditlogs.properties.initiated_by.user.userPrincipalName with runs=5
+          [authentication where event.dataset == "okta.system" and okta.event_type == "user.mfa.okta_verify.deny_push"] by azure.auditlogs.properties.initiated_by.user.userPrincipalName
+        """
+        bad_rule = {
+            "metadata": mk_metadata(["azure", "okta"], comments="Top-level sequence by and runs invalid join"),
+            "rule": mk_rule(
+                name="EQL sequence with top-level by and runs invalid",
+                rule_id="4e5f6a99-4567-4f8d-9f72-1d8e5f3e5f16",
+                description="Invalid: okta subquery uses azure join field.",
+                risk_score=42,
+                query=query,
+            ),
+        }
+        with self.assertRaisesRegex(ValueError, r"Error in both stack and integrations checks"):
+            rc.load_dict(bad_rule)
+
     def test_sequence_okta_missing_in_metadata_but_present_in_dataset(self) -> None:
         """Okta dataset appears in a subquery but is not listed in metadata; dataset should drive schema selection."""
         rc = RuleCollection()