chore(ec2): Change security groups to dict (#4700)

This commit is contained in:
Rubén De la Torre Vico
2024-08-09 20:40:34 +02:00
committed by GitHub
parent ca56ac4e77
commit 18885d0cd7
44 changed files with 127 additions and 132 deletions
+1 -1
View File
@@ -222,7 +222,7 @@ class ec2_securitygroup_with_many_ingress_egress_rules(Check):
max_security_group_rules = ec2_client.audit_config.get(
"max_security_group_rules", 50
)
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
```
```yaml title="config.yaml"
@@ -22,7 +22,7 @@ class dms_instance_no_public_access(Check):
if instance.security_groups:
report.status = "PASS"
report.status_extended = f"DMS Replication Instance {instance.id} is set as publicly accessible but filtered with security groups."
for security_group in ec2_client.security_groups:
for security_group in ec2_client.security_groups.values():
if security_group.id in instance.security_groups:
for ingress_rule in security_group.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_cassandra_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -22,7 +22,7 @@ class ec2_instance_port_cifs_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_elasticsearch_kibana_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -22,7 +22,7 @@ class ec2_instance_port_ftp_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_kafka_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_kerberos_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -22,7 +22,7 @@ class ec2_instance_port_ldap_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_memcached_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_mongodb_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_mysql_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_oracle_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_postgresql_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_rdp_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_redis_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_sqlserver_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_ssh_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -20,7 +20,7 @@ class ec2_instance_port_telnet_exposed_to_internet(Check):
report.resource_tags = instance.tags
is_open_port = False
if instance.security_groups:
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id in instance.security_groups:
for ingress_rule in sg.ingress_rules:
if check_security_group(
@@ -1,13 +1,13 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_all_ports(Check):
def execute(self):
findings = []
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -20,13 +20,13 @@ class ec2_securitygroup_allow_ingress_from_internet_to_all_ports(Check):
report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have all ports open to the Internet."
report.resource_details = security_group.name
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_arn = security_group_arn
report.resource_tags = security_group.tags
for ingress_rule in security_group.ingress_rules:
if check_security_group(ingress_rule, "-1", any_address=True):
ec2_client.set_failed_check(
self.__class__.__name__,
security_group.arn,
security_group_arn,
)
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has all ports open to the Internet."
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.ec2_service import NetworkInterface
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.ec2_service import NetworkInterface
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check):
def execute(self):
findings = []
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -24,12 +24,12 @@ 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."
report.resource_details = security_group.name
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_arn = security_group_arn
report.resource_tags = security_group.tags
# 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -61,7 +61,6 @@ class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check):
):
report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but is exclusively not attached to any network interface."
for eni in enis:
if self.is_allowed_eni_type(eni_type=eni.type):
report.status = "PASS"
report.status_extended = f"Security group {security_group_name} ({security_group_id}) has at least one port open to the Internet but is exclusively attached to an allowed network interface type ({eni.type})."
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018(Check):
def execute(self):
findings = []
check_ports = [27017, 27018]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018(
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_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 MongoDB ports 27017 and 27018 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21(Check):
def execute(self):
findings = []
check_ports = [20, 21]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -24,12 +24,12 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21(Check)
report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have FTP ports 20 and 21 open to the Internet."
report.resource_details = security_group.name
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_arn = security_group_arn
report.resource_tags = security_group.tags
# 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22(Check):
def execute(self):
findings = []
check_ports = [22]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -24,12 +24,12 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22(Check):
report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have SSH port 22 open to the Internet."
report.resource_details = security_group.name
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_arn = security_group_arn
report.resource_tags = security_group.tags
# 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389(Check):
def execute(self):
findings = []
check_ports = [3389]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -24,12 +24,12 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389(Check):
report.status_extended = f"Security group {security_group.name} ({security_group.id}) does not have Microsoft RDP port 3389 open to the Internet."
report.resource_details = security_group.name
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_arn = security_group_arn
report.resource_tags = security_group.tags
# 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,10 +1,10 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888(
@@ -13,7 +13,7 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9
def execute(self):
findings = []
check_ports = [7199, 9160, 8888]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -24,14 +24,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9
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_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 Casandra ports 7199, 8888 and 9160 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,10 +1,10 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601(
@@ -13,7 +13,7 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_ki
def execute(self):
findings = []
check_ports = [9200, 9300, 5601]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -24,14 +24,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_ki
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_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 Elasticsearch/Kibana ports 9200, 9300 and 5601 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092(Check):
def execute(self):
findings = []
check_ports = [9092]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092(Check
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_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 Kafka port 9092 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211(Check):
def execute(self):
findings = []
check_ports = [11211]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211(
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_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 Memcached port 11211 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306(Check):
def execute(self):
findings = []
check_ports = [3306]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306(Check
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_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 MySQL port 3306 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483(Check):
def execute(self):
findings = []
check_ports = [1521, 2483]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483
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_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 Oracle ports 1521 and 2483 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432(Check):
def execute(self):
findings = []
check_ports = [5432]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432(Ch
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_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 Postgres port 5432 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379(Check):
def execute(self):
findings = []
check_ports = [6379]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379(Check
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_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 Redis port 6379 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,10 +1,10 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434(
@@ -13,7 +13,7 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_
def execute(self):
findings = []
check_ports = [1433, 1434]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -24,14 +24,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_
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_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 Microsoft SQL Server ports 1433 and 1434 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -1,17 +1,17 @@
from prowler.lib.check.models import Check, Check_Report_AWS
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23(Check):
def execute(self):
findings = []
check_ports = [23]
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -22,14 +22,14 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23(Check)
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_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 Telnet port 23 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,
security_group_arn,
):
# Loop through every security group's ingress rule and check it
for ingress_rule in security_group.ingress_rules:
@@ -9,7 +9,7 @@ class ec2_securitygroup_allow_wide_open_public_ipv4(Check):
def execute(self):
findings = []
cidr_treshold = 24
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the SG is in use
if ec2_client.provider.scan_unused_services or (
security_group.vpc_id in vpc_client.vpcs
@@ -20,7 +20,7 @@ class ec2_securitygroup_allow_wide_open_public_ipv4(Check):
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_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}) has no potential wide-open non-RFC1918 address."
@@ -6,7 +6,7 @@ from prowler.providers.aws.services.vpc.vpc_client import vpc_client
class ec2_securitygroup_default_restrict_traffic(Check):
def execute(self):
findings = []
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Check if ignoring flag is set and if the VPC and the default SG are in used
if security_group.name == "default" and (
ec2_client.provider.scan_unused_services
@@ -20,7 +20,7 @@ class ec2_securitygroup_default_restrict_traffic(Check):
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_arn = security_group_arn
report.resource_tags = security_group.tags
report.status = "FAIL"
report.status_extended = (
@@ -5,12 +5,12 @@ from prowler.providers.aws.services.ec2.ec2_client import ec2_client
class ec2_securitygroup_from_launch_wizard(Check):
def execute(self):
findings = []
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
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_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}) was not created using the EC2 Launch Wizard."
@@ -6,14 +6,14 @@ from prowler.providers.aws.services.ec2.ec2_client import ec2_client
class ec2_securitygroup_not_used(Check):
def execute(self):
findings = []
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
# Default security groups can not be deleted, so ignore them
if security_group.name != "default":
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_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}) it is being used."
@@ -22,7 +22,7 @@ class ec2_securitygroup_not_used(Check):
for function in awslambda_client.functions.values():
if security_group.id in function.security_groups:
sg_in_lambda = True
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if security_group.id in sg.associated_sgs:
sg_associated = True
if (
@@ -10,12 +10,12 @@ class ec2_securitygroup_with_many_ingress_egress_rules(Check):
max_security_group_rules = ec2_client.audit_config.get(
"max_security_group_rules", 50
)
for security_group in ec2_client.security_groups:
for security_group_arn, security_group in ec2_client.security_groups.items():
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_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}) has {len(security_group.ingress_rules)} inbound rules and {len(security_group.egress_rules)} outbound rules."
@@ -18,7 +18,7 @@ class EC2(AWSService):
self.instances = []
self.__threading_call__(self.__describe_instances__)
self.__threading_call__(self.__get_instance_user_data__, self.instances)
self.security_groups = []
self.security_groups = {}
self.regions_with_sgs = []
self.__threading_call__(self.__describe_security_groups__)
self.network_acls = []
@@ -120,18 +120,15 @@ class EC2(AWSService):
for sg_group in ingress_rule.get("UserIdGroupPairs", []):
if sg_group.get("GroupId"):
associated_sgs.append(sg_group["GroupId"])
self.security_groups.append(
SecurityGroup(
name=sg["GroupName"],
arn=arn,
region=regional_client.region,
id=sg["GroupId"],
ingress_rules=sg["IpPermissions"],
egress_rules=sg["IpPermissionsEgress"],
associated_sgs=associated_sgs,
vpc_id=sg["VpcId"],
tags=sg.get("Tags"),
)
self.security_groups[arn] = SecurityGroup(
name=sg["GroupName"],
region=regional_client.region,
id=sg["GroupId"],
ingress_rules=sg["IpPermissions"],
egress_rules=sg["IpPermissionsEgress"],
associated_sgs=associated_sgs,
vpc_id=sg["VpcId"],
tags=sg.get("Tags"),
)
if sg["GroupName"] != "default":
self.regions_with_sgs.append(regional_client.region)
@@ -267,7 +264,7 @@ class EC2(AWSService):
):
try:
for sg in interface_security_groups:
for security_group in self.security_groups:
for security_group in self.security_groups.values():
if security_group.id == sg["GroupId"]:
security_group.network_interfaces.append(interface)
except Exception as error:
@@ -563,7 +560,6 @@ class NetworkInterface(BaseModel):
class SecurityGroup(BaseModel):
name: str
arn: str
region: str
id: str
vpc_id: str
@@ -40,7 +40,7 @@ class emr_cluster_publicly_accesible(Check):
master_public_security_groups = []
for master_sg in master_node_sg_groups:
master_sg_public = False
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id == master_sg:
for ingress_rule in sg.ingress_rules:
if check_security_group(ingress_rule, -1):
@@ -62,7 +62,7 @@ class emr_cluster_publicly_accesible(Check):
slave_public_security_groups = []
for slave_sg in slave_node_sg_groups:
slave_sg_public = False
for sg in ec2_client.security_groups:
for sg in ec2_client.security_groups.values():
if sg.id == slave_sg:
for ingress_rule in sg.ingress_rules:
if check_security_group(ingress_rule, -1):
@@ -25,7 +25,7 @@ class rds_instance_no_public_access(Check):
report.status_extended = f"RDS Instance {db_instance.id} is set as publicly accessible but filtered with security groups."
db_instance_port = db_instance.endpoint.get("Port")
if db_instance_port:
for security_group in ec2_client.security_groups:
for security_group in ec2_client.security_groups.values():
if security_group.id in db_instance.security_groups:
for ingress_rule in security_group.ingress_rules:
if check_security_group(
@@ -150,11 +150,11 @@ class Test_EC2_Service:
ec2 = EC2(aws_provider)
assert sg_id in str(ec2.security_groups)
for security_group in ec2.security_groups:
for sg_arn, security_group in ec2.security_groups.items():
if security_group.id == sg_id:
assert security_group.name == "test-security-group"
assert (
security_group.arn
sg_arn
== f"arn:{aws_provider.identity.partition}:ec2:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:security-group/{security_group.id}"
)
assert re.match(r"sg-[0-9a-z]{17}", security_group.id)
@@ -524,7 +524,7 @@ class Test_EC2_Service:
{"Key": "string", "Value": "string"},
]
# Check if ENI was added to security group
for sg in ec2.security_groups:
for sg in ec2.security_groups.values():
if sg.id == eni.groups[0]["GroupId"]:
assert sg.network_interfaces == ec2.network_interfaces