crhelper 2.0.12

crhelper simplifies authoring CloudFormation Custom Resources

Custom Resource Helper

Simplify best practice Custom Resource creation, sending responses to CloudFormation and providing exception, timeout trapping, and detailed configurable logging.

Features

• Dead simple to use, reduces the complexity of writing a CloudFormation custom resource
• Guarantees that CloudFormation will get a response even if an exception is raised
• Returns meaningful errors to CloudFormation Stack events in the case of a failure
• Polling enables run times longer than the lambda 15 minute limit
• JSON logging that includes request id's, stack id's and request type to assist in tracing logs relevant to a particular CloudFormation event
• Catches function timeouts and sends CloudFormation a failure response
• Static typing (mypy) compatible

Installation

Install into the root folder of your lambda function

cd my-lambda-function/
pip install crhelper -t .

Example Usage

This blog covers usage in more detail.

from __future__ import print_function
from crhelper import CfnResource
import logging

logger = logging.getLogger(__name__)
# Initialise the helper, all inputs are optional, this example shows the defaults
helper = CfnResource(json_logging=False, log_level='DEBUG', boto_level='CRITICAL', sleep_on_delete=120, ssl_verify=None)

try:
    ## Init code goes here
    pass
except Exception as e:
    helper.init_failure(e)


@helper.create
def create(event, context):
    logger.info("Got Create")
    # Optionally return an ID that will be used for the resource PhysicalResourceId, 
    # if None is returned an ID will be generated. If a poll_create function is defined 
    # return value is placed into the poll event as event['CrHelperData']['PhysicalResourceId']
    #
    # To add response data update the helper.Data dict
    # If poll is enabled data is placed into poll event as event['CrHelperData']
    helper.Data.update({"test": "testdata"})

    # To return an error to cloudformation you raise an exception:
    if not helper.Data.get("test"):
        raise ValueError("this error will show in the cloudformation events log and console.")
    
    return "MyResourceId"


@helper.update
def update(event, context):
    logger.info("Got Update")
    # If the update resulted in a new resource being created, return an id for the new resource. 
    # CloudFormation will send a delete event with the old id when stack update completes


@helper.delete
def delete(event, context):
    logger.info("Got Delete")
    # Delete never returns anything. Should not fail if the underlying resources are already deleted.
    # Desired state.


@helper.poll_create
def poll_create(event, context):
    logger.info("Got create poll")
    # Return a resource id or True to indicate that creation is complete. if True is returned an id 
    # will be generated
    return True


def handler(event, context):
    helper(event, context)

Polling

If you need longer than the max runtime of 15 minutes, you can enable polling by adding additional decorators for poll_create, poll_update or poll_delete. When a poll function is defined for create/update/delete the function will not send a response to CloudFormation and instead a CloudWatch Events schedule will be created to re-invoke the lambda function every 2 minutes. When the function is invoked the matching @helper.poll_ function will be called, logic to check for completion should go here, if the function returns None then the schedule will run again in 2 minutes. Once complete either return a PhysicalResourceID or True to have one generated. The schedule will be deleted and a response sent back to CloudFormation. If you use polling the following additional IAM policy must be attached to the function's IAM role:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "lambda:AddPermission",
        "lambda:RemovePermission",
        "events:PutRule",
        "events:DeleteRule",
        "events:PutTargets",
        "events:RemoveTargets"
      ],
      "Resource": "*"
    }
  ]
}

Certificate Verification

To turn off certification verification, or to use a custom CA bundle path for the underlying boto3 clients used by this library, override the ssl_verify argument with the appropriate values. These can be either:

• False - do not validate SSL certificates. SSL will still be used, but SSL certificates will not be verified.
• path/to/cert/bundle.pem - A filename of the CA cert bundle to uses. You can specify this argument if you want to use a different CA cert bundle than the one used by botocore.

Use CDK to depoy a Custom Resource that uses Custom Resource Helper

You can use the AWS Cloud Development Kit (AWS CDK) to deploy a Custom Resource that uses Custom Resource Helper. AWS CDK is an open-source software development framework for defining cloud infrastructure in code and provisioning it through AWS CloudFormation.

Note: crhelper is not intended to be used with AWS CDK using the Provider construct.

AWS CDK template example

from aws_cdk import (
    ...
    aws_lambda as _lambda,
    CustomResource,
)

crhelperSumResource = _lambda.Function(...)

customResource = CustomResource(
  self, 
  'MyCustomResource'
  serviceToken = crhelperSumResource.function_arn,
  properties = {
    'No1': 1,
    'No2': 2
  },
)

Credits

Decorator implementation inspired by https://github.com/ryansb/cfn-wrapper-python

Log implementation inspired by https://gitlab.com/hadrien/aws_lambda_logging

License

This library is licensed under the Apache 2.0 License.