diff --git a/prowler/config/config.yaml b/prowler/config/config.yaml index b9c9a5686c..7079735af8 100644 --- a/prowler/config/config.yaml +++ b/prowler/config/config.yaml @@ -386,6 +386,18 @@ aws: # Minimum retention period in hours for Kinesis streams min_kinesis_stream_retention_hours: 168 # 7 days + # AWS S3 Configuration + # aws.s3_bucket_object_public + # This check performs a spot-check by sampling object ACLs within a bucket, so + # it is disabled by default. For complete coverage, rely on s3_bucket_acl_prohibited + # which enforces BucketOwnerEnforced Object Ownership (AWS's recommended approach). + # Set s3_bucket_object_public_enabled to True to opt in. + s3_bucket_object_public_enabled: False + # Maximum number of objects to list per bucket (upper bound for the sampling pool) + s3_bucket_object_public_max_objects: 100 + # Number of objects to randomly sample from the listed pool and inspect ACLs for + s3_bucket_object_public_sample_size: 3 + # Detect Secrets plugin configuration detect_secrets_plugins: [ {"name": "ArtifactoryDetector"}, diff --git a/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.metadata.json b/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.metadata.json index a1f7edf3dd..199183f46d 100644 --- a/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.metadata.json +++ b/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.metadata.json @@ -1,17 +1,17 @@ { "Provider": "aws", "CheckID": "s3_bucket_object_public", - "CheckTitle": "Ensure that S3 buckets do not have public objects", + "CheckTitle": "Spot-check S3 bucket objects for public ACLs", "CheckType": [ "Data Protection" ], "ServiceName": "s3", "SubServiceName": "", "ResourceIdTemplate": "arn:partition:s3:::resource", - "Severity": "medium", + "Severity": "low", "ResourceType": "AwsS3Bucket", - "Description": "This check verifies if S3 buckets have public objects by random sampling.", - "Risk": "Public objects can be accessed by anyone on the internet, potentially leaking sensitive data.", + "Description": "Spot-checks a configurable random sample of objects in each S3 bucket and flags any whose ACL grants read access to the AllUsers group. This is a sampling-based check, not a comprehensive audit: ListObjectsV2 returns keys in lexicographical order and only a small sample is inspected, so public objects outside the sample can be missed. The check is disabled by default and must be opted into via the s3_bucket_object_public_enabled configuration flag.", + "Risk": "Public objects can be accessed by anyone on the internet, potentially leaking sensitive data. A bucket can appear private at the bucket-policy level while still containing individual objects with public ACL grants.", "RelatedUrl": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-control-block-public-access.html", "Remediation": { "Code": { @@ -21,14 +21,16 @@ "Terraform": "" }, "Recommendation": { - "Text": "Ensure that your S3 objects are not public. You can block public access at the bucket or account level to prevent this.", - "Url": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-control-block-public-access.html" + "Text": "For complete coverage, enable the s3_bucket_acl_prohibited check, which enforces the BucketOwnerEnforced Object Ownership setting (AWS's recommended approach since April 2023) and prevents public object ACLs entirely. Use this spot-check as a supplementary tool for manual assessments.", + "Url": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/about-object-ownership.html" } }, "Categories": [ "public-access" ], "DependsOn": [], - "RelatedTo": [], - "Notes": "This check samples 3 objects per bucket to verify public access." + "RelatedTo": [ + "s3_bucket_acl_prohibited" + ], + "Notes": "Disabled by default. Configure s3_bucket_object_public_enabled, s3_bucket_object_public_max_objects, and s3_bucket_object_public_sample_size in the Prowler configuration. Because ListObjectsV2 returns keys lexicographically and only a sample is inspected, a PASS does not guarantee the bucket is free of public objects; use s3_bucket_acl_prohibited for full assurance." } diff --git a/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.py b/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.py index e3fd5058cc..b0d2603230 100644 --- a/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.py +++ b/prowler/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public.py @@ -2,75 +2,86 @@ import random from botocore.exceptions import ClientError -from prowler.lib.check.models import Check, Check_Report +from prowler.lib.check.models import Check, Check_Report_AWS from prowler.providers.aws.services.s3.s3_client import s3_client +ALL_USERS_URI = "http://acs.amazonaws.com/groups/global/AllUsers" + class s3_bucket_object_public(Check): def execute(self): findings = [] + if not s3_client.audit_config.get("s3_bucket_object_public_enabled", False): + return findings + + max_objects = s3_client.audit_config.get( + "s3_bucket_object_public_max_objects", 100 + ) + sample_size = s3_client.audit_config.get( + "s3_bucket_object_public_sample_size", 3 + ) + for bucket in s3_client.buckets.values(): - report = Check_Report(self.metadata(), bucket) - - report.resource_id = bucket.name - report.resource_arn = bucket.arn - report.region = bucket.region - report.resource_tags = bucket.tags - - report.status = "PASS" - report.status_extended = ( - f"No public objects found in bucket {bucket.name} (sampled)." - ) + report = Check_Report_AWS(metadata=self.metadata(), resource=bucket) try: regional_client = s3_client.regional_clients[bucket.region] objects = regional_client.list_objects_v2( - Bucket=bucket.name, MaxKeys=100 + Bucket=bucket.name, MaxKeys=max_objects ) - if "Contents" in objects: - all_keys = [obj["Key"] for obj in objects["Contents"]] - sample_keys = random.sample(all_keys, min(len(all_keys), 3)) + contents = objects.get("Contents", []) + if not contents: + report.status = "PASS" + report.status_extended = f"S3 Bucket {bucket.name} is empty." + findings.append(report) + continue - public_objects_found = [] + all_keys = [obj["Key"] for obj in contents] + sample_keys = random.sample(all_keys, min(len(all_keys), sample_size)) + sampled = len(sample_keys) - for key in sample_keys: - acl = regional_client.get_object_acl( - Bucket=bucket.name, Key=key - ) - - for grant in acl.get("Grants", []): - grantee = grant.get("Grantee", {}) - if grantee.get( - "Type" - ) == "Group" and "AllUsers" in grantee.get("URI", ""): - public_objects_found.append(key) - break - - if public_objects_found: - report.status = "FAIL" - report.status_extended = f"S3 Bucket {bucket.name} contains public objects: {', '.join(public_objects_found)}." + public_objects_found = [] + for key in sample_keys: + acl = regional_client.get_object_acl(Bucket=bucket.name, Key=key) + for grant in acl.get("Grants", []): + grantee = grant.get("Grantee", {}) + if ( + grantee.get("Type") == "Group" + and grantee.get("URI") == ALL_USERS_URI + ): + public_objects_found.append(key) + break + if public_objects_found: + report.status = "FAIL" + report.status_extended = ( + f"S3 Bucket {bucket.name} has public objects detected in " + f"spot-check sample of {sampled} objects: " + f"{', '.join(public_objects_found)}." + ) else: - report.status_extended = f"Bucket {bucket.name} is empty." + report.status = "PASS" + report.status_extended = ( + f"No public objects detected in spot-check sample of " + f"{sampled} objects in bucket {bucket.name}. For complete " + f"assurance, ensure ACLs are disabled via Object Ownership " + f"settings." + ) except ClientError as error: + report.status = "MANUAL" if error.response["Error"]["Code"] == "AccessDenied": - report.status = "MANUAL" report.status_extended = ( - f"Access Denied when checking objects in bucket {bucket.name}." + f"Access Denied when spot-checking objects in bucket " + f"{bucket.name}." ) else: - report.status = "MANUAL" report.status_extended = ( - f"Could not check objects in bucket {bucket.name}: {error}" + f"Could not spot-check objects in bucket {bucket.name}: " + f"{error}." ) - except Exception as error: - report.status = "MANUAL" - report.status_extended = ( - f"An error occurred in bucket {bucket.name}: {error}" - ) findings.append(report) diff --git a/tests/providers/aws/services/s3/s3_bucket_object_public/__init__.py b/tests/providers/aws/services/s3/s3_bucket_object_public/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public_test.py b/tests/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public_test.py new file mode 100644 index 0000000000..b3aa2756f0 --- /dev/null +++ b/tests/providers/aws/services/s3/s3_bucket_object_public/s3_bucket_object_public_test.py @@ -0,0 +1,244 @@ +from unittest import mock + +from boto3 import client +from botocore.exceptions import ClientError +from moto import mock_aws + +from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider + +CHECK_MODULE = ( + "prowler.providers.aws.services.s3.s3_bucket_object_public.s3_bucket_object_public" +) + + +class Test_s3_bucket_object_public: + @mock_aws + def test_check_disabled_by_default_returns_no_findings(self): + s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1) + s3_client_us_east_1.create_bucket(Bucket="bucket-disabled") + + from prowler.providers.aws.services.s3.s3_service import S3 + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + s3_service = S3(aws_provider) + with mock.patch(f"{CHECK_MODULE}.s3_client", new=s3_service): + from prowler.providers.aws.services.s3.s3_bucket_object_public.s3_bucket_object_public import ( + s3_bucket_object_public, + ) + + check = s3_bucket_object_public() + result = check.execute() + + assert result == [] + + @mock_aws + def test_bucket_empty_passes(self): + s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name = "bucket-empty" + s3_client_us_east_1.create_bucket(Bucket=bucket_name) + + from prowler.providers.aws.services.s3.s3_service import S3 + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + s3_service = S3(aws_provider) + s3_service.audit_config = { + "s3_bucket_object_public_enabled": True, + "s3_bucket_object_public_max_objects": 100, + "s3_bucket_object_public_sample_size": 3, + } + with mock.patch(f"{CHECK_MODULE}.s3_client", new=s3_service): + from prowler.providers.aws.services.s3.s3_bucket_object_public.s3_bucket_object_public import ( + s3_bucket_object_public, + ) + + check = s3_bucket_object_public() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert result[0].status_extended == ( + f"S3 Bucket {bucket_name} is empty." + ) + assert result[0].resource_id == bucket_name + assert result[0].region == AWS_REGION_US_EAST_1 + + @mock_aws + def test_bucket_with_only_private_objects_passes(self): + s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name = "bucket-private-objects" + s3_client_us_east_1.create_bucket(Bucket=bucket_name) + s3_client_us_east_1.put_object( + Bucket=bucket_name, Key="private-1.txt", Body=b"x" + ) + s3_client_us_east_1.put_object( + Bucket=bucket_name, Key="private-2.txt", Body=b"x" + ) + + from prowler.providers.aws.services.s3.s3_service import S3 + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + s3_service = S3(aws_provider) + s3_service.audit_config = { + "s3_bucket_object_public_enabled": True, + "s3_bucket_object_public_max_objects": 100, + "s3_bucket_object_public_sample_size": 3, + } + with mock.patch(f"{CHECK_MODULE}.s3_client", new=s3_service): + from prowler.providers.aws.services.s3.s3_bucket_object_public.s3_bucket_object_public import ( + s3_bucket_object_public, + ) + + check = s3_bucket_object_public() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "PASS" + assert "No public objects detected in spot-check sample of" in ( + result[0].status_extended + ) + assert bucket_name in result[0].status_extended + assert ( + "For complete assurance, ensure ACLs are disabled via " + "Object Ownership settings." + ) in result[0].status_extended + + @mock_aws + def test_bucket_with_public_object_fails(self): + s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name = "bucket-public-object" + public_key = "public.txt" + s3_client_us_east_1.create_bucket(Bucket=bucket_name) + s3_client_us_east_1.put_object(Bucket=bucket_name, Key=public_key, Body=b"x") + s3_client_us_east_1.put_object_acl( + Bucket=bucket_name, Key=public_key, ACL="public-read" + ) + + from prowler.providers.aws.services.s3.s3_service import S3 + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + s3_service = S3(aws_provider) + s3_service.audit_config = { + "s3_bucket_object_public_enabled": True, + "s3_bucket_object_public_max_objects": 100, + "s3_bucket_object_public_sample_size": 3, + } + with mock.patch(f"{CHECK_MODULE}.s3_client", new=s3_service): + from prowler.providers.aws.services.s3.s3_bucket_object_public.s3_bucket_object_public import ( + s3_bucket_object_public, + ) + + check = s3_bucket_object_public() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "FAIL" + assert public_key in result[0].status_extended + assert ( + f"S3 Bucket {bucket_name} has public objects detected in " + "spot-check sample of" + ) in result[0].status_extended + + @mock_aws + def test_access_denied_on_list_objects_reports_manual(self): + s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name = "bucket-access-denied" + s3_client_us_east_1.create_bucket(Bucket=bucket_name) + + from prowler.providers.aws.services.s3.s3_service import S3 + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + s3_service = S3(aws_provider) + s3_service.audit_config = { + "s3_bucket_object_public_enabled": True, + "s3_bucket_object_public_max_objects": 100, + "s3_bucket_object_public_sample_size": 3, + } + + regional_client = mock.MagicMock() + regional_client.list_objects_v2.side_effect = ClientError( + {"Error": {"Code": "AccessDenied", "Message": "denied"}}, + "ListObjectsV2", + ) + s3_service.regional_clients[AWS_REGION_US_EAST_1] = regional_client + + with mock.patch(f"{CHECK_MODULE}.s3_client", new=s3_service): + from prowler.providers.aws.services.s3.s3_bucket_object_public.s3_bucket_object_public import ( + s3_bucket_object_public, + ) + + check = s3_bucket_object_public() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert result[0].status_extended == ( + f"Access Denied when spot-checking objects in bucket " + f"{bucket_name}." + ) + + @mock_aws + def test_other_client_error_reports_manual(self): + s3_client_us_east_1 = client("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name = "bucket-other-error" + s3_client_us_east_1.create_bucket(Bucket=bucket_name) + + from prowler.providers.aws.services.s3.s3_service import S3 + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ): + s3_service = S3(aws_provider) + s3_service.audit_config = { + "s3_bucket_object_public_enabled": True, + "s3_bucket_object_public_max_objects": 100, + "s3_bucket_object_public_sample_size": 3, + } + + regional_client = mock.MagicMock() + regional_client.list_objects_v2.side_effect = ClientError( + {"Error": {"Code": "InternalError", "Message": "boom"}}, + "ListObjectsV2", + ) + s3_service.regional_clients[AWS_REGION_US_EAST_1] = regional_client + + with mock.patch(f"{CHECK_MODULE}.s3_client", new=s3_service): + from prowler.providers.aws.services.s3.s3_bucket_object_public.s3_bucket_object_public import ( + s3_bucket_object_public, + ) + + check = s3_bucket_object_public() + result = check.execute() + + assert len(result) == 1 + assert result[0].status == "MANUAL" + assert ( + f"Could not spot-check objects in bucket {bucket_name}" + ) in result[0].status_extended