From 103d176a38bd4ca05626a9c1cdbcf4da0df7e929 Mon Sep 17 00:00:00 2001 From: Prowler Bot Date: Tue, 13 Jan 2026 08:39:53 +0100 Subject: [PATCH] chore(aws): fixup AWS EC2 SG lib (#9775) Co-authored-by: Lee Trout Co-authored-by: MrCloudSec Co-authored-by: Sergio Garcia Co-authored-by: HugoPBrito --- prowler/CHANGELOG.md | 1 + .../dms_instance_no_public_access.py | 2 +- ...allow_ingress_from_internet_to_any_port.py | 2 +- .../aws/services/ec2/lib/security_groups.py | 93 +++++++++---------- .../services/ec2/lib/security_groups_test.py | 8 +- 5 files changed, 52 insertions(+), 54 deletions(-) diff --git a/prowler/CHANGELOG.md b/prowler/CHANGELOG.md index 7adffee660..01ef06e722 100644 --- a/prowler/CHANGELOG.md +++ b/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) --- diff --git a/prowler/providers/aws/services/dms/dms_instance_no_public_access/dms_instance_no_public_access.py b/prowler/providers/aws/services/dms/dms_instance_no_public_access/dms_instance_no_public_access.py index 833d5d1fb5..2b491529c9 100644 --- a/prowler/providers/aws/services/dms/dms_instance_no_public_access/dms_instance_no_public_access.py +++ b/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." diff --git a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py b/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py index c4d3978199..f4ed7c20c4 100644 --- a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_any_port/ec2_securitygroup_allow_ingress_from_internet_to_any_port.py +++ b/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, diff --git a/prowler/providers/aws/services/ec2/lib/security_groups.py b/prowler/providers/aws/services/ec2/lib/security_groups.py index a5f240adb5..dabbe4f86d 100644 --- a/prowler/providers/aws/services/ec2/lib/security_groups.py +++ b/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 diff --git a/tests/providers/aws/services/ec2/lib/security_groups_test.py b/tests/providers/aws/services/ec2/lib/security_groups_test.py index 9296a29c20..653bf420bd 100644 --- a/tests/providers/aws/services/ec2/lib/security_groups_test.py +++ b/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