feat(s3): add new check s3_multi_region_access_point_public_access_block (#5552)

Co-authored-by: Sergio <sergio@prowler.com> Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>
2026-07-23 20:42:02 +00:00 · 2024-10-29 20:56:56 +01:00
parent 85777546e8
commit 82ec3e8779
6 changed files with 360 additions and 9 deletions
@@ -0,0 +1,32 @@
+{
+  "Provider": "aws",
+  "CheckID": "s3_multi_region_access_point_public_access_block",
+  "CheckTitle": "Block Public Access Settings enabled on Multi Region Access Points.",
+  "CheckType": [
+    "Software and Configuration Checks/Industry and Regulatory Standards/AWS Foundational Security Best Practices"
+  ],
+  "ServiceName": "s3",
+  "SubServiceName": "",
+  "ResourceIdTemplate": "arn:aws:s3:region:account-id:accesspoint/access-point-name",
+  "Severity": "high",
+  "ResourceType": "AwsS3AccessPoint",
+  "Description": "Ensures that public access is blocked on S3 Access Points.",
+  "Risk": "Leaving S3 multi region access points open to the public in AWS can lead to data exposure, breaches, compliance violations, unauthorized access, and data integrity issues.",
+  "RelatedUrl": "https://aws.amazon.com/es/getting-started/hands-on/getting-started-with-amazon-s3-multi-region-access-points/",
+  "Remediation": {
+    "Code": {
+      "CLI": "",
+      "NativeIaC": "",
+      "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/s3-controls.html#s3-24",
+      "Terraform": ""
+    },
+    "Recommendation": {
+      "Text": "Ensure S3 multi region access points are private by default, applying strict access controls, and regularly auditing permissions to prevent unauthorized public access.",
+      "Url": "https://docs.aws.amazon.com/AmazonS3/latest/userguide/multi-region-access-point-block-public-access.html"
+    }
+  },
+  "Categories": [],
+  "DependsOn": [],
+  "RelatedTo": [],
+  "Notes": ""
+}
@@ -0,0 +1,39 @@
+from prowler.lib.check.models import Check, Check_Report_AWS
+from prowler.providers.aws.services.s3.s3control_client import s3control_client
+
+
+class s3_multi_region_access_point_public_access_block(Check):
+    """Ensure that Multi Region Access Points have Public Access Block enabled.
+
+    This check is useful to ensure that Multi Region Access Points have Public Access Block enabled.
+    """
+
+    def execute(self):
+        """Execute the Multi Region Access Points have Public Access Block enabled check.
+
+        Iterates over all Multi Region Access Points and checks if they have Public Access Block enabled.
+
+        Returns:
+            List[Check_Report_AWS]: A list of reports for each Multi Region Access Point.
+        """
+        findings = []
+        for mr_access_point in s3control_client.multi_region_access_points.values():
+            report = Check_Report_AWS(self.metadata())
+            report.region = mr_access_point.region
+            report.resource_id = mr_access_point.name
+            report.resource_arn = mr_access_point.arn
+            report.status = "PASS"
+            report.status_extended = f"S3 Multi Region Access Point {mr_access_point.name} of buckets {', '.join(mr_access_point.buckets)} does have Public Access Block enabled."
+
+            if not (
+                mr_access_point.public_access_block.block_public_acls
+                and mr_access_point.public_access_block.ignore_public_acls
+                and mr_access_point.public_access_block.block_public_policy
+                and mr_access_point.public_access_block.restrict_public_buckets
+            ):
+                report.status = "FAIL"
+                report.status_extended = f"S3 Multi Region Access Point {mr_access_point.name} of buckets {', '.join(mr_access_point.buckets)} does not have Public Access Block enabled."
+
+            findings.append(report)
+
+        return findings
@@ -470,8 +470,11 @@ class S3Control(AWSService):
        super().__init__(__class__.__name__, provider)
        self.account_public_access_block = self._get_public_access_block()
        self.access_points = {}
+        self.multi_region_access_points = {}
        self.__threading_call__(self._list_access_points)
        self.__threading_call__(self._get_access_point, self.access_points.values())
+        if self.audited_partition == "aws":
+            self._list_multi_region_access_points()

    def _get_public_access_block(self):
        logger.info("S3 - Get account public access block...")
@@ -506,25 +509,59 @@ class S3Control(AWSService):
            )["AccessPointList"]
            for ap in list_access_points:
                self.access_points[ap["AccessPointArn"]] = AccessPoint(
+                    arn=ap["AccessPointArn"],
                    account_id=self.audited_account,
                    name=ap["Name"],
                    bucket=ap["Bucket"],
                    region=regional_client.region,
                )
-        except ClientError as error:
-            if error.response["Error"]["Code"] == "NoSuchMultiRegionAccessPoint":
-                logger.warning(
-                    f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
-                )
-            else:
-                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}"
            )

+    def _list_multi_region_access_points(self):
+        # NOTE: This function is restricted to the us-west-2 region due to AWS limitations on Multi-Region Access Points.
+        # For more details on region restrictions, see the AWS documentation:
+        # https://docs.aws.amazon.com/AmazonS3/latest/userguide/MultiRegionAccessPointRestrictions.html
+        logger.info("S3 - Listing account multi region access points...")
+        try:
+            region = "us-west-2"
+            client = self.session.client(self.service, region)
+            list_multi_region_access_points = client.list_multi_region_access_points(
+                AccountId=self.audited_account
+            ).get("AccessPoints", [])
+            for mr_access_point in list_multi_region_access_points:
+                mr_ap_arn = f"arn:{self.audited_partition}:s3::{self.audited_account}:accesspoint/{mr_access_point['Name']}"
+                bucket_list = []
+                for mrap_region in mr_access_point.get("Regions", []):
+                    bucket_list.append(mrap_region.get("Bucket", ""))
+                self.multi_region_access_points[mr_ap_arn] = MultiRegionAccessPoint(
+                    arn=mr_ap_arn,
+                    account_id=self.audited_account,
+                    name=mr_access_point["Name"],
+                    buckets=bucket_list,
+                    region=region,
+                    public_access_block=PublicAccessBlock(
+                        block_public_acls=mr_access_point.get(
+                            "PublicAccessBlock", {}
+                        ).get("BlockPublicAcls", False),
+                        ignore_public_acls=mr_access_point.get(
+                            "PublicAccessBlock", {}
+                        ).get("IgnorePublicAcls", False),
+                        block_public_policy=mr_access_point.get(
+                            "PublicAccessBlock", {}
+                        ).get("BlockPublicPolicy", False),
+                        restrict_public_buckets=mr_access_point.get(
+                            "PublicAccessBlock", {}
+                        ).get("RestrictPublicBuckets", False),
+                    ),
+                )
+        except Exception as error:
+            logger.error(
+                f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
    def _get_access_point(self, ap):
        logger.info("S3 - Get account access point...")
        try:
@@ -567,6 +604,7 @@ class PublicAccessBlock(BaseModel):


 class AccessPoint(BaseModel):
+    arn: str
    account_id: str
    name: str
    bucket: str
@@ -574,6 +612,15 @@ class AccessPoint(BaseModel):
    region: str


+class MultiRegionAccessPoint(BaseModel):
+    arn: str
+    account_id: str
+    name: str
+    buckets: list[str] = []
+    public_access_block: Optional[PublicAccessBlock]
+    region: str
+
+
 class LifeCycleRule(BaseModel):
    id: str
    status: str
@@ -74,6 +74,7 @@ class Test_s3_access_point_public_access_block:
            s3control_client = mock.MagicMock()
            s3control_client.access_points = {
                arn_us: AccessPoint(
+                    arn=arn_us,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_us,
                    bucket=bucket_name_us,
@@ -86,6 +87,7 @@ class Test_s3_access_point_public_access_block:
                    ),
                ),
                arn_eu: AccessPoint(
+                    arn=arn_eu,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_eu,
                    bucket=bucket_name_eu,
@@ -165,6 +167,7 @@ class Test_s3_access_point_public_access_block:
            s3control_client = mock.MagicMock()
            s3control_client.access_points = {
                arn_us: AccessPoint(
+                    arn=arn_us,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_us,
                    bucket=bucket_name_us,
@@ -177,6 +180,7 @@ class Test_s3_access_point_public_access_block:
                    ),
                ),
                arn_eu: AccessPoint(
+                    arn=arn_eu,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_eu,
                    bucket=bucket_name_eu,
@@ -256,6 +260,7 @@ class Test_s3_access_point_public_access_block:
            s3control_client = mock.MagicMock()
            s3control_client.access_points = {
                arn_us: AccessPoint(
+                    arn=arn_us,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_us,
                    bucket=bucket_name_us,
@@ -268,6 +273,7 @@ class Test_s3_access_point_public_access_block:
                    ),
                ),
                arn_eu: AccessPoint(
+                    arn=arn_eu,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_eu,
                    bucket=bucket_name_eu,
@@ -346,6 +352,7 @@ class Test_s3_access_point_public_access_block:
            s3control_client = mock.MagicMock()
            s3control_client.access_points = {
                arn_us: AccessPoint(
+                    arn=arn_us,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_us,
                    bucket="bucket-us",
@@ -358,6 +365,7 @@ class Test_s3_access_point_public_access_block:
                    ),
                ),
                arn_eu: AccessPoint(
+                    arn=arn_eu,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_eu,
                    bucket="bucket-eu",
@@ -370,6 +378,7 @@ class Test_s3_access_point_public_access_block:
                    ),
                ),
                arn_ap: AccessPoint(
+                    arn=arn_ap,
                    account_id=AWS_ACCOUNT_NUMBER,
                    name=ap_name_ap,
                    bucket="bucket-ap",
@@ -0,0 +1,224 @@
+from unittest import mock
+from unittest.mock import patch
+
+import botocore
+from moto import mock_aws
+
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_WEST_2,
+    set_mocked_aws_provider,
+)
+
+MRAP_NAME = "test-mrap"
+BUCKET_NAME = "test-bucket"
+
+# Original botocore _make_api_call function
+orig = botocore.client.BaseClient._make_api_call
+
+
+# Mocked botocore _make_api_call function
+def mock_make_api_call_pab_enabled(self, operation_name, kwarg):
+    if operation_name == "ListMultiRegionAccessPoints":
+        return {
+            "AccessPoints": [
+                {
+                    "Name": MRAP_NAME,
+                    "Regions": [
+                        {
+                            "Bucket": BUCKET_NAME,
+                            "Region": "us-west-2",
+                        }
+                    ],
+                    "PublicAccessBlock": {
+                        "BlockPublicAcls": True,
+                        "IgnorePublicAcls": True,
+                        "BlockPublicPolicy": True,
+                        "RestrictPublicBuckets": True,
+                    },
+                }
+            ]
+        }
+    # If we don't want to patch the API call
+    return orig(self, operation_name, kwarg)
+
+
+def mock_make_api_call_pab_disabled(self, operation_name, kwarg):
+    if operation_name == "ListMultiRegionAccessPoints":
+        return {
+            "AccessPoints": [
+                {
+                    "Name": MRAP_NAME,
+                    "Regions": [
+                        {
+                            "Bucket": BUCKET_NAME,
+                            "Region": "us-west-2",
+                        }
+                    ],
+                    "PublicAccessBlock": {
+                        "BlockPublicAcls": False,
+                        "IgnorePublicAcls": False,
+                        "BlockPublicPolicy": False,
+                        "RestrictPublicBuckets": False,
+                    },
+                }
+            ]
+        }
+    # If we don't want to patch the API call
+    return orig(self, operation_name, kwarg)
+
+
+def mock_make_api_call_pab_one_disabled(self, operation_name, kwarg):
+    if operation_name == "ListMultiRegionAccessPoints":
+        return {
+            "AccessPoints": [
+                {
+                    "Name": MRAP_NAME,
+                    "Regions": [
+                        {
+                            "Bucket": BUCKET_NAME,
+                            "Region": "us-west-2",
+                        }
+                    ],
+                    "PublicAccessBlock": {
+                        "BlockPublicAcls": False,
+                        "IgnorePublicAcls": True,
+                        "BlockPublicPolicy": True,
+                        "RestrictPublicBuckets": True,
+                    },
+                }
+            ]
+        }
+    # If we don't want to patch the API call
+    return orig(self, operation_name, kwarg)
+
+
+class Test_s3_multi_region_access_point_public_access_block:
+    @mock_aws
+    def test_no_multi_region_access_points(self):
+        from prowler.providers.aws.services.s3.s3_service import S3Control
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block.s3control_client",
+                new=S3Control(aws_provider),
+            ):
+                from prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block import (
+                    s3_multi_region_access_point_public_access_block,
+                )
+
+                check = s3_multi_region_access_point_public_access_block()
+                result = check.execute()
+
+                assert len(result) == 0
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call", new=mock_make_api_call_pab_enabled
+    )
+    def test_multi_region_access_points_with_public_access_block(self):
+        from prowler.providers.aws.services.s3.s3_service import S3Control
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block.s3control_client",
+                new=S3Control(aws_provider),
+            ):
+                from prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block import (
+                    s3_multi_region_access_point_public_access_block,
+                )
+
+                check = s3_multi_region_access_point_public_access_block()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "PASS"
+                assert (
+                    result[0].status_extended
+                    == f"S3 Multi Region Access Point {MRAP_NAME} of buckets {BUCKET_NAME} does have Public Access Block enabled."
+                )
+                assert result[0].resource_id == MRAP_NAME
+                assert (
+                    result[0].resource_arn
+                    == f"arn:{aws_provider.identity.partition}:s3::{AWS_ACCOUNT_NUMBER}:accesspoint/{MRAP_NAME}"
+                )
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call", new=mock_make_api_call_pab_disabled
+    )
+    def test_multi_region_access_points_without_public_access_block(self):
+        from prowler.providers.aws.services.s3.s3_service import S3Control
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block.s3control_client",
+                new=S3Control(aws_provider),
+            ):
+                from prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block import (
+                    s3_multi_region_access_point_public_access_block,
+                )
+
+                check = s3_multi_region_access_point_public_access_block()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"S3 Multi Region Access Point {MRAP_NAME} of buckets {BUCKET_NAME} does not have Public Access Block enabled."
+                )
+                assert result[0].resource_id == MRAP_NAME
+                assert (
+                    result[0].resource_arn
+                    == f"arn:{aws_provider.identity.partition}:s3::{AWS_ACCOUNT_NUMBER}:accesspoint/{MRAP_NAME}"
+                )
+
+    @patch(
+        "botocore.client.BaseClient._make_api_call",
+        new=mock_make_api_call_pab_one_disabled,
+    )
+    def test_multi_region_access_points_without_one_public_access_block(self):
+        from prowler.providers.aws.services.s3.s3_service import S3Control
+
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_WEST_2])
+
+        with mock.patch(
+            "prowler.providers.common.provider.Provider.get_global_provider",
+            return_value=aws_provider,
+        ):
+            with mock.patch(
+                "prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block.s3control_client",
+                new=S3Control(aws_provider),
+            ):
+                from prowler.providers.aws.services.s3.s3_multi_region_access_point_public_access_block.s3_multi_region_access_point_public_access_block import (
+                    s3_multi_region_access_point_public_access_block,
+                )
+
+                check = s3_multi_region_access_point_public_access_block()
+                result = check.execute()
+
+                assert len(result) == 1
+                assert result[0].status == "FAIL"
+                assert (
+                    result[0].status_extended
+                    == f"S3 Multi Region Access Point {MRAP_NAME} of buckets {BUCKET_NAME} does not have Public Access Block enabled."
+                )
+                assert result[0].resource_id == MRAP_NAME
+                assert (
+                    result[0].resource_arn
+                    == f"arn:{aws_provider.identity.partition}:s3::{AWS_ACCOUNT_NUMBER}:accesspoint/{MRAP_NAME}"
+                )