ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-24 13:01:56 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(cloudtrail): trail.region must be home region (#4153)

Browse Source 
Co-authored-by: Sergio <sergio@prowler.com>
This commit is contained in:
Pepe Fagoaga
2024-06-03 19:19:40 +02:00
committed by GitHub
parent 2affed81ad
commit f895e4df6a
15 changed files with 187 additions and 92 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/aws/services/cloudtrail/cloudtrail_bucket_requires_mfa_delete/cloudtrail_bucket_requires_mfa_delete.py
+1 -1
View File
@@ -14,7 +14,7 @@ class cloudtrail_bucket_requires_mfa_delete(Check):
trail_bucket_is_in_account = False
trail_bucket = trail.s3_bucket
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_cloudwatch_logging_enabled/cloudtrail_cloudwatch_logging_enabled.py
+1 -1
View File
@@ -15,7 +15,7 @@ class cloudtrail_cloudwatch_logging_enabled(Check):
for trail in cloudtrail_client.trails.values():
if trail.name:
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_insights_exist/cloudtrail_insights_exist.py
+1 -1
View File
@@ -11,7 +11,7 @@ class cloudtrail_insights_exist(Check):
for trail in cloudtrail_client.trails.values():
if trail.is_logging:
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_kms_encryption_enabled/cloudtrail_kms_encryption_enabled.py
+1 -1
View File
@@ -11,7 +11,7 @@ class cloudtrail_kms_encryption_enabled(Check):
for trail in cloudtrail_client.trails.values():
if trail.name:
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_log_file_validation_enabled/cloudtrail_log_file_validation_enabled.py
+1 -1
View File
@@ -11,7 +11,7 @@ class cloudtrail_log_file_validation_enabled(Check):
for trail in cloudtrail_client.trails.values():
if trail.name:
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_access_logging_enabled/cloudtrail_logs_s3_bucket_access_logging_enabled.py
+1 -1
View File
@@ -14,7 +14,7 @@ class cloudtrail_logs_s3_bucket_access_logging_enabled(Check):
trail_bucket_is_in_account = False
trail_bucket = trail.s3_bucket
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible.py
+1 -1
View File
@@ -14,7 +14,7 @@ class cloudtrail_logs_s3_bucket_is_not_publicly_accessible(Check):
trail_bucket_is_in_account = False
trail_bucket = trail.s3_bucket
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled_logging_management_events/cloudtrail_multi_region_enabled_logging_management_events.py
+50 -43
View File
@@ -8,48 +8,55 @@ class cloudtrail_multi_region_enabled_logging_management_events(Check):
def execute(self):
findings = []
if cloudtrail_client.trails is not None:
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No trail found with multi-region enabled and logging management events."
report.region = cloudtrail_client.region
report.resource_id = cloudtrail_client.audited_account
report.resource_arn = cloudtrail_client.trail_arn_template
for trail in cloudtrail_client.trails.values():
if trail.is_logging:
if trail.is_multiregion:
for event in trail.data_events:
# Classic event selectors
if not event.is_advanced:
# Check if trail has IncludeManagementEvents and ReadWriteType is All
if (
event.event_selector["ReadWriteType"] == "All"
and event.event_selector["IncludeManagementEvents"]
):
report.region = trail.region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
report.status = "PASS"
report.status_extended = f"Trail {trail.name} from home region {trail.home_region} is multi-region, is logging and have management events enabled."
# Advanced event selectors
elif event.is_advanced:
if event.event_selector.get(
"Name"
) == "Management events selector" and all(
[
field["Field"] != "readOnly"
for field in event.event_selector[
"FieldSelectors"
for region in cloudtrail_client.regional_clients.keys():
report = Check_Report_AWS(self.metadata())
report.status = "FAIL"
report.status_extended = "No CloudTrail trails enabled and logging management events were found."
report.region = region
report.resource_id = cloudtrail_client.audited_account
report.resource_arn = cloudtrail_client.trail_arn_template
trail_is_logging_management_events = False
for trail in cloudtrail_client.trails.values():
if trail.region == region or trail.is_multiregion:
if trail.is_logging:
for event in trail.data_events:
# Classic event selectors
if not event.is_advanced:
# Check if trail has IncludeManagementEvents and ReadWriteType is All
if (
event.event_selector["ReadWriteType"] == "All"
and event.event_selector[
"IncludeManagementEvents"
]
]
):
report.region = trail.region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
report.status = "PASS"
report.status_extended = f"Trail {trail.name} from home region {trail.home_region} is multi-region, is logging and have management events enabled."
findings.append(report)
):
trail_is_logging_management_events = True
# Advanced event selectors
elif event.is_advanced:
if event.event_selector.get(
"Name"
) == "Management events selector" and all(
[
field["Field"] != "readOnly"
for field in event.event_selector[
"FieldSelectors"
]
]
):
trail_is_logging_management_events = True
if trail_is_logging_management_events:
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
report.status = "PASS"
if trail.is_multiregion:
report.status_extended = f"Trail {trail.name} from home region {trail.home_region} is multi-region, is logging and have management events enabled."
else:
report.status_extended = f"Trail {trail.name} in region {trail.home_region} is logging and have management events enabled."
# Since there exists a logging trail in that region there is no point in checking the remaining trails
# Store the finding and exit the loop
findings.append(report)
break
if report.status == "FAIL":
findings.append(report)
return findings
prowler/providers/aws/services/cloudtrail/cloudtrail_s3_dataevents_read_enabled/cloudtrail_s3_dataevents_read_enabled.py
+2 -2
View File
@@ -28,7 +28,7 @@ class cloudtrail_s3_dataevents_read_enabled(Check):
in resource["Values"]
):
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
@@ -45,7 +45,7 @@ class cloudtrail_s3_dataevents_read_enabled(Check):
and field_selector["Equals"][0] == "AWS::S3::Object"
):
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_s3_dataevents_write_enabled/cloudtrail_s3_dataevents_write_enabled.py
+2 -2
View File
@@ -28,7 +28,7 @@ class cloudtrail_s3_dataevents_write_enabled(Check):
in resource["Values"]
):
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
@@ -45,7 +45,7 @@ class cloudtrail_s3_dataevents_write_enabled(Check):
and field_selector["Equals"][0] == "AWS::S3::Object"
):
report = Check_Report_AWS(self.metadata())
report.region = trail.region
report.region = trail.home_region
report.resource_id = trail.name
report.resource_arn = trail.arn
report.resource_tags = trail.tags
prowler/providers/aws/services/cloudtrail/cloudtrail_service.py
+20 -10
View File
@@ -36,6 +36,10 @@ class Cloudtrail(AWSService):
describe_trails = regional_client.describe_trails()["trailList"]
trails_count = 0
for trail in describe_trails:
# If a multi region trail was already retrieved in another region
if trail["TrailARN"] in self.trails.keys():
continue
if not self.audit_resources or (
is_resource_filtered(trail["TrailARN"], self.audit_resources)
):
@@ -208,16 +212,21 @@ class Cloudtrail(AWSService):
logger.info("CloudTrail - List Tags...")
try:
for trail in self.trails.values():
# Check if trails are in this account and region
if (
trail.region == trail.home_region
and self.audited_account in trail.arn
):
regional_client = self.regional_clients[trail.region]
response = regional_client.list_tags(ResourceIdList=[trail.arn])[
"ResourceTagList"
][0]
trail.tags = response.get("TagsList")
try:
# Check if trails are in this account and region
if (
trail.region == trail.home_region
and self.audited_account in trail.arn
):
regional_client = self.regional_clients[trail.region]
response = regional_client.list_tags(
ResourceIdList=[trail.arn]
)["ResourceTagList"][0]
trail.tags = response.get("TagsList")
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -234,6 +243,7 @@ class Trail(BaseModel):
is_multiregion: bool = None
home_region: str = None
arn: str = None
# Region holds the region where the trail is audited
region: str
is_logging: bool = None
log_file_validation_enabled: bool = None
tests/providers/aws/services/cloudtrail/cloudtrail_cloudwatch_logging_enabled/cloudtrail_cloudwatch_logging_enabled_test.py
+4
View File
@@ -205,6 +205,7 @@ class Test_cloudtrail_cloudwatch_logging_enabled:
report.status_extended,
f"Multiregion trail {trail_name_us} has been logging the last 24h.",
)
assert report.region == AWS_REGION_US_EAST_1
assert report.resource_tags == []
if (
report.resource_id == trail_name_eu
@@ -217,6 +218,7 @@ class Test_cloudtrail_cloudwatch_logging_enabled:
report.status_extended,
f"Single region trail {trail_name_eu} is not logging in the last 24h.",
)
assert report.region == AWS_REGION_EU_WEST_1
assert report.resource_tags == []
@mock_aws
@@ -293,6 +295,7 @@ class Test_cloudtrail_cloudwatch_logging_enabled:
report.status_extended
== f"Single region trail {trail_name_us} has been logging the last 24h."
)
assert report.region == AWS_REGION_US_EAST_1
assert report.resource_tags == []
if report.resource_id == trail_name_eu:
assert report.resource_id == trail_name_eu
@@ -302,6 +305,7 @@ class Test_cloudtrail_cloudwatch_logging_enabled:
report.status_extended
== f"Single region trail {trail_name_eu} is not logging in the last 24h or not configured to deliver logs."
)
assert report.region == AWS_REGION_EU_WEST_1
assert report.resource_tags == []
@mock_aws
tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_access_logging_enabled/cloudtrail_logs_s3_bucket_access_logging_enabled_test.py
-1
View File
@@ -229,7 +229,6 @@ class Test_cloudtrail_logs_s3_bucket_access_logging_enabled:
@mock_aws
def test_access_denied(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
tests/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled_logging_management_events/cloudtrail_multi_region_enabled_logging_management_events_test.py
+37 -3
View File
@@ -6,6 +6,7 @@ from moto import mock_aws
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_US_EAST_1,
AWS_REGION_US_EAST_2,
set_mocked_aws_provider,
)
@@ -44,7 +45,7 @@ class Test_cloudtrail_multi_region_enabled_logging_management_events:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "No trail found with multi-region enabled and logging management events."
== "No CloudTrail trails enabled and logging management events were found."
)
@mock_aws
@@ -159,7 +160,7 @@ class Test_cloudtrail_multi_region_enabled_logging_management_events:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "No trail found with multi-region enabled and logging management events."
== "No CloudTrail trails enabled and logging management events were found."
)
@mock_aws
@@ -271,7 +272,7 @@ class Test_cloudtrail_multi_region_enabled_logging_management_events:
assert result[0].status == "FAIL"
assert (
result[0].status_extended
== "No trail found with multi-region enabled and logging management events."
== "No CloudTrail trails enabled and logging management events were found."
)
@mock_aws
@@ -299,3 +300,36 @@ class Test_cloudtrail_multi_region_enabled_logging_management_events:
check = cloudtrail_multi_region_enabled_logging_management_events()
result = check.execute()
assert len(result) == 0
def test_no_trails_two_regions(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
Cloudtrail,
)
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_US_EAST_2]
)
with mock.patch(
"prowler.providers.common.provider.Provider.get_global_provider",
return_value=aws_provider,
):
with mock.patch(
"prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled_logging_management_events.cloudtrail_multi_region_enabled_logging_management_events.cloudtrail_client",
new=Cloudtrail(aws_provider),
):
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled_logging_management_events.cloudtrail_multi_region_enabled_logging_management_events import (
cloudtrail_multi_region_enabled_logging_management_events,
)
check = cloudtrail_multi_region_enabled_logging_management_events()
result = check.execute()
assert len(result) == 2
for r in result:
assert r.resource_id == AWS_ACCOUNT_NUMBER
assert r.status == "FAIL"
assert (
r.status_extended
== "No CloudTrail trails enabled and logging management events were found."
)
tests/providers/aws/services/cloudtrail/cloudtrail_service_test.py
+65 -24
View File
@@ -4,6 +4,7 @@ from moto import mock_aws
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import Cloudtrail
from tests.providers.aws.utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_SOUTH_2,
AWS_REGION_EU_WEST_1,
AWS_REGION_US_EAST_1,
set_mocked_aws_provider,
@@ -50,23 +51,14 @@ class Test_Cloudtrail_Service:
@mock_aws
def test_describe_trails(self):
# USA
cloudtrail_client_us_east_1 = client(
"cloudtrail", region_name=AWS_REGION_US_EAST_1
)
s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1)
cloudtrail_client_eu_west_1 = client(
"cloudtrail", region_name=AWS_REGION_EU_WEST_1
)
s3_client_eu_west_1 = client("s3", region_name=AWS_REGION_EU_WEST_1)
trail_name_us = "trail_test_us"
bucket_name_us = "bucket_test_us"
trail_name_eu = "trail_test_eu"
bucket_name_eu = "bucket_test_eu"
s3_client_us_east_1.create_bucket(Bucket=bucket_name_us)
s3_client_eu_west_1.create_bucket(
Bucket=bucket_name_eu,
CreateBucketConfiguration={"LocationConstraint": AWS_REGION_EU_WEST_1},
)
cloudtrail_client_us_east_1.create_trail(
Name=trail_name_us,
S3BucketName=bucket_name_us,
@@ -75,6 +67,18 @@ class Test_Cloudtrail_Service:
{"Key": "test", "Value": "test"},
],
)
# IRELAND
cloudtrail_client_eu_west_1 = client(
"cloudtrail", region_name=AWS_REGION_EU_WEST_1
)
s3_client_eu_west_1 = client("s3", region_name=AWS_REGION_EU_WEST_1)
trail_name_eu = "trail_test_eu"
bucket_name_eu = "bucket_test_eu"
s3_client_eu_west_1.create_bucket(
Bucket=bucket_name_eu,
CreateBucketConfiguration={"LocationConstraint": AWS_REGION_EU_WEST_1},
)
cloudtrail_client_eu_west_1.create_trail(
Name=trail_name_eu,
S3BucketName=bucket_name_eu,
@@ -83,19 +87,60 @@ class Test_Cloudtrail_Service:
{"Key": "test", "Value": "test"},
],
)
# SPAIN
cloudtrail_client_eu_south_2 = client(
"cloudtrail", region_name=AWS_REGION_EU_SOUTH_2
)
s3_client_eu_south_2 = client("s3", region_name=AWS_REGION_EU_SOUTH_2)
trail_name_sp = "trail_test_sp"
bucket_name_sp = "bucket_test_sp"
s3_client_eu_south_2.create_bucket(
Bucket=bucket_name_sp,
CreateBucketConfiguration={"LocationConstraint": AWS_REGION_EU_SOUTH_2},
)
cloudtrail_client_eu_south_2.create_trail(
Name=trail_name_sp,
S3BucketName=bucket_name_sp,
IsMultiRegionTrail=True,
TagsList=[
{"Key": "test", "Value": "test"},
],
)
# We are not going to include AWS_REGION_EU_SOUTH_2 in the audited
# regions, but that trail is regional so it'll appear
aws_provider = set_mocked_aws_provider(
[AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
)
cloudtrail = Cloudtrail(aws_provider)
assert len(cloudtrail.trails) == 2
assert len(cloudtrail.trails) == 3
for trail in cloudtrail.trails.values():
if trail.name:
assert trail.name == trail_name_us or trail.name == trail_name_eu
if trail.name == trail_name_us:
assert not trail.is_multiregion
assert (
trail.home_region == AWS_REGION_US_EAST_1
or trail.home_region == AWS_REGION_EU_WEST_1
)
assert trail.home_region == AWS_REGION_US_EAST_1
assert trail.region == AWS_REGION_US_EAST_1
assert not trail.is_logging
assert not trail.log_file_validation_enabled
assert not trail.latest_cloudwatch_delivery_time
assert trail.s3_bucket == bucket_name_us
assert trail.tags == [
{"Key": "test", "Value": "test"},
]
if trail.name == trail_name_eu:
assert not trail.is_multiregion
assert trail.home_region == AWS_REGION_EU_WEST_1
assert trail.region == AWS_REGION_EU_WEST_1
assert not trail.is_logging
assert not trail.log_file_validation_enabled
assert not trail.latest_cloudwatch_delivery_time
assert trail.s3_bucket == bucket_name_eu
assert trail.tags == [
{"Key": "test", "Value": "test"},
]
if trail.name == trail_name_sp:
assert trail.is_multiregion
assert trail.home_region == AWS_REGION_EU_SOUTH_2
# The region is the first audited region since the trail home region is not audited
assert (
trail.region == AWS_REGION_US_EAST_1
or trail.region == AWS_REGION_EU_WEST_1
@@ -103,13 +148,9 @@ class Test_Cloudtrail_Service:
assert not trail.is_logging
assert not trail.log_file_validation_enabled
assert not trail.latest_cloudwatch_delivery_time
assert (
trail.s3_bucket == bucket_name_eu
or trail.s3_bucket == bucket_name_us
)
assert trail.tags == [
{"Key": "test", "Value": "test"},
]
assert trail.s3_bucket == bucket_name_sp
# No tags since the trail region is not audited and the tags are retrieved from the regional endpoint
assert trail.tags == []
@mock_aws
def test_status_trails(self):