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

STS - Handle AssumeRoleWithSAML as an unsigned request

Browse Source
This commit is contained in:
Bert Blommers 2020-04-22 12:18:27 +01:00
parent 88494c58f9
commit 50111929cc
1 changed files with 22 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
moto/server.py
View File

@ -1,6 +1,7 @@
from __future__ import unicode_literals from __future__ import unicode_literals
import argparse import argparse
import io
import json import json
import re import re
import sys import sys
@ -29,6 +30,7 @@ UNSIGNED_REQUESTS = {
"AWSCognitoIdentityService": ("cognito-identity", "us-east-1"), "AWSCognitoIdentityService": ("cognito-identity", "us-east-1"),
"AWSCognitoIdentityProviderService": ("cognito-idp", "us-east-1"), "AWSCognitoIdentityProviderService": ("cognito-idp", "us-east-1"),
} }
UNSIGNED_ACTIONS = {"AssumeRoleWithSAML": ("sts", "us-east-1")}
class DomainDispatcherApplication(object): class DomainDispatcherApplication(object):
@ -77,9 +79,13 @@ class DomainDispatcherApplication(object):
else: else:
# Unsigned request # Unsigned request
target = environ.get("HTTP_X_AMZ_TARGET") target = environ.get("HTTP_X_AMZ_TARGET")
action = self.get_action_from_body(environ)
if target: if target:
service, _ = target.split(".", 1) service, _ = target.split(".", 1)
service, region = UNSIGNED_REQUESTS.get(service, DEFAULT_SERVICE_REGION) service, region = UNSIGNED_REQUESTS.get(service, DEFAULT_SERVICE_REGION)
elif action and action in UNSIGNED_ACTIONS:
# See if we can match the Action to a known service
service, region = UNSIGNED_ACTIONS.get(action)
else: else:
# S3 is the last resort when the target is also unknown # S3 is the last resort when the target is also unknown
service, region = DEFAULT_SERVICE_REGION service, region = DEFAULT_SERVICE_REGION
@ -130,6 +136,22 @@ class DomainDispatcherApplication(object):
self.app_instances[backend] = app self.app_instances[backend] = app
return app return app
def get_action_from_body(self, environ):
body = None
try:
request_body_size = int(environ.get("CONTENT_LENGTH", 0))
if "wsgi.input" in environ:
body = environ["wsgi.input"].read(request_body_size).decode("utf-8")
body_dict = dict(x.split("=") for x in str(body).split("&"))
return body_dict["Action"]
except ValueError:
pass
finally:
if body:
# We've consumed the body = need to reset it
environ["wsgi.input"] = io.StringIO(body)
return None
def __call__(self, environ, start_response): def __call__(self, environ, start_response):
backend_app = self.get_application(environ) backend_app = self.get_application(environ)
return backend_app(environ, start_response) return backend_app(environ, start_response)