From d222e929aabc30b571af6e9f6ecdcf764c78dacd Mon Sep 17 00:00:00 2001
From: rafcio19 <therafcio9@gmail.com>
Date: Fri, 20 Oct 2023 11:43:47 +0200
Subject: [PATCH] S3: s3 access point reads / writes reach destination bucket
 (#6926)

---
 moto/s3/responses.py                      | 21 +++++++++-
 tests/test_s3control/test_s3control_s3.py | 47 +++++++++++++++++++++--
 2 files changed, 62 insertions(+), 6 deletions(-)

diff --git a/moto/s3/responses.py b/moto/s3/responses.py
index d8f0d286c..907bdce54 100644
--- a/moto/s3/responses.py
+++ b/moto/s3/responses.py
@@ -173,6 +173,10 @@ class S3Response(BaseResponse):
         else:
             return unquote(part)
 
+    @property
+    def is_access_point(self) -> bool:
+        return ".s3-accesspoint." in self.headers["host"]
+
     @property
     def backend(self) -> S3Backend:
         return s3_backends[self.current_account]["global"]
@@ -247,10 +251,23 @@ class S3Response(BaseResponse):
         return "delete" in qs
 
     def parse_bucket_name_from_url(self, request: Any, url: str) -> str:
+        bucket_name = ""
         if self.subdomain_based_buckets(request):
-            return bucket_name_from_url(url)  # type: ignore
+            bucket_name = bucket_name_from_url(url)  # type: ignore
         else:
-            return bucketpath_bucket_name_from_url(url)  # type: ignore
+            bucket_name = bucketpath_bucket_name_from_url(url)  # type: ignore
+
+        if self.is_access_point:
+            # import here to avoid circular dependency error
+            from moto.s3control import s3control_backends
+
+            ap_name = bucket_name[: -(len(self.current_account) + 1)]
+            ap = s3control_backends[self.current_account]["global"].get_access_point(
+                self.current_account, ap_name
+            )
+            bucket_name = ap.bucket
+
+        return bucket_name
 
     def parse_key_name(self, request: Any, url: str) -> str:
         if self.subdomain_based_buckets(request):
diff --git a/tests/test_s3control/test_s3control_s3.py b/tests/test_s3control/test_s3control_s3.py
index 7947496ac..2ea3d3afb 100644
--- a/tests/test_s3control/test_s3control_s3.py
+++ b/tests/test_s3control/test_s3control_s3.py
@@ -5,13 +5,16 @@ from moto import mock_s3, mock_s3control, settings
 from moto.core import DEFAULT_ACCOUNT_ID as ACCOUNT_ID
 
 
+REGION = "us-east-1"
+
+
 if not settings.TEST_SERVER_MODE:
 
     @mock_s3
     @mock_s3control
     def test_pab_are_kept_separate():
-        client = boto3.client("s3control", region_name="us-east-1")
-        s3_client = boto3.client("s3", region_name="us-east-1")
+        client = boto3.client("s3control", region_name=REGION)
+        s3_client = boto3.client("s3", region_name=REGION)
         s3_client.create_bucket(Bucket="bucket")
 
         client.put_public_access_block(
@@ -53,8 +56,8 @@ if not settings.TEST_SERVER_MODE:
     @mock_s3control
     @mock_s3
     def test_pab_are_kept_separate_with_inverse_mocks():
-        client = boto3.client("s3control", region_name="us-east-1")
-        s3_client = boto3.client("s3", region_name="us-east-1")
+        client = boto3.client("s3control", region_name=REGION)
+        s3_client = boto3.client("s3", region_name=REGION)
         s3_client.create_bucket(Bucket="bucket")
 
         client.put_public_access_block(
@@ -92,3 +95,39 @@ if not settings.TEST_SERVER_MODE:
             "BlockPublicPolicy": True,
             "RestrictPublicBuckets": False,
         }
+
+    @mock_s3
+    @mock_s3control
+    def test_access_point_read_write():
+        # Setup
+        bucket = "test-bucket"
+        ap_client = boto3.client("s3control", region_name=REGION)
+        s3_client = boto3.client("s3", region_name=REGION)
+        s3_client.create_bucket(Bucket=bucket)
+
+        read_ap = ap_client.create_access_point(
+            AccountId=ACCOUNT_ID, Name="read-ap", Bucket=bucket
+        )
+        write_ap = ap_client.create_access_point(
+            AccountId=ACCOUNT_ID, Name="write-ap", Bucket=bucket
+        )
+
+        content = b"This is test content"
+        key = "test/object.txt"
+
+        # Execute
+        s3_client.put_object(
+            Bucket=write_ap["AccessPointArn"],
+            Key=key,
+            Body=content,
+            ContentType="text/plain",
+        )
+
+        # Verify
+        assert (
+            s3_client.get_object(Bucket=read_ap["AccessPointArn"], Key=key)[
+                "Body"
+            ].read()
+            == content
+        )
+        assert s3_client.get_object(Bucket=bucket, Key=key)["Body"].read() == content