""" Class that Normalizes a Template based on Resource Metadata """ import json import logging import re from pathlib import Path from typing import Dict from samcli.lib.iac.cdk.utils import is_cdk_project from samcli.lib.utils.resources import AWS_CLOUDFORMATION_STACK CDK_NESTED_STACK_RESOURCE_ID_SUFFIX = ".NestedStack" RESOURCES_KEY = "Resources" PROPERTIES_KEY = "Properties" METADATA_KEY = "Metadata" RESOURCE_CDK_PATH_METADATA_KEY = "aws:cdk:path" ASSET_PATH_METADATA_KEY = "aws:asset:path" ASSET_PROPERTY_METADATA_KEY = "aws:asset:property" IMAGE_ASSET_PROPERTY = "Code.ImageUri" ASSET_DOCKERFILE_PATH_KEY = "aws:asset:dockerfile-path" ASSET_DOCKERFILE_BUILD_ARGS_KEY = "aws:asset:docker-build-args" SAM_RESOURCE_ID_KEY = "SamResourceId" SAM_IS_NORMALIZED = "SamNormalized" SAM_METADATA_DOCKERFILE_KEY = "Dockerfile" SAM_METADATA_DOCKER_CONTEXT_KEY = "DockerContext" SAM_METADATA_DOCKER_BUILD_ARGS_KEY = "DockerBuildArgs" ASSET_BUNDLED_METADATA_KEY = "aws:asset:is-bundled" SAM_METADATA_SKIP_BUILD_KEY = "SkipBuild" # https://github.com/aws/aws-cdk/blob/b1ecd3d49d7ebf97a54a80d06779ef0f0b113c16/packages/%40aws-cdk/assert-internal/lib/canonicalize-assets.ts#L19 CDK_ASSET_PARAMETER_PATTERN = re.compile( "^AssetParameters[0-9a-fA-F]{64}(?:S3Bucket|S3VersionKey|ArtifactHash)[0-9a-fA-F]{8}$" ) BUILD_PROPERTIES_PASCAL_TO_SNAKE_CASE_PATTERN = re.compile(r"(? 1: key = nested_keys.pop(0) target_dict[key] = {} target_dict = target_dict[key] target_dict[nested_keys[0]] = property_value elif property_key or property_value: LOG.info( "WARNING: Ignoring Metadata for Resource %s. Metadata contains only aws:asset:path or " "aws:assert:property but not both", logical_id, ) @staticmethod def _extract_image_asset_metadata(metadata): """ Extract/create relevant metadata properties for image assets Parameters ---------- metadata dict Metadata to use for extracting image assets properties Returns ------- dict metadata properties for image-type lambda function """ asset_path = Path(metadata.get(ASSET_PATH_METADATA_KEY, "")) dockerfile_path = Path(metadata.get(ASSET_DOCKERFILE_PATH_KEY), "") return { SAM_METADATA_DOCKERFILE_KEY: str(dockerfile_path.as_posix()), SAM_METADATA_DOCKER_CONTEXT_KEY: str(asset_path), SAM_METADATA_DOCKER_BUILD_ARGS_KEY: metadata.get(ASSET_DOCKERFILE_BUILD_ARGS_KEY, {}), } @staticmethod def _update_resource_metadata(metadata, updated_values): """ Update the metadata values for image-type lambda functions This method will mutate the template Parameters ---------- metadata dict Metadata dict to be updated updated_values dict Dict of key-value pairs to append to the existing metadata """ for key, val in updated_values.items(): metadata[key] = val @staticmethod def get_resource_id(resource_properties, logical_id): """ Get unique id for a resource. for any resource, the resource id can be the customer defined id if exist, if not exist it can be the cdk-defined resource id, or the logical id if the resource id is not found. Parameters ---------- resource_properties dict Properties of this resource logical_id str LogicalID of the resource Returns ------- str The unique function id """ resource_metadata = resource_properties.get("Metadata", {}) customer_defined_id = resource_metadata.get(SAM_RESOURCE_ID_KEY) if isinstance(customer_defined_id, str) and customer_defined_id: LOG.debug( "Sam customer defined id is more priority than other IDs. Customer defined id for resource %s is %s", logical_id, customer_defined_id, ) return customer_defined_id resource_cdk_path = resource_metadata.get(RESOURCE_CDK_PATH_METADATA_KEY) if not isinstance(resource_cdk_path, str) or not resource_cdk_path: LOG.debug( "There is no customer defined id or cdk path defined for resource %s, so we will use the resource " "logical id as the resource id", logical_id, ) return logical_id # aws:cdk:path metadata format of functions: {stack_id}/{function_id}/Resource # Design doc of CDK path: https://github.com/aws/aws-cdk/blob/master/design/construct-tree.md cdk_path_partitions = resource_cdk_path.split("/") min_cdk_path_partitions_length = 2 LOG.debug("CDK Path for resource %s is %s", logical_id, cdk_path_partitions) if len(cdk_path_partitions) < min_cdk_path_partitions_length: LOG.warning( "Cannot detect function id from aws:cdk:path metadata '%s', using default logical id", resource_cdk_path ) return logical_id cdk_resource_id = ( cdk_path_partitions[-2] if cdk_path_partitions[-1] == "Resource" or ( resource_properties.get("Type", "") == AWS_CLOUDFORMATION_STACK and cdk_path_partitions[-2].endswith(CDK_NESTED_STACK_RESOURCE_ID_SUFFIX) ) else cdk_path_partitions[-1] ) # Check if the Resource is nested Stack if resource_properties.get("Type", "") == AWS_CLOUDFORMATION_STACK and cdk_resource_id.endswith( CDK_NESTED_STACK_RESOURCE_ID_SUFFIX ): cdk_resource_id = cdk_resource_id[: -len(CDK_NESTED_STACK_RESOURCE_ID_SUFFIX)] return cdk_resource_id @staticmethod def normalize_build_properties(build_props) -> Dict: """ Convert PascalCase properties in the template to snake case to be consistent with what Lambda Builders expects from its properties :param build_props: Properties to be passed to Lambda Builders :return: dict of normalized properties """ normalized_props = {} for key, val in build_props.items(): normalized_key = BUILD_PROPERTIES_PASCAL_TO_SNAKE_CASE_PATTERN.sub("_", key).lower() normalized_props[normalized_key] = val return normalized_props