chore: bump deps (#56)

This reverts commit 899c62ce06.

.gitignore

@@ -8,3 +8,5 @@ plan.tfplan
 # IntelliJ IDEA project files
 .idea
 *.iprg
+
+.DS_Store

CODEOWNERS

@@ -1,5 +1,4 @@
 # These owners will be the default owners for everything in
 # the repo, unless a later match takes precedence.
 
-@hashicorp/boundary
-
+* @hashicorp/boundary

go.mod

@@ -1,6 +1,6 @@
 module github.com/hashicorp/boundary-plugin-aws
 
-go 1.21
+go 1.23.3
 
 require (
 	github.com/aws/aws-sdk-go-v2 v1.20.1
@@ -11,14 +11,14 @@ require (
 	github.com/aws/smithy-go v1.14.1
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.4.0
-	github.com/hashicorp/boundary/sdk v0.0.43-0.20240717182311-a20aae98794a
+	github.com/hashicorp/boundary/sdk v0.0.49
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-secure-stdlib/awsutil/v2 v2.0.0
 	github.com/hashicorp/terraform-json v0.14.0
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/stretchr/testify v1.8.4
 	google.golang.org/grpc v1.61.0
-	google.golang.org/protobuf v1.35.1
+	google.golang.org/protobuf v1.33.0
 )
 
 require (
@@ -55,10 +55,10 @@ require (
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/zclconf/go-cty v1.10.0 // indirect
-	golang.org/x/crypto v0.28.0 // indirect
-	golang.org/x/net v0.30.0 // indirect
-	golang.org/x/sys v0.26.0 // indirect
-	golang.org/x/text v0.19.0 // indirect
+	golang.org/x/crypto v0.18.0 // indirect
+	golang.org/x/net v0.20.0 // indirect
+	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/text v0.14.0 // indirect
 	google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240125205218-1f4bbc51befe // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240125205218-1f4bbc51befe // indirect

go.sum

@@ -78,8 +78,8 @@ github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/uuid v1.4.0 h1:MtMxsa51/r9yyhkyLsVeVt0B+BGQZzpQiTQ4eHZ8bc4=
 github.com/google/uuid v1.4.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/hashicorp/boundary/sdk v0.0.43-0.20240717182311-a20aae98794a h1:SwVze6sYE5o+J9qdcgj29auY1k6O8oSo9DC2+2Gb9rw=
-github.com/hashicorp/boundary/sdk v0.0.43-0.20240717182311-a20aae98794a/go.mod h1:9iOT7kDM6mYcSkKxNuZlv8rP7U5BG1kXoevjLLL8lNQ=
+github.com/hashicorp/boundary/sdk v0.0.49 h1:XOb6mSKyrU/wI20+5xTYBHQUP7eIeKcLxKSCpCs4yzM=
+github.com/hashicorp/boundary/sdk v0.0.49/go.mod h1:IHP79to8aIi22FiY58pgBqJL96/U9D8ZAUhS2DdC+Us=
 github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
 github.com/hashicorp/errwrap v1.1.0 h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=
 github.com/hashicorp/errwrap v1.1.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
@@ -210,8 +210,8 @@ go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
-golang.org/x/crypto v0.28.0 h1:GBDwsMXVQi34v5CCYUm2jkJvu4cbtru2U4TN2PSyQnw=
-golang.org/x/crypto v0.28.0/go.mod h1:rmgy+3RHxRZMyY0jjAJShp2zgEdOqj2AO7U0pYmeQ7U=
+golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
+golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/lint v0.0.0-20200302205851-738671d3881b/go.mod h1:3xt1FjdF8hUf6vQPIChWIBhFzV8gjjsPE/fR3IyQdNY=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
@@ -224,8 +224,8 @@ golang.org/x/net v0.0.0-20190603091049-60506f45cf65/go.mod h1:HSz+uSET+XFnRR8LxR
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200301022130-244492dfa37a/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.30.0 h1:AcW1SDZMkb8IpzCdQUaIq2sP4sZ4zw+55h6ynffypl4=
-golang.org/x/net v0.30.0/go.mod h1:2wGyMJ5iFasEhkwi13ChkO/t1ECNC4X4eBKkVFyYFlU=
+golang.org/x/net v0.20.0 h1:aCL9BSgETF1k+blQaYUBx9hJ9LOGP3gAVemcZlf1Kpo=
+golang.org/x/net v0.20.0/go.mod h1:z8BVo6PvndSri0LbOE3hAn0apkU+1YvI6E70E9jsnvY=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -239,14 +239,14 @@ golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220503163025-988cb79eb6c6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.26.0 h1:KHjCJyddX0LoSTb3J+vWpupP9p0oznkqVk/IfjymZbo=
-golang.org/x/sys v0.26.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
+golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
-golang.org/x/text v0.19.0 h1:kTxAhCbGbxhK0IwgSKiMO5awPoDQ0RpfiVYBfK860YM=
-golang.org/x/text v0.19.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
+golang.org/x/text v0.14.0 h1:ScX5w1eTa3QqT8oi6+ziP7dTV1S2+ALU0bI+0zXKWiQ=
+golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
 golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
@@ -269,8 +269,8 @@ google.golang.org/grpc v1.61.0 h1:TOvOcuXn30kRao+gfcvsebNEa5iZIiLkisYEkf7R7o0=
 google.golang.org/grpc v1.61.0/go.mod h1:VUbo7IFqmF1QtCAstipjG0GIoq49KvMe9+h1jFLBNJs=
 google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
 google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
-google.golang.org/protobuf v1.35.1 h1:m3LfL6/Ca+fqnjnlqQXNpFPABW1UD7mjh8KO2mKFytA=
-google.golang.org/protobuf v1.35.1/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE=
+google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=
+google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=

plugin/plugin.go

@@ -32,6 +32,6 @@ type AwsPlugin struct {
 func NewAwsPlugin() *AwsPlugin {
 	return &AwsPlugin{
 		HostPlugin:    &host.HostPlugin{},
-		StoragePlugin: &storage.StoragePlugin{},
+		StoragePlugin: storage.New(),
 	}
 }

plugin/service/storage/plugin.go

@@ -14,6 +14,7 @@ import (
 	"os"
 	"path"
 	"strings"
+	"sync"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -32,6 +33,14 @@ import (
 // Ensure that we are implementing StoragePluginServiceServer
 var _ pb.StoragePluginServiceServer = (*StoragePlugin)(nil)
 
+// clientCache caches previously initialized S3API clients created with a RoleARN attribute.
+type clientCache struct {
+	// cache is a map of cached s3 clients. The key of the map
+	// is the public id of the storage bucket
+	cache map[string]S3API
+	sync.RWMutex
+}
+
 // StoragePlugin implements the StoragePluginServiceServer interface for the
 // AWS storage service plugin.
 type StoragePlugin struct {
@@ -42,6 +51,112 @@ type StoragePlugin struct {
 
 	// testStorageStateOpts are passed in to the stored state to control test behavior
 	testStorageStateOpts []awsStoragePersistedStateOption
+
+	clients *clientCache
+}
+
+// New creates a new StoragePlugin
+func New() *StoragePlugin {
+	c := &clientCache{
+		cache: make(map[string]S3API),
+	}
+	return &StoragePlugin{
+		clients: c,
+	}
+}
+
+// getClient returns an S3API client for the given storage bucket id.
+func (p *StoragePlugin) getClient(ctx context.Context,
+	storageBucketId string,
+	storageState *awsStoragePersistedState,
+	opt ...s3Option) (S3API, error) {
+	if storageBucketId == "" {
+		// No storage bucket ID to key cache on, create new client and return
+		client, err := storageState.s3Client(ctx, opt...)
+		if err != nil {
+			return nil, errors.BadRequestStatus("error creating S3 client: %s", err)
+		}
+		return client, nil
+	}
+
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing options in cache: %w", err)
+	}
+
+	if opts.withCacheRefresh {
+		// Forced cache refresh
+		p.clients.Lock()
+		defer p.clients.Unlock()
+		client, err := storageState.s3Client(ctx, opt...)
+		if err != nil {
+			return nil, errors.BadRequestStatus("error creating S3 client: %s", err)
+		}
+		p.clients.cache[storageBucketId] = client
+		return client, nil
+	}
+
+	p.clients.RLock()
+	client, ok := p.clients.cache[storageBucketId]
+	p.clients.RUnlock()
+
+	if !ok || client.Credentials().Expired() {
+		// We got a cache miss or the credentials has expired, time to refresh
+		p.clients.Lock()
+		defer p.clients.Unlock()
+
+		// Check cache again in case another caller updated it since we got the lock
+		client, ok = p.clients.cache[storageBucketId]
+		if ok && !client.Credentials().Expired() {
+			// Got a cache hit with valid credentials, return cached client
+			return client, nil
+		}
+
+		// Create new client and cache it
+		var err error
+		client, err = storageState.s3Client(ctx, opt...)
+		if err != nil {
+			return nil, errors.BadRequestStatus("error creating S3 client: %s", err)
+		}
+		p.clients.cache[storageBucketId] = client
+	}
+	return client, nil
+}
+
+type s3Caller func(client S3API) (any, error)
+
+// call gets the s3client and passes it to the s3caller function, if a permission error is returned
+// it forces a cache refresh and tries again.
+func (p *StoragePlugin) call(
+	ctx context.Context,
+	fn s3Caller,
+	storageBucketId string,
+	storageState *awsStoragePersistedState,
+	opts ...s3Option) (any, error) {
+
+	s3Client, err := p.getClient(ctx, storageBucketId, storageState, opts...)
+	if err != nil {
+		return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
+	}
+	firstAttempt := true
+	for {
+		resp, err := fn(s3Client)
+		if err != nil {
+			st, _ := errors.ParseAWSError("", err)
+			if st.Code() == codes.PermissionDenied && firstAttempt {
+				// We got a permission error, this is the first time so refresh cache and try again
+				firstAttempt = false
+				opts = append(opts, WithCacheRefresh(true))
+				s3Client, err = p.getClient(ctx, storageBucketId, storageState, opts...)
+				if err != nil {
+					return nil, errors.BadRequestStatus("error refreshing S3 client: %s", err)
+				}
+				continue
+			}
+			return nil, err
+		}
+		return resp, nil
+	}
 }
 
 // OnCreateStorageBucket is called when a storage bucket is created.
@@ -346,19 +461,20 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
 	if storageAttributes.EndpointUrl != "" {
 		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
 	}
-	s3Client, err := storageState.S3Client(ctx, opts...)
-	if err != nil {
-		return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
-	}
 
 	objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
-	resp, err := s3Client.HeadObject(ctx, &s3.HeadObjectInput{
-		Bucket: aws.String(bucket.GetBucketName()),
-		Key:    aws.String(objectKey),
-	})
+	headCall := func(s3Client S3API) (any, error) {
+		return s3Client.HeadObject(ctx, &s3.HeadObjectInput{
+			Bucket: aws.String(bucket.GetBucketName()),
+			Key:    aws.String(objectKey),
+		})
+	}
+	headResp, err := p.call(ctx, headCall, bucket.GetId(), storageState, opts...)
 	if err != nil {
 		return nil, parseS3Error("head object", err, req).Err()
 	}
+	resp := headResp.(*s3.HeadObjectOutput)
+
 	return &pb.HeadObjectResponse{
 		ContentLength: resp.ContentLength,
 		LastModified:  timestamppb.New(*resp.LastModified),
@@ -472,19 +588,19 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
 	if storageAttributes.EndpointUrl != "" {
 		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
 	}
-	s3Client, err := storageState.S3Client(stream.Context(), opts...)
-	if err != nil {
-		return errors.BadRequestStatus("error getting S3 client: %s", err)
-	}
 
 	objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
-	resp, err := s3Client.GetObject(stream.Context(), &s3.GetObjectInput{
-		Bucket: aws.String(bucket.GetBucketName()),
-		Key:    aws.String(objectKey),
-	})
+	getCall := func(s3Client S3API) (any, error) {
+		return s3Client.GetObject(stream.Context(), &s3.GetObjectInput{
+			Bucket: aws.String(bucket.GetBucketName()),
+			Key:    aws.String(objectKey),
+		})
+	}
+	getResp, err := p.call(stream.Context(), getCall, bucket.GetId(), storageState, opts...)
 	if err != nil {
 		return parseS3Error("get object", err, req).Err()
 	}
+	resp := getResp.(*s3.GetObjectOutput)
 
 	defer resp.Body.Close()
 	reader := bufio.NewReader(resp.Body)
@@ -583,10 +699,6 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
 	if storageAttributes.EndpointUrl != "" {
 		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
 	}
-	s3Client, err := storageState.S3Client(ctx, opts...)
-	if err != nil {
-		return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
-	}
 
 	hash := sha256.New()
 	if _, err := io.Copy(hash, file); err != nil {
@@ -598,18 +710,23 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
 	if err != nil {
 		return nil, errors.UnknownStatus("failed to rewind file pointer")
 	}
-
 	objectKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
-	resp, err := s3Client.PutObject(ctx, &s3.PutObjectInput{
-		Bucket:            aws.String(bucket.GetBucketName()),
-		Key:               aws.String(objectKey),
-		Body:              file,
-		ChecksumAlgorithm: types.ChecksumAlgorithmSha256,
-		ChecksumSHA256:    aws.String(checksum),
-	})
+
+	putCall := func(s3Client S3API) (any, error) {
+		return s3Client.PutObject(ctx, &s3.PutObjectInput{
+			Bucket:            aws.String(bucket.GetBucketName()),
+			Key:               aws.String(objectKey),
+			Body:              file,
+			ChecksumAlgorithm: types.ChecksumAlgorithmSha256,
+			ChecksumSHA256:    aws.String(checksum),
+		})
+	}
+	putResp, err := p.call(ctx, putCall, bucket.GetId(), storageState, opts...)
 	if err != nil {
 		return nil, parseS3Error("put object", err, req).Err()
 	}
+	resp := putResp.(*s3.PutObjectOutput)
+
 	if resp.ChecksumSHA256 == nil {
 		return nil, errors.UnknownStatus("missing checksum response from aws")
 	}
@@ -678,10 +795,6 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
 	if storageAttributes.EndpointUrl != "" {
 		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
 	}
-	client, err := storageState.S3Client(ctx, opts...)
-	if err != nil {
-		return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
-	}
 
 	prefix := path.Join(bucket.GetBucketPrefix(), req.GetKeyPrefix())
 	if strings.HasSuffix(req.GetKeyPrefix(), "/") {
@@ -691,10 +804,17 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
 	}
 
 	if !req.Recursive {
-		_, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
-			Bucket: aws.String(bucket.GetBucketName()),
-			Key:    aws.String(prefix),
-		})
+		deleteCall := func(s3Client S3API) (any, error) {
+			return s3Client.DeleteObject(ctx, &s3.DeleteObjectInput{
+				Bucket: aws.String(bucket.GetBucketName()),
+				Key:    aws.String(prefix),
+			})
+		}
+		_, err := p.call(ctx, deleteCall, bucket.GetId(), storageState, opts...)
+		if err != nil {
+			return nil, parseS3Error("delete object", err, req).Err()
+		}
+
 		if err != nil {
 			return nil, parseS3Error("delete object", err, req).Err()
 		}
@@ -704,10 +824,13 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
 	}
 
 	const maxkeys = 1000
-
 	objects := []types.ObjectIdentifier{}
+	client, err := p.getClient(ctx, bucket.GetId(), storageState, opts...)
+	if err != nil {
+		return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
+	}
 	var conToken *string
-	truncated := true
+	truncated, firstAttempt := true, true
 	for truncated {
 		res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
 			Bucket:            aws.String(bucket.GetBucketName()),
@@ -716,6 +839,18 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
 			ContinuationToken: conToken,
 		})
 		if err != nil {
+			st, _ := errors.ParseAWSError("", err)
+			if st.Code() == codes.PermissionDenied && firstAttempt {
+				// This is first attempt and we get a permission error, refresh cache and try again
+				firstAttempt = false
+				opts = append(opts, WithCacheRefresh(true))
+				client, err = p.getClient(ctx, bucket.GetId(), storageState, opts...)
+				if err != nil {
+					return nil, errors.BadRequestStatus("error refreshing S3 client: %s", err)
+				}
+				continue
+			}
+
 			return nil, parseS3Error("list objects", err, req).Err()
 		}
 		truncated = res.IsTruncated
@@ -770,7 +905,7 @@ func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attr
 		opts = append(opts, WithEndpoint(attrs.EndpointUrl))
 	}
 
-	client, err := state.S3Client(ctx, opts...)
+	client, err := state.s3Client(ctx, opts...)
 	if err != nil {
 		return status.New(codes.InvalidArgument, fmt.Sprintf("error getting S3 client: %s", err))
 	}

plugin/service/storage/plugin_test.go

@@ -3917,3 +3917,89 @@ func TestStoragePlugin_DeleteObjects(t *testing.T) {
 		})
 	}
 }
+
+func TestS3ClientCache(t *testing.T) {
+	ctx := context.Background()
+	assert, require := assert.New(t), require.New(t)
+
+	cred1, err := awsutil.NewCredentialsConfig(
+		awsutil.WithRegion("us-east-1"),
+		awsutil.WithAccessKey("cred1"),
+		awsutil.WithSecretKey("secret1"),
+	)
+	require.NoError(err)
+	cred2, err := awsutil.NewCredentialsConfig(
+		awsutil.WithRegion("us-east-1"),
+		awsutil.WithAccessKey("cred2"),
+		awsutil.WithSecretKey("secret2"),
+	)
+	require.NoError(err)
+	cred3, err := awsutil.NewCredentialsConfig(
+		awsutil.WithRegion("us-east-1"),
+		awsutil.WithAccessKey("cred3"),
+		awsutil.WithSecretKey("secret3"),
+	)
+	require.NoError(err)
+
+	p := New()
+	cred, err := credential.NewAwsCredentialPersistedState(
+		credential.WithCredentialsConfig(cred1),
+	)
+	require.NoError(err)
+
+	state, err := newAwsStoragePersistedState(withCredentials(cred))
+	require.NoError(err)
+
+	// Cache should be empty before first call
+	assert.Empty(p.clients.cache)
+
+	client, err := p.getClient(ctx, "id", state)
+	require.NoError(err)
+	require.NotNil(client)
+	assert.Equal(cred1.AccessKey, client.Credentials().AccessKeyID)
+	assert.Equal(cred1.SecretKey, client.Credentials().SecretAccessKey)
+
+	// Update state with new credentials
+	cred, err = credential.NewAwsCredentialPersistedState(
+		credential.WithCredentialsConfig(cred2),
+	)
+	require.NoError(err)
+	state, err = newAwsStoragePersistedState(withCredentials(cred))
+	require.NoError(err)
+
+	// Client is cached and creds valid so it should still grab initial client
+	client, err = p.getClient(ctx, "id", state)
+	require.NoError(err)
+	require.NotNil(client)
+	assert.Equal(cred1.AccessKey, client.Credentials().AccessKeyID)
+	assert.Equal(cred1.SecretKey, client.Credentials().SecretAccessKey)
+
+	// Force credentials to expire
+	c := client.(*s3Client)
+	c.creds.CanExpire = true
+	p.clients.Lock()
+	p.clients.cache["id"] = c
+	p.clients.Unlock()
+
+	// Getting client should now create new client on cred2 since the credentials have expired
+	client, err = p.getClient(ctx, "id", state)
+	require.NoError(err)
+	require.NotNil(client)
+	assert.Equal(cred2.AccessKey, client.Credentials().AccessKeyID)
+	assert.Equal(cred2.SecretKey, client.Credentials().SecretAccessKey)
+
+	// Update state with new credentials again
+	cred, err = credential.NewAwsCredentialPersistedState(
+		credential.WithCredentialsConfig(cred3),
+	)
+	require.NoError(err)
+	state, err = newAwsStoragePersistedState(withCredentials(cred))
+	require.NoError(err)
+
+	// Getting client and force cache refresh should return cred3 based client
+	client, err = p.getClient(ctx, "id", state, WithCacheRefresh(true))
+	require.NoError(err)
+	require.NotNil(client)
+	assert.Equal(cred3.AccessKey, client.Credentials().AccessKeyID)
+	assert.Equal(cred3.SecretKey, client.Credentials().SecretAccessKey)
+}

plugin/service/storage/state.go

@@ -27,6 +27,16 @@ type S3API interface {
 	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)
+	Credentials() aws.Credentials
+}
+
+type s3Client struct {
+	*s3.Client
+	creds aws.Credentials
+}
+
+func (c *s3Client) Credentials() aws.Credentials {
+	return c.creds
 }
 
 var customClient = &http.Client{
@@ -91,9 +101,9 @@ func (s *awsStoragePersistedState) toProto() (*storagebuckets.StorageBucketPersi
 	return &storagebuckets.StorageBucketPersisted{Data: data}, nil
 }
 
-// S3Client returns a configured S3 client based on the session
+// s3Client returns a configured S3 client based on the session
 // information stored in the state.
-func (s *awsStoragePersistedState) S3Client(ctx context.Context, opt ...s3Option) (S3API, error) {
+func (s *awsStoragePersistedState) s3Client(ctx context.Context, opt ...s3Option) (S3API, error) {
 	opts, err := getOpts(opt...)
 	if err != nil {
 		return nil, fmt.Errorf("error parsing options when fetching S3 client: %w", err)
@@ -114,7 +124,16 @@ func (s *awsStoragePersistedState) S3Client(ctx context.Context, opt ...s3Option
 	if s.testS3APIFunc != nil {
 		return s.testS3APIFunc(*awsCfg)
 	}
-	return s3.NewFromConfig(*awsCfg, s3Opts...), nil
+
+	// Retrieve the credentials provider from the client
+	credsProvider := awsCfg.Credentials
+	creds, err := credsProvider.Retrieve(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to retrieve credentials: %v\n", err)
+	}
+
+	c := s3.NewFromConfig(*awsCfg, s3Opts...)
+	return &s3Client{Client: c, creds: creds}, nil
 }
 
 // getOpts iterates the inbound s3Options and returns a struct
@@ -136,13 +155,22 @@ type s3Option func(*s3Options) error
 
 // options = how options are represented
 type s3Options struct {
-	withEndpoint string
+	withEndpoint     string
+	withCacheRefresh bool
 }
 
 func getDefaultS3Options() s3Options {
 	return s3Options{}
 }
 
+// WithCacheRefresh controls if the cache should be forced to refresh
+func WithCacheRefresh(refresh bool) s3Option {
+	return func(o *s3Options) error {
+		o.withCacheRefresh = refresh
+		return nil
+	}
+}
+
 // WithEndpoint contains the endpoint to use
 func WithEndpoint(with string) s3Option {
 	return func(o *s3Options) error {

plugin/service/storage/state_test.go

@@ -64,7 +64,7 @@ func TestAwsStoragePersistedStateS3Client(t *testing.T) {
 			)
 			require.NoError(err)
 
-			clientRaw, err := state.S3Client(context.Background(), tc.s3Opts...)
+			clientRaw, err := state.s3Client(context.Background(), tc.s3Opts...)
 			require.NoError(err)
 			require.NotNil(clientRaw)
 			client, ok := clientRaw.(*testMockS3)