samcli/commands/local/lib/swagger/parser.py

"""Handles Swagger Parsing"""

import logging
from typing import Dict, List, Union

from samcli.commands.local.lib.swagger.integration_uri import IntegrationType, LambdaUri
from samcli.commands.local.lib.validators.identity_source_validator import IdentitySourceValidator
from samcli.local.apigw.authorizers.authorizer import Authorizer
from samcli.local.apigw.authorizers.lambda_authorizer import LambdaAuthorizer
from samcli.local.apigw.exceptions import (
    IncorrectOasWithDefaultAuthorizerException,
    InvalidOasVersion,
    InvalidSecurityDefinition,
    MultipleAuthorizerException,
)
from samcli.local.apigw.route import Route

LOG = logging.getLogger(__name__)


class SwaggerParser:
    _AUTHORIZER_KEY = "x-amazon-apigateway-authorizer"
    _INTEGRATION_KEY = "x-amazon-apigateway-integration"
    _ANY_METHOD_EXTENSION_KEY = "x-amazon-apigateway-any-method"
    _BINARY_MEDIA_TYPES_EXTENSION_KEY = "x-amazon-apigateway-binary-media-types"  # pylint: disable=C0103
    _ANY_METHOD = "ANY"

    _SWAGGER = "swagger"
    _OPENAPI = "openapi"
    _2_X_VERSION = "2."
    _3_X_VERSION = "3."
    _SWAGGER_COMPONENTS = "components"
    _SWAGGER_SECURITY = "security"
    _SWAGGER_SECURITY_SCHEMES = "securitySchemes"
    _SWAGGER_SECURITY_DEFINITIONS = "securityDefinitions"
    _AUTHORIZER_TYPE = "type"
    _AUTHORIZER_PAYLOAD_VERSION = "authorizerPayloadFormatVersion"
    _AUTHORIZER_LAMBDA_URI = "authorizerUri"
    _AUTHORIZER_LAMBDA_VALIDATION = "identityValidationExpression"
    _AUTHORIZER_NAME = "name"
    _AUTHORIZER_IN = "in"
    _AUTHORIZER_IDENTITY_SOURCE = "identitySource"
    _AUTHORIZER_SIMPLE_RESPONSES = "enableSimpleResponses"

    def __init__(self, stack_path: str, swagger):
        """
        Constructs an Swagger Parser object

        :param str stack_path: Path of the stack the resource is located
        :param dict swagger: Dictionary representation of a Swagger document
        """
        self.swagger = swagger or {}
        self.stack_path = stack_path

    def get_binary_media_types(self):
        """
        Get the list of Binary Media Types from Swagger

        Returns
        -------
        list of str
            List of strings that represent the Binary Media Types for the API, defaulting to empty list is None

        """
        return self.swagger.get(self._BINARY_MEDIA_TYPES_EXTENSION_KEY) or []

    def get_authorizers(self, event_type: str = Route.API) -> Dict[str, Authorizer]:
        """
        Parse Swagger document and returns a list of Authorizer objects

        Parameters
        ----------
        event_type: str
            String indicating what type of API Gateway this is

        Returns
        -------
        dict[str, Authorizer]
            A map of authorizer names and Authorizer objects found in the body definition
        """
        authorizers: Dict[str, Authorizer] = {}

        authorizer_dict = {}
        document_version = self._get_document_version()

        if document_version.startswith(SwaggerParser._2_X_VERSION):
            LOG.debug("Parsing Swagger document using 2.0 specification")
            authorizer_dict = self.swagger.get(SwaggerParser._SWAGGER_SECURITY_DEFINITIONS, {})
        elif document_version.startswith(SwaggerParser._3_X_VERSION):
            LOG.debug("Parsing Swagger document using 3.0 specification")
            authorizer_dict = self.swagger.get(SwaggerParser._SWAGGER_COMPONENTS, {}).get(
                SwaggerParser._SWAGGER_SECURITY_SCHEMES, {}
            )
        else:
            raise InvalidOasVersion(
                f"An invalid OpenApi version was detected: '{document_version}', must be one of 2.x or 3.x",
            )

        for auth_name, properties in authorizer_dict.items():
            authorizer_object = properties.get(self._AUTHORIZER_KEY)

            if not authorizer_object:
                LOG.warning("Skip parsing unsupported authorizer '%s'", auth_name)
                continue

            authorizer_type = authorizer_object.get(SwaggerParser._AUTHORIZER_TYPE, "").lower()
            payload_version = authorizer_object.get(SwaggerParser._AUTHORIZER_PAYLOAD_VERSION)

            if event_type == Route.HTTP and payload_version not in LambdaAuthorizer.PAYLOAD_VERSIONS:
                raise InvalidSecurityDefinition(f"Authorizer '{auth_name}' contains an invalid payload version")

            if event_type == Route.API:
                payload_version = LambdaAuthorizer.PAYLOAD_V1

            lambda_name = LambdaUri.get_function_name(authorizer_object.get(SwaggerParser._AUTHORIZER_LAMBDA_URI))

            if not lambda_name:
                LOG.warning("Unable to parse authorizerUri '%s' for authorizer '%s', skipping", lambda_name, auth_name)
                continue

            # only add authorizer if it is Lambda token or request based (not jwt)
            if authorizer_type not in LambdaAuthorizer.VALID_TYPES:
                LOG.warning("Lambda authorizer '%s' type '%s' is unsupported, skipping", auth_name, authorizer_type)
                continue

            identity_sources = self._get_lambda_identity_sources(
                auth_name, authorizer_type, event_type, properties, authorizer_object
            )

            validation_expression = authorizer_object.get(SwaggerParser._AUTHORIZER_LAMBDA_VALIDATION)
            if event_type == Route.HTTP and validation_expression:
                validation_expression = None

                LOG.warning(
                    "Validation expressions is only available on REST APIs, ignoring for Lambda authorizer '%s'",
                    auth_name,
                )

            enable_simple_response = authorizer_object.get(SwaggerParser._AUTHORIZER_SIMPLE_RESPONSES, False)

            if (
                event_type != Route.HTTP
                or payload_version != LambdaAuthorizer.PAYLOAD_V2
                or not isinstance(enable_simple_response, bool)
            ):
                enable_simple_response = False

                if authorizer_object.get(SwaggerParser._AUTHORIZER_SIMPLE_RESPONSES) is not None:
                    LOG.warning(
                        "Simple responses are only available on HTTP APIs with payload version "
                        "2.0, ignoring for Lambda authorizer '%s'",
                        auth_name,
                    )

            # token based authorizers must have an identity source defined
            # this is determined by taking the header key in the properties
            # to form the identity source in a previous method call
            if not identity_sources and authorizer_type == LambdaAuthorizer.TOKEN:
                LOG.warning(
                    "Skip parsing Lambda authorizer '%s', must contain valid "
                    "identity sources for Rest Api based token authorizers",
                    auth_name,
                )
                continue

            lambda_authorizer = LambdaAuthorizer(
                authorizer_name=auth_name,
                type=authorizer_type,
                payload_version=payload_version,
                lambda_name=lambda_name,
                identity_sources=identity_sources,
                validation_string=validation_expression,
                use_simple_response=enable_simple_response,
            )

            authorizers[auth_name] = lambda_authorizer

            LOG.debug("Parsing Lambda authorizer '%s' type '%s'", auth_name, authorizer_type)

        return authorizers

    @staticmethod
    def _get_lambda_identity_sources(
        auth_name: str, auth_type: str, event_type: str, properties: dict, authorizer_object: dict
    ) -> List[str]:
        """
        Parses the properties depending on the Lambda Authorizer type (token or request) and retrieves identity sources

        Parameters
        ----------
        auth_name: str
            Name of the authorizer used for logging
        auth_type: str
            Type of authorizer (token, request)
        event_type: str
            API Gateway type (API, HTTP API)
        properties: dict
            Swagger Lambda Authorizer properties
        authorizer_object: dict
            Lambda Authorizer integration properties
        Returns
        -------
        List[str]
            A list of identity sources
        """
        identity_sources: List[str] = []

        if auth_type == LambdaAuthorizer.TOKEN:
            header_name = properties.get(SwaggerParser._AUTHORIZER_NAME)

            if not properties.get(SwaggerParser._AUTHORIZER_IN) == "header" or not header_name:
                LOG.warning(
                    "Missing properties for Lambda Authorizer '%s', "
                    "property 'in' must be set to 'header' and "
                    "property 'name' must be provided",
                    auth_name,
                )
            elif event_type == Route.HTTP:
                LOG.info("Type 'token' for Lambda Authorizer '%s' is unsupported ", auth_name)
            else:
                identity_sources.append(f"method.request.header.{header_name}")
        else:
            identity_source_string = authorizer_object.get(SwaggerParser._AUTHORIZER_IDENTITY_SOURCE, "")

            # split the identity sources, remove any trailing spaces, and validate
            # we check for false-y string since .split() will return [""] instead of [] on an empty string
            split_identity_source: List[str] = identity_source_string.split(",") if identity_source_string else []

            for identity in split_identity_source:
                trimmed_identity = identity.strip()
                is_valid_format = IdentitySourceValidator.validate_identity_source(trimmed_identity, event_type)

                if not is_valid_format:
                    raise InvalidSecurityDefinition(
                        f"Identity source '{trimmed_identity}' for Lambda Authorizer '{auth_name}' "
                        "is not a valid identity source, check the spelling/format."
                    )

                identity_sources.append(trimmed_identity)

        return identity_sources

    def _get_document_version(self) -> str:
        """
        Helper method to fetch the Swagger document version

        Returns
        -------
        str
            A string representing a version, blank if not found
        """
        document_version = self.swagger.get(SwaggerParser._SWAGGER) or self.swagger.get(SwaggerParser._OPENAPI) or ""

        return str(document_version)

    def get_default_authorizer(self, event_type: str) -> Union[str, None]:
        """
        Parses the body definition to find root level Authorizer definitions

        Parameters
        ----------
        event_type: str
            String representing the type of API the definition body is defined as

        Returns
        -------
        Union[str, None]
            Returns the name of the authorizer, if there is one defined, otherwise None
        """
        document_version = self._get_document_version()
        authorizers = self.swagger.get(SwaggerParser._SWAGGER_SECURITY, [])

        if not authorizers:
            return None

        if not document_version.startswith(SwaggerParser._3_X_VERSION):
            raise IncorrectOasWithDefaultAuthorizerException(
                "Root level definition of default authorizers are only supported for API "
                "resources using an OpenApi 3.x body"
            )

        if len(authorizers) > 1:
            raise MultipleAuthorizerException(
                f"There must only be a single authorizer defined for a single route, found '{len(authorizers)}'"
            )

        if len(authorizers) == 1:
            # user has authorizer defined
            authorizer_object = authorizers[0]
            authorizer_object = list(authorizers[0])

            # make sure that authorizer actually has keys
            if len(authorizer_object) != 1:
                raise InvalidSecurityDefinition(
                    "Invalid default security definition found, there must be an authorizer defined."
                )

            authorizer_name = str(authorizer_object[0])

            LOG.debug("Found default authorizer: %s", authorizer_name)

            return authorizer_name

        return None

    def get_routes(self, event_type=Route.API) -> List[Route]:
        """
        Parses a swagger document and returns a list of APIs configured in the document.

        Swagger documents have the following structure
        {
            "/path1": {    # path
                "get": {   # method
                    "x-amazon-apigateway-integration": {   # integration
                        "type": "aws_proxy",

                        # URI contains the Lambda function ARN that needs to be parsed to get Function Name
                        "uri": {
                            "Fn::Sub":
                                "arn:aws:apigateway:aws:lambda:path/2015-03-31/functions/${LambdaFunction.Arn}/..."
                        }
                    }
                },
                "post": {
                },
            },
            "/path2": {
                ...
            }
        }

        Returns
        -------
        list of list of samcli.commands.local.apigw.local_apigw_service.Route
            List of APIs that are configured in the Swagger document
        """

        result = []
        paths_dict = self.swagger.get("paths", {})

        for full_path, path_config in paths_dict.items():
            for method, method_config in path_config.items():
                function_name = self._get_integration_function_name(method_config)
                if not function_name:
                    LOG.debug(
                        "Lambda function integration not found in Swagger document at path='%s' method='%s'",
                        full_path,
                        method,
                    )
                    continue

                normalized_method = method
                if normalized_method.lower() == self._ANY_METHOD_EXTENSION_KEY:
                    # Convert to a more commonly used method notation
                    normalized_method = self._ANY_METHOD
                payload_format_version = self._get_payload_format_version(method_config)

                authorizers = method_config.get(SwaggerParser._SWAGGER_SECURITY, None)

                authorizer_name = None
                use_default_authorizer = True
                if authorizers is not None:
                    if not isinstance(authorizers, list):
                        raise InvalidSecurityDefinition(
                            "Invalid security definition found, authorizers for "
                            f"path='{full_path}' method='{method}' must be a list"
                        )

                    if len(authorizers) > 1:
                        raise MultipleAuthorizerException(
                            "There must only be a single authorizer defined "
                            f"for path='{full_path}' method='{method}', found '{len(authorizers)}'"
                        )

                    if len(authorizers) == 1 and authorizers[0] != {}:
                        # user has authorizer defined
                        authorizer_object = authorizers[0]
                        authorizer_object = list(authorizers[0])

                        # make sure that authorizer actually has keys
                        if len(authorizer_object) != 1:
                            raise InvalidSecurityDefinition(
                                "Invalid security definition found, authorizers for "
                                f"path='{full_path}' method='{method}' must contain an authorizer"
                            )

                        authorizer_name = str(authorizer_object[0])
                    else:
                        # customer provided empty list, do not use default authorizer
                        use_default_authorizer = False

                route = Route(
                    function_name,
                    full_path,
                    methods=[normalized_method],
                    event_type=event_type,
                    payload_format_version=payload_format_version,
                    operation_name=method_config.get("operationId"),
                    stack_path=self.stack_path,
                    authorizer_name=authorizer_name,
                    use_default_authorizer=use_default_authorizer,
                )
                result.append(route)

        return result

    def _get_integration(self, method_config):
        """
        Get Integration defined in the method configuration.
        Integration configuration is defined under the special "x-amazon-apigateway-integration" key. We care only
        about Lambda integrations, which are of type aws_proxy, and ignore the rest.

        Parameters
        ----------
        method_config : dict
            Dictionary containing the method configuration which might contain integration settings

        Returns
        -------
        dict or None
            integration, if possible. None, if not.
        """
        if not isinstance(method_config, dict) or self._INTEGRATION_KEY not in method_config:
            return None

        integration = method_config[self._INTEGRATION_KEY]

        if (
            integration
            and isinstance(integration, dict)
            and integration.get("type").lower() == IntegrationType.aws_proxy.value
        ):
            # Integration must be "aws_proxy" otherwise we don't care about it
            return integration

        return None

    def _get_integration_function_name(self, method_config):
        """
        Tries to parse the Lambda Function name from the Integration defined in the method configuration.
        Integration configuration is defined under the special "x-amazon-apigateway-integration" key. We care only
        about Lambda integrations, which are of type aws_proxy, and ignore the rest. Integration URI is complex and
        hard to parse. Hence we do our best to extract function name out of integration URI. If not possible, we
        return None.

        Parameters
        ----------
        method_config : dict
            Dictionary containing the method configuration which might contain integration settings

        Returns
        -------
        string or None
            Lambda function name, if possible. None, if not.
        """
        integration = self._get_integration(method_config)
        if integration is None:
            return None

        return LambdaUri.get_function_name(integration.get("uri"))

    def _get_payload_format_version(self, method_config):
        """
        Get the "payloadFormatVersion" from the Integration defined in the method configuration.

        Parameters
        ----------
        method_config : dict
            Dictionary containing the method configuration which might contain integration settings

        Returns
        -------
        string or None
            Payload format version, if exists. None, if not.
        """
        integration = self._get_integration(method_config)
        if integration is None:
            return None

        return integration.get("payloadFormatVersion")