Merge branch 'main' into terrancedejesus/issue5189

Author: terrancedejesus

detection_rules/rule.py

@@ -1074,10 +1074,17 @@ def validate(self, meta: RuleMeta) -> None:  # noqa: ARG002
 
 # All of the possible rule types
 # Sort inverse of any inheritance - see comment in TOMLRuleContents.to_dict
+# ThresholdQueryRuleData needs to be first in this union to handle cases where there is ambiguity between
+# ThresholdAlertSuppression and AlertSuppressionMapping. Since AlertSuppressionMapping has duration as an
+# optional field, ThresholdAlertSuppression objects can be mistakenly loaded as an AlertSuppressionMapping
+# object with group_by and missing_fields_strategy as missing parameters, resulting in an error.
+# Checking the type against ThresholdQueryRuleData first in the union prevent this from occurring.
+# Please also keep issue 1141 in mind when handling union schemas.
+
 AnyRuleData = (
-    EQLRuleData
+    ThresholdQueryRuleData
+    | EQLRuleData
     | ESQLRuleData
-    | ThresholdQueryRuleData
     | ThreatMatchRuleData
     | MachineLearningRuleData
     | QueryRuleData

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "detection_rules"
-version = "1.4.11"
+version = "1.4.12"
 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"

rules/linux/persistence_simple_web_server_connection_accepted.toml

@@ -2,7 +2,7 @@
 creation_date = "2024/12/17"
 integration = ["endpoint"]
 maturity = "production"
-updated_date = "2025/01/15"
+updated_date = "2025/10/13"
 
 [rule]
 author = ["Elastic"]
@@ -13,7 +13,7 @@ payload to the server web root, allowing them to regain remote access to the sys
 an attacker requests the server to execute a command or script via a potential backdoor.
 """
 from = "now-9m"
-index = ["logs-endpoint.events.network*"]
+index = ["logs-endpoint.events.process*", "logs-endpoint.events.network*"]
 language = "eql"
 license = "Elastic License v2"
 name = "Simple HTTP Web Server Connection"
@@ -58,10 +58,13 @@ tags = [
 timestamp_override = "event.ingested"
 type = "eql"
 query = '''
-network where host.os.type == "linux" and event.type == "start" and event.action == "connection_accepted" and (
+sequence by process.entity_id with maxspan=1m
+[process where host.os.type == "linux" and event.type == "start" and 
+ (
   (process.name regex~ """php?[0-9]?\.?[0-9]{0,2}""" and process.command_line like "*-S*") or
   (process.name like "python*" and process.command_line like ("*--cgi*", "*CGIHTTPServer*"))
-)
+ )]
+[network where host.os.type == "linux" and event.type == "start" and event.action == "connection_accepted"]
 '''
 note = """## Triage and analysis
 

tests/test_python_library.py

@@ -3,7 +3,10 @@
 # 2.0; you may not use this file except in compliance with the Elastic License
 # 2.0.
 
+from typing import Any
+
 import eql
+from marshmallow import ValidationError
 
 from detection_rules.rule_loader import RuleCollection
 
@@ -22,26 +25,46 @@ def mk_metadata(integrations: list[str], comments: str = "Test metadata") -> dic
     }
 
 
-def mk_rule(
+def mk_rule(  # noqa: PLR0913
     *,
     name: str,
     rule_id: str,
     description: str,
     risk_score: int,
     query: str,
-) -> dict:
+    language: str = "eql",
+    query_type: str = "eql",
+    threshold: dict[str, Any] | None = None,
+    alert_suppression: dict[str, Any] | None = None,
+    index: list[str] | None = None,
+    threat_language: str | None = None,
+    threat_index: list[str] | None = None,
+    threat_indicator_path: str | None = None,
+    threat_mapping: list[Any] | None = None,
+) -> dict[str, Any]:
     """Create rule dictionary."""
-    return {
+    rule = {
         "author": ["Elastic"],
         "description": description,
-        "language": "eql",
+        "language": language,
         "name": name,
         "risk_score": risk_score,
         "rule_id": rule_id,
         "severity": "low",
-        "type": "eql",
+        "type": query_type,
         "query": query,
+        "alert_suppression": alert_suppression,
     }
+    if threshold is not None:
+        rule["threshold"] = threshold
+    if query_type == "threat_match":
+        rule["index"] = index
+        rule["threat_language"] = threat_language
+        rule["threat_index"] = threat_index
+        rule["threat_indicator_path"] = threat_indicator_path
+        rule["threat_mapping"] = threat_mapping
+
+    return rule
 
 
 class TestEQLInSet(BaseRuleTest):
@@ -283,3 +306,196 @@ def test_sequence_datasetless_subquery_with_metadata_integration_valid(self) ->
             ),
         }
         rc.load_dict(rule)
+
+
+class TestAlertSuppressionValidation(BaseRuleTest):
+    """Tests for alert_suppression field validation in rules."""
+
+    def test_threshold_rule_duration(self) -> None:
+        """Test that a threshold rule with alert_suppression with just duration validates correctly."""
+        rc = RuleCollection()
+        query = """
+        process.name: \"test\"
+        """
+        rule_dict: dict[str, Any] = {
+            "metadata": mk_metadata(
+                ["endpoint", "windows"], comments="New fields added: required_fields, related_integrations, setup"
+            ),
+            "rule": mk_rule(
+                name="Fake Test Rule",
+                rule_id="4fffae5d-8b7d-4e48-88b1-979ed42fd9a3",
+                description="Test Rule.",
+                risk_score=47,
+                query=query,
+                language="kuery",
+                query_type="threshold",
+                threshold={"field": [], "value": 200, "cardinality": []},
+                alert_suppression={"duration": {"value": 5, "unit": "h"}},
+            ),
+        }
+        _ = rc.load_dict(rule_dict)
+
+    def test_query_rule_duration(self) -> None:
+        """Test that a query rule with alert_suppression with group_by and missing_fields_strategy validates correctly."""
+        rc = RuleCollection()
+        query = """
+        process.name: \"test\"
+        """
+        rule_dict: dict[str, Any] = {
+            "metadata": mk_metadata(
+                ["endpoint", "windows"], comments="New fields added: required_fields, related_integrations, setup"
+            ),
+            "rule": mk_rule(
+                name="Fake Test Rule",
+                rule_id="4fffae5d-8b7d-4e48-88b1-979ed42fd9a3",
+                description="Test Rule.",
+                risk_score=47,
+                query=query,
+                language="kuery",
+                query_type="query",
+                threshold=None,
+                alert_suppression={"duration": {"value": 5, "unit": "h"}},
+            ),
+        }
+        with self.assertRaises((ValidationError, TypeError)):
+            _ = rc.load_dict(rule_dict)
+
+    def test_query_rule_group_by_missing_fields(self) -> None:
+        """Test that a query rule with alert_suppression with group_by and missing_fields_strategy validates correctly."""
+        rc = RuleCollection()
+        query = """
+        process.name: \"test\"
+        """
+        rule_dict: dict[str, Any] = {
+            "metadata": mk_metadata(
+                ["endpoint", "windows"], comments="New fields added: required_fields, related_integrations, setup"
+            ),
+            "rule": mk_rule(
+                name="Fake Test Rule",
+                rule_id="4fffae5d-8b7d-4e48-88b1-979ed42fd9a3",
+                description="Test Rule.",
+                risk_score=47,
+                query=query,
+                language="kuery",
+                query_type="query",
+                threshold=None,
+                alert_suppression={"group_by": ["process.id"], "missing_fields_strategy": "suppress"},
+            ),
+        }
+        _ = rc.load_dict(rule_dict)
+
+    def test_query_rule_group_by(self) -> None:
+        """Test that a query rule with alert_suppression with just group_by is not valid."""
+        rc = RuleCollection()
+        query = """
+        process.name: \"test\"
+        """
+        rule_dict: dict[str, Any] = {
+            "metadata": mk_metadata(
+                ["endpoint", "windows"], comments="New fields added: required_fields, related_integrations, setup"
+            ),
+            "rule": mk_rule(
+                name="Fake Test Rule",
+                rule_id="4fffae5d-8b7d-4e48-88b1-979ed42fd9a3",
+                description="Test Rule.",
+                risk_score=47,
+                query=query,
+                language="kuery",
+                query_type="query",
+                threshold=None,
+                alert_suppression={"group_by": ["process.id"]},
+            ),
+        }
+        with self.assertRaises((ValidationError, TypeError)):
+            _ = rc.load_dict(rule_dict)
+
+    def test_query_rule_missing_fields_strategy(self) -> None:
+        """Test that a query rule with alert_suppression with just missing_fields_strategy is not valid."""
+        rc = RuleCollection()
+        query = """
+        process.name: \"test\"
+        """
+        rule_dict: dict[str, Any] = {
+            "metadata": mk_metadata(
+                ["endpoint", "windows"], comments="New fields added: required_fields, related_integrations, setup"
+            ),
+            "rule": mk_rule(
+                name="Fake Test Rule",
+                rule_id="4fffae5d-8b7d-4e48-88b1-979ed42fd9a3",
+                description="Test Rule.",
+                risk_score=47,
+                query=query,
+                language="kuery",
+                query_type="query",
+                threshold=None,
+                alert_suppression={"missing_fields_strategy": "suppress"},
+            ),
+        }
+        with self.assertRaises((ValidationError, TypeError)):
+            _ = rc.load_dict(rule_dict)
+
+    def test_threat_match_rule(self) -> None:
+        """Test that a threat_match rule with alert_suppression with all fields set is valid."""
+        rc = RuleCollection()
+        query = """
+        process.name: \"test\"
+        """
+        rule_dict: dict[str, Any] = {
+            "metadata": mk_metadata(
+                ["endpoint", "windows"], comments="New fields added: required_fields, related_integrations, setup"
+            ),
+            "rule": mk_rule(
+                name="Fake Test Rule",
+                rule_id="4fffae5d-8b7d-4e48-88b1-979ed42fd9a3",
+                description="Test Rule.",
+                risk_score=47,
+                query=query,
+                language="kuery",
+                query_type="threat_match",
+                threshold=None,
+                alert_suppression={
+                    "group_by": ["client.ip"],
+                    "duration": {"value": 12, "unit": "h"},
+                    "missing_fields_strategy": "suppress",
+                },
+                index=["logs-*"],
+                threat_language="kuery",
+                threat_index=["logs-*"],
+                threat_indicator_path="threat.indicator",
+                threat_mapping=[{"entries": [{"field": "client.ip", "type": "mapping", "value": "client.ip"}]}],
+            ),
+        }
+        _ = rc.load_dict(rule_dict)
+
+    def test_threat_match_rule_missing_fields_duration(self) -> None:
+        """Test that a threat_match  rule with alert_suppression with missing_fields_strategy and duration is not valid."""
+        rc = RuleCollection()
+        query = """
+        process.name: \"test\"
+        """
+        rule_dict: dict[str, Any] = {
+            "metadata": mk_metadata(
+                ["endpoint", "windows"], comments="New fields added: required_fields, related_integrations, setup"
+            ),
+            "rule": mk_rule(
+                name="Fake Test Rule",
+                rule_id="4fffae5d-8b7d-4e48-88b1-979ed42fd9a3",
+                description="Test Rule.",
+                risk_score=47,
+                query=query,
+                language="kuery",
+                query_type="threat_match",
+                threshold=None,
+                alert_suppression={
+                    "duration": {"value": 12, "unit": "h"},
+                    "missing_fields_strategy": "suppress",
+                },
+                index=["logs-*"],
+                threat_language="kuery",
+                threat_index=["logs-*"],
+                threat_indicator_path="threat.indicator",
+                threat_mapping=[{"entries": [{"field": "client.ip", "type": "mapping", "value": "client.ip"}]}],
+            ),
+        }
+        with self.assertRaises((ValidationError, TypeError)):
+            _ = rc.load_dict(rule_dict)