Extract region from auth headers for s3 bucket creation (#4983)

moto/core/utils.py

@@ -400,6 +400,14 @@ def aws_api_matches(pattern, string):
         return False
 
 
+def extract_region_from_aws_authorization(string):
+    auth = string or ""
+    region = re.sub(r".*Credential=[^/]+/[^/]+/([^/]+)/.*", r"\1", auth)
+    if region == auth:
+        return None
+    return region
+
+
 class BackendDict(dict):
     def __init__(self, fn, service_name):
         self.fn = fn

moto/s3/responses.py

@@ -4,7 +4,11 @@ import re
 from typing import List, Union
 
 from moto import settings
-from moto.core.utils import amzn_request_id, str_to_rfc_1123_datetime
+from moto.core.utils import (
+    amzn_request_id,
+    extract_region_from_aws_authorization,
+    str_to_rfc_1123_datetime,
+)
 from urllib.parse import parse_qs, urlparse, unquote, urlencode, urlunparse
 
 import xmltodict
@@ -278,7 +282,12 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
     def _bucket_response(self, request, full_url):
         querystring = self._get_querystring(full_url)
         method = request.method
-        region_name = parse_region_from_url(full_url)
+        region_name = parse_region_from_url(full_url, use_default_region=False)
+        if region_name is None:
+            region_name = extract_region_from_aws_authorization(
+                request.headers.get("Authorization", "")
+            )
+            region_name = region_name or DEFAULT_REGION_NAME
 
         bucket_name = self.parse_bucket_name_from_url(request, full_url)
         if not bucket_name:

moto/s3/utils.py

@@ -60,12 +60,12 @@ REGION_URL_REGEX = re.compile(
 )
 
 
-def parse_region_from_url(url):
+def parse_region_from_url(url, use_default_region=True):
     match = REGION_URL_REGEX.search(url)
     if match:
         region = match.group("region1") or match.group("region2")
     else:
-        region = "us-east-1"
+        region = "us-east-1" if use_default_region else None
     return region
 
 

tests/test_s3/test_s3.py

@@ -874,22 +874,17 @@ def test_bucket_location_nondefault():
     )
 
 
-# Test uses current Region to determine whether to throw an error
-# Region is retrieved based on current URL
-# URL will always be localhost in Server Mode, so can't run it there
-if not settings.TEST_SERVER_MODE:
+@mock_s3
+def test_s3_location_should_error_outside_useast1():
+    s3 = boto3.client("s3", region_name="eu-west-1")
 
-    @mock_s3
-    def test_s3_location_should_error_outside_useast1():
-        s3 = boto3.client("s3", region_name="eu-west-1")
+    bucket_name = "asdfasdfsdfdsfasda"
 
-        bucket_name = "asdfasdfsdfdsfasda"
-
-        with pytest.raises(ClientError) as e:
-            s3.create_bucket(Bucket=bucket_name)
-        e.value.response["Error"]["Message"].should.equal(
-            "The unspecified location constraint is incompatible for the region specific endpoint this request was sent to."
-        )
+    with pytest.raises(ClientError) as e:
+        s3.create_bucket(Bucket=bucket_name)
+    e.value.response["Error"]["Message"].should.equal(
+        "The unspecified location constraint is incompatible for the region specific endpoint this request was sent to."
+    )
 
 
 @mock_s3