fix(check_network_acl): check with all rules together (#1350)

providers/aws/services/ec2/ec2_networkacl_allow_ingress_tcp_port_22/ec2_networkacl_allow_ingress_tcp_port_22.py

@@ -9,20 +9,10 @@ class ec2_networkacl_allow_ingress_tcp_port_22(Check):
         tcp_protocol = "6"
         check_port = 22
         for network_acl in ec2_client.network_acls:
-            public = False
             report = Check_Report(self.metadata)
             report.region = network_acl.region
-            for entry in network_acl.entries:
-                # For IPv4
-                if "CidrBlock" in entry:
-                    public = check_network_acl(entry, tcp_protocol, check_port, "IPv4")
-                # For IPv6
-                if "Ipv6CidrBlock" in entry:
-                    public = check_network_acl(entry, tcp_protocol, check_port, "IPv6")
-                # If some entry allows it, that ACL is not securely configured
-                if public:
-                    break
-            if not public:
+            # If some entry allows it, that ACL is not securely configured
+            if not check_network_acl(network_acl.entries, tcp_protocol, check_port):
                 report.status = "PASS"
                 report.status_extended = f"Network ACL {network_acl.id} has not SSH port 22 open to the Internet."
                 report.resource_id = network_acl.id

providers/aws/services/ec2/ec2_networkacl_allow_ingress_tcp_port_3389/ec2_networkacl_allow_ingress_tcp_port_3389.py

@@ -9,17 +9,10 @@ class ec2_networkacl_allow_ingress_tcp_port_3389(Check):
         tcp_protocol = "6"
         check_port = 3389
         for network_acl in ec2_client.network_acls:
-            public = False
             report = Check_Report(self.metadata)
             report.region = network_acl.region
-            for entry in network_acl.entries:
-                # For IPv4
-                if "CidrBlock" in entry:
-                    public = check_network_acl(entry, tcp_protocol, check_port, "IPv4")
-                # For IPv6
-                if "Ipv6CidrBlock" in entry:
-                    public = check_network_acl(entry, tcp_protocol, check_port, "IPv6")
-            if not public:
+            # If some entry allows it, that ACL is not securely configured
+            if not check_network_acl(network_acl.entries, tcp_protocol, check_port):
                 report.status = "PASS"
                 report.status_extended = f"Network ACL {network_acl.id} has not Microsoft RDP port 3389 open to the Internet."
                 report.resource_id = network_acl.id

providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21/ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21.py

@@ -1,6 +1,6 @@
 from lib.check.models import Check, Check_Report
-from providers.aws.services.ec2.ec2_service import check_security_group, ec2_client
-
+from providers.aws.services.ec2.ec2_service import ec2_client
+from providers.aws.services.ec2.lib.security_groups import check_security_group
 
 class ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21(Check):
     def execute(self):

providers/aws/services/ec2/lib/network_acls.py

@@ -1,27 +1,87 @@
+from re import T
 from typing import Any
 
 
-################## Network ACLs
-# Check if the network acls ingress rule has public access to the check_ports using the protocol
-def check_network_acl(entry: Any, protocol: str, port: str, ip_version: str) -> bool:
-    # For IPv4
-    if ip_version == "IPv4":
-        entry_value = "CidrBlock"
-        public_ip = "0.0.0.0/0"
-    # For IPv6
-    elif ip_version == "IPv6":
-        entry_value = "Ipv6CidrBlock"
-        public_ip = "::/0"
+# Network ACLs
+# Check if the network acls rules has ingress public access to the check_ports using the protocol
+def check_network_acl(rules: Any, protocol: str, port: str) -> bool:
 
-    if (
-        entry[entry_value] == public_ip
-        and entry["RuleAction"] == "allow"
-        and not entry["Egress"]
-    ):
-        if entry["Protocol"] == "-1" or (
-            entry["PortRange"]["From"] == port
-            and entry["PortRange"]["To"] == port
-            and entry["Protocol"] == protocol
+    # Spliting IPv6 from IPv4 rules
+    rules_IPv6 = list(
+        filter(lambda rule: rule.get("CidrBlock") is None and not rule["Egress"], rules))
+
+    # For IPv6
+    # Rules must order by RuleNumber
+    for rule in sorted(rules_IPv6, key=lambda rule: rule["RuleNumber"]):
+        if (
+            rule["Ipv6CidrBlock"] == "::/0"
+            and rule["RuleAction"] == "deny"
+            and (
+                rule["Protocol"] == "-1"
+                or
+                (
+                    rule["Protocol"] == protocol
+                    and
+                    rule["PortRange"]["From"] <= port <= rule["PortRange"]["To"]
+                )
+            )
+        ):
+            # Exist IPv6 deny for this port
+            break
+
+        if (
+            rule["Ipv6CidrBlock"] == "::/0"
+            and rule["RuleAction"] == "allow"
+            and (
+                rule["Protocol"] == "-1"
+                or
+                (
+                    rule["Protocol"] == protocol
+                    and
+                    rule["PortRange"]["From"] <= port <= rule["PortRange"]["To"]
+                )
+            )
+        ):
+            # Exist IPv6 allow for this port
+            return True
+
+    # There are not IPv6 Public access here
+
+    # Spliting IPv4 from IPv6 rules
+    rules_IPv4 = list(filter(lambda rule: rule.get("Ipv6CidrBlock") is None and not rule["Egress"], rules))
+
+    # For IPv4
+    # Rules must order by RuleNumber
+    for rule in sorted(rules_IPv4, key=lambda rule: rule["RuleNumber"]):
+        if (
+            rule["CidrBlock"] == "0.0.0.0/0"
+            and rule["RuleAction"] == "deny"
+            and (
+                rule["Protocol"] == "-1"
+                or
+                (
+                    rule["Protocol"] == protocol
+                    and
+                    rule["PortRange"]["From"] <= port <= rule["PortRange"]["To"]
+                )
+            )
+        ):
+
+            # Exist IPv4 deny for this port and if exist IPv6 there are not IPv6 Public access here
+            return False
+
+        if (
+            rule["CidrBlock"] == "0.0.0.0/0"
+            and rule["RuleAction"] == "allow"
+            and (
+                rule["Protocol"] == "-1"
+                or
+                (
+                    rule["Protocol"] == protocol
+                    and
+                    rule["PortRange"]["From"] <= port <= rule["PortRange"]["To"]
+                )
+            )
         ):
             return True