samcli/hook_packages/terraform/hooks/prepare/property_builder.py

"""
Terraform prepare property builder
"""

import logging
from json import loads
from json.decoder import JSONDecodeError
from typing import Any, Dict, Optional

from samcli.hook_packages.terraform.hooks.prepare.constants import (
    REMOTE_DUMMY_VALUE,
    TF_AWS_API_GATEWAY_AUTHORIZER,
    TF_AWS_API_GATEWAY_INTEGRATION,
    TF_AWS_API_GATEWAY_INTEGRATION_RESPONSE,
    TF_AWS_API_GATEWAY_METHOD,
    TF_AWS_API_GATEWAY_RESOURCE,
    TF_AWS_API_GATEWAY_REST_API,
    TF_AWS_API_GATEWAY_STAGE,
    TF_AWS_API_GATEWAY_V2_API,
    TF_AWS_API_GATEWAY_V2_AUTHORIZER,
    TF_AWS_API_GATEWAY_V2_INTEGRATION,
    TF_AWS_API_GATEWAY_V2_ROUTE,
    TF_AWS_API_GATEWAY_V2_STAGE,
    TF_AWS_LAMBDA_FUNCTION,
    TF_AWS_LAMBDA_LAYER_VERSION,
)
from samcli.hook_packages.terraform.hooks.prepare.resource_linking import _resolve_resource_attribute
from samcli.hook_packages.terraform.hooks.prepare.resources.internal import (
    INTERNAL_API_GATEWAY_INTEGRATION,
    INTERNAL_API_GATEWAY_INTEGRATION_RESPONSE,
)
from samcli.hook_packages.terraform.hooks.prepare.types import (
    PropertyBuilder,
    PropertyBuilderMapping,
    ResourceTranslator,
    TFResource,
)
from samcli.lib.hook.exceptions import PrepareHookException
from samcli.lib.utils.packagetype import IMAGE, ZIP
from samcli.lib.utils.resources import AWS_APIGATEWAY_AUTHORIZER as CFN_AWS_APIGATEWAY_AUTHORIZER
from samcli.lib.utils.resources import AWS_APIGATEWAY_METHOD as CFN_AWS_APIGATEWAY_METHOD
from samcli.lib.utils.resources import AWS_APIGATEWAY_RESOURCE as CFN_AWS_APIGATEWAY_RESOURCE
from samcli.lib.utils.resources import AWS_APIGATEWAY_RESTAPI as CFN_AWS_APIGATEWAY_RESTAPI
from samcli.lib.utils.resources import AWS_APIGATEWAY_STAGE as CFN_AWS_APIGATEWAY_STAGE
from samcli.lib.utils.resources import AWS_APIGATEWAY_V2_API as CFN_AWS_APIGATEWAY_V2_API
from samcli.lib.utils.resources import AWS_APIGATEWAY_V2_AUTHORIZER as CFN_AWS_APIGATEWAY_V2_AUTHORIZER
from samcli.lib.utils.resources import AWS_APIGATEWAY_V2_INTEGRATION as CFN_AWS_APIGATEWAY_V2_INTEGRATION
from samcli.lib.utils.resources import AWS_APIGATEWAY_V2_ROUTE as CFN_AWS_APIGATEWAY_V2_ROUTE
from samcli.lib.utils.resources import AWS_APIGATEWAY_V2_STAGE as CFN_AWS_APIGATEWAY_V2_STAGE
from samcli.lib.utils.resources import AWS_LAMBDA_FUNCTION as CFN_AWS_LAMBDA_FUNCTION
from samcli.lib.utils.resources import AWS_LAMBDA_LAYERVERSION as CFN_AWS_LAMBDA_LAYER_VERSION

LOG = logging.getLogger(__name__)


def _build_code_property(tf_properties: dict, resource: TFResource) -> Any:
    """
    Builds the Code property of a CloudFormation AWS Lambda Function out of the
    properties of the equivalent terraform resource

    Parameters
    ----------
    tf_properties: dict
        Properties of the terraform AWS Lambda function resource
    resource: TFResource
        Configuration terraform resource

    Returns
    -------
    dict
        The built Code property of a CloudFormation AWS Lambda Function resource
    """
    filename = tf_properties.get("filename")
    if filename:
        return filename

    code = {}
    tf_cfn_prop_names = [
        ("s3_bucket", "S3Bucket"),
        ("s3_key", "S3Key"),
        ("image_uri", "ImageUri"),
        ("s3_object_version", "S3ObjectVersion"),
    ]
    for tf_prop_name, cfn_prop_name in tf_cfn_prop_names:
        tf_prop_value = tf_properties.get(tf_prop_name)
        if tf_prop_value is not None:
            code[cfn_prop_name] = tf_prop_value

    package_type = tf_properties.get("package_type", ZIP)

    # Get the S3 Bucket details from configuration in case if the customer is creating the S3 bucket in the tf project
    if package_type == ZIP and ("S3Bucket" not in code or "S3Key" not in code or "S3ObjectVersion" not in code):
        s3_bucket_tf_config_value = _resolve_resource_attribute(resource, "s3_bucket")
        s3_key_tf_config_value = _resolve_resource_attribute(resource, "s3_key")
        s3_object_version_tf_config_value = _resolve_resource_attribute(resource, "s3_object_version")
        if "S3Bucket" not in code and s3_bucket_tf_config_value:
            code["S3Bucket"] = REMOTE_DUMMY_VALUE
            code["S3Bucket_config_value"] = s3_bucket_tf_config_value
        if "S3Key" not in code and s3_key_tf_config_value:
            code["S3Key"] = REMOTE_DUMMY_VALUE
            code["S3Key_config_value"] = s3_key_tf_config_value
        if "S3ObjectVersion" not in code and s3_object_version_tf_config_value:
            code["S3ObjectVersion"] = REMOTE_DUMMY_VALUE
            code["S3ObjectVersion_config_value"] = s3_object_version_tf_config_value

    # Get the Image URI details from configuration in case if the customer is creating the ecr repo in the tf project
    if package_type == IMAGE and "ImageUri" not in code:
        image_uri_tf_config_value = _resolve_resource_attribute(resource, "image_uri")
        if image_uri_tf_config_value:
            code["ImageUri"] = REMOTE_DUMMY_VALUE

    return code


def _get_property_extractor(property_name: str) -> PropertyBuilder:
    """
    Returns a PropertyBuilder function to extract the given property from a dict

    Parameters
    ----------
    property_name: str
        The name of the property to extract

    Returns
    -------
    PropertyBuilder
        function that takes in a dict and extracts the given property name from it
    """
    return lambda properties, _: properties.get(property_name)


def _build_lambda_function_environment_property(tf_properties: dict, resource: TFResource) -> Optional[dict]:
    """
    Builds the Environment property of a CloudFormation AWS Lambda Function out of the
    properties of the equivalent terraform resource

    Parameters
    ----------
    tf_properties: dict
        Properties of the terraform AWS Lambda function resource
    resource: TFResource
        Configuration terraform resource

    Returns
    -------
    dict
        The built Environment property of a CloudFormation AWS Lambda Function resource
    """
    environment = tf_properties.get("environment")
    if not environment:
        return None

    for env in environment:
        variables = env.get("variables")
        if variables:
            return {"Variables": variables}

    # no variables
    return None


def _build_lambda_function_image_config_property(tf_properties: dict, resource: TFResource) -> Optional[dict]:
    """
    Builds the ImageConfig property of a CloudFormation AWS Lambda Function out of the
    properties of the equivalent terraform resource

    Parameters
    ----------
    tf_properties: dict
        Properties of the terraform AWS Lambda function resource
    resource: TFResource
        Configuration terraform resource

    Returns
    -------
    dict
        The built ImageConfig property of a CloudFormation AWS Lambda Function resource
    """
    image_config = tf_properties.get("image_config")
    if not image_config:
        return None

    _check_image_config_value(image_config)
    image_config = image_config[0]

    cfn_image_config = {}
    tf_cfn_prop_names = [
        ("command", "Command"),
        ("entry_point", "EntryPoint"),
        ("working_directory", "WorkingDirectory"),
    ]

    for tf_prop_name, cfn_prop_name in tf_cfn_prop_names:
        tf_prop_value = image_config.get(tf_prop_name)
        if tf_prop_value is not None:
            cfn_image_config[cfn_prop_name] = tf_prop_value

    return cfn_image_config


def _check_image_config_value(image_config: Any) -> bool:
    """
    validate if the image_config property value is as SAM CLI expects. If it is not valid, it will raise a
    PrepareHookException.

     Parameters
    ----------
    image_config: Any
        The aws_lambda resource's Image_config property value as read from the terraform plan output.

    Returns
    -------
    bool
        return True, if the image_config value as expects, and raise PrepareHookException if not as expected.
    """
    if not isinstance(image_config, list):
        raise PrepareHookException(
            f"AWS SAM CLI expects that the value of image_config of aws_lambda_function resource in "
            f"the terraform plan output to be of type list instead of {type(image_config)}"
        )
    if len(image_config) > 1:
        raise PrepareHookException(
            f"AWS SAM CLI expects that there is only one item in the  image_config property of "
            f"aws_lambda_function resource in the terraform plan output, but there are "
            f"{len(image_config)} items"
        )
    return True


def _get_json_body(tf_properties: dict, resource: TFResource) -> Any:
    """
    Gets the JSON formatted body value from the API Gateway if there is one

    Parameters
    ----------
    tf_properties: dict
        Properties of the terraform AWS Lambda function resource
    resource: TFResource
        Configuration terraform resource

    Returns
    -------
    Any
        Returns a dictonary if there is a valid body to parse, otherwise return original value
    """
    body = tf_properties.get("body")

    if isinstance(body, str):
        try:
            return loads(body)
        except JSONDecodeError:
            pass

    LOG.debug(f"Failed to load JSON body for API Gateway body, returning original value: '{body}'")

    return body


def _get_cors_v2_api(tf_properties: dict, resource: TFResource) -> Optional[Dict[str, Any]]:
    """
    Extract the cors properties from an aws_apigatewayv2_api since they are in a nested property
    called "cors_configuration" and not in the root of the resource

    e.g.
    resource "aws_apigatewayv2_api" "my_api" {
      name           = "my_api"
      protocol_type  = "HTTP"

      cors_configuration {
        allow_credentials = true
        allow_headers     = ["Content-Type"]
        allow_methods     = ["OPTIONS", "POST", "GET"]
        allow_origins     = ["my-origin.com"]
        max_age           = "500"
      }
    }

    Parameters
    ----------
    tf_properties: dict
        Properties of the terraform AWS Lambda function resource
    resource: TFResource
        Configuration terraform resource

    Returns
    -------
    Optional[Dict[str, Any]]
        Optional dictionary containing the extracted cors properties with CFN property names
    """
    cors = tf_properties.get("cors_configuration")
    if not cors:
        return None

    extractable_configs = cors[0]
    cors_configuration = {}

    def _add_property(cfn_prop, tf_prop):
        property_value = extractable_configs.get(tf_prop)
        if property_value:
            cors_configuration[cfn_prop] = property_value

    _add_property("AllowCredentials", "allow_credentials")
    _add_property("AllowHeaders", "allow_headers")
    _add_property("AllowMethods", "allow_methods")
    _add_property("AllowOrigins", "allow_origins")
    _add_property("ExposeHeaders", "expose_headers")
    _add_property("MaxAge", "max_age")

    return cors_configuration


AWS_LAMBDA_FUNCTION_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "FunctionName": _get_property_extractor("function_name"),
    "Architectures": _get_property_extractor("architectures"),
    "Environment": _build_lambda_function_environment_property,
    "Code": _build_code_property,
    "Handler": _get_property_extractor("handler"),
    "PackageType": _get_property_extractor("package_type"),
    "Runtime": _get_property_extractor("runtime"),
    "Layers": _get_property_extractor("layers"),
    "Timeout": _get_property_extractor("timeout"),
    "MemorySize": _get_property_extractor("memory_size"),
    "ImageConfig": _build_lambda_function_image_config_property,
}

AWS_LAMBDA_LAYER_VERSION_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "LayerName": _get_property_extractor("layer_name"),
    "CompatibleRuntimes": _get_property_extractor("compatible_runtimes"),
    "CompatibleArchitectures": _get_property_extractor("compatible_architectures"),
    "Content": _build_code_property,
}

AWS_API_GATEWAY_REST_API_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "Name": _get_property_extractor("name"),
    "Body": _get_json_body,
    "Parameters": _get_property_extractor("parameters"),
    "BinaryMediaTypes": _get_property_extractor("binary_media_types"),
}

AWS_API_GATEWAY_STAGE_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "RestApiId": _get_property_extractor("rest_api_id"),
    "StageName": _get_property_extractor("stage_name"),
    "Variables": _get_property_extractor("variables"),
}

AWS_API_GATEWAY_RESOURCE_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "RestApiId": _get_property_extractor("rest_api_id"),
    "ParentId": _get_property_extractor("parent_id"),
    "PathPart": _get_property_extractor("path_part"),
}

AWS_API_GATEWAY_METHOD_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "RestApiId": _get_property_extractor("rest_api_id"),
    "ResourceId": _get_property_extractor("resource_id"),
    "HttpMethod": _get_property_extractor("http_method"),
    "OperationName": _get_property_extractor("operation_name"),
    "AuthorizerId": _get_property_extractor("authorizer_id"),
    "AuthorizationType": _get_property_extractor("authorization"),
}

AWS_API_GATEWAY_INTEGRATION_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "RestApiId": _get_property_extractor("rest_api_id"),
    "ResourceId": _get_property_extractor("resource_id"),
    "HttpMethod": _get_property_extractor("http_method"),
    "Uri": _get_property_extractor("uri"),
    "Type": _get_property_extractor("type"),
    "ContentHandling": _get_property_extractor("content_handling"),
    "ConnectionType": _get_property_extractor("connection_type"),
}

AWS_API_GATEWAY_AUTHORIZER_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "Name": _get_property_extractor("name"),
    "RestApiId": _get_property_extractor("rest_api_id"),
    "AuthorizerUri": _get_property_extractor("authorizer_uri"),
    "IdentitySource": _get_property_extractor("identity_source"),
    "Type": _get_property_extractor("type"),
    "IdentityValidationExpression": _get_property_extractor("identity_validation_expression"),
}

AWS_API_GATEWAY_INTEGRATION_RESPONSE_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "RestApiId": _get_property_extractor("rest_api_id"),
    "ResourceId": _get_property_extractor("resource_id"),
    "HttpMethod": _get_property_extractor("http_method"),
    "ResponseParameters": _get_property_extractor("response_parameters"),
}

AWS_API_GATEWAY_V2_API_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "Name": _get_property_extractor("name"),
    "Body": _get_json_body,
    "Target": _get_property_extractor("target"),
    "ProtocolType": _get_property_extractor("protocol_type"),
    "RouteKey": _get_property_extractor("route_key"),
    "CorsConfiguration": _get_cors_v2_api,
}

AWS_API_GATEWAY_V2_ROUTE_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "ApiId": _get_property_extractor("api_id"),
    "RouteKey": _get_property_extractor("route_key"),
    "Target": _get_property_extractor("target"),
    "OperationName": _get_property_extractor("operation_name"),
}

AWS_API_GATEWAY_V2_STAGE_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "ApiId": _get_property_extractor("api_id"),
    "StageName": _get_property_extractor("name"),
    "StageVariables": _get_property_extractor("stage_variables"),
}

AWS_API_GATEWAY_V2_INTEGRATION_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "ApiId": _get_property_extractor("api_id"),
    "IntegrationType": _get_property_extractor("integration_type"),
    "IntegrationMethod": _get_property_extractor("integration_method"),
    "IntegrationUri": _get_property_extractor("integration_uri"),
    "PayloadFormatVersion": _get_property_extractor("payload_format_version"),
}

AWS_API_GATEWAY_V2_AUTHORIZER_PROPERTY_BUILDER_MAPPING: PropertyBuilderMapping = {
    "ApiId": _get_property_extractor("api_id"),
    "AuthorizerType": _get_property_extractor("authorizer_type"),
    "AuthorizerUri": _get_property_extractor("authorizer_uri"),
    "Name": _get_property_extractor("name"),
    "AuthorizerPayloadFormatVersion": _get_property_extractor("authorizer_payload_format_version"),
    "IdentitySource": _get_property_extractor("identity_sources"),
    "EnableSimpleResponses": _get_property_extractor("enable_simple_responses"),
}

RESOURCE_TRANSLATOR_MAPPING: Dict[str, ResourceTranslator] = {
    TF_AWS_LAMBDA_FUNCTION: ResourceTranslator(CFN_AWS_LAMBDA_FUNCTION, AWS_LAMBDA_FUNCTION_PROPERTY_BUILDER_MAPPING),
    TF_AWS_LAMBDA_LAYER_VERSION: ResourceTranslator(
        CFN_AWS_LAMBDA_LAYER_VERSION, AWS_LAMBDA_LAYER_VERSION_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_REST_API: ResourceTranslator(
        CFN_AWS_APIGATEWAY_RESTAPI, AWS_API_GATEWAY_REST_API_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_STAGE: ResourceTranslator(
        CFN_AWS_APIGATEWAY_STAGE, AWS_API_GATEWAY_STAGE_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_RESOURCE: ResourceTranslator(
        CFN_AWS_APIGATEWAY_RESOURCE, AWS_API_GATEWAY_RESOURCE_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_METHOD: ResourceTranslator(
        CFN_AWS_APIGATEWAY_METHOD, AWS_API_GATEWAY_METHOD_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_INTEGRATION: ResourceTranslator(
        INTERNAL_API_GATEWAY_INTEGRATION, AWS_API_GATEWAY_INTEGRATION_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_AUTHORIZER: ResourceTranslator(
        CFN_AWS_APIGATEWAY_AUTHORIZER, AWS_API_GATEWAY_AUTHORIZER_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_INTEGRATION_RESPONSE: ResourceTranslator(
        INTERNAL_API_GATEWAY_INTEGRATION_RESPONSE, AWS_API_GATEWAY_INTEGRATION_RESPONSE_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_V2_API: ResourceTranslator(
        CFN_AWS_APIGATEWAY_V2_API, AWS_API_GATEWAY_V2_API_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_V2_ROUTE: ResourceTranslator(
        CFN_AWS_APIGATEWAY_V2_ROUTE, AWS_API_GATEWAY_V2_ROUTE_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_V2_STAGE: ResourceTranslator(
        CFN_AWS_APIGATEWAY_V2_STAGE, AWS_API_GATEWAY_V2_STAGE_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_V2_INTEGRATION: ResourceTranslator(
        CFN_AWS_APIGATEWAY_V2_INTEGRATION, AWS_API_GATEWAY_V2_INTEGRATION_PROPERTY_BUILDER_MAPPING
    ),
    TF_AWS_API_GATEWAY_V2_AUTHORIZER: ResourceTranslator(
        CFN_AWS_APIGATEWAY_V2_AUTHORIZER, AWS_API_GATEWAY_V2_AUTHORIZER_PROPERTY_BUILDER_MAPPING
    ),
}