samcli/local/lambda_service/local_lambda_invoke_service.py

"""Local Lambda Service that only invokes a function"""

import io
import json
import logging

from flask import Flask, request
from werkzeug.routing import BaseConverter

from samcli.commands.local.lib.exceptions import UnsupportedInlineCodeError
from samcli.lib.utils.stream_writer import StreamWriter
from samcli.local.docker.exceptions import DockerContainerCreationFailedException
from samcli.local.lambdafn.exceptions import FunctionNotFound
from samcli.local.services.base_local_service import BaseLocalService, LambdaOutputParser

from .lambda_error_responses import LambdaErrorResponses

LOG = logging.getLogger(__name__)


class FunctionNamePathConverter(BaseConverter):
    regex = ".+"
    weight = 300
    part_isolating = False

    def to_python(self, value):
        return value

    def to_url(self, value):
        return value


class LocalLambdaInvokeService(BaseLocalService):
    def __init__(self, lambda_runner, port, host, stderr=None, ssl_context=None):
        """
        Creates a Local Lambda Service that will only response to invoking a function

        Parameters
        ----------
        lambda_runner samcli.commands.local.lib.local_lambda.LocalLambdaRunner
            The Lambda runner class capable of invoking the function
        port int
            Optional. port for the service to start listening on
        host str
            Optional. host to start the service on
        ssl_context : (str, str)
            Optional. tuple(str, str) indicating the cert and key files to use to start in https mode
            Defaults to None
        stderr io.BaseIO
            Optional stream where the stderr from Docker container should be written to
        """
        super().__init__(lambda_runner.is_debugging(), port=port, host=host, ssl_context=ssl_context)
        self.lambda_runner = lambda_runner
        self.stderr = stderr

    def create(self):
        """
        Creates a Flask Application that can be started.
        """
        self._app = Flask(__name__)

        # add converter to support nested stack function path
        self._app.url_map.converters["function_path"] = FunctionNamePathConverter

        path = "/2015-03-31/functions/<function_path:function_name>/invocations"
        self._app.add_url_rule(
            path,
            endpoint=path,
            view_func=self._invoke_request_handler,
            methods=["POST"],
            provide_automatic_options=False,
        )

        # setup request validation before Flask calls the view_func
        self._app.before_request(LocalLambdaInvokeService.validate_request)

        self._construct_error_handling()

    @staticmethod
    def validate_request():
        """
        Validates the incoming request

        The following are invalid
            1. The Request data is not json serializable
            2. Query Parameters are sent to the endpoint
            3. The Request Content-Type is not application/json
            4. 'X-Amz-Log-Type' header is not 'None'
            5. 'X-Amz-Invocation-Type' header is not 'RequestResponse'

        Returns
        -------
        flask.Response
            If the request is not valid a flask Response is returned

        None:
            If the request passes all validation
        """
        flask_request = request
        request_data = flask_request.get_data()

        if not request_data:
            request_data = b"{}"

        request_data = request_data.decode("utf-8")

        try:
            json.loads(request_data)
        except ValueError as json_error:
            LOG.debug("Request body was not json. Exception: %s", str(json_error))
            return LambdaErrorResponses.invalid_request_content(
                "Could not parse request body into json: No JSON object could be decoded"
            )

        if flask_request.args:
            LOG.debug("Query parameters are in the request but not supported")
            return LambdaErrorResponses.invalid_request_content("Query Parameters are not supported")

        request_headers = flask_request.headers

        log_type = request_headers.get("X-Amz-Log-Type", "None")
        if log_type != "None":
            LOG.debug("log-type: %s is not supported. None is only supported.", log_type)
            return LambdaErrorResponses.not_implemented_locally(
                "log-type: {} is not supported. None is only supported.".format(log_type)
            )

        invocation_type = request_headers.get("X-Amz-Invocation-Type", "RequestResponse")
        if invocation_type != "RequestResponse":
            LOG.warning("invocation-type: %s is not supported. RequestResponse is only supported.", invocation_type)
            return LambdaErrorResponses.not_implemented_locally(
                "invocation-type: {} is not supported. RequestResponse is only supported.".format(invocation_type)
            )

        return None

    def _construct_error_handling(self):
        """
        Updates the Flask app with Error Handlers for different Error Codes

        """
        self._app.register_error_handler(500, LambdaErrorResponses.generic_service_exception)
        self._app.register_error_handler(404, LambdaErrorResponses.generic_path_not_found)
        self._app.register_error_handler(405, LambdaErrorResponses.generic_method_not_allowed)

    def _invoke_request_handler(self, function_name):
        """
        Request Handler for the Local Lambda Invoke path. This method is responsible for understanding the incoming
        request and invoking the Local Lambda Function

        Parameters
        ----------
        function_name str
            Name of the function to invoke

        Returns
        -------
        A Flask Response response object as if it was returned from Lambda
        """
        flask_request = request

        request_data = flask_request.get_data()

        if not request_data:
            request_data = b"{}"

        request_data = request_data.decode("utf-8")

        stdout_stream_string = io.StringIO()
        stdout_stream_bytes = io.BytesIO()
        stdout_stream_writer = StreamWriter(stdout_stream_string, stdout_stream_bytes, auto_flush=True)

        try:
            self.lambda_runner.invoke(function_name, request_data, stdout=stdout_stream_writer, stderr=self.stderr)
        except FunctionNotFound:
            LOG.debug("%s was not found to invoke.", function_name)
            return LambdaErrorResponses.resource_not_found(function_name)
        except UnsupportedInlineCodeError:
            return LambdaErrorResponses.not_implemented_locally(
                "Inline code is not supported for sam local commands. Please write your code in a separate file."
            )
        except DockerContainerCreationFailedException as ex:
            return LambdaErrorResponses.container_creation_failed(ex.message)

        lambda_response, is_lambda_user_error_response = LambdaOutputParser.get_lambda_output(
            stdout_stream_string, stdout_stream_bytes
        )

        if is_lambda_user_error_response:
            return self.service_response(
                lambda_response, {"Content-Type": "application/json", "x-amz-function-error": "Unhandled"}, 200
            )

        return self.service_response(lambda_response, {"Content-Type": "application/json"}, 200)