add DeleteObjects command to plugin (#38)

plugin/service/storage/plugin.go

@@ -12,6 +12,7 @@ import (
 	"io"
 	"os"
 	"path"
+	"strings"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/aws/retry"
@@ -622,6 +623,129 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
 	}, nil
 }
 
+// DeleteObjects is used to delete one or many objects from an s3 bucket.
+func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjectsRequest) (*pb.DeleteObjectsResponse, error) {
+	if req.GetKeyPrefix() == "" {
+		return nil, status.Error(codes.InvalidArgument, "key prefix is required")
+	}
+
+	bucket := req.GetBucket()
+	if bucket == nil {
+		return nil, status.Error(codes.InvalidArgument, "bucket is required")
+	}
+
+	if bucket.GetBucketName() == "" {
+		return nil, status.Error(codes.InvalidArgument, "bucketName is required")
+	}
+
+	attrs := bucket.GetAttributes()
+	if attrs == nil {
+		return nil, status.Error(codes.InvalidArgument, "attributes is required")
+	}
+
+	storageAttributes, err := getStorageAttributes(attrs)
+	if err != nil {
+		return nil, err
+	}
+
+	credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, false)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set up the creds in our persisted state.
+	credState, err := cred.NewAwsCredentialPersistedState(
+		append([]cred.AwsCredentialPersistedStateOption{
+			cred.WithCredentialsConfig(credConfig),
+		}, p.testCredStateOpts...)...,
+	)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
+	}
+
+	storageState, err := newAwsStoragePersistedState(
+		append([]awsStoragePersistedStateOption{
+			withCredentials(credState),
+		}, p.testStorageStateOpts...)...,
+	)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
+	}
+
+	opts := []s3Option{}
+	if storageAttributes.EndpointUrl != "" {
+		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
+	}
+	client, err := storageState.S3Client(ctx, opts...)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
+	}
+
+	prefix := path.Join(bucket.GetBucketPrefix(), req.GetKeyPrefix())
+	if strings.HasSuffix(req.GetKeyPrefix(), "/") {
+		// path.Join ends by "cleaning" the path, including removing a trailing slash, if
+		// it exists. given that a slash is used to denote a folder, it is required here.
+		prefix += "/"
+	}
+
+	if !req.Recursive {
+		_, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
+			Bucket: aws.String(bucket.GetBucketName()),
+			Key:    aws.String(prefix),
+		})
+		if err != nil {
+			return nil, parseAWSErrCode("error deleting S3 object", err)
+		}
+		return &pb.DeleteObjectsResponse{
+			ObjectsDeleted: uint32(1),
+		}, nil
+	}
+
+	const maxkeys = 1000
+
+	objects := []types.ObjectIdentifier{}
+	var conToken *string
+	truncated := true
+	for truncated {
+		res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+			Bucket:            aws.String(bucket.GetBucketName()),
+			Prefix:            aws.String(prefix),
+			MaxKeys:           maxkeys,
+			ContinuationToken: conToken,
+		})
+		if err != nil {
+			return nil, parseAWSErrCode("error iterating S3 bucket contents", err)
+		}
+		truncated = res.IsTruncated
+		conToken = res.NextContinuationToken
+		for _, o := range res.Contents {
+			objects = append(objects, types.ObjectIdentifier{
+				Key: o.Key,
+			})
+		}
+	}
+
+	deleted := 0
+
+	for i := 0; i < len(objects); i += maxkeys {
+		toDelete := objects[i:min(len(objects), i+maxkeys)] // min is required to avoid an out of bounds panic
+		res, err := client.DeleteObjects(ctx, &s3.DeleteObjectsInput{
+			Bucket: aws.String(bucket.GetBucketName()),
+			Delete: &types.Delete{
+				Objects: toDelete,
+			},
+		})
+		if err != nil {
+			return nil, parseAWSErrCode("error deleting S3 object(s)", err)
+		}
+		deleted += len(res.Deleted)
+	}
+
+	return &pb.DeleteObjectsResponse{
+		ObjectsDeleted: uint32(deleted),
+	}, nil
+}
+
 // dryRunValidation validates that the IAM role policy attached to the given secrets of the
 // storage bucket has the minimum required permissions needed for this plugin to function
 // as expected. This function will create an object in the s3 bucket to validate it has
@@ -672,13 +796,21 @@ func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attr
 		return parseAWSErrCode("error failed to get head object", err)
 	}
 
-	// attempt to delete the test object created for the dry run validation,
-	// this step is allowed to fail because DeleteObject is not a required
-	// operation for the plugin.
-	client.DeleteObject(ctx, &s3.DeleteObjectInput{
+	if res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+		Bucket: aws.String(bucket.GetBucketName()),
+		Prefix: aws.String(objectKey),
+	}); err != nil {
+		return parseAWSErrCode("error failed to list objects", err)
+	} else if res == nil || len(res.Contents) != 1 || *res.Contents[0].Key != objectKey {
+		return status.Errorf(codes.InvalidArgument, "list response did not contain the expected key: %+v", res)
+	}
+
+	if _, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
 		Bucket: aws.String(bucket.GetBucketName()),
 		Key:    aws.String(objectKey),
-	})
+	}); err != nil {
+		return parseAWSErrCode("error failed to delete object", err)
+	}
 
 	return nil
 }

plugin/service/storage/plugin_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/iam"
 	iamTypes "github.com/aws/aws-sdk-go-v2/service/iam/types"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
+	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/aws/aws-sdk-go-v2/service/sts"
 	"github.com/aws/aws-sdk-go-v2/service/sts/types"
 	"github.com/hashicorp/boundary-plugin-aws/internal/credential"
@@ -263,6 +264,15 @@ func TestStoragePlugin_OnCreateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -320,6 +330,15 @@ func TestStoragePlugin_OnCreateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -361,6 +380,15 @@ func TestStoragePlugin_OnCreateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -670,6 +698,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -879,6 +916,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -899,6 +945,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -977,6 +1032,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -1064,6 +1128,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -1107,6 +1180,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -1167,6 +1249,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -1211,6 +1302,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectError(fmt.Errorf("put object fail oops")),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -1256,6 +1356,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -1302,6 +1411,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -1396,6 +1514,15 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -2039,6 +2166,15 @@ func TestStoragePlugin_ValidatePermissions(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -2065,6 +2201,15 @@ func TestStoragePlugin_ValidatePermissions(t *testing.T) {
 						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
 						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
 						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+						testMockS3WithListObjectsV2OutputFunc(func(i *s3.ListObjectsV2Input) *s3.ListObjectsV2Output {
+							return &s3.ListObjectsV2Output{
+								Contents: []s3types.Object{
+									{
+										Key: i.Prefix,
+									},
+								},
+							}
+						}),
 					),
 				),
 			},
@@ -2735,3 +2880,358 @@ func TestStoragePlugin_PutObject(t *testing.T) {
 		})
 	}
 }
+
+func TestStoragePlugin_DeleteObjects(t *testing.T) {
+	validRequest := func() *pb.DeleteObjectsRequest {
+		return &pb.DeleteObjectsRequest{
+			Bucket: &storagebuckets.StorageBucket{
+				BucketName: "external-obj-store",
+				Secrets:    credential.MockStaticCredentialSecrets(),
+				Attributes: &structpb.Struct{
+					Fields: map[string]*structpb.Value{
+						credential.ConstRegion: structpb.NewStringValue("us-west-2"),
+					},
+				},
+			},
+			KeyPrefix: "abc/",
+		}
+	}
+
+	validRecursiveRequest := func() *pb.DeleteObjectsRequest {
+		return &pb.DeleteObjectsRequest{
+			Bucket: &storagebuckets.StorageBucket{
+				BucketName: "external-obj-store",
+				Secrets:    credential.MockStaticCredentialSecrets(),
+				Attributes: &structpb.Struct{
+					Fields: map[string]*structpb.Value{
+						credential.ConstRegion: structpb.NewStringValue("us-west-2"),
+					},
+				},
+			},
+			KeyPrefix: "abc/",
+			Recursive: true,
+		}
+	}
+
+	cases := []struct {
+		name                string
+		req                 *pb.DeleteObjectsRequest
+		credOpts            []credential.AwsCredentialPersistedStateOption
+		storageOpts         []awsStoragePersistedStateOption
+		expectedErrContains string
+		expectedErrCode     codes.Code
+		expected            uint32
+	}{
+		{
+			name:                "empty key prefix",
+			req:                 &pb.DeleteObjectsRequest{},
+			expectedErrContains: "key prefix is required",
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name: "nil bucket",
+			req: &pb.DeleteObjectsRequest{
+				KeyPrefix: "/foo/bar/key",
+			},
+			expectedErrContains: "bucket is required",
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name: "missing bucket name",
+			req: &pb.DeleteObjectsRequest{
+				KeyPrefix: "/foo/bar/key",
+				Bucket:    &storagebuckets.StorageBucket{},
+			},
+			expectedErrContains: "bucketName is required",
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name: "nil attributes",
+			req: &pb.DeleteObjectsRequest{
+				KeyPrefix: "/foo/bar/key",
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "foo",
+					Secrets:    new(structpb.Struct),
+				},
+			},
+			expectedErrContains: "attributes is required",
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name: "error reading attributes",
+			req: &pb.DeleteObjectsRequest{
+				KeyPrefix: "/foo/bar/key",
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "foo",
+					Secrets:    new(structpb.Struct),
+					Attributes: new(structpb.Struct),
+				},
+			},
+			expectedErrContains: "missing required value \"region\"",
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name: "invalid region",
+			req: &pb.DeleteObjectsRequest{
+				KeyPrefix: "/foo/bar/key",
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "foo",
+					Secrets:    new(structpb.Struct),
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							credential.ConstRegion: structpb.NewStringValue("foobar"),
+						},
+					},
+				},
+			},
+			expectedErrContains: "not a valid region: foobar",
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name: "credential persisted state setup error",
+			req:  validRequest(),
+			credOpts: []credential.AwsCredentialPersistedStateOption{
+				func(s *credential.AwsCredentialPersistedState) error {
+					return errors.New(testOptionErr)
+				},
+			},
+			expectedErrContains: fmt.Sprintf("error setting up persisted state: %s", testOptionErr),
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name: "storage persisted state setup error",
+			req:  validRequest(),
+			storageOpts: []awsStoragePersistedStateOption{
+				func(s *awsStoragePersistedState) error {
+					return errors.New(testOptionErr)
+				},
+			},
+			expectedErrContains: fmt.Sprintf("error setting up persisted state: %s", testOptionErr),
+			expectedErrCode:     codes.InvalidArgument,
+		},
+		{
+			name:     "DeleteObject error",
+			req:      validRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithDeleteObjectError(errors.New(testDeleteObjectErr)),
+					),
+				),
+			},
+			expectedErrContains: "error deleting S3 object: test error for DeleteObject",
+			expectedErrCode:     codes.Internal,
+		},
+		{
+			name:     "ListObjectV2 error",
+			req:      validRecursiveRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithListObjectsV2Error(errors.New(testListObjectV2Err)),
+					),
+				),
+			},
+			expectedErrContains: "error iterating S3 bucket contents: test error for ListObjectV2",
+			expectedErrCode:     codes.Internal,
+		},
+		{
+			name:     "DeleteObjects error",
+			req:      validRecursiveRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
+							IsTruncated: false,
+							Contents: []s3types.Object{
+								s3types.Object{
+									Key: aws.String("abc/abc"),
+								},
+							},
+						}),
+						testMockS3WithDeleteObjectsError(errors.New(testDeleteObjectsErr)),
+					),
+				),
+			},
+			expectedErrContains: "error deleting S3 object(s): test error for DeleteObjects",
+			expectedErrCode:     codes.Internal,
+		},
+		{
+			name:     "DeleteObject throttle error",
+			req:      validRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithDeleteObjectError(new(throttleErr)),
+					),
+				),
+			},
+			expectedErrContains: "error deleting S3 object: ThrottlingException",
+			expectedErrCode:     codes.Unavailable,
+		},
+		{
+			name:     "ListObjectsV2 throttle error",
+			req:      validRecursiveRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithListObjectsV2Error(new(throttleErr)),
+					),
+				),
+			},
+			expectedErrContains: "error iterating S3 bucket contents: ThrottlingException",
+			expectedErrCode:     codes.Unavailable,
+		},
+		{
+			name:     "DeleteObjects throttle error",
+			req:      validRecursiveRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
+							IsTruncated: false,
+							Contents: []s3types.Object{
+								s3types.Object{
+									Key: aws.String("abc/abc"),
+								},
+							},
+						}),
+						testMockS3WithDeleteObjectsError(new(throttleErr)),
+					),
+				),
+			},
+			expectedErrContains: "error deleting S3 object(s): ThrottlingException",
+			expectedErrCode:     codes.Unavailable,
+		},
+		{
+			name:     "success",
+			req:      validRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithDeleteObjectOutput(&s3.DeleteObjectOutput{}),
+					),
+				),
+			},
+			expected: 1,
+		},
+		{
+			name:     "recursive success",
+			req:      validRecursiveRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
+							IsTruncated: false,
+							Contents: []s3types.Object{
+								s3types.Object{
+									Key: aws.String("abc/abc1"),
+								},
+								s3types.Object{
+									Key: aws.String("abc/abc2"),
+								},
+								s3types.Object{
+									Key: aws.String("abc/abc3"),
+								},
+							},
+						}),
+						testMockS3WithDeleteObjectsOutput(&s3.DeleteObjectsOutput{
+							Deleted: []s3types.DeletedObject{
+								s3types.DeletedObject{},
+								s3types.DeletedObject{},
+								s3types.DeletedObject{},
+							},
+						}),
+					),
+				),
+			},
+			expected: 3,
+		},
+		{
+			name:     "recursive success empty",
+			req:      validRecursiveRequest(),
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
+							IsTruncated: false,
+							Contents:    []s3types.Object{},
+						}),
+						// no deleted response since it shouldn't be called
+					),
+				),
+			},
+			expected: 0,
+		},
+		{
+			name: "success with dynamic credentials",
+			req: &pb.DeleteObjectsRequest{
+				KeyPrefix: "/foo/bar/key",
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: "foo",
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							credential.ConstRegion:                    structpb.NewStringValue("us-west-2"),
+							credential.ConstRoleArn:                   structpb.NewStringValue("arn:aws:iam::123456789012:role/S3Access"),
+							credential.ConstDisableCredentialRotation: structpb.NewBoolValue(true),
+						},
+					},
+				},
+			},
+			credOpts: validSTSMock(),
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithDeleteObjectOutput(&s3.DeleteObjectOutput{}),
+					),
+				),
+			},
+			expected: 1,
+		},
+		// NOTE: there is no automated test for checking continuation tokens or multiple
+		// delete calls due to not having a stack for output responses. this was tested
+		// manually
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			require := require.New(t)
+			p := &StoragePlugin{
+				testCredStateOpts:    tc.credOpts,
+				testStorageStateOpts: tc.storageOpts,
+			}
+			res, err := p.DeleteObjects(context.Background(), tc.req)
+			if tc.expectedErrContains != "" {
+				require.Error(err)
+				require.Contains(err.Error(), tc.expectedErrContains)
+				require.Equal(status.Code(err).String(), tc.expectedErrCode.String())
+				return
+			}
+			require.NoError(err)
+
+			require.NoError(err)
+			require.NotNil(res)
+			require.Equal(tc.expected, res.ObjectsDeleted)
+		})
+	}
+}

plugin/service/storage/state.go

@@ -25,6 +25,8 @@ type S3API interface {
 	GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error)
 	HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error)
 	PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error)
+	DeleteObjects(ctx context.Context, params *s3.DeleteObjectsInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectsOutput, error)
+	ListObjectsV2(ctx context.Context, params *s3.ListObjectsV2Input, optFns ...func(*s3.Options)) (*s3.ListObjectsV2Output, error)
 }
 
 var customClient = &http.Client{

plugin/service/storage/testing.go

@@ -31,6 +31,9 @@ const (
 	testGetObjectErr         = "test error for GetObject"
 	testPutObjectErr         = "test error for PutObject"
 	testHeadObjectErr        = "test error for HeadObject"
+	testListObjectV2Err      = "test error for ListObjectV2"
+	testDeleteObjectErr      = "test error for DeleteObject"
+	testDeleteObjectsErr     = "test error for DeleteObjects"
 )
 
 // throttleErr is a mocked error used for testing the aws s3 client
@@ -54,6 +57,14 @@ type testMockS3State struct {
 
 	HeadObjectCalled      bool
 	HeadObjectInputParams *s3.HeadObjectInput
+
+	ListObjectsV2Called      bool
+	ListObjectsV2InputParams *s3.ListObjectsV2Input
+
+	DeleteObjectCalled       bool
+	DeleteObjectInputParams  *s3.DeleteObjectInput
+	DeleteObjectsCalled      bool
+	DeleteObjectsInputParams *s3.DeleteObjectsInput
 }
 
 func (s *testMockS3State) Reset() {
@@ -64,6 +75,12 @@ func (s *testMockS3State) Reset() {
 	s.PutObjectBody = nil
 	s.HeadObjectCalled = false
 	s.HeadObjectInputParams = nil
+	s.ListObjectsV2Called = false
+	s.ListObjectsV2InputParams = nil
+	s.DeleteObjectCalled = false
+	s.DeleteObjectInputParams = nil
+	s.DeleteObjectsCalled = false
+	s.DeleteObjectsInputParams = nil
 }
 
 type testMockS3 struct {
@@ -83,6 +100,18 @@ type testMockS3 struct {
 	// mocked responses for headObject
 	HeadObjectOutput *s3.HeadObjectOutput
 	HeadObjectErr    error
+
+	// mocked responses for ListObjectsV2 (needed for delete)
+	ListObjectsV2Output *s3.ListObjectsV2Output
+	ListObjectsV2Err    error
+	// sometimes we need to create a wildcard output
+	ListObjectsV2OutputFunc func(*s3.ListObjectsV2Input) *s3.ListObjectsV2Output
+
+	// mocked responses for DeleteObject(s)
+	DeleteObjectOutput  *s3.DeleteObjectOutput
+	DeleteObjectErr     error
+	DeleteObjectsOutput *s3.DeleteObjectsOutput
+	DeleteObjectsErr    error
 }
 
 type testMockS3Option func(m *testMockS3) error
@@ -129,6 +158,55 @@ func testMockS3WithHeadObjectError(e error) testMockS3Option {
 	}
 }
 
+func testMockS3WithListObjectsV2Output(o *s3.ListObjectsV2Output) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.ListObjectsV2Output = o
+		return nil
+	}
+}
+
+func testMockS3WithListObjectsV2Error(e error) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.ListObjectsV2Err = e
+		return nil
+	}
+}
+
+func testMockS3WithListObjectsV2OutputFunc(o func(*s3.ListObjectsV2Input) *s3.ListObjectsV2Output) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.ListObjectsV2OutputFunc = o
+		return nil
+	}
+}
+
+func testMockS3WithDeleteObjectOutput(o *s3.DeleteObjectOutput) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.DeleteObjectOutput = o
+		return nil
+	}
+}
+
+func testMockS3WithDeleteObjectError(e error) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.DeleteObjectErr = e
+		return nil
+	}
+}
+
+func testMockS3WithDeleteObjectsOutput(o *s3.DeleteObjectsOutput) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.DeleteObjectsOutput = o
+		return nil
+	}
+}
+
+func testMockS3WithDeleteObjectsError(e error) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.DeleteObjectsErr = e
+		return nil
+	}
+}
+
 func newTestMockS3(state *testMockS3State, opts ...testMockS3Option) s3APIFunc {
 	return func(cfgs ...aws.Config) (S3API, error) {
 		m := &testMockS3{
@@ -193,8 +271,47 @@ func (m *testMockS3) PutObject(ctx context.Context, params *s3.PutObjectInput, o
 	return m.PutObjectOutput, nil
 }
 
+func (m *testMockS3) ListObjectsV2(ctx context.Context, params *s3.ListObjectsV2Input, optFns ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) {
+	if m.State != nil {
+		m.State.ListObjectsV2Called = true
+		m.State.ListObjectsV2InputParams = params
+	}
+
+	if m.ListObjectsV2Err != nil {
+		return nil, m.ListObjectsV2Err
+	}
+
+	if m.ListObjectsV2OutputFunc != nil {
+		return m.ListObjectsV2OutputFunc(params), nil
+	}
+
+	return m.ListObjectsV2Output, nil
+}
+
 func (m *testMockS3) DeleteObject(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) {
-	return &s3.DeleteObjectOutput{}, nil
+	if m.State != nil {
+		m.State.DeleteObjectCalled = true
+		m.State.DeleteObjectInputParams = params
+	}
+
+	if m.DeleteObjectErr != nil {
+		return nil, m.DeleteObjectErr
+	}
+
+	return m.DeleteObjectOutput, nil
+}
+
+func (m *testMockS3) DeleteObjects(ctx context.Context, params *s3.DeleteObjectsInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectsOutput, error) {
+	if m.State != nil {
+		m.State.DeleteObjectsCalled = true
+		m.State.DeleteObjectsInputParams = params
+	}
+
+	if m.DeleteObjectsErr != nil {
+		return nil, m.DeleteObjectsErr
+	}
+
+	return m.DeleteObjectsOutput, nil
 }
 
 func (m *testMockS3) HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error) {