feat(aws): New CloudTrail, DLM, DocumentDB, EC2, Account and Support checks

prowler/config/config.yaml

@@ -61,6 +61,10 @@ aws:
   # MEDIUM
   ecr_repository_vulnerability_minimum_severity: "MEDIUM"
 
+  # AWS Trusted Advisor
+  # trustedadvisor_premium_support_plan_subscribed
+  verify_premium_support_plans: True
+
 # Azure Configuration
 azure:
 

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.metadata.json

@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "account_maintain_different_contact_details_to_security_billing_and_operations",
+  "CheckTitle": "Maintain different contact details to security, billing and operations.",
+  "CheckType": [
+    "IAM"
+  ],
+  "ServiceName": "account",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:access-recorder:region:account-id:recorder/resource-id",
+  "Severity": "medium",
+  "ResourceType": "Other",
+  "Description": "Maintain different contact details to security, billing and operations.",
+  "Risk": "Ensure contact email and telephone details for AWS accounts are current and map to more than one individual in your organization. An AWS account supports a number of contact details; and AWS will use these to contact the account owner if activity judged to be in breach of Acceptable Use Policy. If an AWS account is observed to be behaving in a prohibited or suspicious manner; AWS will attempt to contact the account owner by email and phone using the contact details listed. If this is unsuccessful and the account behavior needs urgent mitigation; proactive measures may be taken; including throttling of traffic between the account exhibiting suspicious behavior and the AWS API endpoints and the Internet. This will result in impaired service to and from the account in question.",
+  "RelatedUrl": "https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact.html",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://docs.bridgecrew.io/docs/iam_18-maintain-contact-details#aws-console",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Using the Billing and Cost Management console complete contact details.",
+      "Url": "https://docs.aws.amazon.com/accounts/latest/reference/manage-acct-update-contact.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

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

@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.account.account_client import account_client
+
+
+class account_maintain_different_contact_details_to_security_billing_and_operations(
+    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
+
+        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]

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

@@ -1,11 +1,100 @@
 ################## Account
+from typing import Optional
+from venv import logger
+
+from botocore.client import ClientError
+from pydantic import BaseModel
+
 from prowler.providers.aws.lib.service.service import AWSService
 
 
 class Account(AWSService):
     def __init__(self, audit_info):
         # Call AWSService's __init__
         super().__init__(__class__.__name__, audit_info)
+        self.number_of_contacts = 4
+        self.contact_base = self.__get_contact_information__()
+        self.contacts_billing = self.__get_alternate_contact__("BILLING")
+        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,
+        }
+
+        # 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,
+        }
+
+    def __get_contact_information__(self):
+        try:
+            primary_account_contact = self.client.get_contact_information()[
+                "ContactInformation"
+            ]
+
+            return Contact(
+                type="PRIMARY",
+                name=primary_account_contact.get("FullName"),
+                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")
+
+    def __get_alternate_contact__(self, contact_type: str):
+        try:
+            account_contact = self.client.get_alternate_contact(
+                AlternateContactType=contact_type
+            )["AlternateContact"]
+
+            return Contact(
+                type=contact_type,
+                email=account_contact.get("EmailAddress"),
+                name=account_contact.get("Name"),
+                phone_number=account_contact.get("PhoneNumber"),
+            )
+
+        except ClientError as error:
+            if (
+                error.response["Error"]["Code"] == "ResourceNotFoundException"
+                and error.response["Error"]["Message"]
+                == "No contact of the inputted alternate contact type found."
+            ):
+                logger.warning(
+                    f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                )
+            return Contact(
+                type=contact_type,
+            )
+
+        except Exception as error:
+            logger.error(
+                f"{self.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+            return Contact(
+                type=contact_type,
+            )
 
 
-### This service don't need boto3 calls
+class Contact(BaseModel):
+    type: str
+    email: Optional[str]
+    name: Optional[str]
+    phone_number: Optional[str]

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

@@ -23,7 +23,7 @@ def execute(self):
                             )
                         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 reamaining trails
+                        # Since there exists a logging trail in that region there is no point in checking the remaining trails
                         # Store the finding and exit the loop
                         findings.append(report)
                         break

prowler/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled_logging_management_events/cloudtrail_multi_region_enabled_logging_management_events.metadata.json

@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "cloudtrail_multi_region_enabled_logging_management_events",
+  "CheckTitle": "Ensure CloudTrail logging management events in All Regions",
+  "CheckType": [
+    "CIS AWS Foundations Benchmark"
+  ],
+  "ServiceName": "cloudtrail",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "low",
+  "ResourceType": "AwsCloudTrailTrail",
+  "Description": "Ensure CloudTrail logging management events in All Regions",
+  "Risk": "AWS CloudTrail enables governance, compliance, operational auditing, and risk auditing of your AWS account. To meet FTR requirements, you must have management events enabled for all AWS accounts and in all regions and aggregate these logs into an Amazon Simple Storage Service (Amazon S3) bucket owned by a separate AWS account.",
+  "RelatedUrl": "https://docs.bridgecrew.io/docs/logging_14",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws cloudtrail update-trail --name <trail_name> --is-multi-region-trail",
+      "NativeIaC": "",
+      "Other": "https://docs.bridgecrew.io/docs/logging_14",
+      "Terraform": "https://docs.bridgecrew.io/docs/logging_14#terraform"
+    },
+    "Recommendation": {
+      "Text": "Enable CloudTrail logging management events in All Regions",
+      "Url": "https://docs.bridgecrew.io/docs/logging_14"
+    }
+  },
+  "Categories": [
+    "forensics-ready"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

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

@@ -0,0 +1,54 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
+    cloudtrail_client,
+)
+
+
+class cloudtrail_multi_region_enabled_logging_management_events(Check):
+    def execute(self):
+        findings = []
+        report = Check_Report_AWS(self.metadata())
+        report.status = "FAIL"
+        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.audited_account_arn
+
+        for trail in cloudtrail_client.trails:
+            if trail.is_logging:
+                if trail.is_multiregion:
+                    for event in trail.data_events:
+                        # Classic event selectors
+                        if not event.is_advanced:
+                            # Check if trail has IncludeManagementEvents and ReadWriteType is All
+                            if (
+                                event.event_selector["ReadWriteType"] == "All"
+                                and event.event_selector["IncludeManagementEvents"]
+                            ):
+                                report.region = trail.region
+                                report.resource_id = trail.name
+                                report.resource_arn = trail.arn
+                                report.resource_tags = trail.tags
+                                report.status = "PASS"
+                                report.status_extended = f"Trail {trail.name} from home region {trail.home_region} is multi-region, is logging and have management events enabled."
+
+                        # Advanced event selectors
+                        elif event.is_advanced:
+                            if event.event_selector.get(
+                                "Name"
+                            ) == "Management events selector" and all(
+                                [
+                                    field["Field"] != "readOnly"
+                                    for field in event.event_selector["FieldSelectors"]
+                                ]
+                            ):
+                                report.region = trail.region
+                                report.resource_id = trail.name
+                                report.resource_arn = trail.arn
+                                report.resource_tags = trail.tags
+                                report.status = "PASS"
+                                report.status_extended = f"Trail {trail.name} from home region {trail.home_region} is multi-region, is logging and have management events enabled."
+        findings.append(report)
+        return findings

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

@@ -15,7 +15,7 @@ def execute(self):
         report.status_extended = "No CloudTrail trails have a data event to record all S3 object-level API operations."
         for trail in cloudtrail_client.trails:
             for data_event in trail.data_events:
-                # classic event selectors
+                # Classic event selectors
                 if not data_event.is_advanced:
                     # Check if trail has a data event for all S3 Buckets for write
                     if (
@@ -37,7 +37,7 @@ def execute(self):
                                 report.resource_tags = trail.tags
                                 report.status = "PASS"
                                 report.status_extended = f"Trail {trail.name} from home region {trail.home_region} has a classic data event selector to record all S3 object-level API operations."
-                # advanced event selectors
+                # Advanced event selectors
                 elif data_event.is_advanced:
                     for field_selector in data_event.event_selector["FieldSelectors"]:
                         if (

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

@@ -93,7 +93,7 @@ def __get_event_selectors__(self):
                 for region, client in self.regional_clients.items():
                     if trail.region == region and trail.name:
                         data_events = client.get_event_selectors(TrailName=trail.arn)
-                        # check if key exists and array associated to that key is not empty
+                        # EventSelectors
                         if (
                             "EventSelectors" in data_events
                             and data_events["EventSelectors"]
@@ -103,7 +103,7 @@ def __get_event_selectors__(self):
                                     is_advanced=False, event_selector=event
                                 )
                                 trail.data_events.append(event_selector)
-                        # check if key exists and array associated to that key is not empty
+                        # AdvancedEventSelectors
                         elif (
                             "AdvancedEventSelectors" in data_events
                             and data_events["AdvancedEventSelectors"]

prowler/providers/aws/services/dlm/dlm_client.py

@@ -0,0 +1,4 @@
+from prowler.providers.aws.lib.audit_info.audit_info import current_audit_info
+from prowler.providers.aws.services.dlm.dlm_service import DLM
+
+dlm_client = DLM(current_audit_info)

prowler/providers/aws/services/dlm/dlm_ebs_snapshot_lifecycle_policy_exists/dlm_ebs_snapshot_lifecycle_policy_exists.metadata.json

@@ -0,0 +1,34 @@
+{
+  "Provider": "aws",
+  "CheckID": "dlm_ebs_snapshot_lifecycle_policy_exists",
+  "CheckTitle": "Ensure EBS Snapshot lifecycle policies are defined.",
+  "CheckType": [
+    "Data Protection"
+  ],
+  "ServiceName": "dlm",
+  "SubServiceName": "ebs",
+  "ResourceIdTemplate": "arn:aws:iam::account-id:resource-id",
+  "Severity": "medium",
+  "ResourceType": "Other",
+  "Description": "Ensure EBS Snapshot lifecycle policies are defined.",
+  "Risk": "With AWS DLM service, you can manage the lifecycle of your EBS volume snapshots. By automating the EBS volume backup management using lifecycle policies, you can protect your EBS data by enforcing a regular backup schedule, retain backups as required by auditors or internal compliance.",
+  "RelatedUrl": "https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/snapshot-lifecycle.html#dlm-elements",
+  "Remediation": {
+    "Code": {
+      "CLI": "aws dlm create-lifecycle-policy --region <region> --execution-role-arn <execution-role-arn> --description <description> --state ENABLED --policy-details file://lifecycle-policy-config.json",
+      "NativeIaC": "",
+      "Other": "https://www.trendmicro.com/cloudoneconformity/knowledge-base/aws/DLM/ebs-snapshot-automation.html",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "To use Amazon Data Lifecycle Manager (DLM) service to manage the lifecycle of your EBS volume snapshots, you have to tag your AWS EBS volumes and create data lifecycle policies via Amazon DLM.",
+      "Url": "https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/snapshot-lifecycle.html#dlm-elements"
+    }
+  },
+  "Categories": [
+    "forensics-ready"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

prowler/providers/aws/services/dlm/dlm_ebs_snapshot_lifecycle_policy_exists/dlm_ebs_snapshot_lifecycle_policy_exists.py

@@ -0,0 +1,24 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.dlm.dlm_client import dlm_client
+from prowler.providers.aws.services.ec2.ec2_client import ec2_client
+
+
+class dlm_ebs_snapshot_lifecycle_policy_exists(Check):
+    def execute(self):
+        findings = []
+        for region in dlm_client.lifecycle_policies:
+            if (
+                region in ec2_client.regions_with_snapshots
+                and ec2_client.regions_with_snapshots[region]
+            ):
+                report = Check_Report_AWS(self.metadata())
+                report.status = "FAIL"
+                report.status_extended = "No EBS Snapshot lifecycle policies found."
+                report.region = region
+                report.resource_id = dlm_client.audited_account
+                report.resource_arn = dlm_client.audited_account_arn
+                if dlm_client.lifecycle_policies[region]:
+                    report.status = "PASS"
+                    report.status_extended = "EBS snapshot lifecycle policies found."
+                findings.append(report)
+        return findings