Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

S3: s3 access point reads / writes reach destination bucket (#6926)

Browse Source
This commit is contained in:
rafcio19 2023-10-20 11:43:47 +02:00 committed by GitHub
parent a329641b3a
commit d222e929aa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 62 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
moto/s3/responses.py
View File

@ -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):

47
tests/test_s3control/test_s3control_s3.py
View File

@ -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