feat(waf): add new check waf_global_rule_with_conditions (#5465)

Co-authored-by: Sergio <[email protected]>

prowler/providers/aws/services/waf/waf_global_rule_with_conditions/waf_global_rule_with_conditions.metadata.json

@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "waf_global_rule_with_conditions",
+  "CheckTitle": "AWS WAF Classic Global Rules Should Have at Least One Condition.",
+  "CheckType": [
+    "Software and Configuration Checks/Industry and Regulatory Standards/NIST 800-53 Controls"
+  ],
+  "ServiceName": "waf",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:waf:account-id:rule/rule-id",
+  "Severity": "medium",
+  "ResourceType": "AwsWafRule",
+  "Description": "Ensure that every AWS WAF Classic Global Rule contains at least one condition.",
+  "Risk": "An AWS WAF Classic Global rule without any conditions cannot inspect or filter traffic, potentially allowing malicious requests to pass unchecked.",
+  "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/waf-global-rule-not-empty.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws waf update-rule --rule-id <your-rule-id> --change-token <your-change-token> --updates '[{\"Action\":\"INSERT\",\"Predicate\":{\"Negated\":false,\"Type\":\"IPMatch\",\"DataId\":\"<your-ipset-id>\"}}]' --region <your-region>",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/waf-controls.html#waf-6",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure that every AWS WAF Classic Global rule has at least one condition to properly inspect and manage web traffic.",
+      "Url": "https://docs.aws.amazon.com/waf/latest/developerguide/classic-web-acl-rules-editing.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/waf/waf_global_rule_with_conditions/waf_global_rule_with_conditions.py

@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.waf.waf_client import waf_client
+
+
+class waf_global_rule_with_conditions(Check):
+    def execute(self):
+        findings = []
+        for rule in waf_client.rules.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = rule.region
+            report.resource_id = rule.id
+            report.resource_arn = rule.arn
+            report.resource_tags = rule.tags
+            report.status = "FAIL"
+            report.status_extended = (
+                f"AWS WAF Global Rule {rule.name} does not have any conditions."
+            )
+
+            if rule.predicates:
+                report.status = "PASS"
+                report.status_extended = (
+                    f"AWS WAF Global Rule {rule.name} has at least one condition."
+                )
+
+            findings.append(report)
+
+        return findings

prowler/providers/aws/services/waf/waf_service.py

@@ -1,6 +1,6 @@
-from typing import Optional
+from typing import Dict, List, Optional
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from prowler.lib.logger import logger
 from prowler.lib.scan_filters.scan_filters import is_resource_filtered
@@ -11,46 +11,123 @@ class WAF(AWSService):
     def __init__(self, provider):
         # Call AWSService's __init__
         super().__init__("waf", provider)
+        self.rules = {}
+        self.rule_groups = {}
         self.web_acls = {}
-        self.__threading_call__(self._list_web_acls)
+        self._list_rules()
+        self.__threading_call__(self._get_rule, self.rules.values())
+        self._list_rule_groups()
         self.__threading_call__(
-            self._list_resources_for_web_acl, self.web_acls.values()
+            self._list_activated_rules_in_rule_group, self.rule_groups.values()
         )
+        self._list_web_acls()
+        self.__threading_call__(self._get_web_acl, self.web_acls.values())
 
-    def _list_web_acls(self, regional_client):
-        logger.info("WAF - Listing Regional Web ACLs...")
+    def _list_rules(self):
+        logger.info("WAF - Listing Global Rules...")
         try:
-            for waf in regional_client.list_web_acls()["WebACLs"]:
+            for rule in self.client.list_rules().get("Rules", []):
+                arn = f"arn:{self.audited_partition}:waf:{self.region}:{self.audited_account}:rule/{rule['RuleId']}"
+                self.rules[arn] = Rule(
+                    arn=arn,
+                    id=rule.get("RuleId", ""),
+                    region=self.region,
+                    name=rule.get("Name", ""),
+                )
+
+        except Exception as error:
+            logger.error(
+                f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _get_rule(self, rule):
+        logger.info(f"WAF - Getting Global Rule {rule.name}...")
+        try:
+            get_rule = self.client.get_rule(RuleId=rule.id)
+            for predicate in get_rule.get("Rule", {}).get("Predicates", []):
+                rule.predicates.append(
+                    Predicate(
+                        negated=predicate.get("Negated", False),
+                        type=predicate.get("Type", "IPMatch"),
+                        data_id=predicate.get("DataId", ""),
+                    )
+                )
+
+        except Exception as error:
+            logger.error(
+                f"{rule.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _list_rule_groups(self):
+        logger.info("WAF - Listing Global Rule Groups...")
+        try:
+            for rule_group in self.client.list_rule_groups().get("RuleGroups", []):
+                arn = f"arn:{self.audited_partition}:waf:{self.region}:{self.audited_account}:rulegroup/{rule_group['RuleGroupId']}"
+                self.rule_groups[arn] = RuleGroup(
+                    arn=arn,
+                    region=self.region,
+                    id=rule_group.get("RuleGroupId", ""),
+                    name=rule_group.get("Name", ""),
+                )
+
+        except Exception as error:
+            logger.error(
+                f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _list_activated_rules_in_rule_group(self, rule_group):
+        logger.info(
+            f"WAF - Listing activated rules in Global Rule Group {rule_group.name}..."
+        )
+        try:
+            for rule in self.client.list_activated_rules_in_rule_group(
+                RuleGroupId=rule_group.id
+            ).get("ActivatedRules", []):
+                rule_arn = f"arn:{self.audited_partition}:waf:{self.region}:{self.audited_account}:rule/{rule.get('RuleId', '')}"
+                rule_group.rules.append(self.rules[rule_arn])
+
+        except Exception as error:
+            logger.error(
+                f"{rule_group.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _list_web_acls(self):
+        logger.info("WAF - Listing Global Web ACLs...")
+        try:
+            for waf in self.client.list_web_acls()["WebACLs"]:
                 if not self.audit_resources or (
                     is_resource_filtered(waf["WebACLId"], self.audit_resources)
                 ):
-                    arn = f"arn:{self.audited_partition}:waf:{regional_client.region}:{self.audited_account}:webacl/{waf['WebACLId']}"
+                    arn = f"arn:{self.audited_partition}:waf:{self.region}:{self.audited_account}:webacl/{waf['WebACLId']}"
                     self.web_acls[arn] = WebAcl(
                         arn=arn,
                         name=waf["Name"],
                         id=waf["WebACLId"],
                         albs=[],
-                        region=regional_client.region,
+                        region=self.region,
                     )
 
         except Exception as error:
             logger.error(
-                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
-    def _list_resources_for_web_acl(self, regional_client):
-        logger.info("WAF - Describing resources...")
+    def _get_web_acl(self, acl):
+        logger.info(f"WAF - Getting Global Web ACL {acl.name}...")
         try:
-            for acl in self.web_acls.values():
-                if acl.region == regional_client.region:
-                    for resource in regional_client.list_resources_for_web_acl(
-                        WebACLId=acl.id, ResourceType="APPLICATION_LOAD_BALANCER"
-                    )["ResourceArns"]:
-                        acl.albs.append(resource)
+            get_web_acl = self.client.get_web_acl(WebACLId=acl.id)
+            for rule in get_web_acl.get("WebACL", {}).get("Rules", []):
+                rule_id = rule.get("RuleId", "")
+                if rule.get("Type", "") == "GROUP":
+                    rule_group_arn = f"arn:{self.audited_partition}:waf:{self.region}:{self.audited_account}:rulegroup/{rule_id}"
+                    acl.rule_groups.append(self.rule_groups[rule_group_arn])
+                else:
+                    rule_arn = f"arn:{self.audited_partition}:waf:{self.region}:{self.audited_account}:rule/{rule_id}"
+                    acl.rules.append(self.rules[rule_arn])
 
         except Exception as error:
             logger.error(
-                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                f"{acl.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
 
@@ -99,8 +176,10 @@ def _get_rule(self, rule):
                         data_id=predicate.get("DataId", ""),
                     )
                 )
-        except KeyError:
-            logger.error(f"Rule {rule.name} not found in {rule.region}.")
+        except Exception as error:
+            logger.error(
+                f"{rule.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
 
     def _list_rule_groups(self, regional_client):
         logger.info("WAFRegional - Listing Regional Rule Groups...")
@@ -159,7 +238,7 @@ def _list_web_acls(self, regional_client):
             )
 
     def _get_web_acl(self, acl):
-        logger.info(f"WAFRegional - Getting Web ACL {acl.name}...")
+        logger.info(f"WAFRegional - Getting Regional Web ACL {acl.name}...")
         try:
             get_web_acl = self.regional_clients[acl.region].get_web_acl(WebACLId=acl.id)
             for rule in get_web_acl.get("WebACL", {}).get("Rules", []):
@@ -207,8 +286,8 @@ class Rule(BaseModel):
     id: str
     region: str
     name: str
-    predicates: list[Predicate] = []
-    tags: Optional[list] = []
+    predicates: Optional[List[Predicate]] = Field(default_factory=list)
+    tags: Optional[List[Dict[str, str]]] = Field(default_factory=list)
 
 
 class RuleGroup(BaseModel):
@@ -218,8 +297,8 @@ class RuleGroup(BaseModel):
     id: str
     region: str
     name: str
-    rules: list[Rule] = []
-    tags: Optional[list] = []
+    rules: List[Rule] = Field(default_factory=list)
+    tags: Optional[List[Dict[str, str]]] = Field(default_factory=list)
 
 
 class WebAcl(BaseModel):
@@ -228,8 +307,8 @@ class WebAcl(BaseModel):
     arn: str
     name: str
     id: str
-    albs: list[str]
+    albs: List[str] = Field(default_factory=list)
     region: str
-    rules: list[Rule] = []
-    rule_groups: list[RuleGroup] = []
-    tags: Optional[list] = []
+    rules: List[Rule] = Field(default_factory=list)
+    rule_groups: List[RuleGroup] = Field(default_factory=list)
+    tags: Optional[List[Dict[str, str]]] = Field(default_factory=list)