feat(plugin): add support for storage service

plugin/plugin.go

@@ -2,14 +2,17 @@ package plugin
 
 import (
 	"github.com/hashicorp/boundary-plugin-host-aws/plugin/service/host"
+	"github.com/hashicorp/boundary-plugin-host-aws/plugin/service/storage"
 	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
 )
 
 // Ensure that AWSPlugin implementings the following services:
 //
 //	HostPluginServiceServer
+//	StoragePluginServiceServer
 var (
-	_ pb.HostPluginServiceServer = (*host.HostPlugin)(nil)
+	_ pb.HostPluginServiceServer    = (*host.HostPlugin)(nil)
+	_ pb.StoragePluginServiceServer = (*storage.StoragePlugin)(nil)
 )
 
 // AwsPlugin contains a collection of all aws plugin services.
@@ -17,4 +20,15 @@ type AwsPlugin struct {
 	// HostPlugin implements the HostPluginServiceServer interface for
 	// supporting dynamically sourcing hosts from Amazon EC2
 	*host.HostPlugin
+
+	// StoragePlugin implements the StoragePluginServiceServer interface for
+	// supporting storing and retrieving BSR files from Amazon S3
+	*storage.StoragePlugin
+}
+
+func NewAwsPlugin() *AwsPlugin {
+	return &AwsPlugin{
+		HostPlugin:    &host.HostPlugin{},
+		StoragePlugin: &storage.StoragePlugin{},
+	}
 }

plugin/service/storage/README.md

@@ -0,0 +1,83 @@
+## Getting Started
+
+To create a storage bucket (using default scope created by `boundary dev`):
+
+```
+boundary storage-buckets create \
+ -scope-id p_1234567890 \
+ -name "Example Plugin-Based Storage Bucket" \
+ -description "Description for plugin-based storage bucket" \
+ -plugin-name aws \
+ -bucket-name="session_recording_storage" \
+ -bucket-prefix="foo/bar/zoo" \
+ -worker-filter ‘“aws-access” in “/tags/type”’ \
+ -attr region=REGION \
+ -attr endpoint_url=0.0.0.0 \
+ -secret access_key_id='KEY' \
+ -secret secret_access_key='SECRET'
+```
+
+### Required IAM Privileges
+
+The following IAM privileges, at the very least, are required to be attached to
+a configured IAM user for this provider:
+
+Example policy:
+
+```
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Action": [
+        "s3:PutObject",
+        "s3:GetObject",
+        "s3:GetObjectAttributes"
+      ],
+      "Effect": "Allow",
+      "Resource": "arn:aws:s3:::session_recording_storage/foo/bar/zoo/*"
+    }
+  ]
+}
+```
+
+* `iam:GetUser`, `iam:CreateAccessKey`, and `iam:DeleteAccessKey`, configured to
+  the IAM user to allow credential rotation. Example policy:
+
+```
+{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Action": [
+        "iam:DeleteAccessKey",
+        "iam:GetUser",
+        "iam:CreateAccessKey"
+      ],
+      "Effect": "Allow",
+      "Resource": "arn:aws:iam::123456789012:user/JohnDoe"
+    }
+  ]
+}
+```
+
+### Attributes
+
+The following `attributes` are valid on an AWS storage bucket resource:
+
+- `disable_credential_rotation` (bool): If `true`, credential rotation will not
+  be performed. See the [Credential Rotation](#credential-rotation) section
+  above.
+- `region` (string): The region to configure the storage bucket for.
+- `endpoint-url` (string): The endpoint to configure the storage.
+
+An example of how to utilize the [endpoint attribute](https://docs.aws.amazon.com/vpc/latest/privatelink/vpc-endpoints-s3.html)
+
+### Secrets
+
+The following `secrets` are valid on an AWS storage bucket resource:
+
+- `access_key_id` (string): The access key ID for the IAM user to use with this
+  storage bucket.
+- `secret_access_key` (string): The secret access key for the IAM user to use
+  with this storage bucket.

plugin/service/storage/attributes.go

@@ -0,0 +1,57 @@
+package storage
+
+import (
+	"fmt"
+
+	cred "github.com/hashicorp/boundary-plugin-host-aws/internal/credential"
+	"github.com/hashicorp/boundary-plugin-host-aws/internal/errors"
+	"github.com/hashicorp/boundary-plugin-host-aws/internal/values"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+// StorageAttributes is a Go-native representation of the Attributes
+// map.
+type StorageAttributes struct {
+	*cred.CredentialAttributes
+
+	// EndpointUrl is used for configuring how the aws client will resolve requests.
+	EndpointUrl string
+}
+
+func getStorageAttributes(in *structpb.Struct) (*StorageAttributes, error) {
+	unknownFields := values.StructFields(in)
+	badFields := make(map[string]string)
+
+	var err error
+	credAttributes, err := cred.GetCredentialAttributes(in)
+	if err != nil {
+		return nil, err
+	}
+
+	endpointUrl, err := values.GetStringValue(in, ConstAwsEndpointUrl, false)
+	if err != nil {
+		badFields[fmt.Sprintf("attributes.%s", ConstAwsEndpointUrl)] = err.Error()
+	}
+	delete(unknownFields, ConstAwsEndpointUrl)
+
+	for s := range unknownFields {
+		switch s {
+		// Ignore knownFields from CredentialAttributes
+		case cred.ConstRegion:
+			continue
+		case cred.ConstDisableCredentialRotation:
+			continue
+		default:
+			badFields[fmt.Sprintf("attributes.%s", s)] = "unrecognized field"
+		}
+	}
+
+	if len(badFields) > 0 {
+		return nil, errors.InvalidArgumentError("Invalid arguments in storage bucket attributes", badFields)
+	}
+
+	return &StorageAttributes{
+		CredentialAttributes: credAttributes,
+		EndpointUrl:          endpointUrl,
+	}, nil
+}

plugin/service/storage/attributes_test.go

@@ -0,0 +1,80 @@
+package storage
+
+import (
+	"testing"
+
+	cred "github.com/hashicorp/boundary-plugin-host-aws/internal/credential"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+func TestGetStorageAttributes(t *testing.T) {
+	cases := []struct {
+		name                string
+		in                  map[string]any
+		expected            *StorageAttributes
+		expectedErrContains string
+	}{
+		{
+			name:                "missing region",
+			in:                  map[string]any{},
+			expectedErrContains: "missing required value \"region\"",
+		},
+		{
+			name: "unknown fields",
+			in: map[string]any{
+				"region": "us-west-2",
+				"foo":    true,
+				"bar":    true,
+			},
+			expectedErrContains: "attributes.bar: unrecognized field, attributes.foo: unrecognized field",
+		},
+		{
+			name: "good w/ endpoint",
+			in: map[string]any{
+				cred.ConstRegion:    "us-west-2",
+				ConstAwsEndpointUrl: "0.0.0.0",
+			},
+			expected: &StorageAttributes{
+				CredentialAttributes: &cred.CredentialAttributes{
+					Region: "us-west-2",
+				},
+				EndpointUrl: "0.0.0.0",
+			},
+		},
+		{
+			name: "good w/o endpoint",
+			in: map[string]any{
+				cred.ConstRegion: "us-west-2",
+			},
+			expected: &StorageAttributes{
+				CredentialAttributes: &cred.CredentialAttributes{
+					Region: "us-west-2",
+				},
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			require := require.New(t)
+
+			input, err := structpb.NewStruct(tc.in)
+			require.NoError(err)
+
+			actual, err := getStorageAttributes(input)
+			if tc.expectedErrContains != "" {
+				require.Error(err)
+				require.Contains(err.Error(), tc.expectedErrContains)
+				require.Equal(status.Code(err), codes.InvalidArgument)
+				return
+			}
+
+			require.NoError(err)
+			require.Equal(tc.expected, actual)
+		})
+	}
+}

plugin/service/storage/const.go

@@ -0,0 +1,9 @@
+package storage
+
+const (
+	// ConstAwsEndpointUrl is the key for the endpoint url in the aws s3 client.
+	ConstAwsEndpointUrl = "endpoint_url"
+
+	// defaultStreamChunkSize is the recommened chunk size for sending data through a stream
+	defaultStreamChunkSize = 65536 // 64 KiB
+)

plugin/service/storage/plugin.go

@@ -0,0 +1,715 @@
+package storage
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/base64"
+	"io"
+	"os"
+	"path"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/aws/retry"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/google/uuid"
+	cred "github.com/hashicorp/boundary-plugin-host-aws/internal/credential"
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
+	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/types/known/timestamppb"
+)
+
+// Ensure that we are implementing StoragePluginServiceServer
+var _ pb.StoragePluginServiceServer = (*StoragePlugin)(nil)
+
+// StoragePlugin implements the StoragePluginServiceServer interface for the
+// AWS storage service plugin.
+type StoragePlugin struct {
+	pb.UnimplementedStoragePluginServiceServer
+
+	// testCredStateOpts are passed in to the stored state to control test behavior
+	testCredStateOpts []cred.AwsCredentialPersistedStateOption
+
+	// testStorageStateOpts are passed in to the stored state to control test behavior
+	testStorageStateOpts []awsStoragePersistedStateOption
+}
+
+// OnCreateStorageBucket is called when a storage bucket is created.
+func (p *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCreateStorageBucketRequest) (*pb.OnCreateStorageBucketResponse, error) {
+	bucket := req.GetBucket()
+	if bucket == nil {
+		return nil, status.Error(codes.InvalidArgument, "bucket is required")
+	}
+	if bucket.BucketName == "" {
+		return nil, status.Error(codes.InvalidArgument, "bucketName is required")
+	}
+
+	secrets := bucket.GetSecrets()
+	if secrets == nil {
+		return nil, status.Error(codes.InvalidArgument, "secrets 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(secrets, storageAttributes.Region)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set up the creds in our persisted state.
+	credState, err := cred.NewAwsCredentialPersistedState(
+		append([]cred.AwsCredentialPersistedStateOption{
+			cred.WithAccessKeyId(credConfig.AccessKey),
+			cred.WithSecretAccessKey(credConfig.SecretKey),
+			cred.WithRegion(credConfig.Region),
+		}, p.testCredStateOpts...)...,
+	)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
+	}
+
+	// Try to rotate the credentials if we're not skipping them.
+	if !storageAttributes.DisableCredentialRotation {
+		if err := credState.RotateCreds(); err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "error during credential rotation: %s", err)
+		}
+	} else {
+		// Simply validate if we aren't rotating.
+		if err := credState.ValidateCreds(); err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "error during credential validation: %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)
+	}
+
+	// perform dry run to ensure we can interact with the bucket as expected.
+	if err := dryRunValidation(ctx, storageState, storageAttributes, bucket); err != nil {
+		return nil, err
+	}
+
+	persistedProto, err := storageState.toProto()
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	return &pb.OnCreateStorageBucketResponse{
+		Persisted: persistedProto,
+	}, nil
+}
+
+// OnUpdateStorageBucket is called when a storage bucket is updated.
+func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpdateStorageBucketRequest) (*pb.OnUpdateStorageBucketResponse, error) {
+	// For storage buckets we only need to really validate what is
+	// currently in the new copy of the storage bucket data, so skip
+	// fetching the new copy to keep things a little less messy.
+	bucket := req.GetNewBucket()
+	if bucket == nil {
+		return nil, status.Error(codes.InvalidArgument, "new bucket is required")
+	}
+	if bucket.BucketName == "" {
+		return nil, status.Error(codes.InvalidArgument, "new bucketName is required")
+	}
+
+	var updateSecrets bool
+	secrets := bucket.GetSecrets()
+	if secrets != nil {
+		// We will be updating secrets this run, but what exactly that
+		// means will be determined later.
+		updateSecrets = true
+	}
+
+	attrs := bucket.GetAttributes()
+	if attrs == nil {
+		return nil, status.Error(codes.InvalidArgument, "new bucket attributes is required")
+	}
+
+	storageAttributes, err := getStorageAttributes(attrs)
+	if err != nil {
+		return nil, err
+	}
+
+	// Get the persisted data.
+	// NOTE: We might need to change this at a later time. I'm not too
+	// sure *exactly* what scenarios we might encounter that would
+	// ultimately mean that we would have to handle an empty or missing
+	// state in update, but what we are ultimately assuming here
+	// (implicitly through awsStoragePersistedStateFromProto) is that
+	// the state will exist and be populated. Personally I think this
+	// is fine and important, but this may change in the future.
+	credState, err := cred.AwsCredentialPersistedStateFromProto(
+		req.GetPersisted().GetData(),
+		append(p.testCredStateOpts, cred.WithRegion(storageAttributes.Region))...)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
+	}
+
+	if storageAttributes.DisableCredentialRotation && !updateSecrets {
+		// This is a validate check to make sure that we aren't disabling
+		// rotation for credentials currently being managed by rotation.
+		// This is not allowed.
+		if !credState.CredsLastRotatedTime.IsZero() {
+			return nil, status.Error(codes.FailedPrecondition, "cannot disable rotation for already-rotated credentials")
+		}
+	}
+
+	if updateSecrets {
+		storageSecrets, err := cred.GetCredentialsConfig(secrets, storageAttributes.Region)
+		if err != nil {
+			return nil, err
+		}
+
+		// Replace the credentials. This checks the timestamp on the last
+		// rotation time as well and deletes the credentials if we are
+		// managing them (ie: if we've rotated them before).
+		if err := credState.ReplaceCreds(storageSecrets.AccessKey, storageSecrets.SecretKey); err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "error attempting to replace credentials: %s", err)
+		}
+	}
+
+	if !storageAttributes.DisableCredentialRotation && credState.CredsLastRotatedTime.IsZero() {
+		// If we're enabling rotation now but didn't before, or have
+		// freshly replaced credentials, we can rotate here.
+		if err := credState.RotateCreds(); err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "error during credential rotation: %s", err)
+		}
+	}
+
+	storageState, err := newAwsStoragePersistedState(
+		append([]awsStoragePersistedStateOption{
+			withCredentials(credState),
+		}, p.testStorageStateOpts...)...,
+	)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
+	}
+
+	// perform dry run to ensure we can interact with the bucket as expected.
+	if err := dryRunValidation(ctx, storageState, storageAttributes, bucket); err != nil {
+		return nil, err
+	}
+
+	persistedProto, err := storageState.toProto()
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	return &pb.OnUpdateStorageBucketResponse{
+		Persisted: persistedProto,
+	}, nil
+}
+
+// OnDeleteStorageBucket is called when a storage bucket is deleted.
+func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDeleteStorageBucketRequest) (*pb.OnDeleteStorageBucketResponse, error) {
+	bucket := req.GetBucket()
+	if bucket == nil {
+		return nil, status.Error(codes.InvalidArgument, "bucket 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
+	}
+
+	// Get the persisted data.
+	// NOTE: We return on error here, blocking the delete. This may or
+	// may not be an overzealous approach to maintaining database/state
+	// integrity. May need to be changed at later time if there are
+	// scenarios where we might be deleting things and any secret state
+	// may be corrupt/and or legitimately missing.
+	credState, err := cred.AwsCredentialPersistedStateFromProto(
+		req.GetPersisted().GetData(),
+		append(p.testCredStateOpts, cred.WithRegion(storageAttributes.Region))...)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
+	}
+
+	_, err = newAwsStoragePersistedState(
+		append([]awsStoragePersistedStateOption{
+			withCredentials(credState),
+		}, p.testStorageStateOpts...)...,
+	)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
+	}
+
+	if !credState.CredsLastRotatedTime.IsZero() {
+		// Delete old/existing credentials. This is done with the same
+		// credentials to ensure that it has the proper permissions to do
+		// it.
+		if err := credState.DeleteCreds(); err != nil {
+			return nil, status.Errorf(codes.Aborted, "error removing rotated credentials during storage bucket deletion: %s", err)
+		}
+	}
+
+	return &pb.OnDeleteStorageBucketResponse{}, nil
+}
+
+// HeadObject is called to get the metadata of an object.
+func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectRequest) (*pb.HeadObjectResponse, error) {
+	if req.GetKey() == "" {
+		return nil, status.Error(codes.InvalidArgument, "key is required")
+	}
+
+	bucket := req.GetBucket()
+	if bucket == nil {
+		return nil, status.Error(codes.InvalidArgument, "bucket is required")
+	}
+
+	if bucket.BucketName == "" {
+		return nil, status.Error(codes.InvalidArgument, "bucketName is required")
+	}
+
+	secrets := bucket.GetSecrets()
+	if secrets == nil {
+		return nil, status.Error(codes.InvalidArgument, "secrets 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(secrets, storageAttributes.Region)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set up the creds in our persisted state.
+	credState, err := cred.NewAwsCredentialPersistedState(
+		append([]cred.AwsCredentialPersistedStateOption{
+			cred.WithAccessKeyId(credConfig.AccessKey),
+			cred.WithSecretAccessKey(credConfig.SecretKey),
+			cred.WithRegion(credConfig.Region),
+		}, 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.Region != "" {
+		opts = append(opts, WithRegion(storageAttributes.Region))
+	}
+	if storageAttributes.EndpointUrl != "" {
+		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
+	}
+	s3Client, err := storageState.S3Client(ctx, opts...)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "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),
+	})
+	if err != nil {
+		return nil, parseAWSErrCode("error getting head object from s3", err)
+	}
+	return &pb.HeadObjectResponse{
+		ContentLength: resp.ContentLength,
+		LastModified:  timestamppb.New(*resp.LastModified),
+	}, nil
+}
+
+// ValidatePermissions is called to validate the secrets associated with the storage bucket.
+func (p *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.ValidatePermissionsRequest) (*pb.ValidatePermissionsResponse, error) {
+	bucket := req.GetBucket()
+	if bucket == nil {
+		return nil, status.Error(codes.InvalidArgument, "bucket is required")
+	}
+
+	if bucket.BucketName == "" {
+		return nil, status.Error(codes.InvalidArgument, "bucketName is required")
+	}
+
+	secrets := bucket.GetSecrets()
+	if secrets == nil {
+		return nil, status.Error(codes.InvalidArgument, "secrets 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(secrets, storageAttributes.Region)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set up the creds in our persisted state.
+	credState, err := cred.NewAwsCredentialPersistedState(
+		append([]cred.AwsCredentialPersistedStateOption{
+			cred.WithAccessKeyId(credConfig.AccessKey),
+			cred.WithSecretAccessKey(credConfig.SecretKey),
+			cred.WithRegion(credConfig.Region),
+		}, 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)
+	}
+
+	// perform dry run to ensure we can interact with the bucket as expected.
+	if err := dryRunValidation(ctx, storageState, storageAttributes, bucket); err != nil {
+		return nil, err
+	}
+
+	return &pb.ValidatePermissionsResponse{}, nil
+}
+
+// GetObject is called when retrieving objects from an s3 bucket.
+// GetObject is a blocking call until the stream has been recieved in full.
+func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePluginService_GetObjectServer) error {
+	if req.GetKey() == "" {
+		return status.Error(codes.InvalidArgument, "key is required")
+	}
+
+	bucket := req.GetBucket()
+	if bucket == nil {
+		return status.Error(codes.InvalidArgument, "bucket is required")
+	}
+
+	if bucket.GetBucketName() == "" {
+		return status.Error(codes.InvalidArgument, "bucketName is required")
+	}
+
+	secrets := bucket.GetSecrets()
+	if secrets == nil {
+		return status.Error(codes.InvalidArgument, "secrets is required")
+	}
+
+	attrs := bucket.GetAttributes()
+	if attrs == nil {
+		return status.Error(codes.InvalidArgument, "attributes is required")
+	}
+
+	storageAttributes, err := getStorageAttributes(attrs)
+	if err != nil {
+		return err
+	}
+
+	credConfig, err := cred.GetCredentialsConfig(secrets, storageAttributes.Region)
+	if err != nil {
+		return err
+	}
+
+	// Set up the creds in our persisted state.
+	credState, err := cred.NewAwsCredentialPersistedState(
+		append([]cred.AwsCredentialPersistedStateOption{
+			cred.WithAccessKeyId(credConfig.AccessKey),
+			cred.WithSecretAccessKey(credConfig.SecretKey),
+			cred.WithRegion(credConfig.Region),
+		}, p.testCredStateOpts...)...,
+	)
+	if err != nil {
+		return status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
+	}
+
+	storageState, err := newAwsStoragePersistedState(
+		append([]awsStoragePersistedStateOption{
+			withCredentials(credState),
+		}, p.testStorageStateOpts...)...,
+	)
+	if err != nil {
+		return status.Errorf(codes.InvalidArgument, "error setting up persisted state: %s", err)
+	}
+
+	opts := []s3Option{}
+	if storageAttributes.Region != "" {
+		opts = append(opts, WithRegion(storageAttributes.Region))
+	}
+	if storageAttributes.EndpointUrl != "" {
+		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
+	}
+	s3Client, err := storageState.S3Client(stream.Context(), opts...)
+	if err != nil {
+		return status.Errorf(codes.InvalidArgument, "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),
+	})
+	if err != nil {
+		return parseAWSErrCode("error getting object from s3", err)
+	}
+
+	defer resp.Body.Close()
+	reader := bufio.NewReader(resp.Body)
+	chunkSize := req.GetChunkSize()
+	if chunkSize == 0 {
+		chunkSize = defaultStreamChunkSize
+	}
+	for {
+		buffer := make([]byte, chunkSize)
+		n, err := reader.Read(buffer)
+		if err != nil && err != io.EOF {
+			return status.Errorf(codes.Internal, "error reading chunk from s3: %s", err)
+		}
+		if n > 0 {
+			if err := stream.Send(&pb.GetObjectResponse{
+				FileChunk: buffer[:n],
+			}); err != nil {
+				return status.Errorf(codes.Internal, "error sending chunk to client: %s", err)
+			}
+		}
+		if err == io.EOF {
+			break
+		}
+	}
+
+	return nil
+}
+
+// PutObject is called when putting objects into an s3 bucket.
+func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest) (*pb.PutObjectResponse, error) {
+	if req == nil {
+		return nil, status.Errorf(codes.InvalidArgument, "request is required")
+	}
+	if req.GetKey() == "" {
+		return nil, status.Errorf(codes.InvalidArgument, "key 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")
+	}
+	secrets := bucket.GetSecrets()
+	if secrets == nil {
+		return nil, status.Error(codes.InvalidArgument, "secrets is required")
+	}
+	attrs := bucket.GetAttributes()
+	if attrs == nil {
+		return nil, status.Error(codes.InvalidArgument, "attributes is required")
+	}
+	if req.GetPath() == "" {
+		return nil, status.Errorf(codes.InvalidArgument, "path is required")
+	}
+	file, err := os.Open(req.GetPath())
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to open file")
+	}
+	info, err := file.Stat()
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to read file info")
+	}
+	if info.IsDir() {
+		return nil, status.Errorf(codes.InvalidArgument, "path is not a file")
+	}
+	if info.Size() == 0 {
+		return nil, status.Errorf(codes.InvalidArgument, "file is empty")
+	}
+
+	storageAttributes, err := getStorageAttributes(attrs)
+	if err != nil {
+		return nil, err
+	}
+
+	credConfig, err := cred.GetCredentialsConfig(secrets, storageAttributes.Region)
+	if err != nil {
+		return nil, err
+	}
+
+	// Set up the creds in our persisted state.
+	credState, err := cred.NewAwsCredentialPersistedState(
+		append([]cred.AwsCredentialPersistedStateOption{
+			cred.WithAccessKeyId(credConfig.AccessKey),
+			cred.WithSecretAccessKey(credConfig.SecretKey),
+			cred.WithRegion(credConfig.Region),
+		}, 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.Region != "" {
+		opts = append(opts, WithRegion(storageAttributes.Region))
+	}
+	if storageAttributes.EndpointUrl != "" {
+		opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
+	}
+	s3Client, err := storageState.S3Client(ctx, opts...)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
+	}
+
+	hash := sha256.New()
+	if _, err := io.Copy(hash, file); err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to calcualte hash")
+	}
+	checksum := base64.StdEncoding.EncodeToString(hash.Sum(nil))
+
+	_, err = file.Seek(0, io.SeekStart)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "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),
+	})
+	if err != nil {
+		return nil, parseAWSErrCode("error putting object into s3", err)
+	}
+	if resp.ChecksumSHA256 == nil {
+		return nil, status.Errorf(codes.Internal, "missing checksum response from aws")
+	}
+	if checksum != *resp.ChecksumSHA256 {
+		return nil, status.Errorf(codes.Internal, "mismatched checksum")
+	}
+	decodedChecksum, err := base64.StdEncoding.DecodeString(*resp.ChecksumSHA256)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed to decode checksum value from aws")
+	}
+	return &pb.PutObjectResponse{
+		ChecksumSha_256: decodedChecksum,
+	}, 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
+// PutObject permissions. This function will read the same file it had uploaded to validate
+// it has GetObject permissions. This function will read the metadata of the same file it
+// had uploaded to validate it has GetObjectAttributes permissions.
+func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attrs *StorageAttributes, bucket *storagebuckets.StorageBucket) error {
+	if state == nil {
+		return status.Error(codes.InvalidArgument, "persisted state is required")
+	}
+	if attrs == nil {
+		return status.Error(codes.InvalidArgument, "attributes is required")
+	}
+	if bucket == nil {
+		return status.Error(codes.InvalidArgument, "bucket is required")
+	}
+
+	opts := []s3Option{}
+	if attrs.Region != "" {
+		opts = append(opts, WithRegion(attrs.Region))
+	}
+	if attrs.EndpointUrl != "" {
+		opts = append(opts, WithEndpoint(attrs.EndpointUrl))
+	}
+
+	client, err := state.S3Client(ctx, opts...)
+	if err != nil {
+		return status.Errorf(codes.InvalidArgument, "error getting S3 client: %s", err)
+	}
+
+	objectKey := path.Join(bucket.GetBucketPrefix(), uuid.New().String())
+	if _, err := client.PutObject(ctx, &s3.PutObjectInput{
+		Bucket: aws.String(bucket.GetBucketName()),
+		Key:    aws.String(objectKey),
+		Body:   bytes.NewReader([]byte("hashicorp boundary aws plugin access test")),
+	}); err != nil {
+		return parseAWSErrCode("error failed to put object", err)
+	}
+
+	if _, err := client.GetObject(ctx, &s3.GetObjectInput{
+		Bucket: aws.String(bucket.GetBucketName()),
+		Key:    aws.String(objectKey),
+	}); err != nil {
+		return parseAWSErrCode("error failed to get object", err)
+	}
+
+	if _, err := client.HeadObject(ctx, &s3.HeadObjectInput{
+		Bucket: aws.String(bucket.GetBucketName()),
+		Key:    aws.String(objectKey),
+	}); err != nil {
+		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{
+		Bucket: aws.String(bucket.GetBucketName()),
+		Key:    aws.String(objectKey),
+	})
+
+	return nil
+}
+
+// parseAWSErrCode detects if an aws error is a throttle error.
+// If the error is a throttle error, then an Unavailable status code
+// is returned. Otherwise the default status code is an Internal error.
+func parseAWSErrCode(msg string, err error) error {
+	errCode := codes.Internal
+	retryErr := retry.ThrottleErrorCode{
+		Codes: retry.DefaultThrottleErrorCodes,
+	}
+	if retryErr.IsErrorThrottle(err).Bool() {
+		errCode = codes.Unavailable
+	}
+	return status.Errorf(errCode, "%s: %s", msg, err)
+}

plugin/service/storage/state.go

@@ -0,0 +1,189 @@
+package storage
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/hashicorp/boundary-plugin-host-aws/internal/credential"
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+type S3API interface {
+	DeleteObject(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error)
+	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)
+}
+
+var customClient = &http.Client{
+	Transport: &http.Transport{
+		Proxy: http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+			DualStack: true,
+		}).DialContext,
+	},
+}
+
+type awsStoragePersistedState struct {
+	*credential.AwsCredentialPersistedState
+
+	// testS3APIFunc provides a way to provide a factory for a mock EC2 client
+	testS3APIFunc s3APIFunc
+}
+
+type s3APIFunc func(...aws.Config) (S3API, error)
+
+type awsStoragePersistedStateOption func(s *awsStoragePersistedState) error
+
+func withTestS3APIFunc(f s3APIFunc) awsStoragePersistedStateOption {
+	return func(s *awsStoragePersistedState) error {
+		if s.testS3APIFunc != nil {
+			return errors.New("test API function already set")
+		}
+
+		s.testS3APIFunc = f
+		return nil
+	}
+}
+
+func withCredentials(x *credential.AwsCredentialPersistedState) awsStoragePersistedStateOption {
+	return func(s *awsStoragePersistedState) error {
+		if s.AwsCredentialPersistedState != nil {
+			return errors.New("aws credentials already set")
+		}
+
+		s.AwsCredentialPersistedState = x
+		return nil
+	}
+}
+
+func newAwsStoragePersistedState(opts ...awsStoragePersistedStateOption) (*awsStoragePersistedState, error) {
+	s := new(awsStoragePersistedState)
+	for _, opt := range opts {
+		if err := opt(s); err != nil {
+			return nil, err
+		}
+	}
+	return s, nil
+}
+
+func (s *awsStoragePersistedState) toProto() (*storagebuckets.StorageBucketPersisted, error) {
+
+	data, err := structpb.NewStruct(s.ToMap())
+	if err != nil {
+		return nil, fmt.Errorf("error converting state to structpb.Struct: %w", err)
+	}
+	return &storagebuckets.StorageBucketPersisted{Data: data}, nil
+}
+
+// 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) {
+	opts, err := getOpts(opt...)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing options when fetching S3 client: %w", err)
+	}
+
+	awsConfigOpts := []func(*config.LoadOptions) error{}
+	sess, err := s.GetSession()
+	if err != nil {
+		return nil, fmt.Errorf("error getting AWS session when fetching S3 client: %w", err)
+	}
+	creds, err := sess.Config.Credentials.Get()
+	if err != nil {
+		return nil, fmt.Errorf("error getting AWS session when fetching S3 client: %w", err)
+	}
+	customCredentials := config.WithCredentialsProvider(
+		credentials.NewStaticCredentialsProvider(creds.AccessKeyID, creds.SecretAccessKey, creds.SessionToken),
+	)
+	awsConfigOpts = append(awsConfigOpts, customCredentials)
+	if sess.Config.Region != nil {
+		awsConfigOpts = append(awsConfigOpts, config.WithRegion(*sess.Config.Region))
+	}
+	if opts.withRegion != "" {
+		awsConfigOpts = append(awsConfigOpts, config.WithRegion(opts.withRegion))
+	}
+	if opts.withEndpoint != "" {
+		customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
+			if service == s3.ServiceID {
+				if opts.withRegion != "" {
+					region = opts.withRegion
+				}
+				return aws.Endpoint{
+					PartitionID:   "aws",
+					URL:           opts.withEndpoint,
+					SigningRegion: region,
+				}, nil
+			}
+			// returning EndpointNotFoundError will allow the service to fallback to it's default resolution
+			return aws.Endpoint{}, &aws.EndpointNotFoundError{}
+		})
+		awsConfigOpts = append(awsConfigOpts, config.WithEndpointResolverWithOptions(customResolver))
+	}
+	cfg, err := config.LoadDefaultConfig(ctx, awsConfigOpts...)
+	if err != nil {
+		return nil, fmt.Errorf("error loading aws configuration when fetching S3 client: %w", err)
+	}
+
+	if s.testS3APIFunc != nil {
+		return s.testS3APIFunc(cfg)
+	}
+
+	cfg.HTTPClient = customClient
+
+	return s3.NewFromConfig(cfg), nil
+}
+
+// getOpts iterates the inbound s3Options and returns a struct
+func getOpts(opt ...s3Option) (s3Options, error) {
+	opts := getDefaultS3Options()
+	for _, o := range opt {
+		if o == nil {
+			continue
+		}
+		if err := o(&opts); err != nil {
+			return s3Options{}, err
+		}
+	}
+	return opts, nil
+}
+
+// s3Option - how s3Options are passed as arguments
+type s3Option func(*s3Options) error
+
+// options = how options are represented
+type s3Options struct {
+	withRegion   string
+	withEndpoint string
+}
+
+func getDefaultS3Options() s3Options {
+	return s3Options{}
+}
+
+// WithRegion contains the region to use
+func WithRegion(with string) s3Option {
+	return func(o *s3Options) error {
+		o.withRegion = with
+		return nil
+	}
+}
+
+// WithEndpoint contains the endpoint to use
+func WithEndpoint(with string) s3Option {
+	return func(o *s3Options) error {
+		o.withEndpoint = with
+		return nil
+	}
+}

plugin/service/storage/state_test.go

@@ -0,0 +1,40 @@
+package storage
+
+import (
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/hashicorp/boundary-plugin-host-aws/internal/credential"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAwsStoragePersistedStateS3Client(t *testing.T) {
+	require := require.New(t)
+	cred, err := credential.NewAwsCredentialPersistedState(
+		credential.WithRegion("us-west-2"),
+		credential.WithAccessKeyId("foobar"),
+		credential.WithSecretAccessKey("barfoo"),
+	)
+	require.NoError(err)
+
+	state, err := newAwsStoragePersistedState(
+		withCredentials(cred),
+		withTestS3APIFunc(newTestMockS3(nil)),
+	)
+	require.NoError(err)
+
+	clientRaw, err := state.S3Client(context.Background(), WithRegion("us-west-2"), WithEndpoint("0.0.0.0"))
+	require.NoError(err)
+	require.NotNil(clientRaw)
+	client, ok := clientRaw.(*testMockS3)
+	require.True(ok)
+	require.NotNil(client)
+	require.Equal("us-west-2", client.Region)
+
+	endpoint, err := client.Endpoint.ResolveEndpoint(s3.ServiceID, client.Region)
+	require.NoError(err)
+	require.Equal("0.0.0.0", endpoint.URL)
+	require.Equal("us-west-2", endpoint.SigningRegion)
+	require.Equal("aws", endpoint.PartitionID)
+}

plugin/service/storage/testing.go

@@ -0,0 +1,300 @@
+package storage
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go/service/sts"
+	"github.com/hashicorp/boundary-plugin-host-aws/internal/credential"
+	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
+	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/hashicorp/go-secure-stdlib/awsutil"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/protobuf/types/known/structpb"
+)
+
+const (
+	testOptionErr            = "test option error"
+	testGetUserErr           = "test error for GetUser"
+	testGetCallerIdentityErr = "test error for GetCallerIdentity"
+	testDeleteAccessKeyErr   = "test error for DeleteAccessKey"
+	testGetObjectErr         = "test error for GetObject"
+	testPutObjectErr         = "test error for PutObject"
+	testHeadObjectErr        = "test error for HeadObject"
+)
+
+// throttleErr is a mocked error used for testing the aws s3 client
+type throttleErr struct{}
+
+func (e *throttleErr) ErrorCode() string {
+	return "ThrottlingException"
+}
+
+func (e *throttleErr) Error() string {
+	return "ThrottlingException"
+}
+
+type testMockS3State struct {
+	GetObjectCalled      bool
+	GetObjectInputParams *s3.GetObjectInput
+
+	PutObjectCalled      bool
+	PutObjectInputParams *s3.PutObjectInput
+	PutObjectBody        []byte
+
+	HeadObjectCalled      bool
+	HeadObjectInputParams *s3.HeadObjectInput
+}
+
+func (s *testMockS3State) Reset() {
+	s.GetObjectCalled = false
+	s.GetObjectInputParams = nil
+	s.PutObjectCalled = false
+	s.PutObjectInputParams = nil
+	s.PutObjectBody = nil
+	s.HeadObjectCalled = false
+	s.HeadObjectInputParams = nil
+}
+
+type testMockS3 struct {
+	S3API
+
+	State    *testMockS3State
+	Region   string
+	Endpoint aws.EndpointResolverWithOptions
+
+	// mocked responses for getObject
+	GetObjectOutput *s3.GetObjectOutput
+	GetObjectErr    error
+
+	// mocked responses for putObject
+	PutObjectOutput *s3.PutObjectOutput
+	PutObjectErr    error
+
+	// mocked responses for headObject
+	HeadObjectOutput *s3.HeadObjectOutput
+	HeadObjectErr    error
+}
+
+type testMockS3Option func(m *testMockS3) error
+
+func testMockS3WithPutObjectOutput(o *s3.PutObjectOutput) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.PutObjectOutput = o
+		return nil
+	}
+}
+
+func testMockS3WithPutObjectError(e error) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.PutObjectErr = e
+		return nil
+	}
+}
+
+func testMockS3WithGetObjectOutput(o *s3.GetObjectOutput) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.GetObjectOutput = o
+		return nil
+	}
+}
+
+func testMockS3WithGetObjectError(e error) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.GetObjectErr = e
+		return nil
+	}
+}
+
+func testMockS3WithHeadObjectOutput(o *s3.HeadObjectOutput) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.HeadObjectOutput = o
+		return nil
+	}
+}
+
+func testMockS3WithHeadObjectError(e error) testMockS3Option {
+	return func(m *testMockS3) error {
+		m.HeadObjectErr = e
+		return nil
+	}
+}
+
+func newTestMockS3(state *testMockS3State, opts ...testMockS3Option) s3APIFunc {
+	return func(cfgs ...aws.Config) (S3API, error) {
+		m := &testMockS3{
+			State: state,
+		}
+
+		for _, opt := range opts {
+			if err := opt(m); err != nil {
+				return nil, err
+			}
+		}
+
+		for _, cfg := range cfgs {
+			// Last region takes precedence
+			if cfg.Region != "" {
+				m.Region = cfg.Region
+			}
+			if cfg.EndpointResolverWithOptions != nil {
+				m.Endpoint = cfg.EndpointResolverWithOptions
+			}
+		}
+
+		return m, nil
+	}
+}
+
+func (m *testMockS3) GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error) {
+	if m.State != nil {
+		m.State.GetObjectCalled = true
+		m.State.GetObjectInputParams = params
+	}
+
+	if m.GetObjectErr != nil {
+		return nil, m.GetObjectErr
+	}
+
+	return m.GetObjectOutput, nil
+}
+
+func (m *testMockS3) PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error) {
+	if m.State != nil {
+		m.State.PutObjectCalled = true
+		m.State.PutObjectInputParams = params
+	}
+
+	if m.PutObjectErr != nil {
+		return nil, m.PutObjectErr
+	}
+
+	// Reads and Writes on the pipe are matched one to one except when multiple Reads are needed to consume a single Write.
+	// Therefore, we need to read the entire body to ensure that the pipe is not blocked.
+	if params.Body != nil {
+		data, err := io.ReadAll(params.Body)
+		if err != nil {
+			return nil, err
+		}
+		if m.State == nil || m.State.PutObjectBody == nil {
+			return m.PutObjectOutput, nil
+		}
+		if !bytes.Equal(data, m.State.PutObjectBody) {
+			return nil, fmt.Errorf("expected body %q, got %q", m.State.PutObjectBody, data)
+		}
+	}
+
+	return m.PutObjectOutput, nil
+}
+
+func (m *testMockS3) DeleteObject(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) {
+	return &s3.DeleteObjectOutput{}, nil
+}
+
+func (m *testMockS3) HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error) {
+	if m.State != nil {
+		m.State.HeadObjectCalled = true
+		m.State.HeadObjectInputParams = params
+	}
+
+	if m.HeadObjectErr != nil {
+		return nil, m.HeadObjectErr
+	}
+
+	return m.HeadObjectOutput, nil
+}
+
+type getObjectStreamMock struct {
+	grpc.ServerStream
+	ctx  context.Context
+	err  error
+	data *pb.GetObjectResponse
+}
+
+func (s *getObjectStreamMock) Context() context.Context {
+	return s.ctx
+}
+
+// Send streams a response message to the client.
+// This function should only be called on the server side.
+func (g *getObjectStreamMock) Send(req *pb.GetObjectResponse) error {
+	if req == nil {
+		return fmt.Errorf(`parameter arg "resp GetObjectResponse" cannot be nil.`)
+	}
+	g.data = req
+	return nil
+}
+
+// Recv streams a response message from the server.
+// This function should only be called on the client side.
+func (g *getObjectStreamMock) Recv() (*pb.GetObjectResponse, error) {
+	if g.data == nil {
+		if g.err != nil {
+			return nil, g.err
+		}
+		return nil, io.EOF
+	}
+	resp := &pb.GetObjectResponse{
+		FileChunk: append([]byte{}, g.data.FileChunk...),
+	}
+	g.data = nil
+	return resp, nil
+}
+
+func newGetObjectStreamMock() *getObjectStreamMock {
+	return &getObjectStreamMock{
+		ctx: context.Background(),
+	}
+}
+
+func createTime(t *testing.T, timestamp string) time.Time {
+	require := require.New(t)
+	testTime, err := time.Parse(time.RFC3339, timestamp)
+	require.NoError(err)
+	return testTime
+}
+
+func deepCopyPutObjectRequest(v *pb.PutObjectRequest) *pb.PutObjectRequest {
+	deepCopy := &pb.PutObjectRequest{
+		Key:  v.Key,
+		Path: v.Path,
+	}
+	if v.Bucket != nil {
+		deepCopy.Bucket = &storagebuckets.StorageBucket{
+			BucketName:   v.Bucket.BucketName,
+			BucketPrefix: v.Bucket.BucketPrefix,
+		}
+		if v.Bucket.Attributes != nil {
+			attrs, _ := structpb.NewStruct(v.Bucket.Attributes.AsMap())
+			deepCopy.Bucket.Attributes = attrs
+		}
+		if v.Bucket.Secrets != nil {
+			secrets, _ := structpb.NewStruct(v.Bucket.Secrets.AsMap())
+			deepCopy.Bucket.Secrets = secrets
+		}
+	}
+	return deepCopy
+}
+
+func validSTSMock() []credential.AwsCredentialPersistedStateOption {
+	return []credential.AwsCredentialPersistedStateOption{
+		credential.WithStateTestOpts([]awsutil.Option{
+			awsutil.WithSTSAPIFunc(
+				awsutil.NewMockSTS(
+					awsutil.WithGetCallerIdentityOutput(&sts.GetCallerIdentityOutput{
+						Account: aws.String("0123456789"),
+						Arn:     aws.String("arn:aws:iam::0123456789:user/test"),
+						UserId:  aws.String("test"),
+					}),
+				),
+			),
+		}),
+	}
+}