samtranslator/plugins/globals/globals.py

import copy
from typing import Any, Dict, List, Optional, Union

from samtranslator.model.exceptions import ExceptionWithMessage, InvalidResourceAttributeTypeException
from samtranslator.public.intrinsics import is_intrinsics
from samtranslator.public.sdk.resource import SamResourceType
from samtranslator.swagger.swagger import SwaggerEditor


class Globals:
    """
    Class to parse and process Globals section in SAM template. If a property is specified at Global section for
    say Function, then this class will add it to each resource of AWS::Serverless::Function type.
    """

    # Key of the dictionary containing Globals section in SAM template
    _KEYWORD = "Globals"
    _RESOURCE_PREFIX = "AWS::Serverless::"
    _OPENAPIVERSION = "OpenApiVersion"
    _API_TYPE = "AWS::Serverless::Api"
    _MANAGE_SWAGGER = "__MANAGE_SWAGGER"

    supported_properties = {
        # Everything on Serverless::Function except Role, Policies, FunctionName, Events
        SamResourceType.Function.value: [
            "Handler",
            "Runtime",
            "CodeUri",
            "DeadLetterQueue",
            "Description",
            "MemorySize",
            "Timeout",
            "VpcConfig",
            "Environment",
            "Tags",
            "PropagateTags",
            "Tracing",
            "KmsKeyArn",
            "AutoPublishAlias",
            "AutoPublishAliasAllProperties",
            "Layers",
            "DeploymentPreference",
            "RolePath",
            "PermissionsBoundary",
            "ReservedConcurrentExecutions",
            "ProvisionedConcurrencyConfig",
            "AssumeRolePolicyDocument",
            "EventInvokeConfig",
            "FileSystemConfigs",
            "CodeSigningConfigArn",
            "Architectures",
            "SnapStart",
            "EphemeralStorage",
            "FunctionUrlConfig",
            "RuntimeManagementConfig",
            "LoggingConfig",
            "RecursiveLoop",
            "SourceKMSKeyArn",
        ],
        # Everything except
        #   DefinitionBody: because its hard to reason about merge of Swagger dictionaries
        #   StageName: Because StageName cannot be overridden for Implicit APIs because of the current plugin
        #              architecture
        SamResourceType.Api.value: [
            "Auth",
            "Name",
            "DefinitionUri",
            "CacheClusterEnabled",
            "CacheClusterSize",
            "MergeDefinitions",
            "Variables",
            "EndpointConfiguration",
            "MethodSettings",
            "BinaryMediaTypes",
            "MinimumCompressionSize",
            "Cors",
            "GatewayResponses",
            "AccessLogSetting",
            "CanarySetting",
            "TracingEnabled",
            "OpenApiVersion",
            "Domain",
            "AlwaysDeploy",
            "PropagateTags",
        ],
        SamResourceType.HttpApi.value: [
            "Auth",
            "AccessLogSettings",
            "StageVariables",
            "Tags",
            "CorsConfiguration",
            "DefaultRouteSettings",
            "Domain",
            "RouteSettings",
            "FailOnWarnings",
            "PropagateTags",
        ],
        SamResourceType.SimpleTable.value: ["SSESpecification"],
        SamResourceType.StateMachine.value: ["PropagateTags"],
        SamResourceType.LambdaLayerVersion.value: ["PublishLambdaVersion"],
    }
    # unreleased_properties *must be* part of supported_properties too
    unreleased_properties: Dict[str, List[str]] = {
        SamResourceType.Function.value: ["RuntimeManagementConfig", "RecursiveLoop", "SourceKMSKeyArn"],
    }

    def __init__(self, template: Dict[str, Any]) -> None:
        """
        Constructs an instance of this object

        :param dict template: SAM template to be parsed
        """
        self.supported_resource_section_names = [
            x.replace(self._RESOURCE_PREFIX, "") for x in self.supported_properties
        ]
        # Sort the names for stability in list ordering
        self.supported_resource_section_names.sort()

        self.template_globals: Dict[str, GlobalProperties] = {}

        if self._KEYWORD in template:
            self.template_globals = self._parse(template[self._KEYWORD])  # type: ignore[no-untyped-call]

    def get_template_globals(
        self, logical_id: str, resource_type: str, ignore_globals: Optional[Union[str, List[str]]]
    ) -> "GlobalProperties":
        """
        Get template globals but remove globals based on IgnoreGlobals attribute.

        :param string logical_id: LogicalId of the resource
        :param string resource_type: Type of the resource (Ex: AWS::Serverless::Function)
        :param dict ignore_globals: IgnoreGlobals resource attribute. It can be either 1) "*" string value
            or list of string value, each value should be a valid property in Globals section
        :return dict: processed template globals
        """
        if not ignore_globals:
            return self.template_globals[resource_type]

        if isinstance(ignore_globals, str) and ignore_globals == "*":
            return GlobalProperties({})

        if isinstance(ignore_globals, list):
            global_props: GlobalProperties = copy.deepcopy(self.template_globals[resource_type])
            for key in ignore_globals:
                if key not in global_props.global_properties:
                    raise InvalidResourceAttributeTypeException(
                        logical_id,
                        "IgnoreGlobals",
                        None,
                        f"Resource {logical_id} has invalid resource attribute 'IgnoreGlobals' on item '{key}'.",
                    )
                del global_props.global_properties[key]
            return global_props

        # We raise exception for any non "*" or non-list input
        raise InvalidResourceAttributeTypeException(
            logical_id,
            "IgnoreGlobals",
            None,
            f"Resource {logical_id} has invalid resource attribute 'IgnoreGlobals'.",
        )

    def merge(
        self,
        resource_type: str,
        resource_properties: Dict[str, Any],
        logical_id: str = "",
        ignore_globals: Optional[Union[str, List[str]]] = None,
    ) -> Any:
        """
        Adds global properties to the resource, if necessary. This method is a no-op if there are no global properties
        for this resource type

        :param string resource_type: Type of the resource (Ex: AWS::Serverless::Function)
        :param dict resource_properties: Properties of the resource that need to be merged
        :return dict: Merged properties of the resource
        """

        if resource_type not in self.template_globals:
            # Nothing to do. Return the template unmodified
            return resource_properties

        global_props = self.get_template_globals(logical_id, str(resource_type), ignore_globals)

        return global_props.merge(resource_properties)  # type: ignore[no-untyped-call]

    @classmethod
    def del_section(cls, template: Dict[str, Any]) -> None:
        """
        Helper method to delete the Globals section altogether from the template

        :param dict template: SAM template
        :return: Modified SAM template with Globals section
        """

        if cls._KEYWORD in template:
            del template[cls._KEYWORD]

    @classmethod
    def fix_openapi_definitions(cls, template: Dict[str, Any]) -> None:
        """
        Helper method to postprocess the resources to make sure the swagger doc version matches
        the one specified on the resource with flag OpenApiVersion.

        This is done postprocess in globals because, the implicit api plugin runs before globals, \
        and at that point the global flags aren't applied on each resource, so we do not know \
        whether OpenApiVersion flag is specified. Running the globals plugin before implicit api \
        was a risky change, so we decided to postprocess the openapi version here.

        To make sure we don't modify customer defined swagger, we also check for __MANAGE_SWAGGER flag.

        :param dict template: SAM template
        """
        resources = template.get("Resources", {})

        for _, resource in resources.items():
            if ("Type" in resource) and (resource["Type"] == cls._API_TYPE):
                properties = resource["Properties"]
                if (
                    (cls._OPENAPIVERSION in properties)
                    and (cls._MANAGE_SWAGGER in properties)
                    and SwaggerEditor.safe_compare_regex_with_string(
                        SwaggerEditor._OPENAPI_VERSION_3_REGEX, properties[cls._OPENAPIVERSION]
                    )
                ):
                    if not isinstance(properties[cls._OPENAPIVERSION], str):
                        properties[cls._OPENAPIVERSION] = str(properties[cls._OPENAPIVERSION])
                        resource["Properties"] = properties
                    if "DefinitionBody" in properties:
                        definition_body = properties["DefinitionBody"]
                        definition_body["openapi"] = properties[cls._OPENAPIVERSION]
                        if definition_body.get("swagger"):
                            del definition_body["swagger"]

    def _parse(self, globals_dict):  # type: ignore[no-untyped-def]
        """
        Takes a SAM template as input and parses the Globals section

        :param globals_dict: Dictionary representation of the Globals section
        :return: Processed globals dictionary which can be used to quickly identify properties to merge
        :raises: InvalidResourceException if the input contains properties that we don't support
        """

        _globals = {}
        if not isinstance(globals_dict, dict):
            raise InvalidGlobalsSectionException(self._KEYWORD, "It must be a non-empty dictionary")

        for section_name, properties in globals_dict.items():
            resource_type = self._make_resource_type(section_name)  # type: ignore[no-untyped-call]

            if resource_type not in self.supported_properties:
                raise InvalidGlobalsSectionException(
                    self._KEYWORD,
                    f"'{section_name}' is not supported. "
                    f"Must be one of the following values - {self.supported_resource_section_names}",
                )

            if not isinstance(properties, dict):
                raise InvalidGlobalsSectionException(self._KEYWORD, "Value of ${section} must be a dictionary")

            supported = self.supported_properties[resource_type]
            supported_displayed = [
                prop for prop in supported if prop not in self.unreleased_properties.get(resource_type, [])
            ]
            for key, _ in properties.items():
                if key not in supported:
                    raise InvalidGlobalsSectionException(
                        self._KEYWORD,
                        f"'{key}' is not a supported property of '{section_name}'. "
                        f"Must be one of the following values - {supported_displayed}",
                    )

            # Store all Global properties in a map with key being the AWS::Serverless::* resource type
            _globals[resource_type] = GlobalProperties(properties)

        return _globals

    def _make_resource_type(self, key):  # type: ignore[no-untyped-def]
        return self._RESOURCE_PREFIX + key


class GlobalProperties:
    """
    Object holding the global properties of given type. It also contains methods to perform a merge between
    Global & resource-level properties. Here are the different cases during the merge and how we handle them:

    **Primitive Type (String, Integer, Boolean etc)**
    If either global & local are of primitive types, then we the value at local will overwrite global.

    Example:

      ```
      Global:
        Function:
          Runtime: nodejs

      Function:
         Runtime: python
      ```

    After processing, Function resource will contain:
      ```
      Runtime: python
      ```

    **Different data types**
    If a value at Global is a array, but local is a dictionary, then we will simply use the local value.
    There is no merge to be done here. Similarly for other data type mismatches between global & local value.

    Example:

      ```
      Global:
        Function:
          CodeUri: s3://bucket/key

      Function:
         CodeUri:
           Bucket: foo
           Key: bar
      ```


    After processing, Function resource will contain:
      ```
        CodeUri:
           Bucket: foo
           Key: bar
      ```

    **Arrays**
    If a value is an array at both global & local level, we will simply concatenate them without de-duplicating.
    Customers can easily fix the duplicates:

    Example:

      ```
       Global:
         Function:
           Policy: [Policy1, Policy2]

       Function:
         Policy: [Policy1, Policy3]
      ```

    After processing, Function resource will contain:
    (notice the duplicates)
      ```
       Policy: [Policy1, Policy2, Policy1, Policy3]
      ```

    **Dictionaries**
    If both global & local value is a dictionary, we will recursively merge properties. If a value is one of the above
    types, they will handled according the above rules.

    Example:

      ```
       Global:
         EnvironmentVariables:
           TableName: foo
           DBName: generic-db

       Function:
          EnvironmentVariables:
            DBName: mydb
            ConnectionString: bar
      ```

    After processing, Function resource will contain:
      ```
          EnvironmentVariables:
            TableName: foo
            DBName: mydb
            ConnectionString: bar
      ```

    ***Optional Properties***
    Some resources might have optional properties with default values when it is skipped. If an optional property
    is skipped at local level, an explicitly specified value at global level will be used.

    Example:
      Global:
        DeploymentPreference:
           Enabled: False
           Type: Canary

      Function:
        DeploymentPreference:
          Type: Linear

    After processing, Function resource will contain:
      ```
      DeploymentPreference:
         Enabled: False
         Type: Linear
      ```
    (in other words, Deployments will be turned off for the Function)

    """

    def __init__(self, global_properties) -> None:  # type: ignore[no-untyped-def]
        self.global_properties = global_properties

    def merge(self, local_properties):  # type: ignore[no-untyped-def]
        """
        Merge Global & local level properties according to the above rules

        :return local_properties: Dictionary of local properties
        """
        return self._do_merge(self.global_properties, local_properties)  # type: ignore[no-untyped-call]

    def _do_merge(self, global_value, local_value):  # type: ignore[no-untyped-def]
        """
        Actually perform the merge operation for the given inputs. This method is used as part of the recursion.
        Therefore input values can be of any type. So is the output.

        :param global_value: Global value to be merged
        :param local_value: Local value to be merged
        :return: Merged result
        """

        token_global = self._token_of(global_value)
        token_local = self._token_of(local_value)

        # The following statements codify the rules explained in the doctring above
        if token_global != token_local:
            return self._prefer_local(global_value, local_value)  # type: ignore[no-untyped-call]

        if self.TOKEN.PRIMITIVE == token_global == token_local:
            return self._prefer_local(global_value, local_value)  # type: ignore[no-untyped-call]

        if self.TOKEN.DICT == token_global == token_local:
            return self._merge_dict(global_value, local_value)  # type: ignore[no-untyped-call]

        if self.TOKEN.LIST == token_global == token_local:
            return self._merge_lists(global_value, local_value)  # type: ignore[no-untyped-call]

        raise TypeError(f"Unsupported type of objects. GlobalType={token_global}, LocalType={token_local}")

    def _merge_lists(self, global_list, local_list):  # type: ignore[no-untyped-def]
        """
        Merges the global list with the local list. List merging is simply a concatenation = global + local

        :param global_list: Global value list
        :param local_list: Local value list
        :return: New merged list with the elements shallow copied
        """

        return global_list + local_list

    def _merge_dict(self, global_dict, local_dict):  # type: ignore[no-untyped-def]
        """
        Merges the two dictionaries together

        :param global_dict: Global dictionary to be merged
        :param local_dict: Local dictionary to be merged
        :return: New merged dictionary with values shallow copied
        """

        # Local has higher priority than global. So iterate over local dict and merge into global if keys are overridden
        global_dict = global_dict.copy()

        for key in local_dict:
            if key in global_dict:
                # Both local & global contains the same key. Let's do a merge.
                global_dict[key] = self._do_merge(global_dict[key], local_dict[key])  # type: ignore[no-untyped-call]

            else:
                # Key is not in globals, just in local. Copy it over
                global_dict[key] = local_dict[key]

        return global_dict

    def _prefer_local(self, global_value, local_value):  # type: ignore[no-untyped-def]
        """
        Literally returns the local value whatever it may be. This method is useful to provide a unified implementation
        for cases that don't require special handling.

        :param global_value: Global value
        :param local_value: Local value
        :return: Simply returns the local value
        """
        return local_value

    def _token_of(self, _input: Any) -> str:
        """
        Returns the token type of the input.

        :param _input: Input whose type is to be determined
        :return TOKENS: Token type of the input
        """

        if isinstance(_input, dict):
            # Intrinsic functions are always dicts
            if is_intrinsics(_input):
                # Intrinsic functions are handled *exactly* like a primitive type because
                # they resolve to a primitive type when creating a stack with CloudFormation
                return self.TOKEN.PRIMITIVE
            return self.TOKEN.DICT

        if isinstance(_input, list):
            return self.TOKEN.LIST

        return self.TOKEN.PRIMITIVE

    class TOKEN:
        """
        Enum of tokens used in the merging
        """

        PRIMITIVE = "primitive"
        DICT = "dict"
        LIST = "list"


class InvalidGlobalsSectionException(ExceptionWithMessage):
    """Exception raised when a Globals section is invalid.

    Attributes:
        message -- explanation of the error
    """

    def __init__(self, logical_id, message) -> None:  # type: ignore[no-untyped-def]
        self._logical_id = logical_id
        self._message = message

    @property
    def message(self) -> str:
        return f"'{self._logical_id}' section is invalid. {self._message}"