From 219304848d5e764a2860a63d1ac8047f8b8979a6 Mon Sep 17 00:00:00 2001 From: Harsh Raj Date: Thu, 29 Oct 2026 17:26:46 +0530 Subject: [PATCH] fix(ec2): treat CIDRs that together cover every address as open to the internet (#13191) Co-authored-by: pedrooot --- .../ec2-sg-split-cidrs-any-address.fixed.md | 1 + .../aws/services/ec2/lib/security_groups.py | 31 +++++++++- .../services/ec2/lib/security_groups_test.py | 61 +++++++++++++++++++ .../emr_cluster_publicly_accesible_test.py | 11 ++-- 4 files changed, 98 insertions(+), 6 deletions(-) create mode 100644 prowler/changelog.d/ec2-sg-split-cidrs-any-address.fixed.md diff --git a/prowler/changelog.d/ec2-sg-split-cidrs-any-address.fixed.md b/prowler/changelog.d/ec2-sg-split-cidrs-any-address.fixed.md new file mode 100644 index 0000000000..ee4cc6d9a1 --- /dev/null +++ b/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 diff --git a/prowler/providers/aws/services/ec2/lib/security_groups.py b/prowler/providers/aws/services/ec2/lib/security_groups.py index dabbe4f86d..b8f586dbd5 100644 --- a/prowler/providers/aws/services/ec2/lib/security_groups.py +++ b/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 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 653bf420bd..03acfe8523 100644 --- a/tests/providers/aws/services/ec2/lib/security_groups_test.py +++ b/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 diff --git a/tests/providers/aws/services/emr/emr_cluster_publicly_accesible/emr_cluster_publicly_accesible_test.py b/tests/providers/aws/services/emr/emr_cluster_publicly_accesible/emr_cluster_publicly_accesible_test.py index ab5cd01fc6..c3031a886a 100644 --- a/tests/providers/aws/services/emr/emr_cluster_publicly_accesible/emr_cluster_publicly_accesible_test.py +++ b/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