21615209021

Committed 02 Feb 2026 12:25PM UTC coverage: 86.976% (+0.007%) from 86.969%

Build # 21615209021

Build Type

push

github

Committed by

web-flow

Commit Message

fix lambda state reset method to shutdown esm workers correctly (#13671)

Run Details

0 of 3 new or added lines in 1 file covered. (0.0%)

19 existing lines in 5 files now uncovered.

70493 of 81049 relevant lines covered (86.98%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.38

/localstack-core/localstack/services/lambda_/resource_providers/aws_lambda_function.py

# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

import os
from typing import TypedDict

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceRequest,
)
from localstack.services.lambda_.lambda_utils import get_handler_file_from_name
from localstack.services.lambda_.resource_providers.generated.aws_lambda_function_base import (
    LambdaFunctionProviderBase,
)
from localstack.utils.archives import is_zip_file
from localstack.utils.files import mkdir, new_tmp_dir, rm_rf, save_file
from localstack.utils.strings import is_base64, to_bytes
from localstack.utils.testutil import create_zip_file


class LambdaManagedInstancesCapacityProviderConfig(TypedDict):
    CapacityProviderArn: str | None
    PerExecutionEnvironmentMaxConcurrency: int | None
    ExecutionEnvironmentMemoryGiBPerVCpu: float | None


class CapacityProviderConfig(TypedDict):
    LambdaManagedInstancesCapacityProviderConfig: (
        LambdaManagedInstancesCapacityProviderConfig | None
    )


class LambdaFunctionProperties(TypedDict):
    Code: Code | None
    Role: str | None
    Architectures: list[str] | None
    Arn: str | None
    CapacityProviderConfig: CapacityProviderConfig | None
    CodeSigningConfigArn: str | None
    DeadLetterConfig: DeadLetterConfig | None
    Description: str | None
    Environment: Environment | None
    EphemeralStorage: EphemeralStorage | None
    FileSystemConfigs: list[FileSystemConfig] | None
    FunctionName: str | None
    Handler: str | None
    ImageConfig: ImageConfig | None
    KmsKeyArn: str | None
    Layers: list[str] | None
    MemorySize: int | None
    PackageType: str | None
    ReservedConcurrentExecutions: int | None
    Runtime: str | None
    RuntimeManagementConfig: RuntimeManagementConfig | None
    SnapStart: SnapStart | None
    SnapStartResponse: SnapStartResponse | None
    Tags: list[Tag] | None
    Timeout: int | None
    TracingConfig: TracingConfig | None
    VpcConfig: VpcConfig | None


class TracingConfig(TypedDict):
    Mode: str | None


class VpcConfig(TypedDict):
    SecurityGroupIds: list[str] | None
    SubnetIds: list[str] | None


class RuntimeManagementConfig(TypedDict):
    UpdateRuntimeOn: str | None
    RuntimeVersionArn: str | None


class SnapStart(TypedDict):
    ApplyOn: str | None


class FileSystemConfig(TypedDict):
    Arn: str | None
    LocalMountPath: str | None


class Tag(TypedDict):
    Key: str | None
    Value: str | None


class ImageConfig(TypedDict):
    Command: list[str] | None
    EntryPoint: list[str] | None
    WorkingDirectory: str | None


class DeadLetterConfig(TypedDict):
    TargetArn: str | None


class SnapStartResponse(TypedDict):
    ApplyOn: str | None
    OptimizationStatus: str | None


class Code(TypedDict):
    ImageUri: str | None
    S3Bucket: str | None
    S3Key: str | None
    S3ObjectVersion: str | None
    ZipFile: str | None


class LoggingConfig(TypedDict):
    ApplicationLogLevel: str | None
    LogFormat: str | None
    LogGroup: str | None
    SystemLogLevel: str | None


class Environment(TypedDict):
    Variables: dict | None


class EphemeralStorage(TypedDict):
    Size: int | None


REPEATED_INVOCATION = "repeated_invocation"

# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
PYTHON_CFN_RESPONSE_CONTENT = """
# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0

from __future__ import print_function
import urllib3
import json

SUCCESS = "SUCCESS"
FAILED = "FAILED"

http = urllib3.PoolManager()


def send(event, context, responseStatus, responseData, physicalResourceId=None, noEcho=False, reason=None):
    responseUrl = event['ResponseURL']

    print(responseUrl)

    responseBody = {
        'Status' : responseStatus,
        'Reason' : reason or "See the details in CloudWatch Log Stream: {}".format(context.log_stream_name),
        'PhysicalResourceId' : physicalResourceId or context.log_stream_name,
        'StackId' : event['StackId'],
        'RequestId' : event['RequestId'],
        'LogicalResourceId' : event['LogicalResourceId'],
        'NoEcho' : noEcho,
        'Data' : responseData
    }

    json_responseBody = json.dumps(responseBody)

    print("Response body:")
    print(json_responseBody)

    headers = {
        'content-type' : '',
        'content-length' : str(len(json_responseBody))
    }

    try:
        response = http.request('PUT', responseUrl, headers=headers, body=json_responseBody)
        print("Status code:", response.status)


    except Exception as e:

        print("send(..) failed executing http.request(..):", e)
"""

# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
NODEJS_CFN_RESPONSE_CONTENT = r"""
// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// SPDX-License-Identifier: MIT-0

exports.SUCCESS = "SUCCESS";
exports.FAILED = "FAILED";

exports.send = function(event, context, responseStatus, responseData, physicalResourceId, noEcho) {

    var responseBody = JSON.stringify({
        Status: responseStatus,
        Reason: "See the details in CloudWatch Log Stream: " + context.logStreamName,
        PhysicalResourceId: physicalResourceId || context.logStreamName,
        StackId: event.StackId,
        RequestId: event.RequestId,
        LogicalResourceId: event.LogicalResourceId,
        NoEcho: noEcho || false,
        Data: responseData
    });

    console.log("Response body:\n", responseBody);

    var https = require("https");
    var url = require("url");

    var parsedUrl = url.parse(event.ResponseURL);
    var options = {
        hostname: parsedUrl.hostname,
        port: parsedUrl.port, // Modified line: LS uses port 4566 for https; hard coded 443 causes error
        path: parsedUrl.path,
        method: "PUT",
        headers: {
            "content-type": "",
            "content-length": responseBody.length
        }
    };

    var request = https.request(options, function(response) {
        console.log("Status code: " + parseInt(response.statusCode));
        context.done();
    });

    request.on("error", function(error) {
        console.log("send(..) failed executing https.request(..): " + error);
        context.done();
    });

    request.write(responseBody);
    request.end();
}
"""


def _runtime_supports_inline_code(runtime: str) -> bool:
    return runtime.startswith("python") or runtime.startswith("node")


def _get_lambda_code_param(
    properties: LambdaFunctionProperties,
    _include_arch=False,
):
    # code here is mostly taken directly from legacy implementation
    code = properties.get("Code", {}).copy()

    # TODO: verify only one of "ImageUri" | "S3Bucket" | "ZipFile" is set
    zip_file = code.get("ZipFile")
    if zip_file and not _runtime_supports_inline_code(properties["Runtime"]):
        raise Exception(
            f"Runtime {properties['Runtime']} doesn't support inlining code via the 'ZipFile' property."
        )  # TODO: message not validated
    if zip_file and not is_base64(zip_file) and not is_zip_file(to_bytes(zip_file)):
        tmp_dir = new_tmp_dir()
        try:
            handler_file = get_handler_file_from_name(
                properties["Handler"], runtime=properties["Runtime"]
            )
            tmp_file = os.path.join(tmp_dir, handler_file)
            save_file(tmp_file, zip_file)

            # CloudFormation only includes cfn-response libs if an import is detected
            # TODO: add snapshots for this behavior
            # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
            if properties["Runtime"].lower().startswith("node") and (
                "require('cfn-response')" in zip_file or 'require("cfn-response")' in zip_file
            ):
                # the check if cfn-response is used is pretty simplistic and apparently based on simple string matching
                # having the import commented out will also lead to cfn-response.js being injected
                # this is added under both cfn-response.js and node_modules/cfn-response.js
                cfn_response_mod_dir = os.path.join(tmp_dir, "node_modules")
                mkdir(cfn_response_mod_dir)
                save_file(
                    os.path.join(cfn_response_mod_dir, "cfn-response.js"),
                    NODEJS_CFN_RESPONSE_CONTENT,
                )
                save_file(os.path.join(tmp_dir, "cfn-response.js"), NODEJS_CFN_RESPONSE_CONTENT)
            elif (
                properties["Runtime"].lower().startswith("python")
                and "import cfnresponse" in zip_file
            ):
                save_file(os.path.join(tmp_dir, "cfnresponse.py"), PYTHON_CFN_RESPONSE_CONTENT)

            # create zip file
            zip_file = create_zip_file(tmp_dir, get_content=True)
            code["ZipFile"] = zip_file
        finally:
            rm_rf(tmp_dir)
    if _include_arch and "Architectures" in properties:
        code["Architectures"] = properties.get("Architectures")
    return code


def _transform_function_to_model(function):
    model_properties = [
        "MemorySize",
        "Description",
        "TracingConfig",
        "Timeout",
        "Handler",
        "SnapStartResponse",
        "Role",
        "FileSystemConfigs",
        "FunctionName",
        "Runtime",
        "PackageType",
        "LoggingConfig",
        "Environment",
        "Arn",
        "EphemeralStorage",
        "Architectures",
        "CapacityProviderConfig",
    ]
    response_model = util.select_attributes(function, model_properties)
    response_model["Arn"] = function["FunctionArn"]
    return response_model


class LambdaFunctionProvider(LambdaFunctionProviderBase):
    def create(
        self,
        request: ResourceRequest[LambdaFunctionProperties],
    ) -> ProgressEvent[LambdaFunctionProperties]:
        """
        Create a new resource.

        Primary identifier fields:
          - /properties/FunctionName

        Required properties:
          - Code
          - Role

        Create-only properties:
          - /properties/FunctionName
          - /properties/PackageType
          - /properties/TenancyConfig

        Read-only properties:
          - /properties/Arn
          - /properties/SnapStartResponse
          - /properties/SnapStartResponse/ApplyOn
          - /properties/SnapStartResponse/OptimizationStatus

        IAM permissions required:
          - lambda:CreateFunction
          - lambda:GetFunction
          - lambda:PutFunctionConcurrency
          - iam:PassRole
          - s3:GetObject
          - s3:GetObjectVersion
          - ec2:DescribeSecurityGroups
          - ec2:DescribeSubnets
          - ec2:DescribeVpcs
          - elasticfilesystem:DescribeMountTargets
          - kms:CreateGrant
          - kms:Decrypt
          - kms:Encrypt
          - kms:GenerateDataKey
          - lambda:GetCodeSigningConfig
          - lambda:GetFunctionCodeSigningConfig
          - lambda:GetLayerVersion
          - lambda:GetRuntimeManagementConfig
          - lambda:PutRuntimeManagementConfig
          - lambda:TagResource
          - lambda:GetPolicy
          - lambda:PutFunctionRecursionConfig
          - lambda:GetFunctionRecursionConfig
          - lambda:PutFunctionScalingConfig
          - lambda:PassCapacityProvider

        """
        model = request.desired_state
        lambda_client = request.aws_client_factory.lambda_

        if not request.custom_context.get(REPEATED_INVOCATION):
            request.custom_context[REPEATED_INVOCATION] = True

            name = model.get("FunctionName")
            if not name:
                name = util.generate_default_name(request.stack_name, request.logical_resource_id)
                model["FunctionName"] = name

            kwargs = util.select_attributes(
                model,
                [
                    "Architectures",
                    "DeadLetterConfig",
                    "Description",
                    "FunctionName",
                    "Handler",
                    "ImageConfig",
                    "PackageType",
                    "Layers",
                    "MemorySize",
                    "Runtime",
                    "Role",
                    "Timeout",
                    "TracingConfig",
                    "VpcConfig",
                    "LoggingConfig",
                    "CapacityProviderConfig",
                ],
            )
            if "Timeout" in kwargs:
                kwargs["Timeout"] = int(kwargs["Timeout"])
            if "MemorySize" in kwargs:
                kwargs["MemorySize"] = int(kwargs["MemorySize"])
            if model_tags := model.get("Tags"):
                tags = {}
                for tag in model_tags:
                    tags[tag["Key"]] = tag["Value"]
                kwargs["Tags"] = tags

            # botocore/data/lambda/2015-03-31/service-2.json:1161 (EnvironmentVariableValue)
            # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-function-environment.html
            if "Environment" in model:
                environment_variables = model["Environment"].get("Variables", {})
                kwargs["Environment"] = {
                    "Variables": {k: str(v) for k, v in environment_variables.items()}
                }

            kwargs["Code"] = _get_lambda_code_param(model)

            # For managed instance lambdas, we publish them immediately
            if "CapacityProviderConfig" in kwargs:
                kwargs["Publish"] = True
                kwargs["PublishTo"] = "LATEST_PUBLISHED"

            create_response = lambda_client.create_function(**kwargs)
            # TODO: if version is in the schema, just put it in the model instead of the custom context
            request.custom_context["Version"] = create_response["Version"]  # $LATEST.PUBLISHED
            model["Arn"] = create_response["FunctionArn"]

        if request.custom_context.get("Version") == "$LATEST.PUBLISHED":
            # for managed instance lambdas, we need to wait until the version is published & active
            get_fn_response = lambda_client.get_function(
                FunctionName=model["FunctionName"], Qualifier=request.custom_context["Version"]
            )
        else:
            get_fn_response = lambda_client.get_function(FunctionName=model["Arn"])

        match get_fn_response["Configuration"]["State"]:
            # TODO: explicitly handle new ActiveNonInvocable state?
            case "Pending":
                return ProgressEvent(
                    status=OperationStatus.IN_PROGRESS,
                    resource_model=model,
                    custom_context=request.custom_context,
                )
            case "Active":
                return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)
            case "Inactive":
                # This might happen when setting LAMBDA_KEEPALIVE_MS=0
                return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)
            case "Failed":
                return ProgressEvent(
                    status=OperationStatus.FAILED,
                    resource_model=model,
                    error_code=get_fn_response["Configuration"].get("StateReasonCode", "unknown"),
                    message=get_fn_response["Configuration"].get("StateReason", "unknown"),
                )
            case unknown_state:  # invalid state, should technically never happen
                return ProgressEvent(
                    status=OperationStatus.FAILED,
                    resource_model=model,
                    error_code="InternalException",
                    message=f"Invalid state returned: {unknown_state}",
                )

    def read(
        self,
        request: ResourceRequest[LambdaFunctionProperties],
    ) -> ProgressEvent[LambdaFunctionProperties]:
        """
        Fetch resource information

        IAM permissions required:
          - lambda:GetFunction
          - lambda:GetFunctionCodeSigningConfig
          - lambda:GetFunctionRecursionConfig
          - lambda:GetRuntimeManagementConfig
          - lambda:GetFunctionScalingConfig
        """
        function_name = request.desired_state["FunctionName"]
        lambda_client = request.aws_client_factory.lambda_
        get_fn_response = lambda_client.get_function(FunctionName=function_name)

        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model=_transform_function_to_model(get_fn_response["Configuration"]),
        )

    def delete(
        self,
        request: ResourceRequest[LambdaFunctionProperties],
    ) -> ProgressEvent[LambdaFunctionProperties]:
        """
        Delete a resource

        IAM permissions required:
          - lambda:DeleteFunction
          - lambda:GetFunction
          - ec2:DescribeNetworkInterfaces
        """
        try:
            lambda_client = request.aws_client_factory.lambda_
            lambda_client.delete_function(FunctionName=request.previous_state["FunctionName"])
        except request.aws_client_factory.lambda_.exceptions.ResourceNotFoundException:
            pass
        # any other exception will be propagated
        return ProgressEvent(status=OperationStatus.SUCCESS, resource_model={})

    def update(
        self,
        request: ResourceRequest[LambdaFunctionProperties],
    ) -> ProgressEvent[LambdaFunctionProperties]:
        """
        Update a resource

        IAM permissions required:
          - lambda:DeleteFunctionConcurrency
          - lambda:GetFunction
          - lambda:PutFunctionConcurrency
          - lambda:TagResource
          - lambda:UntagResource
          - lambda:UpdateFunctionConfiguration
          - lambda:UpdateFunctionCode
          - iam:PassRole
          - s3:GetObject
          - s3:GetObjectVersion
          - ec2:DescribeSecurityGroups
          - ec2:DescribeSubnets
          - ec2:DescribeVpcs
          - elasticfilesystem:DescribeMountTargets
          - kms:CreateGrant
          - kms:Decrypt
          - kms:GenerateDataKey
          - lambda:GetRuntimeManagementConfig
          - lambda:PutRuntimeManagementConfig
          - lambda:PutFunctionCodeSigningConfig
          - lambda:DeleteFunctionCodeSigningConfig
          - lambda:GetCodeSigningConfig
          - lambda:GetFunctionCodeSigningConfig
          - lambda:PutFunctionRecursionConfig
          - lambda:GetFunctionRecursionConfig
          - lambda:PutFunctionScalingConfig
          - lambda:PublishVersion
          - lambda:PassCapacityProvider
        """
        client = request.aws_client_factory.lambda_

        # TODO: handle defaults properly
        old_name = request.previous_state["FunctionName"]
        new_name = request.desired_state.get("FunctionName")
        if new_name and old_name != new_name:
            # replacement (!) => shouldn't be handled here but in the engine
            self.delete(request)
            return self.create(request)

        config_keys = [
            "Description",
            "DeadLetterConfig",
            "Environment",
            "Handler",
            "ImageConfig",
            "Layers",
            "MemorySize",
            "Role",
            "Runtime",
            "Timeout",
            "TracingConfig",
            "VpcConfig",
            "LoggingConfig",
            "CapacityProviderConfig",
        ]
        update_config_props = util.select_attributes(request.desired_state, config_keys)
        function_name = request.previous_state["FunctionName"]
        update_config_props["FunctionName"] = function_name

        if "Timeout" in update_config_props:
            update_config_props["Timeout"] = int(update_config_props["Timeout"])
        if "MemorySize" in update_config_props:
            update_config_props["MemorySize"] = int(update_config_props["MemorySize"])
        if "Code" in request.desired_state:
            code = request.desired_state["Code"] or {}
            if not code.get("ZipFile"):
                request.logger.debug(
                    'Updating code for Lambda "%s" from location: %s', function_name, code
                )
            code = _get_lambda_code_param(
                request.desired_state,
                _include_arch=True,
            )
            client.update_function_code(FunctionName=function_name, **code)
            client.get_waiter("function_updated_v2").wait(FunctionName=function_name)
        if "Environment" in update_config_props:
            environment_variables = update_config_props["Environment"].get("Variables", {})
            update_config_props["Environment"]["Variables"] = {
                k: str(v) for k, v in environment_variables.items()
            }
        client.update_function_configuration(**update_config_props)
        client.get_waiter("function_updated_v2").wait(FunctionName=function_name)
        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_model={**request.previous_state, **request.desired_state},
        )

    def list(
        self,
        request: ResourceRequest[LambdaFunctionProperties],
    ) -> ProgressEvent[LambdaFunctionProperties]:
        """
        List available resources of this type
        IAM permissions required:
          - lambda:ListFunctions
        """
        functions = request.aws_client_factory.lambda_.list_functions()
        return ProgressEvent(
            status=OperationStatus.SUCCESS,
            resource_models=[_transform_function_to_model(fn) for fn in functions["Functions"]],
        )

1	# LocalStack Resource Provider Scaffolding v2
2	from __future__ import annotations	1✔
3
4	import os	1✔
5	from typing import TypedDict	1✔
6
7	import localstack.services.cloudformation.provider_utils as util	1✔
8	from localstack.services.cloudformation.resource_provider import (	1✔
9	OperationStatus,
10	ProgressEvent,
11	ResourceRequest,
12	)
13	from localstack.services.lambda_.lambda_utils import get_handler_file_from_name	1✔
14	from localstack.services.lambda_.resource_providers.generated.aws_lambda_function_base import (	1✔
15	LambdaFunctionProviderBase,
16	)
17	from localstack.utils.archives import is_zip_file	1✔
18	from localstack.utils.files import mkdir, new_tmp_dir, rm_rf, save_file	1✔
19	from localstack.utils.strings import is_base64, to_bytes	1✔
20	from localstack.utils.testutil import create_zip_file	1✔
21
22
23	class LambdaManagedInstancesCapacityProviderConfig(TypedDict):	1✔
24	CapacityProviderArn: str \| None	1✔
25	PerExecutionEnvironmentMaxConcurrency: int \| None	1✔
26	ExecutionEnvironmentMemoryGiBPerVCpu: float \| None	1✔
27
28
29	class CapacityProviderConfig(TypedDict):	1✔
30	LambdaManagedInstancesCapacityProviderConfig: (	1✔
31	LambdaManagedInstancesCapacityProviderConfig \| None
32	)
33
34
35	class LambdaFunctionProperties(TypedDict):	1✔
36	Code: Code \| None	1✔
37	Role: str \| None	1✔
38	Architectures: list[str] \| None	1✔
39	Arn: str \| None	1✔
40	CapacityProviderConfig: CapacityProviderConfig \| None	1✔
41	CodeSigningConfigArn: str \| None	1✔
42	DeadLetterConfig: DeadLetterConfig \| None	1✔
43	Description: str \| None	1✔
44	Environment: Environment \| None	1✔
45	EphemeralStorage: EphemeralStorage \| None	1✔
46	FileSystemConfigs: list[FileSystemConfig] \| None	1✔
47	FunctionName: str \| None	1✔
48	Handler: str \| None	1✔
49	ImageConfig: ImageConfig \| None	1✔
50	KmsKeyArn: str \| None	1✔
51	Layers: list[str] \| None	1✔
52	MemorySize: int \| None	1✔
53	PackageType: str \| None	1✔
54	ReservedConcurrentExecutions: int \| None	1✔
55	Runtime: str \| None	1✔
56	RuntimeManagementConfig: RuntimeManagementConfig \| None	1✔
57	SnapStart: SnapStart \| None	1✔
58	SnapStartResponse: SnapStartResponse \| None	1✔
59	Tags: list[Tag] \| None	1✔
60	Timeout: int \| None	1✔
61	TracingConfig: TracingConfig \| None	1✔
62	VpcConfig: VpcConfig \| None	1✔
63
64
65	class TracingConfig(TypedDict):	1✔
66	Mode: str \| None	1✔
67
68
69	class VpcConfig(TypedDict):	1✔
70	SecurityGroupIds: list[str] \| None	1✔
71	SubnetIds: list[str] \| None	1✔
72
73
74	class RuntimeManagementConfig(TypedDict):	1✔
75	UpdateRuntimeOn: str \| None	1✔
76	RuntimeVersionArn: str \| None	1✔
77
78
79	class SnapStart(TypedDict):	1✔
80	ApplyOn: str \| None	1✔
81
82
83	class FileSystemConfig(TypedDict):	1✔
84	Arn: str \| None	1✔
85	LocalMountPath: str \| None	1✔
86
87
88	class Tag(TypedDict):	1✔
89	Key: str \| None	1✔
90	Value: str \| None	1✔
91
92
93	class ImageConfig(TypedDict):	1✔
94	Command: list[str] \| None	1✔
95	EntryPoint: list[str] \| None	1✔
96	WorkingDirectory: str \| None	1✔
97
98
99	class DeadLetterConfig(TypedDict):	1✔
100	TargetArn: str \| None	1✔
101
102
103	class SnapStartResponse(TypedDict):	1✔
104	ApplyOn: str \| None	1✔
105	OptimizationStatus: str \| None	1✔
106
107
108	class Code(TypedDict):	1✔
109	ImageUri: str \| None	1✔
110	S3Bucket: str \| None	1✔
111	S3Key: str \| None	1✔
112	S3ObjectVersion: str \| None	1✔
113	ZipFile: str \| None	1✔
114
115
116	class LoggingConfig(TypedDict):	1✔
117	ApplicationLogLevel: str \| None	1✔
118	LogFormat: str \| None	1✔
119	LogGroup: str \| None	1✔
120	SystemLogLevel: str \| None	1✔
121
122
123	class Environment(TypedDict):	1✔
124	Variables: dict \| None	1✔
125
126
127	class EphemeralStorage(TypedDict):	1✔
128	Size: int \| None	1✔
129
130
131	REPEATED_INVOCATION = "repeated_invocation"	1✔
132
133	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
134	PYTHON_CFN_RESPONSE_CONTENT = """	1✔
135	# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
136	# SPDX-License-Identifier: MIT-0
137
138	from __future__ import print_function
139	import urllib3
140	import json
141
142	SUCCESS = "SUCCESS"
143	FAILED = "FAILED"
144
145	http = urllib3.PoolManager()
146
147
148	def send(event, context, responseStatus, responseData, physicalResourceId=None, noEcho=False, reason=None):
149	responseUrl = event['ResponseURL']
150
151	print(responseUrl)
152
153	responseBody = {
154	'Status' : responseStatus,
155	'Reason' : reason or "See the details in CloudWatch Log Stream: {}".format(context.log_stream_name),
156	'PhysicalResourceId' : physicalResourceId or context.log_stream_name,
157	'StackId' : event['StackId'],
158	'RequestId' : event['RequestId'],
159	'LogicalResourceId' : event['LogicalResourceId'],
160	'NoEcho' : noEcho,
161	'Data' : responseData
162	}
163
164	json_responseBody = json.dumps(responseBody)
165
166	print("Response body:")
167	print(json_responseBody)
168
169	headers = {
170	'content-type' : '',
171	'content-length' : str(len(json_responseBody))
172	}
173
174	try:
175	response = http.request('PUT', responseUrl, headers=headers, body=json_responseBody)
176	print("Status code:", response.status)
177
178
179	except Exception as e:
180
181	print("send(..) failed executing http.request(..):", e)
182	"""
183
184	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
185	NODEJS_CFN_RESPONSE_CONTENT = r"""	1✔
186	// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
187	// SPDX-License-Identifier: MIT-0
188
189	exports.SUCCESS = "SUCCESS";
190	exports.FAILED = "FAILED";
191
192	exports.send = function(event, context, responseStatus, responseData, physicalResourceId, noEcho) {
193
194	var responseBody = JSON.stringify({
195	Status: responseStatus,
196	Reason: "See the details in CloudWatch Log Stream: " + context.logStreamName,
197	PhysicalResourceId: physicalResourceId \|\| context.logStreamName,
198	StackId: event.StackId,
199	RequestId: event.RequestId,
200	LogicalResourceId: event.LogicalResourceId,
201	NoEcho: noEcho \|\| false,
202	Data: responseData
203	});
204
205	console.log("Response body:\n", responseBody);
206
207	var https = require("https");
208	var url = require("url");
209
210	var parsedUrl = url.parse(event.ResponseURL);
211	var options = {
212	hostname: parsedUrl.hostname,
213	port: parsedUrl.port, // Modified line: LS uses port 4566 for https; hard coded 443 causes error
214	path: parsedUrl.path,
215	method: "PUT",
216	headers: {
217	"content-type": "",
218	"content-length": responseBody.length
219	}
220	};
221
222	var request = https.request(options, function(response) {
223	console.log("Status code: " + parseInt(response.statusCode));
224	context.done();
225	});
226
227	request.on("error", function(error) {
228	console.log("send(..) failed executing https.request(..): " + error);
229	context.done();
230	});
231
232	request.write(responseBody);
233	request.end();
234	}
235	"""
236
237
238	def _runtime_supports_inline_code(runtime: str) -> bool:	1✔
239	return runtime.startswith("python") or runtime.startswith("node")	1✔
240
241
242	def _get_lambda_code_param(	1✔
243	properties: LambdaFunctionProperties,
244	_include_arch=False,
245	):
246	# code here is mostly taken directly from legacy implementation
247	code = properties.get("Code", {}).copy()	1✔
248
249	# TODO: verify only one of "ImageUri" \| "S3Bucket" \| "ZipFile" is set
250	zip_file = code.get("ZipFile")	1✔
251	if zip_file and not _runtime_supports_inline_code(properties["Runtime"]):	1✔
UNCOV 252	raise Exception(	×
253	f"Runtime {properties['Runtime']} doesn't support inlining code via the 'ZipFile' property."
254	) # TODO: message not validated
255	if zip_file and not is_base64(zip_file) and not is_zip_file(to_bytes(zip_file)):	1✔
256	tmp_dir = new_tmp_dir()	1✔
257	try:	1✔
258	handler_file = get_handler_file_from_name(	1✔
259	properties["Handler"], runtime=properties["Runtime"]
260	)
261	tmp_file = os.path.join(tmp_dir, handler_file)	1✔
262	save_file(tmp_file, zip_file)	1✔
263
264	# CloudFormation only includes cfn-response libs if an import is detected
265	# TODO: add snapshots for this behavior
266	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
267	if properties["Runtime"].lower().startswith("node") and (	1✔
268	"require('cfn-response')" in zip_file or 'require("cfn-response")' in zip_file
269	):
270	# the check if cfn-response is used is pretty simplistic and apparently based on simple string matching
271	# having the import commented out will also lead to cfn-response.js being injected
272	# this is added under both cfn-response.js and node_modules/cfn-response.js
273	cfn_response_mod_dir = os.path.join(tmp_dir, "node_modules")	×
274	mkdir(cfn_response_mod_dir)	×
UNCOV 275	save_file(	×
276	os.path.join(cfn_response_mod_dir, "cfn-response.js"),
277	NODEJS_CFN_RESPONSE_CONTENT,
278	)
UNCOV 279	save_file(os.path.join(tmp_dir, "cfn-response.js"), NODEJS_CFN_RESPONSE_CONTENT)	×
280	elif (	1✔
281	properties["Runtime"].lower().startswith("python")
282	and "import cfnresponse" in zip_file
283	):
UNCOV 284	save_file(os.path.join(tmp_dir, "cfnresponse.py"), PYTHON_CFN_RESPONSE_CONTENT)	×
285
286	# create zip file
287	zip_file = create_zip_file(tmp_dir, get_content=True)	1✔
288	code["ZipFile"] = zip_file	1✔
289	finally:
290	rm_rf(tmp_dir)	1✔
291	if _include_arch and "Architectures" in properties:	1✔
UNCOV 292	code["Architectures"] = properties.get("Architectures")	×
293	return code	1✔
294
295
296	def _transform_function_to_model(function):	1✔
UNCOV 297	model_properties = [	×
298	"MemorySize",
299	"Description",
300	"TracingConfig",
301	"Timeout",
302	"Handler",
303	"SnapStartResponse",
304	"Role",
305	"FileSystemConfigs",
306	"FunctionName",
307	"Runtime",
308	"PackageType",
309	"LoggingConfig",
310	"Environment",
311	"Arn",
312	"EphemeralStorage",
313	"Architectures",
314	"CapacityProviderConfig",
315	]
316	response_model = util.select_attributes(function, model_properties)	×
317	response_model["Arn"] = function["FunctionArn"]	×
UNCOV 318	return response_model	×
319
320
321	class LambdaFunctionProvider(LambdaFunctionProviderBase):	1✔
322	def create(	1✔
323	self,
324	request: ResourceRequest[LambdaFunctionProperties],
325	) -> ProgressEvent[LambdaFunctionProperties]:
326	"""
327	Create a new resource.
328
329	Primary identifier fields:
330	- /properties/FunctionName
331
332	Required properties:
333	- Code
334	- Role
335
336	Create-only properties:
337	- /properties/FunctionName
338	- /properties/PackageType
339	- /properties/TenancyConfig
340
341	Read-only properties:
342	- /properties/Arn
343	- /properties/SnapStartResponse
344	- /properties/SnapStartResponse/ApplyOn
345	- /properties/SnapStartResponse/OptimizationStatus
346
347	IAM permissions required:
348	- lambda:CreateFunction
349	- lambda:GetFunction
350	- lambda:PutFunctionConcurrency
351	- iam:PassRole
352	- s3:GetObject
353	- s3:GetObjectVersion
354	- ec2:DescribeSecurityGroups
355	- ec2:DescribeSubnets
356	- ec2:DescribeVpcs
357	- elasticfilesystem:DescribeMountTargets
358	- kms:CreateGrant
359	- kms:Decrypt
360	- kms:Encrypt
361	- kms:GenerateDataKey
362	- lambda:GetCodeSigningConfig
363	- lambda:GetFunctionCodeSigningConfig
364	- lambda:GetLayerVersion
365	- lambda:GetRuntimeManagementConfig
366	- lambda:PutRuntimeManagementConfig
367	- lambda:TagResource
368	- lambda:GetPolicy
369	- lambda:PutFunctionRecursionConfig
370	- lambda:GetFunctionRecursionConfig
371	- lambda:PutFunctionScalingConfig
372	- lambda:PassCapacityProvider
373
374	"""
375	model = request.desired_state	1✔
376	lambda_client = request.aws_client_factory.lambda_	1✔
377
378	if not request.custom_context.get(REPEATED_INVOCATION):	1✔
379	request.custom_context[REPEATED_INVOCATION] = True	1✔
380
381	name = model.get("FunctionName")	1✔
382	if not name:	1✔
383	name = util.generate_default_name(request.stack_name, request.logical_resource_id)	1✔
384	model["FunctionName"] = name	1✔
385
386	kwargs = util.select_attributes(	1✔
387	model,
388	[
389	"Architectures",
390	"DeadLetterConfig",
391	"Description",
392	"FunctionName",
393	"Handler",
394	"ImageConfig",
395	"PackageType",
396	"Layers",
397	"MemorySize",
398	"Runtime",
399	"Role",
400	"Timeout",
401	"TracingConfig",
402	"VpcConfig",
403	"LoggingConfig",
404	"CapacityProviderConfig",
405	],
406	)
407	if "Timeout" in kwargs:	1✔
408	kwargs["Timeout"] = int(kwargs["Timeout"])	1✔
409	if "MemorySize" in kwargs:	1✔
410	kwargs["MemorySize"] = int(kwargs["MemorySize"])	1✔
411	if model_tags := model.get("Tags"):	1✔
412	tags = {}	1✔
413	for tag in model_tags:	1✔
414	tags[tag["Key"]] = tag["Value"]	1✔
415	kwargs["Tags"] = tags	1✔
416
417	# botocore/data/lambda/2015-03-31/service-2.json:1161 (EnvironmentVariableValue)
418	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-function-environment.html
419	if "Environment" in model:	1✔
420	environment_variables = model["Environment"].get("Variables", {})	1✔
421	kwargs["Environment"] = {	1✔
422	"Variables": {k: str(v) for k, v in environment_variables.items()}
423	}
424
425	kwargs["Code"] = _get_lambda_code_param(model)	1✔
426
427	# For managed instance lambdas, we publish them immediately
428	if "CapacityProviderConfig" in kwargs:	1✔
429	kwargs["Publish"] = True	×
430	kwargs["PublishTo"] = "LATEST_PUBLISHED"	×
431
432	create_response = lambda_client.create_function(**kwargs)	1✔
433	# TODO: if version is in the schema, just put it in the model instead of the custom context
434	request.custom_context["Version"] = create_response["Version"] # $LATEST.PUBLISHED	1✔
435	model["Arn"] = create_response["FunctionArn"]	1✔
436
437	if request.custom_context.get("Version") == "$LATEST.PUBLISHED":	1✔
438	# for managed instance lambdas, we need to wait until the version is published & active
439	get_fn_response = lambda_client.get_function(	×
440	FunctionName=model["FunctionName"], Qualifier=request.custom_context["Version"]
441	)
442	else:
443	get_fn_response = lambda_client.get_function(FunctionName=model["Arn"])	1✔
444
445	match get_fn_response["Configuration"]["State"]:	1✔
446	# TODO: explicitly handle new ActiveNonInvocable state?
447	case "Pending":	1✔
448	return ProgressEvent(	1✔
449	status=OperationStatus.IN_PROGRESS,
450	resource_model=model,
451	custom_context=request.custom_context,
452	)
453	case "Active":	1✔
454	return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)	1✔
455	case "Inactive":	×
456	# This might happen when setting LAMBDA_KEEPALIVE_MS=0
457	return ProgressEvent(status=OperationStatus.SUCCESS, resource_model=model)	×
458	case "Failed":	×
459	return ProgressEvent(	×
460	status=OperationStatus.FAILED,
461	resource_model=model,
462	error_code=get_fn_response["Configuration"].get("StateReasonCode", "unknown"),
463	message=get_fn_response["Configuration"].get("StateReason", "unknown"),
464	)
465	case unknown_state: # invalid state, should technically never happen	×
466	return ProgressEvent(	×
467	status=OperationStatus.FAILED,
468	resource_model=model,
469	error_code="InternalException",
470	message=f"Invalid state returned: {unknown_state}",
471	)
472
473	def read(	1✔
474	self,
475	request: ResourceRequest[LambdaFunctionProperties],
476	) -> ProgressEvent[LambdaFunctionProperties]:
477	"""
478	Fetch resource information
479
480	IAM permissions required:
481	- lambda:GetFunction
482	- lambda:GetFunctionCodeSigningConfig
483	- lambda:GetFunctionRecursionConfig
484	- lambda:GetRuntimeManagementConfig
485	- lambda:GetFunctionScalingConfig
486	"""
UNCOV 487	function_name = request.desired_state["FunctionName"]	×
488	lambda_client = request.aws_client_factory.lambda_	×
UNCOV 489	get_fn_response = lambda_client.get_function(FunctionName=function_name)	×
490
UNCOV 491	return ProgressEvent(	×
492	status=OperationStatus.SUCCESS,
493	resource_model=_transform_function_to_model(get_fn_response["Configuration"]),
494	)
495
496	def delete(	1✔
497	self,
498	request: ResourceRequest[LambdaFunctionProperties],
499	) -> ProgressEvent[LambdaFunctionProperties]:
500	"""
501	Delete a resource
502
503	IAM permissions required:
504	- lambda:DeleteFunction
505	- lambda:GetFunction
506	- ec2:DescribeNetworkInterfaces
507	"""
508	try:	1✔
509	lambda_client = request.aws_client_factory.lambda_	1✔
510	lambda_client.delete_function(FunctionName=request.previous_state["FunctionName"])	1✔
511	except request.aws_client_factory.lambda_.exceptions.ResourceNotFoundException:	1✔
512	pass	1✔
513	# any other exception will be propagated
514	return ProgressEvent(status=OperationStatus.SUCCESS, resource_model={})	1✔
515
516	def update(	1✔
517	self,
518	request: ResourceRequest[LambdaFunctionProperties],
519	) -> ProgressEvent[LambdaFunctionProperties]:
520	"""
521	Update a resource
522
523	IAM permissions required:
524	- lambda:DeleteFunctionConcurrency
525	- lambda:GetFunction
526	- lambda:PutFunctionConcurrency
527	- lambda:TagResource
528	- lambda:UntagResource
529	- lambda:UpdateFunctionConfiguration
530	- lambda:UpdateFunctionCode
531	- iam:PassRole
532	- s3:GetObject
533	- s3:GetObjectVersion
534	- ec2:DescribeSecurityGroups
535	- ec2:DescribeSubnets
536	- ec2:DescribeVpcs
537	- elasticfilesystem:DescribeMountTargets
538	- kms:CreateGrant
539	- kms:Decrypt
540	- kms:GenerateDataKey
541	- lambda:GetRuntimeManagementConfig
542	- lambda:PutRuntimeManagementConfig
543	- lambda:PutFunctionCodeSigningConfig
544	- lambda:DeleteFunctionCodeSigningConfig
545	- lambda:GetCodeSigningConfig
546	- lambda:GetFunctionCodeSigningConfig
547	- lambda:PutFunctionRecursionConfig
548	- lambda:GetFunctionRecursionConfig
549	- lambda:PutFunctionScalingConfig
550	- lambda:PublishVersion
551	- lambda:PassCapacityProvider
552	"""
553	client = request.aws_client_factory.lambda_	1✔
554
555	# TODO: handle defaults properly
556	old_name = request.previous_state["FunctionName"]	1✔
557	new_name = request.desired_state.get("FunctionName")	1✔
558	if new_name and old_name != new_name:	1✔
559	# replacement (!) => shouldn't be handled here but in the engine
560	self.delete(request)	1✔
561	return self.create(request)	1✔
562
563	config_keys = [	1✔
564	"Description",
565	"DeadLetterConfig",
566	"Environment",
567	"Handler",
568	"ImageConfig",
569	"Layers",
570	"MemorySize",
571	"Role",
572	"Runtime",
573	"Timeout",
574	"TracingConfig",
575	"VpcConfig",
576	"LoggingConfig",
577	"CapacityProviderConfig",
578	]
579	update_config_props = util.select_attributes(request.desired_state, config_keys)	1✔
580	function_name = request.previous_state["FunctionName"]	1✔
581	update_config_props["FunctionName"] = function_name	1✔
582
583	if "Timeout" in update_config_props:	1✔
584	update_config_props["Timeout"] = int(update_config_props["Timeout"])	1✔
585	if "MemorySize" in update_config_props:	1✔
586	update_config_props["MemorySize"] = int(update_config_props["MemorySize"])	1✔
587	if "Code" in request.desired_state:	1✔
588	code = request.desired_state["Code"] or {}	1✔
589	if not code.get("ZipFile"):	1✔
590	request.logger.debug(	1✔
591	'Updating code for Lambda "%s" from location: %s', function_name, code
592	)
593	code = _get_lambda_code_param(	1✔
594	request.desired_state,
595	_include_arch=True,
596	)
597	client.update_function_code(FunctionName=function_name, **code)	1✔
598	client.get_waiter("function_updated_v2").wait(FunctionName=function_name)	1✔
599	if "Environment" in update_config_props:	1✔
600	environment_variables = update_config_props["Environment"].get("Variables", {})	1✔
601	update_config_props["Environment"]["Variables"] = {	1✔
602	k: str(v) for k, v in environment_variables.items()
603	}
604	client.update_function_configuration(**update_config_props)	1✔
605	client.get_waiter("function_updated_v2").wait(FunctionName=function_name)	1✔
606	return ProgressEvent(	1✔
607	status=OperationStatus.SUCCESS,
608	resource_model={request.previous_state, request.desired_state},
609	)
610
611	def list(	1✔
612	self,
613	request: ResourceRequest[LambdaFunctionProperties],
614	) -> ProgressEvent[LambdaFunctionProperties]:
615	"""
616	List available resources of this type
617	IAM permissions required:
618	- lambda:ListFunctions
619	"""
UNCOV 620	functions = request.aws_client_factory.lambda_.list_functions()	×
UNCOV 621	return ProgressEvent(	×
622	status=OperationStatus.SUCCESS,
623	resource_models=[_transform_function_to_model(fn) for fn in functions["Functions"]],
624	)

localstack / localstack / 21615209021

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous