5f1fdd48-92f9-4508-ac88-80c0245b2bad

Committed 09 Apr 2025 07:07PM UTC coverage: 86.783% (-0.03%) from 86.809%

Build # 5f1fdd48-92f9-4508-ac88-80c0245b2bad

Build Type

push

circleci

Committed by

web-flow

Commit Message

EC2: generate security group ids using id manager concept (#12494)

Co-authored-by: Mathieu Cloutier <cloutier.mat0@gmail.com>

Coverage Stats

14 of 14 new or added lines in 1 file covered. (100.0%)

615 existing lines in 23 files now uncovered.

63754 of 73464 relevant lines covered (86.78%)

0.87 hits per line

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

88.21

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

# LocalStack Resource Provider Scaffolding v2
from __future__ import annotations

import os
from pathlib import Path
from typing import Optional, TypedDict

import localstack.services.cloudformation.provider_utils as util
from localstack.services.cloudformation.resource_provider import (
    OperationStatus,
    ProgressEvent,
    ResourceProvider,
    ResourceRequest,
)
from localstack.services.lambda_.lambda_utils import get_handler_file_from_name
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 LambdaFunctionProperties(TypedDict):
    Code: Optional[Code]
    Role: Optional[str]
    Architectures: Optional[list[str]]
    Arn: Optional[str]
    CodeSigningConfigArn: Optional[str]
    DeadLetterConfig: Optional[DeadLetterConfig]
    Description: Optional[str]
    Environment: Optional[Environment]
    EphemeralStorage: Optional[EphemeralStorage]
    FileSystemConfigs: Optional[list[FileSystemConfig]]
    FunctionName: Optional[str]
    Handler: Optional[str]
    ImageConfig: Optional[ImageConfig]
    KmsKeyArn: Optional[str]
    Layers: Optional[list[str]]
    MemorySize: Optional[int]
    PackageType: Optional[str]
    ReservedConcurrentExecutions: Optional[int]
    Runtime: Optional[str]
    RuntimeManagementConfig: Optional[RuntimeManagementConfig]
    SnapStart: Optional[SnapStart]
    SnapStartResponse: Optional[SnapStartResponse]
    Tags: Optional[list[Tag]]
    Timeout: Optional[int]
    TracingConfig: Optional[TracingConfig]
    VpcConfig: Optional[VpcConfig]


class TracingConfig(TypedDict):
    Mode: Optional[str]


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


class RuntimeManagementConfig(TypedDict):
    UpdateRuntimeOn: Optional[str]
    RuntimeVersionArn: Optional[str]


class SnapStart(TypedDict):
    ApplyOn: Optional[str]


class FileSystemConfig(TypedDict):
    Arn: Optional[str]
    LocalMountPath: Optional[str]


class Tag(TypedDict):
    Key: Optional[str]
    Value: Optional[str]


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


class DeadLetterConfig(TypedDict):
    TargetArn: Optional[str]


class SnapStartResponse(TypedDict):
    ApplyOn: Optional[str]
    OptimizationStatus: Optional[str]


class Code(TypedDict):
    ImageUri: Optional[str]
    S3Bucket: Optional[str]
    S3Key: Optional[str]
    S3ObjectVersion: Optional[str]
    ZipFile: Optional[str]


class LoggingConfig(TypedDict):
    ApplicationLogLevel: Optional[str]
    LogFormat: Optional[str]
    LogGroup: Optional[str]
    SystemLogLevel: Optional[str]


class Environment(TypedDict):
    Variables: Optional[dict]


class EphemeralStorage(TypedDict):
    Size: Optional[int]


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",
    ]
    response_model = util.select_attributes(function, model_properties)
    response_model["Arn"] = function["FunctionArn"]
    return response_model


class LambdaFunctionProvider(ResourceProvider[LambdaFunctionProperties]):
    TYPE = "AWS::Lambda::Function"  # Autogenerated. Don't change
    SCHEMA = util.get_schema_path(Path(__file__))  # Autogenerated. Don't change

    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

        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:AddPermission
          - lambda:RemovePermission
          - lambda:GetResourcePolicy
          - lambda:PutResourcePolicy

        """
        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",
                ],
            )
            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)
            create_response = lambda_client.create_function(**kwargs)
            model["Arn"] = create_response["FunctionArn"]

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

localstack / localstack / 5f1fdd48-92f9-4508-ac88-80c0245b2bad

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