diff --git a/samcli/lib/providers/api_collector.py b/samcli/lib/providers/api_collector.py index 3c71bbcf07..c25c4c002d 100644 --- a/samcli/lib/providers/api_collector.py +++ b/samcli/lib/providers/api_collector.py @@ -217,7 +217,11 @@ def get_api(self) -> Api: @staticmethod def normalize_cors_methods(routes: List[Route], cors: Optional[Cors]) -> List[Route]: """ - Adds OPTIONS method to all the route methods if cors exists + Adds OPTIONS method to route methods if cors exists while preserving + existing OPTIONS ownership for each function and path, regardless of + operation name. In get_api(), authorizers are linked before this step, so + synthesized OPTIONS prefers a route without a linked local authorizer. If + every sibling has one, the first route remains the fallback owner. Parameters ----------- @@ -229,52 +233,117 @@ def normalize_cors_methods(routes: List[Route], cors: Optional[Cors]) -> List[Ro Return ------- - A list of routes without duplicate routes with the same function_name and method + A list of routes with existing OPTIONS ownership preserved and synthesized + OPTIONS assigned to at most one route per group """ + if not cors: + return routes - def add_options_to_route(route: Route) -> Route: - if "OPTIONS" not in route.methods: - route.methods.append("OPTIONS") - return route + grouped_routes: Dict[Tuple[str, Optional[str], str], List[Route]] = {} - return routes if not cors else [add_options_to_route(route) for route in routes] + for route in routes: + key = (route.stack_path, route.function_name, route.path) + grouped_routes.setdefault(key, []).append(route) + + result: List[Route] = [] + + for route_group in grouped_routes.values(): + options_claimed = any("OPTIONS" in route.methods for route in route_group) + + if not options_claimed: + owner = next( + (route for route in route_group if route.authorizer_object is None), + route_group[0], + ) + owner.methods.append("OPTIONS") + + result.extend(route_group) + + return result @staticmethod def dedupe_function_routes(routes: List[Route]) -> List[Route]: """ - Remove duplicate routes that have the same function_name and method + Remove duplicate routes that have the same function_name, path, and method while preserving method-specific + operation names. route: list(Route) List of Routes Return ------- - A list of routes without duplicate routes with the same stack_path, function_name and method + A list of routes without duplicate routes with the same stack_path, function_name, path, and method """ - grouped_routes: Dict[str, Route] = {} + grouped_routes: Dict[Tuple[str, Optional[str], str], List[Route]] = {} for route in routes: - key = "{}-{}-{}-{}".format(route.stack_path, route.function_name, route.path, route.operation_name or "") - config = grouped_routes.get(key, None) - methods = route.methods - if config: - methods += config.methods - sorted_methods = sorted(methods) - # Prefer route-specific CORS over None - cors = route.cors if route.cors is not None else (config.cors if config else None) - grouped_routes[key] = Route( - function_name=route.function_name, - path=route.path, - methods=sorted_methods, - event_type=route.event_type, - payload_format_version=route.payload_format_version, - operation_name=route.operation_name, - stack_path=route.stack_path, - authorizer_name=route.authorizer_name, - authorizer_object=route.authorizer_object, - cors=cors, + key = (route.stack_path, route.function_name, route.path) + grouped_routes.setdefault(key, []).append(route) + + result: List[Route] = [] + + def has_same_authorizer(first: Route, second: Route) -> bool: + return ( + first.authorizer_name == second.authorizer_name and first.authorizer_object == second.authorizer_object ) - return list(grouped_routes.values()) + + def can_merge(first: Route, second: Route) -> bool: + return has_same_authorizer(first, second) and (first.operation_name or "") == (second.operation_name or "") + + for route_group in grouped_routes.values(): + merged_routes: List[Route] = [] + group_cors = next((route.cors for route in route_group if route.cors is not None), None) + + # Process broader routes first so a more specific route can own overlapping methods, even when operation + # names differ. Only routes with the same authorizer and operation-name metadata are merged into one Route. + for route in sorted(route_group, key=lambda item: len(item.methods), reverse=True): + methods = list(dict.fromkeys(route.methods)) + + for existing_route in merged_routes: + if not can_merge(existing_route, route): + existing_route.methods = [method for method in existing_route.methods if method not in methods] + + matching_route = next( + (existing_route for existing_route in merged_routes if can_merge(existing_route, route)), + None, + ) + + if matching_route: + matching_route.methods = sorted(set(matching_route.methods + methods)) + + if matching_route.payload_format_version is None: + matching_route.payload_format_version = route.payload_format_version + + if route.cors is not None: + matching_route.cors = route.cors + + # Authorizers are already resolved by _link_authorizers() before + # deduplication, so use_default_authorizer does not affect this merge. + continue + + merged_routes.append( + Route( + function_name=route.function_name, + path=route.path, + methods=sorted(methods), + event_type=route.event_type, + payload_format_version=route.payload_format_version, + operation_name=route.operation_name, + stack_path=route.stack_path, + authorizer_name=route.authorizer_name, + authorizer_object=route.authorizer_object, + use_default_authorizer=route.use_default_authorizer, + cors=route.cors, + ) + ) + + for merged_route in merged_routes: + if merged_route.cors is None: + merged_route.cors = group_cors + + result.extend(route for route in merged_routes if route.methods) + + return result def add_binary_media_types(self, logical_id: str, binary_media_types: Optional[List[str]]) -> None: """ diff --git a/samcli/lib/providers/sam_api_provider.py b/samcli/lib/providers/sam_api_provider.py index aec497429b..7fcf70643f 100644 --- a/samcli/lib/providers/sam_api_provider.py +++ b/samcli/lib/providers/sam_api_provider.py @@ -553,11 +553,13 @@ def _convert_event_route( @staticmethod def merge_routes(collector: ApiCollector) -> List[Route]: """ - Quite often, an API is defined both in Implicit and Explicit Route definitions. In such cases, Implicit API - definition wins because that conveys clear intent that the API is backed by a function. This method will - merge two such list of routes with the right order of precedence. If a Path+Method combination is defined - in both the places, only one wins. - In a multi-stack situation, the API defined in the top level wins. + Quite often, an API is defined in both implicit and explicit route definitions. The implicit API normally + wins because that conveys clear intent that the API is backed by a function. When a later expanded ANY route + overlaps a single-method route from the same function, both are retained only if the narrower route explicitly + declares different authorizer intent. Downstream deduplication then preserves that method-level authorization, + including when the routes have different operation names. A payload format version is inherited only when the + later route actually replaces the earlier route. In a multi-stack situation, the API defined in the top level + wins. Parameters ---------- @@ -581,8 +583,8 @@ def merge_routes(collector: ApiCollector) -> List[Route]: else: explicit_routes.extend(apis) - # We will use "path+method" combination as key to this dictionary and store the Api config for this combination. - # If an path+method combo already exists, then overwrite it if and only if this is an implicit API + # Use the "path+method" combination as the key. Later routes normally overwrite earlier routes, subject to + # the narrow authorizer-preservation exception below. all_routes: Dict[str, Route] = {} # By adding implicit APIs to the end of the list, they will be iterated last. If a configuration was already @@ -594,13 +596,34 @@ def merge_routes(collector: ApiCollector) -> List[Route]: ) for config in all_configs: - # Normalize the methods before de-duping to allow an ANY method in implicit API to override a regular HTTP - # method on explicit route. + # Normalize methods before de-duping so an ANY route normally overrides a regular HTTP method. + # A narrow route with explicit, different authorizer intent is conditionally preserved below. for normalized_method in config.methods: key = config.path + normalized_method route = all_routes.get(key) + + # Preserve a single-method route when it explicitly declares different + # raw authorizer intent and both routes can be reconciled downstream. + if ( + route + and len(route.methods) == 1 + and set(config.methods) == set(Route.ANY_HTTP_METHODS) + and route.function_name == config.function_name + and route.stack_path == config.stack_path + and route.event_type == config.event_type + and (route.authorizer_name is not None or not route.use_default_authorizer) + and ( + route.authorizer_name != config.authorizer_name + or route.use_default_authorizer != config.use_default_authorizer + ) + ): + continue + + # Inherit only from a route that config is about to replace. A preserved route must not mutate the + # shared expanded ANY route used by the other method keys. if route and route.payload_format_version and config.payload_format_version is None: config.payload_format_version = route.payload_format_version + all_routes[key] = config result = set(all_routes.values()) # Assign to a set() to de-dupe diff --git a/tests/unit/commands/local/lib/test_api_collector.py b/tests/unit/commands/local/lib/test_api_collector.py index fdda3db3d4..9e7138394a 100644 --- a/tests/unit/commands/local/lib/test_api_collector.py +++ b/tests/unit/commands/local/lib/test_api_collector.py @@ -160,3 +160,376 @@ def test_link_authorizers(self, routes, authorizers, default_authorizer, expecte self.api_collector._link_authorizers() self.assertEqual(self.api_collector._route_per_resource, {self.apigw_id: expected_routes}) + + +class TestApiCollector_dedupe_function_routes(TestCase): + def test_preserves_options_route_with_different_authorizer(self): + routes = [ + Route( + function_name="func", + path="/{proxy+}", + methods=["ANY"], + authorizer_name="MyAuthorizer", + ), + Route( + function_name="func", + path="/{proxy+}", + methods=["OPTIONS"], + authorizer_name=None, + use_default_authorizer=False, + ), + ] + + actual = ApiCollector.dedupe_function_routes(routes) + + expected = [ + Route( + function_name="func", + path="/{proxy+}", + methods=["GET", "DELETE", "PUT", "POST", "HEAD", "PATCH"], + authorizer_name="MyAuthorizer", + ), + Route( + function_name="func", + path="/{proxy+}", + methods=["OPTIONS"], + authorizer_name=None, + use_default_authorizer=False, + ), + ] + + self.assertCountEqual(expected, actual) + + def test_reconciles_overlapping_routes_with_different_operation_names(self): + routes = [ + Route( + function_name="func", + path="/x", + methods=["ANY"], + authorizer_name="MyAuthorizer", + ), + Route( + function_name="func", + path="/x", + methods=["OPTIONS"], + operation_name="Preflight", + authorizer_name=None, + use_default_authorizer=False, + ), + ] + + actual = ApiCollector.dedupe_function_routes(routes) + + options_routes = [route for route in actual if "OPTIONS" in route.methods] + protected_route = next(route for route in actual if route.authorizer_name == "MyAuthorizer") + + self.assertEqual(len(actual), 2) + self.assertEqual(len(options_routes), 1) + self.assertEqual(options_routes[0].methods, ["OPTIONS"]) + self.assertEqual(options_routes[0].operation_name, "Preflight") + self.assertIsNone(options_routes[0].authorizer_name) + self.assertFalse(options_routes[0].use_default_authorizer) + self.assertNotIn("OPTIONS", protected_route.methods) + + def test_preserves_distinct_operation_names_for_disjoint_methods(self): + routes = [ + Route( + function_name="func", + path="/x", + methods=["GET"], + operation_name="GetX", + ), + Route( + function_name="func", + path="/x", + methods=["POST"], + operation_name="PostX", + ), + ] + + actual = ApiCollector.dedupe_function_routes(routes) + + routes_by_operation = {route.operation_name: route for route in actual} + self.assertEqual(len(actual), 2) + self.assertEqual(routes_by_operation["GetX"].methods, ["GET"]) + self.assertEqual(routes_by_operation["PostX"].methods, ["POST"]) + + def test_specific_operation_owns_overlap_with_same_authorizer(self): + routes = [ + Route( + function_name="func", + path="/x", + methods=["ANY"], + authorizer_name="MyAuthorizer", + ), + Route( + function_name="func", + path="/x", + methods=["OPTIONS"], + operation_name="Preflight", + authorizer_name="MyAuthorizer", + ), + ] + + actual = ApiCollector.dedupe_function_routes(routes) + + options_routes = [route for route in actual if "OPTIONS" in route.methods] + broad_route = next(route for route in actual if route.operation_name is None) + + self.assertEqual(len(actual), 2) + self.assertEqual(len(options_routes), 1) + self.assertEqual(options_routes[0].operation_name, "Preflight") + self.assertNotIn("OPTIONS", broad_route.methods) + + def test_preserves_cors_when_routes_split_by_authorizer(self): + cors = object() + + routes = [ + Route( + function_name="func", + path="/{proxy+}", + methods=["ANY"], + authorizer_name="MyAuthorizer", + ), + Route( + function_name="func", + path="/{proxy+}", + methods=["OPTIONS"], + authorizer_name=None, + use_default_authorizer=False, + cors=cors, + ), + ] + + actual = ApiCollector.dedupe_function_routes(routes) + + self.assertEqual(len(actual), 2) + self.assertTrue(all(route.cors is cors for route in actual)) + + def test_merges_routes_with_same_resolved_authorizer(self): + routes = [ + Route( + function_name="func", + path="/x", + methods=["GET"], + authorizer_name=None, + use_default_authorizer=True, + ), + Route( + function_name="func", + path="/x", + methods=["POST"], + authorizer_name=None, + use_default_authorizer=False, + ), + ] + + actual = ApiCollector.dedupe_function_routes(routes) + + self.assertEqual(len(actual), 1) + self.assertEqual(sorted(actual[0].methods), ["GET", "POST"]) + + def test_cors_normalization_does_not_readd_options_to_protected_route(self): + routes = [ + Route( + function_name="func", + path="/{proxy+}", + methods=["ANY"], + authorizer_name="MyAuthorizer", + ), + Route( + function_name="func", + path="/{proxy+}", + methods=["OPTIONS"], + authorizer_name=None, + use_default_authorizer=False, + ), + ] + + deduped_routes = ApiCollector.dedupe_function_routes(routes) + actual = ApiCollector.normalize_cors_methods(deduped_routes, object()) + + options_routes = [route for route in actual if "OPTIONS" in route.methods] + + self.assertEqual(len(options_routes), 1) + self.assertIsNone(options_routes[0].authorizer_name) + + def test_cors_normalization_groups_routes_across_operation_names(self): + authorizer = Authorizer( + authorizer_name="MyAuthorizer", + type="request", + payload_version="1.0", + ) + routes = [ + Route( + function_name="func", + path="/x", + methods=["GET"], + operation_name="GetX", + authorizer_name="MyAuthorizer", + authorizer_object=authorizer, + ), + Route( + function_name="func", + path="/x", + methods=["POST"], + operation_name="PostX", + authorizer_name=None, + use_default_authorizer=False, + ), + ] + + actual = ApiCollector.normalize_cors_methods(routes, object()) + + options_routes = [route for route in actual if "OPTIONS" in route.methods] + self.assertEqual(len(options_routes), 1) + self.assertEqual(options_routes[0].operation_name, "PostX") + self.assertIsNone(options_routes[0].authorizer_object) + + @parameterized.expand( + [ + ("protected_first", ["GET", "POST"]), + ("unprotected_first", ["POST", "GET"]), + ] + ) + def test_cors_synthesis_prefers_route_without_authorizer(self, _, method_order): + collector = ApiCollector() + authorizer = Authorizer( + authorizer_name="MyAuthorizer", + type="request", + payload_version="1.0", + ) + routes_by_method = { + "GET": Route( + function_name="func", + path="/x", + methods=["GET"], + authorizer_name="MyAuthorizer", + ), + "POST": Route( + function_name="func", + path="/x", + methods=["POST"], + authorizer_name=None, + use_default_authorizer=False, + ), + } + + collector.add_authorizers("api", {"MyAuthorizer": authorizer}) + collector.add_routes("api", [routes_by_method[method] for method in method_order]) + collector.cors = object() + + actual = collector.get_api().routes + options_routes = [route for route in actual if "OPTIONS" in route.methods] + get_route = next(route for route in actual if "GET" in route.methods) + + self.assertEqual(len(options_routes), 1) + self.assertIn("POST", options_routes[0].methods) + self.assertIsNone(options_routes[0].authorizer_name) + self.assertIsNone(options_routes[0].authorizer_object) + self.assertNotIn("OPTIONS", get_route.methods) + self.assertEqual(get_route.authorizer_name, "MyAuthorizer") + self.assertIs(get_route.authorizer_object, authorizer) + + def test_cors_synthesis_falls_back_to_first_route_when_all_routes_are_authorized(self): + first_authorizer = Authorizer( + authorizer_name="FirstAuthorizer", + type="request", + payload_version="1.0", + ) + second_authorizer = Authorizer( + authorizer_name="SecondAuthorizer", + type="request", + payload_version="1.0", + ) + routes = [ + Route( + function_name="func", + path="/x", + methods=["GET"], + authorizer_name="FirstAuthorizer", + authorizer_object=first_authorizer, + ), + Route( + function_name="func", + path="/x", + methods=["POST"], + authorizer_name="SecondAuthorizer", + authorizer_object=second_authorizer, + ), + ] + + actual = ApiCollector.normalize_cors_methods(ApiCollector.dedupe_function_routes(routes), object()) + options_routes = [route for route in actual if "OPTIONS" in route.methods] + + self.assertEqual(len(options_routes), 1) + self.assertIn("GET", options_routes[0].methods) + self.assertIs(options_routes[0].authorizer_object, first_authorizer) + + def test_preserves_payload_format_version_when_merging_routes(self): + routes = [ + Route( + function_name="func", + path="/x", + methods=["ANY"], + event_type=Route.HTTP, + authorizer_name=None, + ), + Route( + function_name="func", + path="/x", + methods=["GET"], + event_type=Route.HTTP, + payload_format_version="1.0", + authorizer_name=None, + ), + ] + + actual = ApiCollector.dedupe_function_routes(routes) + + self.assertEqual(len(actual), 1) + self.assertEqual(actual[0].payload_format_version, "1.0") + + def test_get_api_preserves_explicit_unauthorized_options_with_cors(self): + collector = ApiCollector() + + authorizer = Authorizer( + authorizer_name="MyAuthorizer", + type="request", + payload_version="1.0", + ) + + collector.add_authorizers("api", {"MyAuthorizer": authorizer}) + collector.set_default_authorizer("api", "MyAuthorizer") + + collector.add_routes( + "api", + [ + Route( + function_name="func", + path="/{proxy+}", + methods=["ANY"], + ), + Route( + function_name="func", + path="/{proxy+}", + methods=["OPTIONS"], + authorizer_name=None, + use_default_authorizer=False, + ), + ], + ) + + collector.cors = object() + + api = collector.get_api() + + options_routes = [route for route in api.routes if "OPTIONS" in route.methods] + get_routes = [route for route in api.routes if "GET" in route.methods] + + self.assertEqual(len(options_routes), 1) + self.assertIsNone(options_routes[0].authorizer_name) + + self.assertEqual(len(get_routes), 1) + self.assertEqual(get_routes[0].authorizer_name, "MyAuthorizer") + self.assertIs(get_routes[0].authorizer_object, authorizer) diff --git a/tests/unit/commands/local/lib/test_api_provider.py b/tests/unit/commands/local/lib/test_api_provider.py index 246c03f673..d0fcec8601 100644 --- a/tests/unit/commands/local/lib/test_api_provider.py +++ b/tests/unit/commands/local/lib/test_api_provider.py @@ -9,6 +9,7 @@ from samcli.lib.providers.api_provider import ApiProvider from samcli.lib.providers.sam_api_provider import SamApiProvider from samcli.lib.providers.cfn_api_provider import CfnApiProvider +from samcli.local.apigw.route import Route class TestApiProvider_init(TestCase): @@ -240,6 +241,82 @@ def test_apis_in_child_stack_overridden_by_apis_in_parents_within_implicit_or_ex ] self.assertEqual(SamApiProvider.merge_routes(collector), [route1]) + def test_preserved_route_does_not_propagate_payload_format_version_to_any(self): + options_route = Route( + function_name="func", + path="/x", + methods=["OPTIONS"], + event_type=Route.HTTP, + payload_format_version="1.0", + authorizer_name=None, + use_default_authorizer=False, + ) + any_route = Route( + function_name="func", + path="/x", + methods=["ANY"], + event_type=Route.HTTP, + authorizer_name="MyAuth", + ) + collector = [(SamApiProvider.IMPLICIT_HTTP_API_RESOURCE_ID, [options_route, any_route])] + + actual = SamApiProvider.merge_routes(collector) + + self.assertEqual(len(actual), 2) + self.assertTrue(any(route is options_route for route in actual)) + self.assertTrue(any(route is any_route for route in actual)) + self.assertEqual(options_route.payload_format_version, "1.0") + self.assertIsNone(any_route.payload_format_version) + + def test_preserves_explicit_authorizer_intent_when_operation_names_differ(self): + options_route = Route( + function_name="func", + path="/x", + methods=["OPTIONS"], + operation_name="Preflight", + authorizer_name=None, + use_default_authorizer=False, + ) + any_route = Route( + function_name="func", + path="/x", + methods=["ANY"], + authorizer_name="MyAuth", + ) + collector = [("Api1", [options_route, any_route])] + + actual = SamApiProvider.merge_routes(collector) + + self.assertEqual(len(actual), 2) + self.assertTrue(any(route is options_route for route in actual)) + self.assertTrue(any(route is any_route for route in actual)) + + def test_overriding_any_inherits_payload_format_version(self): + options_route = Route( + function_name="func", + path="/x", + methods=["OPTIONS"], + event_type=Route.HTTP, + payload_format_version="1.0", + authorizer_name=None, + use_default_authorizer=False, + ) + any_route = Route( + function_name="func", + path="/x", + methods=["ANY"], + event_type=Route.HTTP, + authorizer_name=None, + use_default_authorizer=False, + ) + collector = [(SamApiProvider.IMPLICIT_HTTP_API_RESOURCE_ID, [options_route, any_route])] + + actual = SamApiProvider.merge_routes(collector) + + self.assertEqual(len(actual), 1) + self.assertIs(actual[0], any_route) + self.assertEqual(any_route.payload_format_version, "1.0") + class TestApiProvider_check_implicit_api_resource_ids(TestCase): @patch("samcli.lib.providers.sam_base_provider.SamBaseProvider.get_template") diff --git a/tests/unit/commands/local/lib/test_sam_api_provider.py b/tests/unit/commands/local/lib/test_sam_api_provider.py index 9dbed45a5b..507cbe0d37 100644 --- a/tests/unit/commands/local/lib/test_sam_api_provider.py +++ b/tests/unit/commands/local/lib/test_sam_api_provider.py @@ -7,6 +7,7 @@ from parameterized import parameterized from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException +from samcli.lib.providers.api_collector import ApiCollector from samcli.lib.providers.api_provider import ApiProvider from samcli.lib.providers.provider import Cors, Stack from samcli.lib.providers.sam_api_provider import SamApiProvider @@ -591,6 +592,29 @@ def test_must_prefer_implicit_with_any_method(self): provider = ApiProvider(make_mock_stacks_from_template(self.template)) self.assertCountEqual(expected_routes, provider.routes) + def test_must_prefer_implicit_any_for_same_function_with_same_authorizer_intent(self): + implicit_routes = { + "Event1": { + "Type": "Api", + "Properties": { + "Path": "/path", + "Method": "ANY", + }, + } + } + + explicit_routes = [Route(path="/path", methods=["GET"], function_name="ImplicitFunc")] + + self.template["Resources"]["Api1"]["Properties"]["DefinitionBody"] = make_swagger(explicit_routes) + self.template["Resources"]["ImplicitFunc"]["Properties"]["Events"] = implicit_routes + + collector = ApiCollector() + SamApiProvider().extract_resources(make_mock_stacks_from_template(self.template), collector) + + expected_routes = [Route(path="/path", methods=["ANY"], function_name="ImplicitFunc")] + + self.assertCountEqual(expected_routes, collector.routes) + def test_with_any_method_on_both(self): implicit_routes = { "Event1": { @@ -1850,6 +1874,345 @@ def test_global_cors(self): class TestSamApiUsingAuthorizers(TestCase): + def test_cors_synthesis_prefers_swagger_method_without_authorizer(self): + swagger = make_swagger( + [ + Route(path="/x", methods=["GET"], function_name="SamFunc1"), + Route(path="/x", methods=["POST"], function_name="SamFunc1"), + ] + ) + swagger["paths"]["/x"]["GET"]["security"] = [{"MyAuthorizer": []}] + swagger["paths"]["/x"]["POST"]["security"] = [] + authorizer_arn = "arn:aws:lambda:us-east-1:123456789012:function:AuthFunc" + + template = { + "Resources": { + "Api1": { + "Type": "AWS::Serverless::Api", + "Properties": { + "StageName": "Prod", + "Cors": "'*'", + "DefinitionBody": swagger, + "Auth": { + "Authorizers": { + "MyAuthorizer": { + "FunctionArn": authorizer_arn, + "FunctionPayloadType": "REQUEST", + } + } + }, + }, + }, + "SamFunc1": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + }, + }, + "AuthFunc": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + }, + }, + } + } + + provider = ApiProvider(make_mock_stacks_from_template(template)) + + options_routes = [route for route in provider.routes if "OPTIONS" in route.methods] + get_route = next(route for route in provider.routes if "GET" in route.methods) + + self.assertEqual(len(options_routes), 1) + self.assertIn("POST", options_routes[0].methods) + self.assertIsNone(options_routes[0].authorizer_name) + self.assertIsNone(options_routes[0].authorizer_object) + self.assertNotIn("OPTIONS", get_route.methods) + self.assertEqual(get_route.authorizer_name, "MyAuthorizer") + self.assertIsInstance(get_route.authorizer_object, LambdaAuthorizer) + + def test_any_authorizer_applies_to_swagger_method_without_security(self): + swagger = make_swagger([Route(path="/x", methods=["GET"], function_name="SamFunc1")]) + authorizer_arn = "arn:aws:lambda:us-east-1:123456789012:function:AuthFunc" + + template = { + "Resources": { + "Api1": { + "Type": "AWS::Serverless::Api", + "Properties": { + "StageName": "Prod", + "DefinitionBody": swagger, + "Auth": { + "Authorizers": { + "MyAuthorizer": { + "FunctionArn": authorizer_arn, + "FunctionPayloadType": "REQUEST", + } + } + }, + }, + }, + "SamFunc1": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + "Events": { + "Any": { + "Type": "Api", + "Properties": { + "Path": "/x", + "Method": "ANY", + "RestApiId": "Api1", + "Auth": {"Authorizer": "MyAuthorizer"}, + }, + } + }, + }, + }, + "AuthFunc": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + }, + }, + } + } + + provider = ApiProvider(make_mock_stacks_from_template(template)) + + get_routes = [route for route in provider.routes if "GET" in route.methods] + + self.assertEqual(len(get_routes), 1) + self.assertEqual(get_routes[0].authorizer_name, "MyAuthorizer") + self.assertIsInstance(get_routes[0].authorizer_object, LambdaAuthorizer) + + def test_preflight_operation_id_preserves_unauthenticated_options(self): + swagger = make_swagger([Route(path="/x", methods=["OPTIONS"], function_name="SamFunc1")]) + swagger["paths"]["/x"]["OPTIONS"].update({"operationId": "Preflight", "security": []}) + authorizer_arn = "arn:aws:lambda:us-east-1:123456789012:function:AuthFunc" + + template = { + "Resources": { + "Api1": { + "Type": "AWS::Serverless::Api", + "Properties": { + "StageName": "Prod", + "Cors": "'*'", + "DefinitionBody": swagger, + "Auth": { + "Authorizers": { + "MyAuthorizer": { + "FunctionArn": authorizer_arn, + "FunctionPayloadType": "REQUEST", + } + } + }, + }, + }, + "SamFunc1": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + "Events": { + "Any": { + "Type": "Api", + "Properties": { + "Path": "/x", + "Method": "ANY", + "RestApiId": "Api1", + "Auth": {"Authorizer": "MyAuthorizer"}, + }, + } + }, + }, + }, + "AuthFunc": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + }, + }, + } + } + + provider = ApiProvider(make_mock_stacks_from_template(template)) + + options_routes = [route for route in provider.routes if "OPTIONS" in route.methods] + protected_routes = [route for route in provider.routes if route.authorizer_name == "MyAuthorizer"] + + self.assertEqual(len(provider.routes), 2) + self.assertEqual(len(options_routes), 1) + self.assertEqual(len(protected_routes), 1) + protected_route = protected_routes[0] + self.assertEqual(options_routes[0].methods, ["OPTIONS"]) + self.assertEqual(options_routes[0].operation_name, "Preflight") + self.assertIsNone(options_routes[0].authorizer_name) + self.assertIsNone(options_routes[0].authorizer_object) + self.assertFalse(options_routes[0].use_default_authorizer) + self.assertNotIn("OPTIONS", protected_route.methods) + self.assertEqual(set(protected_route.methods), set(Route.ANY_HTTP_METHODS) - {"OPTIONS"}) + self.assertIsNone(protected_route.operation_name) + self.assertEqual(protected_route.authorizer_name, "MyAuthorizer") + self.assertIsInstance(protected_route.authorizer_object, LambdaAuthorizer) + + def test_extract_resources_preserves_options_when_declared_before_any(self): + template = { + "Resources": { + "SamFunc1": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + "Events": { + "Options": { + "Type": "Api", + "Properties": { + "Path": "/{proxy+}", + "Method": "OPTIONS", + "Auth": {"Authorizer": "NONE"}, + }, + }, + "Any": { + "Type": "Api", + "Properties": { + "Path": "/{proxy+}", + "Method": "ANY", + "Auth": {"Authorizer": "MyAuthorizer"}, + }, + }, + }, + }, + } + } + } + + collector = ApiCollector() + + SamApiProvider().extract_resources( + make_mock_stacks_from_template(template), + collector, + ) + + options_routes = [route for route in collector.routes if route.methods == ["OPTIONS"]] + + self.assertEqual(len(options_routes), 1) + self.assertIsNone(options_routes[0].authorizer_name) + self.assertFalse(options_routes[0].use_default_authorizer) + + def test_extract_resources_does_not_propagate_options_payload_version_to_any(self): + template = { + "Resources": { + "SamFunc1": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + "Events": { + "Options": { + "Type": "HttpApi", + "Properties": { + "Path": "/x", + "Method": "OPTIONS", + "PayloadFormatVersion": "1.0", + "Auth": {"Authorizer": "NONE"}, + }, + }, + "Any": { + "Type": "HttpApi", + "Properties": { + "Path": "/x", + "Method": "ANY", + "Auth": {"Authorizer": "MyAuth"}, + }, + }, + }, + }, + } + } + } + collector = ApiCollector() + + SamApiProvider().extract_resources( + make_mock_stacks_from_template(template), + collector, + ) + + self.assertEqual(len(collector.routes), 2) + options_route = next(route for route in collector.routes if route.methods == ["OPTIONS"]) + any_route = next(route for route in collector.routes if set(route.methods) == set(Route.ANY_HTTP_METHODS)) + + self.assertEqual(options_route.payload_format_version, "1.0") + self.assertIsNone(options_route.authorizer_name) + self.assertFalse(options_route.use_default_authorizer) + self.assertIsNone(any_route.payload_format_version) + self.assertEqual(any_route.authorizer_name, "MyAuth") + + def test_extract_resources_preserves_explicit_options_with_implicit_any(self): + template = { + "Resources": { + "Api1": { + "Type": "AWS::Serverless::Api", + "Properties": { + "StageName": "Prod", + }, + }, + "SamFunc1": { + "Type": "AWS::Serverless::Function", + "Properties": { + "CodeUri": "/usr/foo/bar", + "Runtime": "python3.11", + "Handler": "index.handler", + "Events": { + "Options": { + "Type": "Api", + "Properties": { + "Path": "/{proxy+}", + "Method": "OPTIONS", + "RestApiId": "Api1", + "Auth": {"Authorizer": "NONE"}, + }, + }, + "Any": { + "Type": "Api", + "Properties": { + "Path": "/{proxy+}", + "Method": "ANY", + "Auth": {"Authorizer": "MyAuthorizer"}, + }, + }, + }, + }, + }, + } + } + + collector = ApiCollector() + + SamApiProvider().extract_resources( + make_mock_stacks_from_template(template), + collector, + ) + + options_routes = [route for route in collector.routes if route.methods == ["OPTIONS"]] + + self.assertEqual(len(options_routes), 1) + self.assertIsNone(options_routes[0].authorizer_name) + self.assertFalse(options_routes[0].use_default_authorizer) + @parameterized.expand( [(SamApiProvider()._extract_from_serverless_api,), (SamApiProvider()._extract_from_serverless_http,)] )