diff --git a/prowler/changelog.d/cloudtrail-threat-detection-resource-identity.fixed.md b/prowler/changelog.d/cloudtrail-threat-detection-resource-identity.fixed.md new file mode 100644 index 0000000000..659f0afa66 --- /dev/null +++ b/prowler/changelog.d/cloudtrail-threat-detection-resource-identity.fixed.md @@ -0,0 +1 @@ +`cloudtrail_threat_detection_enumeration`, `cloudtrail_threat_detection_privilege_escalation`, and `cloudtrail_threat_detection_llm_jacking` now emit stable principal resource identities with consistent metadata diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_service.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_service.py index d597e362d6..886d8f6c94 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_service.py +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_service.py @@ -1,5 +1,6 @@ +import json from datetime import datetime, timedelta -from typing import Optional +from typing import Any, Optional from botocore.client import ClientError from pydantic.v1 import BaseModel @@ -294,3 +295,84 @@ class Trail(BaseModel): data_events: list[Event_Selector] = [] tags: Optional[list] = [] has_insight_selectors: str = None + + +class CloudTrailThreatDetectionResource(BaseModel): + id: str + name: str + arn: str + region: str + identity_type: str + source_arn: str + + +def normalize_cloudtrail_identity( + user_identity: dict, region: str +) -> Optional[CloudTrailThreatDetectionResource]: + source_arn = user_identity.get("arn") + if not source_arn: + return None + + identity_type = user_identity.get("type", "Unknown") + identity_arn = source_arn + if identity_type == "AssumedRole": + identity_arn = ( + user_identity.get("sessionContext", {}).get("sessionIssuer", {}).get("arn") + or source_arn + ) + + resource_component = identity_arn.split(":", 5)[-1] + name = resource_component.rsplit("/", 1)[-1] + if identity_type == "AssumedRole" and identity_arn == source_arn: + name = resource_component.removeprefix("assumed-role/").split("/", 1)[0] + + return CloudTrailThreatDetectionResource( + id=resource_component, + name=name, + arn=identity_arn, + region=region, + identity_type=identity_type, + source_arn=source_arn, + ) + + +def get_cloudtrail_threat_detection_identities( + cloudtrail_client: Any, actions: list[str], minutes: int +) -> dict[str, tuple[CloudTrailThreatDetectionResource, set[str]]]: + identities = {} + multiregion_trail = next( + (trail for trail in cloudtrail_client.trails.values() if trail.is_multiregion), + None, + ) + trails_to_scan = ( + [multiregion_trail] if multiregion_trail else cloudtrail_client.trails.values() + ) + + for trail in trails_to_scan: + for action in actions: + for event_log in cloudtrail_client._lookup_events( + trail=trail, event_name=action, minutes=minutes + ): + event = json.loads(event_log["CloudTrailEvent"]) + resource = normalize_cloudtrail_identity( + event.get("userIdentity", {}), cloudtrail_client.region + ) + if resource: + identities.setdefault(resource.arn, (resource, set()))[1].add( + action + ) + + return identities + + +def get_cloudtrail_account_resource( + account_id: str, account_arn: str, region: str +) -> CloudTrailThreatDetectionResource: + return CloudTrailThreatDetectionResource( + id=account_id, + name=account_id, + arn=account_arn, + region=region, + identity_type="AWSAccount", + source_arn=account_arn, + ) diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.metadata.json b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.metadata.json index 61d1cc12a5..1cbb2d992e 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.metadata.json +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.metadata.json @@ -11,7 +11,7 @@ "SubServiceName": "", "ResourceIdTemplate": "", "Severity": "critical", - "ResourceType": "AwsCloudTrailTrail", + "ResourceType": "Other", "ResourceGroup": "monitoring", "Description": "**CloudTrail activity** is analyzed for AWS identities executing a broad mix of discovery APIs like `List*`, `Describe*`, and `Get*` within a recent time window.\n\nAn identity exceeding a configurable ratio of these actions indicates potential enumeration behavior by that principal.", "Risk": "Concentrated discovery activity signals **reconnaissance** with valid credentials. Adversaries can map assets and policies to enable **privilege escalation**, target data stores for **exfiltration** (confidentiality), and identify services to disrupt (availability), supporting stealthy lateral movement.", 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 6b6af71f02..f2fda54c3d 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 @@ -1,9 +1,11 @@ -import json - from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.cloudtrail.cloudtrail_client import ( cloudtrail_client, ) +from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + get_cloudtrail_account_resource, + get_cloudtrail_threat_detection_identities, +) default_threat_detection_enumeration_actions = [ "CreateIndex", @@ -114,71 +116,26 @@ class cloudtrail_threat_detection_enumeration(Check): "threat_detection_enumeration_actions", default_threat_detection_enumeration_actions, ) - potential_enumeration = {} found_potential_enumeration = False - multiregion_trail = None - # Check if any trail is multi-region so we only need to check once - for trail in cloudtrail_client.trails.values(): - if trail.is_multiregion: - multiregion_trail = trail - break - trails_to_scan = ( - cloudtrail_client.trails.values() - if not multiregion_trail - else [multiregion_trail] + potential_enumeration = get_cloudtrail_threat_detection_identities( + cloudtrail_client, enumeration_actions, threat_detection_minutes ) - for trail in trails_to_scan: - for event_name in enumeration_actions: - for event_log in cloudtrail_client._lookup_events( - trail=trail, - event_name=event_name, - minutes=threat_detection_minutes, - ): - event_log = json.loads(event_log["CloudTrailEvent"]) - if ( - "arn" in event_log["userIdentity"] - ): # Ignore event logs without ARN since they are AWS services - if ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) not in potential_enumeration: - potential_enumeration[ - ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) - ] = set() - potential_enumeration[ - ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) - ].add(event_name) - for aws_identity, actions in potential_enumeration.items(): + for resource, actions in potential_enumeration.values(): identity_threshold = round(len(actions) / len(enumeration_actions), 2) - aws_identity_type = aws_identity[1] - aws_identity_arn = aws_identity[0] if len(actions) / len(enumeration_actions) > threshold: found_potential_enumeration = True - report = Check_Report_AWS( - metadata=self.metadata(), resource=cloudtrail_client.trails - ) - report.region = cloudtrail_client.region - report.resource_id = aws_identity_arn.split("/")[-1] - report.resource_arn = aws_identity_arn + report = Check_Report_AWS(metadata=self.metadata(), resource=resource) report.status = "FAIL" - report.status_extended = f"Potential enumeration attack detected from AWS {aws_identity_type} {aws_identity_arn.split('/')[-1]} with a threshold of {identity_threshold}." + report.status_extended = f"Potential enumeration attack detected from AWS {resource.identity_type} {resource.name} with a threshold of {identity_threshold}." findings.append(report) if not found_potential_enumeration: - report = Check_Report_AWS( - metadata=self.metadata(), resource=cloudtrail_client.trails - ) - report.region = cloudtrail_client.region - report.resource_id = cloudtrail_client.audited_account - report.resource_arn = cloudtrail_client._get_trail_arn_template( - cloudtrail_client.region + resource = get_cloudtrail_account_resource( + cloudtrail_client.audited_account, + cloudtrail_client.audited_account_arn, + cloudtrail_client.region, ) + report = Check_Report_AWS(metadata=self.metadata(), resource=resource) report.status = "PASS" report.status_extended = "No potential enumeration attack detected." findings.append(report) diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.metadata.json b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.metadata.json index d430137462..6932d39740 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.metadata.json +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.metadata.json @@ -14,7 +14,7 @@ "SubServiceName": "", "ResourceIdTemplate": "", "Severity": "critical", - "ResourceType": "AwsCloudTrailTrail", + "ResourceType": "Other", "ResourceGroup": "monitoring", "Description": "**CloudTrail Bedrock activity** is analyzed per identity for a high diversity of LLM-related API calls (e.g., `InvokeModel`, `InvokeModelWithResponseStream`, `GetFoundationModelAvailability`). *If an identity's share of these actions exceeds a configured threshold over a recent window*, it is surfaced as potential **LLM-jacking** behavior.", "Risk": "Such patterns suggest **stolen credential** abuse to drive LLM usage.\n- Availability: cost exhaustion and service disruption\n- Confidentiality: leakage of prompts/outputs and model settings\n- Integrity: misuse of permissions for broader access\nAttackers may use reverse proxies to resell access and obfuscate sources.", diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.py index ea8b2a24d1..9949f7cc4f 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.py +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.py @@ -1,9 +1,11 @@ -import json - from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.cloudtrail.cloudtrail_client import ( cloudtrail_client, ) +from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + get_cloudtrail_account_resource, + get_cloudtrail_threat_detection_identities, +) default_threat_detection_llm_jacking_actions = [ "PutUseCaseForModelAccess", @@ -36,71 +38,26 @@ class cloudtrail_threat_detection_llm_jacking(Check): "threat_detection_llm_jacking_actions", default_threat_detection_llm_jacking_actions, ) - potential_llm_jacking = {} found_potential_llm_jacking = False - multiregion_trail = None - # Check if any trail is multi-region so we only need to check once - for trail in cloudtrail_client.trails.values(): - if trail.is_multiregion: - multiregion_trail = trail - break - trails_to_scan = ( - cloudtrail_client.trails.values() - if not multiregion_trail - else [multiregion_trail] + potential_llm_jacking = get_cloudtrail_threat_detection_identities( + cloudtrail_client, llm_jacking_actions, threat_detection_minutes ) - for trail in trails_to_scan: - for event_name in llm_jacking_actions: - for event_log in cloudtrail_client._lookup_events( - trail=trail, - event_name=event_name, - minutes=threat_detection_minutes, - ): - event_log = json.loads(event_log["CloudTrailEvent"]) - if ( - "arn" in event_log["userIdentity"] - ): # Ignore event logs without ARN since they are AWS services - if ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) not in potential_llm_jacking: - potential_llm_jacking[ - ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) - ] = set() - potential_llm_jacking[ - ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) - ].add(event_name) - for aws_identity, actions in potential_llm_jacking.items(): + for resource, actions in potential_llm_jacking.values(): identity_threshold = round(len(actions) / len(llm_jacking_actions), 2) - aws_identity_type = aws_identity[1] - aws_identity_arn = aws_identity[0] if len(actions) / len(llm_jacking_actions) > threshold: found_potential_llm_jacking = True - report = Check_Report_AWS( - metadata=self.metadata(), resource=cloudtrail_client.trails - ) - report.region = cloudtrail_client.region - report.resource_id = aws_identity_arn.split("/")[-1] - report.resource_arn = aws_identity_arn + report = Check_Report_AWS(metadata=self.metadata(), resource=resource) report.status = "FAIL" - report.status_extended = f"Potential LLM Jacking attack detected from AWS {aws_identity_type} {aws_identity_arn.split('/')[-1]} with a threshold of {identity_threshold}." + report.status_extended = f"Potential LLM Jacking attack detected from AWS {resource.identity_type} {resource.name} with a threshold of {identity_threshold}." findings.append(report) if not found_potential_llm_jacking: - report = Check_Report_AWS( - metadata=self.metadata(), resource=cloudtrail_client.trails - ) - report.region = cloudtrail_client.region - report.resource_id = cloudtrail_client.audited_account - report.resource_arn = cloudtrail_client._get_trail_arn_template( - cloudtrail_client.region + resource = get_cloudtrail_account_resource( + cloudtrail_client.audited_account, + cloudtrail_client.audited_account_arn, + cloudtrail_client.region, ) + report = Check_Report_AWS(metadata=self.metadata(), resource=resource) report.status = "PASS" report.status_extended = "No potential LLM Jacking attack detected." findings.append(report) diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.metadata.json b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.metadata.json index 284dde4895..de0b8b0802 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.metadata.json +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.metadata.json @@ -10,7 +10,7 @@ "SubServiceName": "", "ResourceIdTemplate": "", "Severity": "critical", - "ResourceType": "AwsCloudTrailTrail", + "ResourceType": "Other", "ResourceGroup": "monitoring", "Description": "**CloudTrail** activity is analyzed for **identities** executing high-risk actions linked to **privilege escalation** (e.g., `Attach*Policy`, `PassRole`, `AssumeRole`, `CreateAccessKey`). Identities exceeding a configurable share of such events within a *recent time window* are highlighted for investigation.", "Risk": "Escalation patterns can grant elevated entitlements, enabling:\n- Confidentiality loss via unauthorized data/secret access\n- Integrity compromise by changing IAM policies/roles\n- Availability impact by tampering with logging or resources\nThis also facilitates lateral movement and persistence.", 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 3e6158df97..9416a492f5 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 @@ -1,9 +1,11 @@ -import json - from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.cloudtrail.cloudtrail_client import ( cloudtrail_client, ) +from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + get_cloudtrail_account_resource, + get_cloudtrail_threat_detection_identities, +) default_threat_detection_privilege_escalation_actions = [ "AddPermission", @@ -74,72 +76,29 @@ class cloudtrail_threat_detection_privilege_escalation(Check): default_threat_detection_privilege_escalation_actions, ) - potential_privilege_escalation = {} found_potential_privilege_escalation = False - multiregion_trail = None - # Check if any trail is multi-region so we only need to check once - for trail in cloudtrail_client.trails.values(): - if trail.is_multiregion: - multiregion_trail = trail - break - trails_to_scan = ( - cloudtrail_client.trails.values() - if not multiregion_trail - else [multiregion_trail] + potential_privilege_escalation = get_cloudtrail_threat_detection_identities( + cloudtrail_client, + privilege_escalation_actions, + threat_detection_minutes, ) - for trail in trails_to_scan: - 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, - ): - event_log = json.loads(event_log["CloudTrailEvent"]) - if ( - "arn" in event_log["userIdentity"] - ): # Ignore event logs without ARN since they are AWS services - if ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) not in potential_privilege_escalation: - potential_privilege_escalation[ - ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) - ] = set() - potential_privilege_escalation[ - ( - event_log["userIdentity"]["arn"], - event_log["userIdentity"]["type"], - ) - ].add(event_name) - for aws_identity, actions in potential_privilege_escalation.items(): + for resource, actions in potential_privilege_escalation.values(): identity_threshold = round( len(actions) / len(privilege_escalation_actions), 2 ) - aws_identity_type = aws_identity[1] - aws_identity_arn = aws_identity[0] if len(actions) / len(privilege_escalation_actions) > threshold: found_potential_privilege_escalation = True - report = Check_Report_AWS( - metadata=self.metadata(), resource=cloudtrail_client.trails - ) - report.region = cloudtrail_client.region - report.resource_id = aws_identity_arn.split("/")[-1] - report.resource_arn = aws_identity_arn + report = Check_Report_AWS(metadata=self.metadata(), resource=resource) report.status = "FAIL" - report.status_extended = f"Potential privilege escalation attack detected from AWS {aws_identity_type} {aws_identity_arn.split('/')[-1]} with a threshold of {identity_threshold}." + report.status_extended = f"Potential privilege escalation attack detected from AWS {resource.identity_type} {resource.name} with a threshold of {identity_threshold}." findings.append(report) if not found_potential_privilege_escalation: - report = Check_Report_AWS( - metadata=self.metadata(), resource=cloudtrail_client.trails - ) - report.region = cloudtrail_client.region - report.resource_id = cloudtrail_client.audited_account - report.resource_arn = cloudtrail_client._get_trail_arn_template( - cloudtrail_client.region + resource = get_cloudtrail_account_resource( + cloudtrail_client.audited_account, + cloudtrail_client.audited_account_arn, + cloudtrail_client.region, ) + report = Check_Report_AWS(metadata=self.metadata(), resource=resource) report.status = "PASS" report.status_extended = ( "No potential privilege escalation attack detected." diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py index 01f7322d70..5857b46815 100644 --- a/tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py +++ b/tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py @@ -1,7 +1,14 @@ +import json +from unittest import mock + from boto3 import client from moto import mock_aws -from prowler.providers.aws.services.cloudtrail.cloudtrail_service import Cloudtrail +from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + get_cloudtrail_threat_detection_identities, + normalize_cloudtrail_identity, +) from tests.providers.aws.utils import ( AWS_ACCOUNT_NUMBER, AWS_REGION_EU_SOUTH_2, @@ -343,3 +350,227 @@ class Test_Cloudtrail_Service: if trail.name: if trail.name == trail_name_us: assert trail.tags == [{"Key": "test", "Value": tag}] + + +class Test_normalize_cloudtrail_identity: + def test_iam_user_with_path(self): + identity_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:user/engineering/platform/attacker" + ) + + resource = normalize_cloudtrail_identity( + {"type": "IAMUser", "arn": identity_arn}, AWS_REGION_US_EAST_1 + ) + + assert resource.id == "user/engineering/platform/attacker" + assert resource.name == "attacker" + assert resource.arn == identity_arn + assert resource.region == AWS_REGION_US_EAST_1 + assert resource.identity_type == "IAMUser" + assert resource.source_arn == identity_arn + + def test_assumed_role_with_session_issuer(self): + source_arn = ( + f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/platform/admin/session-one" + ) + role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/platform/admin" + + resource = normalize_cloudtrail_identity( + { + "type": "AssumedRole", + "arn": source_arn, + "sessionContext": {"sessionIssuer": {"arn": role_arn}}, + }, + AWS_REGION_US_EAST_1, + ) + + assert resource.id == "role/platform/admin" + assert resource.name == "admin" + assert resource.arn == role_arn + assert resource.identity_type == "AssumedRole" + assert resource.source_arn == source_arn + + def test_assumed_role_without_session_issuer(self): + source_arn = ( + f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/platform-admin/session-one" + ) + + resource = normalize_cloudtrail_identity( + {"type": "AssumedRole", "arn": source_arn}, AWS_REGION_US_EAST_1 + ) + + assert resource.id == "assumed-role/platform-admin/session-one" + assert resource.name == "platform-admin" + assert resource.arn == source_arn + assert resource.source_arn == source_arn + + def test_user_and_role_with_same_leaf_name_are_distinct(self): + user = normalize_cloudtrail_identity( + { + "type": "IAMUser", + "arn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:user/team/operator", + }, + AWS_REGION_US_EAST_1, + ) + role = normalize_cloudtrail_identity( + { + "type": "AssumedRole", + "arn": f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/operator/session", + "sessionContext": { + "sessionIssuer": { + "arn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/team/operator" + } + }, + }, + AWS_REGION_US_EAST_1, + ) + + assert user.id == "user/team/operator" + assert role.id == "role/team/operator" + + def test_roles_with_same_session_name_are_distinct(self): + first_role = normalize_cloudtrail_identity( + { + "type": "AssumedRole", + "arn": f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/first/shared-session", + "sessionContext": { + "sessionIssuer": { + "arn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/first" + } + }, + }, + AWS_REGION_US_EAST_1, + ) + second_role = normalize_cloudtrail_identity( + { + "type": "AssumedRole", + "arn": f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/second/shared-session", + "sessionContext": { + "sessionIssuer": { + "arn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/second" + } + }, + }, + AWS_REGION_US_EAST_1, + ) + + assert first_role.id == "role/first" + assert second_role.id == "role/second" + + def test_federated_user(self): + identity_arn = f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:federated-user/external-user" + + resource = normalize_cloudtrail_identity( + {"type": "FederatedUser", "arn": identity_arn}, + AWS_REGION_US_EAST_1, + ) + + assert resource.id == "federated-user/external-user" + assert resource.name == "external-user" + assert resource.arn == identity_arn + + def test_root(self): + identity_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + + resource = normalize_cloudtrail_identity( + {"type": "Root", "arn": identity_arn}, AWS_REGION_US_EAST_1 + ) + + assert resource.id == "root" + assert resource.name == "root" + assert resource.arn == identity_arn + + def test_unknown_identity_with_arn(self): + identity_arn = ( + f"arn:aws:custom:us-east-1:{AWS_ACCOUNT_NUMBER}:resource/path/name" + ) + + resource = normalize_cloudtrail_identity( + {"type": "UnknownType", "arn": identity_arn}, AWS_REGION_US_EAST_1 + ) + + assert resource.id == "resource/path/name" + assert resource.name == "name" + assert resource.arn == identity_arn + + def test_identity_without_arn_is_ignored(self): + assert ( + normalize_cloudtrail_identity({"type": "AWSService"}, AWS_REGION_US_EAST_1) + is None + ) + + +class Test_get_cloudtrail_threat_detection_identities: + def test_same_role_sessions_aggregate_by_canonical_role_arn(self): + role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/platform/admin" + cloudtrail_client = mock.MagicMock() + cloudtrail_client.region = AWS_REGION_US_EAST_1 + cloudtrail_client.trails = {"trail": mock.MagicMock(is_multiregion=False)} + + def lookup_events(trail, event_name, minutes): + session_name = "session-one" if event_name == "ActionOne" else "session-two" + return [ + { + "CloudTrailEvent": json.dumps( + { + "userIdentity": { + "type": "AssumedRole", + "arn": f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/admin/{session_name}", + "sessionContext": {"sessionIssuer": {"arn": role_arn}}, + } + } + ) + } + ] + + cloudtrail_client._lookup_events = lookup_events + + identities = get_cloudtrail_threat_detection_identities( + cloudtrail_client, ["ActionOne", "ActionTwo"], 60 + ) + + assert list(identities) == [role_arn] + resource, actions = identities[role_arn] + assert resource.id == "role/platform/admin" + assert actions == {"ActionOne", "ActionTwo"} + + def test_different_roles_with_same_session_name_remain_distinct(self): + first_role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/first" + second_role_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/second" + cloudtrail_client = mock.MagicMock() + cloudtrail_client.region = AWS_REGION_US_EAST_1 + cloudtrail_client.trails = {"trail": mock.MagicMock(is_multiregion=False)} + cloudtrail_client._lookup_events = lambda trail, event_name, minutes: [ + { + "CloudTrailEvent": json.dumps( + { + "userIdentity": { + "type": "AssumedRole", + "arn": f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/first/shared-session", + "sessionContext": { + "sessionIssuer": {"arn": first_role_arn} + }, + } + } + ) + }, + { + "CloudTrailEvent": json.dumps( + { + "userIdentity": { + "type": "AssumedRole", + "arn": f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/second/shared-session", + "sessionContext": { + "sessionIssuer": {"arn": second_role_arn} + }, + } + } + ) + }, + ] + + identities = get_cloudtrail_threat_detection_identities( + cloudtrail_client, ["ActionOne"], 60 + ) + + assert set(identities) == {first_role_arn, second_role_arn} 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 index 8c237f98ef..bff85cd069 100644 --- 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 @@ -1,7 +1,10 @@ +import json +from types import SimpleNamespace from unittest import mock from moto import mock_aws +from prowler.lib.outputs.finding import Finding from tests.providers.aws.utils import ( AWS_ACCOUNT_NUMBER, AWS_REGION_US_EAST_1, @@ -17,6 +20,7 @@ def mock_get_trail_arn_template(region=None, *_) -> str: def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> list: + del trail, minutes return [ { "CloudTrailEvent": '{"eventName": "DescribeAccessEntry", "userIdentity": {"type": "IAMUser", "principalId": "EXAMPLE6E4XEGITWATV6R", "arn": "arn:aws:iam::123456789012:user/Attacker", "accountId": "123456789012", "accessKeyId": "AKIAIOSFODNN7EXAMPLE", "userName": "Attacker", "sessionContext": {"sessionIssuer": {}, "webIdFederationData": {}, "attributes": {"creationDate": "2023-07-19T21:11:57Z", "mfaAuthenticated": "false"}}}}' @@ -30,6 +34,7 @@ def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> def mock__get_lookup_events_aws_service__( trail=None, event_name=None, minutes=None, *_ ) -> list: + del trail, minutes return [ { "CloudTrailEvent": '{"eventName": "DescribeAccessEntry", "userIdentity": {"type": "AWSService", "principalId": "EXAMPLE6E4XEGITWATV6R", "accountId": "123456789012", "sessionContext": {"sessionIssuer": {}, "webIdFederationData": {}, "attributes": {"creationDate": "2023-07-19T21:11:57Z", "mfaAuthenticated": "false"}}}}' @@ -40,6 +45,34 @@ def mock__get_lookup_events_aws_service__( ] +def mock__get_lookup_events_assumed_role__( + trail=None, event_name=None, minutes=None, *_ +) -> list: + del trail, minutes + session_name = ( + "enumeration-session-one" + if event_name == "DescribeAccessEntry" + else "enumeration-session-two" + ) + return [ + { + "CloudTrailEvent": json.dumps( + { + "userIdentity": { + "type": "AssumedRole", + "arn": f"arn:aws:sts::{AWS_ACCOUNT_NUMBER}:assumed-role/platform-attacker/{session_name}", + "sessionContext": { + "sessionIssuer": { + "arn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/security/platform-attacker" + } + }, + } + } + ) + } + ] + + class Test_cloudtrail_threat_detection_enumeration: @mock_aws def test_no_trails(self): @@ -48,6 +81,9 @@ class Test_cloudtrail_threat_detection_enumeration: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 with ( @@ -75,10 +111,8 @@ class Test_cloudtrail_threat_detection_enumeration: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + assert result[0].resource["identity_type"] == "AWSAccount" @mock_aws def test_no_potential_enumeration(self): @@ -92,6 +126,9 @@ class Test_cloudtrail_threat_detection_enumeration: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_enumeration_actions": ENUMERATION_ACTIONS, @@ -127,10 +164,7 @@ class Test_cloudtrail_threat_detection_enumeration: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" @mock_aws def test_potential_enumeration(self): @@ -144,6 +178,9 @@ class Test_cloudtrail_threat_detection_enumeration: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_enumeration_actions": ENUMERATION_ACTIONS, @@ -178,12 +215,15 @@ class Test_cloudtrail_threat_detection_enumeration: result[0].status_extended == "Potential enumeration attack detected from AWS IAMUser Attacker with a threshold of 1.0." ) - assert result[0].resource_id == "Attacker" + assert result[0].resource_id == "user/Attacker" assert result[0].region == AWS_REGION_US_EAST_1 assert ( result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:user/Attacker" ) + assert result[0].resource["name"] == "Attacker" + assert result[0].resource["identity_type"] == "IAMUser" + assert result[0].check_metadata.ResourceType == "Other" @mock_aws def test_big_threshold(self): @@ -197,6 +237,9 @@ class Test_cloudtrail_threat_detection_enumeration: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_enumeration_actions": ENUMERATION_ACTIONS, @@ -232,10 +275,7 @@ class Test_cloudtrail_threat_detection_enumeration: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" @mock_aws def test_potential_enumeration_from_aws_service(self): @@ -249,6 +289,9 @@ class Test_cloudtrail_threat_detection_enumeration: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_enumeration_actions": ENUMERATION_ACTIONS, @@ -284,7 +327,61 @@ class Test_cloudtrail_threat_detection_enumeration: ) 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:iam::{AWS_ACCOUNT_NUMBER}:root" + + @mock_aws + def test_assumed_role_sessions_aggregate_into_one_finding(self): + aws_provider = set_mocked_aws_provider() + cloudtrail_client = mock.MagicMock() + cloudtrail_client.trails = {"us-east-1": mock.MagicMock()} + cloudtrail_client.trails["us-east-1"].is_multiregion = False + cloudtrail_client.audited_account = AWS_ACCOUNT_NUMBER + cloudtrail_client.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) + cloudtrail_client.region = AWS_REGION_US_EAST_1 + cloudtrail_client.audit_config = { + "threat_detection_enumeration_actions": [ + "DescribeAccessEntry", + "DescribeAccountAttributes", + ], + "threat_detection_enumeration_threshold": 0.6, + "threat_detection_enumeration_minutes": 1440, + } + cloudtrail_client._lookup_events = mock__get_lookup_events_assumed_role__ + + with ( + mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), + mock.patch( + "prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration.cloudtrail_client", + new=cloudtrail_client, + ), + ): + from prowler.providers.aws.services.cloudtrail.cloudtrail_threat_detection_enumeration.cloudtrail_threat_detection_enumeration import ( + cloudtrail_threat_detection_enumeration, + ) + + result = cloudtrail_threat_detection_enumeration().execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert result[0].resource_id == "role/security/platform-attacker" assert ( result[0].resource_arn - == f"arn:aws:cloudtrail:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:trail" + == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:role/security/platform-attacker" ) + assert result[0].resource["identity_type"] == "AssumedRole" + assert result[0].resource["source_arn"].startswith("arn:aws:sts::") + + finding = Finding.generate_output( + aws_provider, + result[0], + SimpleNamespace(unix_timestamp=False, bulk_checks_metadata={}), + ) + assert finding.resource_name == "role/security/platform-attacker" + assert finding.resource_uid == result[0].resource_arn + assert finding.resource_metadata == result[0].resource + assert finding.uid.endswith("-role/security/platform-attacker") diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_test.py index bf22af9e87..54d8d3de7a 100644 --- a/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_test.py +++ b/tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_test.py @@ -17,6 +17,7 @@ def mock_get_trail_arn_template(region=None, *_) -> str: def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> list: + del trail, minutes return [ { "CloudTrailEvent": '{"eventName": "InvokeModel", "userIdentity": {"type": "IAMUser", "principalId": "EXAMPLE6E4XEGITWATV6R", "arn": "arn:aws:iam::123456789012:user/Attacker", "accountId": "123456789012", "accessKeyId": "AKIAIOSFODNN7EXAMPLE", "userName": "Attacker", "sessionContext": {"sessionIssuer": {}, "webIdFederationData": {}, "attributes": {"creationDate": "2023-07-19T21:11:57Z", "mfaAuthenticated": "false"}}}}' @@ -30,6 +31,7 @@ def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> def mock__get_lookup_events_aws_service__( trail=None, event_name=None, minutes=None, *_ ) -> list: + del trail, minutes return [ { "CloudTrailEvent": '{"eventName": "InvokeModel", "userIdentity": {"type": "AWSService", "principalId": "EXAMPLE6E4XEGITWATV6R", "accountId": "123456789012", "accessKeyId": "AKIAIOSFODNN7EXAMPLE", "sessionContext": {"sessionIssuer": {}, "webIdFederationData": {}, "attributes": {"creationDate": "2023-07-19T21:11:57Z", "mfaAuthenticated": "false"}}}}' @@ -48,6 +50,9 @@ class Test_cloudtrail_threat_detection_llm_jacking: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 with ( @@ -75,10 +80,8 @@ class Test_cloudtrail_threat_detection_llm_jacking: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + assert result[0].resource["identity_type"] == "AWSAccount" @mock_aws def test_no_potential_llm_jacking(self): @@ -89,6 +92,9 @@ class Test_cloudtrail_threat_detection_llm_jacking: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_llm_jacking_actions": [], @@ -124,10 +130,7 @@ class Test_cloudtrail_threat_detection_llm_jacking: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" @mock_aws def test_potential_priviledge_escalation(self): @@ -138,6 +141,9 @@ class Test_cloudtrail_threat_detection_llm_jacking: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_llm_jacking_actions": [ @@ -175,12 +181,14 @@ class Test_cloudtrail_threat_detection_llm_jacking: result[0].status_extended == "Potential LLM Jacking attack detected from AWS IAMUser Attacker with a threshold of 1.0." ) - assert result[0].resource_id == "Attacker" + assert result[0].resource_id == "user/Attacker" assert result[0].region == AWS_REGION_US_EAST_1 assert ( result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:user/Attacker" ) + assert result[0].resource["identity_type"] == "IAMUser" + assert result[0].check_metadata.ResourceType == "Other" @mock_aws def test_bigger_threshold(self): @@ -191,6 +199,9 @@ class Test_cloudtrail_threat_detection_llm_jacking: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_llm_jacking_actions": [ @@ -229,10 +240,7 @@ class Test_cloudtrail_threat_detection_llm_jacking: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" @mock_aws def test_potential_enumeration_from_aws_service(self): @@ -243,6 +251,9 @@ class Test_cloudtrail_threat_detection_llm_jacking: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_llm_jacking_actions": [ @@ -281,7 +292,4 @@ class Test_cloudtrail_threat_detection_llm_jacking: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" 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 index b413494f1c..58ba6a27d7 100644 --- 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 @@ -17,6 +17,7 @@ def mock_get_trail_arn_template(region=None, *_) -> str: def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> list: + del trail, minutes return [ { "CloudTrailEvent": '{"eventName": "CreateLoginProfile", "userIdentity": {"type": "IAMUser", "principalId": "EXAMPLE6E4XEGITWATV6R", "arn": "arn:aws:iam::123456789012:user/Attacker", "accountId": "123456789012", "accessKeyId": "AKIAIOSFODNN7EXAMPLE", "userName": "Attacker", "sessionContext": {"sessionIssuer": {}, "webIdFederationData": {}, "attributes": {"creationDate": "2023-07-19T21:11:57Z", "mfaAuthenticated": "false"}}}}' @@ -30,6 +31,7 @@ def mock__get_lookup_events__(trail=None, event_name=None, minutes=None, *_) -> def mock__get_lookup_events_aws_service__( trail=None, event_name=None, minutes=None, *_ ) -> list: + del trail, minutes return [ { "CloudTrailEvent": '{"eventName": "CreateLoginProfile", "userIdentity": {"type": "AWSService", "principalId": "EXAMPLE6E4XEGITWATV6R", "accountId": "123456789012", "accessKeyId": "AKIAIOSFODNN7EXAMPLE", "sessionContext": {"sessionIssuer": {}, "webIdFederationData": {}, "attributes": {"creationDate": "2023-07-19T21:11:57Z", "mfaAuthenticated": "false"}}}}' @@ -48,6 +50,9 @@ class Test_cloudtrail_threat_detection_privilege_escalation: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 with ( @@ -76,10 +81,8 @@ class Test_cloudtrail_threat_detection_privilege_escalation: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + assert result[0].resource["identity_type"] == "AWSAccount" @mock_aws def test_no_potential_priviledge_escalation(self): @@ -90,6 +93,9 @@ class Test_cloudtrail_threat_detection_privilege_escalation: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_privilege_escalation_actions": [], @@ -126,10 +132,7 @@ class Test_cloudtrail_threat_detection_privilege_escalation: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" @mock_aws def test_potential_priviledge_escalation(self): @@ -140,6 +143,9 @@ class Test_cloudtrail_threat_detection_privilege_escalation: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_privilege_escalation_actions": [ @@ -177,12 +183,14 @@ class Test_cloudtrail_threat_detection_privilege_escalation: result[0].status_extended == "Potential privilege escalation attack detected from AWS IAMUser Attacker with a threshold of 1.0." ) - assert result[0].resource_id == "Attacker" + assert result[0].resource_id == "user/Attacker" assert result[0].region == AWS_REGION_US_EAST_1 assert ( result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:user/Attacker" ) + assert result[0].resource["identity_type"] == "IAMUser" + assert result[0].check_metadata.ResourceType == "Other" @mock_aws def test_bigger_threshold(self): @@ -193,6 +201,9 @@ class Test_cloudtrail_threat_detection_privilege_escalation: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_privilege_escalation_actions": [ @@ -232,10 +243,7 @@ class Test_cloudtrail_threat_detection_privilege_escalation: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" @mock_aws def test_potential_enumeration_from_aws_service(self): @@ -246,6 +254,9 @@ class Test_cloudtrail_threat_detection_privilege_escalation: 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.audited_account_arn = ( + f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + ) cloudtrail_client.region = AWS_REGION_US_EAST_1 cloudtrail_client.audit_config = { "threat_detection_privilege_escalation_actions": [ @@ -285,7 +296,4 @@ class Test_cloudtrail_threat_detection_privilege_escalation: ) 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" - ) + assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"