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

MotoServer: skip using HTTP_HOST header when determining request origin (#7225)

Browse Source
This commit is contained in:
Bert Blommers 2024-01-19 20:25:17 +00:00
parent 8199a88446
commit 09c5751671
2 changed files with 17 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
moto/moto_server/werkzeug_app.py
View File

@ -63,7 +63,9 @@ class DomainDispatcherApplication:
self.app_instances: Dict[str, Flask] = {}
self.backend_url_patterns = backend_index.backend_url_patterns
def get_backend_for_host(self, host: str) -> Any:
def get_backend_for_host(self, host: Optional[str]) -> Any:
if host is None:
return None
if host == "moto_api":
return host
@ -82,8 +84,8 @@ class DomainDispatcherApplication:
)
def infer_service_region_host(
self, body: Optional[str], environ: Dict[str, Any]
) -> str:
self, environ: Dict[str, Any], path: str
) -> Optional[str]:
auth = environ.get("HTTP_AUTHORIZATION")
target = environ.get("HTTP_X_AMZ_TARGET")
service = None
@ -109,25 +111,24 @@ class DomainDispatcherApplication:
# infer a service-region. A reduced set of services still use
# the deprecated SigV2, ergo prefer S3 as most likely default.
# https://docs.aws.amazon.com/general/latest/gr/signature-version-2.html
service, region = DEFAULT_SERVICE_REGION
return None
else:
# Unsigned request
body = self._get_body(environ)
action = self.get_action_from_body(body)
if target:
service, _ = target.split(".", 1)
service, region = UNSIGNED_REQUESTS.get(service, DEFAULT_SERVICE_REGION)
service, region = UNSIGNED_REQUESTS.get(service, (None, None))
elif action and action in UNSIGNED_ACTIONS:
# See if we can match the Action to a known service
service, region = UNSIGNED_ACTIONS[action]
if not service:
service, region = self.get_service_from_body(body, environ)
if not service:
service, region = self.get_service_from_path(environ)
service, region = self.get_service_from_path(path)
if not service:
# S3 is the last resort when the target is also unknown
service, region = DEFAULT_SERVICE_REGION
return None
path = environ.get("PATH_INFO", "")
if service in ["budgets", "cloudfront"]:
# Global Services - they do not have/expect a region
host = f"{service}.amazonaws.com"
@ -177,14 +178,16 @@ class DomainDispatcherApplication:
elif path_info.startswith("/latest/meta-data/"):
host = "instance_metadata"
else:
host = environ["HTTP_HOST"].split(":")[0]
host = None
with self.lock:
backend = self.get_backend_for_host(host)
if not backend:
# No regular backend found; try parsing body/other headers
body = self._get_body(environ)
host = self.infer_service_region_host(body, environ)
host = self.infer_service_region_host(environ, path_info)
if host is None:
service, region = DEFAULT_SERVICE_REGION
host = f"{service}.{region}.amazonaws.com"
backend = self.get_backend_for_host(host)
app = self.app_instances.get(backend, None)
@ -239,12 +242,11 @@ class DomainDispatcherApplication:
return None
def get_service_from_path(
self, environ: Dict[str, Any]
self, path_info: str
) -> Tuple[Optional[str], Optional[str]]:
# Moto sometimes needs to send a HTTP request to itself
# In which case it will send a request to 'http://localhost/service_region/whatever'
try:
path_info = environ.get("PATH_INFO", "/")
service, region = path_info[1 : path_info.index("/", 1)].split("_")
return service, region
except (AttributeError, KeyError, ValueError):

2
tests/test_dynamodb_v20111205/test_servermode.py
View File

@ -18,7 +18,7 @@ def test_table_list():
raise SkipTest("Only run test with external server")
headers = {
"X-Amz-Target": "DynamoDB_20111205.ListTables",
"Host": "dynamodb.us-east-1.amazonaws.com",
"AUTHORIZATION": "AWS4-HMAC-SHA256 Credential=ACCESS_KEY/20220226/us-east-1/dynamodb/aws4_request, SignedHeaders=content-type;host;x-amz-date;x-amz-target, Signature=sig",
}
requests.post(settings.test_server_mode_endpoint() + "/moto-api/reset")
res = requests.get(settings.test_server_mode_endpoint(), headers=headers)