diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py new file mode 100644 index 0000000000..f79780093f --- /dev/null +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py @@ -0,0 +1,52 @@ +from prowler.lib.logger import logger +from prowler.providers.aws.services.cloudtrail.cloudtrail_client import ( + cloudtrail_client, +) +from prowler.providers.aws.services.s3.s3_client import s3_client + + +def fixer(resource_id: str, region: str) -> bool: + """ + Modify the CloudTrail's associated S3 bucket's public access settings to ensure the bucket is not publicly accessible. + Specifically, this fixer configures the S3 bucket's public access block settings to block all public access. + Requires the s3:PutBucketPublicAccessBlock permissions. + Permissions: + { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Action": "s3:PutBucketPublicAccessBlock", + "Resource": "*" + } + ] + } + Args: + resource_id (str): The CloudTrail name. + region (str): AWS region where the CloudTrail and S3 bucket exist. + Returns: + bool: True if the operation is successful (policy and ACL updated), False otherwise. + """ + try: + regional_client = s3_client.regional_clients[region] + for trail in cloudtrail_client.trails.values(): + if trail.name == resource_id: + trail_bucket = trail.s3_bucket + + regional_client.put_public_access_block( + Bucket=trail_bucket, + PublicAccessBlockConfiguration={ + "BlockPublicAcls": True, + "IgnorePublicAcls": True, + "BlockPublicPolicy": True, + "RestrictPublicBuckets": True, + }, + ) + + except Exception as error: + logger.error( + f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + return False + else: + return True diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py new file mode 100644 index 0000000000..bd384748c2 --- /dev/null +++ b/tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py @@ -0,0 +1,145 @@ +from unittest import mock + +import botocore +import botocore.client +from boto3 import client +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_REGION_EU_WEST_1, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +mock_make_api_call = botocore.client.BaseClient._make_api_call + + +def mock_make_api_call_error(self, operation_name, kwarg): + if operation_name == "PutPublicAccessBlock": + raise botocore.exceptions.ClientError( + { + "Error": { + "Code": "InvalidPermission.NotFound", + "Message": "The specified rule does not exist in this security group.", + } + }, + operation_name, + ) + return mock_make_api_call(self, operation_name, kwarg) + + +class Test_cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer: + @mock_aws + def test_trail_bucket_public_acl(self): + aws_provider = set_mocked_aws_provider( + [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1] + ) + s3_client = client("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name_us = "bucket_test_us" + s3_client.create_bucket(Bucket=bucket_name_us) + s3_client.put_bucket_acl( + AccessControlPolicy={ + "Grants": [ + { + "Grantee": { + "DisplayName": "test", + "EmailAddress": "", + "ID": "test_ID", + "Type": "Group", + "URI": "http://acs.amazonaws.com/groups/global/AllUsers", + }, + "Permission": "READ", + }, + ], + "Owner": {"DisplayName": "test", "ID": "test_id"}, + }, + Bucket=bucket_name_us, + ) + + trail_name_us = "trail_test_us" + cloudtrail_client = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + cloudtrail_client.create_trail( + Name=trail_name_us, S3BucketName=bucket_name_us, IsMultiRegionTrail=False + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + from prowler.providers.aws.services.s3.s3_service import S3 + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), mock.patch( + "prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.s3_client", + new=S3(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer import ( + fixer, + ) + + assert fixer(trail_name_us, AWS_REGION_US_EAST_1) + + @mock_aws + def test_trail_bucket_public_acl_error(self): + with mock.patch( + "botocore.client.BaseClient._make_api_call", new=mock_make_api_call_error + ): + aws_provider = set_mocked_aws_provider( + [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1] + ) + s3_client = client("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name_us = "bucket_test_us" + s3_client.create_bucket(Bucket=bucket_name_us) + s3_client.put_bucket_acl( + AccessControlPolicy={ + "Grants": [ + { + "Grantee": { + "DisplayName": "test", + "EmailAddress": "", + "ID": "test_ID", + "Type": "Group", + "URI": "http://acs.amazonaws.com/groups/global/AllUsers", + }, + "Permission": "READ", + }, + ], + "Owner": {"DisplayName": "test", "ID": "test_id"}, + }, + Bucket=bucket_name_us, + ) + + trail_name_us = "trail_test_us" + cloudtrail_client = client("cloudtrail", region_name=AWS_REGION_US_EAST_1) + cloudtrail_client.create_trail( + Name=trail_name_us, + S3BucketName=bucket_name_us, + IsMultiRegionTrail=False, + ) + + from prowler.providers.aws.services.cloudtrail.cloudtrail_service import ( + Cloudtrail, + ) + from prowler.providers.aws.services.s3.s3_service import S3 + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.cloudtrail_client", + new=Cloudtrail(aws_provider), + ), mock.patch( + "prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.s3_client", + new=S3(aws_provider), + ): + # Test Check + from prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer import ( + fixer, + ) + + assert not fixer(trail_name_us, AWS_REGION_US_EAST_1)