mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-24 13:01:56 +00:00
fix(s3): address review feedback on s3_bucket_object_public
Make the check opt-in via three audit_config keys (disabled by default), lower severity to low, and clarify that this is a spot-check sample. - prowler/config/config.yaml: add s3_bucket_object_public_enabled (False), _max_objects (100), _sample_size (3) with guidance pointing to s3_bucket_acl_prohibited for comprehensive coverage. - Check early-returns when disabled; otherwise reads max_objects and sample_size from audit_config. Uses Check_Report_AWS. Compares the AllUsers group URI exactly instead of substring-matching. PASS message discloses the sample size and recommends disabling ACLs via Object Ownership for full assurance. - metadata.json: Severity -> low; Description, Notes and Recommendation disclose the sampling limitation and cross-reference s3_bucket_acl_prohibited; RelatedTo links the two checks. - Unit tests cover: disabled-by-default (no findings), empty bucket PASS, private-objects PASS, public-object FAIL, AccessDenied MANUAL, other ClientError MANUAL.
This commit is contained in:
Synchx00
@@ -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"},
|
||||
|
||||
+10
-8
@@ -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."
|
||||
}
|
||||
|
||||
+54
-43
@@ -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)
|
||||
|
||||
|
||||
+244
@@ -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
|
||||
Reference in New Issue
Block a user