import copy from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple from boto3 import Session from samtranslator.feature_toggle.feature_toggle import ( FeatureToggle, FeatureToggleDefaultConfigProvider, ) from samtranslator.internal.types import GetManagedPolicyMap from samtranslator.intrinsics.actions import FindInMapAction from samtranslator.intrinsics.resolver import IntrinsicsResolver from samtranslator.intrinsics.resource_refs import SupportedResourceReferences from samtranslator.metrics.method_decorator import MetricsMethodWrapperSingleton from samtranslator.metrics.metrics import DummyMetricsPublisher, Metrics from samtranslator.model import Resource, ResourceResolver, ResourceTypeResolver, sam_resources from samtranslator.model.api.api_generator import SharedApiUsagePlan from samtranslator.model.eventsources.push import Api from samtranslator.model.exceptions import ( DuplicateLogicalIdException, ExceptionWithMessage, InvalidDocumentException, InvalidEventException, InvalidResourceException, InvalidTemplateException, ) from samtranslator.model.preferences.deployment_preference_collection import DeploymentPreferenceCollection from samtranslator.model.sam_resources import SamConnector from samtranslator.parser.parser import Parser from samtranslator.plugins import BasePlugin, LifeCycleEvents from samtranslator.plugins.api.default_definition_body_plugin import DefaultDefinitionBodyPlugin from samtranslator.plugins.application.serverless_app_plugin import ServerlessAppPlugin from samtranslator.plugins.globals.globals_plugin import GlobalsPlugin from samtranslator.plugins.policies.policy_templates_plugin import PolicyTemplatesForResourcePlugin from samtranslator.plugins.sam_plugins import SamPlugins from samtranslator.policy_template_processor.processor import PolicyTemplatesProcessor from samtranslator.sdk.parameter import SamParameterValues from samtranslator.translator.arn_generator import ArnGenerator from samtranslator.translator.verify_logical_id import verify_unique_logical_id from samtranslator.utils.actions import ResolveDependsOn from samtranslator.utils.traverse import traverse from samtranslator.validator.value_validator import sam_expect class Translator: """Translates SAM templates into CloudFormation templates""" def __init__( self, managed_policy_map: Optional[Dict[str, str]], sam_parser: Parser, plugins: Optional[List[BasePlugin]] = None, boto_session: Optional[Session] = None, metrics: Optional[Metrics] = None, ) -> None: """ :param dict managed_policy_map: Map of managed policy names to the ARNs :param sam_parser: Instance of a SAM Parser :param list of samtranslator.plugins.BasePlugin plugins: List of plugins to be installed in the translator, in addition to the default ones. """ self.managed_policy_map = managed_policy_map self.plugins = plugins self.sam_parser = sam_parser self.feature_toggle: Optional[FeatureToggle] = None self.boto_session = boto_session self.metrics = metrics if metrics else Metrics("ServerlessTransform", DummyMetricsPublisher()) MetricsMethodWrapperSingleton.set_instance(self.metrics) self.document_errors: List[ExceptionWithMessage] = [] if self.boto_session: ArnGenerator.BOTO_SESSION_REGION_NAME = self.boto_session.region_name def _get_function_names( self, resource_dict: Dict[str, Any], intrinsics_resolver: IntrinsicsResolver ) -> Dict[str, str]: """ :param resource_dict: AWS::Serverless::Function resource is provided as input :param intrinsics_resolver: to resolve intrinsics for function_name :return: a dictionary containing api_logical_id as the key and concatenated String of all function_names associated with this api as the value """ if resource_dict.get("Type", "").strip() == "AWS::Serverless::Function": events_properties = resource_dict.get("Properties", {}).get("Events", {}) events = list(events_properties.values()) if events_properties else [] for item in events: # If the function event type is `Api` then gets the function name and # adds to the function_names dict with key as the api_name and value as the function_name item_properties = item.get("Properties", {}) if item.get("Type") == "Api" and item_properties.get("RestApiId"): rest_api = item_properties.get("RestApiId") api_name = Api.get_rest_api_id_string(rest_api) if not isinstance(api_name, str): continue raw_function_name = resource_dict.get("Properties", {}).get("FunctionName") resolved_function_name = intrinsics_resolver.resolve_parameter_refs( copy.deepcopy(raw_function_name) ) if not resolved_function_name: continue self.function_names.setdefault(api_name, "") self.function_names[api_name] += str(resolved_function_name) return self.function_names def translate( # noqa: PLR0912, PLR0915 self, sam_template: Dict[str, Any], parameter_values: Dict[str, Any], feature_toggle: Optional[FeatureToggle] = None, passthrough_metadata: Optional[bool] = False, get_managed_policy_map: Optional[GetManagedPolicyMap] = None, ) -> Dict[str, Any]: """Loads the SAM resources from the given SAM manifest, replaces them with their corresponding CloudFormation resources, and returns the resulting CloudFormation template. :param dict sam_template: the SAM manifest, as loaded by json.load() or yaml.load(), or as provided by \ CloudFormation transforms. :param dict parameter_values: Map of template parameter names to their values. It is a required parameter that should at least be an empty map. By providing an empty map, the caller explicitly opts-into the idea that some functionality that relies on resolving parameter references might not work as expected (ex: auto-creating new Lambda Version when CodeUri contains reference to template parameter). This is why this parameter is required :returns: a copy of the template with SAM resources replaced with the corresponding CloudFormation, which may \ be dumped into a valid CloudFormation JSON or YAML template """ self.feature_toggle = feature_toggle or FeatureToggle( FeatureToggleDefaultConfigProvider(), stage=None, account_id=None, region=None ) self.function_names: Dict[Any, Any] = {} self.redeploy_restapi_parameters = {} sam_parameter_values = SamParameterValues(parameter_values) sam_parameter_values.add_default_parameter_values(sam_template) sam_parameter_values.add_pseudo_parameter_values(self.boto_session) parameter_values = sam_parameter_values.parameter_values # Create & Install plugins sam_plugins = prepare_plugins(self.plugins, parameter_values) self.sam_parser.parse(sam_template=sam_template, parameter_values=parameter_values, sam_plugins=sam_plugins) # replaces Connectors attributes with serverless Connector resources resources = sam_template.get("Resources", {}) embedded_connectors = self._get_embedded_connectors(resources) connector_resources = self._update_resources(embedded_connectors) resources.update(connector_resources) self._delete_connectors_attribute(resources) template = copy.deepcopy(sam_template) macro_resolver = ResourceTypeResolver(sam_resources) intrinsics_resolver = IntrinsicsResolver(parameter_values) # ResourceResolver is used by connector, its "resources" will be # updated in-place by other transforms so connector transform # can see the transformed resources. resource_resolver = ResourceResolver(template.get("Resources", {})) mappings_resolver = IntrinsicsResolver( template.get("Mappings", {}), {FindInMapAction.intrinsic_name: FindInMapAction()} ) deployment_preference_collection = DeploymentPreferenceCollection() supported_resource_refs = SupportedResourceReferences() shared_api_usage_plan = SharedApiUsagePlan() changed_logical_ids = {} route53_record_set_groups: Dict[Any, Any] = {} for logical_id, resource_dict in self._get_resources_to_iterate(sam_template, macro_resolver): try: macro = macro_resolver.resolve_resource_type(resource_dict).from_dict( logical_id, resource_dict, sam_plugins=sam_plugins ) kwargs = macro.resources_to_link(sam_template["Resources"]) kwargs["managed_policy_map"] = self.managed_policy_map kwargs["get_managed_policy_map"] = get_managed_policy_map kwargs["intrinsics_resolver"] = intrinsics_resolver kwargs["mappings_resolver"] = mappings_resolver kwargs["deployment_preference_collection"] = deployment_preference_collection kwargs["conditions"] = template.get("Conditions") kwargs["resource_resolver"] = resource_resolver kwargs["original_template"] = sam_template # add the value of FunctionName property if the function is referenced with the api resource self.redeploy_restapi_parameters["function_names"] = self._get_function_names( resource_dict, intrinsics_resolver ) kwargs["redeploy_restapi_parameters"] = self.redeploy_restapi_parameters kwargs["shared_api_usage_plan"] = shared_api_usage_plan kwargs["feature_toggle"] = self.feature_toggle kwargs["route53_record_set_groups"] = route53_record_set_groups translated = macro.to_cloudformation(**kwargs) supported_resource_refs = macro.get_resource_references(translated, supported_resource_refs) # Some resources mutate their logical ids. Track those to change all references to them: if logical_id != macro.logical_id: changed_logical_ids[logical_id] = macro.logical_id del template["Resources"][logical_id] for resource in translated: if verify_unique_logical_id(resource, sam_template["Resources"]): # For each generated resource, pass through existing metadata that may exist on the original SAM resource. _r = resource.to_dict() if ( resource_dict.get("Metadata") and passthrough_metadata and not template["Resources"].get(resource.logical_id) ): _r[resource.logical_id]["Metadata"] = resource_dict["Metadata"] template["Resources"].update(_r) else: self.document_errors.append( DuplicateLogicalIdException(logical_id, resource.logical_id, resource.resource_type) ) except (InvalidResourceException, InvalidEventException, InvalidTemplateException) as e: self.document_errors.append(e) if deployment_preference_collection.any_enabled(): template["Resources"].update(deployment_preference_collection.get_codedeploy_application().to_dict()) if deployment_preference_collection.needs_resource_condition(): new_conditions = deployment_preference_collection.create_aggregate_deployment_condition() if new_conditions: template.get("Conditions", {}).update(new_conditions) if not deployment_preference_collection.can_skip_service_role(): template["Resources"].update(deployment_preference_collection.get_codedeploy_iam_role().to_dict()) for logical_id in deployment_preference_collection.enabled_logical_ids(): try: template["Resources"].update( deployment_preference_collection.deployment_group(logical_id).to_dict() ) except InvalidResourceException as e: self.document_errors.append(e) # Run the after-transform plugin target try: sam_plugins.act(LifeCycleEvents.after_transform_template, template) except (InvalidDocumentException, InvalidResourceException, InvalidTemplateException) as e: self.document_errors.append(e) # Cleanup if "Transform" in template: del template["Transform"] if len(self.document_errors) == 0: resolveDependsOn = ResolveDependsOn(resolution_data=changed_logical_ids) # Initializes ResolveDependsOn template = traverse(template, [resolveDependsOn]) template = intrinsics_resolver.resolve_sam_resource_id_refs(template, changed_logical_ids) return intrinsics_resolver.resolve_sam_resource_refs(template, supported_resource_refs) raise InvalidDocumentException(self.document_errors) # private methods def _get_resources_to_iterate( self, sam_template: Dict[str, Any], macro_resolver: ResourceTypeResolver ) -> List[Tuple[str, Dict[str, Any]]]: """ Returns a list of resources to iterate, order them based on the following order: 1. AWS::Serverless::Function - because API Events need to modify the corresponding Serverless::Api resource. 2. AWS::Serverless::StateMachine - because API Events need to modify the corresponding Serverless::Api resource. 3. AWS::Serverless::Api 4. Anything else 5. AWS::Serverless::Connector - because connector profiles only work with raw CloudFormation resources This is necessary because a Function or State Machine resource with API Events will modify the API resource's Swagger JSON. Therefore API resource needs to be parsed only after all the Swagger modifications are complete. :param dict sam_template: SAM template :param macro_resolver: Resolver that knows if a resource can be processed or not :return list: List containing tuple of (logicalId, resource_dict) in the order of processing """ functions = [] statemachines = [] apis = [] others = [] connectors = [] resources = sam_template["Resources"] for logicalId, resource in resources.items(): data = (logicalId, resource) # Skip over the resource if it is not a SAM defined Resource if not macro_resolver.can_resolve(resource): continue if resource["Type"] == "AWS::Serverless::Function": functions.append(data) elif resource["Type"] == "AWS::Serverless::StateMachine": statemachines.append(data) elif resource["Type"] in ("AWS::Serverless::Api", "AWS::Serverless::HttpApi"): apis.append(data) elif resource["Type"] == "AWS::Serverless::Connector": connectors.append(data) else: others.append(data) return functions + statemachines + apis + others + connectors @staticmethod def _update_resources(connectors_list: List[Resource]) -> Dict[str, Any]: connector_resources = {} for connector in connectors_list: connector_resources.update(connector.to_dict()) return connector_resources @staticmethod def _delete_connectors_attribute(resources: Dict[str, Any]) -> None: for resource in resources.values(): if "Connectors" not in resource: continue del resource["Connectors"] def _get_embedded_connectors(self, resources: Dict[str, Any]) -> List[Resource]: """ Loops through the SAM Template resources to find any connectors that have been attached to the resources. Converts those attached connectors into Connector resources and returns a list of them :param dict resources: Dict of resources from the SAM template :return List[SamConnector]: List of the generated SAM Connectors """ connectors = [] # Loop through the resources in the template and see if any connectors have been attached for source_logical_id, resource in resources.items(): if "Connectors" not in resource: continue try: sam_expect( resource.get("Connectors"), source_logical_id, f"{source_logical_id}.Connectors", is_resource_attribute=True, ).to_be_a_map() except InvalidResourceException as e: self.document_errors.append(e) continue for connector_logical_id, connector_dict in resource["Connectors"].items(): try: full_connector_logical_id = source_logical_id + connector_logical_id # can't use sam_expect since this is neither a property nor a resource attribute if not isinstance(connector_dict, dict): raise InvalidResourceException( full_connector_logical_id, f"{source_logical_id}.{full_connector_logical_id} should be a map.", ) generated_connector = self._get_generated_connector( source_logical_id, full_connector_logical_id, connector_logical_id, connector_dict, ) if not verify_unique_logical_id(generated_connector, resources): raise DuplicateLogicalIdException( source_logical_id, full_connector_logical_id, generated_connector.resource_type ) connectors.append(generated_connector) except (InvalidResourceException, DuplicateLogicalIdException) as e: self.document_errors.append(e) return connectors def _get_generated_connector( self, source_logical_id: str, full_connector_logical_id: str, connector_logical_id: str, connector_dict: Dict[str, Any], ) -> Resource: """ Generates the connector resource from the embedded connector :param str source_logical_id: Logical id of the resource the connector is attached to :param str full_connector_logical_id: source_logical_id + connector_logical_id :param str connector_logical_id: Logical id of the connector defined by the user :param dict connector_dict: The properties of the connector including the Destination, Permissions and optionally the SourceReference :return: The generated SAMConnector resource """ connector = copy.deepcopy(connector_dict) connector["Type"] = SamConnector.resource_type properties = sam_expect( connector.get("Properties"), source_logical_id, f"Connectors.{connector_logical_id}.Properties", is_resource_attribute=True, ).to_be_a_map() properties["Source"] = {"Id": source_logical_id} if "SourceReference" in properties: source_reference = sam_expect( properties.get("SourceReference"), source_logical_id, f"Connectors.{connector_logical_id}.Properties.SourceReference", ).to_be_a_map() # can't allow user to override the Id using SourceReference if "Id" in source_reference: raise InvalidResourceException(connector_logical_id, "'Id' shouldn't be defined in 'SourceReference'.") properties["Source"].update(source_reference) del properties["SourceReference"] return SamConnector.from_dict(full_connector_logical_id, connector) def prepare_plugins(plugins: Optional[List[BasePlugin]], parameters: Optional[Dict[str, Any]] = None) -> SamPlugins: """ Creates & returns a plugins object with the given list of plugins installed. In addition to the given plugins, we will also install a few "required" plugins that are necessary to provide complete support for SAM template spec. :param plugins: list of samtranslator.plugins.BasePlugin plugins: List of plugins to install :param parameters: Dictionary of parameter values :return samtranslator.plugins.SamPlugins: Instance of `SamPlugins` """ if parameters is None: parameters = {} required_plugins = [ DefaultDefinitionBodyPlugin(), make_implicit_rest_api_plugin(), make_implicit_http_api_plugin(), GlobalsPlugin(), make_policy_template_for_function_plugin(), ] plugins = plugins or [] # If a ServerlessAppPlugin does not yet exist, create one and add to the beginning of the required plugins list. if not any(isinstance(plugin, ServerlessAppPlugin) for plugin in plugins): required_plugins.insert(0, ServerlessAppPlugin(parameters=parameters)) # Execute customer's plugins first before running SAM plugins. It is very important to retain this order because # other plugins will be dependent on this ordering. return SamPlugins(plugins + required_plugins) if TYPE_CHECKING: from samtranslator.plugins.api.implicit_http_api_plugin import ImplicitHttpApiPlugin from samtranslator.plugins.api.implicit_rest_api_plugin import ImplicitRestApiPlugin def make_implicit_rest_api_plugin() -> "ImplicitRestApiPlugin": # This is necessary to prevent a circular dependency on imports when loading package from samtranslator.plugins.api.implicit_rest_api_plugin import ImplicitRestApiPlugin return ImplicitRestApiPlugin() def make_implicit_http_api_plugin() -> "ImplicitHttpApiPlugin": # This is necessary to prevent a circular dependency on imports when loading package from samtranslator.plugins.api.implicit_http_api_plugin import ImplicitHttpApiPlugin return ImplicitHttpApiPlugin() def make_policy_template_for_function_plugin() -> PolicyTemplatesForResourcePlugin: """ Constructs an instance of policy templates processing plugin using default policy templates JSON data :return plugins.policies.policy_templates_plugin.PolicyTemplatesForResourcePlugin: Instance of the plugin """ policy_templates = PolicyTemplatesProcessor.get_default_policy_templates_json() processor = PolicyTemplatesProcessor(policy_templates) return PolicyTemplatesForResourcePlugin(processor)