mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 04:21:52 +00:00
fix(ec2): unique finding per Security Group in high risk ports check (#5697)
This commit is contained in:
Mario Rodriguez Lopez
committed by
GitHub
GitHub
parent
5d0f498425
commit
5de13bdd8a
+30
-26
@@ -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
|
||||
|
||||
+21
-43
@@ -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 == []
|
||||
|
||||
Reference in New Issue
Block a user