Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
'Status': status,
'Reason': 'See the details in CloudWatch Log Stream',
'PhysicalResourceId': event['ServiceToken'],
'StackId': event['StackId'],
'RequestId': event['RequestId'],
'LogicalResourceId': event['LogicalResourceId'],
'Data': data
}
print('Response = ' + json.dumps(responseBody))
print(event)
lambda_client = get_client('lambda')
function_name = os.environ['AWS_LAMBDA_FUNCTION_NAME']
print('Deleting Lambda')
lambda_client.delete_function(FunctionName=function_name)
requests.put(event['ResponseURL'], data=json.dumps(responseBody))
def respond_cloudformation(event, status, data=None):
responseBody = {
'Status': status,
'Reason': 'See the details in CloudWatch Log Stream',
'PhysicalResourceId': event['ServiceToken'],
'StackId': event['StackId'],
'RequestId': event['RequestId'],
'LogicalResourceId': event['LogicalResourceId'],
'Data': data
}
print('Response = ' + json.dumps(responseBody))
print(event)
requests.put(event['ResponseURL'], data=json.dumps(responseBody))
'StackId': event['StackId'],
'RequestId': event['RequestId'],
'LogicalResourceId': event['LogicalResourceId'],
'NoEcho': noEcho,
'Data': responseData,
})
log.info("| response body:\n" + body)
headers = {
'content-type' : '',
'content-length' : str(len(body))
}
try:
from botocore.vendored import requests
response = requests.put(responseUrl, data=body, headers=headers)
log.info("| status code: " + response.reason)
response.raise_for_status()
except Exception as e:
log.error("| unable to send response to CloudFormation")
raise e
responseBody['PhysicalResourceId'] = physicalResourceId or context.log_stream_name
responseBody['StackId'] = event['StackId']
responseBody['RequestId'] = event['RequestId']
responseBody['LogicalResourceId'] = event['LogicalResourceId']
responseBody['NoEcho'] = noEcho
responseBody['Data'] = responseData
json_responseBody = json.dumps(responseBody)
headers = {
'content-type' : '',
'content-length' : str(len(json_responseBody))
}
try:
response = requests.put(responseUrl,
data=json_responseBody,
headers=headers)
except Exception as e:
error = e
if (response_status == Status.FAILED) and 'result' in response_data:
reason = "%s %s" % (response_data['result'], reason)
body = {
"Status": response_status,
"Reason": reason,
"PhysicalResourceId": physical_resource_id,
"StackId": event['StackId'],
"RequestId": event['RequestId'],
"LogicalResourceId": event['LogicalResourceId'],
"Data": response_data
}
response_body = json.dumps(body)
logger.debug("Response body: %s", response_body)
try:
response = requests.put(event['ResponseURL'],
data=response_body)
logger.debug("Status code: %s" % response.status_code)
# logger.debug("Status message: %s" % response.status_message)
# how do we get the status message?
except Exception as e:
logger.error("send(..) failed executing https.request(..): %s" %
e.message)
responseBody['StackId'] = event['StackId']
responseBody['RequestId'] = event['RequestId']
responseBody['LogicalResourceId'] = event['LogicalResourceId']
responseBody['NoEcho'] = noEcho
responseBody['Data'] = responseData
body = json.dumps(responseBody)
logger.info("| response body:\n" + body)
headers = {
'content-type' : '',
'content-length' : str(len(body))
}
try:
response = requests.put(responseUrl, data=body, headers=headers)
logger.info("| status code: " + response.reason)
except Exception as e:
logger.error("| unable to send response to CloudFormation")
logger.exception(e)
responseBody['RequestId'] = event['RequestId']
responseBody['LogicalResourceId'] = event['LogicalResourceId']
if responseData and responseData != {} and responseData != [] and isinstance(responseData, dict):
responseBody['Data'] = responseData
json_responseBody = json.dumps(responseBody)
logger.debug("Response body:\n" + json_responseBody)
headers = {
'content-type': '',
'content-length': str(len(json_responseBody))
}
try:
response = requests.put(responseUrl,
data=json_responseBody,
headers=headers)
logger.info("CloudFormation returned status code: " + response.reason)
except Exception as e:
logger.error("send(..) failed executing requests.put(..): " + str(e))
raise
'RequestId': event['RequestId'],
'LogicalResourceId': event['LogicalResourceId'],
'Data': responseData
}
json_responseBody = json.dumps(responseBody)
print("Response body:\n" + json_responseBody)
headers = {
'content-type': '',
'content-length': str(len(json_responseBody))
}
try:
response = requests.put(responseUrl,
data=json_responseBody,
headers=headers)
print("Status code: " + response.reason)
except Exception as e:
print("send(..) failed executing requests.put(..): " + str(e))
return