From bd87351ea705f55eb70b9e1506389a54af145816 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pedro=20Mart=C3=ADn?= Date: Wed, 17 Apr 2024 14:01:39 +0200 Subject: [PATCH] chore(aws): Add CloudTrail Threat Detection tests (#3804) Co-authored-by: Pepe Fagoaga --- ...cloudtrail_threat_detection_enumeration.py | 27 ++- ...l_threat_detection_privilege_escalation.py | 28 +-- .../cloudtrail/cloudtrail_service_test.py | 51 +++++ ...trail_threat_detection_enumeration_test.py | 213 ++++++++++++++++++ ...eat_detection_privilege_escalation_test.py | 213 ++++++++++++++++++ 5 files changed, 504 insertions(+), 28 deletions(-) create mode 100644 tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_test.py create mode 100644 tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_test.py diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.py index 3bf7a75297..2f7cb8db8d 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.py +++ b/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 diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.py index a41eb8e6b9..7cea5430ee 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.py +++ b/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 diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py index ed35eaf137..caf8bb7ee4 100644 --- a/tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py +++ b/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}] diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_test.py new file mode 100644 index 0000000000..7265f426a7 --- /dev/null +++ b/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" + ) diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_test.py new file mode 100644 index 0000000000..dbfca19d42 --- /dev/null +++ b/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" + )