An AWS Lambda can be triggered by events on an S3 bucket. A very useful feature when e.g. a file upload is made and some operation should be done on the uploaded file. The triggered Lambda can then read the file and perform these operations.

In AWS CloudFormation a Lambda trigger is set on the bucket. When the bucket is defined and deployed in the CloudFormation template a NotificationConfiguration containing a LambdaFunctionConfiguration is set on the bucket. However, setting up such a trigger on an existing bucket (i.e. the bucket exist outside of the stack) is a bit more complex and one method is described here.

Bucket notification configuration using Custom Resource

A solution is to setup the notification configuration on the existing bucket using an AWS CloudFormation Custom resource. An introduction to CloudFormation custom resources can be found in a previous note.

AWS has an article on the subject, however, that description will not allow multiple notification configurations on a bucket since it will overwrite the existing notification configuration (and delete all notification configurations in the delete case).

The solution presented here allows multiple stacks setting bucket notification configuration on the same bucket and has some additional error handling to prevent errors when creating or deleting the stack.

The S3 bucket notification configuration is set as properties on the custom resource. When the lambda resource handler is triggered by the custom resource it reads the notification configuration and adds it to the bucket. When the stack is deleted the lambda will also be triggered and should remove the notification configuration from the bucket. An example is described below.

CloudFormation Template

In this example the existing bucket is passed in as the parameter TestBucket to the stack (the bucket name could also be imported using the intrinsic function Fn::ImportValue). The idea is that a .png file added using the prefix images (e.g. <bucket>/images/faces/me.png) should trigger the Lambda TestLambda provided in the template.

The Custom Resource

    Type: Custom::S3BucketNotifications
          - BucketNotificationsHandler
          - Arn
        Ref: TestBucket
          - Events:
              - s3:ObjectCreated:*
                  - Name: suffix
                    Value: .png
                  - Name: prefix
                    Value: images
                - TestLambda
                - Arn
      Managed: false
      - AllowTestBucketInvokeTestLambda

The ServiceToken property is specifying the Lambda resource handler that will be invoked when the Custom Resource is created. The rest of the properties are to be read by that Lambda resource handler.

The NotificationConfiguration is the actual configuration to be set on the bucket. Other than that there is the Managed and BucketName properties, the latter is obvious but Managed might need a clarification. Managed set to true means that the bucket is created in this template. That’s not the case here, hence Managed is set to false.

When the bucket exist outside of the stack the bucket notification handler need to be cautious that there could be other notifications on the bucket. Those notifications should not be affected by the bucket notification handler. In the other case, when the bucket is Managed (i.e. created in the stack) no such considerations need to be made.

Custom Resource handler

This is the actual bucket notification handler. A lambda with inline python code. The code is actually what AWS CDK is generating when setting bucket notification configuration using the CDK.

    Type: AWS::Lambda::Function
      Description: AWS CloudFormation handler for "Custom::S3BucketNotifications" resources (@aws-cdk/aws-s3)
      Handler: index.handler
          - BucketNotificationsHandlerRole
          - Arn
      Runtime: python3.9
      Timeout: 300
        ZipFile: |
          import boto3  # type: ignore
          import json
          import logging
          import urllib.request

          s3 = boto3.client("s3")
          l = boto3.client("lambda")

          CONFIGURATION_TYPES = ["TopicConfigurations", "QueueConfigurations", "LambdaFunctionConfigurations"]

          def handler(event: dict, context):
              response_status = "SUCCESS"
              error_message = ""
                  props = event["ResourceProperties"]
                  bucket = props["BucketName"]
                  notification_configuration = props["NotificationConfiguration"]
                  request_type = event["RequestType"]
                  managed = props.get('Managed', 'true').lower() == 'true'
                  stack_id = event['StackId']

                  if managed:
                    config = handle_managed(request_type, notification_configuration)
                    config = handle_unmanaged(bucket, stack_id, request_type, notification_configuration)

                  put_bucket_notification_configuration(bucket, config)
              except Exception as e:
                  logging.exception("Failed to put bucket notification configuration")
                  response_status = "FAILED"
                  error_message = f"Error: {str(e)}. "
                  submit_response(event, context, response_status, error_message)

          def handle_managed(request_type, notification_configuration):
            if request_type == 'Delete':
              return {}
            return notification_configuration

          def handle_unmanaged(bucket, stack_id, request_type, notification_configuration):

            # find external notifications
            external_notifications = find_external_notifications(bucket, stack_id)

            # if delete, that's all we need
            if request_type == 'Delete':
              return external_notifications

            def with_id(notification):
              notification['Id'] = f"{stack_id}-{hash(json.dumps(notification, sort_keys=True))}"
              return notification

            # otherwise, merge external with incoming config and augment with id
            notifications = {}
            for t in CONFIGURATION_TYPES:
              external = external_notifications.get(t, [])
              incoming = [with_id(n) for n in notification_configuration.get(t, [])]
              notifications[t] = external + incoming
            return notifications

          def find_external_notifications(bucket, stack_id):
            existing_notifications = get_bucket_notification_configuration(bucket)
            external_notifications = {}
            for t in CONFIGURATION_TYPES:
              # if the notification was created by us, we know what id to expect
              # so we can filter by it.
              external_notifications[t] = [n for n in existing_notifications
                .get(t, []) if not n['Id'].startswith(f"{stack_id}-") and lambda_exist(n['LambdaFunctionArn'])]

            return external_notifications

          def lambda_exist(lambda_arn):
            except Exception as e:
              if 'ResourceNotFound' in str(e):
                print(f'lambda {lambda_arn} does not exist. {e}')
                return False
            return True

          def get_bucket_notification_configuration(bucket):
            return s3.get_bucket_notification_configuration(Bucket=bucket)

          def put_bucket_notification_configuration(bucket, notification_configuration):
            s3.put_bucket_notification_configuration(Bucket=bucket, NotificationConfiguration=notification_configuration)

          def submit_response(event: dict, context, response_status: str, error_message: str):
              response_body = json.dumps(
                      "Status": response_status,
                      "Reason": f"{error_message}See the details in CloudWatch Log Stream: {context.log_stream_name}",
                      "PhysicalResourceId": event.get("PhysicalResourceId") or event["LogicalResourceId"],
                      "StackId": event["StackId"],
                      "RequestId": event["RequestId"],
                      "LogicalResourceId": event["LogicalResourceId"],
                      "NoEcho": False,
              headers = {"content-type": "", "content-length": str(len(response_body))}
                  req = urllib.request.Request(url=event["ResponseURL"], headers=headers, data=response_body, method="PUT")
                  with urllib.request.urlopen(req) as response:
                  print("Status code: " + response.reason)
              except Exception as e:
                  print("send(..) failed executing request.urlopen(..): " + str(e))
      - BucketNotificationsHandlerRoleDefaultPolicy
      - BucketNotificationsHandlerRole

Basic idea

The Lambda will find all notifications on the bucket and will then add or remove the notification configuration for the current stack to/from the list of notification configurations and then re-set the configuration on the bucket.

Handle non-existing lambda trigger destination

As mentioned the custom resource lambda handler code is what CDK is generating - with one addition. There is a check added to make sure the lambda that should be triggered exists (the call to lambda_exist(). Otherwise, if the lambda does not exist any operation (create, update or delete) on the stack would fail. This is a corner case but can happen when having multiple CloudFormation stacks where the stacks contain a lambda that is triggered by an external S3 bucket and the stacks are deleted in parallel.

The call Traceback for the error would look like;

Failed to put bucket notification configuration
Traceback (most recent call last):
File "/var/task/", line 27, in handler
put_bucket_notification_configuration(bucket, config)
File "/var/task/", line 87, in put_bucket_notification_configuration
s3.put_bucket_notification_configuration(Bucket=bucket, NotificationConfiguration=notification_configuration)
File "/var/runtime/botocore/", line 391, in _api_call
return self._make_api_call(operation_name, kwargs)
File "/var/runtime/botocore/", line 719, in _make_api_call
raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidArgument) when calling the PutBucketNotificationConfiguration operation: Unable to validate the following destination configurations

I.e. the destination notification configuration is incorrect since one of the lambdas does not exist.


The S3 bucket need to have InvokeFunction permission on the target Lambda

    Type: AWS::Lambda::Permission
      Action: lambda:InvokeFunction
          - TestLambda
          - Arn
        Ref: AWS::AccountId
          - ''
          - - 'arn:aws:s3:::'
            - !Ref TestBucket

The custom resource handler lambda should assume Lambda Basic Execution Role (for logging to CloudWatch Logs):

    Type: AWS::IAM::Role
          - Action: sts:AssumeRole
            Effect: Allow
        Version: "2012-10-17"
        - Fn::Join:
            - ""
            - - "arn:"
              - Ref: AWS::Partition
              - :iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

The custom resource handler lambda need permission to add bucket notification on the bucket

    Type: AWS::IAM::Policy
          - Action: s3:*BucketNotification
            Effect: Allow
            Resource: "*"
        Version: "2012-10-17"
      PolicyName: BucketNotificationsHandlerRoleDefaultPolicy
        - Ref: BucketNotificationsHandlerRole


A working example can be found here.

Deploy the template using AWS Console or AWS Cli using:

$ aws cloudformation create-stack --stack-name bucket-notification-demo --template-body file://bucketnotification.yaml --capabilities CAPABILITY_IAM --parameters ParameterKey=TestBucket,ParameterValue=<your-bucket>


AWS Custom Resource

AWS Notification config on existing bucket with cloudformation