From e006290228ed55b7e4e5375bca66304994524d07 Mon Sep 17 00:00:00 2001 From: stealthycoin <stealthycoin@users.noreply.github.com> Date: Mon, 28 Aug 2017 10:51:29 -0700 Subject: [PATCH] Add support for local custom authorizers BuiltinAuthorizer will now work in local mode. Authorizer types that cannot be run locally will now simply allow the requests to proceed as if there were no authorizer to allow for local testing. --- chalice/app.pyi | 5 +- chalice/cli/__init__.py | 2 +- chalice/cli/factory.py | 6 +- chalice/local.py | 520 ++++++++++++++++++++++++++++++++------- tests/unit/conftest.py | 50 ++++ tests/unit/test_app.py | 105 +++----- tests/unit/test_local.py | 510 +++++++++++++++++++++++++++++++++++++- 7 files changed, 1028 insertions(+), 170 deletions(-) diff --git a/chalice/app.pyi b/chalice/app.pyi index 5d589ef38a..45887a510a 100644 --- a/chalice/app.pyi +++ b/chalice/app.pyi @@ -1,4 +1,5 @@ from typing import Dict, List, Any, Callable, Union, Optional +from chalice.local import LambdaContext __version__ = ... # type: str @@ -109,9 +110,7 @@ class Chalice(object): api = ... # type: APIGateway routes = ... # type: Dict[str, Dict[str, RouteEntry]] current_request = ... # type: Request - # TODO: Change lambda_context to a real type once we have one for local - # API Gateway - lambda_context = ... # type: Any + lambda_context = ... # type: LambdaContext debug = ... # type: bool authorizers = ... # type: Dict[str, Dict[str, Any]] builtin_auth_handlers = ... # type: List[BuiltinAuthConfig] diff --git a/chalice/cli/__init__.py b/chalice/cli/__init__.py index 7e42c069d3..ea27b77b19 100644 --- a/chalice/cli/__init__.py +++ b/chalice/cli/__init__.py @@ -99,7 +99,7 @@ def run_local_server(factory, port, env): # The app-specific logger (app.log) will still continue # to work. logging.basicConfig(stream=sys.stdout) - server = factory.create_local_server(app_obj, port) + server = factory.create_local_server(app_obj, config, port) server.serve_forever() diff --git a/chalice/cli/factory.py b/chalice/cli/factory.py index 26cc1a12d0..06db382244 100644 --- a/chalice/cli/factory.py +++ b/chalice/cli/factory.py @@ -185,6 +185,6 @@ def load_project_config(self): with open(config_file) as f: return json.loads(f.read()) - def create_local_server(self, app_obj, port): - # type: (Chalice, int) -> local.LocalDevServer - return local.create_local_server(app_obj, port) + def create_local_server(self, app_obj, config, port): + # type: (Chalice, Config, int) -> local.LocalDevServer + return local.create_local_server(app_obj, config, port) diff --git a/chalice/local.py b/chalice/local.py index 7e0b04e4fc..2b72363561 100644 --- a/chalice/local.py +++ b/chalice/local.py @@ -4,27 +4,98 @@ """ from __future__ import print_function +import re +import time +import uuid import base64 import functools +import warnings from collections import namedtuple from six.moves.BaseHTTPServer import HTTPServer from six.moves.BaseHTTPServer import BaseHTTPRequestHandler -from typing import List, Any, Dict, Tuple, Callable # noqa +from typing import List, Any, Dict, Tuple, Callable, Optional, Union # noqa + +from chalice.app import Chalice # noqa +from chalice.app import CORSConfig # noqa +from chalice.app import ChaliceAuthorizer # noqa +from chalice.app import RouteEntry # noqa +from chalice.app import Request # noqa +from chalice.app import AuthResponse # noqa +from chalice.app import BuiltinAuthConfig # noqa +from chalice.config import Config # noqa -from chalice.app import Chalice, CORSConfig # noqa from chalice.compat import urlparse, parse_qs MatchResult = namedtuple('MatchResult', ['route', 'captured', 'query_params']) EventType = Dict[str, Any] +ContextType = Dict[str, Any] +HeaderType = Dict[str, Any] +ResponseType = Dict[str, Any] HandlerCls = Callable[..., 'ChaliceRequestHandler'] ServerCls = Callable[..., 'HTTPServer'] -def create_local_server(app_obj, port): - # type: (Chalice, int) -> LocalDevServer - return LocalDevServer(app_obj, port) +class Clock(object): + def time(self): + # type: () -> float + return time.time() + + +def create_local_server(app_obj, config, port): + # type: (Chalice, Config, int) -> LocalDevServer + return LocalDevServer(app_obj, config, port) + + +class LocalARNBuilder(object): + ARN_FORMAT = ('arn:aws:execute-api:{region}:{account_id}' + ':{api_id}/{stage}/{method}/{resource_path}') + LOCAL_REGION = 'mars-west-1' + LOCAL_ACCOUNT_ID = '123456789012' + LOCAL_API_ID = 'ymy8tbxw7b' + LOCAL_STAGE = 'api' + + def build_arn(self, method, path): + # type: (str, str) -> str + # In API Gateway the method and URI are separated by a / so typically + # the uri portion omits the leading /. In the case where the entire + # url is just '/' API Gateway adds a / to the end so that the arn end + # with a '//'. + if path != '/': + path = path[1:] + return self.ARN_FORMAT.format( + region=self.LOCAL_REGION, + account_id=self.LOCAL_ACCOUNT_ID, + api_id=self.LOCAL_API_ID, + stage=self.LOCAL_STAGE, + method=method, + resource_path=path + ) + + +class ARNMatcher(object): + def __init__(self, target_arn): + # type: (str) -> None + self._arn = target_arn + + def _resource_match(self, resource): + # type: (str) -> bool + # Arn matching supports two special case characetrs that are not + # escapable. * represents a glob which translates to a non-greedy + # match of any number of characters. ? which is any single character. + # These are easy to translate to a regex using .*? and . respectivly. + escaped_resource = re.escape(resource) + resource_regex = escaped_resource.replace(r'\?', '.').replace( + r'\*', '.*?') + return re.match(resource_regex, self._arn) is not None + + def does_any_resource_match(self, resources): + # type: (List[str]) -> bool + for resource in resources: + if self._resource_match(resource): + return True + return False class RouteMatcher(object): @@ -94,7 +165,7 @@ def create_lambda_event(self, method, path, headers, body=None): 'sourceIp': self.LOCAL_SOURCE_IP }, }, - 'headers': dict(headers), + 'headers': {k.lower(): v for k, v in headers.items()}, 'queryStringParameters': view_route.query_params, 'pathParameters': view_route.captured, 'stageVariables': {}, @@ -109,98 +180,288 @@ def create_lambda_event(self, method, path, headers, body=None): return event -class ChaliceRequestHandler(BaseHTTPRequestHandler): - - protocol = 'HTTP/1.1' +class LocalGatewayException(Exception): + CODE = 0 + + def __init__(self, headers, body=None): + # type: (HeaderType, Optional[bytes]) -> None + self.headers = headers + self.body = body + + +class ForbiddenError(LocalGatewayException): + CODE = 403 + + +class NotAuthorizedError(LocalGatewayException): + CODE = 401 + + +class NoOptionsRouteDefined(LocalGatewayException): + CODE = 403 + + +class LambdaContext(object): + def __init__(self, function_name, memory_size, + max_runtime_ms=3000, time_source=None): + # type: (str, int, int, Optional[Clock]) -> None + if time_source is None: + time_source = Clock() + self._time_source = time_source + self._start_time = self._current_time_millis() + self._max_runtime = max_runtime_ms + + # Below are properties that are found on the real LambdaContext passed + # by lambda and their associated documentation. + + # Name of the Lambda function that is executing. + self.function_name = function_name + + # The Lambda function version that is executing. If an alias is used + # to invoke the function, then function_version will be the version + # the alias points to. + # Chalice local obviously does not support versioning so it will always + # be set to $LATEST. + self.function_version = '$LATEST' + + # The ARN used to invoke this function. It can be function ARN or + # alias ARN. An unqualified ARN executes the $LATEST version and + # aliases execute the function version it is pointing to. + self.invoked_function_arn = '' + + # Memory limit, in MB, you configured for the Lambda function. You set + # the memory limit at the time you create a Lambda function and you + # can change it later. + self.memory_limit_in_mb = memory_size + + # AWS request ID associated with the request. This is the ID returned + # to the client that called the invoke method. + self.aws_request_id = str(uuid.uuid4()) + + # The name of the CloudWatch log group where you can find logs written + # by your Lambda function. + self.log_group_name = '' + + # The name of the CloudWatch log stream where you can find logs + # written by your Lambda function. The log stream may or may not + # change for each invocation of the Lambda function. + # + # The value is null if your Lambda function is unable to create a log + # stream, which can happen if the execution role that grants necessary + # permissions to the Lambda function does not include permissions for + # the CloudWatch Logs actions. + self.log_stream_name = '' + + # The last two attributes have the following comment in the + # documentation: + # Information about the client application and device when invoked + # through the AWS Mobile SDK, it can be null. + # Chalice local doens't need to set these since they are specifically + # for the mobile SDK. + self.identity = None + self.client_context = None + + def _current_time_millis(self): + # type: () -> float + return self._time_source.time() * 1000 + + def get_remaining_time_in_millis(self): + # type: () -> float + runtime = self._current_time_millis() - self._start_time + return self._max_runtime - runtime + + +class LocalGatewayAuthorizer(object): + """A class for running user defined authorizers in local mode.""" + def __init__(self, app_object): + # type: (Chalice) -> None + self._app_object = app_object + self._arn_builder = LocalARNBuilder() + + def authorize(self, lambda_event, lambda_context): + # type: (EventType, LambdaContext) -> Tuple[EventType, LambdaContext] + method = lambda_event['requestContext']['httpMethod'] + route_entry = self._route_for_event(lambda_event) + if not route_entry: + return lambda_event, lambda_context + authorizer = route_entry.authorizer + if not authorizer: + return lambda_event, lambda_context + if not isinstance(authorizer, ChaliceAuthorizer): + # Currently the only supported local authorizer is the + # BuiltinAuthConfig type. Anything else we will err on the side of + # allowing local testing by simply admiting the request. Otherwise + # there is no way for users to test their code in local mode. + warnings.warn( + '%s is not a supported in local mode. All requests made ' + 'against a route will be authorized to allow local testing.' + % authorizer.__class__.__name__ + ) + return lambda_event, lambda_context + path = lambda_event['requestContext']['resourcePath'] + arn = self._arn_builder.build_arn(method, path) + auth_event = self._prepare_authorizer_event(arn, lambda_event, + lambda_context) + auth_result = authorizer(auth_event, lambda_context) + authed = self._check_can_invoke_view_function(arn, auth_result) + if authed: + lambda_event = self._update_lambda_event(lambda_event, auth_result) + else: + raise ForbiddenError( + {'x-amzn-RequestId': lambda_context.aws_request_id, + 'x-amzn-ErrorType': 'AccessDeniedException'}, + (b'{"Message": ' + b'"User is not authorized to access this resource"}')) + return lambda_event, lambda_context + + def _check_can_invoke_view_function(self, arn, auth_result): + # type: (str, ResponseType) -> bool + policy = auth_result.get('policyDocument', {}) + statements = policy.get('Statement', []) + allow_resource_statements = [] + for statement in statements: + if statement.get('Effect') == 'Allow' and \ + statement.get('Action') == 'execute-api:Invoke': + for resource in statement.get('Resource'): + allow_resource_statements.append(resource) + + arn_matcher = ARNMatcher(arn) + return arn_matcher.does_any_resource_match(allow_resource_statements) + + def _route_for_event(self, lambda_event): + # type: (EventType) -> Optional[RouteEntry] + # Authorizer had to be made into an Any type since mypy couldn't + # detect that app.ChaliceAuthorizer was callable. + resource_path = lambda_event.get( + 'requestContext', {}).get('resourcePath') + http_method = lambda_event['requestContext']['httpMethod'] + try: + route_entry = self._app_object.routes[resource_path][http_method] + except KeyError: + # If a key error is raised when trying to get the route entry + # then this route does not support this method. A method error + # will be raised by the chalice handler method. We can ignore it + # here by returning no authorizer to avoid duplicating the logic. + return None + return route_entry + + def _update_lambda_event(self, lambda_event, auth_result): + # type: (EventType, ResponseType) -> EventType + auth_context = auth_result['context'] + auth_context.update({ + 'principalId': auth_result['principalId'] + }) + lambda_event['requestContext']['authorizer'] = auth_context + return lambda_event - def __init__(self, request, client_address, server, app_object): - # type: (bytes, Tuple[str, int], HTTPServer, Chalice) -> None - self.app_object = app_object + def _prepare_authorizer_event(self, arn, lambda_event, lambda_context): + # type: (str, EventType, LambdaContext) -> EventType + """Translate event for an authorizer input.""" + authorizer_event = lambda_event.copy() + authorizer_event['type'] = 'TOKEN' + try: + authorizer_event['authorizationToken'] = authorizer_event.get( + 'headers', {})['authorization'] + except KeyError: + raise NotAuthorizedError( + {'x-amzn-RequestId': lambda_context.aws_request_id, + 'x-amzn-ErrorType': 'UnauthorizedException'}, + b'{"message":"Unauthorized"}') + authorizer_event['methodArn'] = arn + return authorizer_event + + +class LocalGateway(object): + """A class for faking the behavior of API Gateway.""" + def __init__(self, app_object, config): + # type: (Chalice, Config) -> None + self._app_object = app_object + self._config = config self.event_converter = LambdaEventConverter( RouteMatcher(list(app_object.routes)), - self.app_object.api.binary_types + self._app_object.api.binary_types + ) + self._authorizer = LocalGatewayAuthorizer(app_object) + + def _generate_lambda_context(self): + # type: () -> LambdaContext + return LambdaContext( + function_name=self._config.function_name, + memory_size=self._config.lambda_memory_size, + runtime_millis=self._config.lambda_timeout ) - BaseHTTPRequestHandler.__init__( - self, request, client_address, server) # type: ignore - # Force BaseHTTPRequestHandler to use HTTP/1.1 - # Chrome ignores many headers from HTTP/1.0 servers. - BaseHTTPRequestHandler.protocol_version = "HTTP/1.1" - - def _generic_handle(self): - # type: () -> None - lambda_event = self._generate_lambda_event() - self._do_invoke_view_function(lambda_event) - - def _handle_binary(self, response): - # type: (Dict[str,Any]) -> Dict[str,Any] - if response.get('isBase64Encoded'): - body = base64.b64decode(response['body']) - response['body'] = body - return response - - def _do_invoke_view_function(self, lambda_event): - # type: (EventType) -> None - lambda_context = None - response = self.app_object(lambda_event, lambda_context) - response = self._handle_binary(response) - self._send_http_response(lambda_event, response) - - def _send_http_response(self, lambda_event, response): - # type: (EventType, Dict[str, Any]) -> None - self.send_response(response['statusCode']) - self.send_header('Content-Length', str(len(response['body']))) - content_type = response['headers'].pop( - 'Content-Type', 'application/json') - self.send_header('Content-Type', content_type) - headers = response['headers'] - for header in headers: - self.send_header(header, headers[header]) - self.end_headers() - body = response['body'] - if not isinstance(body, bytes): - body = body.encode('utf-8') - self.wfile.write(body) - def _generate_lambda_event(self): - # type: () -> EventType - content_length = int(self.headers.get('content-length', '0')) - body = None - if content_length > 0: - body = self.rfile.read(content_length) - # mypy doesn't like dict(self.headers) so I had to use a - # dictcomp instead to make it happy. - converted_headers = {key: value for key, value in self.headers.items()} + def _generate_lambda_event(self, method, path, headers, body): + # type: (str, str, HeaderType, str) -> EventType lambda_event = self.event_converter.create_lambda_event( - method=self.command, path=self.path, headers=converted_headers, + method=method, path=path, headers=headers, body=body, ) return lambda_event - do_GET = do_PUT = do_POST = do_HEAD = do_DELETE = do_PATCH = \ - _generic_handle - - def do_OPTIONS(self): - # type: () -> None - # This can either be because the user's provided an OPTIONS method - # *or* this is a preflight request, which chalice automatically - # sets up for you. - lambda_event = self._generate_lambda_event() - if self._has_user_defined_options_method(lambda_event): - self._do_invoke_view_function(lambda_event) - else: - # Otherwise this is a preflight request which we automatically - # generate. - self._send_autogen_options_response(lambda_event) - def _has_user_defined_options_method(self, lambda_event): # type: (EventType) -> bool route_key = lambda_event['requestContext']['resourcePath'] - return 'OPTIONS' in self.app_object.routes[route_key] + return 'OPTIONS' in self._app_object.routes[route_key] + + def handle_request(self, method, path, headers, body): + # type: (str, str, HeaderType, str) -> ResponseType + lambda_context = self._generate_lambda_context() + try: + lambda_event = self._generate_lambda_event( + method, path, headers, body) + except ValueError: + # API Gateway will return a different error on route not found + # depending on whether or not we have an authorization token in our + # request. Since we do not do that check until we actually find + # the authorizer that we will call we do not have that information + # available at this point. Instead we just check to see if that + # header is present and change our response if it is. This will + # need to be refactored later if we decide to more closely mirror + # how API Gateway does their auth and routing. + error_headers = {'x-amzn-RequestId': lambda_context.aws_request_id, + 'x-amzn-ErrorType': 'UnauthorizedException'} + auth_header = headers.get('authorization') + if auth_header is None: + auth_header = headers.get('Authorization') + if auth_header is not None: + raise ForbiddenError( + error_headers, + (b'{"message": "Authorization header requires ' + b'\'Credential\'' + b' parameter. Authorization header requires \'Signature\'' + b' parameter. Authorization header requires ' + b'\'SignedHeaders\' parameter. Authorization header ' + b'requires existence of either a \'X-Amz-Date\' or a' + b' \'Date\' header. Authorization=%s"}' + % auth_header.encode('ascii'))) + raise ForbiddenError( + error_headers, + b'{"message": "Missing Authentication Token"}') - def _send_autogen_options_response(self, lambda_event): - # type:(EventType) -> None + # This can either be because the user's provided an OPTIONS method + # *or* this is a preflight request, which chalice automatically + # responds to without invoking a user defined route. + if method == 'OPTIONS' and \ + not self._has_user_defined_options_method(lambda_event): + # No options route was defined for this path. API Gateway should + # automatically generate our CORS headers. + options_headers = self._autogen_options_headers(lambda_event) + raise NoOptionsRouteDefined(options_headers) + # The authorizer call will be a noop if there is no authorizer method + # defined for route. Otherwise it will raise a ForbiddenError + # which will be caught by the handler that called this and a 403 will + # be sent back over the wire. + lambda_event, lambda_context = self._authorizer.authorize( + lambda_event, lambda_context) + response = self._app_object(lambda_event, lambda_context) + response = self._handle_binary(response) + return response + + def _autogen_options_headers(self, lambda_event): + # type:(EventType) -> HeaderType route_key = lambda_event['requestContext']['resourcePath'] - route_dict = self.app_object.routes[route_key] + route_dict = self._app_object.routes[route_key] route_methods = list(route_dict.keys()) # Chalice ensures that routes with multiple views have the same @@ -214,7 +475,6 @@ def _send_autogen_options_response(self, lambda_event): # So our local version needs to add this manually to our set of allowed # headers. route_methods.append('OPTIONS') - route_methods = route_methods # The Access-Control-Allow-Methods header is not added by the # CORSConfig object it is added to the API Gateway route during @@ -222,21 +482,101 @@ def _send_autogen_options_response(self, lambda_event): cors_headers.update({ 'Access-Control-Allow-Methods': '%s' % ','.join(route_methods) }) + return cors_headers + + def _handle_binary(self, response): + # type: (Dict[str,Any]) -> Dict[str,Any] + if response.get('isBase64Encoded'): + body = base64.b64decode(response['body']) + response['body'] = body + return response + + +class ChaliceRequestHandler(BaseHTTPRequestHandler): + """A class for mapping raw HTTP events to and from LocalGateway.""" + protocol_version = 'HTTP/1.1' + + def __init__(self, request, client_address, server, app_object, config): + # type: (bytes, Tuple[str, int], HTTPServer, Chalice, Config) -> None + self.local_gateway = LocalGateway(app_object, config) + BaseHTTPRequestHandler.__init__( + self, request, client_address, server) # type: ignore + + def _parse_payload(self): + # type: () -> Tuple[HeaderType, str] + body = None + content_length = int(self.headers.get('content-length', '0')) + if content_length > 0: + body = self.rfile.read(content_length) + # mypy doesn't like dict(self.headers) so I had to use a + # dictcomp instead to make it happy. + converted_headers = {key: value for key, value in self.headers.items()} + return converted_headers, body + + def _generic_handle(self): + # type: () -> None + headers, body = self._parse_payload() + try: + response = self.local_gateway.handle_request( + method=self.command, + path=self.path, + headers=headers, + body=body + ) + status_code = response['statusCode'] + headers = response['headers'] + body = response['body'] + self._send_http_response(status_code, headers, body) + except LocalGatewayException as e: + self._send_error_response(e) + + def _send_error_response(self, error): + # type: (LocalGatewayException) -> None + code = error.CODE + headers = error.headers + body = error.body + self._send_http_response(code, headers, body) + + def _send_http_response(self, code, headers, body): + # type: (int, HeaderType, Optional[Union[str,bytes]]) -> None + if body is None: + self._send_http_response_no_body(code, headers) + else: + self._send_http_response_with_body(code, headers, body) + + def _send_http_response_with_body(self, code, headers, body): + # type: (int, HeaderType, Union[str,bytes]) -> None + self.send_response(code) + self.send_header('Content-Length', str(len(body))) + content_type = headers.pop( + 'Content-Type', 'application/json') + self.send_header('Content-Type', content_type) + for header_name, header_value in headers.items(): + self.send_header(header_name, header_value) + self.end_headers() + if not isinstance(body, bytes): + body = body.encode('utf-8') + self.wfile.write(body) + + do_GET = do_PUT = do_POST = do_HEAD = do_DELETE = do_PATCH = do_OPTIONS = \ + _generic_handle - self.send_response(200) - for k, v in cors_headers.items(): + def _send_http_response_no_body(self, code, headers): + # type: (int, HeaderType) -> None + self.send_response(code) + for k, v in headers.items(): self.send_header(k, v) self.end_headers() class LocalDevServer(object): - def __init__(self, app_object, port, handler_cls=ChaliceRequestHandler, - server_cls=HTTPServer): - # type: (Chalice, int, HandlerCls, ServerCls) -> None + def __init__(self, app_object, config, port, + handler_cls=ChaliceRequestHandler, server_cls=HTTPServer): + # type: (Chalice, Config, int, HandlerCls, ServerCls) -> None self.app_object = app_object self.port = port self._wrapped_handler = functools.partial( - handler_cls, app_object=app_object) + handler_cls, app_object=app_object, config=config) self.server = server_cls(('', port), self._wrapped_handler) def handle_single_request(self): diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index a42dd1673b..437ce5c3f0 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -1,3 +1,4 @@ +import json from pytest import fixture from chalice.app import Chalice @@ -32,3 +33,52 @@ def foo(): return {} return app + + +@fixture +def create_event(): + def create_event_inner(uri, method, path, content_type='application/json'): + return { + 'requestContext': { + 'httpMethod': method, + 'resourcePath': uri, + }, + 'headers': { + 'Content-Type': content_type, + }, + 'pathParameters': path, + 'queryStringParameters': {}, + 'body': "", + 'stageVariables': {}, + } + return create_event_inner + + +@fixture +def create_empty_header_event(): + def create_empty_header_event_inner(uri, method, path, + content_type='application/json'): + return { + 'requestContext': { + 'httpMethod': method, + 'resourcePath': uri, + }, + 'headers': None, + 'pathParameters': path, + 'queryStringParameters': {}, + 'body': "", + 'stageVariables': {}, + } + return create_empty_header_event_inner + + +@fixture +def create_event_with_body(): + def create_event_with_body_inner(body, uri='/', method='POST', + content_type='application/json'): + event = create_event()(uri, method, {}, content_type) + if content_type == 'application/json': + body = json.dumps(body) + event['body'] = body + return event + return create_event_with_body_inner diff --git a/tests/unit/test_app.py b/tests/unit/test_app.py index fc9cdae36f..d0676d3540 100644 --- a/tests/unit/test_app.py +++ b/tests/unit/test_app.py @@ -53,46 +53,6 @@ def create_request_with_content_type(content_type): ) -def create_event(uri, method, path, content_type='application/json'): - return { - 'requestContext': { - 'httpMethod': method, - 'resourcePath': uri, - }, - 'headers': { - 'Content-Type': content_type, - }, - 'pathParameters': path, - 'queryStringParameters': {}, - 'body': "", - 'stageVariables': {}, - } - - -def create_empty_header_event(uri, method, path, - content_type='application/json'): - return { - 'requestContext': { - 'httpMethod': method, - 'resourcePath': uri, - }, - 'headers': None, - 'pathParameters': path, - 'queryStringParameters': {}, - 'body': "", - 'stageVariables': {}, - } - - -def create_event_with_body(body, uri='/', method='POST', - content_type='application/json'): - event = create_event(uri, method, {}, content_type) - if content_type == 'application/json': - body = json.dumps(body) - event['body'] = body - return event - - def assert_response_body_is(response, body): assert json.loads(response['body']) == body @@ -221,13 +181,13 @@ def test_error_on_unknown_event(sample_app): assert json_response_body(raw_response)['Code'] == 'InternalServerError' -def test_can_route_api_call_to_view_function(sample_app): +def test_can_route_api_call_to_view_function(sample_app, create_event): event = create_event('/index', 'GET', {}) response = sample_app(event, context=None) assert_response_body_is(response, {'hello': 'world'}) -def test_can_call_to_dict_on_current_request(sample_app): +def test_can_call_to_dict_on_current_request(sample_app, create_event): @sample_app.route('/todict') def todict(): return sample_app.current_request.to_dict() @@ -243,34 +203,35 @@ def todict(): assert isinstance(json.loads(json.dumps(response)), dict) -def test_will_pass_captured_params_to_view(sample_app): +def test_will_pass_captured_params_to_view(sample_app, create_event): event = create_event('/name/{name}', 'GET', {'name': 'james'}) response = sample_app(event, context=None) response = json_response_body(response) assert response == {'provided-name': 'james'} -def test_error_on_unsupported_method(sample_app): +def test_error_on_unsupported_method(sample_app, create_event): event = create_event('/name/{name}', 'POST', {'name': 'james'}) raw_response = sample_app(event, context=None) assert raw_response['statusCode'] == 405 assert json_response_body(raw_response)['Code'] == 'MethodNotAllowedError' -def test_error_on_unsupported_method_gives_feedback_on_method(sample_app): +def test_error_on_unsupported_method_gives_feedback_on_method(sample_app, + create_event): method = 'POST' event = create_event('/name/{name}', method, {'name': 'james'}) raw_response = sample_app(event, context=None) assert 'POST' in json_response_body(raw_response)['Message'] -def test_no_view_function_found(sample_app): +def test_no_view_function_found(sample_app, create_event): bad_path = create_event('/noexist', 'GET', {}) with pytest.raises(app.ChaliceError): sample_app(bad_path, context=None) -def test_can_access_context(): +def test_can_access_context(create_event): demo = app.Chalice('app-name') @demo.route('/index') @@ -286,7 +247,7 @@ def index_view(): assert result == serialized_lambda_context -def test_can_access_raw_body(): +def test_can_access_raw_body(create_event): demo = app.Chalice('app-name') @demo.route('/index') @@ -300,7 +261,7 @@ def index_view(): assert result == {'rawbody': '{"hello": "world"}'} -def test_raw_body_cache_returns_same_result(): +def test_raw_body_cache_returns_same_result(create_event): demo = app.Chalice('app-name') @demo.route('/index') @@ -320,7 +281,7 @@ def index_view(): assert result['rawbody'] == result['rawbody2'] -def test_can_have_views_of_same_route_but_different_methods(): +def test_can_have_views_of_same_route_but_different_methods(create_event): demo = app.Chalice('app-name') @demo.route('/index', methods=['GET']) @@ -356,7 +317,7 @@ def index_view_dup(): return {'foo': 'bar'} -def test_json_body_available_with_right_content_type(): +def test_json_body_available_with_right_content_type(create_event): demo = app.Chalice('demo-app') @demo.route('/', methods=['POST']) @@ -371,7 +332,7 @@ def index(): assert result == {'foo': 'bar'} -def test_cant_access_json_body_with_wrong_content_type(): +def test_cant_access_json_body_with_wrong_content_type(create_event): demo = app.Chalice('demo-app') @demo.route('/', methods=['POST'], content_types=['application/xml']) @@ -388,7 +349,7 @@ def index(): assert raw_body == '<Message>hello</Message>' -def test_json_body_available_on_multiple_content_types(): +def test_json_body_available_on_multiple_content_types(create_event_with_body): demo = app.Chalice('demo-app') @demo.route('/', methods=['POST'], @@ -415,7 +376,8 @@ def index(): assert raw_body == '{"foo": "bar"}' -def test_json_body_available_with_lowercase_content_type_key(): +def test_json_body_available_with_lowercase_content_type_key( + create_event_with_body): demo = app.Chalice('demo-app') @demo.route('/', methods=['POST']) @@ -441,7 +403,7 @@ def index_post(): return {'foo': 'bar'} -def test_content_type_validation_raises_error_on_unknown_types(): +def test_content_type_validation_raises_error_on_unknown_types(create_event): demo = app.Chalice('demo-app') @demo.route('/', methods=['POST'], content_types=['application/xml']) @@ -457,7 +419,7 @@ def index(): assert 'application/bad-xml' in json_response['Message'] -def test_content_type_with_charset(): +def test_content_type_with_charset(create_event): demo = app.Chalice('demo-app') @demo.route('/', content_types=['application/json']) @@ -469,7 +431,7 @@ def index(): assert response == {'foo': 'bar'} -def test_can_return_response_object(): +def test_can_return_response_object(create_event): demo = app.Chalice('app-name') @demo.route('/index') @@ -483,7 +445,7 @@ def index_view(): 'headers': {'Content-Type': 'application/json'}} -def test_headers_have_basic_validation(): +def test_headers_have_basic_validation(create_event): demo = app.Chalice('app-name') @demo.route('/index') @@ -499,7 +461,7 @@ def index_view(): assert json.loads(response['body'])['Code'] == 'InternalServerError' -def test_empty_headers_have_basic_validation(): +def test_empty_headers_have_basic_validation(create_empty_header_event): demo = app.Chalice('app-name') @demo.route('/index') @@ -512,7 +474,7 @@ def index_view(): assert response['statusCode'] == 200 -def test_no_content_type_is_still_allowed(): +def test_no_content_type_is_still_allowed(create_event): # When the content type validation happens in API gateway, it appears # to assume a default of application/json, so the chalice handler needs # to emulate that behavior. @@ -530,7 +492,7 @@ def index(): assert json_response == {'success': True} -def test_can_base64_encode_binary_media_types_bytes(): +def test_can_base64_encode_binary_media_types_bytes(create_event): demo = app.Chalice('demo-app') @demo.route('/index') @@ -549,7 +511,8 @@ def index_view(): assert response['headers']['Content-Type'] == 'application/octet-stream' -def test_can_return_text_even_with_binary_content_type_configured(): +def test_can_return_text_even_with_binary_content_type_configured( + create_event): demo = app.Chalice('demo-app') @demo.route('/index') @@ -604,7 +567,7 @@ def test_route_inequality(view_function): assert not a == b -def test_exceptions_raised_as_chalice_errors(sample_app): +def test_exceptions_raised_as_chalice_errors(sample_app, create_event): @sample_app.route('/error') def raise_error(): @@ -620,7 +583,7 @@ def raise_error(): assert raw_response['statusCode'] == 500 -def test_original_exception_raised_in_debug_mode(sample_app): +def test_original_exception_raised_in_debug_mode(sample_app, create_event): sample_app.debug = True @sample_app.route('/error') @@ -636,7 +599,8 @@ def raise_error(): assert 'You will see this error' in response['body'] -def test_chalice_view_errors_propagate_in_non_debug_mode(sample_app): +def test_chalice_view_errors_propagate_in_non_debug_mode(sample_app, + create_event): @sample_app.route('/notfound') def notfound(): raise NotFoundError("resource not found") @@ -647,7 +611,7 @@ def notfound(): assert json_response_body(raw_response)['Code'] == 'NotFoundError' -def test_chalice_view_errors_propagate_in_debug_mode(sample_app): +def test_chalice_view_errors_propagate_in_debug_mode(sample_app, create_event): @sample_app.route('/notfound') def notfound(): raise NotFoundError("resource not found") @@ -668,7 +632,7 @@ def test_case_insensitive_mapping(): assert repr({'header': 'Value'}) in repr(mapping) -def test_unknown_kwargs_raise_error(sample_app): +def test_unknown_kwargs_raise_error(sample_app, create_event): with pytest.raises(TypeError): @sample_app.route('/foo', unknown_kwargs='foo') def badkwargs(): @@ -721,7 +685,7 @@ def test_json_body_available_when_content_type_matches(content_type, is_json): assert request.json_body is None -def test_can_receive_binary_data(): +def test_can_receive_binary_data(create_event_with_body): content_type = 'application/octet-stream' demo = app.Chalice('demo-app') @@ -743,7 +707,8 @@ def bincat(): assert response['body'] == body -def test_cannot_receive_base64_string_with_binary_response(): +def test_cannot_receive_base64_string_with_binary_response( + create_event_with_body): content_type = 'application/octet-stream' demo = app.Chalice('demo-app') @@ -1238,7 +1203,7 @@ def test_aws_execution_env_set(): ) -def test_can_use_out_of_order_args(): +def test_can_use_out_of_order_args(create_event): demo = app.Chalice('demo-app') # Note how the url params and function args are out of order. diff --git a/tests/unit/test_local.py b/tests/unit/test_local.py index d28afb2a44..bd745a3ec6 100644 --- a/tests/unit/test_local.py +++ b/tests/unit/test_local.py @@ -1,3 +1,4 @@ +import re import json import decimal import pytest @@ -7,6 +8,38 @@ from chalice import app from chalice import local, BadRequestError, CORSConfig from chalice import Response +from chalice import IAMAuthorizer +from chalice.config import Config +from chalice.local import LambdaContext +from chalice.local import LocalARNBuilder +from chalice.local import LocalGateway +from chalice.local import LocalGatewayAuthorizer +from chalice.local import NotAuthorizedError +from chalice.local import ForbiddenError + + +AWS_REQUEST_ID_PATTERN = re.compile( + '^[0-9a-f]{8}-([0-9a-f]{4}-){3}[0-9a-f]{12}$', + re.I) + + +class FakeTimeSource(object): + def __init__(self, times): + """Create a fake source of second-precision time. + + :type time: List + :param time: List of times that the time source should return in the + order it should return them. These should be in seconds. + """ + self._times = times + + def time(self): + """Get the next time. + + This is for mimicing the Clock interface used in local. + """ + time = self._times.pop(0) + return time class ChaliceStubbedHandler(local.ChaliceRequestHandler): @@ -22,6 +55,19 @@ def finish(self): pass +@pytest.fixture +def arn_builder(): + return LocalARNBuilder() + + +@pytest.fixture +def lambda_context_args(): + # LambdaContext has several positional args before the ones that we + # care about for the timing tests, this gives reasonable defaults for + # those arguments. + return ['lambda_name', 256] + + @fixture def sample_app(): demo = app.Chalice('demo-app') @@ -93,14 +139,120 @@ def binary_round_trip(): return demo +@fixture +def demo_app_auth(): + demo = app.Chalice('app-name') + + @demo.authorizer() + def auth_with_explicit_policy(auth_request): + token = auth_request.token + if token == 'allow': + return { + 'context': {}, + 'principalId': 'user', + 'policyDocument': { + 'Version': '2012-10-17', + 'Statement': [ + { + 'Action': 'execute-api:Invoke', + 'Effect': 'Allow', + 'Resource': + ["arn:aws:execute-api:mars-west-1:123456789012:" + "ymy8tbxw7b/api/GET/explicit"] + } + ] + } + } + else: + return { + 'context': {}, + 'principalId': '', + 'policyDocument': { + 'Version': '2012-10-17', + 'Statement': [ + { + 'Action': 'execute-api:Invoke', + 'Effect': 'Deny', + 'Resource': + ["arn:aws:execute-api:mars-west-1:123456789012:" + "ymy8tbxw7b/api/GET/explicit"] + } + ] + } + } + + @demo.authorizer() + def demo_auth(auth_request): + token = auth_request.token + if token == 'allow': + return app.AuthResponse(routes=['/index'], principal_id='user') + else: + return app.AuthResponse(routes=[], principal_id='user') + + @demo.authorizer() + def all_auth(auth_request): + token = auth_request.token + if token == 'allow': + return app.AuthResponse(routes=['*'], principal_id='user') + else: + return app.AuthResponse(routes=[], principal_id='user') + + @demo.authorizer() + def landing_page_auth(auth_request): + token = auth_request.token + if token == 'allow': + return app.AuthResponse(routes=['/'], principal_id='user') + else: + return app.AuthResponse(routes=[], principal_id='user') + + iam_authorizer = IAMAuthorizer() + + @demo.route('/', authorizer=landing_page_auth) + def landing_view(): + return {} + + @demo.route('/index', authorizer=demo_auth) + def index_view(): + return {} + + @demo.route('/secret', authorizer=demo_auth) + def secret_view(): + return {} + + @demo.route('/secret/{value}', authorizer=all_auth) + def secret_view_value(value): + return {'secret': value} + + @demo.route('/explicit', authorizer=auth_with_explicit_policy) + def explicit(): + return {} + + @demo.route('/iam', authorizer=iam_authorizer) + def iam_route(): + return {} + + return demo + + @fixture def handler(sample_app): - chalice_handler = ChaliceStubbedHandler(None, ('127.0.0.1', 2000), None, - app_object=sample_app) + config = Config() + chalice_handler = ChaliceStubbedHandler( + None, ('127.0.0.1', 2000), None, app_object=sample_app, config=config) chalice_handler.sample_app = sample_app return chalice_handler +@fixture +def auth_handler(demo_app_auth): + config = Config() + chalice_handler = ChaliceStubbedHandler( + None, ('127.0.0.1', 2000), None, app_object=demo_app_auth, + config=config) + chalice_handler.sample_app = demo_app_auth + return chalice_handler + + def _get_raw_body_from_response_stream(handler): # This is going to include things like status code and # response headers in the raw stream. We just care about the @@ -292,6 +444,17 @@ def test_content_type_included_once(handler): assert len(content_header_lines) == 1 +def test_can_deny_unauthed_request(auth_handler): + set_current_request(auth_handler, method='GET', path='/index') + auth_handler.do_GET() + value = auth_handler.wfile.getvalue() + response_lines = value.splitlines() + assert b'HTTP/1.1 401 Unauthorized' in response_lines + assert b'x-amzn-ErrorType: UnauthorizedException' in response_lines + assert b'Content-Type: application/json' in response_lines + assert b'{"message":"Unauthorized"}' in response_lines + + @pytest.mark.parametrize('actual_url,matched_url', [ ('/foo', '/foo'), ('/foo/bar', '/foo/bar'), @@ -404,5 +567,346 @@ def test_can_create_lambda_event_for_post_with_formencoded_body(): def test_can_provide_port_to_local_server(sample_app): - dev_server = local.create_local_server(sample_app, port=23456) + dev_server = local.create_local_server(sample_app, None, port=23456) assert dev_server.server.server_port == 23456 + + +class TestLambdaContext(object): + def test_can_get_remaining_time_once(self, lambda_context_args): + time_source = FakeTimeSource([0, 5]) + context = LambdaContext(*lambda_context_args, runtime_millis=10000, + time_source=time_source) + time_remaining = context.get_remaining_time_in_millis() + assert time_remaining == 5000 + + def test_can_get_remaining_time_multiple(self, lambda_context_args): + time_source = FakeTimeSource([0, 3, 7, 9]) + context = LambdaContext(*lambda_context_args, runtime_millis=10000, + time_source=time_source) + + time_remaining = context.get_remaining_time_in_millis() + assert time_remaining == 7000 + time_remaining = context.get_remaining_time_in_millis() + assert time_remaining == 3000 + time_remaining = context.get_remaining_time_in_millis() + assert time_remaining == 1000 + + def test_does_populate_aws_request_id_with_valid_uuid(self, + lambda_context_args): + context = LambdaContext(*lambda_context_args) + assert AWS_REQUEST_ID_PATTERN.match(context.aws_request_id) + + def test_does_set_version_to_latest(self, lambda_context_args): + context = LambdaContext(*lambda_context_args) + assert context.function_version == '$LATEST' + + +class TestLocalGateway(object): + def test_can_invoke_function(self): + demo = app.Chalice('app-name') + + @demo.route('/') + def index_view(): + return {'foo': 'bar'} + + gateway = LocalGateway(demo, Config()) + response = gateway.handle_request('GET', '/', {}, '') + body = json.loads(response['body']) + assert body['foo'] == 'bar' + + def test_does_populate_context(self): + demo = app.Chalice('app-name') + + @demo.route('/context') + def context_view(): + context = demo.lambda_context + return { + 'name': context.function_name, + 'memory': context.memory_limit_in_mb, + 'version': context.function_version, + 'timeout': context.get_remaining_time_in_millis(), + 'request_id': context.aws_request_id, + } + + disk_config = { + 'lambda_timeout': 10, + 'lambda_memory_size': 256, + } + config = Config(chalice_stage='api', config_from_disk=disk_config) + gateway = LocalGateway(demo, config) + response = gateway.handle_request('GET', '/context', {}, '') + body = json.loads(response['body']) + assert body['name'] == 'api_handler' + assert body['memory'] == 256 + assert body['version'] == '$LATEST' + assert body['timeout'] <= 10 + assert AWS_REQUEST_ID_PATTERN.match(body['request_id']) + + def test_can_validate_route_with_variables(self, demo_app_auth): + gateway = LocalGateway(demo_app_auth, Config()) + response = gateway.handle_request( + 'GET', '/secret/foobar', {'Authorization': 'allow'}, '') + json_body = json.loads(response['body']) + assert json_body['secret'] == 'foobar' + + def test_does_deny_unauthed_request(self, demo_app_auth): + gateway = LocalGateway(demo_app_auth, Config()) + with pytest.raises(ForbiddenError) as ei: + gateway.handle_request( + 'GET', '/index', {'Authorization': 'deny'}, '') + + assert ('{"Message": ' + '"User is not authorized to ' + 'access this resource"}') in ei.value.body + + def test_does_throw_unauthorized_when_no_auth_token_present_on_valid_route( + self, demo_app_auth): + gateway = LocalGateway(demo_app_auth, Config()) + with pytest.raises(NotAuthorizedError) as ei: + gateway.handle_request( + 'GET', '/index', {}, '') + assert '{"message":"Unauthorized"}' in ei.value.body + + def test_does_deny_with_forbidden_when_route_not_found( + self, demo_app_auth): + gateway = LocalGateway(demo_app_auth, Config()) + with pytest.raises(ForbiddenError) as ei: + gateway.handle_request('GET', '/badindex', {}, '') + exception_body = ei.value.body + assert 'Missing Authentication Token' in exception_body + + def test_does_deny_with_forbidden_when_auth_token_present( + self, demo_app_auth): + gateway = LocalGateway(demo_app_auth, Config()) + with pytest.raises(ForbiddenError) as ei: + gateway.handle_request('GET', '/badindex', + {'Authorization': 'foobar'}, '') + # The message should be a more complicated error message to do with + # signing the request. It always ends with the Authorization token + # that we passed up, so we can check for that. + exception_body = ei.value.body + assert 'Authorization=foobar' in exception_body + + +class TestLocalBuiltinAuthorizers(object): + def test_can_authorize_empty_path(self, lambda_context_args, + demo_app_auth, create_event): + # Ensures that / routes work since that is a special case in the + # API Gateway arn generation where an extra / is appended to the end + # of the arn. + authorizer = LocalGatewayAuthorizer(demo_app_auth) + event = create_event('/', 'GET', {}) + event['headers']['authorization'] = 'allow' + context = LambdaContext(*lambda_context_args) + event, context = authorizer.authorize(event, context) + assert event['requestContext']['authorizer']['principalId'] == 'user' + + def test_can_call_method_without_auth(self, lambda_context_args, + create_event): + demo = app.Chalice('app-name') + + @demo.route('/index') + def index_view(): + return {} + + authorizer = LocalGatewayAuthorizer(demo) + original_event = create_event('/index', 'GET', {}) + original_context = LambdaContext(*lambda_context_args) + event, context = authorizer.authorize(original_event, original_context) + # Assert that when the authorizer.authorize is called and there is no + # authorizer defined for a particular route that it is a noop. + assert original_event == event + assert original_context == context + + def test_does_raise_not_authorized_error(self, demo_app_auth, + lambda_context_args, + create_event): + authorizer = LocalGatewayAuthorizer(demo_app_auth) + event = create_event('/index', 'GET', {}) + context = LambdaContext(*lambda_context_args) + with pytest.raises(NotAuthorizedError): + authorizer.authorize(event, context) + + def test_does_authorize_valid_requests(self, demo_app_auth, + lambda_context_args, create_event): + authorizer = LocalGatewayAuthorizer(demo_app_auth) + event = create_event('/index', 'GET', {}) + event['headers']['authorization'] = 'allow' + context = LambdaContext(*lambda_context_args) + event, context = authorizer.authorize(event, context) + assert event['requestContext']['authorizer']['principalId'] == 'user' + + def test_does_authorize_unsupported_authorizer(self, demo_app_auth, + lambda_context_args, + capsys, create_event): + authorizer = LocalGatewayAuthorizer(demo_app_auth) + event = create_event('/iam', 'GET', {}) + context = LambdaContext(*lambda_context_args) + new_event, new_context = authorizer.authorize(event, context) + _, err = capsys.readouterr() + assert event == new_event + assert context == new_context + assert ('UserWarning: IAMAuthorizer is not a supported in local ' + 'mode. All requests made against a route will be authorized' + ' to allow local testing.') in err + + def test_cannot_access_view_without_permission(self, demo_app_auth, + lambda_context_args, + create_event): + authorizer = LocalGatewayAuthorizer(demo_app_auth) + event = create_event('/secret', 'GET', {}) + event['headers']['authorization'] = 'allow' + context = LambdaContext(*lambda_context_args) + with pytest.raises(ForbiddenError): + authorizer.authorize(event, context) + + def test_can_understand_explicit_auth_policy(self, demo_app_auth, + lambda_context_args, + create_event): + authorizer = LocalGatewayAuthorizer(demo_app_auth) + event = create_event('/explicit', 'GET', {}) + event['headers']['authorization'] = 'allow' + context = LambdaContext(*lambda_context_args) + event, context = authorizer.authorize(event, context) + assert event['requestContext']['authorizer']['principalId'] == 'user' + + def test_can_understand_explicit_deny_policy(self, demo_app_auth, + lambda_context_args, + create_event): + # Our auto-generated policies from the AuthResponse object do not + # contain any Deny clauses, however we also allow the user to return + # a dictionary that is transated into a policy, so we have to + # account for the ability for a user to set an explicit deny policy. + # It should behave exactly as not getting permission added with an + # allow. + authorizer = LocalGatewayAuthorizer(demo_app_auth) + event = create_event('/explicit', 'GET', {}) + context = LambdaContext(*lambda_context_args) + with pytest.raises(NotAuthorizedError): + authorizer.authorize(event, context) + + +class TestArnBuilder(object): + def test_can_create_basic_arn(self, arn_builder): + arn = ('arn:aws:execute-api:mars-west-1:123456789012:ymy8tbxw7b' + '/api/GET/resource') + built_arn = arn_builder.build_arn('GET', '/resource') + assert arn == built_arn + + def test_can_create_root_arn(self, arn_builder): + arn = ('arn:aws:execute-api:mars-west-1:123456789012:ymy8tbxw7b' + '/api/GET//') + built_arn = arn_builder.build_arn('GET', '/') + assert arn == built_arn + + def test_can_create_multi_part_arn(self, arn_builder): + arn = ('arn:aws:execute-api:mars-west-1:123456789012:ymy8tbxw7b' + '/api/GET/path/to/resource') + built_arn = arn_builder.build_arn('GET', '/path/to/resource') + assert arn == built_arn + + def test_can_create_glob_method_arn(self, arn_builder): + arn = ('arn:aws:execute-api:mars-west-1:123456789012:ymy8tbxw7b' + '/api/*/resource') + built_arn = arn_builder.build_arn('*', '/resource') + assert arn == built_arn + + +@pytest.mark.parametrize('arn,pattern', [ + ('mars-west-2:123456789012:ymy8tbxw7b/api/GET/foo', + 'mars-west-2:123456789012:ymy8tbxw7b/api/GET/foo' + ), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-1:123456789012:ymy8tbxw7b/api/GET/*' + ), + ('mars-west-1:123456789012:ymy8tbxw7b/api/PUT/foobar', + 'mars-west-1:123456789012:ymy8tbxw7b/api/???/foobar' + ), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-1:123456789012:ymy8tbxw7b/api/???/*' + ), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-1:123456789012:*/api/GET/*' + ), + ('mars-west-2:123456789012:ymy8tbxw7b/api/GET/foobar', + '*' + ), + ('mars-west-2:123456789012:ymy8tbxw7b/api/GET/foo.bar', + 'mars-west-2:123456789012:ymy8tbxw7b/*/GET/*') +]) +def test_can_allow_route_arns(arn, pattern): + prefix = 'arn:aws:execute-api:' + full_arn = '%s%s' % (prefix, arn) + full_pattern = '%s%s' % (prefix, pattern) + matcher = local.ARNMatcher(full_arn) + does_match = matcher.does_any_resource_match([full_pattern]) + assert does_match is True + + +@pytest.mark.parametrize('arn,pattern', [ + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-1:123456789012:ymy8tbxw7b/api/PUT/*' + ), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-1:123456789012:ymy8tbxw7b/api/??/foobar' + ), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-2:123456789012:ymy8tbxw7b/api/???/*' + ), + ('mars-west-2:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-2:123456789012:ymy8tbxw7b/*/GET/foo...') +]) +def test_can_deny_route_arns(arn, pattern): + prefix = 'arn:aws:execute-api:' + full_arn = '%s%s' % (prefix, arn) + full_pattern = '%s%s' % (prefix, pattern) + matcher = local.ARNMatcher(full_arn) + does_match = matcher.does_any_resource_match([full_pattern]) + assert does_match is False + + +@pytest.mark.parametrize('arn,patterns', [ + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + [ + 'mars-west-1:123456789012:ymy8tbxw7b/api/PUT/*', + 'mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar' + ]), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + [ + 'mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-1:123456789012:ymy8tbxw7b/api/PUT/*' + ]), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + [ + 'mars-west-1:123456789012:ymy8tbxw7b/api/PUT/foobar', + '*' + ]) +]) +def test_can_allow_multiple_resource_arns(arn, patterns): + prefix = 'arn:aws:execute-api:' + full_arn = '%s%s' % (prefix, arn) + full_patterns = ['%s%s' % (prefix, pattern) for pattern in patterns] + matcher = local.ARNMatcher(full_arn) + does_match = matcher.does_any_resource_match(full_patterns) + assert does_match is True + + +@pytest.mark.parametrize('arn,patterns', [ + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + [ + 'mars-west-1:123456789012:ymy8tbxw7b/api/POST/*', + 'mars-west-1:123456789012:ymy8tbxw7b/api/PUT/foobar' + ]), + ('mars-west-1:123456789012:ymy8tbxw7b/api/GET/foobar', + [ + 'mars-west-2:123456789012:ymy8tbxw7b/api/GET/foobar', + 'mars-west-2:123456789012:ymy8tbxw7b/api/*/*' + ]) +]) +def test_can_deny_multiple_resource_arns(arn, patterns): + prefix = 'arn:aws:execute-api:' + full_arn = '%s%s' % (prefix, arn) + full_patterns = ['%s%s' % (prefix, pattern) for pattern in patterns] + matcher = local.ARNMatcher(full_arn) + does_match = matcher.does_any_resource_match(full_patterns) + assert does_match is False