diff --git a/botocore/handlers.py b/botocore/handlers.py index 4c121620c0..418381afe0 100644 --- a/botocore/handlers.py +++ b/botocore/handlers.py @@ -36,7 +36,12 @@ logger = logging.getLogger(__name__) -LabelRE = re.compile('[a-z0-9][a-z0-9\-]*[a-z0-9]') +LABEL_RE = re.compile('[a-z0-9][a-z0-9\-]*[a-z0-9]') +RESTRICTED_REGIONS = [ + 'us-gov-west-1', + 'fips-gov-west-1', +] + def decode_console_output(event_name, shape, value, **kwargs): @@ -91,7 +96,7 @@ def check_dns_name(bucket_name): if n == 1: if not bucket_name.isalnum(): return False - match = LabelRE.match(bucket_name) + match = LABEL_RE.match(bucket_name) if match is None or match.end() != len(bucket_name): return False return True @@ -114,7 +119,7 @@ def fix_s3_host(event_name, endpoint, request, auth, **kwargs): bucket_name = path_parts[1] logger.debug('Checking for DNS compatible bucket for: %s', request.url) - if check_dns_name(bucket_name): + if check_dns_name(bucket_name) and _allowed_region(endpoint.region_name): # If the operation is on a bucket, the auth_path must be # terminated with a '/' character. if len(path_parts) == 2: @@ -132,6 +137,10 @@ def fix_s3_host(event_name, endpoint, request, auth, **kwargs): bucket_name) +def _allowed_region(region_name): + return region_name not in RESTRICTED_REGIONS + + def register_retries_for_service(service, **kwargs): if not hasattr(service, 'retry'): return diff --git a/tests/unit/test_s3_addressing.py b/tests/unit/test_s3_addressing.py index 26be381723..cee2b7cdba 100644 --- a/tests/unit/test_s3_addressing.py +++ b/tests/unit/test_s3_addressing.py @@ -70,6 +70,15 @@ def test_list_objects_dns_name_non_classic(self): self.assertEqual(prepared_request.url, 'https://safename.s3.amazonaws.com/') + def test_list_objects_in_restricted_regions(self): + self.endpoint = self.s3.get_endpoint('us-gov-west-1') + op = self.s3.get_operation('ListObjects') + params = op.build_parameters(bucket='safename') + prepared_request = self.get_prepared_request(op, params) + # Note how we keep the region specific endpoint here. + self.assertEqual(prepared_request.url, + 'https://s3-us-gov-west-1.amazonaws.com/safename') + def test_list_objects_non_dns_name_non_classic(self): self.endpoint = self.s3.get_endpoint('us-west-2') op = self.s3.get_operation('ListObjects')