chore(aws): Add CloudTrail Threat Detection tests (#3804)

Co-authored-by: Pepe Fagoaga <pepe@prowler.com>

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

@@ -5,20 +5,19 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
 )
 
-THRESHOLD = cloudtrail_client.audit_config.get(
-    "threat_detection_enumeration_threshold", 0.1
-)
-THREAT_DETECTION_MINUTES = cloudtrail_client.audit_config.get(
-    "threat_detection_enumeration_minutes", 1440
-)
-ENUMERATION_ACTIONS = cloudtrail_client.audit_config.get(
-    "threat_detection_enumeration_actions", []
-)
-
 
 class cloudtrail_threat_detection_enumeration(Check):
     def execute(self):
         findings = []
+        threshold = cloudtrail_client.audit_config.get(
+            "threat_detection_enumeration_threshold", 0.1
+        )
+        threat_detection_minutes = cloudtrail_client.audit_config.get(
+            "threat_detection_enumeration_minutes", 1440
+        )
+        enumeration_actions = cloudtrail_client.audit_config.get(
+            "threat_detection_enumeration_actions", []
+        )
         potential_enumeration = {}
         found_potential_enumeration = False
         multiregion_trail = None
@@ -33,11 +32,11 @@ class cloudtrail_threat_detection_enumeration(Check):
             else [multiregion_trail]
         )
         for trail in trails_to_scan:
-            for event_name in ENUMERATION_ACTIONS:
+            for event_name in enumeration_actions:
                 for event_log in cloudtrail_client.__lookup_events__(
                     trail=trail,
                     event_name=event_name,
-                    minutes=THREAT_DETECTION_MINUTES,
+                    minutes=threat_detection_minutes,
                 ):
                     event_log = json.loads(event_log["CloudTrailEvent"])
                     if ".amazonaws.com" not in event_log["sourceIPAddress"]:
@@ -47,8 +46,8 @@ class cloudtrail_threat_detection_enumeration(Check):
                             event_name
                         )
         for source_ip, actions in potential_enumeration.items():
-            ip_threshold = round(len(actions) / len(ENUMERATION_ACTIONS), 2)
-            if len(actions) / len(ENUMERATION_ACTIONS) > THRESHOLD:
+            ip_threshold = round(len(actions) / len(enumeration_actions), 2)
+            if len(actions) / len(enumeration_actions) > threshold:
                 found_potential_enumeration = True
                 report = Check_Report_AWS(self.metadata())
                 report.region = cloudtrail_client.region

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

@@ -5,20 +5,20 @@ from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
     cloudtrail_client,
 )
 
-THRESHOLD = cloudtrail_client.audit_config.get(
-    "threat_detection_privilege_escalation_threshold", 0.1
-)
-THREAT_DETECTION_MINUTES = cloudtrail_client.audit_config.get(
-    "threat_detection_privilege_escalation_minutes", 1440
-)
-PRIVILEGE_ESCALATION_ACTIONS = cloudtrail_client.audit_config.get(
-    "threat_detection_privilege_escalation_actions", []
-)
-
 
 class cloudtrail_threat_detection_privilege_escalation(Check):
     def execute(self):
         findings = []
+        threshold = cloudtrail_client.audit_config.get(
+            "threat_detection_privilege_escalation_threshold", 0.1
+        )
+        threat_detection_minutes = cloudtrail_client.audit_config.get(
+            "threat_detection_privilege_escalation_minutes", 1440
+        )
+        privilege_escalation_actions = cloudtrail_client.audit_config.get(
+            "threat_detection_privilege_escalation_actions", []
+        )
+
         potential_privilege_escalation = {}
         found_potential_privilege_escalation = False
         multiregion_trail = None
@@ -33,11 +33,11 @@ class cloudtrail_threat_detection_privilege_escalation(Check):
             else [multiregion_trail]
         )
         for trail in trails_to_scan:
-            for event_name in PRIVILEGE_ESCALATION_ACTIONS:
+            for event_name in privilege_escalation_actions:
                 for event_log in cloudtrail_client.__lookup_events__(
                     trail=trail,
                     event_name=event_name,
-                    minutes=THREAT_DETECTION_MINUTES,
+                    minutes=threat_detection_minutes,
                 ):
                     event_log = json.loads(event_log["CloudTrailEvent"])
                     if ".amazonaws.com" not in event_log["sourceIPAddress"]:
@@ -52,8 +52,8 @@ class cloudtrail_threat_detection_privilege_escalation(Check):
                             event_log["sourceIPAddress"]
                         ].add(event_name)
         for source_ip, actions in potential_privilege_escalation.items():
-            ip_threshold = round(len(actions) / len(PRIVILEGE_ESCALATION_ACTIONS), 2)
-            if len(actions) / len(PRIVILEGE_ESCALATION_ACTIONS) > THRESHOLD:
+            ip_threshold = round(len(actions) / len(privilege_escalation_actions), 2)
+            if len(actions) / len(privilege_escalation_actions) > threshold:
                 found_potential_privilege_escalation = True
                 report = Check_Report_AWS(self.metadata())
                 report.region = cloudtrail_client.region

tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py

@@ -251,3 +251,54 @@ class Test_Cloudtrail_Service:
                         trail.data_events[0].event_selector == data_events_response[0]
                     )
                     assert trail.data_events[0].is_advanced
+
+    @mock_aws
+    def test__lookup_events__(self):
+        cloudtrail_client_us_east_1 = client(
+            "cloudtrail", region_name=AWS_REGION_US_EAST_1
+        )
+        s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1)
+        trail_name_us = "trail_test_us"
+        bucket_name_us = "bucket_test_us"
+        s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
+        cloudtrail_client_us_east_1.create_trail(
+            Name=trail_name_us,
+            S3BucketName=bucket_name_us,
+            IsMultiRegionTrail=False,
+            EnableLogFileValidation=True,
+        )
+        cloudtrail_client_us_east_1.start_logging(Name=trail_name_us)
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
+        )
+        cloudtrail = Cloudtrail(aws_provider)
+        assert len(cloudtrail.trails) == len(aws_provider.identity.audited_regions)
+
+    @mock_aws
+    def test__list_tags_for_resource__(self):
+        tag = "test-tag"
+        cloudtrail_client_us_east_1 = client(
+            "cloudtrail", region_name=AWS_REGION_US_EAST_1
+        )
+        s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1)
+        trail_name_us = "trail_test_us"
+        bucket_name_us = "bucket_test_us"
+        s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
+        cloudtrail_client_us_east_1.create_trail(
+            Name=trail_name_us,
+            S3BucketName=bucket_name_us,
+            IsMultiRegionTrail=False,
+            EnableLogFileValidation=True,
+            TagsList=[
+                {"Key": "test", "Value": tag},
+            ],
+        )
+        aws_provider = set_mocked_aws_provider(
+            [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
+        )
+        cloudtrail = Cloudtrail(aws_provider)
+        assert len(cloudtrail.trails) == len(aws_provider.identity.audited_regions)
+        for trail in cloudtrail.trails.values():
+            if trail.name:
+                if trail.name == trail_name_us:
+                    assert trail.tags == [{"Key": "test", "Value": tag}]

tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_test.py

@@ -0,0 +1,213 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+def mock__get_trail_arn_template__(region=None, *_) -> str:
+    if region:
+        return f"arn:aws:cloudtrail:{region}:{AWS_ACCOUNT_NUMBER}:trail"
+    else:
+        return f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+
+
+def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> list:
+    return [
+        {
+            "CloudTrailEvent": '{"sourceIPAddress": "172.28.7.0/24", "eventName": "DescribeAccessEntry"}'
+        },
+        {
+            "CloudTrailEvent": '{"sourceIPAddress": "172.28.7.0/24", "eventName": "DescribeAccountAttributes"}'
+        },
+    ]
+
+
+class Test_cloudtrail_threat_detection_enumeration:
+    @mock_aws
+    def test_no_trails(self):
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {}
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration import (
+                cloudtrail_threat_detection_enumeration,
+            )
+
+            check = cloudtrail_threat_detection_enumeration()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended == "No potential enumeration attack detected."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )
+
+    @mock_aws
+    def test_no_potential_enumeration(self):
+        ENUMERATION_ACTIONS = []
+        THRESHOLD = 0.1
+        THREAT_DETECTION_MINUTES = 1440
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {"us-east-1": mock.MagicMock()}
+        cloudtrail_client.trails["us-east-1"].is_multiregion = False
+        cloudtrail_client.trails["us-east-1"].name = "trail_test_us"
+        cloudtrail_client.trails["us-east-1"].s3_bucket_name = "bucket_test_us"
+        cloudtrail_client.trails["us-east-1"].region = "us-east-1"
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+        cloudtrail_client.audit_config = {
+            "threat_detection_enumeration_actions": ENUMERATION_ACTIONS,
+            "threat_detection_enumeration_threshold": THRESHOLD,
+            "threat_detection_enumeration_minutes": THREAT_DETECTION_MINUTES,
+        }
+
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration import (
+                cloudtrail_threat_detection_enumeration,
+            )
+
+            check = cloudtrail_threat_detection_enumeration()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended == "No potential enumeration attack detected."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )
+
+    @mock_aws
+    def test_potential_enumeration(self):
+        ENUMERATION_ACTIONS = ["DescribeAccessEntry", "DescribeAccountAttributes"]
+        THRESHOLD = 0.1
+        THREAT_DETECTION_MINUTES = 1440
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {"us-east-1": mock.MagicMock()}
+        cloudtrail_client.trails["us-east-1"].is_multiregion = False
+        cloudtrail_client.trails["us-east-1"].name = "trail_test_us"
+        cloudtrail_client.trails["us-east-1"].s3_bucket_name = "bucket_test_us"
+        cloudtrail_client.trails["us-east-1"].region = "us-east-1"
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+        cloudtrail_client.audit_config = {
+            "threat_detection_enumeration_actions": ENUMERATION_ACTIONS,
+            "threat_detection_enumeration_threshold": THRESHOLD,
+            "threat_detection_enumeration_minutes": THREAT_DETECTION_MINUTES,
+        }
+
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration import (
+                cloudtrail_threat_detection_enumeration,
+            )
+
+            check = cloudtrail_threat_detection_enumeration()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Potential enumeration attack detected from source IP 172.28.7.0/24 with an threshold of 1.0."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )
+
+    @mock_aws
+    def test_big_threshold(self):
+        ENUMERATION_ACTIONS = ["DescribeAccessEntry", "DescribeAccountAttributes"]
+        THRESHOLD = 2.0
+        THREAT_DETECTION_MINUTES = 1440
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {"us-east-1": mock.MagicMock()}
+        cloudtrail_client.trails["us-east-1"].is_multiregion = False
+        cloudtrail_client.trails["us-east-1"].name = "trail_test_us"
+        cloudtrail_client.trails["us-east-1"].s3_bucket_name = "bucket_test_us"
+        cloudtrail_client.trails["us-east-1"].region = "us-east-1"
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+        cloudtrail_client.audit_config = {
+            "threat_detection_enumeration_actions": ENUMERATION_ACTIONS,
+            "threat_detection_enumeration_threshold": THRESHOLD,
+            "threat_detection_enumeration_minutes": THREAT_DETECTION_MINUTES,
+        }
+
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration import (
+                cloudtrail_threat_detection_enumeration,
+            )
+
+            check = cloudtrail_threat_detection_enumeration()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended == "No potential enumeration attack detected."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )

tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_test.py

@@ -0,0 +1,213 @@
+from unittest import mock
+
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+def mock__get_trail_arn_template__(region=None, *_) -> str:
+    if region:
+        return f"arn:aws:cloudtrail:{region}:{AWS_ACCOUNT_NUMBER}:trail"
+    else:
+        return f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+
+
+def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> list:
+    return [
+        {
+            "CloudTrailEvent": '{"sourceIPAddress": "172.28.7.0/24", "eventName": "CreateLoginProfile"}'
+        },
+        {
+            "CloudTrailEvent": '{"sourceIPAddress": "172.28.7.0/24", "eventName": "UpdateLoginProfile"}'
+        },
+    ]
+
+
+class Test_cloudtrail_threat_detection_privilege_escalation:
+    @mock_aws
+    def test_no_trails(self):
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {}
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation import (
+                cloudtrail_threat_detection_privilege_escalation,
+            )
+
+            check = cloudtrail_threat_detection_privilege_escalation()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "No potential privilege escalation attack detected."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )
+
+    @mock_aws
+    def test_no_potential_priviledge_escalation(self):
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {"us-east-1": mock.MagicMock()}
+        cloudtrail_client.trails["us-east-1"].is_multiregion = False
+        cloudtrail_client.trails["us-east-1"].name = "trail_test_us"
+        cloudtrail_client.trails["us-east-1"].s3_bucket_name = "bucket_test_us"
+        cloudtrail_client.trails["us-east-1"].region = "us-east-1"
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+        cloudtrail_client.audit_config = {
+            "threat_detection_privilege_escalation_actions": [],
+            "threat_detection_privilege_escalation_threshold": 0.1,
+            "threat_detection_privilege_escalation_minutes": 1440,
+        }
+
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation import (
+                cloudtrail_threat_detection_privilege_escalation,
+            )
+
+            check = cloudtrail_threat_detection_privilege_escalation()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "No potential privilege escalation attack detected."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )
+
+    @mock_aws
+    def test_potential_priviledge_escalation(self):
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {"us-east-1": mock.MagicMock()}
+        cloudtrail_client.trails["us-east-1"].is_multiregion = False
+        cloudtrail_client.trails["us-east-1"].name = "trail_test_us"
+        cloudtrail_client.trails["us-east-1"].s3_bucket_name = "bucket_test_us"
+        cloudtrail_client.trails["us-east-1"].region = "us-east-1"
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+        cloudtrail_client.audit_config = {
+            "threat_detection_privilege_escalation_actions": [
+                "CreateLoginProfile",
+                "UpdateLoginProfile",
+            ],
+            "threat_detection_privilege_escalation_threshold": 0.1,
+            "threat_detection_privilege_escalation_minutes": 1440,
+        }
+
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation import (
+                cloudtrail_threat_detection_privilege_escalation,
+            )
+
+            check = cloudtrail_threat_detection_privilege_escalation()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "FAIL"
+            assert (
+                result[0].status_extended
+                == "Potential privilege escalation attack detected from source IP 172.28.7.0/24 with an threshold of 1.0."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )
+
+    @mock_aws
+    def test_bigger_threshold(self):
+        cloudtrail_client = mock.MagicMock()
+        cloudtrail_client.trails = {"us-east-1": mock.MagicMock()}
+        cloudtrail_client.trails["us-east-1"].is_multiregion = False
+        cloudtrail_client.trails["us-east-1"].name = "trail_test_us"
+        cloudtrail_client.trails["us-east-1"].s3_bucket_name = "bucket_test_us"
+        cloudtrail_client.trails["us-east-1"].region = "us-east-1"
+        cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER
+        cloudtrail_client.region = AWS_REGION_US_EAST_1
+        cloudtrail_client.audit_config = {
+            "threat_detection_privilege_escalation_actions": [
+                "CreateLoginProfile",
+                "UpdateLoginProfile",
+            ],
+            "threat_detection_privilege_escalation_threshold": 2.0,
+            "threat_detection_privilege_escalation_minutes": 1440,
+        }
+
+        cloudtrail_client.__lookup_events__ = mock__get_lookup_events__
+        cloudtrail_client.__get_trail_arn_template__ = mock__get_trail_arn_template__
+
+        with mock.patch(
+            "prowler.providers.common.common.get_global_provider",
+            return_value=set_mocked_aws_provider(),
+        ), mock.patch(
+            "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation.cloudtrail_client",
+            new=cloudtrail_client,
+        ):
+            # Test Check
+            from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_privilege_escalation.cloudtrail_threat_detection_privilege_escalation import (
+                cloudtrail_threat_detection_privilege_escalation,
+            )
+
+            check = cloudtrail_threat_detection_privilege_escalation()
+            result = check.execute()
+
+            assert len(result) == 1
+            assert result[0].status == "PASS"
+            assert (
+                result[0].status_extended
+                == "No potential privilege escalation attack detected."
+            )
+            assert result[0].resource_id == AWS_ACCOUNT_NUMBER
+            assert result[0].region == AWS_REGION_US_EAST_1
+            assert (
+                result[0].resource_arn
+                == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail"
+            )