chore(backport): put latest changes of v4 to v3 (#4144)

This commit is contained in:
Sergio Garcia
2024-05-30 12:16:36 -04:00
committed by GitHub
parent fb2cd0dbb0
commit bdc42a5251
61 changed files with 2281 additions and 142 deletions
@@ -95,7 +95,8 @@ checks_v3_to_v2_mapping = {
"ec2_networkacl_allow_ingress_any_port": "extra7138",
"ec2_networkacl_allow_ingress_tcp_port_22": "check45",
"ec2_networkacl_allow_ingress_tcp_port_3389": "check46",
"ec2_securitygroup_allow_ingress_from_internet_to_any_port": "extra748",
"ec2_securitygroup_allow_ingress_from_internet_to_all_ports": "extra748",
"ec2_securitygroup_allow_ingress_from_internet_to_any_port": "extra74",
"ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018": "extra753",
"ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21": "extra7134",
"ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22": "check41",
+14 -1
View File
@@ -30,9 +30,10 @@ If EBS default encyption is not enabled, sensitive information at rest is not pr
- `ec2_ebs_default_encryption`
If your Security groups are not properly configured the attack surface is increased, nonetheless, Prowler will detect those security groups that are being used (they are attached) to only notify those that are being used. This logic applies to the 15 checks related to open ports in security groups.
If your Security groups are not properly configured the attack surface is increased, nonetheless, Prowler will detect those security groups that are being used (they are attached) to only notify those that are being used. This logic applies to the 15 checks related to open ports in security groups and the check for the default security group.
- `ec2_securitygroup_allow_ingress_from_internet_to_port_X` (15 checks)
- `ec2_securitygroup_default_restrict_traffic`
Prowler will also check for used Network ACLs to only alerts those with open ports that are being used.
@@ -69,3 +70,15 @@ You should enable Public Access Block at the account level to prevent the exposu
VPC Flow Logs provide visibility into network traffic that traverses the VPC and can be used to detect anomalous traffic or insight during security workflows. Nevertheless, Prowler will only check if the Flow Logs are enabled for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).
- `vpc_flow_logs_enabled`
VPC subnets must not have public IP addresses by default to prevent the exposure of your resources to the internet. Prowler will only check this configuration for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).
- `vpc_subnet_no_public_ip_by_default`
VPCs should have separate private and public subnets to prevent the exposure of your resources to the internet. Prowler will only check this configuration for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).
- `vpc_subnet_separate_private_public`
VPCs should have subnets in different availability zones to prevent a single point of failure. Prowler will only check this configuration for those VPCs that are in use, in other words, only the VPCs where you have ENIs (network interfaces).
- `vpc_subnet_different_az`
+2 -2
View File
@@ -11,7 +11,7 @@ prowler <provider> --slack
![Prowler Slack Message](img/slack-prowler-message.png)
???+ note
Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_ID environment variables.
Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_NAME environment variables.
### Configuration
@@ -35,4 +35,4 @@ To configure the Slack Integration, follow the next steps:
4. Set the following environment variables that Prowler will read:
- `SLACK_API_TOKEN`: the *Slack App OAuth Token* that was previously get.
- `SLACK_CHANNEL_ID`: the name of your Slack Channel where Prowler will send the message.
- `SLACK_CHANNEL_NAME`: the name of your Slack Channel where Prowler will send the message.
+9 -3
View File
@@ -204,17 +204,23 @@ def prowler():
stats = extract_findings_statistics(findings)
if args.slack:
if "SLACK_API_TOKEN" in os.environ and "SLACK_CHANNEL_ID" in os.environ:
if "SLACK_API_TOKEN" in os.environ and (
"SLACK_CHANNEL_NAME" in os.environ or "SLACK_CHANNEL_ID" in os.environ
):
_ = send_slack_message(
os.environ["SLACK_API_TOKEN"],
os.environ["SLACK_CHANNEL_ID"],
(
os.environ["SLACK_CHANNEL_NAME"]
if "SLACK_CHANNEL_NAME" in os.environ
else os.environ["SLACK_CHANNEL_ID"]
),
stats,
provider,
audit_info,
)
else:
logger.critical(
"Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_ID environment variables (see more in https://docs.prowler.cloud/en/latest/tutorials/integrations/#slack)."
"Slack integration needs SLACK_API_TOKEN and SLACK_CHANNEL_NAME environment variables (see more in https://docs.prowler.cloud/en/latest/tutorials/integrations/#slack)."
)
sys.exit(1)
+2 -1
View File
@@ -24,7 +24,8 @@ def get_check_compliance(finding, provider, output_options):
compliance_fw = compliance.Framework
if compliance.Version:
compliance_fw = f"{compliance_fw}-{compliance.Version}"
if compliance.Provider == provider.upper():
# compliance.Provider == "Azure" or "GCP" or "AWS"
if compliance.Provider.upper() == provider.upper():
if compliance_fw not in check_compliance:
check_compliance[compliance_fw] = []
for requirement in compliance.Requirements:
@@ -1280,6 +1280,7 @@
"braket": {
"regions": {
"aws": [
"eu-north-1",
"eu-west-2",
"us-east-1",
"us-west-1",
@@ -8021,14 +8022,18 @@
"quicksight": {
"regions": {
"aws": [
"af-south-1",
"ap-northeast-1",
"ap-northeast-2",
"ap-south-1",
"ap-southeast-1",
"ap-southeast-2",
"ap-southeast-3",
"ca-central-1",
"eu-central-1",
"eu-central-2",
"eu-north-1",
"eu-south-1",
"eu-west-1",
"eu-west-2",
"eu-west-3",
@@ -8296,14 +8301,18 @@
"aws": [
"ap-south-1",
"ca-central-1",
"eu-central-2",
"eu-south-2",
"eu-west-3",
"sa-east-1",
"us-east-1",
"us-east-2",
"us-west-1",
"us-west-2"
],
"aws-cn": [
"cn-north-1"
"cn-north-1",
"cn-northwest-1"
],
"aws-us-gov": []
}
@@ -19,6 +19,17 @@ class AWSService:
- Also handles if the AWS Service is Global
"""
failed_checks = set()
@classmethod
def set_failed_check(cls, check_id=None, arn=None):
if check_id is not None and arn is not None:
cls.failed_checks.add((check_id.split(".")[-1], arn))
@classmethod
def is_failed_check(cls, check_id, arn):
return (check_id.split(".")[-1], arn) in cls.failed_checks
def __init__(self, service: str, audit_info: AWS_Audit_Info, global_service=False):
# Audit Information
self.audit_info = audit_info
@@ -10,7 +10,7 @@
"ResourceType": "AwsCloudFormationStack",
"Description": "Find secrets in CloudFormation outputs",
"Risk": "Secrets hardcoded into CloudFormation outputs can be used by malware and bad actors to gain lateral access to other services.",
"RelatedUrl": "https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-secretsmanager-secret-generatesecretstring.html",
"RelatedUrl": "https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/outputs-section-structure.html",
"Remediation": {
"Code": {
"CLI": "https://docs.prowler.com/checks/aws/secrets-policies/bc_aws_secrets_2#cli-command",
@@ -1,5 +1,6 @@
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
@@ -21,9 +22,16 @@ class ec2_securitygroup_allow_ingress_from_internet_to_all_ports(Check):
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_tags = security_group.tags
if security_group.public_ports:
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has all ports open to the Internet."
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,
)
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has all ports open to the Internet."
break
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -22,7 +25,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check):
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_tags = security_group.tags
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -30,6 +37,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_any_port(Check):
):
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has at least one port open to the Internet."
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 a specific port."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018(
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018(
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has MongoDB ports 27017 and 27018 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 the specific MongoDB ports 27017 and 27018."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21(Check)
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_tags = security_group.tags
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21(Check)
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has FTP ports 20 and 21 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 the specific FTP ports 20 and 21."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22(Check):
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_tags = security_group.tags
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22(Check):
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has SSH port 22 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 the specific SSH port 22."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389(Check):
report.resource_id = security_group.id
report.resource_arn = security_group.arn
report.resource_tags = security_group.tags
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389(Check):
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Microsoft RDP port 3389 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 the specific Microsoft RDP port 3389."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -25,7 +28,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -34,6 +41,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Casandra ports 7199, 8888 and 9160 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 the specific Cassandra ports 7199, 8888 and 9160."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -25,7 +28,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_ki
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -34,6 +41,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_ki
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Elasticsearch/Kibana ports 9200, 9300 and 5601 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 the specific Elasticsearch/Kibana ports 9200, 9300 and 5601."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092(Check
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092(Check
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Kafka port 9092 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 the specific Kafka port 9092."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211(
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211(
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Memcached port 11211 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 the specific Memcached port 11211."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306(Check
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -34,6 +41,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306(Check
report.resource_details = security_group.name
report.resource_id = security_group.id
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 the specific MySQL port 3306."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Oracle ports 1521 and 2483 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 the specific Oracle ports 1521 and 2483."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432(Ch
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432(Ch
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Postgres port 5432 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 the specific Postgres port 5432."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379(Check
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379(Check
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Redis port 6379 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 the specific Redis port 6379."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -25,7 +28,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -34,6 +41,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Microsoft SQL Server ports 1433 and 1434 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 the specific Microsoft SQL Server ports 1433 and 1434."
findings.append(report)
return findings
@@ -1,5 +1,8 @@
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_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
@@ -23,7 +26,11 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23(Check)
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."
if not security_group.public_ports:
# 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:
if check_security_group(
@@ -32,6 +39,9 @@ class ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23(Check)
report.status = "FAIL"
report.status_extended = f"Security group {security_group.name} ({security_group.id}) has Telnet port 23 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 the specific Telnet port 23."
findings.append(report)
return findings
@@ -1,19 +1,27 @@
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
class ec2_securitygroup_default_restrict_traffic(Check):
def execute(self):
findings = []
for security_group in ec2_client.security_groups:
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
# Find default security group
if security_group.name == "default":
# Check if ignoring flag is set and if the VPC and the default SG are in used
if security_group.name == "default" and (
not ec2_client.audit_info.ignore_unused_services
or (
security_group.vpc_id in vpc_client.vpcs
and vpc_client.vpcs[security_group.vpc_id].in_use
and len(security_group.network_interfaces) > 0
)
):
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 = "FAIL"
report.status_extended = (
f"Default Security Group ({security_group.id}) rules allow traffic."
@@ -7,7 +7,6 @@ from pydantic import BaseModel
from prowler.lib.logger import logger
from prowler.lib.scan_filters.scan_filters import is_resource_filtered
from prowler.providers.aws.lib.service.service import AWSService
from prowler.providers.aws.services.ec2.lib.security_groups import check_security_group
################## EC2
@@ -122,16 +121,7 @@ class EC2(AWSService):
):
associated_sgs = []
# check if sg has public access to all ports
all_public_ports = False
for ingress_rule in sg["IpPermissions"]:
if (
check_security_group(
ingress_rule, "-1", any_address=True
)
and "ec2_securitygroup_allow_ingress_from_internet_to_all_ports"
in self.audited_checks
):
all_public_ports = True
# check associated security groups
for sg_group in ingress_rule.get("UserIdGroupPairs", []):
if sg_group.get("GroupId"):
@@ -144,7 +134,6 @@ class EC2(AWSService):
id=sg["GroupId"],
ingress_rules=sg["IpPermissions"],
egress_rules=sg["IpPermissionsEgress"],
public_ports=all_public_ports,
associated_sgs=associated_sgs,
vpc_id=sg["VpcId"],
tags=sg.get("Tags"),
@@ -465,7 +454,6 @@ class SecurityGroup(BaseModel):
region: str
id: str
vpc_id: str
public_ports: bool
associated_sgs: list
network_interfaces: list[NetworkInterface] = []
ingress_rules: list[dict]
@@ -117,9 +117,11 @@ class OpenSearchService(AWSService):
domain.internal_user_database = describe_domain["DomainStatus"][
"AdvancedSecurityOptions"
]["InternalUserDatabaseEnabled"]
domain.saml_enabled = describe_domain["DomainStatus"][
"AdvancedSecurityOptions"
]["SAMLOptions"]["Enabled"]
domain.saml_enabled = (
describe_domain["DomainStatus"]["AdvancedSecurityOptions"]
.get("SAMLOptions", {})
.get("Enabled", False)
)
domain.update_available = describe_domain["DomainStatus"][
"ServiceSoftwareOptions"
]["UpdateAvailable"]
@@ -42,7 +42,7 @@ class rds_instance_transport_encrypted(Check):
for parameter in db_instance.parameters:
if (
parameter["ParameterName"] == "rds.force_ssl"
and parameter["ParameterValue"] == "1"
and parameter.get("ParameterValue", "0") == "1"
):
report.status = "PASS"
report.status_extended = f"RDS Instance {db_instance.id} connections use SSL encryption."
@@ -50,10 +50,29 @@ class rds_instance_transport_encrypted(Check):
for parameter in db_instance.parameters:
if (
parameter["ParameterName"] == "require_secure_transport"
and parameter["ParameterValue"] == "1"
and parameter.get("ParameterValue", "0") == "1"
):
report.status = "PASS"
report.status_extended = f"RDS Instance {db_instance.id} connections use SSL encryption."
findings.append(report)
for db_cluster in rds_client.db_clusters:
report = Check_Report_AWS(self.metadata())
report.region = rds_client.db_clusters[db_cluster].region
report.resource_id = rds_client.db_clusters[db_cluster].id
report.resource_arn = db_cluster
report.resource_tags = rds_client.db_clusters[db_cluster].tags
report.status = "FAIL"
report.status_extended = f"RDS Cluster {rds_client.db_clusters[db_cluster].id} connections are not encrypted."
# Check RDS Clusters that support TLS encryption
if rds_client.db_clusters[db_cluster].force_ssl == "1":
report.status = "PASS"
report.status_extended = f"RDS Cluster {rds_client.db_clusters[db_cluster].id} connections use SSL encryption."
if rds_client.db_clusters[db_cluster].require_secure_transport == "ON":
report.status = "PASS"
report.status_extended = f"RDS Cluster {rds_client.db_clusters[db_cluster].id} connections use SSL encryption."
findings.append(report)
return findings
@@ -17,6 +17,7 @@ class RDS(AWSService):
self.db_clusters = {}
self.db_snapshots = []
self.db_engines = {}
self.db_cluster_parameters = {}
self.db_cluster_snapshots = []
self.__threading_call__(self.__describe_db_instances__)
self.__threading_call__(self.__describe_db_parameters__)
@@ -169,6 +170,9 @@ class RDS(AWSService):
is_resource_filtered(db_cluster_arn, self.audit_resources)
):
if cluster["Engine"] != "docdb":
describe_db_parameters_paginator = (
regional_client.get_paginator("describe_db_parameters")
)
db_cluster = DBCluster(
id=cluster["DBClusterIdentifier"],
arn=db_cluster_arn,
@@ -192,8 +196,25 @@ class RDS(AWSService):
region=regional_client.region,
tags=cluster.get("TagList", []),
)
for page in describe_db_parameters_paginator.paginate(
DBParameterGroupName=cluster["DBClusterParameterGroup"]
):
for parameter in page["Parameters"]:
if parameter["ParameterName"] == "rds.force_ssl":
db_cluster.force_ssl = parameter[
"ParameterValue"
]
if (
parameter["ParameterName"]
== "require_secure_transport"
):
db_cluster.require_secure_transport = parameter[
"ParameterValue"
]
# We must use a unique value as the dict key to have unique keys
self.db_clusters[db_cluster_arn] = db_cluster
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -324,7 +345,9 @@ class DBCluster(BaseModel):
deletion_protection: bool
auto_minor_version_upgrade: bool
multi_az: bool
parameter_group: Optional[str]
parameter_group: str
force_ssl: Optional[bool]
require_secure_transport: Optional[str]
region: str
tags: Optional[list] = []
@@ -28,9 +28,9 @@ class VPC(AWSService):
self.__describe_flow_logs__()
self.__describe_peering_route_tables__()
self.__describe_vpc_endpoint_service_permissions__()
self.__describe_network_interfaces__()
self.vpc_subnets = {}
self.__threading_call__(self.__describe_vpc_subnets__)
self.__describe_network_interfaces__()
def __describe_vpcs__(self, regional_client):
logger.info("VPC - Describing VPCs...")
@@ -192,6 +192,19 @@ class VPC(AWSService):
)["NetworkInterfaces"]
if enis:
vpc.in_use = True
for subnet in vpc.subnets:
enis = regional_client.describe_network_interfaces(
Filters=[
{
"Name": "subnet-id",
"Values": [
subnet.id,
],
},
]
)["NetworkInterfaces"]
if enis:
subnet.in_use = True
except Exception as error:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -395,6 +408,7 @@ class VpcSubnet(BaseModel):
cidr_block: Optional[str]
availability_zone: str
public: bool
in_use: bool = False
nat_gateway: bool
region: str
mapPublicIpOnLaunch: bool
@@ -6,28 +6,29 @@ class vpc_subnet_different_az(Check):
def execute(self):
findings = []
for vpc in vpc_client.vpcs.values():
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
)
report.resource_id = vpc.id
report.resource_arn = vpc.arn
if vpc.subnets:
availability_zone = None
for subnet in vpc.subnets:
if (
availability_zone
and subnet.availability_zone != availability_zone
):
report.status = "PASS"
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has subnets in more than one availability zone."
break
availability_zone = subnet.availability_zone
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only subnets in {availability_zone}."
if not vpc_client.audit_info.ignore_unused_services or vpc.in_use:
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
)
report.resource_id = vpc.id
report.resource_arn = vpc.arn
if vpc.subnets:
availability_zone = None
for subnet in vpc.subnets:
if (
availability_zone
and subnet.availability_zone != availability_zone
):
report.status = "PASS"
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has subnets in more than one availability zone."
break
availability_zone = subnet.availability_zone
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only subnets in {availability_zone}."
findings.append(report)
findings.append(report)
return findings
@@ -7,17 +7,19 @@ class vpc_subnet_no_public_ip_by_default(Check):
findings = []
for vpc in vpc_client.vpcs.values():
for subnet in vpc.subnets:
report = Check_Report_AWS(self.metadata())
report.region = subnet.region
report.resource_tags = subnet.tags
report.resource_id = subnet.id
report.resource_arn = subnet.arn
if subnet.mapPublicIpOnLaunch:
report.status = "FAIL"
report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} assigns public IP by default."
else:
report.status = "PASS"
report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} does NOT assign public IP by default."
findings.append(report)
# Check if ignoring flag is set and if the VPC Subnet is in use
if not vpc_client.audit_info.ignore_unused_services or subnet.in_use:
report = Check_Report_AWS(self.metadata())
report.region = subnet.region
report.resource_tags = subnet.tags
report.resource_id = subnet.id
report.resource_arn = subnet.arn
if subnet.mapPublicIpOnLaunch:
report.status = "FAIL"
report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} assigns public IP by default."
else:
report.status = "PASS"
report.status_extended = f"VPC subnet {subnet.name if subnet.name else subnet.id} does NOT assign public IP by default."
findings.append(report)
return findings
@@ -6,28 +6,29 @@ class vpc_subnet_separate_private_public(Check):
def execute(self):
findings = []
for vpc in vpc_client.vpcs.values():
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
)
report.resource_id = vpc.id
report.resource_arn = vpc.arn
if vpc.subnets:
public = False
private = False
for subnet in vpc.subnets:
if subnet.public:
public = True
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only public subnets."
if not subnet.public:
private = True
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only private subnets."
if public and private:
report.status = "PASS"
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has private and public subnets."
findings.append(report)
if not vpc_client.audit_info.ignore_unused_services or vpc.in_use:
report = Check_Report_AWS(self.metadata())
report.region = vpc.region
report.resource_tags = vpc.tags
report.status = "FAIL"
report.status_extended = (
f"VPC {vpc.name if vpc.name else vpc.id} has no subnets."
)
report.resource_id = vpc.id
report.resource_arn = vpc.arn
if vpc.subnets:
public = False
private = False
for subnet in vpc.subnets:
if subnet.public:
public = True
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only public subnets."
if not subnet.public:
private = True
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has only private subnets."
if public and private:
report.status = "PASS"
report.status_extended = f"VPC {vpc.name if vpc.name else vpc.id} has private and public subnets."
findings.append(report)
return findings
@@ -8,26 +8,26 @@ class iam_custom_role_has_permissions_to_administer_resource_locks(Check):
def execute(self) -> Check_Report_Azure:
findings = []
for subscription, roles in iam_client.custom_roles.items():
for role in roles:
exits_role_with_permission_over_locks = False
for custom_role in roles:
if exits_role_with_permission_over_locks:
break
report = Check_Report_Azure(self.metadata())
report.subscription = subscription
report.resource_id = role.id
report.resource_name = role.name
has_lock_permission = False
for permission_item in role.permissions:
if has_lock_permission:
report.resource_id = custom_role.id
report.resource_name = custom_role.name
report.status = "FAIL"
report.status_extended = f"Role {custom_role.name} from subscription {subscription} has no permission to administer resource locks."
for permission_item in custom_role.permissions:
if exits_role_with_permission_over_locks:
break
for action in permission_item.actions:
if has_lock_permission:
break
if search("^Microsoft.Authorization/locks/.*", action):
report.status = "PASS"
report.status_extended = f"Role {role.name} from subscription {subscription} has permission to administer resource locks."
has_lock_permission = True
report.status_extended = f"Role {custom_role.name} from subscription {subscription} has permission to administer resource locks."
exits_role_with_permission_over_locks = True
break
else:
report.status = "FAIL"
report.status_extended = f"Role {role.name} from subscription {subscription} has no permission to administer resource locks."
break
findings.append(report)
findings.append(report)
return findings
@@ -8,20 +8,20 @@ class iam_subscription_roles_owner_custom_not_created(Check):
def execute(self) -> Check_Report_Azure:
findings = []
for subscription, roles in iam_client.custom_roles.items():
for role in roles:
for custom_role in roles:
report = Check_Report_Azure(self.metadata())
report.subscription = subscription
report.resource_id = role.id
report.resource_name = role.name
report.resource_id = custom_role.id
report.resource_name = custom_role.name
report.status = "PASS"
report.status_extended = f"Role {role.name} from subscription {subscription} is not a custom owner role."
for scope in role.assignable_scopes:
report.status_extended = f"Role {custom_role.name} from subscription {subscription} is not a custom owner role."
for scope in custom_role.assignable_scopes:
if search("^/.*", scope):
for permission_item in role.permissions:
for permission_item in custom_role.permissions:
for action in permission_item.actions:
if action == "*":
report.status = "FAIL"
report.status_extended = f"Role {role.name} from subscription {subscription} is a custom owner role."
report.status_extended = f"Role {custom_role.name} from subscription {subscription} is a custom owner role."
break
findings.append(report)
+13 -5
View File
@@ -3,6 +3,7 @@ import sys
from google import auth
from googleapiclient import discovery
from googleapiclient.errors import HttpError
from prowler.lib.logger import logger
@@ -80,18 +81,25 @@ class GCP_Provider:
previous_request=request, previous_response=response
)
return project_ids
except HttpError as http_error:
if "Cloud Resource Manager API has not been used" in str(http_error):
logger.critical(
"Cloud Resource Manager API has not been used before or it is disabled. Enable it by visiting https://console.developers.google.com/apis/api/cloudresourcemanager.googleapis.com/ then retry."
)
sys.exit(1)
else:
logger.error(
f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {http_error}"
)
except Exception as error:
if error.__class__.__name__ == "RefreshError":
logger.critical(
"Google Cloud SDK has not been authenticated or the credentials have expired. Authenticate by running 'gcloud auth application-default login' then retry."
)
elif "Cloud Resource Manager API has not been used" in str(error):
logger.critical(
"Cloud Resource Manager API has not been used before or it is disabled. Enable it by visiting https://console.developers.google.com/apis/api/cloudresourcemanager.googleapis.com/ then retry."
)
else:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
finally:
return project_ids
@@ -60,3 +60,33 @@ class Test_AWSService:
assert not hasattr(service, "regional_clients")
assert service.region == AWS_REGION_US_EAST_1
assert service.client.__class__.__name__ == "CloudFront"
def test_AWSService_set_failed_check(self):
AWSService.failed_checks.clear()
check_id = "ec2_securitygroup_allow_ingress_from_internet_to_all_ports"
arn = "arn:aws:ec2:eu-central-1:123456789:security-group/sg-12345678"
assert (check_id, arn) not in AWSService.failed_checks
AWSService.set_failed_check(check_id, arn)
assert (check_id, arn) in AWSService.failed_checks
def test_AWSService_is_failed_check(self):
AWSService.failed_checks.clear()
check_id = "ec2_securitygroup_allow_ingress_from_internet_to_all_ports"
arn = "arn:aws:ec2:eu-central-1:123456789:security-group/sg-12345678"
assert not AWSService.is_failed_check(check_id, arn)
AWSService.set_failed_check(check_id, arn)
assert AWSService.is_failed_check(check_id, arn)
assert not AWSService.is_failed_check(
check_id,
"arn:aws:ec2:eu-central-1:123456789:security-group/sg-87654321",
)
@@ -331,3 +331,76 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_all_ports:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_set_failed_check_called_correctly(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
}
],
)
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1],
expected_checks=[
"ec2_securitygroup_allow_ingress_from_internet_to_all_ports"
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
from prowler.providers.aws.services.vpc.vpc_service import VPC
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
), mock.patch(
"prowler.providers.aws.services.vpc.vpc_service.VPC",
new=VPC(current_audit_info),
), mock.patch(
"prowler.providers.aws.lib.service.service.AWSService.set_failed_check"
) as mock_set_failed_check:
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check = ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
result = check.execute()
# Verify set_failed_check was called with the correct arguments
mock_set_failed_check.assert_called_with(
"ec2_securitygroup_allow_ingress_from_internet_to_all_ports",
f"arn:{current_audit_info.audited_partition}:ec2:{AWS_REGION_US_EAST_1}:{current_audit_info.audited_account}:security-group/{default_sg_id}",
)
# Additional assertions to verify the check result
assert len(result) == 3
for sg in result:
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 all ports open to the Internet."
)
assert (
sg.resource_arn
== f"arn:{current_audit_info.audited_partition}:ec2:{AWS_REGION_US_EAST_1}:{current_audit_info.audited_account}:security-group/{default_sg_id}"
)
assert sg.resource_details == default_sg_name
assert sg.resource_tags == []
@@ -359,3 +359,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_any_port:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_any_port.ec2_securitygroup_allow_ingress_from_internet_to_any_port.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_any_port.ec2_securitygroup_allow_ingress_from_internet_to_any_port.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_any_port.ec2_securitygroup_allow_ingress_from_internet_to_any_port import (
ec2_securitygroup_allow_ingress_from_internet_to_any_port,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_any_port()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 a specific port."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_2
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018.ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018.ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018.ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018 import (
ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_port_mongodb_27017_27018()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific MongoDB ports 27017 and 27018."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21.ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21.ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21.ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_ftp_port_20_21()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific FTP ports 20 and 21."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -256,3 +256,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_22()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific SSH port 22."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -251,3 +251,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389:
)
assert sg.resource_details == default_sg_name
assert sg.resource_tags == []
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_3389()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Microsoft RDP port 3389."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_cassandra_7199_9160_8888()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Cassandra ports 7199, 8888 and 9160."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsear
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_elasticsearch_kibana_9200_9300_5601()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Elasticsearch/Kibana ports 9200, 9300 and 5601."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_kafka_9092()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Kafka port 9092."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_1
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_memcached_11211()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Memcached port 11211."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_mysql_3306()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific MySQL port 3306."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_oracle_1521_2483()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Oracle ports 1521 and 2483."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -332,3 +332,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_54
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_postgres_5432()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Postgres port 5432."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -319,3 +319,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379:
)
assert result[0].resource_details == sg["GroupName"]
assert result[0].resource_tags == []
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_redis_6379()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Redis port 6379."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_sql_server_1433_1434()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Microsoft SQL Server ports 1433 and 1434."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -261,3 +261,95 @@ class Test_ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23:
assert len(result) == 1
assert result[0].status == "PASS"
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_ec2_non_compliant_default_sg_pass_to_avoid_fail_twice(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="10.0.0.0/16")
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg_name = default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"FromPort": -1,
"ToPort": -1,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
{
"IpProtocol": "tcp",
"FromPort": 22,
"ToPort": 22,
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
},
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_client",
new=EC2(current_audit_info),
) as ec2_client_instance, mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.vpc_client",
new=VPC(current_audit_info),
) as vpc_client_instance:
# Run check for all ports
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_all_ports.ec2_securitygroup_allow_ingress_from_internet_to_all_ports import (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports,
)
check_all_ports = (
ec2_securitygroup_allow_ingress_from_internet_to_all_ports()
)
result_all_ports = check_all_ports.execute()
# Verify that the all ports check has detected the issue
assert any(
sg.status == "FAIL" and sg.resource_id == default_sg_id
for sg in result_all_ports
)
# use the same mock objects for the specific port check
with mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23.ec2_client",
new=ec2_client_instance,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23.vpc_client",
new=vpc_client_instance,
):
# Now run the specific port check
from prowler.providers.aws.services.ec2.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23.ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23 import (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23,
)
check_specific_port = (
ec2_securitygroup_allow_ingress_from_internet_to_tcp_port_telnet_23()
)
result_specific_port = check_specific_port.execute()
# One default sg per region
assert len(result_specific_port) == 3
# Search changed sg
for sg in result_specific_port:
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 the specific Telnet port 23."
)
assert sg.resource_tags == []
assert sg.resource_details == default_sg_name
@@ -130,6 +130,58 @@ class Test_ec2_securitygroup_default_restrict_traffic:
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_id == default_sg_id
@mock_aws
def test_ec2_non_compliant_sg_ingress_rule_but_unused(self):
# Create EC2 Mocked Resources
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
default_sg = ec2_client.describe_security_groups(GroupNames=["default"])[
"SecurityGroups"
][0]
default_sg_id = default_sg["GroupId"]
default_sg["GroupName"]
ec2_client.authorize_security_group_ingress(
GroupId=default_sg_id,
IpPermissions=[
{"IpProtocol": "-1", "IpRanges": [{"CidrIp": "10.0.0.16/0"}]}
],
)
ec2_client.revoke_security_group_egress(
GroupId=default_sg_id,
IpPermissions=[
{
"IpProtocol": "-1",
"IpRanges": [{"CidrIp": "0.0.0.0/0"}],
"Ipv6Ranges": [],
"PrefixListIds": [],
"UserIdGroupPairs": [],
}
],
)
from prowler.providers.aws.services.ec2.ec2_service import EC2
current_audit_info = set_mocked_aws_audit_info(
audited_regions=[AWS_REGION_US_EAST_1], ignore_unused_services=True
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
), mock.patch(
"prowler.providers.aws.services.ec2.ec2_securitygroup_default_restrict_traffic.ec2_securitygroup_default_restrict_traffic.ec2_client",
new=EC2(current_audit_info),
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_securitygroup_default_restrict_traffic.ec2_securitygroup_default_restrict_traffic import (
ec2_securitygroup_default_restrict_traffic,
)
check = ec2_securitygroup_default_restrict_traffic()
result = check.execute()
# One default sg per region
assert len(result) == 0
@mock_aws
def test_ec2_non_compliant_sg_egress_rule(self):
# Create EC2 Mocked Resources
@@ -173,7 +173,6 @@ class Test_EC2_Service:
"UserIdGroupPairs": [],
}
]
assert security_group.public_ports
assert security_group.tags == [
{"Key": "test", "Value": "test"},
]
@@ -148,12 +148,18 @@ class Test_rds_instance_deletion_protection:
@mock_aws
def test_rds_instance_without_cluster_deletion_protection(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mysql8.0",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="postgres",
DatabaseName="staging-postgres",
DeletionProtection=False,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[
@@ -206,12 +212,18 @@ class Test_rds_instance_deletion_protection:
@mock_aws
def test_rds_instance_with_cluster_deletion_protection(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mysql8.0",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="postgres",
DatabaseName="staging-postgres",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[
@@ -112,6 +112,17 @@ class Test_rds_instance_transport_encrypted:
DBParameterGroupFamily="default.aurora-postgresql14",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-postgresql",
DatabaseName="staging-postgres",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
@@ -119,7 +130,7 @@ class Test_rds_instance_transport_encrypted:
DBName="aurora-postgres",
DBInstanceClass="db.m1.small",
DBParameterGroupName="test",
DBClusterIdentifier="cluster-postgres",
DBClusterIdentifier="db-cluster-1",
)
from prowler.providers.aws.services.rds.rds_service import RDS
@@ -141,7 +152,19 @@ class Test_rds_instance_transport_encrypted:
check = rds_instance_transport_encrypted()
result = check.execute()
assert len(result) == 0
assert len(result) == 1
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 connections are not encrypted."
)
assert result[0].resource_id == "db-cluster-1"
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_postgres_rds_instance_no_ssl(self):
@@ -390,3 +413,129 @@ class Test_rds_instance_transport_encrypted:
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:db:db-master-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_postgres_clustered_instance_ssl(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.aurora-postgresql14",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-postgresql",
DatabaseName="staging-postgres",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
conn.modify_db_parameter_group(
DBParameterGroupName="test",
Parameters=[
{
"ParameterName": "rds.force_ssl",
"ParameterValue": "1",
"ApplyMethod": "immediate",
},
],
)
from prowler.providers.aws.services.rds.rds_service import RDS
audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_transport_encrypted.rds_instance_transport_encrypted.rds_client",
new=RDS(audit_info),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_transport_encrypted.rds_instance_transport_encrypted import (
rds_instance_transport_encrypted,
)
check = rds_instance_transport_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 connections use SSL encryption."
)
assert result[0].resource_id == "db-cluster-1"
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
)
assert result[0].resource_tags == []
@mock_aws
def test_rds_aurora_mysql_clustered_instance_ssl(self):
conn = client("rds", region_name=AWS_REGION_US_EAST_1)
conn.create_db_parameter_group(
DBParameterGroupName="test",
DBParameterGroupFamily="default.mysql8.0",
Description="test parameter group",
)
conn.create_db_cluster(
DBClusterIdentifier="db-cluster-1",
AllocatedStorage=10,
Engine="aurora-mysql",
DatabaseName="staging-mysql",
DeletionProtection=True,
DBClusterParameterGroupName="test",
MasterUsername="test",
MasterUserPassword="password",
Tags=[],
)
conn.modify_db_parameter_group(
DBParameterGroupName="test",
Parameters=[
{
"ParameterName": "require_secure_transport",
"ParameterValue": "ON",
"ApplyMethod": "immediate",
},
],
)
from prowler.providers.aws.services.rds.rds_service import RDS
audit_info = set_mocked_aws_audit_info([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=audit_info,
):
with mock.patch(
"prowler.providers.aws.services.rds.rds_instance_transport_encrypted.rds_instance_transport_encrypted.rds_client",
new=RDS(audit_info),
):
# Test Check
from prowler.providers.aws.services.rds.rds_instance_transport_encrypted.rds_instance_transport_encrypted import (
rds_instance_transport_encrypted,
)
check = rds_instance_transport_encrypted()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== "RDS Cluster db-cluster-1 connections use SSL encryption."
)
assert result[0].resource_id == "db-cluster-1"
assert result[0].region == AWS_REGION_US_EAST_1
assert (
result[0].resource_arn
== f"arn:aws:rds:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:cluster:db-cluster-1"
)
assert result[0].resource_tags == []
@@ -169,3 +169,31 @@ class Test_vpc_subnet_different_az:
assert result.region == AWS_REGION_US_EAST_1
if not found:
assert False
@mock_aws
def test_vpc_no_subnets_but_unused(self):
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
ec2_client.create_vpc(CidrBlock="172.28.7.0/24", InstanceTenancy="default")
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = set_mocked_aws_audit_info(
audited_regions=[AWS_REGION_US_EAST_1], ignore_unused_services=True
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_subnet_different_az.vpc_subnet_different_az.vpc_client",
new=VPC(current_audit_info),
):
from prowler.providers.aws.services.vpc.vpc_subnet_different_az.vpc_subnet_different_az import (
vpc_subnet_different_az,
)
check = vpc_subnet_different_az()
results = check.execute()
assert len(results) == 0
@@ -105,3 +105,51 @@ class Test_vpc_subnet_no_public_ip_by_default:
result.status_extended
== f"VPC subnet {subnet_private['Subnet']['SubnetId']} does NOT assign public IP by default."
)
@mock_aws
def test_vpc_with_map_ip_on_launch_but_unused(self):
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
vpc = ec2_client.create_vpc(
CidrBlock="172.28.7.0/24", InstanceTenancy="default"
)
subnet_private = ec2_client.create_subnet(
VpcId=vpc["Vpc"]["VpcId"],
CidrBlock="172.28.7.192/26",
AvailabilityZone=f"{AWS_REGION_US_EAST_1}a",
TagSpecifications=[
{
"ResourceType": "subnet",
"Tags": [
{"Key": "Name", "Value": "subnet_name"},
],
},
],
)
ec2_client.modify_subnet_attribute(
SubnetId=subnet_private["Subnet"]["SubnetId"],
MapPublicIpOnLaunch={"Value": True},
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = set_mocked_aws_audit_info(
audited_regions=[AWS_REGION_US_EAST_1], ignore_unused_services=True
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_subnet_no_public_ip_by_default.vpc_subnet_no_public_ip_by_default.vpc_client",
new=VPC(current_audit_info),
):
from prowler.providers.aws.services.vpc.vpc_subnet_no_public_ip_by_default.vpc_subnet_no_public_ip_by_default import (
vpc_subnet_no_public_ip_by_default,
)
check = vpc_subnet_no_public_ip_by_default()
results = check.execute()
assert len(results) == 0
@@ -132,6 +132,48 @@ class Test_vpc_subnet_separate_private_public:
if not found:
assert False
@mock_aws
def test_vpc_subnet_only_public_but_unused(self):
# Create EC2 Mocked Resources
ec2 = resource("ec2", region_name=AWS_REGION_US_EAST_1)
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock="10.0.0.0/18")
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
# Create IGW and attach to VPC
igw = ec2.create_internet_gateway()
vpc.attach_internet_gateway(InternetGatewayId=igw.id)
# Set IGW as default route for public subnet
route_table = ec2.create_route_table(VpcId=vpc.id)
route_table.associate_with_subnet(SubnetId=subnet.id)
ec2_client.create_route(
RouteTableId=route_table.id,
DestinationCidrBlock="0.0.0.0/0",
GatewayId=igw.id,
)
from prowler.providers.aws.services.vpc.vpc_service import VPC
current_audit_info = set_mocked_aws_audit_info(
audited_regions=[AWS_REGION_US_EAST_1], ignore_unused_services=True
)
with mock.patch(
"prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
new=current_audit_info,
):
with mock.patch(
"prowler.providers.aws.services.vpc.vpc_subnet_separate_private_public.vpc_subnet_separate_private_public.vpc_client",
new=VPC(current_audit_info),
):
from prowler.providers.aws.services.vpc.vpc_subnet_separate_private_public.vpc_subnet_separate_private_public import (
vpc_subnet_separate_private_public,
)
check = vpc_subnet_separate_private_public()
results = check.execute()
assert len(results) == 0
@mock_aws
def test_vpc_subnet_private_and_public(self):
ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1)
@@ -110,3 +110,64 @@ class Test_iam_custom_role_has_permissions_to_administer_resource_locks:
== defender_client.custom_roles[AZURE_SUBSCRIPTION][0].id
)
assert result[0].resource_name == role_name
def test_iam_custom_owner_roles_created_with_lock_administration_permissions(
self,
):
defender_client = mock.MagicMock
role_name = "test-role"
role_name2 = "test-role2"
defender_client.custom_roles = {
AZURE_SUBSCRIPTION: [
Role(
id=str(uuid4()),
name=role_name,
type="CustomRole",
assignable_scopes=["/.*", "/test"],
permissions=[
Permission(
actions=[
"Microsoft.Authorization/locks/*",
"microsoft.aadiam/azureADMetrics/read",
]
)
],
),
Role(
id=str(uuid4()),
name=role_name2,
type="CustomRole",
assignable_scopes=["/.*", "/test"],
permissions=[
Permission(
actions=[
"Microsoft.Authorization/locks/*",
"microsoft.aadiam/azureADMetrics/read",
]
)
],
),
]
}
with mock.patch(
"prowler.providers.azure.services.iam.iam_custom_role_has_permissions_to_administer_resource_locks.iam_custom_role_has_permissions_to_administer_resource_locks.iam_client",
new=defender_client,
):
from prowler.providers.azure.services.iam.iam_custom_role_has_permissions_to_administer_resource_locks.iam_custom_role_has_permissions_to_administer_resource_locks import (
iam_custom_role_has_permissions_to_administer_resource_locks,
)
check = iam_custom_role_has_permissions_to_administer_resource_locks()
result = check.execute()
assert len(result) == 1
assert result[0].status == "PASS"
assert (
result[0].status_extended
== f"Role {role_name} from subscription {AZURE_SUBSCRIPTION} has permission to administer resource locks."
)
assert result[0].subscription == AZURE_SUBSCRIPTION
assert (
result[0].resource_id
== defender_client.custom_roles[AZURE_SUBSCRIPTION][0].id
)