ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-05 03:12:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(cloudtrail): align threat detection resource identities

Browse Source 
- Normalize principal identities across threat-detection checks
- Aggregate assumed-role sessions by canonical role
- Add regression coverage and changelog entry
This commit is contained in:
Hugo P.Brito
2026-09-02 11:29:45 +01:00
parent 18453e592e
commit 845d555fc9
12 changed files with 524 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/changelog.d/cloudtrail-threat-detection-resource-identity.fixed.md
+1
View File
@@ -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
prowler/providers/aws/services/cloudtrail/cloudtrail_service.py
+83 -1
View File
@@ -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,
)
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.metadata.json
+1 -1
View File
@@ -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.",
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration.py
+14 -57
View File
@@ -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)
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.metadata.json
+1 -1
View File
@@ -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.",
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking.py
+14 -57
View File
@@ -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)
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.metadata.json
+1 -1
View File
@@ -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.",
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation.py
+16 -57
View File
@@ -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."
tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py
+232 -1
View File
@@ -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}
tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_test.py
+111 -14
View File
@@ -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")
tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_test.py
+25 -17
View File
@@ -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"
tests/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_test.py
+25 -17
View File
@@ -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"