fix(ec2): treat CIDRs that together cover every address as open to the internet (#13191)

Co-authored-by: pedrooot <pedromarting3@gmail.com>

prowler/changelog.d/ec2-sg-split-cidrs-any-address.fixed.md

@@ -0,0 +1 @@
+Security group internet exposure checks now treat CIDRs in one ingress rule that together cover 0.0.0.0/0 or ::/0 as open to the internet

prowler/providers/aws/services/ec2/lib/security_groups.py

@@ -35,7 +35,7 @@ def check_security_group(
 
     @param ports: List of ports to check. If not provided all ports will be checked unless all_ports is False. (Default: None)
 
-    @param any_address: If True, only 0.0.0.0/0 or "::/0" will be public and do not search for public addresses. (Default: False)
+    @param any_address: If True, only 0.0.0.0/0 or "::/0", or CIDRs in the same ingress rule that together cover every address of one IP version, will be public and do not search for public addresses. (Default: False)
 
     @param all_ports: If True, empty ports list will be treated as all ports. (Default: False)
 
@@ -44,8 +44,14 @@ def check_security_group(
     if ports is None:
         ports = []
 
+    # With any_address, CIDRs that together cover every address in one rule, such as
+    # 0.0.0.0/1 and 128.0.0.0/1, are as open as 0.0.0.0/0.
+    rule_covers_all_addresses = any_address and _covers_all_addresses(ingress_rule)
+
     # Check for all traffic ingress rules regardless of the protocol
     if ingress_rule["IpProtocol"] == "-1":
+        if rule_covers_all_addresses:
+            return True
         for ip_ingress_rule in ingress_rule["IpRanges"]:
             if _is_cidr_public(ip_ingress_rule["CidrIp"], any_address):
                 return True
@@ -79,7 +85,7 @@ def check_security_group(
 
         for ip_ingress_rule in all_ingress_rules:
             # We only check public CIDRs
-            if _is_cidr_public(
+            if rule_covers_all_addresses or _is_cidr_public(
                 ip_ingress_rule.get("CidrIp", ip_ingress_rule.get("CidrIpv6")),
                 any_address,
             ):
@@ -102,6 +108,27 @@ def check_security_group(
     return False
 
 
+def _covers_all_addresses(ingress_rule: Any) -> bool:
+    """
+    Check if the CIDRs of one IP version in the ingress rule together cover the whole address space
+
+    @param ingress_rule: AWS Security Group IpPermissions Ingress Rule
+    """
+    for key, cidr_key in (("IpRanges", "CidrIp"), ("Ipv6Ranges", "CidrIpv6")):
+        try:
+            networks = [
+                ipaddress.ip_network(ip_range[cidr_key], strict=False)
+                for ip_range in ingress_rule.get(key, [])
+            ]
+        except ValueError:
+            continue
+        if any(
+            network.prefixlen == 0 for network in ipaddress.collapse_addresses(networks)
+        ):
+            return True
+    return False
+
+
 def _is_cidr_public(cidr: str, any_address: bool = False) -> bool:
     """
     Check if an input CIDR is public

tests/providers/aws/services/ec2/lib/security_groups_test.py

@@ -411,3 +411,64 @@ class Test_check_security_group:
             port, port, TRANSPORT_PROTOCOL_ALL, [], [IP_V6_ALL_CIDRS]
         )
         assert check_security_group(ingress_rule, TRANSPORT_PROTOCOL_ALL, None, True)
+
+    # TCP Protocol - two halves of IPv4 in one rule - Ingress 22 to 22 - check 22 - Any Address - Open
+    def test_split_ipv4_halves_open_22_tcp_any_address(self):
+        port = 22
+        ingress_rule = self.ingress_rule_generator(
+            port, port, TRANSPORT_PROTOCOL_TCP, ["0.0.0.0/1", "128.0.0.0/1"], []
+        )
+        assert check_security_group(ingress_rule, TRANSPORT_PROTOCOL_TCP, [port], True)
+
+    # TCP Protocol - two halves of IPv6 in one rule - Ingress 22 to 22 - check 22 - Any Address - Open
+    def test_split_ipv6_halves_open_22_tcp_any_address(self):
+        port = 22
+        ingress_rule = self.ingress_rule_generator(
+            port, port, TRANSPORT_PROTOCOL_TCP, [], ["::/1", "8000::/1"]
+        )
+        assert check_security_group(ingress_rule, TRANSPORT_PROTOCOL_TCP, [port], True)
+
+    # Any protocol - two halves of IPv4 in one rule - check 22 - Any Address - Open
+    def test_split_ipv4_halves_any_protocol_any_address(self):
+        ingress_rule = self.ingress_rule_generator(
+            0, 65535, TRANSPORT_PROTOCOL_ALL, ["0.0.0.0/1", "128.0.0.0/1"], []
+        )
+        assert check_security_group(ingress_rule, TRANSPORT_PROTOCOL_TCP, [22], True)
+
+    # TCP Protocol - one half of IPv4 - Ingress 22 to 22 - check 22 - Any Address - Closed
+    def test_one_ipv4_half_open_22_tcp_any_address(self):
+        port = 22
+        ingress_rule = self.ingress_rule_generator(
+            port, port, TRANSPORT_PROTOCOL_TCP, ["0.0.0.0/1"], []
+        )
+        assert not check_security_group(
+            ingress_rule, TRANSPORT_PROTOCOL_TCP, [port], True
+        )
+
+    # TCP Protocol - one half of IPv4 and one half of IPv6 - Ingress 22 to 22 - check 22 - Any Address - Closed
+    def test_one_ipv4_half_and_one_ipv6_half_open_22_tcp_any_address(self):
+        port = 22
+        ingress_rule = self.ingress_rule_generator(
+            port, port, TRANSPORT_PROTOCOL_TCP, ["0.0.0.0/1"], ["8000::/1"]
+        )
+        assert not check_security_group(
+            ingress_rule, TRANSPORT_PROTOCOL_TCP, [port], True
+        )
+
+    # TCP Protocol - two halves of IPv4 - Ingress 80 to 80 - check 22 - Any Address - Closed
+    def test_split_ipv4_halves_other_port_any_address(self):
+        ingress_rule = self.ingress_rule_generator(
+            80, 80, TRANSPORT_PROTOCOL_TCP, ["0.0.0.0/1", "128.0.0.0/1"], []
+        )
+        assert not check_security_group(
+            ingress_rule, TRANSPORT_PROTOCOL_TCP, [22], True
+        )
+
+    # The rule passed in is not modified, since fixers send it back to AWS
+    def test_split_ipv4_halves_rule_unchanged(self):
+        ingress_rule = self.ingress_rule_generator(
+            22, 22, TRANSPORT_PROTOCOL_TCP, ["0.0.0.0/1", "128.0.0.0/1"], []
+        )
+        before = repr(ingress_rule)
+        check_security_group(ingress_rule, TRANSPORT_PROTOCOL_TCP, [22], True)
+        assert repr(ingress_rule) == before

tests/providers/aws/services/emr/emr_cluster_publicly_accesible/emr_cluster_publicly_accesible_test.py

@@ -13,12 +13,11 @@ from tests.providers.aws.utils import (
 
 
 class Test_emr_cluster_publicly_accesible:
+    @mock_aws
     def test_no_clusters(self):
         # EMR Client
         emr_client = mock.MagicMock
         emr_client.clusters = {}
-        # EC2 Client
-        ec2_client = mock.MagicMock
 
         with (
             mock.patch(
@@ -26,8 +25,12 @@ class Test_emr_cluster_publicly_accesible:
                 new=emr_client,
             ),
             mock.patch(
-                "prowler.providers.aws.services.ec2.ec2_service.EC2",
-                new=ec2_client,
+                "prowler.providers.common.provider.Provider.get_global_provider",
+                return_value=set_mocked_aws_provider(),
+            ),
+            mock.patch(
+                "prowler.providers.aws.services.emr.emr_cluster_publicly_accesible.emr_cluster_publicly_accesible.ec2_client",
+                new=mock.MagicMock(),
             ),
         ):
             # Test Check