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

docs/developer-guide/services.md

@@ -175,6 +175,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
 

prowler/lib/outputs/summary_table.py

@@ -40,11 +40,13 @@ def display_summary_table(
  entity_type = "Project ID/s"
  audited_entities = ", ".join(audit_info.project_ids)
 
- 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,
@@ -68,16 +70,17 @@ def display_summary_table(
  ):
  add_service_to_table(findings_table, current)
 
- current["Total"] = 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
 
  current["Total"] += 1
  if finding.status == "PASS":
  pass_count += 1
+ current["Pass"] += 1
  elif finding.status == "FAIL":
  fail_count += 1
  if finding.check_metadata.Severity == "critical":
@@ -157,7 +160,8 @@ 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"])
  findings_table["Status"].append(current["Status"])

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.py

@@ -6,22 +6,26 @@ class account_maintain_different_contact_details_to_security_billing_and_operati
  Check
 ):
  def execute(self):
- report = Check_Report_AWS(self.metadata())
- report.region = account_client.region
- report.resource_id = account_client.audited_account
- report.resource_arn = account_client.audited_account_arn
+ findings = []
+ if account_client.contact_base:
+ report = Check_Report_AWS(self.metadata())
+ report.region = account_client.region
+ report.resource_id = account_client.audited_account
+ report.resource_arn = account_client.audited_account_arn
 
- if (
- len(account_client.contact_phone_numbers)
- == 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
- ):
- report.status = "PASS"
- report.status_extended = "SECURITY, BILLING and OPERATIONS contacts found and they are different between each other and between ROOT contact."
- 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]
+ if (
+ len(account_client.contact_phone_numbers)
+ == 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
+ ):
+ report.status = "PASS"
+ report.status_extended = "SECURITY, BILLING and OPERATIONS contacts found and they are different between each other and between ROOT contact."
+ 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."
+ findings.append(report)
+ return findings

prowler/providers/aws/services/account/account_service.py

@@ -18,28 +18,29 @@ def __init__(self, audit_info):
  self.contacts_security = self.__get_alternate_contact__("SECURITY")
  self.contacts_operations = self.__get_alternate_contact__("OPERATIONS")
 
- # Set of contact phone numbers
- self.contact_phone_numbers = {
- self.contact_base.phone_number,
- self.contacts_billing.phone_number,
- self.contacts_security.phone_number,
- self.contacts_operations.phone_number,
- }
+ if self.contact_base:
+ # Set of contact phone numbers
+ self.contact_phone_numbers = {
+ self.contact_base.phone_number,
+ self.contacts_billing.phone_number,
+ self.contacts_security.phone_number,
+ self.contacts_operations.phone_number,
+ }
 
- # Set of contact names
- self.contact_names = {
- self.contact_base.name,
- self.contacts_billing.name,
- self.contacts_security.name,
- self.contacts_operations.name,
- }
+  # Set of contact names
+  self.contact_names = {
+  self.contact_base.name,
+  self.contacts_billing.name,
+  self.contacts_security.name,
+  self.contacts_operations.name,
+  }
 
- # Set of contact emails
- self.contact_emails = {
- self.contacts_billing.email,
- self.contacts_security.email,
- self.contacts_operations.email,
- }
+  # Set of contact emails
+  self.contact_emails = {
+  self.contacts_billing.email,
+  self.contacts_security.email,
+  self.contacts_operations.email,
+  }
 
  def __get_contact_information__(self):
  try:
@@ -53,10 +54,16 @@ def __get_contact_information__(self):
  phone_number=primary_account_contact.get("PhoneNumber"),
  )
  except Exception as error:
- logger.error(
- f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
- )
- return Contact(type="PRIMARY")
+ 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}"
+ )
+ return Contact(type="PRIMARY")
 
  def __get_alternate_contact__(self, contact_type: str):
  try:

prowler/providers/aws/services/backup/backup_service.py

@@ -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 @@ def __list_backup_vaults__(self, regional_client):
  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 @@ def __list_backup_vaults__(self, regional_client):
  ),
  )
  )
-
+ 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}"

prowler/providers/aws/services/backup/backup_vaults_encrypted/backup_vaults_encrypted.py

@@ -5,24 +5,24 @@
 class backup_vaults_encrypted(Check):
  def execute(self):
  findings = []
-
- for backup_vault in backup_client.backup_vaults:
- # By default we assume that the result is fail
- report = Check_Report_AWS(self.metadata())
- report.status = "FAIL"
- report.status_extended = (
- f"Backup Vault {backup_vault.name} is not encrypted."
- )
- report.resource_arn = backup_vault.arn
- report.resource_id = backup_vault.name
- report.region = backup_vault.region
- # if it is encrypted we only change the status and the status extended
- if backup_vault.encryption:
- report.status = "PASS"
+ 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())
+ report.status = "FAIL"
  report.status_extended = (
- f"Backup Vault {backup_vault.name} is encrypted."
+ f"Backup Vault {backup_vault.name} is not encrypted."
  )
- # then we store the finding
- findings.append(report)
+ report.resource_arn = backup_vault.arn
+ report.resource_id = backup_vault.name
+ report.region = backup_vault.region
+ # if it is encrypted we only change the status and the status extended
+ if backup_vault.encryption:
+ report.status = "PASS"
+ report.status_extended = (
+ f"Backup Vault {backup_vault.name} is encrypted."
+ )
+ # then we store the finding
+ findings.append(report)
 
  return findings

prowler/providers/aws/services/backup/backup_vaults_exist/backup_vaults_exist.py

@@ -5,18 +5,19 @@
 class backup_vaults_exist(Check):
  def execute(self):
  findings = []
- report = Check_Report_AWS(self.metadata())
- report.status = "FAIL"
- report.status_extended = "No Backup Vault exist."
- report.resource_arn = backup_client.backup_vault_arn_template
- report.resource_id = backup_client.audited_account
- report.region = backup_client.region
- if backup_client.backup_vaults:
- report.status = "PASS"
- report.status_extended = f"At least one backup vault exists: {backup_client.backup_vaults[0].name}."
- report.resource_arn = backup_client.backup_vaults[0].arn
- report.resource_id = backup_client.backup_vaults[0].name
- report.region = backup_client.backup_vaults[0].region
+ if backup_client.backup_vaults is not None:
+ report = Check_Report_AWS(self.metadata())
+ report.status = "FAIL"
+ report.status_extended = "No Backup Vault exist."
+ report.resource_arn = backup_client.backup_vault_arn_template
+ report.resource_id = backup_client.audited_account
+ report.region = backup_client.region
+ if backup_client.backup_vaults:
+ report.status = "PASS"
+ report.status_extended = f"At least one backup vault exists: {backup_client.backup_vaults[0].name}."
+ report.resource_arn = backup_client.backup_vaults[0].arn
+ report.resource_id = backup_client.backup_vaults[0].name
+ report.region = backup_client.backup_vaults[0].region
 
- findings.append(report)
+  findings.append(report)
  return findings

prowler/providers/aws/services/cloudtrail/cloudtrail_bucket_requires_mfa_delete/cloudtrail_bucket_requires_mfa_delete.py

@@ -8,28 +8,29 @@
 class cloudtrail_bucket_requires_mfa_delete(Check):
  def execute(self):
  findings = []
- for trail in cloudtrail_client.trails.values():
- if trail.is_logging:
- trail_bucket_is_in_account = False
- trail_bucket = trail.s3_bucket
- report = Check_Report_AWS(self.metadata())
- report.region = trail.region
- report.resource_id = trail.name
- report.resource_arn = trail.arn
- report.resource_tags = trail.tags
- report.status = "FAIL"
- report.status_extended = f"Trail {trail.name} bucket ({trail_bucket}) does not have MFA delete enabled."
- for bucket in s3_client.buckets:
- if trail_bucket == bucket.name:
- trail_bucket_is_in_account = True
- if bucket.mfa_delete:
- report.status = "PASS"
- report.status_extended = f"Trail {trail.name} bucket ({trail_bucket}) has MFA delete enabled."
- # check if trail bucket is a cross account bucket
- if not trail_bucket_is_in_account:
- report.status = "INFO"
- report.status_extended = f"Trail {trail.name} bucket ({trail_bucket}) is a cross-account bucket in another account out of Prowler's permissions scope, please check it manually."
+ if cloudtrail_client.trails is not None:
+ for trail in cloudtrail_client.trails.values():
+ if trail.is_logging:
+ trail_bucket_is_in_account = False
+ trail_bucket = trail.s3_bucket
+ report = Check_Report_AWS(self.metadata())
+ report.region = trail.region
+ report.resource_id = trail.name
+ report.resource_arn = trail.arn
+ report.resource_tags = trail.tags
+ report.status = "FAIL"
+ report.status_extended = f"Trail {trail.name} bucket ({trail_bucket}) does not have MFA delete enabled."
+ for bucket in s3_client.buckets:
+ if trail_bucket == bucket.name:
+ trail_bucket_is_in_account = True
+ if bucket.mfa_delete:
+ report.status = "PASS"
+ report.status_extended = f"Trail {trail.name} bucket ({trail_bucket}) has MFA delete enabled."
+ # check if trail bucket is a cross account bucket
+ if not trail_bucket_is_in_account:
+ report.status = "INFO"
+ report.status_extended = f"Trail {trail.name} bucket ({trail_bucket}) is a cross-account bucket in another account out of Prowler's permissions scope, please check it manually."
 
- findings.append(report)
+  findings.append(report)
 
  return findings

prowler/providers/aws/services/cloudtrail/cloudtrail_cloudwatch_logging_enabled/cloudtrail_cloudwatch_logging_enabled.py

@@ -11,37 +11,38 @@
 class cloudtrail_cloudwatch_logging_enabled(Check):
  def execute(self):
  findings = []
- for trail in cloudtrail_client.trails.values():
- if trail.name:
- report = Check_Report_AWS(self.metadata())
- report.region = trail.region
- report.resource_id = trail.name
- report.resource_arn = trail.arn
- 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."
- )
- else:
- report.status_extended = f"Single region trail {trail.name} has been logging the last 24h."
- if trail.latest_cloudwatch_delivery_time:
- last_log_delivery = (
- datetime.now().replace(tzinfo=timezone.utc)
- - trail.latest_cloudwatch_delivery_time
- )
- if last_log_delivery > timedelta(days=maximum_time_without_logging):
+ if cloudtrail_client.trails is not None:
+ for trail in cloudtrail_client.trails.values():
+ if trail.name:
+ report = Check_Report_AWS(self.metadata())
+ report.region = trail.region
+ report.resource_id = trail.name
+ report.resource_arn = trail.arn
+ 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."
+ else:
+ report.status_extended = f"Single region trail {trail.name} has been logging the last 24h."
+ if trail.latest_cloudwatch_delivery_time:
+ last_log_delivery = (
+ datetime.now().replace(tzinfo=timezone.utc)
+ - trail.latest_cloudwatch_delivery_time
+ )
+ 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."
+ else:
+ report.status_extended = f"Single region trail {trail.name} is not logging in the last 24h."
+ else:
  report.status = "FAIL"
  if trail.is_multiregion:
- report.status_extended = f"Multiregion trail {trail.name} is not logging in the last 24h."
+ report.status_extended = f"Multiregion trail {trail.name} is not logging in the last 24h or not configured to deliver logs."
  else:
- report.status_extended = f"Single region trail {trail.name} is not logging in the last 24h."
- else:
- report.status = "FAIL"
- if trail.is_multiregion:
- report.status_extended = f"Multiregion trail {trail.name} is not logging in the last 24h or not configured to deliver logs."
- else:
- report.status_extended = f"Single region trail {trail.name} is not logging in the last 24h or not configured to deliver logs."
- findings.append(report)
+ report.status_extended = f"Single region trail {trail.name} is not logging in the last 24h or not configured to deliver logs."
+ findings.append(report)
 
  return findings