fix(s3-deployment): updating memoryLimit or vpc results in stack update failure (#17530)

This fixes the issue where updating the memoryLimit or vpc of the
BucketDeployment resource would result in a stack update failure. In
order to fix this the ID of the BucketDeployment CustomResource includes
info on the memoryLimit and vpc in the same way that the Lambda function
does. This means that when either of these values are updated, the
BucketDeployment CustomResource will be recreated along with the Lambda
function.

If anyone is setting `retainOnDelete=false` (default is `true`) then
this change would result in the data in the bucket being deleted. In
order to avoid this scenario, this PR introduces a bucket tag that
controls whether or not a BucketDeployment resource can delete data from
the bucket.

The BucketDeployment resource will now add a tag to the deployment
bucket with a format like `aws-cdk:cr-owned:{keyPrefix}:{uniqueHash}`.
For example:

```
{
  Key: 'aws-cdk:cr-owned:deploy/here/:240D17B3',
  Value: 'true',
}
```

Each bucket+keyPrefix can be "owned" by 1 or more BucketDeployment
resources. Since there are some scenarios where multiple BucketDeployment
resources can deploy to the same bucket and key prefix
(e.g. using include/exclude) we also append part of the id to
make the key unique.

As long as a bucket+keyPrefix is "owned" by a BucketDeployment
resource, another CR cannot delete data. There are a couple of
scenarios where this comes into play.

1. If the LogicalResourceId of the CustomResource changes
(e.g. memoryLimit is updated) CloudFormation will first issue a 'Create'
to create the new CustomResource and will update the Tag on the bucket.
CloudFormation will then issue a 'Delete' on the old CustomResource
and since the new CR "owns" the Bucket+keyPrefix (indicated by the
presence of the tag), the old CR will not delete the contents of the bucket

2. If the BucketDeployment resource is deleted _and_ it is the only CR
for that bucket+keyPrefix then CloudFormation will first remove the tag
from the bucket and then issue a "Delete" to the CR. Since there are no
tags indicating that this bucket+keyPrefix is "owned" then it will delete
the contents.

3. If the BucketDeployment resource is deleted _and_ it is *not* the only
CR for that bucket:keyPrefix then CloudFormation will first remove the tag
from the bucket and then issue a "Delete" to the CR.
Since there are other CRs that also "own" that bucket+keyPrefix
(indicated by the presence of tags), there will
still be a tag on the bucket and the contents will not be removed. The
contents will only be removed after _all_ BucketDeployment resource that
"own" the bucket+keyPrefix have been removed.

4. If the BucketDeployment resource _and_ the S3 Bucket are both removed,
then CloudFormation will first issue a "Delete" to the CR and since there
is a tag on the bucket the contents will not be removed. If you want the
contents of the bucket to be removed on bucket deletion, then
`autoDeleteObjects` property should be set to true on the Bucket.

Scenario 3 & 4 are changes to the existing functionality in that they
now do *not* delete data in some scenarios when they did previously.

fixes #7128


----

*By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license*

Author: corymhall

packages/@aws-cdk/aws-ecs/test/ec2/integ.environment-file.expected.json

@@ -8,11 +8,181 @@
           "BlockPublicPolicy": true,
           "IgnorePublicAcls": true,
           "RestrictPublicBuckets": true
+        },
+        "Tags": [
+          {
+            "Key": "aws-cdk:auto-delete-objects",
+            "Value": "true"
+          },
+          {
+            "Key": "aws-cdk:cr-owned:f8f0a91c",
+            "Value": "true"
+          }
+        ]
+      },
+      "UpdateReplacePolicy": "Delete",
+      "DeletionPolicy": "Delete"
+    },
+    "BucketPolicyE9A3008A": {
+      "Type": "AWS::S3::BucketPolicy",
+      "Properties": {
+        "Bucket": {
+          "Ref": "Bucket83908E77"
+        },
+        "PolicyDocument": {
+          "Statement": [
+            {
+              "Action": [
+                "s3:GetBucket*",
+                "s3:List*",
+                "s3:DeleteObject*"
+              ],
+              "Effect": "Allow",
+              "Principal": {
+                "AWS": {
+                  "Fn::GetAtt": [
+                    "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092",
+                    "Arn"
+                  ]
+                }
+              },
+              "Resource": [
+                {
+                  "Fn::GetAtt": [
+                    "Bucket83908E77",
+                    "Arn"
+                  ]
+                },
+                {
+                  "Fn::Join": [
+                    "",
+                    [
+                      {
+                        "Fn::GetAtt": [
+                          "Bucket83908E77",
+                          "Arn"
+                        ]
+                      },
+                      "/*"
+                    ]
+                  ]
+                }
+              ]
+            }
+          ],
+          "Version": "2012-10-17"
+        }
+      }
+    },
+    "BucketAutoDeleteObjectsCustomResourceBAFD23C2": {
+      "Type": "Custom::S3AutoDeleteObjects",
+      "Properties": {
+        "ServiceToken": {
+          "Fn::GetAtt": [
+            "CustomS3AutoDeleteObjectsCustomResourceProviderHandler9D90184F",
+            "Arn"
+          ]
+        },
+        "BucketName": {
+          "Ref": "Bucket83908E77"
         }
       },
+      "DependsOn": [
+        "BucketPolicyE9A3008A"
+      ],
       "UpdateReplacePolicy": "Delete",
       "DeletionPolicy": "Delete"
     },
+    "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092": {
+      "Type": "AWS::IAM::Role",
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Action": "sts:AssumeRole",
+              "Effect": "Allow",
+              "Principal": {
+                "Service": "lambda.amazonaws.com"
+              }
+            }
+          ]
+        },
+        "ManagedPolicyArns": [
+          {
+            "Fn::Sub": "arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+          }
+        ]
+      }
+    },
+    "CustomS3AutoDeleteObjectsCustomResourceProviderHandler9D90184F": {
+      "Type": "AWS::Lambda::Function",
+      "Properties": {
+        "Code": {
+          "S3Bucket": {
+            "Ref": "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3Bucket2C6C817C"
+          },
+          "S3Key": {
+            "Fn::Join": [
+              "",
+              [
+                {
+                  "Fn::Select": [
+                    0,
+                    {
+                      "Fn::Split": [
+                        "||",
+                        {
+                          "Ref": "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3VersionKeyFA215BD6"
+                        }
+                      ]
+                    }
+                  ]
+                },
+                {
+                  "Fn::Select": [
+                    1,
+                    {
+                      "Fn::Split": [
+                        "||",
+                        {
+                          "Ref": "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3VersionKeyFA215BD6"
+                        }
+                      ]
+                    }
+                  ]
+                }
+              ]
+            ]
+          }
+        },
+        "Timeout": 900,
+        "MemorySize": 128,
+        "Handler": "__entrypoint__.handler",
+        "Role": {
+          "Fn::GetAtt": [
+            "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092",
+            "Arn"
+          ]
+        },
+        "Runtime": "nodejs12.x",
+        "Description": {
+          "Fn::Join": [
+            "",
+            [
+              "Lambda function for auto-deleting objects in ",
+              {
+                "Ref": "Bucket83908E77"
+              },
+              " S3 bucket."
+            ]
+          ]
+        }
+      },
+      "DependsOn": [
+        "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092"
+      ]
+    },
     "Vpc8378EB38": {
       "Type": "AWS::EC2::VPC",
       "Properties": {
@@ -1087,7 +1257,6 @@
         "DestinationBucketName": {
           "Ref": "Bucket83908E77"
         },
-        "RetainOnDelete": false,
         "Prune": true
       },
       "UpdateReplacePolicy": "Delete",
@@ -1219,7 +1388,7 @@
       "Properties": {
         "Code": {
           "S3Bucket": {
-            "Ref": "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3BucketD1AD544E"
+            "Ref": "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3Bucket1BE31DB0"
           },
           "S3Key": {
             "Fn::Join": [
@@ -1232,7 +1401,7 @@
                       "Fn::Split": [
                         "||",
                         {
-                          "Ref": "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3VersionKey93A19D70"
+                          "Ref": "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3VersionKeyDC38E49C"
                         }
                       ]
                     }
@@ -1245,7 +1414,7 @@
                       "Fn::Split": [
                         "||",
                         {
-                          "Ref": "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3VersionKey93A19D70"
+                          "Ref": "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3VersionKeyDC38E49C"
                         }
                       ]
                     }
@@ -1332,9 +1501,17 @@
     }
   },
   "Parameters": {
-    "SsmParameterValueawsserviceecsoptimizedamiamazonlinux2recommendedimageidC96584B6F00A464EAD1953AFF4B05118Parameter": {
-      "Type": "AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>",
-      "Default": "/aws/service/ecs/optimized-ami/amazon-linux-2/recommended/image_id"
+    "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3Bucket2C6C817C": {
+      "Type": "String",
+      "Description": "S3 bucket for asset \"84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709\""
+    },
+    "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3VersionKeyFA215BD6": {
+      "Type": "String",
+      "Description": "S3 key for asset version \"84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709\""
+    },
+    "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709ArtifactHash17D48178": {
+      "Type": "String",
+      "Description": "Artifact hash for asset \"84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709\""
     },
     "AssetParameterse9882ab123687399f934da0d45effe675ecc8ce13b40cb946f3e1d6141fe8d68S3BucketAEADE8C7": {
       "Type": "String",
@@ -1348,17 +1525,17 @@
       "Type": "String",
       "Description": "Artifact hash for asset \"e9882ab123687399f934da0d45effe675ecc8ce13b40cb946f3e1d6141fe8d68\""
     },
-    "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3BucketD1AD544E": {
+    "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3Bucket1BE31DB0": {
       "Type": "String",
-      "Description": "S3 bucket for asset \"a3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8\""
+      "Description": "S3 bucket for asset \"983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c\""
     },
-    "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3VersionKey93A19D70": {
+    "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3VersionKeyDC38E49C": {
       "Type": "String",
-      "Description": "S3 key for asset version \"a3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8\""
+      "Description": "S3 key for asset version \"983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c\""
     },
-    "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8ArtifactHash238275D6": {
+    "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cArtifactHashBA6352EA": {
       "Type": "String",
-      "Description": "Artifact hash for asset \"a3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8\""
+      "Description": "Artifact hash for asset \"983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c\""
     },
     "AssetParameters972240f9dd6e036a93d5f081af9a24315b2053828ac049b3b19b2fa12d7ae64aS3Bucket1F1A8472": {
       "Type": "String",
@@ -1383,6 +1560,10 @@
     "AssetParameters872561bf078edd1685d50c9ff821cdd60d2b2ddfb0013c4087e79bf2bb50724dArtifactHashC2522C05": {
       "Type": "String",
       "Description": "Artifact hash for asset \"872561bf078edd1685d50c9ff821cdd60d2b2ddfb0013c4087e79bf2bb50724d\""
+    },
+    "SsmParameterValueawsserviceecsoptimizedamiamazonlinux2recommendedimageidC96584B6F00A464EAD1953AFF4B05118Parameter": {
+      "Type": "AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>",
+      "Default": "/aws/service/ecs/optimized-ami/amazon-linux-2/recommended/image_id"
     }
   }
 }

packages/@aws-cdk/aws-ecs/test/ec2/integ.environment-file.ts

@@ -13,6 +13,7 @@ const stack = new cdk.Stack(app, 'aws-ecs-integ');
 const bucket = new s3.Bucket(stack, 'Bucket', {
   blockPublicAccess: s3.BlockPublicAccess.BLOCK_ALL,
   removalPolicy: cdk.RemovalPolicy.DESTROY,
+  autoDeleteObjects: true,
 });
 const vpc = new ec2.Vpc(stack, 'Vpc', { maxAzs: 2 });
 
@@ -47,7 +48,6 @@ const taskDefinition = new ecs.Ec2TaskDefinition(stack, 'TaskDefinition', {
 // deploy an envfile to S3 and delete when the bucket is deleted
 const envFileDeployment = new s3deployment.BucketDeployment(stack, 'EnvFileDeployment', {
   destinationBucket: bucket,
-  retainOnDelete: false,
   sources: [s3deployment.Source.asset(path.join(__dirname, '../demo-envfiles'))],
 });
 

packages/@aws-cdk/aws-s3-deployment/lib/bucket-deployment.ts

@@ -15,6 +15,8 @@ import { ISource, SourceConfig } from './source';
 // eslint-disable-next-line no-duplicate-imports, import/order
 import { Construct as CoreConstruct } from '@aws-cdk/core';
 
+// tag key has a limit of 128 characters
+const CUSTOM_RESOURCE_OWNER_TAG = 'aws-cdk:cr-owned';
 /**
  * Properties for `BucketDeployment`.
  */
@@ -32,6 +34,8 @@ export interface BucketDeploymentProps {
   /**
    * Key prefix in the destination bucket.
    *
+   * Must be <=104 characters
+   *
    * @default "/" (unzip to root of the destination bucket)
    */
   readonly destinationKeyPrefix?: string;
@@ -292,7 +296,7 @@ export class BucketDeployment extends CoreConstruct {
       filesystem: accessPoint ? lambda.FileSystem.fromEfsAccessPoint(
         accessPoint,
         mountPath,
-      ): undefined,
+      ) : undefined,
     });
 
     const handlerRole = handler.role;
@@ -309,7 +313,8 @@ export class BucketDeployment extends CoreConstruct {
       }));
     }
 
-    new cdk.CustomResource(this, 'CustomResource', {
+    const crUniqueId = `CustomResource${this.renderUniqueId(props.memoryLimit, props.vpc)}`;
+    const cr = new cdk.CustomResource(this, crUniqueId, {
       serviceToken: handler.functionArn,
       resourceType: 'Custom::CDKBucketDeployment',
       properties: {
@@ -328,10 +333,65 @@ export class BucketDeployment extends CoreConstruct {
       },
     });
 
+    let prefix: string = props.destinationKeyPrefix ?
+      `:${props.destinationKeyPrefix}` :
+      '';
+    prefix += `:${cr.node.addr.substr(-8)}`;
+    const tagKey = CUSTOM_RESOURCE_OWNER_TAG + prefix;
+
+    // destinationKeyPrefix can be 104 characters before we hit
+    // the tag key limit of 128
+    // '/this/is/a/random/key/prefix/that/is/a/lot/of/characters/do/we/think/that/it/will/ever/be/this/long?????'
+    // better to throw an error here than wait for CloudFormation to fail
+    if (tagKey.length > 128) {
+      throw new Error('The BucketDeployment construct requires that the "destinationKeyPrefix" be <=104 characters');
+    }
+
+    /*
+     * This will add a tag to the deployment bucket in the format of
+     * `aws-cdk:cr-owned:{keyPrefix}:{uniqueHash}`
+     *
+     * For example:
+     * {
+     *   Key: 'aws-cdk:cr-owned:deploy/here/:240D17B3',
+     *   Value: 'true',
+     * }
+     *
+     * This will allow for scenarios where there is a single S3 Bucket that has multiple
+     * BucketDeployment resources deploying to it. Each bucket + keyPrefix can be "owned" by
+     * 1 or more BucketDeployment resources. Since there are some scenarios where multiple BucketDeployment
+     * resources can deploy to the same bucket and key prefix (e.g. using include/exclude) we
+     * also append part of the id to make the key unique.
+     *
+     * As long as a bucket + keyPrefix is "owned" by a BucketDeployment resource, another CR
+     * cannot delete data. There are a couple of scenarios where this comes into play.
+     *
+     * 1. If the LogicalResourceId of the CustomResource changes (e.g. the crUniqueId changes)
+     * CloudFormation will first issue a 'Create' to create the new CustomResource and will
+     * update the Tag on the bucket. CloudFormation will then issue a 'Delete' on the old CustomResource
+     * and since the new CR "owns" the Bucket+keyPrefix it will not delete the contents of the bucket
+     *
+     * 2. If the BucketDeployment resource is deleted _and_ it is the only CR for that bucket+keyPrefix
+     * then CloudFormation will first remove the tag from the bucket and then issue a "Delete" to the
+     * CR. Since there are no tags indicating that this bucket+keyPrefix is "owned" then it will delete
+     * the contents.
+     *
+     * 3. If the BucketDeployment resource is deleted _and_ it is *not* the only CR for that bucket:keyPrefix
+     * then CloudFormation will first remove the tag from the bucket and then issue a "Delete" to the CR.
+     * Since there are other CRs that also "own" that bucket+keyPrefix there will still be a tag on the bucket
+     * and the contents will not be removed.
+     *
+     * 4. If the BucketDeployment resource _and_ the S3 Bucket are both removed, then CloudFormation will first
+     * issue a "Delete" to the CR and since there is a tag on the bucket the contents will not be removed. If you
+     * want the contents of the bucket to be removed on bucket deletion, then `autoDeleteObjects` property should
+     * be set to true on the Bucket.
+     */
+    cdk.Tags.of(props.destinationBucket).add(tagKey, 'true');
+
   }
 
-  private renderSingletonUuid(memoryLimit?: number, vpc?: ec2.IVpc) {
-    let uuid = '8693BB64-9689-44B6-9AAF-B0CC9EB8756C';
+  private renderUniqueId(memoryLimit?: number, vpc?: ec2.IVpc) {
+    let uuid = '';
 
     // if user specify a custom memory limit, define another singleton handler
     // with this configuration. otherwise, it won't be possible to use multiple
@@ -355,6 +415,14 @@ export class BucketDeployment extends CoreConstruct {
     return uuid;
   }
 
+  private renderSingletonUuid(memoryLimit?: number, vpc?: ec2.IVpc) {
+    let uuid = '8693BB64-9689-44B6-9AAF-B0CC9EB8756C';
+
+    uuid += this.renderUniqueId(memoryLimit, vpc);
+
+    return uuid;
+  }
+
   /**
    * Function to get/create a stack singleton instance of EFS FileSystem per vpc.
    *
@@ -453,7 +521,7 @@ export class CacheControl {
      * The raw cache control setting.
      */
     public readonly value: any,
-  ) {}
+  ) { }
 }
 
 /**
@@ -551,7 +619,7 @@ export class Expires {
      * The raw expiration date expression.
      */
     public readonly value: any,
-  ) {}
+  ) { }
 }
 
 /**

packages/@aws-cdk/aws-s3-deployment/lib/lambda/index.py

@@ -15,11 +15,14 @@
 logger.setLevel(logging.INFO)
 
 cloudfront = boto3.client('cloudfront')
+s3 = boto3.client('s3')
 
 CFN_SUCCESS = "SUCCESS"
 CFN_FAILED = "FAILED"
 ENV_KEY_MOUNT_PATH = "MOUNT_PATH"
 
+CUSTOM_RESOURCE_OWNER_TAG = "aws-cdk:cr-owned"
+
 def handler(event, context):
 
     def cfn_error(message=None):
@@ -68,9 +71,9 @@ def cfn_error(message=None):
 
         s3_source_zips = map(lambda name, key: "s3://%s/%s" % (name, key), source_bucket_names, source_object_keys)
         s3_dest = "s3://%s/%s" % (dest_bucket_name, dest_bucket_prefix)
-
         old_s3_dest = "s3://%s/%s" % (old_props.get("DestinationBucketName", ""), old_props.get("DestinationBucketKeyPrefix", ""))
 
+
         # obviously this is not
         if old_s3_dest == "s3:///":
             old_s3_dest = None
@@ -89,7 +92,8 @@ def cfn_error(message=None):
 
         # delete or create/update (only if "retain_on_delete" is false)
         if request_type == "Delete" and not retain_on_delete:
-            aws_command("s3", "rm", s3_dest, "--recursive")
+            if not bucket_owned(dest_bucket_name, dest_bucket_prefix):
+                aws_command("s3", "rm", s3_dest, "--recursive")
 
         # if we are updating without retention and the destination changed, delete first
         if request_type == "Update" and not retain_on_delete and old_s3_dest != s3_dest:
@@ -233,3 +237,21 @@ def cfn_send(event, context, responseStatus, responseData={}, physicalResourceId
     except Exception as e:
         logger.error("| unable to send response to CloudFormation")
         logger.exception(e)
+
+
+#---------------------------------------------------------------------------------------------------
+# check if bucket is owned by a custom resource
+# if it is then we don't want to delete content
+def bucket_owned(bucketName, keyPrefix):
+    tag = CUSTOM_RESOURCE_OWNER_TAG
+    if keyPrefix != "":
+        tag = tag + ':' + keyPrefix
+    try:
+        request = s3.get_bucket_tagging(
+            Bucket=bucketName,
+        )
+        return any((x["Key"].startswith(tag)) for x in request["TagSet"])
+    except Exception as e:
+        logger.info("| error getting tags from bucket")
+        logger.exception(e)
+        return False

packages/@aws-cdk/aws-s3-deployment/test/bucket-deployment.test.ts

@@ -1,5 +1,6 @@
 import '@aws-cdk/assert-internal/jest';
 import * as path from 'path';
+import { MatchStyle, objectLike } from '@aws-cdk/assert-internal';
 import * as cloudfront from '@aws-cdk/aws-cloudfront';
 import * as ec2 from '@aws-cdk/aws-ec2';
 import * as iam from '@aws-cdk/aws-iam';
@@ -922,3 +923,120 @@ test('deployment allows vpc and subnets to be implicitly supplied to lambda', ()
     },
   });
 });
+
+test('resource id includes memory and vpc', () => {
+
+  // GIVEN
+  const stack = new cdk.Stack();
+  const bucket = new s3.Bucket(stack, 'Dest');
+  const vpc: ec2.IVpc = new ec2.Vpc(stack, 'SomeVpc2', {});
+
+  // WHEN
+  new s3deploy.BucketDeployment(stack, 'DeployWithVpc2', {
+    sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'))],
+    destinationBucket: bucket,
+    vpc,
+    memoryLimit: 256,
+  });
+
+  // THEN
+  expect(stack).toMatchTemplate({
+    Resources: objectLike({
+      DeployWithVpc2CustomResource256MiBc8a39596cb8641929fcf6a288bc9db5ab7b0f656ad3C5F6E78: objectLike({
+        Type: 'Custom::CDKBucketDeployment',
+      }),
+    }),
+  }, MatchStyle.SUPERSET);
+});
+
+test('bucket includes custom resource owner tag', () => {
+
+  // GIVEN
+  const stack = new cdk.Stack();
+  const bucket = new s3.Bucket(stack, 'Dest');
+  const vpc: ec2.IVpc = new ec2.Vpc(stack, 'SomeVpc2', {});
+
+  // WHEN
+  new s3deploy.BucketDeployment(stack, 'DeployWithVpc2', {
+    sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'))],
+    destinationBucket: bucket,
+    destinationKeyPrefix: '/a/b/c',
+    vpc,
+    memoryLimit: 256,
+  });
+
+  // THEN
+  expect(stack).toHaveResource('AWS::S3::Bucket', {
+    Tags: [{
+      Key: 'aws-cdk:cr-owned:/a/b/c:971e1fa8',
+      Value: 'true',
+    }],
+  });
+});
+
+test('throws if destinationKeyPrefix is too long', () => {
+
+  // GIVEN
+  const stack = new cdk.Stack();
+  const bucket = new s3.Bucket(stack, 'Dest');
+
+  // WHEN
+  expect(() => new s3deploy.BucketDeployment(stack, 'DeployWithVpc2', {
+    sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'))],
+    destinationBucket: bucket,
+    destinationKeyPrefix: '/this/is/a/random/key/prefix/that/is/a/lot/of/characters/do/we/think/that/it/will/ever/be/this/long??????',
+    memoryLimit: 256,
+  })).toThrow(/The BucketDeployment construct requires that/);
+
+});
+
+test('bucket has multiple deployments', () => {
+
+  // GIVEN
+  const stack = new cdk.Stack();
+  const bucket = new s3.Bucket(stack, 'Dest');
+  const vpc: ec2.IVpc = new ec2.Vpc(stack, 'SomeVpc2', {});
+
+  // WHEN
+  new s3deploy.BucketDeployment(stack, 'DeployWithVpc2', {
+    sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'))],
+    destinationBucket: bucket,
+    destinationKeyPrefix: '/a/b/c',
+    vpc,
+    memoryLimit: 256,
+  });
+
+  new s3deploy.BucketDeployment(stack, 'DeployWithVpc2Exclude', {
+    sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'), {
+      exclude: ['index.html'],
+    })],
+    destinationBucket: bucket,
+    destinationKeyPrefix: '/a/b/c',
+    vpc,
+    memoryLimit: 256,
+  });
+
+  new s3deploy.BucketDeployment(stack, 'DeployWithVpc3', {
+    sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'))],
+    destinationBucket: bucket,
+    destinationKeyPrefix: '/x/z',
+  });
+
+  // THEN
+  expect(stack).toHaveResource('AWS::S3::Bucket', {
+    Tags: [
+      {
+        Key: 'aws-cdk:cr-owned:/a/b/c:6da0a4ab',
+        Value: 'true',
+      },
+      {
+        Key: 'aws-cdk:cr-owned:/a/b/c:971e1fa8',
+        Value: 'true',
+      },
+      {
+        Key: 'aws-cdk:cr-owned:/x/z:2db04622',
+        Value: 'true',
+      },
+    ],
+  });
+});

packages/@aws-cdk/aws-s3-deployment/test/integ.bucket-deployment-cloudfront.expected.json

@@ -2,9 +2,181 @@
   "Resources": {
     "Destination3E3DC043D": {
       "Type": "AWS::S3::Bucket",
+      "Properties": {
+        "Tags": [
+          {
+            "Key": "aws-cdk:auto-delete-objects",
+            "Value": "true"
+          },
+          {
+            "Key": "aws-cdk:cr-owned:4685d093",
+            "Value": "true"
+          }
+        ]
+      },
+      "UpdateReplacePolicy": "Delete",
+      "DeletionPolicy": "Delete"
+    },
+    "Destination3Policy685DA6C5": {
+      "Type": "AWS::S3::BucketPolicy",
+      "Properties": {
+        "Bucket": {
+          "Ref": "Destination3E3DC043D"
+        },
+        "PolicyDocument": {
+          "Statement": [
+            {
+              "Action": [
+                "s3:GetBucket*",
+                "s3:List*",
+                "s3:DeleteObject*"
+              ],
+              "Effect": "Allow",
+              "Principal": {
+                "AWS": {
+                  "Fn::GetAtt": [
+                    "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092",
+                    "Arn"
+                  ]
+                }
+              },
+              "Resource": [
+                {
+                  "Fn::GetAtt": [
+                    "Destination3E3DC043D",
+                    "Arn"
+                  ]
+                },
+                {
+                  "Fn::Join": [
+                    "",
+                    [
+                      {
+                        "Fn::GetAtt": [
+                          "Destination3E3DC043D",
+                          "Arn"
+                        ]
+                      },
+                      "/*"
+                    ]
+                  ]
+                }
+              ]
+            }
+          ],
+          "Version": "2012-10-17"
+        }
+      }
+    },
+    "Destination3AutoDeleteObjectsCustomResourceC26EC7B7": {
+      "Type": "Custom::S3AutoDeleteObjects",
+      "Properties": {
+        "ServiceToken": {
+          "Fn::GetAtt": [
+            "CustomS3AutoDeleteObjectsCustomResourceProviderHandler9D90184F",
+            "Arn"
+          ]
+        },
+        "BucketName": {
+          "Ref": "Destination3E3DC043D"
+        }
+      },
+      "DependsOn": [
+        "Destination3Policy685DA6C5"
+      ],
       "UpdateReplacePolicy": "Delete",
       "DeletionPolicy": "Delete"
     },
+    "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092": {
+      "Type": "AWS::IAM::Role",
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Action": "sts:AssumeRole",
+              "Effect": "Allow",
+              "Principal": {
+                "Service": "lambda.amazonaws.com"
+              }
+            }
+          ]
+        },
+        "ManagedPolicyArns": [
+          {
+            "Fn::Sub": "arn:${AWS::Partition}:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+          }
+        ]
+      }
+    },
+    "CustomS3AutoDeleteObjectsCustomResourceProviderHandler9D90184F": {
+      "Type": "AWS::Lambda::Function",
+      "Properties": {
+        "Code": {
+          "S3Bucket": {
+            "Ref": "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3Bucket2C6C817C"
+          },
+          "S3Key": {
+            "Fn::Join": [
+              "",
+              [
+                {
+                  "Fn::Select": [
+                    0,
+                    {
+                      "Fn::Split": [
+                        "||",
+                        {
+                          "Ref": "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3VersionKeyFA215BD6"
+                        }
+                      ]
+                    }
+                  ]
+                },
+                {
+                  "Fn::Select": [
+                    1,
+                    {
+                      "Fn::Split": [
+                        "||",
+                        {
+                          "Ref": "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3VersionKeyFA215BD6"
+                        }
+                      ]
+                    }
+                  ]
+                }
+              ]
+            ]
+          }
+        },
+        "Timeout": 900,
+        "MemorySize": 128,
+        "Handler": "__entrypoint__.handler",
+        "Role": {
+          "Fn::GetAtt": [
+            "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092",
+            "Arn"
+          ]
+        },
+        "Runtime": "nodejs12.x",
+        "Description": {
+          "Fn::Join": [
+            "",
+            [
+              "Lambda function for auto-deleting objects in ",
+              {
+                "Ref": "Destination3E3DC043D"
+              },
+              " S3 bucket."
+            ]
+          ]
+        }
+      },
+      "DependsOn": [
+        "CustomS3AutoDeleteObjectsCustomResourceProviderRole3B1BD092"
+      ]
+    },
     "DistributionCFDistribution882A7313": {
       "Type": "AWS::CloudFront::Distribution",
       "Properties": {
@@ -295,7 +467,7 @@
       "Properties": {
         "Code": {
           "S3Bucket": {
-            "Ref": "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3BucketD1AD544E"
+            "Ref": "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3Bucket1BE31DB0"
           },
           "S3Key": {
             "Fn::Join": [
@@ -308,7 +480,7 @@
                       "Fn::Split": [
                         "||",
                         {
-                          "Ref": "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3VersionKey93A19D70"
+                          "Ref": "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3VersionKeyDC38E49C"
                         }
                       ]
                     }
@@ -321,7 +493,7 @@
                       "Fn::Split": [
                         "||",
                         {
-                          "Ref": "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3VersionKey93A19D70"
+                          "Ref": "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3VersionKeyDC38E49C"
                         }
                       ]
                     }
@@ -353,6 +525,18 @@
     }
   },
   "Parameters": {
+    "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3Bucket2C6C817C": {
+      "Type": "String",
+      "Description": "S3 bucket for asset \"84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709\""
+    },
+    "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709S3VersionKeyFA215BD6": {
+      "Type": "String",
+      "Description": "S3 key for asset version \"84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709\""
+    },
+    "AssetParameters84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709ArtifactHash17D48178": {
+      "Type": "String",
+      "Description": "Artifact hash for asset \"84e9b89449fe2573e51d08cc143e21116ed4608c6db56afffcb4ad85c8130709\""
+    },
     "AssetParameterse9882ab123687399f934da0d45effe675ecc8ce13b40cb946f3e1d6141fe8d68S3BucketAEADE8C7": {
       "Type": "String",
       "Description": "S3 bucket for asset \"e9882ab123687399f934da0d45effe675ecc8ce13b40cb946f3e1d6141fe8d68\""
@@ -365,17 +549,17 @@
       "Type": "String",
       "Description": "Artifact hash for asset \"e9882ab123687399f934da0d45effe675ecc8ce13b40cb946f3e1d6141fe8d68\""
     },
-    "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3BucketD1AD544E": {
+    "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3Bucket1BE31DB0": {
       "Type": "String",
-      "Description": "S3 bucket for asset \"a3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8\""
+      "Description": "S3 bucket for asset \"983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c\""
     },
-    "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8S3VersionKey93A19D70": {
+    "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cS3VersionKeyDC38E49C": {
       "Type": "String",
-      "Description": "S3 key for asset version \"a3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8\""
+      "Description": "S3 key for asset version \"983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c\""
     },
-    "AssetParametersa3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8ArtifactHash238275D6": {
+    "AssetParameters983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4cArtifactHashBA6352EA": {
       "Type": "String",
-      "Description": "Artifact hash for asset \"a3058ccb468d757ebb89df5363a1c20f5307c6911136f29d00e1a68c9b2aa7e8\""
+      "Description": "Artifact hash for asset \"983c442a2fe823a8b4ebb18d241a5150ae15103dacbf3f038c7c6343e565aa4c\""
     },
     "AssetParametersfc4481abf279255619ff7418faa5d24456fef3432ea0da59c95542578ff0222eS3Bucket9CD8B20A": {
       "Type": "String",

packages/@aws-cdk/aws-s3-deployment/test/integ.bucket-deployment-cloudfront.ts

@@ -10,6 +10,7 @@ class TestBucketDeployment extends cdk.Stack {
 
     const bucket = new s3.Bucket(this, 'Destination3', {
       removalPolicy: cdk.RemovalPolicy.DESTROY,
+      autoDeleteObjects: true, // needed for integration test cleanup
     });
     const distribution = new cloudfront.CloudFrontWebDistribution(this, 'Distribution', {
       originConfigs: [

packages/@aws-cdk/aws-s3-deployment/test/integ.bucket-deployment.expected.json

@@ -12,6 +12,7 @@ class TestBucketDeployment extends cdk.Stack {
       websiteIndexDocument: 'index.html',
       publicReadAccess: false,
       removalPolicy: cdk.RemovalPolicy.DESTROY,
+      autoDeleteObjects: true, // needed for integration test cleanup
     });
 
     new s3deploy.BucketDeployment(this, 'DeployMe', {
@@ -29,7 +30,10 @@ class TestBucketDeployment extends cdk.Stack {
       retainOnDelete: false, // default is true, which will block the integration test cleanup
     });
 
-    const bucket2 = new s3.Bucket(this, 'Destination2');
+    const bucket2 = new s3.Bucket(this, 'Destination2', {
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+      autoDeleteObjects: true, // needed for integration test cleanup
+    });
 
     new s3deploy.BucketDeployment(this, 'DeployWithPrefix', {
       sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'))],
@@ -38,7 +42,10 @@ class TestBucketDeployment extends cdk.Stack {
       retainOnDelete: false, // default is true, which will block the integration test cleanup
     });
 
-    const bucket3 = new s3.Bucket(this, 'Destination3');
+    const bucket3 = new s3.Bucket(this, 'Destination3', {
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+      autoDeleteObjects: true, // needed for integration test cleanup
+    });
 
     new s3deploy.BucketDeployment(this, 'DeployWithMetadata', {
       sources: [s3deploy.Source.asset(path.join(__dirname, 'my-website'))],

packages/@aws-cdk/aws-s3-deployment/test/integ.bucket-deployment.ts

@@ -242,23 +242,54 @@ def test_create_update_with_metadata(self):
         )
 
     def test_delete_no_retain(self):
-        invoke_handler("Delete", {
-            "SourceBucketNames": ["<source-bucket>"],
-            "SourceObjectKeys": ["<source-object-key>"],
-            "DestinationBucketName": "<dest-bucket-name>",
-            "RetainOnDelete": "false"
-        }, physical_id="<physicalid>")
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'GetBucketTagging':
+                assert kwarg['Bucket'] == '<dest-bucket-name>'
+                return {'TagSet': [{'Key': 'random', 'Value': '<logical-resource-id>'}]}
+            raise ClientError({'Error': {'Code': '500', 'Message': 'Unsupported operation'}}, operation_name)
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            invoke_handler("Delete", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<dest-bucket-name>",
+                "RetainOnDelete": "false"
+            }, physical_id="<physicalid>")
 
         self.assertAwsCommands(["s3", "rm", "s3://<dest-bucket-name>/", "--recursive"])
 
+    # In a replace the logcal id of the custom resource will change
+    # so the custom resource that gets the Delete event will no longer
+    # "own" the bucket
+    def test_replace_no_retain(self):
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'GetBucketTagging':
+                assert kwarg['Bucket'] == '<dest-bucket-name>'
+                return {'TagSet': [{'Key': 'aws-cdk:cr-owned:-bucket>', 'Value': '<some-other-logical-id>'}]}
+            raise ClientError({'Error': {'Code': '500', 'Message': 'Unsupported operation'}}, operation_name)
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            invoke_handler("Delete", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<dest-bucket-name>",
+                "RetainOnDelete": "false"
+            }, physical_id="<physicalid>")
+
+        self.assertAwsCommands()
+
     def test_delete_with_dest_key(self):
-        invoke_handler("Delete", {
-            "SourceBucketNames": ["<source-bucket>"],
-            "SourceObjectKeys": ["<source-object-key>"],
-            "DestinationBucketName": "<dest-bucket-name>",
-            "DestinationBucketKeyPrefix": "<dest-key-prefix>",
-            "RetainOnDelete": "false"
-        }, physical_id="<physicalid>")
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'GetBucketTagging':
+                assert kwarg['Bucket'] == '<dest-bucket-name>'
+                return {'TagSet': [{'Key': 'random-key', 'Value': '<logical-resource-id>'}]}
+            raise ClientError({'Error': {'Code': '500', 'Message': 'Unsupported operation'}}, operation_name)
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            invoke_handler("Delete", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<dest-bucket-name>",
+                "DestinationBucketKeyPrefix": "<dest-key-prefix>",
+                "RetainOnDelete": "false"
+            }, physical_id="<physicalid>")
 
         self.assertAwsCommands(["s3", "rm", "s3://<dest-bucket-name>/<dest-key-prefix>", "--recursive"])
 
@@ -285,12 +316,18 @@ def test_delete_with_retain_implicit_default(self):
         self.assertAwsCommands()
 
     def test_delete_with_retain_explicitly_false(self):
-        invoke_handler("Delete", {
-            "SourceBucketNames": ["<source-bucket>"],
-            "SourceObjectKeys": ["<source-object-key>"],
-            "DestinationBucketName": "<dest-bucket-name>",
-            "RetainOnDelete": "false"
-        }, physical_id="<physicalid>")
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'GetBucketTagging':
+                assert kwarg['Bucket'] == '<dest-bucket-name>'
+                return {'TagSet': [{'Key': 'random-key', 'Value': '<logical-resource-id>'}]}
+            raise ClientError({'Error': {'Code': '500', 'Message': 'Unsupported operation'}}, operation_name)
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            invoke_handler("Delete", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<dest-bucket-name>",
+                "RetainOnDelete": "false"
+            }, physical_id="<physicalid>")
 
         self.assertAwsCommands(
             ["s3", "rm", "s3://<dest-bucket-name>/", "--recursive"]
@@ -466,6 +503,7 @@ def test_update_new_dest_prefix_retain_implicit(self):
     #
 
     def test_physical_id_allocated_on_create_and_reused_afterwards(self):
+
         create_resp = invoke_handler("Create", {
             "SourceBucketNames": ["<source-bucket>"],
             "SourceObjectKeys": ["<source-object-key>"],
@@ -487,12 +525,17 @@ def test_physical_id_allocated_on_create_and_reused_afterwards(self):
         self.assertEqual(update_resp['PhysicalResourceId'], phid)
 
         # now issue a delete, and make sure this also applies
-        delete_resp = invoke_handler("Delete", {
-            "SourceBucketNames": ["<source-bucket>"],
-            "SourceObjectKeys": ["<source-object-key>"],
-            "DestinationBucketName": "<dest-bucket-name>",
-            "RetainOnDelete": "false"
-        }, physical_id=phid)
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'GetBucketTagging':
+                return {'TagSet': [{'Key': 'aws-cdk:cr-owned:-bucket>', 'Value': '<logical-resource-id>'}]}
+            raise ClientError({'Error': {'Code': '500', 'Message': 'Unsupported operation'}}, operation_name)
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            delete_resp = invoke_handler("Delete", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<dest-bucket-name>",
+                "RetainOnDelete": "false"
+            }, physical_id=phid)
         self.assertEqual(delete_resp['PhysicalResourceId'], phid)
 
     def test_fails_when_physical_id_not_present_in_update(self):
@@ -507,16 +550,42 @@ def test_fails_when_physical_id_not_present_in_update(self):
         self.assertEqual(update_resp['Reason'], "invalid request: request type is 'Update' but 'PhysicalResourceId' is not defined")
 
     def test_fails_when_physical_id_not_present_in_delete(self):
-        update_resp = invoke_handler("Delete", {
-            "SourceBucketNames": ["<source-bucket>"],
-            "SourceObjectKeys": ["<source-object-key>"],
-            "DestinationBucketName": "<new-dest-bucket-name>",
-        }, old_resource_props={
-            "DestinationBucketName": "<dest-bucket-name>",
-        }, expected_status="FAILED")
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'GetBucketTagging':
+                return {'TagSet': [{'Key': 'aws-cdk:cr-owned:-bucket>', 'Value': '<logical-resource-id>'}]}
+            raise ClientError({'Error': {'Code': '500', 'Message': 'Unsupported operation'}}, operation_name)
+
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            update_resp = invoke_handler("Delete", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<new-dest-bucket-name>",
+            }, old_resource_props={
+                "DestinationBucketName": "<dest-bucket-name>",
+            }, expected_status="FAILED")
 
         self.assertEqual(update_resp['Reason'], "invalid request: request type is 'Delete' but 'PhysicalResourceId' is not defined")
 
+    # no bucket tags removes content
+    def test_no_tags_on_bucket(self):
+        def mock_make_api_call(self, operation_name, kwarg):
+            if operation_name == 'GetBucketTagging':
+                raise ClientError({'Error': {'Code': 'NoSuchTagSet', 'Message': 'The TagSet does not exist'}}, operation_name)
+            raise ClientError({'Error': {'Code': '500', 'Message': 'Unsupported operation'}}, operation_name)
+
+        with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
+            invoke_handler("Delete", {
+                "SourceBucketNames": ["<source-bucket>"],
+                "SourceObjectKeys": ["<source-object-key>"],
+                "DestinationBucketName": "<dest-bucket-name>",
+                "RetainOnDelete": "false"
+            }, physical_id="<physicalid>")
+
+        self.assertAwsCommands(
+            ["s3", "rm", "s3://<dest-bucket-name>/", "--recursive"]
+        )
+
+
 
     # asserts that a given list of "aws xxx" commands have been invoked (in order)
     def assertAwsCommands(self, *expected):