feat(check): New CloudTrail check cloudtrail_insights_exist (#2184)

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

@@ -0,0 +1,36 @@
+{
+  "Provider": "aws",
+  "CheckID": "cloudtrail_insights_exist",
+  "CheckTitle": "Ensure CloudTrail Insight is enabled",
+  "CheckType": [
+    "Software and Configuration Checks",
+    "Industry and Regulatory Standards",
+    "CIS AWS Foundations Benchmark"
+  ],
+  "ServiceName": "cloudtrail",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:partition:service:region:account-id:resource-id",
+  "Severity": "low",
+  "ResourceType": "AwsCloudTrailTrail",
+  "Description": "Ensure CloudTrail Insight is enabled",
+  "Risk": "CloudTrail Insights provides a powerful way to search and analyze CloudTrail log data using pre-built queries and machine learning algorithms. This can help you to identify potential security threats and suspicious activity in near real-time, such as unauthorized access attempts, policy changes, or resource modifications.",
+  "RelatedUrl": "",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Enable CloudTrail Insight",
+      "Url": "https://docs.aws.amazon.com/awscloudtrail/latest/userguide/logging-insights-events-with-cloudtrail.html"
+    }
+  },
+  "Categories": [
+    "forensics-ready"
+  ],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}

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

@@ -0,0 +1,27 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
+    cloudtrail_client,
+)
+
+
+class cloudtrail_insights_exist(Check):
+    def execute(self):
+        findings = []
+        for trail in cloudtrail_client.trails:
+            if trail.is_logging:
+                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} has not 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"
+                    )
+                findings.append(report)
+        return findings

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

@@ -22,6 +22,7 @@ class Cloudtrail:
         self.trails = []
         self.__threading_call__(self.__get_trails__)
         self.__get_trail_status__()
+        self.__get_insight_selectors__()
         self.__get_event_selectors__()
         self.__list_tags_for_resource__()
 
@@ -69,6 +70,7 @@ class Cloudtrail:
                             kms_key=kms_key_id,
                             log_group_arn=log_group_arn,
                             data_events=[],
+                            has_insight_selectors=trail["HasInsightSelectors"],
                         )
                     )
             if trails_count == 0:
@@ -134,6 +136,37 @@ class Cloudtrail:
                 f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
 
+    def __get_insight_selectors__(self):
+        logger.info("Cloudtrail - Getting trail insihgt selectors...")
+
+        try:
+            for trail in self.trails:
+                for region, client in self.regional_clients.items():
+                    if trail.region == region and trail.name:
+                        insight_selectors = None
+                        trail.has_insight_selectors = None
+                        try:
+                            client_insight_selectors = client.get_insight_selectors(
+                                TrailName=trail.arn
+                            )
+                            insight_selectors = client_insight_selectors.get(
+                                "InsightSelectors"
+                            )
+                        except Exception as error:
+                            logger.error(
+                                f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+                            )
+                            continue
+                        if insight_selectors:
+                            trail.has_insight_selectors = insight_selectors[0].get(
+                                "InsightType"
+                            )
+
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
     def __list_tags_for_resource__(self):
         logger.info("CloudTrail - List Tags...")
         try:
@@ -173,3 +206,4 @@ class Trail(BaseModel):
     log_group_arn: str = None
     data_events: list[Event_Selector] = []
     tags: Optional[list] = []
+    has_insight_selectors: str = None

tests/providers/aws/services/cloudtrail/cloudtrail_insights_exist/cloudtrail_insights_exist_test.py

@@ -0,0 +1,140 @@
+from unittest import mock
+
+from boto3 import client, session
+from moto import mock_cloudtrail, mock_s3
+
+from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
+from prowler.providers.aws.services.cloudtrail.cloudtrail_service import Cloudtrail
+
+AWS_ACCOUNT_NUMBER = 123456789012
+
+
+class Test_cloudtrail_insights_exist:
+    def set_mocked_audit_info(self):
+        audit_info = AWS_Audit_Info(
+            session_config=None,
+            original_session=None,
+            audit_session=session.Session(
+                profile_name=None,
+                botocore_session=None,
+            ),
+            audited_account=AWS_ACCOUNT_NUMBER,
+            audited_user_id=None,
+            audited_partition="aws",
+            audited_identity_arn=None,
+            profile=None,
+            profile_region=None,
+            credentials=None,
+            assumed_role_info=None,
+            audited_regions=["us-east-1", "eu-west-1"],
+            organizations_metadata=None,
+            audit_resources=None,
+        )
+        return audit_info
+
+    @mock_cloudtrail
+    def test_no_trails(self):
+        current_audit_info = self.set_mocked_audit_info()
+
+        with mock.patch(
+            "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
+            new=current_audit_info,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist.cloudtrail_client",
+                new=Cloudtrail(current_audit_info),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist import (
+                    cloudtrail_insights_exist,
+                )
+
+                check = cloudtrail_insights_exist()
+                result = check.execute()
+                assert len(result) == 0
+
+    @mock_cloudtrail
+    @mock_s3
+    def test_trails_with_no_insight_selector(self):
+        current_audit_info = self.set_mocked_audit_info()
+
+        cloudtrail_client_us_east_1 = client("cloudtrail", region_name="us-east-1")
+        s3_client_us_east_1 = client("s3", region_name="us-east-1")
+        trail_name_us = "trail_test_us_with_no_insight_selector"
+        bucket_name_us = "bucket_test_us"
+        s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
+        trail_us = cloudtrail_client_us_east_1.create_trail(
+            Name=trail_name_us, S3BucketName=bucket_name_us, IsMultiRegionTrail=False
+        )
+        cloudtrail_client_us_east_1.start_logging(Name=trail_name_us)
+        cloudtrail_client_us_east_1.get_trail_status(Name=trail_name_us)
+
+        with mock.patch(
+            "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
+            new=current_audit_info,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist.cloudtrail_client",
+                new=Cloudtrail(current_audit_info),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist import (
+                    cloudtrail_insights_exist,
+                )
+
+                check = cloudtrail_insights_exist()
+                result = check.execute()
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"Trail {trail_name_us} has not insight selectors and it is logging"
+                )
+                assert result[0].resource_id == trail_name_us
+                assert result[0].region == "us-east-1"
+                assert result[0].resource_arn == trail_us["TrailARN"]
+
+    @mock_cloudtrail
+    @mock_s3
+    def test_trails_with_insight_selector(self):
+        current_audit_info = self.set_mocked_audit_info()
+
+        cloudtrail_client_us_east_1 = client("cloudtrail", region_name="us-east-1")
+        s3_client_us_east_1 = client("s3", region_name="us-east-1")
+        trail_name_us = "trail_test_us_with_insight_selector"
+        bucket_name_us = "bucket_test_us"
+        s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
+        trail_us = cloudtrail_client_us_east_1.create_trail(
+            Name=trail_name_us, S3BucketName=bucket_name_us, IsMultiRegionTrail=False
+        )
+        cloudtrail_client_us_east_1.start_logging(Name=trail_name_us)
+        cloudtrail_client_us_east_1.get_trail_status(Name=trail_name_us)
+        cloudtrail_client_us_east_1.put_insight_selectors(
+            TrailName=trail_name_us,
+            InsightSelectors=[{"InsightType": "ApiErrorRateInsight"}],
+        )
+
+        with mock.patch(
+            "prowler.providers.aws.lib.audit_info.audit_info.current_audit_info",
+            new=current_audit_info,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist.cloudtrail_client",
+                new=Cloudtrail(current_audit_info),
+            ):
+                # Test Check
+                from prowler.providers.aws.services.cloudtrail.cloudtrail_insights_exist.cloudtrail_insights_exist import (
+                    cloudtrail_insights_exist,
+                )
+
+                check = cloudtrail_insights_exist()
+                result = check.execute()
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"Trail {trail_name_us} has insight selectors and it is logging"
+                )
+                assert result[0].resource_id == trail_name_us
+                assert result[0].region == "us-east-1"
+                assert result[0].resource_arn == trail_us["TrailARN"]