20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

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

86.54

/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 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 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(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",
                    "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
        """
        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",
            "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]:
        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 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 LambdaManagedInstancesCapacityProviderConfig(TypedDict):	1✔
23	CapacityProviderArn: str \| None	1✔
24	PerExecutionEnvironmentMaxConcurrency: int \| None	1✔
25	ExecutionEnvironmentMemoryGiBPerVCpu: float \| None	1✔
26
27
28	class CapacityProviderConfig(TypedDict):	1✔
29	LambdaManagedInstancesCapacityProviderConfig: (	1✔
30	LambdaManagedInstancesCapacityProviderConfig \| None
31	)
32
33
34	class LambdaFunctionProperties(TypedDict):	1✔
35	Code: Code \| None	1✔
36	Role: str \| None	1✔
37	Architectures: list[str] \| None	1✔
38	Arn: str \| None	1✔
39	CapacityProviderConfig: CapacityProviderConfig \| None	1✔
40	CodeSigningConfigArn: str \| None	1✔
41	DeadLetterConfig: DeadLetterConfig \| None	1✔
42	Description: str \| None	1✔
43	Environment: Environment \| None	1✔
44	EphemeralStorage: EphemeralStorage \| None	1✔
45	FileSystemConfigs: list[FileSystemConfig] \| None	1✔
46	FunctionName: str \| None	1✔
47	Handler: str \| None	1✔
48	ImageConfig: ImageConfig \| None	1✔
49	KmsKeyArn: str \| None	1✔
50	Layers: list[str] \| None	1✔
51	MemorySize: int \| None	1✔
52	PackageType: str \| None	1✔
53	ReservedConcurrentExecutions: int \| None	1✔
54	Runtime: str \| None	1✔
55	RuntimeManagementConfig: RuntimeManagementConfig \| None	1✔
56	SnapStart: SnapStart \| None	1✔
57	SnapStartResponse: SnapStartResponse \| None	1✔
58	Tags: list[Tag] \| None	1✔
59	Timeout: int \| None	1✔
60	TracingConfig: TracingConfig \| None	1✔
61	VpcConfig: VpcConfig \| None	1✔
62
63
64	class TracingConfig(TypedDict):	1✔
65	Mode: str \| None	1✔
66
67
68	class VpcConfig(TypedDict):	1✔
69	SecurityGroupIds: list[str] \| None	1✔
70	SubnetIds: list[str] \| None	1✔
71
72
73	class RuntimeManagementConfig(TypedDict):	1✔
74	UpdateRuntimeOn: str \| None	1✔
75	RuntimeVersionArn: str \| None	1✔
76
77
78	class SnapStart(TypedDict):	1✔
79	ApplyOn: str \| None	1✔
80
81
82	class FileSystemConfig(TypedDict):	1✔
83	Arn: str \| None	1✔
84	LocalMountPath: str \| None	1✔
85
86
87	class Tag(TypedDict):	1✔
88	Key: str \| None	1✔
89	Value: str \| None	1✔
90
91
92	class ImageConfig(TypedDict):	1✔
93	Command: list[str] \| None	1✔
94	EntryPoint: list[str] \| None	1✔
95	WorkingDirectory: str \| None	1✔
96
97
98	class DeadLetterConfig(TypedDict):	1✔
99	TargetArn: str \| None	1✔
100
101
102	class SnapStartResponse(TypedDict):	1✔
103	ApplyOn: str \| None	1✔
104	OptimizationStatus: str \| None	1✔
105
106
107	class Code(TypedDict):	1✔
108	ImageUri: str \| None	1✔
109	S3Bucket: str \| None	1✔
110	S3Key: str \| None	1✔
111	S3ObjectVersion: str \| None	1✔
112	ZipFile: str \| None	1✔
113
114
115	class LoggingConfig(TypedDict):	1✔
116	ApplicationLogLevel: str \| None	1✔
117	LogFormat: str \| None	1✔
118	LogGroup: str \| None	1✔
119	SystemLogLevel: str \| None	1✔
120
121
122	class Environment(TypedDict):	1✔
123	Variables: dict \| None	1✔
124
125
126	class EphemeralStorage(TypedDict):	1✔
127	Size: int \| None	1✔
128
129
130	REPEATED_INVOCATION = "repeated_invocation"	1✔
131
132	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
133	PYTHON_CFN_RESPONSE_CONTENT = """	1✔
134	# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
135	# SPDX-License-Identifier: MIT-0
136
137	from __future__ import print_function
138	import urllib3
139	import json
140
141	SUCCESS = "SUCCESS"
142	FAILED = "FAILED"
143
144	http = urllib3.PoolManager()
145
146
147	def send(event, context, responseStatus, responseData, physicalResourceId=None, noEcho=False, reason=None):
148	responseUrl = event['ResponseURL']
149
150	print(responseUrl)
151
152	responseBody = {
153	'Status' : responseStatus,
154	'Reason' : reason or "See the details in CloudWatch Log Stream: {}".format(context.log_stream_name),
155	'PhysicalResourceId' : physicalResourceId or context.log_stream_name,
156	'StackId' : event['StackId'],
157	'RequestId' : event['RequestId'],
158	'LogicalResourceId' : event['LogicalResourceId'],
159	'NoEcho' : noEcho,
160	'Data' : responseData
161	}
162
163	json_responseBody = json.dumps(responseBody)
164
165	print("Response body:")
166	print(json_responseBody)
167
168	headers = {
169	'content-type' : '',
170	'content-length' : str(len(json_responseBody))
171	}
172
173	try:
174	response = http.request('PUT', responseUrl, headers=headers, body=json_responseBody)
175	print("Status code:", response.status)
176
177
178	except Exception as e:
179
180	print("send(..) failed executing http.request(..):", e)
181	"""
182
183	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
184	NODEJS_CFN_RESPONSE_CONTENT = r"""	1✔
185	// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
186	// SPDX-License-Identifier: MIT-0
187
188	exports.SUCCESS = "SUCCESS";
189	exports.FAILED = "FAILED";
190
191	exports.send = function(event, context, responseStatus, responseData, physicalResourceId, noEcho) {
192
193	var responseBody = JSON.stringify({
194	Status: responseStatus,
195	Reason: "See the details in CloudWatch Log Stream: " + context.logStreamName,
196	PhysicalResourceId: physicalResourceId \|\| context.logStreamName,
197	StackId: event.StackId,
198	RequestId: event.RequestId,
199	LogicalResourceId: event.LogicalResourceId,
200	NoEcho: noEcho \|\| false,
201	Data: responseData
202	});
203
204	console.log("Response body:\n", responseBody);
205
206	var https = require("https");
207	var url = require("url");
208
209	var parsedUrl = url.parse(event.ResponseURL);
210	var options = {
211	hostname: parsedUrl.hostname,
212	port: parsedUrl.port, // Modified line: LS uses port 4566 for https; hard coded 443 causes error
213	path: parsedUrl.path,
214	method: "PUT",
215	headers: {
216	"content-type": "",
217	"content-length": responseBody.length
218	}
219	};
220
221	var request = https.request(options, function(response) {
222	console.log("Status code: " + parseInt(response.statusCode));
223	context.done();
224	});
225
226	request.on("error", function(error) {
227	console.log("send(..) failed executing https.request(..): " + error);
228	context.done();
229	});
230
231	request.write(responseBody);
232	request.end();
233	}
234	"""
235
236
237	def _runtime_supports_inline_code(runtime: str) -> bool:	1✔
238	return runtime.startswith("python") or runtime.startswith("node")	1✔
239
240
241	def _get_lambda_code_param(	1✔
242	properties: LambdaFunctionProperties,
243	_include_arch=False,
244	):
245	# code here is mostly taken directly from legacy implementation
246	code = properties.get("Code", {}).copy()	1✔
247
248	# TODO: verify only one of "ImageUri" \| "S3Bucket" \| "ZipFile" is set
249	zip_file = code.get("ZipFile")	1✔
250	if zip_file and not _runtime_supports_inline_code(properties["Runtime"]):	1✔
251	raise Exception(	×
252	f"Runtime {properties['Runtime']} doesn't support inlining code via the 'ZipFile' property."
253	) # TODO: message not validated
254	if zip_file and not is_base64(zip_file) and not is_zip_file(to_bytes(zip_file)):	1✔
255	tmp_dir = new_tmp_dir()	1✔
256	try:	1✔
257	handler_file = get_handler_file_from_name(	1✔
258	properties["Handler"], runtime=properties["Runtime"]
259	)
260	tmp_file = os.path.join(tmp_dir, handler_file)	1✔
261	save_file(tmp_file, zip_file)	1✔
262
263	# CloudFormation only includes cfn-response libs if an import is detected
264	# TODO: add snapshots for this behavior
265	# https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html
266	if properties["Runtime"].lower().startswith("node") and (	1✔
267	"require('cfn-response')" in zip_file or 'require("cfn-response")' in zip_file
268	):
269	# the check if cfn-response is used is pretty simplistic and apparently based on simple string matching
270	# having the import commented out will also lead to cfn-response.js being injected
271	# this is added under both cfn-response.js and node_modules/cfn-response.js
272	cfn_response_mod_dir = os.path.join(tmp_dir, "node_modules")	×
273	mkdir(cfn_response_mod_dir)	×
274	save_file(	×
275	os.path.join(cfn_response_mod_dir, "cfn-response.js"),
276	NODEJS_CFN_RESPONSE_CONTENT,
277	)
278	save_file(os.path.join(tmp_dir, "cfn-response.js"), NODEJS_CFN_RESPONSE_CONTENT)	×
279	elif (	1✔
280	properties["Runtime"].lower().startswith("python")
281	and "import cfnresponse" in zip_file
282	):
283	save_file(os.path.join(tmp_dir, "cfnresponse.py"), PYTHON_CFN_RESPONSE_CONTENT)	×
284
285	# create zip file
286	zip_file = create_zip_file(tmp_dir, get_content=True)	1✔
287	code["ZipFile"] = zip_file	1✔
288	finally:
289	rm_rf(tmp_dir)	1✔
290	if _include_arch and "Architectures" in properties:	1✔
291	code["Architectures"] = properties.get("Architectures")	×
292	return code	1✔
293
294
295	def _transform_function_to_model(function):	1✔
296	model_properties = [	×
297	"MemorySize",
298	"Description",
299	"TracingConfig",
300	"Timeout",
301	"Handler",
302	"SnapStartResponse",
303	"Role",
304	"FileSystemConfigs",
305	"FunctionName",
306	"Runtime",
307	"PackageType",
308	"LoggingConfig",
309	"Environment",
310	"Arn",
311	"EphemeralStorage",
312	"Architectures",
313	"CapacityProviderConfig",
314	]
315	response_model = util.select_attributes(function, model_properties)	×
316	response_model["Arn"] = function["FunctionArn"]	×
317	return response_model	×
318
319
320	class LambdaFunctionProvider(ResourceProvider[LambdaFunctionProperties]):	1✔
321	TYPE = "AWS::Lambda::Function" # Autogenerated. Don't change	1✔
322	SCHEMA = util.get_schema_path(Path(__file__)) # Autogenerated. Don't change	1✔
323
324	def create(	1✔
325	self,
326	request: ResourceRequest[LambdaFunctionProperties],
327	) -> ProgressEvent[LambdaFunctionProperties]:
328	"""
329	Create a new resource.
330
331	Primary identifier fields:
332	- /properties/FunctionName
333
334	Required properties:
335	- Code
336	- Role
337
338	Create-only properties:
339	- /properties/FunctionName
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:AddPermission
370	- lambda:RemovePermission
371	- lambda:GetResourcePolicy
372	- lambda:PutResourcePolicy
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	"""
484	function_name = request.desired_state["FunctionName"]	×
485	lambda_client = request.aws_client_factory.lambda_	×
486	get_fn_response = lambda_client.get_function(FunctionName=function_name)	×
487
488	return ProgressEvent(	×
489	status=OperationStatus.SUCCESS,
490	resource_model=_transform_function_to_model(get_fn_response["Configuration"]),
491	)
492
493	def delete(	1✔
494	self,
495	request: ResourceRequest[LambdaFunctionProperties],
496	) -> ProgressEvent[LambdaFunctionProperties]:
497	"""
498	Delete a resource
499
500	IAM permissions required:
501	- lambda:DeleteFunction
502	- ec2:DescribeNetworkInterfaces
503	"""
504	try:	1✔
505	lambda_client = request.aws_client_factory.lambda_	1✔
506	lambda_client.delete_function(FunctionName=request.previous_state["FunctionName"])	1✔
507	except request.aws_client_factory.lambda_.exceptions.ResourceNotFoundException:	1✔
508	pass	1✔
509	# any other exception will be propagated
510	return ProgressEvent(status=OperationStatus.SUCCESS, resource_model={})	1✔
511
512	def update(	1✔
513	self,
514	request: ResourceRequest[LambdaFunctionProperties],
515	) -> ProgressEvent[LambdaFunctionProperties]:
516	"""
517	Update a resource
518
519	IAM permissions required:
520	- lambda:DeleteFunctionConcurrency
521	- lambda:GetFunction
522	- lambda:PutFunctionConcurrency
523	- lambda:ListTags
524	- lambda:TagResource
525	- lambda:UntagResource
526	- lambda:UpdateFunctionConfiguration
527	- lambda:UpdateFunctionCode
528	- iam:PassRole
529	- s3:GetObject
530	- s3:GetObjectVersion
531	- ec2:DescribeSecurityGroups
532	- ec2:DescribeSubnets
533	- ec2:DescribeVpcs
534	- elasticfilesystem:DescribeMountTargets
535	- kms:CreateGrant
536	- kms:Decrypt
537	- kms:GenerateDataKey
538	- lambda:GetRuntimeManagementConfig
539	- lambda:PutRuntimeManagementConfig
540	- lambda:PutFunctionCodeSigningConfig
541	- lambda:DeleteFunctionCodeSigningConfig
542	- lambda:GetCodeSigningConfig
543	- lambda:GetFunctionCodeSigningConfig
544	- lambda:GetPolicy
545	- lambda:AddPermission
546	- lambda:RemovePermission
547	- lambda:GetResourcePolicy
548	- lambda:PutResourcePolicy
549	- lambda:DeleteResourcePolicy
550	"""
551	client = request.aws_client_factory.lambda_	1✔
552
553	# TODO: handle defaults properly
554	old_name = request.previous_state["FunctionName"]	1✔
555	new_name = request.desired_state.get("FunctionName")	1✔
556	if new_name and old_name != new_name:	1✔
557	# replacement (!) => shouldn't be handled here but in the engine
558	self.delete(request)	×
559	return self.create(request)	×
560
561	config_keys = [	1✔
562	"Description",
563	"DeadLetterConfig",
564	"Environment",
565	"Handler",
566	"ImageConfig",
567	"Layers",
568	"MemorySize",
569	"Role",
570	"Runtime",
571	"Timeout",
572	"TracingConfig",
573	"VpcConfig",
574	"LoggingConfig",
575	"CapacityProviderConfig",
576	]
577	update_config_props = util.select_attributes(request.desired_state, config_keys)	1✔
578	function_name = request.previous_state["FunctionName"]	1✔
579	update_config_props["FunctionName"] = function_name	1✔
580
581	if "Timeout" in update_config_props:	1✔
582	update_config_props["Timeout"] = int(update_config_props["Timeout"])	1✔
583	if "MemorySize" in update_config_props:	1✔
584	update_config_props["MemorySize"] = int(update_config_props["MemorySize"])	1✔
585	if "Code" in request.desired_state:	1✔
586	code = request.desired_state["Code"] or {}	1✔
587	if not code.get("ZipFile"):	1✔
588	request.logger.debug(	1✔
589	'Updating code for Lambda "%s" from location: %s', function_name, code
590	)
591	code = _get_lambda_code_param(	1✔
592	request.desired_state,
593	_include_arch=True,
594	)
595	client.update_function_code(FunctionName=function_name, **code)	1✔
596	client.get_waiter("function_updated_v2").wait(FunctionName=function_name)	1✔
597	if "Environment" in update_config_props:	1✔
598	environment_variables = update_config_props["Environment"].get("Variables", {})	1✔
599	update_config_props["Environment"]["Variables"] = {	1✔
600	k: str(v) for k, v in environment_variables.items()
601	}
602	client.update_function_configuration(**update_config_props)	1✔
603	client.get_waiter("function_updated_v2").wait(FunctionName=function_name)	1✔
604	return ProgressEvent(	1✔
605	status=OperationStatus.SUCCESS,
606	resource_model={request.previous_state, request.desired_state},
607	)
608
609	def list(	1✔
610	self,
611	request: ResourceRequest[LambdaFunctionProperties],
612	) -> ProgressEvent[LambdaFunctionProperties]:
613	functions = request.aws_client_factory.lambda_.list_functions()	×
614	return ProgressEvent(	×
615	status=OperationStatus.SUCCESS,
616	resource_models=[_transform_function_to_model(fn) for fn in functions["Functions"]],
617	)

localstack / localstack / 20565403496

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