From 5de13bdd8a3ddb6e83e4dc8d42183fd22d8a0731 Mon Sep 17 00:00:00 2001 From: Mario Rodriguez Lopez <101330800+MarioRgzLpz@users.noreply.github.com> Date: Fri, 8 Nov 2024 20:08:27 +0100 Subject: [PATCH] fix(ec2): unique finding per Security Group in high risk ports check (#5697) --- ...ss_from_internet_to_high_risk_tcp_ports.py | 56 ++++++++-------- ...om_internet_to_high_risk_tcp_ports_test.py | 64 ++++++------------- 2 files changed, 51 insertions(+), 69 deletions(-) diff --git a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.py b/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.py index 764cdf5c30..fc3cf66955 100644 --- a/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.py +++ b/prowler/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports.py @@ -17,34 +17,38 @@ class ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports(Check and vpc_client.vpcs[security_group.vpc_id].in_use and len(security_group.network_interfaces) > 0 ): - check_ports = ec2_client.audit_config.get( - "ec2_high_risk_ports", - [25, 110, 135, 143, 445, 3000, 4333, 5000, 5500, 8080, 8088], - ) - for port in check_ports: - report = Check_Report_AWS(self.metadata()) - report.region = security_group.region - report.resource_details = security_group.name - report.resource_id = security_group.id - report.resource_arn = security_group_arn - report.resource_tags = security_group.tags - report.status = "PASS" - report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have port {port} open to the Internet." - # only proceed if check "..._to_all_ports" did not run or did not FAIL to avoid to report open ports twice - if not ec2_client.is_failed_check( - ec2_securitygroup_allow_ingress_from_internet_to_all_ports.__name__, - security_group_arn, - ): - # Loop through every security group's ingress rule and check it - for ingress_rule in security_group.ingress_rules: + report = Check_Report_AWS(self.metadata()) + report.region = security_group.region + report.resource_details = security_group.name + report.resource_id = security_group.id + report.resource_arn = security_group_arn + report.resource_tags = security_group.tags + report.status = "PASS" + report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have any high-risk port open to the Internet." + # only proceed if check "..._to_all_ports" did not run or did not FAIL to avoid to report open ports twice + if not ec2_client.is_failed_check( + ec2_securitygroup_allow_ingress_from_internet_to_all_ports.__name__, + security_group_arn, + ): + check_ports = ec2_client.audit_config.get( + "ec2_high_risk_ports", + [25, 110, 135, 143, 445, 3000, 4333, 5000, 5500, 8080, 8088], + ) + # Loop through every security group's ingress rule and check it + open_ports = [] + for ingress_rule in security_group.ingress_rules: + for port in check_ports: if check_security_group( ingress_rule, "tcp", [port], any_address=True ): - report.status = "FAIL" - report.status_extended = f"Security group {security_group.name} ({security_group.id}) has port {port} (high risk port) open to the Internet." - break - else: - report.status_extended = f"Security group {security_group.name} ({security_group.id}) has all ports open to the Internet and therefore was not checked against port {port}." - findings.append(report) + open_ports.append(port) + + if open_ports: + report.status = "FAIL" + open_ports_str = ", ".join(map(str, open_ports)) + report.status_extended = f"Security group {security_group.name} ({security_group.id}) has the following high-risk ports open to the Internet: {open_ports_str}." + else: + report.status_extended = f"Security group {security_group.name} ({security_group.id}) has all ports open to the Internet and therefore was not checked against high-risk ports." + findings.append(report) return findings diff --git a/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports_test.py b/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports_test.py index dbaaff49fa..2811c00f0b 100644 --- a/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports_test.py +++ b/tests/providers/aws/services/ec2/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports/ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports_test.py @@ -59,11 +59,9 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports: ) result = check.execute() - # One default sg per region, each port should be checked - expected_findings_count = ( - len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3 - ) # 3 security groups - assert len(result) == expected_findings_count + assert ( + len(result) == 3 + ) # 3 security groups one default sg per region and one added # Search changed sg for sg in result: # All are compliant by default @@ -132,21 +130,17 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports: ) result = check.execute() - # One default sg per region, each port should be checked - expected_findings_count = ( - len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3 - ) # 3 security groups - assert len(result) == expected_findings_count + assert ( + len(result) == 3 + ) # 3 security groups one default sg per region and one added # Search changed sg - iterator = 0 for sg in result: - port = ec2_client.audit_config["ec2_high_risk_ports"][iterator] if sg.resource_id == default_sg_id: assert sg.status == "FAIL" assert sg.region == AWS_REGION_US_EAST_1 assert ( sg.status_extended - == f"Security group {default_sg_name} ({default_sg_id}) has port {port} (high risk port) open to the Internet." + == f"Security group {default_sg_name} ({default_sg_id}) has the following high-risk ports open to the Internet: 25, 110, 135, 143, 445, 3000, 4333, 5000, 5500, 8080, 8088." ) assert ( sg.resource_arn @@ -154,9 +148,6 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports: ) assert sg.resource_details == default_sg_name assert sg.resource_tags == [] - iterator = (iterator + 1) % len( - ec2_client.audit_config["ec2_high_risk_ports"] - ) @mock_aws def test_ec2_compliant_default_sg(self): @@ -222,21 +213,17 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports: ) result = check.execute() - # One default sg per region, each port should be checked - expected_findings_count = ( - len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3 - ) # 3 security groups - assert len(result) == expected_findings_count + assert ( + len(result) == 3 + ) # 3 security groups one default sg per region and one added # Search changed sg - iterator = 0 for sg in result: - port = ec2_client.audit_config["ec2_high_risk_ports"][iterator] if sg.resource_id == default_sg_id: assert sg.status == "PASS" assert sg.region == AWS_REGION_US_EAST_1 assert ( sg.status_extended - == f"Security group {default_sg_name} ({default_sg_id}) does not have port {port} open to the Internet." + == f"Security group {default_sg_name} ({default_sg_id}) does not have any high-risk port open to the Internet." ) assert ( sg.resource_arn @@ -244,9 +231,6 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports: ) assert sg.resource_details == default_sg_name assert sg.resource_tags == [] - iterator = (iterator + 1) % len( - ec2_client.audit_config["ec2_high_risk_ports"] - ) @mock_aws def test_ec2_default_sgs_ignoring(self): @@ -350,10 +334,7 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports: ) result = check.execute() - expected_findings_count = len( - ec2_client.audit_config["ec2_high_risk_ports"] - ) # 1 security group in use - assert len(result) == expected_findings_count + assert len(result) == 1 # 1 security group added for sg in result: assert sg.status == "PASS" assert sg.region == AWS_REGION_US_EAST_1 @@ -452,24 +433,21 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_high_risk_tcp_ports: ) result_specific_port = check_specific_port.execute() - # One default sg per region, each port should be checked - expected_findings_count = ( - len(ec2_client.audit_config["ec2_high_risk_ports"]) * 3 - ) # 2 default security groups + 1 added - assert len(result_specific_port) == expected_findings_count + assert ( + len(result_specific_port) == 3 + ) # 3 security groups one default sg per region and one added # Search changed sg - iterator = 0 for sg in result_specific_port: - port = ec2_client.audit_config["ec2_high_risk_ports"][iterator] if sg.resource_id == default_sg_id: assert sg.status == "PASS" assert sg.region == AWS_REGION_US_EAST_1 assert ( sg.status_extended - == f"Security group {sg.resource_details} ({sg.resource_id}) has all ports open to the Internet and therefore was not checked against port {port}." + == f"Security group {default_sg_name} ({default_sg_id}) has all ports open to the Internet and therefore was not checked against high-risk ports." + ) + assert ( + sg.resource_arn + == f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{aws_provider.identity.account}:security-group/{default_sg_id}" ) - assert sg.resource_tags == [] assert sg.resource_details == default_sg_name - iterator = (iterator + 1) % len( - ec2_client.audit_config["ec2_high_risk_ports"] - ) + assert sg.resource_tags == []