1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
|
from __future__ import unicode_literals
import argparse
import json
import re
import sys
from threading import Lock
import six
from flask import Flask
from flask.testing import FlaskClient
from six.moves.urllib.parse import urlencode
from werkzeug.routing import BaseConverter
from werkzeug.serving import run_simple
from moto.backends import BACKENDS
from moto.core.utils import convert_flask_to_httpretty_response
HTTP_METHODS = ["GET", "POST", "PUT", "DELETE", "HEAD", "PATCH"]
class DomainDispatcherApplication(object):
"""
Dispatch requests to different applications based on the "Host:" header
value. We'll match the host header value with the url_bases of each backend.
"""
def __init__(self, create_app, service=None):
self.create_app = create_app
self.lock = Lock()
self.app_instances = {}
self.service = service
def get_backend_for_host(self, host):
if host == 'moto_api':
return host
if self.service:
return self.service
if host in BACKENDS:
return host
for backend_name, backend in BACKENDS.items():
for url_base in list(backend.values())[0].url_bases:
if re.match(url_base, 'http://%s' % host):
return backend_name
raise RuntimeError('Invalid host: "%s"' % host)
def get_application(self, environ):
path_info = environ.get('PATH_INFO', '')
# The URL path might contain non-ASCII text, for instance unicode S3 bucket names
if six.PY2 and isinstance(path_info, str):
path_info = six.u(path_info)
if six.PY3 and isinstance(path_info, six.binary_type):
path_info = path_info.decode('utf-8')
if path_info.startswith("/moto-api") or path_info == "/favicon.ico":
host = "moto_api"
elif path_info.startswith("/latest/meta-data/"):
host = "instance_metadata"
else:
host = environ['HTTP_HOST'].split(':')[0]
if host in {'localhost', 'motoserver'} or host.startswith("192.168."):
# Fall back to parsing auth header to find service
# ['Credential=sdffdsa', '20170220', 'us-east-1', 'sns', 'aws4_request']
try:
_, _, region, service, _ = environ['HTTP_AUTHORIZATION'].split(",")[0].split()[
1].split("/")
except (KeyError, ValueError):
# Some cognito-idp endpoints (e.g. change password) do not receive an auth header.
if environ.get('HTTP_X_AMZ_TARGET', '').startswith('AWSCognitoIdentityProviderService'):
service = 'cognito-idp'
else:
service = 's3'
region = 'us-east-1'
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)
app = self.app_instances.get(backend, None)
if app is None:
app = self.create_app(backend)
self.app_instances[backend] = app
return app
def __call__(self, environ, start_response):
backend_app = self.get_application(environ)
return backend_app(environ, start_response)
class RegexConverter(BaseConverter):
# http://werkzeug.pocoo.org/docs/routing/#custom-converters
def __init__(self, url_map, *items):
super(RegexConverter, self).__init__(url_map)
self.regex = items[0]
class AWSTestHelper(FlaskClient):
def action_data(self, action_name, **kwargs):
"""
Method calls resource with action_name and returns data of response.
"""
opts = {"Action": action_name}
opts.update(kwargs)
res = self.get("/?{0}".format(urlencode(opts)),
headers={"Host": "{0}.us-east-1.amazonaws.com".format(self.application.service)})
return res.data.decode("utf-8")
def action_json(self, action_name, **kwargs):
"""
Method calls resource with action_name and returns object obtained via
deserialization of output.
"""
return json.loads(self.action_data(action_name, **kwargs))
def create_backend_app(service):
from werkzeug.routing import Map
# Create the backend_app
backend_app = Flask(__name__)
backend_app.debug = True
backend_app.service = service
# Reset view functions to reset the app
backend_app.view_functions = {}
backend_app.url_map = Map()
backend_app.url_map.converters['regex'] = RegexConverter
backend = list(BACKENDS[service].values())[0]
for url_path, handler in backend.flask_paths.items():
if handler.__name__ == 'dispatch':
endpoint = '{0}.dispatch'.format(handler.__self__.__name__)
else:
endpoint = None
original_endpoint = endpoint
index = 2
while endpoint in backend_app.view_functions:
# HACK: Sometimes we map the same view to multiple url_paths. Flask
# requries us to have different names.
endpoint = original_endpoint + str(index)
index += 1
backend_app.add_url_rule(
url_path,
endpoint=endpoint,
methods=HTTP_METHODS,
view_func=convert_flask_to_httpretty_response(handler),
strict_slashes=False,
)
backend_app.test_client_class = AWSTestHelper
return backend_app
def main(argv=sys.argv[1:]):
parser = argparse.ArgumentParser()
# Keep this for backwards compat
parser.add_argument(
"service",
type=str,
nargs='?', # http://stackoverflow.com/a/4480202/731592
default=None)
parser.add_argument(
'-H', '--host', type=str,
help='Which host to bind',
default='127.0.0.1')
parser.add_argument(
'-p', '--port', type=int,
help='Port number to use for connection',
default=5000)
parser.add_argument(
'-r', '--reload',
action='store_true',
help='Reload server on a file change',
default=False
)
parser.add_argument(
'-s', '--ssl',
action='store_true',
help='Enable SSL encrypted connection with auto-generated certificate (use https://... URL)',
default=False
)
parser.add_argument(
'-c', '--ssl-cert', type=str,
help='Path to SSL certificate',
default=None)
parser.add_argument(
'-k', '--ssl-key', type=str,
help='Path to SSL private key',
default=None)
args = parser.parse_args(argv)
# Wrap the main application
main_app = DomainDispatcherApplication(
create_backend_app, service=args.service)
main_app.debug = True
ssl_context = None
if args.ssl_key and args.ssl_cert:
ssl_context = (args.ssl_cert, args.ssl_key)
elif args.ssl:
ssl_context = 'adhoc'
run_simple(args.host, args.port, main_app,
threaded=True, use_reloader=args.reload,
ssl_context=ssl_context)
if __name__ == '__main__':
main()
|
