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

Merge pull request #2330 from spulec/host-parsing

Browse Source 
Host parsing
This commit is contained in:
Steve Pulec 2019-07-23 22:51:45 -05:00 committed by GitHub
commit 5dbdff7ca7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 25 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
moto/server.py
View File

@ -58,9 +58,7 @@ class DomainDispatcherApplication(object):
if re.match(url_base, 'http://%s' % host):
return backend_name
raise RuntimeError('Invalid host: "%s"' % host)
def infer_service_region(self, environ):
def infer_service_region_host(self, environ):
auth = environ.get('HTTP_AUTHORIZATION')
if auth:
# Signed request
@ -70,21 +68,35 @@ class DomainDispatcherApplication(object):
try:
credential_scope = auth.split(",")[0].split()[1]
_, _, region, service, _ = credential_scope.split("/")
return service, region
except ValueError:
# Signature format does not match, this is exceptional and we can't
# 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
return DEFAULT_SERVICE_REGION
service, region = DEFAULT_SERVICE_REGION
else:
# Unsigned request
target = environ.get('HTTP_X_AMZ_TARGET')
if target:
service, _ = target.split('.', 1)
return UNSIGNED_REQUESTS.get(service, DEFAULT_SERVICE_REGION)
# S3 is the last resort when the target is also unknown
return DEFAULT_SERVICE_REGION
service, region = UNSIGNED_REQUESTS.get(service, DEFAULT_SERVICE_REGION)
else:
# S3 is the last resort when the target is also unknown
service, region = DEFAULT_SERVICE_REGION
if service == 'dynamodb':
if environ['HTTP_X_AMZ_TARGET'].startswith('DynamoDBStreams'):
host = 'dynamodbstreams'
else:
dynamo_api_version = environ['HTTP_X_AMZ_TARGET'].split("_")[1].split(".")[0]
# If Newer API version, use dynamodb2
if dynamo_api_version > "20111205":
host = "dynamodb2"
else:
host = "{service}.{region}.amazonaws.com".format(
service=service, region=region)
return host
def get_application(self, environ):
path_info = environ.get('PATH_INFO', '')
@ -101,22 +113,14 @@ class DomainDispatcherApplication(object):
host = "instance_metadata"
else:
host = environ['HTTP_HOST'].split(':')[0]
if host in {'localhost', 'motoserver'} or host.startswith("192.168."):
service, region = self.infer_service_region(environ)
if service == 'dynamodb':
if environ['HTTP_X_AMZ_TARGET'].startswith('DynamoDBStreams'):
host = 'dynamodbstreams'
else:
dynamo_api_version = environ['HTTP_X_AMZ_TARGET'].split("_")[1].split(".")[0]
# If Newer API version, use dynamodb2
if dynamo_api_version > "20111205":
host = "dynamodb2"
else:
host = "{service}.{region}.amazonaws.com".format(
service=service, region=region)
with self.lock:
backend = self.get_backend_for_host(host)
if not backend:
# No regular backend found; try parsing other headers
host = self.infer_service_region_host(environ)
backend = self.get_backend_for_host(host)
app = self.app_instances.get(backend, None)
if app is None:
app = self.create_app(backend)

6
tests/test_core/test_server.py
View File

@ -38,12 +38,6 @@ def test_domain_dispatched():
keys[0].should.equal('EmailResponse.dispatch')
def test_domain_without_matches():
dispatcher = DomainDispatcherApplication(create_backend_app)
dispatcher.get_application.when.called_with(
{"HTTP_HOST": "not-matching-anything.com"}).should.throw(RuntimeError)
def test_domain_dispatched_with_service():
# If we pass a particular service, always return that.
dispatcher = DomainDispatcherApplication(create_backend_app, service="s3")