import re

import os

import imp

import sys

import json

import uuid

import time

import base64

import hashlib

import logging

import functools

import threading

import traceback

from io import BytesIO

from datetime import datetime

from flask import Flask, Response, jsonify, request

from six.moves import cStringIO as StringIO

from six.moves.urllib.parse import urlparse

from localstack import config

from localstack.constants import APPLICATION_JSON, TEST_AWS_ACCOUNT_ID

from localstack.utils.aws import aws_stack, aws_responses

from localstack.utils.common import (

to_str, to_bytes, load_file, save_file, TMP_FILES, ensure_readable, short_uid, long_uid, json_safe,

mkdir, unzip, is_zip_file, run, run_safe, first_char_to_lower, run_for_max_seconds, parse_request_data,

timestamp_millis, timestamp, now_utc, safe_requests, FuncThread, isoformat_milliseconds, synchronized)

from localstack.services.awslambda import lambda_executors

from localstack.services.generic_proxy import RegionBackend

from localstack.services.awslambda.lambda_utils import (

DOTNET_LAMBDA_RUNTIMES, multi_value_dict_for_list, get_handler_file_from_name,

LAMBDA_DEFAULT_HANDLER, LAMBDA_DEFAULT_RUNTIME, LAMBDA_DEFAULT_STARTING_POSITION)

from localstack.utils.analytics import event_publisher

from localstack.utils.http_utils import parse_chunked_data

from localstack.utils.aws.aws_models import LambdaFunction, CodeSigningConfig

from localstack.services.cloudformation.service_models import LAMBDA_POLICY_NAME_PATTERN

# logger

LOG = logging.getLogger(__name__)

# constants

APP_NAME = 'lambda_api'

PATH_ROOT = '/2015-03-31'

ARCHIVE_FILE_PATTERN = '%s/lambda.handler.*.jar' % config.TMP_FOLDER

LAMBDA_SCRIPT_PATTERN = '%s/lambda_script_*.py' % config.TMP_FOLDER

LAMBDA_ZIP_FILE_NAME = 'original_lambda_archive.zip'

LAMBDA_JAR_FILE_NAME = 'original_lambda_archive.jar'

# default timeout in seconds

LAMBDA_DEFAULT_TIMEOUT = 3

INVALID_PARAMETER_VALUE_EXCEPTION = 'InvalidParameterValueException'

VERSION_LATEST = '$LATEST'

FUNCTION_MAX_SIZE = 69905067

BATCH_SIZE_RANGES = {

'kafka': (100, 10000),

'kinesis': (100, 10000),

'dynamodb': (100, 1000),

'sqs': (10, 10)

}

app = Flask(APP_NAME)

# mutex for access to CWD and ENV

EXEC_MUTEX = threading.RLock(1)

# whether to use Docker for execution

DO_USE_DOCKER = None

# start characters indicating that a lambda result should be parsed as JSON

JSON_START_CHAR_MAP = {

list: ('[',),

tuple: ('[',),

dict: ('{',),

str: ('"',),

bytes: ('"',),

bool: ('t', 'f'),

type(None): ('n',),

int: ('0', '1', '2', '3', '4', '5', '6', '7', '8', '9'),

float: ('0', '1', '2', '3', '4', '5', '6', '7', '8', '9')

}

POSSIBLE_JSON_TYPES = (str, bytes)

JSON_START_TYPES = tuple(set(JSON_START_CHAR_MAP.keys()) - set(POSSIBLE_JSON_TYPES))

JSON_START_CHARS = tuple(set(functools.reduce(lambda x, y: x + y, JSON_START_CHAR_MAP.values())))

# SQS listener thread settings

SQS_LISTENER_THREAD = {}

SQS_POLL_INTERVAL_SEC = 1

# lambda executor instance

LAMBDA_EXECUTOR = lambda_executors.AVAILABLE_EXECUTORS.get(config.LAMBDA_EXECUTOR, lambda_executors.DEFAULT_EXECUTOR)

# IAM policy constants

IAM_POLICY_VERSION = '2012-10-17'

# Whether to check if the handler function exists while creating lambda function

CHECK_HANDLER_ON_CREATION = False

# Marker name to indicate that a bucket represents the local file system. This is used for testing

# Serverless applications where we mount the Lambda code directly into the container from the host OS.

BUCKET_MARKER_LOCAL = '__local__'

class LambdaRegion(RegionBackend):

def __init__(self):

# map ARN strings to lambda function objects

self.lambdas = {}

# map ARN strings to CodeSigningConfig object

self.code_signing_configs = {}

# list of event source mappings for the API

self.event_source_mappings = []

class ClientError(Exception):

def __init__(self, msg, code=400):

super(ClientError, self).__init__(msg)

self.code = code

self.msg = msg

def get_response(self):

if isinstance(self.msg, Response):

return self.msg

return error_response(self.msg, self.code)

class LambdaContext(object):

DEFAULT_MEMORY_LIMIT = 1536

def __init__(self, func_details, qualifier=None, context=None):

self.function_name = func_details.name()

self.function_version = func_details.get_qualifier_version(qualifier)

self.client_context = context.get('client_context')

self.invoked_function_arn = func_details.arn()

if qualifier:

self.invoked_function_arn += ':' + qualifier

self.cognito_identity = context.get('identity')

self.aws_request_id = str(uuid.uuid4())

self.memory_limit_in_mb = func_details.memory_size or self.DEFAULT_MEMORY_LIMIT

self.log_group_name = '/aws/lambda/%s' % self.function_name

self.log_stream_name = '%s/[1]%s' % (timestamp(format='%Y/%m/%d'), short_uid())

def get_remaining_time_in_millis(self):

# TODO implement!

return 1000 * 60

def cleanup():

region = LambdaRegion.get()

region.lambdas = {}

region.event_source_mappings = []

LAMBDA_EXECUTOR.cleanup()

def func_arn(function_name):

return aws_stack.lambda_function_arn(function_name)

def func_qualifier(function_name, qualifier=None):

region = LambdaRegion.get()

arn = aws_stack.lambda_function_arn(function_name)

details = region.lambdas.get(arn)

if not details:

return details

if details.qualifier_exists(qualifier):

return '{}:{}'.format(arn, qualifier)

return arn

def check_batch_size_range(source_arn, batch_size=None):

source = source_arn.split(':')[2].lower()

source = 'kafka' if 'secretsmanager' in source else source

batch_size_entry = BATCH_SIZE_RANGES.get(source)

if not batch_size_entry:

raise ValueError(

INVALID_PARAMETER_VALUE_EXCEPTION, 'Unsupported event source type'

)

batch_size = batch_size or batch_size_entry[0]

if batch_size > batch_size_entry[1]:

raise ValueError(

INVALID_PARAMETER_VALUE_EXCEPTION,

'BatchSize {} exceeds the max of {}'.format(batch_size, batch_size_entry[1])

)

return batch_size

def add_function_mapping(lambda_name, lambda_handler, lambda_cwd=None):

region = LambdaRegion.get()

arn = func_arn(lambda_name)

lambda_details = region.lambdas[arn]

lambda_details.versions.get(VERSION_LATEST)['Function'] = lambda_handler

lambda_details.cwd = lambda_cwd or lambda_details.cwd

def build_mapping_obj(data):

mapping = {}

function_name = data['FunctionName']

enabled = data.get('Enabled', True)

batch_size = data.get('BatchSize')

mapping['UUID'] = str(uuid.uuid4())

mapping['FunctionArn'] = func_arn(function_name)

mapping['LastProcessingResult'] = 'OK'

mapping['StateTransitionReason'] = 'User action'

mapping['LastModified'] = float(time.mktime(datetime.utcnow().timetuple()))

mapping['State'] = 'Enabled' if enabled in [True, None] else 'Disabled'

if 'SelfManagedEventSource' in data:

source_arn = data['SourceAccessConfigurations'][0]['URI']

mapping['SelfManagedEventSource'] = data['SelfManagedEventSource']

mapping['Topics'] = data['Topics']

mapping['SourceAccessConfigurations'] = data['SourceAccessConfigurations']

else:

source_arn = data['EventSourceArn']

mapping['EventSourceArn'] = source_arn

mapping['StartingPosition'] = LAMBDA_DEFAULT_STARTING_POSITION

batch_size = check_batch_size_range(source_arn, batch_size)

mapping['BatchSize'] = batch_size

return mapping

def add_event_source(data):

region = LambdaRegion.get()

mapping = build_mapping_obj(data)

region.event_source_mappings.append(mapping)

return mapping

def update_event_source(uuid_value, data):

region = LambdaRegion.get()

function_name = data.get('FunctionName') or ''

batch_size = None

enabled = data.get('Enabled', True)

for mapping in region.event_source_mappings:

if uuid_value == mapping['UUID']:

if function_name:

mapping['FunctionArn'] = func_arn(function_name)

batch_size = data.get('BatchSize')

if 'SelfManagedEventSource' in mapping:

batch_size = check_batch_size_range(

mapping['SourceAccessConfigurations'][0]['URI'],

batch_size or mapping['BatchSize'])

else:

batch_size = check_batch_size_range(mapping['EventSourceArn'], batch_size or mapping['BatchSize'])

mapping['State'] = 'Enabled' if enabled in [True, None] else 'Disabled'

mapping['LastModified'] = float(time.mktime(datetime.utcnow().timetuple()))

mapping['BatchSize'] = batch_size

if 'SourceAccessConfigurations' in (mapping and data):

mapping['SourceAccessConfigurations'] = data['SourceAccessConfigurations']

return mapping

return {}

def delete_event_source(uuid_value):

region = LambdaRegion.get()

for i, m in enumerate(region.event_source_mappings):

if uuid_value == m['UUID']:

return region.event_source_mappings.pop(i)

return {}

@synchronized(lock=EXEC_MUTEX)

def use_docker():

global DO_USE_DOCKER

if DO_USE_DOCKER is None:

DO_USE_DOCKER = False

if 'docker' in config.LAMBDA_EXECUTOR:

try:

run('docker images', print_error=False)

DO_USE_DOCKER = True

except Exception:

pass

return DO_USE_DOCKER

def fix_proxy_path_params(path_params):

proxy_path_param_value = path_params.get('proxy+')

if not proxy_path_param_value:

return

del path_params['proxy+']

path_params['proxy'] = proxy_path_param_value

def message_attributes_to_lower(message_attrs):

""" Convert message attribute details (first characters) to lower case (e.g., stringValue, dataType). """

message_attrs = message_attrs or {}

for _, attr in message_attrs.items():

if not isinstance(attr, dict):

continue

for key, value in dict(attr).items():

attr[first_char_to_lower(key)] = attr.pop(key)

return message_attrs

def process_apigateway_invocation(func_arn, path, payload, stage, api_id, headers={},

resource_path=None, method=None, path_params={}, query_string_params=None,

stage_variables={}, request_context={}, event_context={}):

try:

resource_path = resource_path or path

event = construct_invocation_event(method, resource_path, headers, payload, query_string_params)

path_params = dict(path_params)

fix_proxy_path_params(path_params)

event['pathParameters'] = path_params

event['resource'] = resource_path

event['requestContext'] = request_context

event['stageVariables'] = stage_variables

LOG.debug('Running Lambda function %s from API Gateway invocation: %s %s' % (func_arn, method or 'GET', path))

asynchronous = not config.SYNCHRONOUS_API_GATEWAY_EVENTS

inv_result = run_lambda(func_arn=func_arn, event=event, context=event_context, asynchronous=asynchronous)

return inv_result.result

except Exception as e:

LOG.warning('Unable to run Lambda function on API Gateway message: %s %s' % (e, traceback.format_exc()))

def construct_invocation_event(method, path, headers, data, query_string_params={}):

query_string_params = query_string_params or parse_request_data(method, path, '')

event = {

'path': path,

'headers': dict(headers),

'multiValueHeaders': multi_value_dict_for_list(headers),

'body': data,

'isBase64Encoded': False,

'httpMethod': method,

'queryStringParameters': query_string_params,

'multiValueQueryStringParameters': multi_value_dict_for_list(query_string_params)

}

return event

def process_sns_notification(func_arn, topic_arn, subscription_arn, message, message_id,

message_attributes, unsubscribe_url, subject='',):

event = {

'Records': [{

'EventSource': 'localstack:sns',

'EventVersion': '1.0',

'EventSubscriptionArn': subscription_arn,

'Sns': {

'Type': 'Notification',

'MessageId': message_id,

'TopicArn': topic_arn,

'Subject': subject,

'Message': message,

'Timestamp': timestamp_millis(),

'SignatureVersion': '1',

# TODO Add a more sophisticated solution with an actual signature

# Hardcoded

'Signature': 'EXAMPLEpH+..',

'SigningCertUrl': 'https://sns.us-east-1.amazonaws.com/SimpleNotificationService-000000000.pem',

'UnsubscribeUrl': unsubscribe_url,

'MessageAttributes': message_attributes

}

}]

}

inv_result = run_lambda(func_arn=func_arn, event=event, context={}, asynchronous=not config.SYNCHRONOUS_SNS_EVENTS)

return inv_result.result

def process_kinesis_records(records, stream_name):

def chunks(lst, n):

# Yield successive n-sized chunks from lst.

for i in range(0, len(lst), n):

yield lst[i:i + n]

# feed records into listening lambdas

try:

stream_arn = aws_stack.kinesis_stream_arn(stream_name)

sources = get_event_sources(source_arn=stream_arn)

for source in sources:

arn = source['FunctionArn']

for chunk in chunks(records, source['BatchSize']):

event = {

'Records': [

{

'eventID': 'shardId-000000000000:{0}'.format(rec['sequenceNumber']),

'eventSourceARN': stream_arn,

'eventSource': 'aws:kinesis',

'eventVersion': '1.0',

'eventName': 'aws:kinesis:record',

'invokeIdentityArn': 'arn:aws:iam::{0}:role/lambda-role'.format(TEST_AWS_ACCOUNT_ID),

'awsRegion': aws_stack.get_region(),

'kinesis': rec

}

for rec in chunk

]

}

run_lambda(func_arn=arn, event=event, context={}, asynchronous=not config.SYNCHRONOUS_KINESIS_EVENTS)

except Exception as e:

LOG.warning('Unable to run Lambda function on Kinesis records: %s %s' % (e, traceback.format_exc()))

def start_lambda_sqs_listener():

if SQS_LISTENER_THREAD:

return

def send_event_to_lambda(queue_arn, queue_url, lambda_arn, messages, region):

def delete_messages(result, func_arn, event, error=None, dlq_sent=None, **kwargs):

if error and not dlq_sent:

# Skip deleting messages from the queue in case of processing errors AND if

# the message has not yet been sent to a dead letter queue (DLQ).

# We'll pick them up and retry next time they become available on the queue.

return

sqs_client = aws_stack.connect_to_service('sqs')

entries = [{'Id': r['receiptHandle'], 'ReceiptHandle': r['receiptHandle']} for r in records]

sqs_client.delete_message_batch(QueueUrl=queue_url, Entries=entries)

records = []

for msg in messages:

message_attrs = message_attributes_to_lower(msg.get('MessageAttributes'))

records.append({

'body': msg['Body'],

'receiptHandle': msg['ReceiptHandle'],

'md5OfBody': msg['MD5OfBody'],

'eventSourceARN': queue_arn,

'eventSource': lambda_executors.EVENT_SOURCE_SQS,

'awsRegion': region,

'messageId': msg['MessageId'],

'attributes': msg.get('Attributes', {}),

'messageAttributes': message_attrs,

'md5OfMessageAttributes': msg.get('MD5OfMessageAttributes'),

'sqs': True,

})

event = {'Records': records}

# TODO implement retries, based on "RedrivePolicy.maxReceiveCount" in the queue settings

run_lambda(func_arn=lambda_arn, event=event, context={}, asynchronous=True, callback=delete_messages)

def listener_loop(*args):

while True:

try:

sources = get_event_sources(source_arn=r'.*:sqs:.*')

if not sources:

# Temporarily disable polling if no event sources are configured

# anymore. The loop will get restarted next time a message

# arrives and if an event source is configured.

SQS_LISTENER_THREAD.pop('_thread_')

return

sqs_client = aws_stack.connect_to_service('sqs')

for source in sources:

queue_arn = source['EventSourceArn']

lambda_arn = source['FunctionArn']

batch_size = max(min(source.get('BatchSize', 1), 10), 1)

try:

region_name = queue_arn.split(':')[3]

queue_url = aws_stack.sqs_queue_url_for_arn(queue_arn)

result = sqs_client.receive_message(

QueueUrl=queue_url,

MessageAttributeNames=['All'],

MaxNumberOfMessages=batch_size

)

messages = result.get('Messages')

if not messages:

continue

send_event_to_lambda(queue_arn, queue_url, lambda_arn, messages, region=region_name)

except Exception as e:

LOG.debug('Unable to poll SQS messages for queue %s: %s' % (queue_arn, e))

except Exception:

pass

finally:

time.sleep(SQS_POLL_INTERVAL_SEC)

LOG.debug('Starting SQS message polling thread for Lambda API')

SQS_LISTENER_THREAD['_thread_'] = FuncThread(listener_loop)

SQS_LISTENER_THREAD['_thread_'].start()

def process_sqs_message(queue_name, region_name=None):

# feed message into the first listening lambda (message should only get processed once)

try:

region_name = region_name or aws_stack.get_region()

queue_arn = aws_stack.sqs_queue_arn(queue_name, region_name=region_name)

sources = get_event_sources(source_arn=queue_arn)

arns = [s.get('FunctionArn') for s in sources]

source = (sources or [None])[0]

if not source:

return False

LOG.debug('Found %s source mappings for event from SQS queue %s: %s' % (len(arns), queue_arn, arns))

start_lambda_sqs_listener()

return True

except Exception as e:

LOG.warning('Unable to run Lambda function on SQS messages: %s %s' % (e, traceback.format_exc()))

def get_event_sources(func_name=None, source_arn=None):

region = LambdaRegion.get()

result = []

for m in region.event_source_mappings:

if not func_name or (m['FunctionArn'] in [func_name, func_arn(func_name)]):

if _arn_match(mapped=m['EventSourceArn'], searched=source_arn):

result.append(m)

return result

def _arn_match(mapped, searched):

if not searched or mapped == searched:

return True

# Some types of ARNs can end with a path separated by slashes, for

# example the ARN of a DynamoDB stream is tableARN/stream/ID. It's

# a little counterintuitive that a more specific mapped ARN can

# match a less specific ARN on the event, but some integration tests

# rely on it for things like subscribing to a stream and matching an

# event labeled with the table ARN.

if re.match(r'^%s$' % searched, mapped):

return True

if mapped.startswith(searched):

suffix = mapped[len(searched):]

return suffix[0] == '/'

return False

def get_function_version(arn, version):

region = LambdaRegion.get()

func = region.lambdas.get(arn)

return format_func_details(func, version=version, always_add_version=True)

def publish_new_function_version(arn):

region = LambdaRegion.get()

func_details = region.lambdas.get(arn)

versions = func_details.versions

max_version_number = func_details.max_version()

next_version_number = max_version_number + 1

latest_hash = versions.get(VERSION_LATEST).get('CodeSha256')

max_version = versions.get(str(max_version_number))

max_version_hash = max_version.get('CodeSha256') if max_version else ''

if latest_hash != max_version_hash:

versions[str(next_version_number)] = {

'CodeSize': versions.get(VERSION_LATEST).get('CodeSize'),

'CodeSha256': versions.get(VERSION_LATEST).get('CodeSha256'),

'Function': versions.get(VERSION_LATEST).get('Function'),

'RevisionId': str(uuid.uuid4())

}

max_version_number = next_version_number

return get_function_version(arn, str(max_version_number))

def do_list_versions(arn):

region = LambdaRegion.get()

versions = [get_function_version(arn, version) for version in region.lambdas.get(arn).versions.keys()]

return sorted(versions, key=lambda k: str(k.get('Version')))

def do_update_alias(arn, alias, version, description=None):

region = LambdaRegion.get()

new_alias = {

'AliasArn': arn + ':' + alias,

'FunctionVersion': version,

'Name': alias,

'Description': description or '',

'RevisionId': str(uuid.uuid4())

}

region.lambdas.get(arn).aliases[alias] = new_alias

return new_alias

def run_lambda(func_arn, event, context={}, version=None,

suppress_output=False, asynchronous=False, callback=None):

region_name = func_arn.split(':')[3]

region = LambdaRegion.get(region_name)

if suppress_output:

stdout_ = sys.stdout

stderr_ = sys.stderr

stream = StringIO()

sys.stdout = stream

sys.stderr = stream

try:

func_arn = aws_stack.fix_arn(func_arn)

func_details = region.lambdas.get(func_arn)

if not func_details:

result = not_found_error(msg='The resource specified in the request does not exist.')

return lambda_executors.InvocationResult(result)

# forward invocation to external endpoint, if configured

invocation_type = 'Event' if asynchronous else 'RequestResponse'

invoke_result = forward_to_external_url(func_details, event, context, invocation_type)

if invoke_result is not None:

return invoke_result

context = LambdaContext(func_details, version, context)

result = LAMBDA_EXECUTOR.execute(func_arn, func_details, event, context=context,

version=version, asynchronous=asynchronous, callback=callback)

except Exception as e:

exc_type, exc_value, exc_traceback = sys.exc_info()

response = {

'errorType': str(exc_type.__name__),

'errorMessage': str(e),

'stackTrace': traceback.format_tb(exc_traceback)

}

LOG.info('Error executing Lambda function %s: %s %s' % (func_arn, e, traceback.format_exc()))

log_output = e.log_output if isinstance(e, lambda_executors.InvocationException) else ''

return lambda_executors.InvocationResult(Response(json.dumps(response), status=500), log_output)

finally:

if suppress_output:

sys.stdout = stdout_

sys.stderr = stderr_

return result

def exec_lambda_code(script, handler_function='handler', lambda_cwd=None, lambda_env=None):

if lambda_cwd or lambda_env:

EXEC_MUTEX.acquire()

if lambda_cwd:

previous_cwd = os.getcwd()

os.chdir(lambda_cwd)

sys.path = [lambda_cwd] + sys.path

if lambda_env:

previous_env = dict(os.environ)

os.environ.update(lambda_env)

# generate lambda file name

lambda_id = 'l_%s' % short_uid()

lambda_file = LAMBDA_SCRIPT_PATTERN.replace('*', lambda_id)

save_file(lambda_file, script)

# delete temporary .py and .pyc files on exit

TMP_FILES.append(lambda_file)

TMP_FILES.append('%sc' % lambda_file)

try:

pre_sys_modules_keys = set(sys.modules.keys())

try:

handler_module = imp.load_source(lambda_id, lambda_file)

module_vars = handler_module.__dict__

finally:

# the above import can bring files for the function

# (eg settings.py) into the global namespace. subsequent

# calls can pick up file from another function, causing

# general issues.

post_sys_modules_keys = set(sys.modules.keys())

for key in post_sys_modules_keys:

if key not in pre_sys_modules_keys:

sys.modules.pop(key)

except Exception as e:

LOG.error('Unable to exec: %s %s' % (script, traceback.format_exc()))

raise e

finally:

if lambda_cwd or lambda_env:

if lambda_cwd:

os.chdir(previous_cwd)

sys.path.pop(0)

if lambda_env:

os.environ = previous_env

EXEC_MUTEX.release()

return module_vars[handler_function]

def get_handler_function_from_name(handler_name, runtime=LAMBDA_DEFAULT_RUNTIME):

if runtime.startswith(tuple(DOTNET_LAMBDA_RUNTIMES)):

return handler_name.split(':')[-1]

return handler_name.split('.')[-1]

def error_response(msg, code=500, error_type='InternalFailure'):

LOG.info(msg)

return aws_responses.flask_error_response_json(msg, code=code, error_type=error_type)

def get_zip_bytes(function_code):

"""Returns the ZIP file contents from a FunctionCode dict.

:type function_code: dict

:param function_code: https://docs.aws.amazon.com/lambda/latest/dg/API_FunctionCode.html

:returns: bytes of the Zip file.

"""

function_code = function_code or {}

if 'S3Bucket' in function_code:

s3_client = aws_stack.connect_to_service('s3')

bytes_io = BytesIO()

try:

s3_client.download_fileobj(function_code['S3Bucket'], function_code['S3Key'], bytes_io)

zip_file_content = bytes_io.getvalue()

except Exception as e:

raise ClientError('Unable to fetch Lambda archive from S3: %s' % e, 404)

elif 'ZipFile' in function_code:

zip_file_content = function_code['ZipFile']

zip_file_content = base64.b64decode(zip_file_content)

elif 'ImageUri' in function_code:

zip_file_content = None

else:

raise ClientError('No valid Lambda archive specified: %s' % list(function_code.keys()))

return zip_file_content

def get_java_handler(zip_file_content, main_file, func_details=None):

"""Creates a Java handler from an uploaded ZIP or JAR.

:type zip_file_content: bytes

:param zip_file_content: ZIP file bytes.

:type handler: str

:param handler: The lambda handler path.

:type main_file: str

:param main_file: Filepath to the uploaded ZIP or JAR file.

:returns: function or flask.Response

"""

if is_zip_file(zip_file_content):

def execute(event, context):

result = lambda_executors.EXECUTOR_LOCAL.execute_java_lambda(

event, context, main_file=main_file, func_details=func_details)

return result

return execute

raise ClientError(error_response(

'Unable to extract Java Lambda handler - file is not a valid zip/jar file', 400, error_type='ValidationError'))

def set_archive_code(code, lambda_name, zip_file_content=None):

region = LambdaRegion.get()

# get metadata

lambda_arn = func_arn(lambda_name)

lambda_details = region.lambdas[lambda_arn]

is_local_mount = code.get('S3Bucket') == BUCKET_MARKER_LOCAL

if is_local_mount and config.LAMBDA_REMOTE_DOCKER:

msg = 'Please note that Lambda mounts (bucket name "%s") cannot be used with LAMBDA_REMOTE_DOCKER=1'

raise Exception(msg % BUCKET_MARKER_LOCAL)

# Stop/remove any containers that this arn uses.

LAMBDA_EXECUTOR.cleanup(lambda_arn)

if is_local_mount:

# Mount or use a local folder lambda executors can reference

# WARNING: this means we're pointing lambda_cwd to a local path in the user's

# file system! We must ensure that there is no data loss (i.e., we must *not* add

# this folder to TMP_FILES or similar).

lambda_details.cwd = code.get('S3Key')

return code['S3Key']

# get file content

zip_file_content = zip_file_content or get_zip_bytes(code)

if not zip_file_content:

return

# Save the zip file to a temporary file that the lambda executors can reference

code_sha_256 = base64.standard_b64encode(hashlib.sha256(zip_file_content).digest())

latest_version = lambda_details.get_version(VERSION_LATEST)

latest_version['CodeSize'] = len(zip_file_content)

latest_version['CodeSha256'] = code_sha_256.decode('utf-8')

tmp_dir = '%s/zipfile.%s' % (config.TMP_FOLDER, short_uid())

mkdir(tmp_dir)

tmp_file = '%s/%s' % (tmp_dir, LAMBDA_ZIP_FILE_NAME)

save_file(tmp_file, zip_file_content)

TMP_FILES.append(tmp_dir)

lambda_details.cwd = tmp_dir

return tmp_dir

def set_function_code(code, lambda_name, lambda_cwd=None):

def _set_and_configure():

lambda_handler = do_set_function_code(code, lambda_name, lambda_cwd=lambda_cwd)

add_function_mapping(lambda_name, lambda_handler, lambda_cwd)

# unzipping can take some time - limit the execution time to avoid client/network timeout issues

run_for_max_seconds(25, _set_and_configure)

return {'FunctionName': lambda_name}

def do_set_function_code(code, lambda_name, lambda_cwd=None):

def generic_handler(event, context):

raise ClientError(('Unable to find executor for Lambda function "%s". Note that ' +

'Node.js, Golang, and .Net Core Lambdas currently require LAMBDA_EXECUTOR=docker') % lambda_name)

region = LambdaRegion.get()

arn = func_arn(lambda_name)

lambda_details = region.lambdas[arn]

runtime = lambda_details.runtime

lambda_environment = lambda_details.envvars

handler_name = lambda_details.handler = lambda_details.handler or LAMBDA_DEFAULT_HANDLER

code_passed = code

code = code or lambda_details.code

is_local_mount = code.get('S3Bucket') == BUCKET_MARKER_LOCAL

zip_file_content = None

if code_passed:

lambda_cwd = lambda_cwd or set_archive_code(code_passed, lambda_name)

if not is_local_mount:

# Save the zip file to a temporary file that the lambda executors can reference

zip_file_content = get_zip_bytes(code_passed)

else:

lambda_cwd = lambda_cwd or lambda_details.cwd

if not lambda_cwd:

return

# get local lambda working directory

tmp_file = os.path.join(lambda_cwd, LAMBDA_ZIP_FILE_NAME)

if not zip_file_content:

zip_file_content = load_file(tmp_file, mode='rb')

# Set the appropriate lambda handler.

lambda_handler = generic_handler

is_java = lambda_executors.is_java_lambda(runtime)

if is_java:

# The Lambda executors for Docker subclass LambdaExecutorContainers, which

# runs Lambda in Docker by passing all *.jar files in the function working

# directory as part of the classpath. Obtain a Java handler function below.

lambda_handler = get_java_handler(zip_file_content, tmp_file, func_details=lambda_details)

if not is_local_mount:

# Lambda code must be uploaded in Zip format

if not is_zip_file(zip_file_content):

raise ClientError(

'Uploaded Lambda code for runtime ({}) is not in Zip format'.format(runtime))

# Unzip the Lambda archive contents

unzip(tmp_file, lambda_cwd)

# Obtain handler details for any non-Java Lambda function

if not is_java:

handler_file = get_handler_file_from_name(handler_name, runtime=runtime)

handler_function = get_handler_function_from_name(handler_name, runtime=runtime)

main_file = '%s/%s' % (lambda_cwd, handler_file)

if CHECK_HANDLER_ON_CREATION and not os.path.exists(main_file):

# Raise an error if (1) this is not a local mount lambda, or (2) we're

# running Lambdas locally (not in Docker), or (3) we're using remote Docker.

# -> We do *not* want to raise an error if we're using local mount in non-remote Docker

if not is_local_mount or not use_docker() or config.LAMBDA_REMOTE_DOCKER:

file_list = run('cd "%s"; du -d 3 .' % lambda_cwd)

config_debug = ('Config for local mount, docker, remote: "%s", "%s", "%s"' %

(is_local_mount, use_docker(), config.LAMBDA_REMOTE_DOCKER))

LOG.debug('Lambda archive content:\n%s' % file_list)

raise ClientError(error_response(

'Unable to find handler script (%s) in Lambda archive. %s' % (main_file, config_debug),

400, error_type='ValidationError'))

if runtime.startswith('python') and not use_docker():

try:

# make sure the file is actually readable, then read contents

ensure_readable(main_file)

zip_file_content = load_file(main_file, mode='rb')

# extract handler

lambda_handler = exec_lambda_code(

zip_file_content,

handler_function=handler_function,

lambda_cwd=lambda_cwd,

lambda_env=lambda_environment)

except Exception as e:

raise ClientError('Unable to get handler function from lambda code.', e)

return lambda_handler

def do_list_functions():

funcs = []

region = LambdaRegion.get()

this_region = aws_stack.get_region()

for f_arn, func in region.lambdas.items():

if type(func) != LambdaFunction:

continue

# filter out functions of current region

func_region = f_arn.split(':')[3]

if func_region != this_region:

continue

func_name = f_arn.split(':function:')[-1]

arn = func_arn(func_name)

func_details = region.lambdas.get(arn)

if not func_details:

# this can happen if we're accessing Lambdas from a different region (ARN mismatch)

continue

details = format_func_details(func_details)

details['Tags'] = func.tags

funcs.append(details)

return funcs

def format_func_details(func_details, version=None, always_add_version=False):

version = version or VERSION_LATEST

func_version = func_details.get_version(version)

result = {

'CodeSha256': func_version.get('CodeSha256'),

'Role': func_details.role,

'KMSKeyArn': func_details.kms_key_arn,

'Version': version,

'VpcConfig': func_details.vpc_config,

'FunctionArn': func_details.arn(),

'FunctionName': func_details.name(),

'CodeSize': func_version.get('CodeSize'),

'Handler': func_details.handler,

'Runtime': func_details.runtime,

'Timeout': func_details.timeout,

'Description': func_details.description,

'MemorySize': func_details.memory_size,

'LastModified': isoformat_milliseconds(func_details.last_modified) + '+0000',

'TracingConfig': {'Mode': 'PassThrough'},

'RevisionId': func_version.get('RevisionId'),

'State': 'Active',

'LastUpdateStatus': 'Successful',

'PackageType': func_details.package_type,

'ImageConfig': func_details.image_config

}

if func_details.dead_letter_config:

result['DeadLetterConfig'] = func_details.dead_letter_config

if func_details.envvars:

result['Environment'] = {

'Variables': func_details.envvars

}

if (always_add_version or version != VERSION_LATEST) and len(result['FunctionArn'].split(':')) <= 7:

result['FunctionArn'] += ':%s' % version

return result

def forward_to_external_url(func_details, event, context, invocation_type):

""" If LAMBDA_FORWARD_URL is configured, forward the invocation of this Lambda to the configured URL. """

if not config.LAMBDA_FORWARD_URL:

return

func_name = func_details.name()

url = '%s%s/functions/%s/invocations' % (config.LAMBDA_FORWARD_URL, PATH_ROOT, func_name)

headers = aws_stack.mock_aws_request_headers('lambda')

headers['X-Amz-Invocation-Type'] = invocation_type

headers['X-Amz-Log-Type'] = 'Tail'

client_context = context.get('client_context')

if client_context:

headers['X-Amz-Client-Context'] = client_context

data = json.dumps(event) if isinstance(event, dict) else str(event)

LOG.debug('Forwarding Lambda invocation to LAMBDA_FORWARD_URL: %s' % config.LAMBDA_FORWARD_URL)

result = safe_requests.post(url, data, headers=headers)

content = run_safe(lambda: to_str(result.content)) or result.content

LOG.debug('Received result from external Lambda endpoint (status %s): %s' % (result.status_code, content))

result = aws_responses.requests_to_flask_response(result)

result = lambda_executors.InvocationResult(result)

return result

def forward_to_fallback_url(func_arn, data):

""" If LAMBDA_FALLBACK_URL is configured, forward the invocation of this non-existing

Lambda to the configured URL. """

if not config.LAMBDA_FALLBACK_URL:

return

lambda_name = aws_stack.lambda_function_name(func_arn)

if config.LAMBDA_FALLBACK_URL.startswith('dynamodb://'):

table_name = urlparse(config.LAMBDA_FALLBACK_URL.replace('dynamodb://', 'http://')).netloc

dynamodb = aws_stack.connect_to_service('dynamodb')

item = {

'id': {'S': short_uid()},

'timestamp': {'N': str(now_utc())},

'payload': {'S': data},

'function_name': {'S': lambda_name}

}

aws_stack.create_dynamodb_table(table_name, partition_key='id')

dynamodb.put_item(TableName=table_name, Item=item)

return ''

if re.match(r'^https?://.+', config.LAMBDA_FALLBACK_URL):

headers = {

'lambda-function-name': lambda_name,

'Content-Type': APPLICATION_JSON

}

response = safe_requests.post(config.LAMBDA_FALLBACK_URL, data, headers=headers)

content = response.content

try:

# parse the response into a dictionary to get details

# like function error etc.

content = json.loads(content)

except Exception:

pass

return content

raise ClientError('Unexpected value for LAMBDA_FALLBACK_URL: %s' % config.LAMBDA_FALLBACK_URL)

def get_lambda_policy(function, qualifier=None):

iam_client = aws_stack.connect_to_service('iam')

policies = iam_client.list_policies(Scope='Local', MaxItems=500)['Policies']

docs = []

for p in policies:

# !TODO: Cache policy documents instead of running N+1 API calls here!

versions = iam_client.list_policy_versions(PolicyArn=p['Arn'])['Versions']

default_version = [v for v in versions if v.get('IsDefaultVersion')]

versions = default_version or versions

doc = versions[0]['Document']

doc = doc if isinstance(doc, dict) else json.loads(doc)

if not isinstance(doc['Statement'], list):

doc['Statement'] = [doc['Statement']]

for stmt in doc['Statement']:

stmt['Principal'] = stmt.get('Principal') or {'AWS': TEST_AWS_ACCOUNT_ID}

doc['PolicyArn'] = p['Arn']

doc['Id'] = 'default'

docs.append(doc)

res_qualifier = func_qualifier(function, qualifier)

policy = [d for d in docs if d['Statement'][0]['Resource'] == res_qualifier]

return (policy or [None])[0]

def not_found_error(ref=None, msg=None):

if not msg:

msg = 'The resource you requested does not exist.'

if ref:

msg = '%s not found: %s' % ('Function' if ':function:' in ref else 'Resource', ref)

return error_response(msg, 404, error_type='ResourceNotFoundException')