chore: Move shared to lib/ for AWS (#1321)

* chore: Move shared to lib/

* chore: Move shared to lib/ for AWS

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>

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

@@ -1,5 +1,6 @@
 from lib.check.models import Check, Check_Report
-from providers.aws.services.ec2.ec2_service import check_network_acl, ec2_client
+from providers.aws.services.ec2.ec2_service import ec2_client
+from providers.aws.services.ec2.lib.network_acls import check_network_acl
 
 
 class ec2_networkacl_allow_ingress_tcp_port_22(Check):

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

@@ -1,5 +1,6 @@
 from lib.check.models import Check, Check_Report
-from providers.aws.services.ec2.ec2_service import check_network_acl, ec2_client
+from providers.aws.services.ec2.ec2_service import ec2_client
+from providers.aws.services.ec2.lib.network_acls import check_network_acl
 
 
 class ec2_networkacl_allow_ingress_tcp_port_3389(Check):

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

@@ -1,5 +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_any_port(Check):

providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22.py

@@ -1,5 +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_port_22(Check):

providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389.py

@@ -1,5 +1,7 @@
 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_port_3389(Check):
     def execute(self):

providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306.py

@@ -1,5 +1,7 @@
 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_port_mysql_3306(Check):
     def execute(self):

providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483/ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483.py

@@ -1,5 +1,7 @@
 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_port_oracle_1521_2483(Check):
     def execute(self):
@@ -23,4 +25,4 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483
                     report.resource_id = security_group.id
                 findings.append(report)
 
-        return findings
+        return findings

providers/aws/services/ec2/ec2_service.py

@@ -1,6 +1,5 @@
 import threading
 from dataclasses import dataclass
-from typing import Any
 
 from lib.logger import logger
 from providers.aws.aws_provider import current_audit_info, generate_regional_clients
@@ -238,76 +237,3 @@ class NetworkACL:
 
 
 ec2_client = EC2(current_audit_info)
-
-################## Security Groups
-# Check if the security group ingress rule has public access to the check_ports using the protocol
-def check_security_group(ingress_rule: Any, protocol: str, ports: list = []) -> bool:
-    public_IPv4 = "0.0.0.0/0"
-    public_IPv6 = "::/0"
-
-    # Check for all traffic ingress rules regardless of the protocol
-    if ingress_rule["IpProtocol"] == "-1" and (
-        (
-            "0.0.0.0/0" in str(ingress_rule["IpRanges"])
-            or "::/0" in str(ingress_rule["Ipv6Ranges"])
-        )
-    ):
-        return True
-
-    # Check for specific ports in ingress rules
-    if "FromPort" in ingress_rule:
-        # All ports
-        if ingress_rule["FromPort"] == 0 and ingress_rule["ToPort"] == 65535:
-            return True
-
-        # If there is a port range
-        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
-        else:
-            ingress_port_range = []
-            ingress_port_range.append(int(ingress_rule["FromPort"]))
-
-        # Test Security Group
-        for port in ports:
-            if (
-                (
-                    public_IPv4 in str(ingress_rule["IpRanges"])
-                    or public_IPv6 in str(ingress_rule["Ipv6Ranges"])
-                )
-                and port in ingress_port_range
-                and ingress_rule["IpProtocol"] == protocol
-            ):
-                return True
-    return False
-
-
-################## 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"
-
-    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
-        ):
-            return True
-
-    return False

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

@@ -0,0 +1,28 @@
+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"
+
+    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
+        ):
+            return True
+
+    return False

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

@@ -0,0 +1,48 @@
+from typing import Any
+
+
+################## Security Groups
+# Check if the security group ingress rule has public access to the check_ports using the protocol
+def check_security_group(ingress_rule: Any, protocol: str, ports: list = []) -> bool:
+    public_IPv4 = "0.0.0.0/0"
+    public_IPv6 = "::/0"
+
+    # Check for all traffic ingress rules regardless of the protocol
+    if ingress_rule["IpProtocol"] == "-1" and (
+        (
+            "0.0.0.0/0" in str(ingress_rule["IpRanges"])
+            or "::/0" in str(ingress_rule["Ipv6Ranges"])
+        )
+    ):
+        return True
+
+    # Check for specific ports in ingress rules
+    if "FromPort" in ingress_rule:
+        # All ports
+        if ingress_rule["FromPort"] == 0 and ingress_rule["ToPort"] == 65535:
+            return True
+
+        # If there is a port range
+        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
+        else:
+            ingress_port_range = []
+            ingress_port_range.append(int(ingress_rule["FromPort"]))
+
+        # Test Security Group
+        for port in ports:
+            if (
+                (
+                    public_IPv4 in str(ingress_rule["IpRanges"])
+                    or public_IPv6 in str(ingress_rule["Ipv6Ranges"])
+                )
+                and port in ingress_port_range
+                and ingress_rule["IpProtocol"] == protocol
+            ):
+                return True
+    return False