Repository: getmoto/moto
Pull request: S3: s3 access point reads / writes reach destination bucket
Resolves https://github.com/getmoto/moto/issues/6912

Problem statement:
s3.get_object() raises NoSuchBucket when using mocked Access Point alias or ARN
I'm trying to test an application that uses several S3 Access Points to access a common bucket and I want to verify it can do so correctly. I'm mocking both `s3control` and `s3` in my test configuration and using an S3ControlClient to create the access points for my bucket.

When I try to read an object from the Access Point, I get a `NoSuchBucket` error. I couldn't find anywhere in the s3control code where it interacts with the S3 backend to register the Access Point as an alias for the mocked bucket.

## Example code

### requirements.txt

```
# Using Python 3.9.16
boto3==1.26.90
botocore==1.29.90
mypy_boto3==1.26.90
boto3-stubs[s3,s3control]==1.26.90
botocore-stubs==1.29.90
moto==4.2.5
pytest==7.2.2
```

### test_access_point.py

```python
import pytest
import moto
import boto3
from mypy_boto3_s3 import S3Client
from mypy_boto3_s3control import S3ControlClient

@pytest.fixture(autouse=True)
def aws_credentials(monkeypatch):
    for key in ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_SECURITY_TOKEN", "AWS_SESSION_TOKEN"]:
        monkeypatch.setenv(key, "testing")


@pytest.fixture(autouse=True)
def mock_s3() -> S3Client:
    with moto.mock_s3():
        yield boto3.client("s3")


@pytest.fixture(autouse=True)
@pytest.mark.usefixtures("mock_s3")
def mock_s3control() -> S3ControlClient:
    with moto.mock_s3control():
        yield boto3.client("s3control")


@pytest.fixture
def s3_bucket(mock_s3: S3Client) -> str:
    bucket = "test-bucket"
    mock_s3.create_bucket(Bucket=bucket)
    return bucket


@pytest.fixture
def read_access_point(mock_s3control: S3ControlClient, s3_bucket: str) -> str:
    access_point = mock_s3control.create_access_point(AccountId="123456789012", Name="read-ap", Bucket=s3_bucket)
    return access_point["AccessPointArn"]


@pytest.fixture
def write_access_point(mock_s3control: S3ControlClient, s3_bucket: str) -> str:
    access_point = mock_s3control.create_access_point(AccountId="123456789012", Name="write-ap", Bucket=s3_bucket)
    return access_point["AccessPointArn"]


def test_s3_read_write(mock_s3: S3Client, s3_bucket: str):
    content = b"This is test content"
    key = "test/object.txt"

    mock_s3.put_object(Bucket=s3_bucket, Key=key, Body=content, ContentType="text/plain")
    assert mock_s3.get_object(Bucket=s3_bucket, Key=key)["Body"].read() == content


def test_access_point_read_write(mock_s3: S3Client, s3_bucket: str, read_access_point: str, write_access_point: str):
    content = b"This is test content"
    key = "test/object.txt"

    mock_s3.put_object(Bucket=write_access_point, Key=key, Body=content, ContentType="text/plain")
    assert mock_s3.get_object(Bucket=read_access_point, Key=key)["Body"].read() == content
    assert mock_s3.get_object(Bucket=s3_bucket, Key=key)["Body"].read() == content
```

## Expected Behavior

The expected behavior is the `test_access_point_read_write()` test passes. Creating an S3 Access Point with the S3 Control mock registers the alias with the S3 mock so it resolves those aliases to the underlying bucket.

## Actual Behavior

`s3.put_object()` and `s3.get_object()` raise `NoSuchBucket` when providing an S3 Access Point Alias or ARN as the `Bucket`.

Hints from issue discussion:
Hi @mrburrito, this doesn't seem to be supported yet in Moto - marking it as an enhancement.
@bblommers let me take a look

Commit message: feat: access points read and write thru access points

Patch to review:
diff --git a/moto/s3/models.py b/moto/s3/models.py
index 05899720501f..15a161d185be 100644
--- a/moto/s3/models.py
+++ b/moto/s3/models.py
@@ -1762,6 +1762,20 @@ def get_bucket(self, bucket_name: str) -> FakeBucket:
             account_id = s3_backends.bucket_accounts[bucket_name]
             return s3_backends[account_id]["global"].get_bucket(bucket_name)
 
+        # when access point is used the url that's passed is of following format:
+        # AccessPointName-AccountId.s3-accesspoint.*Region*.amazonaws.com
+        # only the first part is passed as bucket name: AccessPointName-AccountId
+        if bucket_name.endswith(f"-{self.account_id}"):
+
+            # import here to avoid circular dependency error
+            from moto.s3control import s3control_backends
+
+            ap_name = bucket_name[: -(len(self.account_id) + 1)]
+            ap = s3control_backends[self.account_id]["global"].get_access_point(
+                self.account_id, ap_name
+            )
+            return self.buckets[ap.bucket]
+
         raise MissingBucket(bucket=bucket_name)
 
     def head_bucket(self, bucket_name: str) -> FakeBucket:
diff --git a/tests/test_s3control/test_s3control_s3.py b/tests/test_s3control/test_s3control_s3.py
index 7947496aced7..2ea3d3afb292 100644
--- a/tests/test_s3control/test_s3control_s3.py
+++ b/tests/test_s3control/test_s3control_s3.py
@@ -5,13 +5,16 @@
 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 @@ def test_pab_are_kept_separate():
     @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 @@ def test_pab_are_kept_separate_with_inverse_mocks():
             "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


Code review comments:
diff_hunk: @@ -1762,6 +1762,20 @@ def get_bucket(self, bucket_name: str) -> FakeBucket:
             account_id = s3_backends.bucket_accounts[bucket_name]
             return s3_backends[account_id]["global"].get_bucket(bucket_name)
 
+        # when access point is used the url that's passed is of following format:
+        # AccessPointName-AccountId.s3-accesspoint.*Region*.amazonaws.com
+        # only the first part is passed as bucket name: AccessPointName-AccountId
+        if bucket_name.endswith(f"-{self.account_id}"):
path: moto/s3/models.py
text: @user1:
This could be a legitimate (non-ap) bucket name, so I'm a little hesitant about this part.

Can we use the `host` instead, to check if the URL contains `s3-accesspoint`?
That means we can't use this feature in ServerMode, as the host will always be `localhost` - but at least it doesn't break people's workflows.

@user1:
Or even better, is there some header that differentiates the two?
I think I looked at this before and couldn't find a difference between S3 and S3 AP requests, aside from the host, but maybe I missed something.

@author:
I had a look but there is no special header in the request to indicate access point call; `host` header it is.