Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return None
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': str(data)}
}
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):
response = safe_requests.post(config.LAMBDA_FALLBACK_URL, data)
return response.content
raise ClientError('Unexpected value for LAMBDA_FALLBACK_URL: %s' % config.LAMBDA_FALLBACK_URL)
with zipfile.ZipFile(BytesIO(zip_file_content)) as zip_ref:
# TODO: check if this is still needed (probably not)
jar_entries = [e for e in zip_ref.infolist() if e.filename.endswith('.jar')]
if len(jar_entries) == 1:
zip_file_content = zip_ref.read(jar_entries[0].filename)
LOG.info('Found single jar file %s with %s bytes in Lambda zip archive' %
(jar_entries[0].filename, len(zip_file_content)))
main_file = new_tmp_file()
save_file(main_file, zip_file_content)
if is_zip_file(zip_file_content):
def execute(event, context):
result, log_output = lambda_executors.EXECUTOR_LOCAL.execute_java_lambda(
event, context, handler=handler, main_file=main_file)
return result
return execute, zip_file_content
raise ClientError(error_response(
'Unable to extract Java Lambda handler - file is not a valid zip/jar file', 400, error_type='ValidationError'))
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.
"""
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)
else:
raise ClientError('No valid Lambda archive specified.')
return zip_file_content
:param function_code: https://docs.aws.amazon.com/lambda/latest/dg/API_FunctionCode.html
:returns: bytes of the Zip file.
"""
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)
else:
raise ClientError('No valid Lambda archive specified.')
return zip_file_content
'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)
add_function_mapping(lambda_name, lambda_handler, lambda_cwd)
return {'FunctionName': lambda_name}