chore(aws): fixup AWS EC2 SG lib (#9775)

Co-authored-by: Lee Trout <ltrout@watchpointlabs.com>
Co-authored-by: MrCloudSec <hello@mistercloudsec.com>
Co-authored-by: Sergio Garcia <sergargar1@gmail.com>
Co-authored-by: HugoPBrito <hugopbrit@gmail.com>

prowler/CHANGELOG.md

@@ -6,6 +6,7 @@ All notable changes to the **Prowler SDK** are documented in this file.
 
 ### Fixed
 - Fix OCI authentication error handling and validation [(#9738)](https://github.com/prowler-cloud/prowler/pull/9738)
+- Fixup AWS EC2 SG library [(#9216)](https://github.com/prowler-cloud/prowler/pull/9216)
 
 ---
 

prowler/providers/aws/services/dms/dms_instance_no_public_access/dms_instance_no_public_access.py

@@ -25,8 +25,8 @@ class dms_instance_no_public_access(Check):
                                 if check_security_group(
                                     ingress_rule,
                                     "-1",
-                                    ports=None,
                                     any_address=True,
+                                    all_ports=True,
                                 ):
                                     report.status = "FAIL"
                                     report.status_extended = f"DMS Replication Instance {instance.id} is set as publicly accessible and security group {security_group.name} ({security_group.id}) is open to the Internet."

prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py

@@ -31,7 +31,7 @@ class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check):
                     report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have any port open to the Internet."
                     for ingress_rule in security_group.ingress_rules:
                         if check_security_group(
-                            ingress_rule, "-1", ports=None, any_address=True
+                            ingress_rule, "-1", any_address=True, all_ports=True
                         ):
                             self.check_enis(
                                 report=report,

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

@@ -3,10 +3,14 @@ from typing import Any
 
 
 def check_security_group(
-    ingress_rule: Any, protocol: str, ports: list = [], any_address: bool = False
+    ingress_rule: Any,
+    protocol: str,
+    ports: list | None = None,
+    any_address: bool = False,
+    all_ports: bool = False,
 ) -> bool:
     """
-    Check if the security group ingress rule has public access to the check_ports using the protocol
+    Check if the security group ingress rule has public access to the check_ports using the protocol.
 
     @param ingress_rule: AWS Security Group IpPermissions Ingress Rule
     {
@@ -29,13 +33,17 @@ def check_security_group(
 
     @param protocol: Protocol to check. If -1, all protocols will be checked.
 
-
-    @param ports: List of ports to check. If empty, any port will be checked. If None, any port will be checked. (Default: [])
+    @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 all_ports: If True, empty ports list will be treated as all ports. (Default: False)
+
     @return: True if the security group has public access to the check_ports using the protocol
     """
+    if ports is None:
+        ports = []
+
     # Check for all traffic ingress rules regardless of the protocol
     if ingress_rule["IpProtocol"] == "-1":
         for ip_ingress_rule in ingress_rule["IpRanges"]:
@@ -54,54 +62,42 @@ def check_security_group(
 
     # Check for specific ports in ingress rules
     if "FromPort" in ingress_rule:
-        # If there is a port range
+
+        # If the ports are not the same create a covering range.
+        # Note range is exclusive of the end value so we add 1 to the ToPort.
         if ingress_rule["FromPort"] != ingress_rule["ToPort"]:
-            # Calculate port range, adding 1
-            diff = (ingress_rule["ToPort"] - ingress_rule["FromPort"]) + 1
-            ingress_port_range = []
-            for x in range(diff):
-                ingress_port_range.append(int(ingress_rule["FromPort"]) + x)
-        # If FromPort and ToPort are the same
+            ingress_port_range = set(
+                range(ingress_rule["FromPort"], ingress_rule["ToPort"] + 1)
+            )
         else:
-            ingress_port_range = []
-            ingress_port_range.append(int(ingress_rule["FromPort"]))
+            ingress_port_range = {int(ingress_rule["FromPort"])}
 
-        # Test Security Group
-        # IPv4
-        for ip_ingress_rule in ingress_rule["IpRanges"]:
-            if _is_cidr_public(ip_ingress_rule["CidrIp"], any_address):
-                # If there are input ports to check
-                if ports:
-                    for port in ports:
-                        if (
-                            port in ingress_port_range
-                            and ingress_rule["IpProtocol"] == protocol
-                        ):
-                            return True
-                # If empty input ports check if all ports are open
-                if len(set(ingress_port_range)) == 65536:
-                    return True
-                # If None input ports check if any port is open
-                if ports is None:
-                    return True
+        # Combine IPv4 and IPv6 ranges to facilitate a single check loop.
+        all_ingress_rules = []
+        all_ingress_rules.extend(ingress_rule["IpRanges"])
+        all_ingress_rules.extend(ingress_rule["Ipv6Ranges"])
 
-        # IPv6
-        for ip_ingress_rule in ingress_rule["Ipv6Ranges"]:
-            if _is_cidr_public(ip_ingress_rule["CidrIpv6"], any_address):
-                # If there are input ports to check
-                if ports:
-                    for port in ports:
-                        if (
-                            port in ingress_port_range
-                            and ingress_rule["IpProtocol"] == protocol
-                        ):
-                            return True
-                # If empty input ports check if all ports are open
-                if len(set(ingress_port_range)) == 65536:
-                    return True
-                # If None input ports check if any port is open
-                if ports is None:
-                    return True
+        for ip_ingress_rule in all_ingress_rules:
+            # We only check public CIDRs
+            if _is_cidr_public(
+                ip_ingress_rule.get("CidrIp", ip_ingress_rule.get("CidrIpv6")),
+                any_address,
+            ):
+                for port in ports:
+                    if port in ingress_port_range and (
+                        ingress_rule["IpProtocol"] == protocol or protocol == "-1"
+                    ):
+                        # Direct match for a port in the specified port range
+                        return True
+
+                # We did not find a specific port for the given protocol for
+                # a public cidr so let's see if all the ports are open
+                all_ports_open = len(ingress_port_range) == 65536
+
+                # Use the all_ports flag to determine if empty ports should be treated as all ports.
+                empty_ports_same_as_all_ports_open = all_ports and not ports
+
+                return all_ports_open or empty_ports_same_as_all_ports_open
 
     return False
 
@@ -120,3 +116,4 @@ def _is_cidr_public(cidr: str, any_address: bool = False) -> bool:
         return True
     if not any_address:
         return ipaddress.ip_network(cidr).is_global
+    return False

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

@@ -48,7 +48,7 @@ class Test_is_cidr_public:
         with pytest.raises(ValueError) as ex:
             _is_cidr_public(cidr)
 
-        assert ex.type == ValueError
+        assert ex.type is ValueError
         assert ex.match(f"{cidr} has host bits set")
 
     def test__is_cidr_public_Public_IPv6_all_IPs_any_address_false(self):
@@ -77,7 +77,7 @@ class Test_is_cidr_public:
 
 
 class Test_check_security_group:
-    def generate_ip_ranges_list(self, input_ip_ranges: [str], v4=True):
+    def generate_ip_ranges_list(self, input_ip_ranges: list[str], v4=True):
         cidr_ranges = "CidrIp" if v4 else "CidrIpv6"
         return [{cidr_ranges: ip, "Description": ""} for ip in input_ip_ranges]
 
@@ -86,8 +86,8 @@ class Test_check_security_group:
         from_port: int,
         to_port: int,
         ip_protocol: str,
-        input_ipv4_ranges: [str],
-        input_ipv6_ranges: [str],
+        input_ipv4_ranges: list[str],
+        input_ipv6_ranges: list[str],
     ):
         """
         ingress_rule_generator returns the following AWS Security Group IpPermissions Ingress Rule based on the input arguments