[FR] Add Alert Suppression for Addtional Rule Types

detection_rules/cli_utils.py

@@ -132,6 +132,9 @@ def rule_prompt(path=None, rule_type=None, required_only=True, save=True, verbos
 
     for name, options in props.items():
 
+        if name == 'index' and kwargs.get("type") == "esql":
+            continue
+
         if name == 'type':
             contents[name] = rule_type
             continue

detection_rules/rule.py

@@ -758,20 +758,14 @@ def validates_index_and_data_view_id(self, data, **kwargs):
         if data.get('index') and data.get('data_view_id'):
             raise ValidationError("Only one of index or data_view_id should be set.")
 
-    @validates_schema
-    def validates_query_data(self, data, **kwargs):
-        """Custom validation for query rule type and subclasses."""
-        # alert suppression is only valid for query rule type and not any of its subclasses
-        if data.get('alert_suppression') and data['type'] not in ('query', 'threshold'):
-            raise ValidationError("Alert suppression is only valid for query and threshold rule types.")
-
 
 @dataclass(frozen=True)
 class MachineLearningRuleData(BaseRuleData):
     type: Literal["machine_learning"]
 
     anomaly_threshold: int
     machine_learning_job_id: Union[str, List[str]]
+    alert_suppression: Optional[AlertSuppressionMapping] = field(metadata=dict(metadata=dict(min_compat="8.15")))
 
 
 @dataclass(frozen=True)
@@ -811,6 +805,7 @@ class HistoryWindowStart:
 
     type: Literal["new_terms"]
     new_terms: NewTermsMapping
+    alert_suppression: Optional[AlertSuppressionMapping] = field(metadata=dict(metadata=dict(min_compat="8.14")))
 
     @pre_load
     def preload_data(self, data: dict, **kwargs) -> dict:
@@ -849,6 +844,7 @@ class EQLRuleData(QueryRuleData):
     timestamp_field: Optional[str] = field(metadata=dict(metadata=dict(min_compat="8.0")))
     event_category_override: Optional[str] = field(metadata=dict(metadata=dict(min_compat="8.0")))
     tiebreaker_field: Optional[str] = field(metadata=dict(metadata=dict(min_compat="8.0")))
+    alert_suppression: Optional[AlertSuppressionMapping] = field(metadata=dict(metadata=dict(min_compat="8.14")))
 
     def convert_relative_delta(self, lookback: str) -> int:
         now = len("now")
@@ -905,6 +901,7 @@ class ESQLRuleData(QueryRuleData):
     type: Literal["esql"]
     language: Literal["esql"]
     query: str
+    alert_suppression: Optional[AlertSuppressionMapping] = field(metadata=dict(metadata=dict(min_compat="8.15")))
 
     @validates_schema
     def validates_esql_data(self, data, **kwargs):
@@ -939,6 +936,7 @@ class ThreatMapEntry:
     threat_language: Optional[definitions.FilterLanguages]
     threat_index: List[str]
     threat_indicator_path: Optional[str]
+    alert_suppression: Optional[AlertSuppressionMapping] = field(metadata=dict(metadata=dict(min_compat="8.13")))
 
     def validate_query(self, meta: RuleMeta) -> None:
         super(ThreatMatchRuleData, self).validate_query(meta)

tests/test_all_rules.py

@@ -24,7 +24,7 @@
                                           load_integrations_manifests,
                                           load_integrations_schemas)
 from detection_rules.packaging import current_stack_version
-from detection_rules.rule import (AlertSuppressionMapping, QueryRuleData, QueryValidator,
+from detection_rules.rule import (AlertSuppressionMapping, EQLRuleData, QueryRuleData, QueryValidator,
                                   ThresholdAlertSuppression, TOMLRuleContents)
 from detection_rules.rule_loader import FILE_PATTERN, RULES_CONFIG
 from detection_rules.rule_validators import EQLValidator, KQLValidator
@@ -1352,8 +1352,7 @@ class TestAlertSuppression(BaseRuleTest):
     def test_group_field_in_schemas(self):
         """Test to ensure the fields are defined is in ECS/Beats/Integrations schema."""
         for rule in self.all_rules:
-            rule_type = rule.contents.data.get('type')
-            if rule_type in ('query', 'threshold') and rule.contents.data.get('alert_suppression'):
+            if rule.contents.data.get('alert_suppression'):
                 if isinstance(rule.contents.data.alert_suppression, AlertSuppressionMapping):
                     group_by_fields = rule.contents.data.alert_suppression.group_by
                 elif isinstance(rule.contents.data.alert_suppression, ThresholdAlertSuppression):
@@ -1381,3 +1380,17 @@ def test_group_field_in_schemas(self):
                     if fld not in schema.keys():
                         self.fail(f"{self.rule_str(rule)} alert suppression field {fld} not \
                             found in ECS, Beats, or non-ecs schemas")
+
+    @unittest.skipIf(PACKAGE_STACK_VERSION < Version.parse("8.14.0"),
+                     "Test only applicable to 8.14+ stacks for eql non-sequence rule alert suppression feature.")
+    def test_eql_non_sequence_support_only(self):
+        for rule in self.all_rules:
+            if (
+                isinstance(rule.contents.data, EQLRuleData) and rule.contents.data.get("alert_suppression")
+                and rule.contents.data.is_sequence  # noqa: W503
+            ):
+                # is_sequence method not yet available during schema validation
+                # so we have to check in a unit test
+                self.fail(
+                    f"{self.rule_str(rule)} Sequence rules cannot have alert suppression"
+                )