Addition of s3control support in server mode (#4885)
This commit is contained in:
Cristopher Pinzón
GitHub
parent
e84cc20abe
commit
8b81481d3e
@ -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({})
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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"
|
||||
|
||||
Loading…
Reference in New Issue
Block a user