fb37c2a2-4bd7-49b4-891b-ede18134b4b1

Committed 20 Jan 2025 04:22PM UTC coverage: 86.826% (-0.02%) from 86.844%

Build # fb37c2a2-4bd7-49b4-891b-ede18134b4b1

Build Type

push

circleci

Committed by

web-flow

Commit Message

Add tests for KMS resource tagging and untagging (#12121)

Coverage Stats

61076 of 70343 relevant lines covered (86.83%)

0.87 hits per line

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

87.89

/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 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
          - kms:Decrypt
          - lambda:GetCodeSigningConfig
          - lambda:GetFunctionCodeSigningConfig
          - lambda:GetRuntimeManagementConfig
          - lambda:PutRuntimeManagementConfig

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

localstack / localstack / fb37c2a2-4bd7-49b4-891b-ede18134b4b1

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