""" Utilities to manipulate template """ import itertools import os import pathlib import jmespath import yaml from botocore.utils import set_value_from_jmespath from samcli.commands.exceptions import UserException from samcli.lib.samlib.resource_metadata_normalizer import ASSET_PATH_METADATA_KEY, ResourceMetadataNormalizer from samcli.lib.utils import graphql_api from samcli.lib.utils.packagetype import IMAGE, ZIP from samcli.lib.utils.resources import ( AWS_LAMBDA_FUNCTION, AWS_SERVERLESS_FUNCTION, AWS_SERVERLESS_GRAPHQLAPI, METADATA_WITH_LOCAL_PATHS, RESOURCES_WITH_LOCAL_PATHS, get_packageable_resource_paths, ) from samcli.yamlhelper import yaml_dump, yaml_parse class TemplateNotFoundException(UserException): pass class TemplateFailedParsingException(UserException): pass def get_template_data(template_file): """ Read the template file, parse it as JSON/YAML and return the template as a dictionary. Parameters ---------- template_file : string Path to the template to read Returns ------- Template data as a dictionary """ if not pathlib.Path(template_file).exists(): raise TemplateNotFoundException("Template file not found at {}".format(template_file)) with open(template_file, "r", encoding="utf-8") as fp: try: return yaml_parse(fp.read()) except (ValueError, yaml.YAMLError) as ex: raise TemplateFailedParsingException("Failed to parse template: {}".format(str(ex))) from ex def move_template(src_template_path, dest_template_path, template_dict): """ Move the SAM/CloudFormation template from ``src_template_path`` to ``dest_template_path``. For convenience, this method accepts a dictionary of template data ``template_dict`` that will be written to the destination instead of reading from the source file. SAM/CloudFormation template can contain certain properties whose value is a relative path to a local file/folder. This path is always relative to the template's location. Before writing the template to ``dest_template_path`, we will update these paths to be relative to the new location. This methods updates resource properties supported by ``aws cloudformation package`` command: https://docs.aws.amazon.com/cli/latest/reference/cloudformation/package.html You must use this method if you are reading a template from one location, modifying it, and writing it back to a different location. Parameters ---------- src_template_path : str Path to the original location of the template dest_template_path : str Path to the destination location where updated template should be written to template_dict : dict Dictionary containing template contents. This dictionary will be updated & written to ``dest`` location. """ original_root = os.path.dirname(src_template_path) new_root = os.path.dirname(dest_template_path) # Next up, we will be writing the template to a different location. Before doing so, we should # update any relative paths in the template to be relative to the new location. modified_template = _update_relative_paths(template_dict, original_root, new_root) # if a stack only has image functions, the directory for that directory won't be created. # here we make sure the directory the destination template file to write to exists. os.makedirs(os.path.dirname(dest_template_path), exist_ok=True) with open(dest_template_path, "w") as fp: fp.write(yaml_dump(modified_template)) def _update_relative_paths(template_dict, original_root, new_root): """ SAM/CloudFormation template can contain certain properties whose value is a relative path to a local file/folder. This path is usually relative to the template's location. If the template is being moved from original location ``original_root`` to new location ``new_root``, use this method to update these paths to be relative to ``new_root``. After this method is complete, it is safe to write the template to ``new_root`` without breaking any relative paths. This methods updates resource properties supported by ``aws cloudformation package`` command: https://docs.aws.amazon.com/cli/latest/reference/cloudformation/package.html If a property is either an absolute path or a S3 URI, this method will not update them. Parameters ---------- template_dict : dict Dictionary containing template contents. This dictionary will be updated & written to ``dest`` location. original_root : str Path to the directory where all paths were originally set relative to. This is usually the directory containing the template originally new_root : str Path to the new directory that all paths set relative to after this method completes. Returns ------- Updated dictionary """ for resource_type, properties in template_dict.get("Metadata", {}).items(): if resource_type not in METADATA_WITH_LOCAL_PATHS: # Unknown resource. Skipping continue for path_prop_name in METADATA_WITH_LOCAL_PATHS[resource_type]: path = properties.get(path_prop_name) updated_path = _resolve_relative_to(path, original_root, new_root) if not updated_path: # This path does not need to get updated continue properties[path_prop_name] = updated_path for _, resource in template_dict.get("Resources", {}).items(): resource_type = resource.get("Type") if resource_type not in RESOURCES_WITH_LOCAL_PATHS: # Unknown resource. Skipping continue for path_prop_name in RESOURCES_WITH_LOCAL_PATHS[resource_type]: properties = resource.get("Properties", {}) if ( resource_type in [AWS_SERVERLESS_FUNCTION, AWS_LAMBDA_FUNCTION] and properties.get("PackageType", ZIP) == IMAGE ): continue # SAM GraphQLApi has many instances of CODE_ARTIFACT_PROPERTY and all of them must be updated if resource_type == AWS_SERVERLESS_GRAPHQLAPI and path_prop_name == graphql_api.CODE_ARTIFACT_PROPERTY: # to be able to set different nested properties to S3 uri, paths are necessary # jmespath doesn't provide that functionality, thus custom implementation paths_values = graphql_api.find_all_paths_and_values(path_prop_name, properties) for property_path, property_value in paths_values: updated_path = _resolve_relative_to(property_value, original_root, new_root) if not updated_path: # This path does not need to get updated continue set_value_from_jmespath(properties, property_path, updated_path) path = jmespath.search(path_prop_name, properties) updated_path = _resolve_relative_to(path, original_root, new_root) if not updated_path: # This path does not need to get updated continue set_value_from_jmespath(properties, path_prop_name, updated_path) metadata = resource.get("Metadata", {}) if ASSET_PATH_METADATA_KEY in metadata: path = metadata.get(ASSET_PATH_METADATA_KEY, "") updated_path = _resolve_relative_to(path, original_root, new_root) if not updated_path: # This path does not need to get updated continue metadata[ASSET_PATH_METADATA_KEY] = updated_path # AWS::Includes can be anywhere within the template dictionary. Hence we need to recurse through the # dictionary in a separate method to find and update relative paths in there template_dict = _update_aws_include_relative_path(template_dict, original_root, new_root) return template_dict def _update_aws_include_relative_path(template_dict, original_root, new_root): """ Update relative paths in "AWS::Include" directive. This directive can be present at any part of the template, and not just within resources. """ for key, val in template_dict.items(): if key == "Fn::Transform": if isinstance(val, dict) and val.get("Name") == "AWS::Include": path = val.get("Parameters", {}).get("Location", {}) updated_path = _resolve_relative_to(path, original_root, new_root) if not updated_path: # This path does not need to get updated continue val["Parameters"]["Location"] = updated_path # Recurse through all dictionary values elif isinstance(val, dict): _update_aws_include_relative_path(val, original_root, new_root) elif isinstance(val, list): for item in val: if isinstance(item, dict): _update_aws_include_relative_path(item, original_root, new_root) return template_dict def _resolve_relative_to(path, original_root, new_root): """ If the given ``path`` is a relative path, then assume it is relative to ``original_root``. This method will update the path to be resolve it relative to ``new_root`` and return. Examples ------- # Assume a file called template.txt at location /tmp/original/root/template.txt expressed as relative path # We are trying to update it to be relative to /tmp/new/root instead of the /tmp/original/root >>> result = _resolve_relative_to("template.txt", \ "/tmp/original/root", \ "/tmp/new/root") >>> result ../../original/root/template.txt Returns ------- Updated path if the given path is a relative path. None, if the path is not a relative path. """ if ( not isinstance(path, str) or path.startswith("s3://") or path.startswith("http://") or path.startswith("https://") or os.path.isabs(path) ): # Value is definitely NOT a relative path. It is either a S3 URi or Absolute path or not a string at all return None # Value is definitely a relative path. Change it relative to the destination directory return os.path.relpath( # Resolve the paths to take care of symlinks os.path.normpath(os.path.join(pathlib.Path(original_root).resolve(), path)), pathlib.Path(new_root).resolve(), # Absolute original path w.r.t ``original_root`` ) # Resolve the original path with respect to ``new_root`` def get_template_parameters(template_file): """ Get Parameters from a template file. Parameters ---------- template_file : string Path to the template to read Returns ------- Template Parameters as a dictionary """ template_dict = get_template_data(template_file=template_file) ResourceMetadataNormalizer.normalize(template_dict, True) return template_dict.get("Parameters", dict()) def get_template_artifacts_format(template_file): """ Get a list of template artifact formats based on PackageType wherever the underlying resource have the actual need to be packaged. :param template_file: :return: list of artifact formats """ template_dict = get_template_data(template_file=template_file) # Get a list of Resources where the artifacts format matter for packaging. packageable_resources = get_packageable_resource_paths() artifacts = [] for _, resource in template_dict.get("Resources", {}).items(): # First check if the resources are part of package-able resource types. if resource.get("Type") in packageable_resources.keys(): # Flatten list of locations per resource type. locations = list(itertools.chain(*packageable_resources.get(resource.get("Type")))) for location in locations: properties = resource.get("Properties", {}) # Search for package-able location within resource properties. if jmespath.search(location, properties): artifacts.append(properties.get("PackageType", ZIP)) return artifacts def get_template_function_resource_ids(template_file, artifact): """ Get a list of function logical ids from template file. Function resource types include AWS::Lambda::Function AWS::Serverless::Function :param template_file: template file location. :param artifact: artifact of type IMAGE or ZIP :return: list of artifact formats """ template_dict = get_template_data(template_file=template_file) _function_resource_ids = [] for resource_id, resource in template_dict.get("Resources", {}).items(): if resource.get("Properties", {}).get("PackageType", ZIP) == artifact and resource.get("Type") in [ AWS_SERVERLESS_FUNCTION, AWS_LAMBDA_FUNCTION, ]: _function_resource_ids.append(resource_id) return _function_resource_ids