fix(aws): not show findings when AccessDenieds (#3803)

This commit is contained in:
Sergio Garcia
2024-04-29 17:42:44 +02:00
committed by GitHub
parent b36152484d
commit 35c8ea5e3f
97 changed files with 2332 additions and 1126 deletions
+2
View File
@@ -178,6 +178,8 @@ class <Service>(ServiceParentClass):
f"{<item>.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
```
???+note
To avoid fake findings, when Prowler can't retrieve the items, because an Access Denied or similar error, we set that items value as `None`.
#### Service Models
+4 -2
View File
@@ -345,8 +345,10 @@ def prowler():
global_provider,
global_provider.output_options,
)
# Only display compliance table if there are findings and it is a default execution
if findings and default_execution:
# Only display compliance table if there are findings (not all MANUAL) and it is a default execution
if (
findings and not all(finding.status == "MANUAL" for finding in findings)
) and default_execution:
compliance_overview = False
if not compliance_framework:
compliance_framework = get_available_compliance_frameworks(provider)
+8 -5
View File
@@ -40,11 +40,13 @@ def display_summary_table(
entity_type = "Context"
audited_entities = provider.identity.context
if findings:
# Check if there are findings and that they are not all MANUAL
if findings and not all(finding.status == "MANUAL" for finding in findings):
current = {
"Service": "",
"Provider": "",
"Total": 0,
"Pass": 0,
"Critical": 0,
"High": 0,
"Medium": 0,
@@ -70,9 +72,9 @@ def display_summary_table(
):
add_service_to_table(findings_table, current)
current["Total"] = current["Muted"] = current["Critical"] = current[
"High"
] = current["Medium"] = current["Low"] = 0
current["Total"] = current["Pass"] = current["Muted"] = current[
"Critical"
] = current["High"] = current["Medium"] = current["Low"] = 0
current["Service"] = finding.check_metadata.ServiceName
current["Provider"] = finding.check_metadata.Provider
@@ -83,6 +85,7 @@ def display_summary_table(
current["Muted"] += 1
if finding.status == "PASS":
pass_count += 1
current["Pass"] += 1
elif finding.status == "FAIL":
fail_count += 1
if finding.check_metadata.Severity == "critical":
@@ -155,7 +158,7 @@ def add_service_to_table(findings_table, current):
)
current["Status"] = f"{Fore.RED}FAIL ({total_fails}){Style.RESET_ALL}"
else:
current["Status"] = f"{Fore.GREEN}PASS ({current['Total']}){Style.RESET_ALL}"
current["Status"] = f"{Fore.GREEN}PASS ({current['Pass']}){Style.RESET_ALL}"
findings_table["Provider"].append(current["Provider"])
findings_table["Service"].append(current["Service"])
+9 -2
View File
@@ -789,7 +789,7 @@ class AwsProvider(Provider):
def get_aws_enabled_regions(self, current_session: Session) -> set:
"""get_aws_enabled_regions returns a set of enabled AWS regions"""
try:
# EC2 Client to check enabled regions
service = "ec2"
default_region = self.get_default_region(service)
@@ -797,10 +797,17 @@ class AwsProvider(Provider):
enabled_regions = set()
# With AllRegions=False we only get the enabled regions for the account
for region in ec2_client.describe_regions(AllRegions=False).get("Regions", []):
for region in ec2_client.describe_regions(AllRegions=False).get(
"Regions", []
):
enabled_regions.add(region.get("RegionName"))
return enabled_regions
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return set()
# TODO: review this function
# Maybe this should be done within the AwsProvider and not in __main__.py
@@ -6,6 +6,8 @@ class account_maintain_different_contact_details_to_security_billing_and_operati
Check
):
def execute(self):
findings = []
if account_client.contact_base:
report = Check_Report_AWS(self.metadata())
report.region = account_client.region
report.resource_id = account_client.audited_account
@@ -14,7 +16,8 @@ class account_maintain_different_contact_details_to_security_billing_and_operati
if (
len(account_client.contact_phone_numbers)
== account_client.number_of_contacts
and len(account_client.contact_names) == account_client.number_of_contacts
and len(account_client.contact_names)
== account_client.number_of_contacts
# This is because the primary contact has no email field
and len(account_client.contact_emails)
== account_client.number_of_contacts - 1
@@ -24,4 +27,5 @@ class account_maintain_different_contact_details_to_security_billing_and_operati
else:
report.status = "FAIL"
report.status_extended = "SECURITY, BILLING and OPERATIONS contacts not found or they are not different between each other and between ROOT contact."
return [report]
findings.append(report)
return findings
@@ -18,6 +18,7 @@ class Account(AWSService):
self.contacts_security = self.__get_alternate_contact__("SECURITY")
self.contacts_operations = self.__get_alternate_contact__("OPERATIONS")
if self.contact_base:
# Set of contact phone numbers
self.contact_phone_numbers = {
self.contact_base.phone_number,
@@ -53,6 +54,12 @@ class Account(AWSService):
phone_number=primary_account_contact.get("PhoneNumber"),
)
except Exception as error:
if error.response["Error"]["Code"] == "AccessDeniedException":
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return None
else:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
@@ -1,6 +1,7 @@
from datetime import datetime
from typing import Optional
from botocore.client import ClientError
from pydantic import BaseModel
from prowler.lib.logger import logger
@@ -37,6 +38,8 @@ class Backup(AWSService):
self.audit_resources,
)
):
if self.backup_vaults is None:
self.backup_vaults = []
self.backup_vaults.append(
BackupVault(
arn=configuration.get("BackupVaultArn"),
@@ -55,7 +58,13 @@ class Backup(AWSService):
),
)
)
except ClientError as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if error.response["Error"]["Code"] == "AccessDeniedException":
if not self.backup_vaults:
self.backup_vaults = None
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -5,7 +5,7 @@ from prowler.providers.aws.services.backup.backup_client import backup_client
class backup_vaults_encrypted(Check):
def execute(self):
findings = []
if backup_client.backup_vaults:
for backup_vault in backup_client.backup_vaults:
# By default we assume that the result is fail
report = Check_Report_AWS(self.metadata())
@@ -5,6 +5,7 @@ from prowler.providers.aws.services.backup.backup_client import backup_client
class backup_vaults_exist(Check):
def execute(self):
findings = []
if backup_client.backup_vaults is not None:
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No Backup Vault exist."
@@ -8,6 +8,7 @@ from prowler.providers.aws.services.s3.s3_client import s3_client
class cloudtrail_bucket_requires_mfa_delete(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
if trail.is_logging:
trail_bucket_is_in_account = False
@@ -11,6 +11,7 @@ maximum_time_without_logging = 1
class cloudtrail_cloudwatch_logging_enabled(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
if trail.name:
report = Check_Report_AWS(self.metadata())
@@ -20,9 +21,7 @@ class cloudtrail_cloudwatch_logging_enabled(Check):
report.resource_tags = trail.tags
report.status = "PASS"
if trail.is_multiregion:
report.status_extended = (
f"Multiregion trail {trail.name} has been logging the last 24h."
)
report.status_extended = f"Multiregion trail {trail.name} has been logging the last 24h."
else:
report.status_extended = f"Single region trail {trail.name} has been logging the last 24h."
if trail.latest_cloudwatch_delivery_time:
@@ -30,7 +29,9 @@ class cloudtrail_cloudwatch_logging_enabled(Check):
datetime.now().replace(tzinfo=timezone.utc)
- trail.latest_cloudwatch_delivery_time
)
if last_log_delivery > timedelta(days=maximum_time_without_logging):
if last_log_delivery > timedelta(
days=maximum_time_without_logging
):
report.status = "FAIL"
if trail.is_multiregion:
report.status_extended = f"Multiregion trail {trail.name} is not logging in the last 24h."
@@ -7,6 +7,7 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
class cloudtrail_insights_exist(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
if trail.is_logging:
report = Check_Report_AWS(self.metadata())
@@ -18,8 +19,6 @@ class cloudtrail_insights_exist(Check):
report.status_extended = f"Trail {trail.name} does not have insight selectors and it is logging."
if trail.has_insight_selectors:
report.status = "PASS"
report.status_extended = (
f"Trail {trail.name} has insight selectors and it is logging."
)
report.status_extended = f"Trail {trail.name} has insight selectors and it is logging."
findings.append(report)
return findings
@@ -7,6 +7,7 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
class cloudtrail_kms_encryption_enabled(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
if trail.name:
report = Check_Report_AWS(self.metadata())
@@ -26,13 +27,9 @@ class cloudtrail_kms_encryption_enabled(Check):
if trail.kms_key:
report.status = "PASS"
if trail.is_multiregion:
report.status_extended = (
f"Multiregion trail {trail.name} has encryption enabled."
)
report.status_extended = f"Multiregion trail {trail.name} has encryption enabled."
else:
report.status_extended = (
f"Single region trail {trail.name} has encryption enabled."
)
report.status_extended = f"Single region trail {trail.name} has encryption enabled."
findings.append(report)
return findings
@@ -7,6 +7,7 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
class cloudtrail_log_file_validation_enabled(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
if trail.name:
report = Check_Report_AWS(self.metadata())
@@ -16,9 +17,7 @@ class cloudtrail_log_file_validation_enabled(Check):
report.resource_tags = trail.tags
report.status = "FAIL"
if trail.is_multiregion:
report.status_extended = (
f"Multiregion trail {trail.name} log file validation disabled."
)
report.status_extended = f"Multiregion trail {trail.name} log file validation disabled."
else:
report.status_extended = f"Single region trail {trail.name} log file validation disabled."
if trail.log_file_validation_enabled:
@@ -8,6 +8,7 @@ from prowler.providers.aws.services.s3.s3_client import s3_client
class cloudtrail_logs_s3_bucket_access_logging_enabled(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
if trail.name:
trail_bucket_is_in_account = False
@@ -8,6 +8,7 @@ from prowler.providers.aws.services.s3.s3_client import s3_client
class cloudtrail_logs_s3_bucket_is_not_publicly_accessible(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
if trail.name:
trail_bucket_is_in_account = False
@@ -7,6 +7,7 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
class cloudtrail_multi_region_enabled(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for region in cloudtrail_client.regional_clients.keys():
report = Check_Report_AWS(self.metadata())
report.region = region
@@ -18,9 +19,7 @@ class cloudtrail_multi_region_enabled(Check):
report.resource_arn = trail.arn
report.resource_tags = trail.tags
if trail.is_multiregion:
report.status_extended = (
f"Trail {trail.name} is multiregion and it is logging."
)
report.status_extended = f"Trail {trail.name} is multiregion and it is logging."
else:
report.status_extended = f"Trail {trail.name} is not multiregion and it is logging."
# Since there exists a logging trail in that region there is no point in checking the remaining trails
@@ -7,11 +7,10 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
class cloudtrail_multi_region_enabled_logging_management_events(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No trail found with multi-region enabled and logging management events."
)
report.status_extended = "No trail found with multi-region enabled and logging management events."
report.region = cloudtrail_client.region
report.resource_id = cloudtrail_client.audited_account
report.resource_arn = cloudtrail_client.trail_arn_template
@@ -41,7 +40,9 @@ class cloudtrail_multi_region_enabled_logging_management_events(Check):
) == "Management events selector" and all(
[
field["Field"] != "readOnly"
for field in event.event_selector["FieldSelectors"]
for field in event.event_selector[
"FieldSelectors"
]
]
):
report.region = trail.region
@@ -8,6 +8,7 @@ from prowler.providers.aws.services.s3.s3_client import s3_client
class cloudtrail_s3_dataevents_read_enabled(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
for data_event in trail.data_events:
# classic event selectors
@@ -36,7 +37,9 @@ class cloudtrail_s3_dataevents_read_enabled(Check):
findings.append(report)
# advanced event selectors
elif data_event.is_advanced:
for field_selector in data_event.event_selector["FieldSelectors"]:
for field_selector in data_event.event_selector[
"FieldSelectors"
]:
if (
field_selector["Field"] == "resources.type"
and field_selector["Equals"][0] == "AWS::S3::Object"
@@ -8,6 +8,7 @@ from prowler.providers.aws.services.s3.s3_client import s3_client
class cloudtrail_s3_dataevents_write_enabled(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
for trail in cloudtrail_client.trails.values():
for data_event in trail.data_events:
# Classic event selectors
@@ -36,7 +37,9 @@ class cloudtrail_s3_dataevents_write_enabled(Check):
findings.append(report)
# Advanced event selectors
elif data_event.is_advanced:
for field_selector in data_event.event_selector["FieldSelectors"]:
for field_selector in data_event.event_selector[
"FieldSelectors"
]:
if (
field_selector["Field"] == "resources.type"
and field_selector["Equals"][0] == "AWS::S3::Object"
@@ -17,6 +17,7 @@ class Cloudtrail(AWSService):
self.trail_arn_template = f"arn:{self.audited_partition}:cloudtrail:{self.region}:{self.audited_account}:trail"
self.trails = {}
self.__threading_call__(self.__get_trails__)
if self.trails:
self.__get_trail_status__()
self.__get_insight_selectors__()
self.__get_event_selectors__()
@@ -45,6 +46,8 @@ class Cloudtrail(AWSService):
kms_key_id = trail["KmsKeyId"]
if "CloudWatchLogsLogGroupArn" in trail:
log_group_arn = trail["CloudWatchLogsLogGroupArn"]
if self.trails is None:
self.trails = {}
self.trails[trail["TrailARN"]] = Trail(
name=trail["Name"],
is_multiregion=trail["IsMultiRegionTrail"],
@@ -61,12 +64,24 @@ class Cloudtrail(AWSService):
has_insight_selectors=trail.get("HasInsightSelectors"),
)
if trails_count == 0:
if self.trails is None:
self.trails = {}
self.trails[self.__get_trail_arn_template__(regional_client.region)] = (
Trail(
region=regional_client.region,
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDeniedException":
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if not self.trails:
self.trails = None
else:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -15,11 +15,14 @@ class cloudwatch_changes_to_network_acls_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateNetworkAcl.+\$\.eventName\s*=\s*.?CreateNetworkAclEntry.+\$\.eventName\s*=\s*.?DeleteNetworkAcl.+\$\.eventName\s*=\s*.?DeleteNetworkAclEntry.+\$\.eventName\s*=\s*.?ReplaceNetworkAclEntry.+\$\.eventName\s*=\s*.?ReplaceNetworkAclAssociation.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_changes_to_network_gateways_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateCustomerGateway.+\$\.eventName\s*=\s*.?DeleteCustomerGateway.+\$\.eventName\s*=\s*.?AttachInternetGateway.+\$\.eventName\s*=\s*.?CreateInternetGateway.+\$\.eventName\s*=\s*.?DeleteInternetGateway.+\$\.eventName\s*=\s*.?DetachInternetGateway.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_changes_to_network_route_tables_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?ec2.amazonaws.com.+\$\.eventName\s*=\s*.?CreateRoute.+\$\.eventName\s*=\s*.?CreateRouteTable.+\$\.eventName\s*=\s*.?ReplaceRoute.+\$\.eventName\s*=\s*.?ReplaceRouteTableAssociation.+\$\.eventName\s*=\s*.?DeleteRouteTable.+\$\.eventName\s*=\s*.?DeleteRoute.+\$\.eventName\s*=\s*.?DisassociateRouteTable.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_changes_to_vpcs_alarm_configured(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateVpc.+\$\.eventName\s*=\s*.?DeleteVpc.+\$\.eventName\s*=\s*.?ModifyVpcAttribute.+\$\.eventName\s*=\s*.?AcceptVpcPeeringConnection.+\$\.eventName\s*=\s*.?CreateVpcPeeringConnection.+\$\.eventName\s*=\s*.?DeleteVpcPeeringConnection.+\$\.eventName\s*=\s*.?RejectVpcPeeringConnection.+\$\.eventName\s*=\s*.?AttachClassicLinkVpc.+\$\.eventName\s*=\s*.?DetachClassicLinkVpc.+\$\.eventName\s*=\s*.?DisableVpcClassicLink.+\$\.eventName\s*=\s*.?EnableVpcClassicLink.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -5,6 +5,7 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class cloudwatch_cross_account_sharing_disabled(Check):
def execute(self):
findings = []
if iam_client.roles is not None:
report = Check_Report_AWS(self.metadata())
report.status = "PASS"
report.status_extended = "CloudWatch doesn't allow cross-account sharing."
@@ -16,6 +17,8 @@ class cloudwatch_cross_account_sharing_disabled(Check):
report.resource_arn = role.arn
report.resource_id = role.name
report.status = "FAIL"
report.status_extended = "CloudWatch has allowed cross-account sharing."
report.status_extended = (
"CloudWatch has allowed cross-account sharing."
)
findings.append(report)
return findings
@@ -5,6 +5,7 @@ from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
class cloudwatch_log_group_kms_encryption_enabled(Check):
def execute(self):
findings = []
if logs_client.log_groups:
for log_group in logs_client.log_groups:
report = Check_Report_AWS(self.metadata())
report.region = log_group.region
@@ -16,8 +17,6 @@ class cloudwatch_log_group_kms_encryption_enabled(Check):
report.status_extended = f"Log Group {log_group.name} does have AWS KMS key {log_group.kms_id} associated."
else:
report.status = "FAIL"
report.status_extended = (
f"Log Group {log_group.name} does not have AWS KMS keys associated."
)
report.status_extended = f"Log Group {log_group.name} does not have AWS KMS keys associated."
findings.append(report)
return findings
@@ -11,10 +11,13 @@ from prowler.providers.aws.services.cloudwatch.logs_client import logs_client
class cloudwatch_log_group_no_secrets_in_logs(Check):
def execute(self):
findings = []
if logs_client.log_groups:
for log_group in logs_client.log_groups:
report = Check_Report_AWS(self.metadata())
report.status = "PASS"
report.status_extended = f"No secrets found in {log_group.name} log group."
report.status_extended = (
f"No secrets found in {log_group.name} log group."
)
report.region = log_group.region
report.resource_id = log_group.name
report.resource_arn = log_group.arn
@@ -40,8 +43,13 @@ class cloudwatch_log_group_no_secrets_in_logs(Check):
flagged_event["timestamp"]
)
)
if cloudwatch_timestamp not in log_stream_secrets.keys():
log_stream_secrets[cloudwatch_timestamp] = SecretsDict()
if (
cloudwatch_timestamp
not in log_stream_secrets.keys()
):
log_stream_secrets[cloudwatch_timestamp] = (
SecretsDict()
)
try:
log_event_data = dumps(
@@ -10,6 +10,7 @@ class cloudwatch_log_group_retention_policy_specific_days_enabled(Check):
specific_retention_days = logs_client.audit_config.get(
"log_group_retention_days", 365
)
if logs_client.log_groups:
for log_group in logs_client.log_groups:
report = Check_Report_AWS(self.metadata())
report.region = log_group.region
@@ -17,11 +17,14 @@ class cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_change
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?config.amazonaws.com.+\$\.eventName\s*=\s*.?StopConfigurationRecorder.+\$\.eventName\s*=\s*.?DeleteDeliveryChannel.+\$\.eventName\s*=\s*.?PutDeliveryChannel.+\$\.eventName\s*=\s*.?PutConfigurationRecorder.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -17,11 +17,14 @@ class cloudwatch_log_metric_filter_and_alarm_for_cloudtrail_configuration_change
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?CreateTrail.+\$\.eventName\s*=\s*.?UpdateTrail.+\$\.eventName\s*=\s*.?DeleteTrail.+\$\.eventName\s*=\s*.?StartLogging.+\$\.eventName\s*=\s*.?StopLogging.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_authentication_failures(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?ConsoleLogin.+\$\.errorMessage\s*=\s*.?Failed authentication.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_aws_organizations_changes(Check):
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?organizations\.amazonaws\.com.+\$\.eventName\s*=\s*.?AcceptHandshake.+\$\.eventName\s*=\s*.?AttachPolicy.+\$\.eventName\s*=\s*.?CancelHandshake.+\$\.eventName\s*=\s*.?CreateAccount.+\$\.eventName\s*=\s*.?CreateOrganization.+\$\.eventName\s*=\s*.?CreateOrganizationalUnit.+\$\.eventName\s*=\s*.?CreatePolicy.+\$\.eventName\s*=\s*.?DeclineHandshake.+\$\.eventName\s*=\s*.?DeleteOrganization.+\$\.eventName\s*=\s*.?DeleteOrganizationalUnit.+\$\.eventName\s*=\s*.?DeletePolicy.+\$\.eventName\s*=\s*.?EnableAllFeatures.+\$\.eventName\s*=\s*.?EnablePolicyType.+\$\.eventName\s*=\s*.?InviteAccountToOrganization.+\$\.eventName\s*=\s*.?LeaveOrganization.+\$\.eventName\s*=\s*.?DetachPolicy.+\$\.eventName\s*=\s*.?DisablePolicyType.+\$\.eventName\s*=\s*.?MoveAccount.+\$\.eventName\s*=\s*.?RemoveAccountFromOrganization.+\$\.eventName\s*=\s*.?UpdateOrganizationalUnit.+\$\.eventName\s*=\s*.?UpdatePolicy.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_disable_or_scheduled_deletion_of_kms_cmk(Chec
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?kms.amazonaws.com.+\$\.eventName\s*=\s*.?DisableKey.+\$\.eventName\s*=\s*.?ScheduleKeyDeletion.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_for_s3_bucket_policy_changes(Check):
def execute(self):
pattern = r"\$\.eventSource\s*=\s*.?s3.amazonaws.com.+\$\.eventName\s*=\s*.?PutBucketAcl.+\$\.eventName\s*=\s*.?PutBucketPolicy.+\$\.eventName\s*=\s*.?PutBucketCors.+\$\.eventName\s*=\s*.?PutBucketLifecycle.+\$\.eventName\s*=\s*.?PutBucketReplication.+\$\.eventName\s*=\s*.?DeleteBucketPolicy.+\$\.eventName\s*=\s*.?DeleteBucketCors.+\$\.eventName\s*=\s*.?DeleteBucketLifecycle.+\$\.eventName\s*=\s*.?DeleteBucketReplication.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_policy_changes(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?DeleteGroupPolicy.+\$\.eventName\s*=\s*.?DeleteRolePolicy.+\$\.eventName\s*=\s*.?DeleteUserPolicy.+\$\.eventName\s*=\s*.?PutGroupPolicy.+\$\.eventName\s*=\s*.?PutRolePolicy.+\$\.eventName\s*=\s*.?PutUserPolicy.+\$\.eventName\s*=\s*.?CreatePolicy.+\$\.eventName\s*=\s*.?DeletePolicy.+\$\.eventName\s*=\s*.?CreatePolicyVersion.+\$\.eventName\s*=\s*.?DeletePolicyVersion.+\$\.eventName\s*=\s*.?AttachRolePolicy.+\$\.eventName\s*=\s*.?DetachRolePolicy.+\$\.eventName\s*=\s*.?AttachUserPolicy.+\$\.eventName\s*=\s*.?DetachUserPolicy.+\$\.eventName\s*=\s*.?AttachGroupPolicy.+\$\.eventName\s*=\s*.?DetachGroupPolicy.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_root_usage(Check):
def execute(self):
pattern = r"\$\.userIdentity\.type\s*=\s*.?Root.+\$\.userIdentity\.invokedBy NOT EXISTS.+\$\.eventType\s*!=\s*.?AwsServiceEvent.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_security_group_changes(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?AuthorizeSecurityGroupIngress.+\$\.eventName\s*=\s*.?AuthorizeSecurityGroupEgress.+\$\.eventName\s*=\s*.?RevokeSecurityGroupIngress.+\$\.eventName\s*=\s*.?RevokeSecurityGroupEgress.+\$\.eventName\s*=\s*.?CreateSecurityGroup.+\$\.eventName\s*=\s*.?DeleteSecurityGroup.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_sign_in_without_mfa(Check):
def execute(self):
pattern = r"\$\.eventName\s*=\s*.?ConsoleLogin.+\$\.additionalEventData\.MFAUsed\s*!=\s*.?Yes.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -15,11 +15,14 @@ class cloudwatch_log_metric_filter_unauthorized_api_calls(Check):
def execute(self):
pattern = r"\$\.errorCode\s*=\s*.?\*UnauthorizedOperation.+\$\.errorCode\s*=\s*.?AccessDenied\*.?"
findings = []
if (
cloudtrail_client.trails is not None
and logs_client.metric_filters is not None
and cloudwatch_client.metric_alarms is not None
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
"No CloudWatch log groups found with metric filters or alarms associated."
)
report.status_extended = "No CloudWatch log groups found with metric filters or alarms associated."
report.region = logs_client.region
report.resource_id = logs_client.audited_account
report.resource_arn = logs_client.log_group_arn_template
@@ -16,6 +16,7 @@ class CloudWatch(AWSService):
super().__init__(__class__.__name__, provider)
self.metric_alarms = []
self.__threading_call__(self.__describe_alarms__)
if self.metric_alarms:
self.__list_tags_for_resource__()
def __describe_alarms__(self, regional_client):
@@ -33,6 +34,8 @@ class CloudWatch(AWSService):
namespace = None
if "Namespace" in alarm:
namespace = alarm["Namespace"]
if self.metric_alarms is None:
self.metric_alarms = []
self.metric_alarms.append(
MetricAlarm(
arn=alarm["AlarmArn"],
@@ -42,6 +45,17 @@ class CloudWatch(AWSService):
region=regional_client.region,
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDenied":
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if not self.metric_alarms:
self.metric_alarms = None
else:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -72,6 +86,7 @@ class Logs(AWSService):
self.log_groups = []
self.__threading_call__(self.__describe_metric_filters__)
self.__threading_call__(self.__describe_log_groups__)
if self.log_groups:
if (
"cloudwatch_log_group_no_secrets_in_logs"
in provider.audit_metadata.expected_checks
@@ -94,6 +109,8 @@ class Logs(AWSService):
if not self.audit_resources or (
is_resource_filtered(arn, self.audit_resources)
):
if self.metric_filters is None:
self.metric_filters = []
self.metric_filters.append(
MetricFilter(
arn=arn,
@@ -104,6 +121,17 @@ class Logs(AWSService):
region=regional_client.region,
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDeniedException":
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if not self.metric_filters:
self.metric_filters = []
else:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -126,6 +154,8 @@ class Logs(AWSService):
if not retention_days:
never_expire = True
retention_days = 9999
if self.log_groups is None:
self.log_groups = []
self.log_groups.append(
LogGroup(
arn=log_group["arn"],
@@ -136,6 +166,17 @@ class Logs(AWSService):
region=regional_client.region,
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDeniedException":
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if not self.log_groups:
self.log_groups = None
else:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -12,6 +12,7 @@ def check_cloudwatch_log_metric_filter(
):
# 1. Iterate for CloudWatch Log Group in CloudTrail trails
log_groups = []
if trails is not None:
for trail in trails.values():
if trail.log_group_arn:
log_groups.append(trail.log_group_arn.split(":")[6])
@@ -52,6 +52,11 @@ class FMS(AWSService):
):
# FMS is not enabled in this account
self.fms_admin_account = False
else:
logger.error(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
)
self.fms_admin_account = None
except Exception as error:
logger.error(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
@@ -5,12 +5,11 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_password_policy_expires_passwords_within_90_days_or_less(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.password_policy:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.password_policy_arn_template
report.resource_id = iam_client.audited_account
# Check if password policy exists
if iam_client.password_policy:
# Check if password policy expiration exists
if iam_client.password_policy.max_age:
if iam_client.password_policy.max_age <= 90:
@@ -22,9 +21,6 @@ class iam_password_policy_expires_passwords_within_90_days_or_less(Check):
else:
report.status = "FAIL"
report.status_extended = "Password expiration is not set."
else:
report.status = "FAIL"
report.status_extended = "Password policy cannot be found."
findings.append(report)
return findings
@@ -5,12 +5,11 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_password_policy_lowercase(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.password_policy:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.password_policy_arn_template
report.resource_id = iam_client.audited_account
# Check if password policy exists
if iam_client.password_policy:
# Check if lowercase flag is set
if iam_client.password_policy.lowercase:
report.status = "PASS"
@@ -20,8 +19,5 @@ class iam_password_policy_lowercase(Check):
else:
report.status = "FAIL"
report.status_extended = "IAM password policy does not require at least one lowercase letter."
else:
report.status = "FAIL"
report.status_extended = "Password policy cannot be found."
findings.append(report)
return findings
@@ -5,12 +5,11 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_password_policy_minimum_length_14(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.password_policy:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.password_policy_arn_template
report.resource_id = iam_client.audited_account
# Check if password policy exists
if iam_client.password_policy:
# Check password policy length
if (
iam_client.password_policy.length
@@ -23,8 +22,5 @@ class iam_password_policy_minimum_length_14(Check):
else:
report.status = "FAIL"
report.status_extended = "IAM password policy does not require minimum length of 14 characters."
else:
report.status = "FAIL"
report.status_extended = "Password policy cannot be found."
findings.append(report)
return findings
@@ -5,12 +5,11 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_password_policy_number(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.password_policy:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.password_policy_arn_template
report.resource_id = iam_client.audited_account
# Check if password policy exists
if iam_client.password_policy:
# Check if number flag is set
if iam_client.password_policy.numbers:
report.status = "PASS"
@@ -22,8 +21,5 @@ class iam_password_policy_number(Check):
report.status_extended = (
"IAM password policy does not require at least one number."
)
else:
report.status = "FAIL"
report.status_extended = "Password policy cannot be found."
findings.append(report)
return findings
@@ -5,12 +5,11 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_password_policy_reuse_24(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.password_policy:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.password_policy_arn_template
report.resource_id = iam_client.audited_account
# Check if password policy exists
if iam_client.password_policy:
# Check if reuse prevention flag is set
if (
iam_client.password_policy.reuse_prevention
@@ -25,8 +24,5 @@ class iam_password_policy_reuse_24(Check):
report.status_extended = (
"IAM password policy reuse prevention is less than 24 or not set."
)
else:
report.status = "FAIL"
report.status_extended = "Password policy cannot be found."
findings.append(report)
return findings
@@ -5,12 +5,11 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_password_policy_symbol(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.password_policy:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.password_policy_arn_template
report.resource_id = iam_client.audited_account
# Check if password policy exists
if iam_client.password_policy:
# Check if symbol flag is set
if iam_client.password_policy.symbols:
report.status = "PASS"
@@ -22,8 +21,5 @@ class iam_password_policy_symbol(Check):
report.status_extended = (
"IAM password policy does not require at least one symbol."
)
else:
report.status = "FAIL"
report.status_extended = "Password policy cannot be found."
findings.append(report)
return findings
@@ -5,12 +5,11 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_password_policy_uppercase(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.password_policy:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_arn = iam_client.password_policy_arn_template
report.resource_id = iam_client.audited_account
# Check if password policy exists
if iam_client.password_policy:
# Check if uppercase flag is set
if iam_client.password_policy.uppercase:
report.status = "PASS"
@@ -20,8 +19,5 @@ class iam_password_policy_uppercase(Check):
else:
report.status = "FAIL"
report.status_extended = "IAM password policy does not require at least one uppercase letter."
else:
report.status = "FAIL"
report.status_extended = "Password policy cannot be found."
findings.append(report)
return findings
@@ -5,6 +5,7 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_role_administratoraccess_policy(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.roles:
for role in iam_client.roles:
if (
not role.is_service_role
@@ -15,9 +16,7 @@ class iam_role_administratoraccess_policy(Check):
report.resource_id = role.name
report.resource_tags = role.tags
report.status = "PASS"
report.status_extended = (
f"IAM Role {role.name} does not have AdministratorAccess policy."
)
report.status_extended = f"IAM Role {role.name} does not have AdministratorAccess policy."
for policy in role.attached_policies:
if policy["PolicyName"] == "AdministratorAccess":
report.status_extended = f"IAM Role {role.name} has AdministratorAccess policy attached."
@@ -5,6 +5,7 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_role_cross_account_readonlyaccess_policy(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.roles:
for role in iam_client.roles:
if (
not role.is_service_role
@@ -32,9 +33,9 @@ class iam_role_cross_account_readonlyaccess_policy(Check):
if isinstance(
statement["Principal"]["AWS"], list
):
for aws_account in statement["Principal"][
"AWS"
]:
for aws_account in statement[
"Principal"
]["AWS"]:
if (
iam_client.audited_account
not in aws_account
@@ -46,7 +47,8 @@ class iam_role_cross_account_readonlyaccess_policy(Check):
if (
iam_client.audited_account
not in statement["Principal"]["AWS"]
or "*" == statement["Principal"]["AWS"]
or "*"
== statement["Principal"]["AWS"]
):
cross_account_access = True
else:
@@ -58,7 +60,9 @@ class iam_role_cross_account_readonlyaccess_policy(Check):
and "AWS" in statement["Principal"]
):
if isinstance(statement["Principal"]["AWS"], list):
for aws_account in statement["Principal"]["AWS"]:
for aws_account in statement["Principal"][
"AWS"
]:
if (
iam_client.audited_account
not in aws_account
@@ -8,6 +8,7 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_role_cross_service_confused_deputy_prevention(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.roles:
for role in iam_client.roles:
# This check should only be performed against service roles (avoid Service Linked Roles since the trust relationship cannot be changed)
if role.is_service_role and "aws-service-role" not in role.arn:
@@ -5,6 +5,7 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_securityaudit_role_created(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.entities_role_attached_to_securityaudit_policy is not None:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = "SecurityAudit"
@@ -14,6 +15,8 @@ class iam_securityaudit_role_created(Check):
report.status_extended = f"SecurityAudit policy attached to role {iam_client.entities_role_attached_to_securityaudit_policy[0]['RoleName']}."
else:
report.status = "FAIL"
report.status_extended = "SecurityAudit policy is not attached to any role."
report.status_extended = (
"SecurityAudit policy is not attached to any role."
)
findings.append(report)
return findings
@@ -117,6 +117,16 @@ class IAM(AWSService):
is_service_role=is_service_role(role),
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDenied":
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
roles = None
else:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -219,11 +229,28 @@ class IAM(AWSService):
except ClientError as error:
if error.response["Error"]["Code"] == "NoSuchEntity":
# Password policy does not exist
stored_password_policy = None
# Password policy is the IAM default
stored_password_policy = PasswordPolicy(
length=8,
symbols=False,
numbers=False,
uppercase=False,
lowercase=False,
allow_change=True,
expiration=False,
max_age=None,
reuse_prevention=None,
hard_expiry=None,
)
logger.warning(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
elif error.response["Error"]["Code"] == "AccessDenied":
# User does not have permission to get password policy
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
stored_password_policy = None
else:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -395,11 +422,12 @@ class IAM(AWSService):
def __list_attached_role_policies__(self):
logger.info("IAM - List Attached User Policies...")
try:
if self.roles:
for role in self.roles:
try:
attached_role_policies = []
list_attached_role_policies_paginator = self.client.get_paginator(
"list_attached_role_policies"
list_attached_role_policies_paginator = (
self.client.get_paginator("list_attached_role_policies")
)
for page in list_attached_role_policies_paginator.paginate(
RoleName=role.name
@@ -548,6 +576,7 @@ class IAM(AWSService):
def __list_inline_role_policies__(self):
logger.info("IAM - List Inline Role Policies...")
if self.roles:
for role in self.roles:
try:
inline_role_policies = []
@@ -615,6 +644,17 @@ class IAM(AWSService):
roles = self.client.list_entities_for_policy(
PolicyArn=policy_arn, EntityFilter="Role"
)["PolicyRoles"]
return roles
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDenied":
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
roles = None
else:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -717,9 +757,12 @@ class IAM(AWSService):
def __list_tags_for_resource__(self):
logger.info("IAM - List Tags...")
try:
if self.roles:
for role in self.roles:
try:
response = self.client.list_role_tags(RoleName=role.name)["Tags"]
response = self.client.list_role_tags(RoleName=role.name)[
"Tags"
]
role.tags = response
except ClientError as error:
if error.response["Error"]["Code"] == "NoSuchEntity":
@@ -5,6 +5,7 @@ from prowler.providers.aws.services.iam.iam_client import iam_client
class iam_support_role_created(Check):
def execute(self) -> Check_Report_AWS:
findings = []
if iam_client.entities_role_attached_to_support_policy is not None:
report = Check_Report_AWS(self.metadata())
report.region = iam_client.region
report.resource_id = "AWSSupportServiceRolePolicy"
@@ -7,6 +7,7 @@ from prowler.providers.aws.services.resourceexplorer2.resourceexplorer2_client i
class resourceexplorer2_indexes_found(Check):
def execute(self):
findings = []
if resource_explorer_2_client.indexes is not None:
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No Resource Explorer Indexes found."
@@ -1,3 +1,4 @@
from botocore.client import ClientError
from pydantic import BaseModel
from prowler.lib.logger import logger
@@ -23,6 +24,8 @@ class ResourceExplorer2(AWSService):
if not self.audit_resources or (
is_resource_filtered(index["Arn"], self.audit_resources)
):
if self.indexes is None:
self.indexes = []
self.indexes.append(
Indexes(
arn=index["Arn"],
@@ -30,6 +33,17 @@ class ResourceExplorer2(AWSService):
type=index["Type"],
)
)
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDeniedException":
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if not self.indexes:
self.indexes = None
else:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -7,6 +7,7 @@ from prowler.providers.aws.services.ssmincidents.ssmincidents_client import (
class ssmincidents_enabled_with_plans(Check):
def execute(self):
findings = []
if ssmincidents_client.replication_set is not None:
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No SSM Incidents replication set exists."
@@ -45,6 +45,17 @@ class SSMIncidents(AWSService):
arn=replication_set,
)
]
except ClientError as error:
if error.response["Error"]["Code"] == "AccessDeniedException":
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if not self.replication_set:
self.replication_set = None
else:
logger.error(
f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
@@ -7,6 +7,7 @@ from prowler.providers.aws.services.trustedadvisor.trustedadvisor_client import
class trustedadvisor_errors_and_warnings(Check):
def execute(self):
findings = []
if trustedadvisor_client.premium_support:
if trustedadvisor_client.premium_support.enabled:
if trustedadvisor_client.checks:
for check in trustedadvisor_client.checks:
@@ -7,7 +7,12 @@ from prowler.providers.aws.services.trustedadvisor.trustedadvisor_client import
class trustedadvisor_premium_support_plan_subscribed(Check):
def execute(self):
findings = []
if trustedadvisor_client.audit_config.get("verify_premium_support_plans", True):
if (
trustedadvisor_client.premium_support
and trustedadvisor_client.audit_config.get(
"verify_premium_support_plans", True
)
):
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = (
@@ -22,7 +22,6 @@ class TrustedAdvisor(AWSService):
support_region = "us-east-1"
else:
support_region = "us-gov-west-1"
self.client = self.session.client(self.service, region_name=support_region)
self.client.region = support_region
self.__describe_services__()
@@ -106,6 +105,19 @@ class TrustedAdvisor(AWSService):
f" {error.__class__.__name__}[{error.__traceback__.tb_lineno}]:"
f" {error}"
)
elif error.response["Error"]["Code"] == "AccessDeniedException":
logger.error(
f"{self.region} --"
f" {error.__class__.__name__}[{error.__traceback__.tb_lineno}]:"
f" {error}"
)
self.premium_support = None
else:
logger.error(
f"{self.region} --"
f" {error.__class__.__name__}[{error.__traceback__.tb_lineno}]:"
f" {error}"
)
except Exception as error:
logger.error(
@@ -101,3 +101,29 @@ class Test_account_maintain_different_contact_details_to_security_billing_and_op
assert result[0].region == AWS_REGION
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == AWS_ACCOUNT_ARN
def test_access_denied(self):
account_client = mock.MagicMock
account_client.region = AWS_REGION
account_client.audited_account = AWS_ACCOUNT_NUMBER
account_client.audited_account_arn = AWS_ACCOUNT_ARN
account_client.contact_base = None
with mock.patch(
"prowler.providers.aws.services.account.account_service.Account",
new=account_client,
), mock.patch(
"prowler.providers.aws.services.account.account_client.account_client",
new=account_client,
):
# Test Check
from prowler.providers.aws.services.account.account_maintain_different_contact_details_to_security_billing_and_operations.account_maintain_different_contact_details_to_security_billing_and_operations import (
account_maintain_different_contact_details_to_security_billing_and_operations,
)
check = (
account_maintain_different_contact_details_to_security_billing_and_operations()
)
result = check.execute()
assert len(result) == 0
@@ -85,3 +85,28 @@ class Test_backup_vaults_exist:
assert result[0].resource_id == "MyBackupVault"
assert result[0].resource_arn == backup_vault_arn
assert result[0].region == AWS_REGION
def test_access_denied(self):
backup_client = mock.MagicMock
backup_client.audited_account = AWS_ACCOUNT_NUMBER
backup_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
backup_client.region = AWS_REGION
backup_client.audited_partition = "aws"
backup_client.backup_vault_arn_template = f"arn:{backup_client.audited_partition}:backup:{backup_client.region}:{backup_client.audited_account}:backup-vault"
backup_client.__get_backup_vault_arn_template__ = mock.MagicMock(
return_value=backup_client.backup_vault_arn_template
)
backup_client.backup_vaults = None
with mock.patch(
"prowler.providers.aws.services.backup.backup_service.Backup",
new=backup_client,
):
# Test Check
from prowler.providers.aws.services.backup.backup_vaults_exist.backup_vaults_exist import (
backup_vaults_exist,
)
check = backup_vaults_exist()
result = check.execute()
assert len(result) == 0
@@ -254,3 +254,35 @@ class Test_cloudtrail_bucket_requires_mfa_delete:
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_arn == trail_us["TrailARN"]
assert result[0].resource_tags == []
@mock_aws
@patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_getbucketversioning_mfadelete_enabled,
)
def test_access_denied(self):
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_bucket_requires_mfa_delete.cloudtrail_bucket_requires_mfa_delete.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as cloudtrail_client, mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_bucket_requires_mfa_delete.cloudtrail_bucket_requires_mfa_delete.s3_client",
new=S3(aws_provider),
) as s3_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_bucket_requires_mfa_delete.cloudtrail_bucket_requires_mfa_delete import (
cloudtrail_bucket_requires_mfa_delete,
)
cloudtrail_client.trails = None
s3_client.buckets = []
check = cloudtrail_bucket_requires_mfa_delete()
result = check.execute()
assert len(result) == 0
@@ -303,3 +303,34 @@ class Test_cloudtrail_cloudwatch_logging_enabled:
== f"Single region trail {trail_name_eu} is not logging in the last 24h or not configured to deliver logs."
)
assert report.resource_tags == []
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
),
):
with mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_cloudwatch_logging_enabled.cloudtrail_cloudwatch_logging_enabled.cloudtrail_client",
new=Cloudtrail(
set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
)
),
) as service_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_cloudwatch_logging_enabled.cloudtrail_cloudwatch_logging_enabled import (
cloudtrail_cloudwatch_logging_enabled,
)
service_client.trails = None
check = cloudtrail_cloudwatch_logging_enabled()
result = check.execute()
assert len(result) == 0
@@ -128,3 +128,26 @@ class Test_cloudtrail_insights_exist:
assert result[0].region == AWS_REGION_US_EAST_1
assert result[0].resource_arn == trail_us["TrailARN"]
assert result[0].resource_tags == []
@mock_aws
def test_access_denied(self):
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as service_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist import (
cloudtrail_insights_exist,
)
service_client.trails = None
check = cloudtrail_insights_exist()
result = check.execute()
assert len(result) == 0
@@ -137,3 +137,32 @@ class Test_cloudtrail_kms_encryption_enabled:
assert result[0].resource_arn == trail_us["TrailARN"]
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
),
), mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_kms_encryption_enabled.cloudtrail_kms_encryption_enabled.cloudtrail_client",
new=Cloudtrail(
set_mocked_aws_provider([AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1])
),
) as service_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_kms_encryption_enabled.cloudtrail_kms_encryption_enabled import (
cloudtrail_kms_encryption_enabled,
)
service_client.trails = None
check = cloudtrail_kms_encryption_enabled()
result = check.execute()
assert len(result) == 0
@@ -155,3 +155,31 @@ class Test_cloudtrail_log_file_validation_enabled:
assert report.resource_arn == trail_eu["TrailARN"]
assert report.resource_tags == []
assert report.region == AWS_REGION_EU_WEST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
),
), mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_log_file_validation_enabled.cloudtrail_log_file_validation_enabled.cloudtrail_client",
new=Cloudtrail(
set_mocked_aws_provider([AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1])
),
) as service_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_log_file_validation_enabled.cloudtrail_log_file_validation_enabled import (
cloudtrail_log_file_validation_enabled,
)
service_client.trails = None
check = cloudtrail_log_file_validation_enabled()
result = check.execute()
assert len(result) == 0
@@ -226,3 +226,40 @@ class Test_cloudtrail_logs_s3_bucket_access_logging_enabled:
assert result[0].resource_arn == trail_us["TrailARN"]
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
from prowler.providers.aws.services.s3.s3_service import S3
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
),
), mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_access_logging_enabled.cloudtrail_logs_s3_bucket_access_logging_enabled.cloudtrail_client",
new=Cloudtrail(
set_mocked_aws_provider([AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1])
),
) as cloudtrail_client, mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_access_logging_enabled.cloudtrail_logs_s3_bucket_access_logging_enabled.s3_client",
new=S3(
set_mocked_aws_provider([AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1])
),
) as s3_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_access_logging_enabled.cloudtrail_logs_s3_bucket_access_logging_enabled import (
cloudtrail_logs_s3_bucket_access_logging_enabled,
)
cloudtrail_client.trails = None
s3_client.buckets = []
check = cloudtrail_logs_s3_bucket_access_logging_enabled()
result = check.execute()
assert len(result) == 0
@@ -289,3 +289,40 @@ class Test_cloudtrail_logs_s3_bucket_is_not_publicly_accessible:
)
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
from prowler.providers.aws.services.s3.s3_service import S3
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
),
), mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_client",
new=Cloudtrail(
set_mocked_aws_provider([AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1])
),
) as cloudtrail_client, mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.s3_client",
new=S3(
set_mocked_aws_provider([AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1])
),
) as s3_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible import (
cloudtrail_logs_s3_bucket_is_not_publicly_accessible,
)
cloudtrail_client.trails = None
s3_client.buckets = []
check = cloudtrail_logs_s3_bucket_is_not_publicly_accessible()
result = check.execute()
assert len(result) == 0
@@ -289,3 +289,31 @@ class Test_cloudtrail_multi_region_enabled:
assert report.resource_id == trail_name_us
assert report.resource_arn == trail_us["TrailARN"]
assert report.resource_tags == []
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled.cloudtrail_multi_region_enabled.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as service_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled.cloudtrail_multi_region_enabled import (
cloudtrail_multi_region_enabled,
)
service_client.trails = None
check = cloudtrail_multi_region_enabled()
result = check.execute()
assert len(result) == 0
@@ -273,3 +273,29 @@ class Test_cloudtrail_multi_region_enabled_logging_management_events:
result[0].status_extended
== "No trail found with multi-region enabled and logging management events."
)
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled_logging_management_events.cloudtrail_multi_region_enabled_logging_management_events.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as cloudtrail_client:
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled_logging_management_events.cloudtrail_multi_region_enabled_logging_management_events import (
cloudtrail_multi_region_enabled_logging_management_events,
)
cloudtrail_client.trails = None
check = cloudtrail_multi_region_enabled_logging_management_events()
result = check.execute()
assert len(result) == 0
@@ -391,3 +391,33 @@ class Test_cloudtrail_s3_dataevents_read_enabled:
assert result[0].resource_arn == trail_us["TrailARN"]
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
from prowler.providers.aws.services.s3.s3_service import S3
aws_provider = set_mocked_aws_provider()
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_s3_dataevents_read_enabled.cloudtrail_s3_dataevents_read_enabled.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as cloudtrail_client, mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_s3_dataevents_read_enabled.cloudtrail_s3_dataevents_read_enabled.s3_client",
new=S3(aws_provider),
):
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_s3_dataevents_read_enabled.cloudtrail_s3_dataevents_read_enabled import (
cloudtrail_s3_dataevents_read_enabled,
)
cloudtrail_client.trails = None
check = cloudtrail_s3_dataevents_read_enabled()
result = check.execute()
assert len(result) == 0
@@ -389,3 +389,33 @@ class Test_cloudtrail_s3_dataevents_write_enabled:
assert result[0].resource_arn == trail_us["TrailARN"]
assert result[0].resource_tags == []
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
from prowler.providers.aws.services.s3.s3_service import S3
aws_provider = set_mocked_aws_provider()
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_s3_dataevents_write_enabled.cloudtrail_s3_dataevents_write_enabled.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as cloudtrail_service, mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_s3_dataevents_write_enabled.cloudtrail_s3_dataevents_write_enabled.s3_client",
new=S3(aws_provider),
):
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_s3_dataevents_write_enabled.cloudtrail_s3_dataevents_write_enabled import (
cloudtrail_s3_dataevents_write_enabled,
)
cloudtrail_service.trails = None
check = cloudtrail_s3_dataevents_write_enabled()
result = check.execute()
assert len(result) == 0
@@ -561,3 +561,51 @@ class Test_cloudwatch_changes_to_network_acls_alarm_configured:
== f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:metric-filter/test-filter"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
from prowler.providers.aws.services.cloudwatch.cloudwatch_service import (
CloudWatch,
Logs,
)
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
)
from prowler.providers.common.models import Audit_Metadata
aws_provider.audit_metadata = Audit_Metadata(
services_scanned=0,
# We need to set this check to call __describe_log_groups__
expected_checks=["cloudwatch_log_group_no_secrets_in_logs"],
completed_checks=0,
audit_progress=0,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_changes_to_network_acls_alarm_configured.cloudwatch_changes_to_network_acls_alarm_configured.logs_client",
new=Logs(aws_provider),
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_changes_to_network_acls_alarm_configured.cloudwatch_changes_to_network_acls_alarm_configured.cloudwatch_client",
new=CloudWatch(aws_provider),
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_changes_to_network_acls_alarm_configured.cloudwatch_changes_to_network_acls_alarm_configured.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as cloudtrail_client:
# Test Check
from prowler.providers.aws.services.cloudwatch.cloudwatch_changes_to_network_acls_alarm_configured.cloudwatch_changes_to_network_acls_alarm_configured import (
cloudwatch_changes_to_network_acls_alarm_configured,
)
cloudtrail_client.trails = None
check = cloudwatch_changes_to_network_acls_alarm_configured()
result = check.execute()
assert len(result) == 0
@@ -99,3 +99,39 @@ class Test_cloudwatch_cross_account_sharing_disabled:
== "CloudWatch has allowed cross-account sharing."
)
assert result[0].resource_id == "CloudWatch-CrossAccountSharingRole"
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
from prowler.providers.common.models import Audit_Metadata
aws_provider.audit_metadata = Audit_Metadata(
services_scanned=0,
# We need to set this check to call __describe_log_groups__
expected_checks=["cloudwatch_log_group_no_secrets_in_logs"],
completed_checks=0,
audit_progress=0,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_cross_account_sharing_disabled.cloudwatch_cross_account_sharing_disabled.iam_client",
new=IAM(aws_provider),
) as iam_client:
# Test Check
from prowler.providers.aws.services.cloudwatch.cloudwatch_cross_account_sharing_disabled.cloudwatch_cross_account_sharing_disabled import (
cloudwatch_cross_account_sharing_disabled,
)
iam_client.roles = None
check = cloudwatch_cross_account_sharing_disabled()
result = check.execute()
assert len(result) == 0
@@ -138,3 +138,39 @@ class Test_cloudwatch_log_group_kms_encryption_enabled:
== "Log Group test does have AWS KMS key test_kms_id associated."
)
assert result[0].resource_id == "test"
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
from prowler.providers.common.models import Audit_Metadata
aws_provider.audit_metadata = Audit_Metadata(
services_scanned=0,
# We need to set this check to call __describe_log_groups__
expected_checks=["cloudwatch_log_group_no_secrets_in_logs"],
completed_checks=0,
audit_progress=0,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_kms_encryption_enabled.cloudwatch_log_group_kms_encryption_enabled.logs_client",
new=Logs(aws_provider),
) as service_client:
# Test Check
from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_kms_encryption_enabled.cloudwatch_log_group_kms_encryption_enabled import (
cloudwatch_log_group_kms_encryption_enabled,
)
service_client.log_groups = None
check = cloudwatch_log_group_kms_encryption_enabled()
result = check.execute()
assert len(result) == 0
@@ -154,3 +154,39 @@ class Test_cloudwatch_log_group_no_secrets_in_logs:
"Potential secrets found in log group", result[0].status_extended
)
assert result[0].resource_id == "test"
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
from prowler.providers.common.models import Audit_Metadata
aws_provider.audit_metadata = Audit_Metadata(
services_scanned=0,
# We need to set this check to call __describe_log_groups__
expected_checks=["cloudwatch_log_group_no_secrets_in_logs"],
completed_checks=0,
audit_progress=0,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_no_secrets_in_logs.cloudwatch_log_group_no_secrets_in_logs.logs_client",
new=Logs(aws_provider),
) as logs_client:
# Test Check
from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_no_secrets_in_logs.cloudwatch_log_group_no_secrets_in_logs import (
cloudwatch_log_group_no_secrets_in_logs,
)
logs_client.log_groups = None
check = cloudwatch_log_group_no_secrets_in_logs()
result = check.execute()
assert len(result) == 0
@@ -207,3 +207,47 @@ class Test_cloudwatch_log_group_retention_policy_specific_days_enabled:
== f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:log-group:test"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
# Generate Logs Client
logs_client = client("logs", region_name=AWS_REGION_US_EAST_1)
# Request Logs group
logs_client.create_log_group(
logGroupName="test",
)
logs_client.put_retention_policy(logGroupName="test", retentionInDays=7)
from prowler.providers.aws.services.cloudwatch.cloudwatch_service import Logs
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
aws_provider._audit_config = {"log_group_retention_days": 365}
from prowler.providers.common.models import Audit_Metadata
aws_provider.audit_metadata = Audit_Metadata(
services_scanned=0,
# We need to set this check to call __describe_log_groups__
expected_checks=["cloudwatch_log_group_no_secrets_in_logs"],
completed_checks=0,
audit_progress=0,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_retention_policy_specific_days_enabled.cloudwatch_log_group_retention_policy_specific_days_enabled.logs_client",
new=Logs(aws_provider),
) as service_client:
# Test Check
from prowler.providers.aws.services.cloudwatch.cloudwatch_log_group_retention_policy_specific_days_enabled.cloudwatch_log_group_retention_policy_specific_days_enabled import (
cloudwatch_log_group_retention_policy_specific_days_enabled,
)
service_client.log_groups = None
check = cloudwatch_log_group_retention_policy_specific_days_enabled()
result = check.execute()
assert len(result) == 0
@@ -575,3 +575,53 @@ class Test_cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_c
== f"arn:aws:logs:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:metric-filter/test-filter"
)
assert result[0].region == AWS_REGION_US_EAST_1
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
from prowler.providers.aws.services.cloudwatch.cloudwatch_service import (
CloudWatch,
Logs,
)
aws_provider = set_mocked_aws_provider(
[AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1]
)
from prowler.providers.common.models import Audit_Metadata
aws_provider.audit_metadata = Audit_Metadata(
services_scanned=0,
# We need to set this check to call __describe_log_groups__
expected_checks=["cloudwatch_log_group_no_secrets_in_logs"],
completed_checks=0,
audit_progress=0,
)
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled.logs_client",
new=Logs(aws_provider),
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled.cloudwatch_client",
new=CloudWatch(aws_provider),
), mock.patch(
"prowler.providers.aws.services.cloudwatch.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled.cloudtrail_client",
new=Cloudtrail(aws_provider),
) as cloudtrail_client:
# Test Check
from prowler.providers.aws.services.cloudwatch.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled.cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled import (
cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled,
)
check = (
cloudwatch_log_metric_filter_and_alarm_for_aws_config_configuration_changes_enabled()
)
cloudtrail_client.trails = None
result = check.execute()
assert len(result) == 0
@@ -26,6 +26,24 @@ class Test_fms_policy_compliant:
assert len(result) == 0
def test_access_denied(self):
fms_client = mock.MagicMock
fms_client.region = AWS_REGION_US_EAST_1
fms_client.fms_admin_account = None
with mock.patch(
"prowler.providers.aws.services.fms.fms_service.FMS",
new=fms_client,
):
# Test Check
from prowler.providers.aws.services.fms.fms_policy_compliant.fms_policy_compliant import (
fms_policy_compliant,
)
check = fms_policy_compliant()
result = check.execute()
assert len(result) == 0
def test_fms_admin_with_non_compliant_policies(self):
fms_client = mock.MagicMock
fms_client.audited_account = AWS_ACCOUNT_NUMBER
@@ -145,3 +145,25 @@ class Test_iam_password_policy_expires_passwords_within_90_days_or_less:
"Password expiration is set lower than 90 days",
result[0].status_extended,
)
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_password_policy_expires_passwords_within_90_days_or_less.iam_password_policy_expires_passwords_within_90_days_or_less.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_password_policy_expires_passwords_within_90_days_or_less.iam_password_policy_expires_passwords_within_90_days_or_less import (
iam_password_policy_expires_passwords_within_90_days_or_less,
)
service_client.password_policy = None
check = iam_password_policy_expires_passwords_within_90_days_or_less()
result = check.execute()
assert len(result) == 0
@@ -85,3 +85,25 @@ class Test_iam_password_policy_lowercase:
== f"arn:aws:iam:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:password-policy"
)
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_password_policy_lowercase.iam_password_policy_lowercase.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_password_policy_lowercase.iam_password_policy_lowercase import (
iam_password_policy_lowercase,
)
service_client.password_policy = None
check = iam_password_policy_lowercase()
result = check.execute()
assert len(result) == 0
@@ -129,3 +129,25 @@ class Test_iam_password_policy_minimum_length_14:
== f"arn:aws:iam:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:password-policy"
)
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_password_policy_minimum_length_14.iam_password_policy_minimum_length_14.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_password_policy_minimum_length_14.iam_password_policy_minimum_length_14 import (
iam_password_policy_minimum_length_14,
)
service_client.password_policy = None
check = iam_password_policy_minimum_length_14()
result = check.execute()
assert len(result) == 0
@@ -92,3 +92,25 @@ class Test_iam_password_policy_number:
== f"arn:aws:iam:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:password-policy"
)
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_password_policy_number.iam_password_policy_number.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_password_policy_number.iam_password_policy_number import (
iam_password_policy_number,
)
service_client.password_policy = None
check = iam_password_policy_number()
result = check.execute()
assert len(result) == 0
@@ -89,3 +89,25 @@ class Test_iam_password_policy_reuse_24:
== f"arn:aws:iam:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:password-policy"
)
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_password_policy_reuse_24.iam_password_policy_reuse_24.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_password_policy_reuse_24.iam_password_policy_reuse_24 import (
iam_password_policy_reuse_24,
)
service_client.password_policy = None
check = iam_password_policy_reuse_24()
result = check.execute()
assert len(result) == 0
@@ -92,3 +92,25 @@ class Test_iam_password_policy_symbol:
== f"arn:aws:iam:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:password-policy"
)
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_password_policy_symbol.iam_password_policy_symbol.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_password_policy_symbol.iam_password_policy_symbol import (
iam_password_policy_symbol,
)
service_client.password_policy = None
check = iam_password_policy_symbol()
result = check.execute()
assert len(result) == 0
@@ -89,3 +89,25 @@ class Test_iam_password_policy_uppercase:
== f"arn:aws:iam:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:password-policy"
)
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
from prowler.providers.aws.services.iam.iam_service import IAM
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.iam.iam_password_policy_uppercase.iam_password_policy_uppercase.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_password_policy_uppercase.iam_password_policy_uppercase import (
iam_password_policy_uppercase,
)
service_client.password_policy = None
check = iam_password_policy_uppercase()
result = check.execute()
assert len(result) == 0
@@ -264,3 +264,26 @@ class Test_iam_role_administratoraccess_policy:
check = iam_role_administratoraccess_policy()
result = check.execute()
assert len(result) == 0
@mock_aws(config={"iam": {"load_aws_managed_policies": True}})
def test_access_denied(self):
iam_client = mock.MagicMock
iam_client.roles = None
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.iam.iam_role_administratoraccess_policy.iam_role_administratoraccess_policy.iam_client",
new=iam_client,
):
# Test Check
from prowler.providers.aws.services.iam.iam_role_administratoraccess_policy.iam_role_administratoraccess_policy import (
iam_role_administratoraccess_policy,
)
check = iam_role_administratoraccess_policy()
result = check.execute()
assert len(result) == 0
@@ -85,3 +85,25 @@ class Test_iam_securityaudit_role_created:
assert result[0].resource_id == "SecurityAudit"
assert result[0].resource_arn == "arn:aws:iam::aws:policy/SecurityAudit"
assert result[0].region == "us-east-1"
@mock_aws(config={"iam": {"load_aws_managed_policies": True}})
def test_access_denied(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
from prowler.providers.aws.services.iam.iam_service import IAM
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.iam.iam_securityaudit_role_created.iam_securityaudit_role_created.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_securityaudit_role_created.iam_securityaudit_role_created import (
iam_securityaudit_role_created,
)
service_client.entities_role_attached_to_securityaudit_policy = None
check = iam_securityaudit_role_created()
result = check.execute()
assert len(result) == 0
@@ -95,3 +95,25 @@ class Test_iam_support_role_created:
result[0].resource_arn
== "arn:aws:iam::aws:policy/aws-service-role/AWSSupportServiceRolePolicy"
)
@mock_aws(config={"iam": {"load_aws_managed_policies": True}})
def test_access_denied(self):
aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
from prowler.providers.aws.services.iam.iam_service import IAM
with mock.patch(
"prowler.providers.common.common.get_global_provider",
return_value=aws_provider,
), mock.patch(
"prowler.providers.aws.services.iam.iam_support_role_created.iam_support_role_created.iam_client",
new=IAM(aws_provider),
) as service_client:
from prowler.providers.aws.services.iam.iam_support_role_created.iam_support_role_created import (
iam_support_role_created,
)
service_client.entities_role_attached_to_support_policy = None
check = iam_support_role_created()
result = check.execute()
assert len(result) == 0
@@ -80,3 +80,31 @@ class Test_resourceexplorer2_indexes_found:
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == INDEX_ARN
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
resourceexplorer2_client = mock.MagicMock
resourceexplorer2_client.indexes = None
resourceexplorer2_client.audited_account = AWS_ACCOUNT_NUMBER
resourceexplorer2_client.audited_account_arn = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
)
resourceexplorer2_client.audited_partition = "aws"
resourceexplorer2_client.region = AWS_REGION_US_EAST_1
resourceexplorer2_client.index_arn_template = f"arn:{resourceexplorer2_client.audited_partition}:resource-explorer:{resourceexplorer2_client.region}:{resourceexplorer2_client.audited_account}:index"
resourceexplorer2_client.__get_index_arn_template__ = mock.MagicMock(
return_value=resourceexplorer2_client.index_arn_template
)
with mock.patch(
"prowler.providers.aws.services.resourceexplorer2.resourceexplorer2_service.ResourceExplorer2",
new=resourceexplorer2_client,
):
# Test Check
from prowler.providers.aws.services.resourceexplorer2.resourceexplorer2_indexes_found.resourceexplorer2_indexes_found import (
resourceexplorer2_indexes_found,
)
check = resourceexplorer2_indexes_found()
result = check.execute()
# Assertions
assert len(result) == 0
@@ -164,3 +164,30 @@ class Test_ssmincidents_enabled_with_plans:
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == REPLICATION_SET_ARN
assert result[0].region == AWS_REGION_US_EAST_1
def test_access_denied(self):
ssmincidents_client = mock.MagicMock
ssmincidents_client.audited_account = AWS_ACCOUNT_NUMBER
ssmincidents_client.audited_partition = "aws"
ssmincidents_client.audited_account_arn = (
f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
)
ssmincidents_client.region = AWS_REGION_US_EAST_1
ssmincidents_client.replication_set_arn_template = f"arn:{ssmincidents_client.audited_partition}:ssm-incidents:{ssmincidents_client.region}:{ssmincidents_client.audited_account}:replication-set"
ssmincidents_client.__get_replication_set_arn_template__ = mock.MagicMock(
return_value=ssmincidents_client.replication_set_arn_template
)
ssmincidents_client.replication_set = None
with mock.patch(
"prowler.providers.aws.services.ssmincidents.ssmincidents_service.SSMIncidents",
new=ssmincidents_client,
):
# Test Check
from prowler.providers.aws.services.ssmincidents.ssmincidents_enabled_with_plans.ssmincidents_enabled_with_plans import (
ssmincidents_enabled_with_plans,
)
check = ssmincidents_enabled_with_plans()
result = check.execute()
assert len(result) == 0
@@ -144,3 +144,30 @@ class Test_trustedadvisor_errors_and_warnings:
check = trustedadvisor_errors_and_warnings()
result = check.execute()
assert len(result) == 0
def test_access_denied(self):
trustedadvisor_client = mock.MagicMock
trustedadvisor_client.checks = []
trustedadvisor_client.premium_support = None
trustedadvisor_client.audited_account = AWS_ACCOUNT_NUMBER
trustedadvisor_client.audited_account_arn = AWS_ACCOUNT_ARN
trustedadvisor_client.checks.append(
Check(
id=CHECK_NAME,
name=CHECK_NAME,
arn=CHECK_ARN,
region=AWS_REGION_US_EAST_1,
status="not_available",
)
)
with mock.patch(
"prowler.providers.aws.services.trustedadvisor.trustedadvisor_service.TrustedAdvisor",
trustedadvisor_client,
):
from prowler.providers.aws.services.trustedadvisor.trustedadvisor_errors_and_warnings.trustedadvisor_errors_and_warnings import (
trustedadvisor_errors_and_warnings,
)
check = trustedadvisor_errors_and_warnings()
result = check.execute()
assert len(result) == 0
@@ -86,3 +86,30 @@ class Test_trustedadvisor_premium_support_plan_subscribed:
result[0].resource_arn
== f"arn:aws:trusted-advisor:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:account"
)
def test_access_denied(self):
trustedadvisor_client = mock.MagicMock
trustedadvisor_client.checks = []
trustedadvisor_client.premium_support = None
trustedadvisor_client.audited_account = AWS_ACCOUNT_NUMBER
trustedadvisor_client.audited_account_arn = AWS_ACCOUNT_ARN
trustedadvisor_client.audited_partition = "aws"
trustedadvisor_client.region = AWS_REGION_US_EAST_1
# Set verify_premium_support_plans config
trustedadvisor_client.audit_config = {"verify_premium_support_plans": True}
trustedadvisor_client.account_arn_template = f"arn:{trustedadvisor_client.audited_partition}:trusted-advisor:{trustedadvisor_client.region}:{trustedadvisor_client.audited_account}:account"
trustedadvisor_client.__get_account_arn_template__ = mock.MagicMock(
return_value=trustedadvisor_client.account_arn_template
)
with mock.patch(
"prowler.providers.aws.services.trustedadvisor.trustedadvisor_service.TrustedAdvisor",
trustedadvisor_client,
):
from prowler.providers.aws.services.trustedadvisor.trustedadvisor_premium_support_plan_subscribed.trustedadvisor_premium_support_plan_subscribed import (
trustedadvisor_premium_support_plan_subscribed,
)
check = trustedadvisor_premium_support_plan_subscribed()
result = check.execute()
assert len(result) == 0