""" Terraform translate to CFN implementation This method contains the logic required to translate the `terraform show` JSON output into a Cloudformation template """ import hashlib import logging from typing import Any, Dict, Iterator, List, Tuple, Type, Union from samcli.hook_packages.terraform.hooks.prepare.constants import ( CFN_CODE_PROPERTIES, REMOTE_DUMMY_VALUE, SAM_METADATA_RESOURCE_NAME_ATTRIBUTE, TF_AWS_API_GATEWAY_INTEGRATION, TF_AWS_API_GATEWAY_INTEGRATION_RESPONSE, TF_AWS_API_GATEWAY_METHOD, TF_AWS_API_GATEWAY_REST_API, TF_AWS_API_GATEWAY_V2_API, ) from samcli.hook_packages.terraform.hooks.prepare.enrich import enrich_resources_and_generate_makefile from samcli.hook_packages.terraform.hooks.prepare.property_builder import ( RESOURCE_TRANSLATOR_MAPPING, PropertyBuilderMapping, ) from samcli.hook_packages.terraform.hooks.prepare.resource_linking import ( _build_module, _resolve_resource_attribute, ) from samcli.hook_packages.terraform.hooks.prepare.resources.apigw import ( RESTAPITranslationValidator, add_integration_responses_to_methods, add_integrations_to_methods, ) from samcli.hook_packages.terraform.hooks.prepare.resources.internal import INTERNAL_PREFIX from samcli.hook_packages.terraform.hooks.prepare.resources.resource_links import ( MULTIPLE_DESTINATIONS_RESOURCE_LINKS, RESOURCE_LINKS, ) from samcli.hook_packages.terraform.hooks.prepare.resources.resource_properties import get_resource_property_mapping from samcli.hook_packages.terraform.hooks.prepare.types import ( CodeResourceProperties, ConstantValue, References, ResolvedReference, ResourceProperties, ResourceTranslationProperties, ResourceTranslationValidator, SamMetadataResource, TFModule, TFResource, ) from samcli.hook_packages.terraform.hooks.prepare.utilities import get_configuration_address from samcli.hook_packages.terraform.lib.utils import ( _calculate_configuration_attribute_value_hash, build_cfn_logical_id, get_sam_metadata_planned_resource_value_attribute, ) from samcli.lib.hook.exceptions import PrepareHookException from samcli.lib.utils.colors import Colored, Colors from samcli.lib.utils.resources import AWS_LAMBDA_FUNCTION as CFN_AWS_LAMBDA_FUNCTION SAM_METADATA_RESOURCE_TYPE = "null_resource" SAM_METADATA_NAME_PREFIX = "sam_metadata_" AWS_PROVIDER_NAME = "registry.terraform.io/hashicorp/aws" NULL_RESOURCE_PROVIDER_NAME = "registry.terraform.io/hashicorp/null" LOG = logging.getLogger(__name__) TRANSLATION_VALIDATORS: Dict[str, Type[ResourceTranslationValidator]] = { TF_AWS_API_GATEWAY_REST_API: RESTAPITranslationValidator, TF_AWS_API_GATEWAY_V2_API: RESTAPITranslationValidator, } def _get_modules(root_module: dict, root_tf_module: TFModule) -> Iterator[Tuple[dict, TFModule]]: """ Iterator helper method to find any child modules for processing. Parameters ---------- root_module: dict The root level planned values dictionary root_tf_module: TFModule The TFModule class representation of the configuration values Yields ------ Tuple[dict, TFModule] A tuple of the current module's planned values and TFModule representation of configuration values """ queue = [(root_module, root_tf_module)] while queue: modules = queue.pop(0) yield modules _add_child_modules_to_queue(*modules, queue) def _check_unresolvable_values(root_module: dict, root_tf_module: TFModule) -> None: """ Checks the planned values and configuration values if there are any properties that are unresolved, or unknown, until the Terraform project is applied. Parameters ---------- root_module: dict The root level planned values dictionary root_tf_module: TFModule The TFModule class representation of the configuration values """ for curr_module, curr_tf_module in _get_modules(root_module, root_tf_module): # iterate over resources for current module for resource in curr_module.get("resources", []): resource_type = resource.get("type") resource_name = resource.get("name") resource_mode = resource.get("mode") resource_mapper = RESOURCE_TRANSLATOR_MAPPING.get(resource_type) if not resource_mapper: continue resource_values = resource.get("values") resource_address = ( f"data.{resource_type}.{resource_name}" if resource_mode == "data" else f"{resource_type}.{resource_name}" ) config_resource_address = get_configuration_address(resource_address) config_resource = curr_tf_module.resources[config_resource_address] for prop_builder in resource_mapper.property_builder_mapping.values(): planned_values = prop_builder(resource_values, config_resource) config_values = prop_builder(config_resource.attributes, config_resource) if config_values and not planned_values: LOG.warning( Colored().color_log( msg="\nUnresolvable attributes discovered in project, " "run terraform apply to resolve them.\n", color=Colors.WARNING, ), extra=dict(markup=True), ) return def translate_to_cfn( tf_json: dict, output_directory_path: str, terraform_application_dir: str, project_root_dir: str ) -> dict: """ Translates the json output of a terraform show into CloudFormation Parameters ---------- tf_json: dict A terraform show json output output_directory_path: str the string path to write the metadata file and makefile terraform_application_dir: str the terraform configuration root module directory. project_root_dir: str the project root directory where terraform configurations, src code, and other modules exist Returns ------- dict The CloudFormation resulting from translating tf_json """ # setup root_module and cfn dict root_module = tf_json.get("planned_values", {}).get("root_module") cfn_dict: dict = {"AWSTemplateFormatVersion": "2010-09-09", "Resources": {}} if not root_module: return cfn_dict LOG.debug("Mapping Lambda functions to their corresponding layers.") input_vars: Dict[str, Union[ConstantValue, References]] = { var_name: ConstantValue(value=var_value.get("value")) for var_name, var_value in tf_json.get("variables", {}).items() } root_tf_module = _build_module("", tf_json.get("configuration", {}).get("root_module"), input_vars, None) # to map s3 object sources to respective functions later # this dictionary will map between the hash value of the S3 Bucket attributes, and a tuple of the planned value # source code path, and the configuration value of the source code path. s3_hash_to_source: Dict[str, Tuple[str, List[Union[ConstantValue, ResolvedReference]]]] = {} # map code/imageuri to Lambda resources # the key is the hash value of lambda code/imageuri # the value is the list of pair of the resource logical id, and the lambda cfn resource dict lambda_resources_to_code_map: Dict[str, List[Tuple[Dict, str]]] = {} sam_metadata_resources: List[SamMetadataResource] = [] resource_property_mapping: Dict[str, ResourceProperties] = get_resource_property_mapping() _check_unresolvable_values(root_module, root_tf_module) # create and iterate over queue of modules to handle child modules for curr_module, curr_tf_module in _get_modules(root_module, root_tf_module): curr_module_address = curr_module.get("address") # iterate over resources for current module resources = curr_module.get("resources", {}) for resource in resources: resource_provider = resource.get("provider_name") resource_type = resource.get("type") resource_values = resource.get("values") resource_full_address = resource.get("address") resource_name = resource.get("name") resource_mode = resource.get("mode") resource_address = ( f"data.{resource_type}.{resource_name}" if resource_mode == "data" else f"{resource_type}.{resource_name}" ) config_resource_address = get_configuration_address(resource_address) if config_resource_address not in curr_tf_module.resources: raise PrepareHookException( f"There is no configuration resource for resource address {resource_full_address} and " f"configuration address {config_resource_address}" ) config_resource = curr_tf_module.resources[config_resource_address] if ( resource_provider == NULL_RESOURCE_PROVIDER_NAME and resource_type == SAM_METADATA_RESOURCE_TYPE and resource_name.startswith(SAM_METADATA_NAME_PREFIX) ): _add_metadata_resource_to_metadata_list( SamMetadataResource(curr_module_address, resource, config_resource), resource, sam_metadata_resources, ) continue # only process supported provider if resource_provider != AWS_PROVIDER_NAME: continue # store S3 sources if resource_type == "aws_s3_object": s3_bucket = ( resource_values.get("bucket") if "bucket" in resource_values else _resolve_resource_attribute(config_resource, "bucket") ) s3_key = ( resource_values.get("key") if "key" in resource_values else _resolve_resource_attribute(config_resource, "key") ) obj_hash = _get_s3_object_hash(s3_bucket, s3_key) code_artifact = resource_values.get("source") config_code_artifact = ( code_artifact if code_artifact else _resolve_resource_attribute(config_resource, "source") ) s3_hash_to_source[obj_hash] = (code_artifact, config_code_artifact) resource_translator = RESOURCE_TRANSLATOR_MAPPING.get(resource_type) # resource type not supported if not resource_translator: continue # translate TF resource "values" to CFN properties LOG.debug("Processing resource %s", resource_full_address) translated_properties = _translate_properties( resource_values, resource_translator.property_builder_mapping, config_resource ) translated_resource: Dict = { "Type": resource_translator.cfn_name, "Properties": translated_properties, "Metadata": {"SamResourceId": resource_full_address}, } # Only set the SkipBuild metadata if it's a resource that can be built if resource_translator.cfn_name in CFN_CODE_PROPERTIES: translated_resource["Metadata"]["SkipBuild"] = True # build CFN logical ID from resource address logical_id = build_cfn_logical_id(resource_full_address) # Add resource to cfn dict if not translated_resource.get("Type", "").startswith(INTERNAL_PREFIX): # Internal resources are ones used for the purpose of translation, they are not real CFN resources. # These are usually resources that exist in other IaCs that don't map 1:1 with CFN resources, but their # properties need to be mapped to other, existing CFN resources. cfn_dict["Resources"][logical_id] = translated_resource resource_translation_properties = ResourceTranslationProperties( resource=resource, translated_resource=translated_resource, config_resource=config_resource, logical_id=logical_id, resource_full_address=resource_full_address, ) if resource_type in resource_property_mapping: resource_properties: ResourceProperties = resource_property_mapping[resource_type] resource_properties.collect(resource_translation_properties) if isinstance(resource_properties, CodeResourceProperties): resource_properties.add_lambda_resources_to_code_map( resource_translation_properties, translated_properties, lambda_resources_to_code_map ) if resource_type in TRANSLATION_VALIDATORS: validator = TRANSLATION_VALIDATORS[resource_type](resource=resource, config_resource=config_resource) validator.validate() # map s3 object sources to corresponding functions LOG.debug("Mapping S3 object sources to corresponding functions") _map_s3_sources_to_functions(s3_hash_to_source, cfn_dict.get("Resources", {}), lambda_resources_to_code_map) _handle_linking(resource_property_mapping) add_integrations_to_methods( resource_property_mapping.get(TF_AWS_API_GATEWAY_METHOD, ResourceProperties()).cfn_resources, resource_property_mapping.get(TF_AWS_API_GATEWAY_INTEGRATION, ResourceProperties()).cfn_resources, ) add_integration_responses_to_methods( resource_property_mapping.get(TF_AWS_API_GATEWAY_METHOD, ResourceProperties()).cfn_resources, resource_property_mapping.get(TF_AWS_API_GATEWAY_INTEGRATION_RESPONSE, ResourceProperties()).cfn_resources, ) if sam_metadata_resources: LOG.debug("Enrich the mapped resources with the sam metadata information and generate Makefile") enrich_resources_and_generate_makefile( sam_metadata_resources, cfn_dict.get("Resources", {}), output_directory_path, terraform_application_dir, lambda_resources_to_code_map, project_root_dir, ) else: LOG.debug("There is no sam metadata resources, no enrichment or Makefile is required") # check if there is still any dummy remote values for lambda resource imagesUri or S3 attributes _check_dummy_remote_values(cfn_dict.get("Resources", {})) return cfn_dict def _handle_linking(resource_property_mapping: Dict[str, ResourceProperties]) -> None: for link in RESOURCE_LINKS: link.linking_func( resource_property_mapping[link.source].terraform_config, resource_property_mapping[link.source].cfn_resources, resource_property_mapping[link.dest].terraform_resources, ) for multiple_destinations_link in MULTIPLE_DESTINATIONS_RESOURCE_LINKS: destinations: Dict[str, Dict] = {} for dest_resource_type in multiple_destinations_link.destinations: destinations = { **destinations, **resource_property_mapping[dest_resource_type].terraform_resources, } multiple_destinations_link.linking_func( resource_property_mapping[multiple_destinations_link.source].terraform_config, resource_property_mapping[multiple_destinations_link.source].cfn_resources, destinations, ) def _add_child_modules_to_queue(curr_module: Dict, curr_module_configuration: TFModule, modules_queue: List) -> None: """ Iterate over the children modules of current module and add each module with its related child module configuration to the modules_queue. Parameters ---------- curr_module: Dict The current module in the planned values curr_module_configuration: TFModule The current module configuration modules_queue: List The list of modules """ child_modules = curr_module.get("child_modules") if child_modules: for child_module in child_modules: config_child_module_address = ( get_configuration_address(child_module["address"]) if "address" in child_module else None ) module_name = ( config_child_module_address[config_child_module_address.rfind(".") + 1 :] if config_child_module_address else None ) child_tf_module = curr_module_configuration.child_modules.get(module_name) if module_name else None if child_tf_module is None: raise PrepareHookException( f"Module {config_child_module_address} exists in terraform planned_value, but does not exist " "in terraform configuration" ) modules_queue.append((child_module, child_tf_module)) def _add_metadata_resource_to_metadata_list( sam_metadata_resource: SamMetadataResource, sam_metadata_resource_planned_values: Dict, sam_metadata_resources: List[SamMetadataResource], ) -> None: """ Prioritize the metadata resources that has resource name value to overwrite the metadata resources that does not have resource name value. Parameters ---------- sam_metadata_resource: SamMetadataResource The mapped metadata resource sam_metadata_resource_planned_values: Dict The metadata resource in planned values section sam_metadata_resources: List[SamMetadataResource] The list of metadata resources """ if get_sam_metadata_planned_resource_value_attribute( sam_metadata_resource_planned_values, SAM_METADATA_RESOURCE_NAME_ATTRIBUTE ): sam_metadata_resources.append(sam_metadata_resource) else: sam_metadata_resources.insert(0, sam_metadata_resource) def _translate_properties( tf_properties: dict, property_builder_mapping: PropertyBuilderMapping, resource: TFResource ) -> dict: """ Translates the properties of a terraform resource into the equivalent properties of a CloudFormation resource Parameters ---------- tf_properties: dict The terraform properties to translate property_builder_mapping: PropertyBuilderMapping A mapping of the CloudFormation property name to a function for building that property resource: TFResource The terraform configuration resource that can be used to retrieve some attributes values if needed Returns ------- dict The CloudFormation properties resulting from translating tf_properties """ cfn_properties = {} for cfn_property_name, cfn_property_builder in property_builder_mapping.items(): cfn_property_value = cfn_property_builder(tf_properties, resource) if cfn_property_value is not None: cfn_properties[cfn_property_name] = cfn_property_value return cfn_properties def _map_s3_sources_to_functions( s3_hash_to_source: Dict[str, Tuple[str, List[Union[ConstantValue, ResolvedReference]]]], cfn_resources: Dict[str, Any], lambda_resources_to_code_map: Dict[str, List[Tuple[Dict, str]]], ) -> None: """ Maps the source property of terraform AWS S3 object resources into the the Code property of CloudFormation AWS Lambda Function resources, and append the hash value of the artifacts path to the lambda resources code map. Parameters ---------- s3_hash_to_source: Dict[str, Tuple[str, List[Union[ConstantValue, ResolvedReference]]]] Mapping of S3 object hash to S3 object source and the S3 Object configuration source value cfn_resources: dict CloudFormation resources lambda_resources_to_code_map: Dict the map between lambda resources code path values, and the lambda resources logical ids """ for resource_logical_id, resource in cfn_resources.items(): resource_type = resource.get("Type") if resource_type in CFN_CODE_PROPERTIES: code_property = CFN_CODE_PROPERTIES[resource_type] code = resource.get("Properties").get(code_property) # mapping not possible if function doesn't have bucket and key if isinstance(code, str): continue bucket = code.get("S3Bucket_config_value") if "S3Bucket_config_value" in code else code.get("S3Bucket") key = code.get("S3Key_config_value") if "S3Key_config_value" in code else code.get("S3Key") if bucket and key: obj_hash = _get_s3_object_hash(bucket, key) source = s3_hash_to_source.get(obj_hash) if source: if source[0]: tf_address = resource.get("Metadata", {}).get("SamResourceId") LOG.debug( "Found S3 object resource with matching bucket and key for function %s." " Setting function's Code property to the matching S3 object's source: %s", tf_address, source[0], ) resource["Properties"][code_property] = source[0] references = source[0] or source[1] res_type = "zip" if resource_type == CFN_AWS_LAMBDA_FUNCTION else "layer" if references: hash_value = f"{res_type}_{_calculate_configuration_attribute_value_hash(references)}" resources_list = lambda_resources_to_code_map.get(hash_value, []) resources_list.append((resource, resource_logical_id)) lambda_resources_to_code_map[hash_value] = resources_list def _check_dummy_remote_values(cfn_resources: Dict[str, Any]) -> None: """ Check if there is any lambda function/layer that has a dummy remote value for its code.imageuri or code.s3 attributes, and raise a validation error for it. Parameters ---------- cfn_resources: dict CloudFormation resources """ for _, resource in cfn_resources.items(): resource_type = resource.get("Type") if resource_type in CFN_CODE_PROPERTIES: code_property = CFN_CODE_PROPERTIES[resource_type] code = resource.get("Properties").get(code_property) # there is no code property, this is the expected behaviour in image package type functions if code is None: continue # its value is a path to a local source code if isinstance(code, str): continue bucket = code.get("S3Bucket") key = code.get("S3Key") image_uri = code.get("ImageUri") if (bucket and bucket == REMOTE_DUMMY_VALUE) or (key and key == REMOTE_DUMMY_VALUE): raise PrepareHookException( f"Lambda resource {resource.get('Metadata', {}).get('SamResourceId')} is referring to an S3 bucket " f"that is not created yet, and there is no sam metadata resource set for it to build its code " f"locally" ) if image_uri and image_uri == REMOTE_DUMMY_VALUE: raise PrepareHookException( f"Lambda resource {resource.get('Metadata', {}).get('SamResourceId')} is referring to an image uri " "that is not created yet, and there is no sam metadata resource set for it to build its image " "locally." ) def _get_s3_object_hash( bucket: Union[str, List[Union[ConstantValue, ResolvedReference]]], key: Union[str, List[Union[ConstantValue, ResolvedReference]]], ) -> str: """ Creates a hash for an AWS S3 object out of the bucket and key Parameters ---------- bucket: Union[str, List[Union[ConstantValue, ResolvedReference]]] bucket for the S3 object key: Union[str, List[Union[ConstantValue, ResolvedReference]]] key for the S3 object Returns ------- str hash for the given bucket and key """ md5 = hashlib.md5() md5.update(_calculate_configuration_attribute_value_hash(bucket).encode()) md5.update(_calculate_configuration_attribute_value_hash(key).encode()) # TODO: Hash version if it exists in addition to key and bucket return md5.hexdigest()