From 8b81481d3eb4e970da19410d11fc1fe284cb4436 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Cristopher=20Pinz=C3=B3n?= Date: Thu, 24 Feb 2022 16:35:07 -0500 Subject: [PATCH] Addition of s3control support in server mode (#4885) --- moto/s3control/responses.py | 25 ++--- moto/server.py | 5 + tests/test_s3control/test_s3control.py | 137 ++++++++++++------------- 3 files changed, 81 insertions(+), 86 deletions(-) diff --git a/moto/s3control/responses.py b/moto/s3control/responses.py index ab8520a4c..a073e2268 100644 --- a/moto/s3control/responses.py +++ b/moto/s3control/responses.py @@ -13,37 +13,38 @@ class S3ControlResponse(BaseResponse): def public_access_block(cls, request, full_url, headers): response_instance = S3ControlResponse() try: - return response_instance._public_access_block(request, headers) + return response_instance._public_access_block(request) except S3ClientError as err: return err.code, {}, err.description @amzn_request_id - def _public_access_block(self, request, headers): + def _public_access_block(self, request): if request.method == "GET": - return self.get_public_access_block(headers) + return self.get_public_access_block(request) elif request.method == "PUT": - return self.put_public_access_block(request, headers) + return self.put_public_access_block(request) elif request.method == "DELETE": - return self.delete_public_access_block(headers) + return self.delete_public_access_block(request) - def get_public_access_block(self, headers): - account_id = headers["x-amz-account-id"] + def get_public_access_block(self, request): + account_id = request.headers.get("x-amz-account-id") public_block_config = s3control_backend.get_public_access_block( account_id=account_id, ) template = self.response_template(S3_PUBLIC_ACCESS_BLOCK_CONFIGURATION) return 200, {}, template.render(public_block_config=public_block_config) - def put_public_access_block(self, request, headers): - account_id = headers["x-amz-account-id"] - pab_config = self._parse_pab_config(request.body) + def put_public_access_block(self, request): + account_id = request.headers.get("x-amz-account-id") + data = request.body if hasattr(request, "body") else request.data + pab_config = self._parse_pab_config(data) s3control_backend.put_public_access_block( account_id, pab_config["PublicAccessBlockConfiguration"] ) return 201, {}, json.dumps({}) - def delete_public_access_block(self, headers): - account_id = headers["x-amz-account-id"] + def delete_public_access_block(self, request): + account_id = request.headers.get("x-amz-account-id") s3control_backend.delete_public_access_block(account_id=account_id,) return 204, {}, json.dumps({}) diff --git a/moto/server.py b/moto/server.py index e8c3da3ef..bcdb9b4ca 100644 --- a/moto/server.py +++ b/moto/server.py @@ -118,6 +118,7 @@ class DomainDispatcherApplication(object): # S3 is the last resort when the target is also unknown service, region = DEFAULT_SERVICE_REGION + path = environ.get("PATH_INFO", "") if service in ["budgets", "cloudfront"]: # Global Services - they do not have/expect a region host = f"{service}.amazonaws.com" @@ -145,6 +146,10 @@ class DomainDispatcherApplication(object): host = "ingest.{service}.{region}.amazonaws.com".format( service=service, region=region ) + elif service == "s3" and ( + path.startswith("/v20180820/") or "s3-control" in environ["HTTP_HOST"] + ): + host = "s3control" else: host = "{service}.{region}.amazonaws.com".format( service=service, region=region diff --git a/tests/test_s3control/test_s3control.py b/tests/test_s3control/test_s3control.py index 944b5c12c..3374c455e 100644 --- a/tests/test_s3control/test_s3control.py +++ b/tests/test_s3control/test_s3control.py @@ -4,87 +4,76 @@ import sure # noqa # pylint: disable=unused-import from boto3 import Session from botocore.client import ClientError -from moto import settings, mock_s3control +from moto import mock_s3control -# All tests for s3-control cannot be run under the server without a modification of the -# hosts file on your system. This is due to the fact that the URL to the host is in the form of: -# ACCOUNT_ID.s3-control.amazonaws.com <-- That Account ID part is the problem. If you want to -# make use of the moto server, update your hosts file for `THE_ACCOUNT_ID_FOR_MOTO.localhost` -# and this will work fine. -if not settings.TEST_SERVER_MODE: +@mock_s3control +def test_get_public_access_block_for_account(): + from moto.s3.models import ACCOUNT_ID - @mock_s3control - def test_get_public_access_block_for_account(): - from moto.s3.models import ACCOUNT_ID + client = boto3.client("s3control", region_name="us-west-2") - client = boto3.client("s3control", region_name="us-west-2") + # With an invalid account ID: + with pytest.raises(ClientError) as ce: + client.get_public_access_block(AccountId="111111111111") + assert ce.value.response["Error"]["Code"] == "AccessDenied" - # With an invalid account ID: - with pytest.raises(ClientError) as ce: - client.get_public_access_block(AccountId="111111111111") - assert ce.value.response["Error"]["Code"] == "AccessDenied" + # Without one defined: + with pytest.raises(ClientError) as ce: + client.get_public_access_block(AccountId=ACCOUNT_ID) + assert ce.value.response["Error"]["Code"] == "NoSuchPublicAccessBlockConfiguration" - # Without one defined: - with pytest.raises(ClientError) as ce: - client.get_public_access_block(AccountId=ACCOUNT_ID) - assert ( - ce.value.response["Error"]["Code"] == "NoSuchPublicAccessBlockConfiguration" - ) - - # Put a with an invalid account ID: - with pytest.raises(ClientError) as ce: - client.put_public_access_block( - AccountId="111111111111", - PublicAccessBlockConfiguration={"BlockPublicAcls": True}, - ) - assert ce.value.response["Error"]["Code"] == "AccessDenied" - - # Put with an invalid PAB: - with pytest.raises(ClientError) as ce: - client.put_public_access_block( - AccountId=ACCOUNT_ID, PublicAccessBlockConfiguration={} - ) - assert ce.value.response["Error"]["Code"] == "InvalidRequest" - assert ( - "Must specify at least one configuration." - in ce.value.response["Error"]["Message"] - ) - - # Correct PAB: + # Put a with an invalid account ID: + with pytest.raises(ClientError) as ce: client.put_public_access_block( - AccountId=ACCOUNT_ID, - PublicAccessBlockConfiguration={ - "BlockPublicAcls": True, - "IgnorePublicAcls": True, - "BlockPublicPolicy": True, - "RestrictPublicBuckets": True, - }, + AccountId="111111111111", + PublicAccessBlockConfiguration={"BlockPublicAcls": True}, ) + assert ce.value.response["Error"]["Code"] == "AccessDenied" - # Get the correct PAB (for all regions): - for region in Session().get_available_regions("s3control"): - region_client = boto3.client("s3control", region_name=region) - assert region_client.get_public_access_block(AccountId=ACCOUNT_ID)[ - "PublicAccessBlockConfiguration" - ] == { - "BlockPublicAcls": True, - "IgnorePublicAcls": True, - "BlockPublicPolicy": True, - "RestrictPublicBuckets": True, - } - - # Delete with an invalid account ID: - with pytest.raises(ClientError) as ce: - client.delete_public_access_block(AccountId="111111111111") - assert ce.value.response["Error"]["Code"] == "AccessDenied" - - # Delete successfully: - client.delete_public_access_block(AccountId=ACCOUNT_ID) - - # Confirm that it's deleted: - with pytest.raises(ClientError) as ce: - client.get_public_access_block(AccountId=ACCOUNT_ID) - assert ( - ce.value.response["Error"]["Code"] == "NoSuchPublicAccessBlockConfiguration" + # Put with an invalid PAB: + with pytest.raises(ClientError) as ce: + client.put_public_access_block( + AccountId=ACCOUNT_ID, PublicAccessBlockConfiguration={} ) + assert ce.value.response["Error"]["Code"] == "InvalidRequest" + assert ( + "Must specify at least one configuration." + in ce.value.response["Error"]["Message"] + ) + + # Correct PAB: + client.put_public_access_block( + AccountId=ACCOUNT_ID, + PublicAccessBlockConfiguration={ + "BlockPublicAcls": True, + "IgnorePublicAcls": True, + "BlockPublicPolicy": True, + "RestrictPublicBuckets": True, + }, + ) + + # Get the correct PAB (for all regions): + for region in Session().get_available_regions("s3control"): + region_client = boto3.client("s3control", region_name=region) + assert region_client.get_public_access_block(AccountId=ACCOUNT_ID)[ + "PublicAccessBlockConfiguration" + ] == { + "BlockPublicAcls": True, + "IgnorePublicAcls": True, + "BlockPublicPolicy": True, + "RestrictPublicBuckets": True, + } + + # Delete with an invalid account ID: + with pytest.raises(ClientError) as ce: + client.delete_public_access_block(AccountId="111111111111") + assert ce.value.response["Error"]["Code"] == "AccessDenied" + + # Delete successfully: + client.delete_public_access_block(AccountId=ACCOUNT_ID) + + # Confirm that it's deleted: + with pytest.raises(ClientError) as ce: + client.get_public_access_block(AccountId=ACCOUNT_ID) + assert ce.value.response["Error"]["Code"] == "NoSuchPublicAccessBlockConfiguration"