Large File Support (#21)

* disable multipart puts and add a large file test

* read random bytes rather than generate from string

go.mod

@@ -5,12 +5,13 @@ go 1.22.2
 require (
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.5.0
-	github.com/hashicorp/boundary/sdk v0.0.42
+	github.com/hashicorp/boundary/sdk v0.0.43-0.20240717182311-a20aae98794a
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/minio/madmin-go/v3 v3.0.48
 	github.com/minio/minio-go/v7 v7.0.67
 	github.com/ory/dockertest/v3 v3.10.0
 	github.com/stretchr/testify v1.8.4
-	google.golang.org/grpc v1.59.0
+	google.golang.org/grpc v1.61.0
 	google.golang.org/protobuf v1.33.0
 )
 
@@ -35,7 +36,6 @@ require (
 	github.com/hashicorp/eventlogger v0.2.6-0.20231025104552-802587e608f0 // indirect
 	github.com/hashicorp/eventlogger/filters/encrypt v0.1.8-0.20231025104552-802587e608f0 // indirect
 	github.com/hashicorp/go-kms-wrapping/v2 v2.0.14 // indirect
-	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/go-uuid v1.0.3 // indirect
 	github.com/imdario/mergo v0.3.13 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
@@ -79,11 +79,13 @@ require (
 	golang.org/x/crypto v0.22.0 // indirect
 	golang.org/x/mod v0.9.0 // indirect
 	golang.org/x/net v0.24.0 // indirect
-	golang.org/x/sync v0.3.0 // indirect
+	golang.org/x/sync v0.5.0 // indirect
 	golang.org/x/sys v0.19.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/tools v0.7.0 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b // 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
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

go.sum

@@ -63,8 +63,8 @@ github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaU
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
 github.com/google/uuid v1.5.0 h1:1p67kYwdtXjb0gL0BPiP1Av9wiZPo5A8z2cWkTZ+eyU=
 github.com/google/uuid v1.5.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
-github.com/hashicorp/boundary/sdk v0.0.42 h1:iRDdnM6/6q9qLtBP7/cpq/FWQGf69TFSsBQFxgGK8Jw=
-github.com/hashicorp/boundary/sdk v0.0.42/go.mod h1:QJhm4rky2YMtk9DS5Rj8QJhDKB/6zQbVrm3R0QgdYmQ=
+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/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=
@@ -126,8 +126,8 @@ github.com/klauspost/cpuid/v2 v2.2.6 h1:ndNyv040zDGIDh8thGkXYjnFtiN02M1PVVF+JE/4
 github.com/klauspost/cpuid/v2 v2.2.6/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
-github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
-github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
+github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
@@ -289,8 +289,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.3.0 h1:ftCYgMx6zT/asHUrPw8BLLscYtGznsLAnjq5RH9P66E=
-golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
+golang.org/x/sync v0.5.0 h1:60k92dhOjHxJkrqnwsfl8KuaHbn/5dl0lUPUklKo3qE=
+golang.org/x/sync v0.5.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -337,10 +337,14 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b h1:ZlWIi1wSK56/8hn4QcBp/j9M7Gt3U/3hZw3mC7vDICo=
-google.golang.org/genproto/googleapis/rpc v0.0.0-20231016165738-49dd2c1f3d0b/go.mod h1:swOH3j0KzcDDgGUWr+SNpyTen5YrXjS3eyPzFYKc6lc=
-google.golang.org/grpc v1.59.0 h1:Z5Iec2pjwb+LEOqzpB2MR12/eKFhDPhuqW91O+4bwUk=
-google.golang.org/grpc v1.59.0/go.mod h1:aUPDwccQo6OTjy7Hct4AfBPD1GptF4fyUjIkQ9YtF98=
+google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac h1:ZL/Teoy/ZGnzyrqK/Optxxp2pmVh+fmJ97slxSRyzUg=
+google.golang.org/genproto v0.0.0-20240116215550-a9fa1716bcac/go.mod h1:+Rvu7ElI+aLzyDQhpHMFMMltsD6m7nqpuWDd2CwJw3k=
+google.golang.org/genproto/googleapis/api v0.0.0-20240125205218-1f4bbc51befe h1:0poefMBYvYbs7g5UkjS6HcxBPaTRAmznle9jnxYoAI8=
+google.golang.org/genproto/googleapis/api v0.0.0-20240125205218-1f4bbc51befe/go.mod h1:4jWUdICTdgc3Ibxmr8nAJiiLHwQBY0UI0XZcEMaFKaA=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240125205218-1f4bbc51befe h1:bQnxqljG/wqi4NTXu2+DJ3n7APcEA882QZ1JvhQAq9o=
+google.golang.org/genproto/googleapis/rpc v0.0.0-20240125205218-1f4bbc51befe/go.mod h1:PAREbraiVEVGVdTZsVWjSbbTtSyGbAgIIvni8a8CD5s=
+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.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI=

plugin/service/storage/plugin.go

@@ -19,6 +19,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/hashicorp/boundary/sdk/pbs/controller/api/resources/storagebuckets"
 	pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
+	"github.com/hashicorp/go-multierror"
 	madmin "github.com/minio/madmin-go/v3"
 	"github.com/minio/minio-go/v7"
 	"github.com/minio/minio-go/v7/pkg/credentials"
@@ -60,11 +61,6 @@ func (sp *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCr
 		return nil, err
 	}
 
-	err = ensureServiceAccount(ctx, sa, sec)
-	if err != nil {
-		return nil, status.Errorf(codes.InvalidArgument, "failed to ensure service account: %v", err)
-	}
-
 	deleteOldCredsFn := func() error { return nil }
 	if !sa.DisableCredentialRotation {
 		newSec, fn, err := rotateCredentials(ctx, bucket.GetBucketName(), sa, sec)
@@ -84,9 +80,15 @@ func (sp *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCr
 		return nil, status.Errorf(codes.Unknown, "failed to create minio sdk client: %v", err)
 	}
 
-	err = dryRun(ctx, cl, bucket)
-	if err != nil {
-		return nil, status.Errorf(codes.Unknown, "failed to verify provided minio environment: %v", err)
+	p, dryRunErr := dryRun(ctx, cl, bucket)
+	if dryRunErr.Len() > 0 {
+		st := status.New(codes.InvalidArgument, fmt.Sprintf("failed to verify provided minio environment: %v", dryRunErr.Unwrap()))
+
+		st, err = st.WithDetails(&pb.StorageBucketCredentialState{State: p})
+		if err != nil {
+			st = status.New(codes.Unknown, fmt.Sprintf("failed to verify provided minio environment: %v", dryRunErr.Unwrap()))
+		}
+		return nil, st.Err()
 	}
 
 	persistedData, err := structpb.NewStruct(sec.AsMap())
@@ -178,11 +180,6 @@ func (sp *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUp
 			return nil, err
 		}
 
-		err = ensureServiceAccount(ctx, nsa, newSec)
-		if err != nil {
-			return nil, status.Errorf(codes.InvalidArgument, "failed to ensure service account: %s", err)
-		}
-
 		if !nsa.DisableCredentialRotation {
 			// We have new incoming credentials and we're also rotating,
 			// therefore we use incoming credentials to create a new service
@@ -219,8 +216,16 @@ func (sp *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUp
 		return nil, status.Errorf(codes.InvalidArgument, "failed to create minio sdk client: %s", err)
 	}
 
-	if err = dryRun(ctx, cl, newBucket); err != nil {
-		return nil, status.Errorf(codes.InvalidArgument, "failed to verify provided minio environment: %s", err)
+	if p, dryRunErr := dryRun(ctx, cl, newBucket); dryRunErr != nil {
+		if dryRunErr.Len() > 0 {
+			st := status.New(codes.InvalidArgument, fmt.Sprintf("failed to verify provided minio environment: %v", dryRunErr.Unwrap()))
+
+			st, err = st.WithDetails(&pb.StorageBucketCredentialState{State: p})
+			if err != nil {
+				st = status.New(codes.InvalidArgument, fmt.Sprintf("failed to verify provided minio environment: %v", dryRunErr.Unwrap()))
+			}
+			return nil, st.Err()
+		}
 	}
 
 	persisted, err := structpb.NewStruct(sec.AsMap())
@@ -292,9 +297,13 @@ func (sp *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.Valida
 		return nil, err
 	}
 
-	err = ensureServiceAccount(ctx, sa, sec)
-	if err != nil {
-		return nil, status.Errorf(codes.InvalidArgument, "failed to ensure service account: %v", err)
+	if !sec.LastRotatedTime.IsZero() {
+		// since these are existing secrets, we only want to validate the
+		// service account if the credentials have been rotated previously
+		err = ensureServiceAccount(ctx, sa, sec)
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "failed to ensure service account: %v", err)
+		}
 	}
 
 	cl, err := minio.New(sa.EndpointUrl, &minio.Options{
@@ -306,9 +315,15 @@ func (sp *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.Valida
 		return nil, status.Errorf(codes.Unknown, "failed to create minio sdk client: %v", err)
 	}
 
-	err = dryRun(ctx, cl, bucket)
-	if err != nil {
-		return nil, status.Errorf(codes.Unknown, "failed to verify provided minio environment: %v", err)
+	p, dryRunErr := dryRun(ctx, cl, bucket)
+	if dryRunErr.Unwrap() != nil {
+		st := status.New(codes.FailedPrecondition, fmt.Sprintf("failed to verify provided minio environment: %v", dryRunErr.Unwrap()))
+
+		st, err = st.WithDetails(&pb.StorageBucketCredentialState{State: p})
+		if err != nil {
+			st = status.New(codes.Unknown, fmt.Sprintf("failed to verify provided minio environment: %v", dryRunErr.Unwrap()))
+		}
+		return nil, st.Err()
 	}
 
 	return &pb.ValidatePermissionsResponse{}, nil
@@ -344,13 +359,13 @@ func (sp *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReque
 		Region: sa.Region,
 	})
 	if err != nil {
-		return nil, status.Errorf(codes.Unknown, "failed to create minio sdk client: %v", err)
+		return nil, errorToStatus(codes.Unknown, fmt.Sprintf("failed to create minio sdk client: %v", err), err, req).Err()
 	}
 
 	objKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
 	oi, err := cl.StatObject(ctx, bucket.GetBucketName(), objKey, minio.StatObjectOptions{})
 	if err != nil {
-		return nil, status.Errorf(codes.Unknown, "failed to stat object: %v", err)
+		return nil, errorToStatus(codes.Unknown, fmt.Sprintf("failed to stat object: %v", err), err, req).Err()
 	}
 
 	return &pb.HeadObjectResponse{
@@ -389,13 +404,13 @@ func (sp *StoragePlugin) GetObject(req *pb.GetObjectRequest, objServer pb.Storag
 		Region: sa.Region,
 	})
 	if err != nil {
-		return status.Errorf(codes.Unknown, "failed to create minio sdk client: %v", err)
+		return errorToStatus(codes.Unknown, fmt.Sprintf("failed to create minio sdk client: %v", err), err, req).Err()
 	}
 
 	objKey := path.Join(bucket.GetBucketPrefix(), req.GetKey())
 	obj, err := cl.GetObject(objServer.Context(), bucket.GetBucketName(), objKey, minio.GetObjectOptions{})
 	if err != nil {
-		return status.Errorf(codes.Unknown, "failed to get object: %v", err)
+		return errorToStatus(codes.Unknown, fmt.Sprintf("failed to get object: %v", err), err, req).Err()
 	}
 	defer obj.Close()
 
@@ -478,7 +493,7 @@ func (sp *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest
 		Region: sa.Region,
 	})
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "failed to create minio sdk client: %v", err)
+		return nil, errorToStatus(codes.Internal, fmt.Sprintf("failed to create minio sdk client: %v", err), err, req).Err()
 	}
 
 	hash := sha256.New()
@@ -497,18 +512,24 @@ func (sp *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest
 			"x-amz-checksum-algorithm": "SHA256",
 			"x-amz-checksum-sha256":    checksum,
 		},
+		DisableMultipart: true,
 	})
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "failed to put object into minio: %v", err)
+		return nil, errorToStatus(codes.Internal, fmt.Sprintf("failed to put object into minio: %v", err), err, req).Err()
 	}
 
-	if res.ChecksumSHA256 == "" {
-		return nil, status.Error(codes.Internal, "missing checksum response from minio")
+	resChecksum := res.ChecksumSHA256
+	if resChecksum == "" {
+		// We were successful in putting the object but we didn't get a file
+		// SHA256 checksum from the server. Given that the call was successful,
+		// assume everything went ok and force the checksum to match so we can
+		// send it correctly back to Boundary.
+		resChecksum = checksum
 	}
-	if checksum != res.ChecksumSHA256 {
+	if checksum != resChecksum {
 		return nil, status.Error(codes.Internal, "mismatched checksum")
 	}
-	decodedChecksum, err := base64.StdEncoding.DecodeString(res.ChecksumSHA256)
+	decodedChecksum, err := base64.StdEncoding.DecodeString(resChecksum)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "failed to decode checksum value from minio: %v", err)
 	}
@@ -548,7 +569,7 @@ func (sp *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObject
 		Region: sa.Region,
 	})
 	if err != nil {
-		return nil, status.Errorf(codes.Unknown, "failed to create minio sdk client: %v", err)
+		return nil, errorToStatus(codes.Unknown, fmt.Sprintf("failed to create minio sdk client: %v", err), err, req).Err()
 	}
 
 	prefix := path.Join(bucket.GetBucketPrefix(), req.GetKeyPrefix())
@@ -560,7 +581,7 @@ func (sp *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObject
 
 	if !req.Recursive {
 		if err := cl.RemoveObject(ctx, bucket.GetBucketName(), prefix, minio.RemoveObjectOptions{}); err != nil {
-			return nil, status.Errorf(codes.Unknown, "error deleting minio object: %v", err)
+			return nil, errorToStatus(codes.Unknown, fmt.Sprintf("error deleting minio object: %v", err), err, req).Err()
 		}
 		return &pb.DeleteObjectsResponse{
 			ObjectsDeleted: uint32(1),
@@ -573,7 +594,7 @@ func (sp *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObject
 		Recursive: true,
 	}) {
 		if obj.Err != nil {
-			return nil, status.Errorf(codes.Unknown, "error iterating minio bucket contents: %v", err)
+			return nil, errorToStatus(codes.Unknown, fmt.Sprintf("error iterating minio bucket contents: %v", obj.Err), obj.Err, req).Err()
 		}
 		objects = append(objects, obj)
 	}
@@ -590,7 +611,7 @@ func (sp *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObject
 	removed := 0
 	for res := range cl.RemoveObjectsWithResult(ctx, bucket.GetBucketName(), emitter, minio.RemoveObjectsOptions{}) {
 		if res.Err != nil {
-			return nil, status.Errorf(codes.Unknown, "error deleting minio object(s): %v", err)
+			return nil, errorToStatus(codes.Unknown, fmt.Sprintf("error deleting minio object(s): %v", res.Err), res.Err, req).Err()
 		}
 		removed++
 	}
@@ -600,37 +621,145 @@ func (sp *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObject
 	}, nil
 }
 
-func dryRun(ctx context.Context, cl *minio.Client, bucket *storagebuckets.StorageBucket) error {
-	objectKey := path.Join(bucket.GetBucketPrefix(), uuid.New().String())
+// validateWritePermission checks if the plugin has write permission for an object.
+func validateWritePermission(ctx context.Context, cl *minio.Client, bucket *storagebuckets.StorageBucket, objectKey string) (*pb.Permission, *multierror.Error) {
+	const op = "storage.validateWritePermission"
+	var combinedErrors *multierror.Error
+
+	if cl == nil {
+		err := fmt.Errorf("%s: minio client is nil", op)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, multierror.Append(combinedErrors, err)
+	}
+	if bucket == nil {
+		err := fmt.Errorf("%s: storage bucket is nil", op)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, multierror.Append(combinedErrors, err)
+	}
+
 	rd := bytes.NewReader([]byte("hashicorp boundary minio plugin access test"))
 	_, err := cl.PutObject(ctx, bucket.GetBucketName(), objectKey, rd, rd.Size(), minio.PutObjectOptions{})
 	if err != nil {
-		return fmt.Errorf("failed to put object at %q: %w", objectKey, err)
+		combinedErrors = multierror.Append(combinedErrors, fmt.Errorf("failed to put object at %q: %w", objectKey, err))
+		return errorToPermission(err), combinedErrors
 	}
 
-	_, err = cl.StatObject(ctx, bucket.GetBucketName(), objectKey, minio.StatObjectOptions{})
+	return &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}, nil
+}
+
+// validateReadPermission checks if the plugin has read permission for an object.
+func validateReadPermission(ctx context.Context, cl *minio.Client, bucket *storagebuckets.StorageBucket, objectKey string) (*pb.Permission, *multierror.Error) {
+	const op = "storage.validateReadPermission"
+	var combinedErrors *multierror.Error
+
+	if cl == nil {
+		err := fmt.Errorf("%s: minio client is nil", op)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, multierror.Append(combinedErrors, err)
+	}
+	if bucket == nil {
+		err := fmt.Errorf("%s: storage bucket is nil", op)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, multierror.Append(combinedErrors, err)
+	}
+	if objectKey == "" {
+		err := fmt.Errorf("%s: object key is empty", op)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, multierror.Append(combinedErrors, err)
+	}
+
+	permission := &pb.Permission{State: pb.StateType_STATE_TYPE_OK}
+	_, err := cl.StatObject(ctx, bucket.GetBucketName(), objectKey, minio.StatObjectOptions{})
 	if err != nil {
-		return fmt.Errorf("failed to stat object at %q: %w", objectKey, err)
+		permission = errorToPermission(err)
+		combinedErrors = multierror.Append(combinedErrors, fmt.Errorf("failed to stat object at %q: %w", objectKey, err))
 	}
 
 	_, err = cl.GetObject(ctx, bucket.GetBucketName(), objectKey, minio.GetObjectOptions{})
 	if err != nil {
-		return fmt.Errorf("failed to get object at %q: %w", objectKey, err)
+		p := errorToPermission(err)
+		if permission.GetState() == pb.StateType_STATE_TYPE_OK {
+			permission = p
+		} else {
+			if p != nil && p.GetErrorDetails() != "" {
+				permission.ErrorDetails = fmt.Sprintf("%s, %s", permission.GetErrorDetails(), p.GetErrorDetails())
+			}
+		}
+		combinedErrors = multierror.Append(combinedErrors, fmt.Errorf("failed to get object at %q: %w", objectKey, err))
 	}
 
 	oiCh := cl.ListObjects(ctx, bucket.GetBucketName(), minio.ListObjectsOptions{Recursive: true})
 	for oi := range oiCh {
 		if oi.Err != nil {
-			return fmt.Errorf("failed to list objects in bucket %q: %w", bucket.GetBucketName(), oi.Err)
+			if permission.GetState() == pb.StateType_STATE_TYPE_OK {
+				permission = errorToPermission(oi.Err)
+			} else {
+				p := errorToPermission(err)
+				if p != nil && p.GetErrorDetails() != "" {
+					permission.ErrorDetails = fmt.Sprintf("%s, %s", permission.GetErrorDetails(), p.GetErrorDetails())
+				}
+			}
+			combinedErrors = multierror.Append(fmt.Errorf("failed to list objects in bucket %q: %w", bucket.GetBucketName(), oi.Err))
 		}
 	}
 
-	err = cl.RemoveObject(ctx, bucket.GetBucketName(), objectKey, minio.RemoveObjectOptions{})
-	if err != nil {
-		return fmt.Errorf("failed to remove object at %q: %w", objectKey, err)
+	if combinedErrors != nil {
+		return permission, combinedErrors
 	}
 
-	return nil
+	permission.CheckedAt = timestamppb.Now()
+	return permission, nil
+}
+
+// validateDeletePermission checks if the plugin has delete permission for an object.
+func validateDeletePermission(ctx context.Context, cl *minio.Client, bucket *storagebuckets.StorageBucket, objectKey string) (*pb.Permission, *multierror.Error) {
+	const op = "storage.validateDeletePermission"
+
+	var combinedErrors *multierror.Error
+
+	if cl == nil {
+		err := fmt.Errorf("%s: minio client is nil", op)
+		combinedErrors = multierror.Append(combinedErrors, err)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, combinedErrors
+	}
+	if bucket == nil {
+		err := fmt.Errorf("%s: storage bucket is nil", op)
+		combinedErrors = multierror.Append(combinedErrors, err)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, combinedErrors
+	}
+	if objectKey == "" {
+		err := fmt.Errorf("%s: object key is empty", op)
+		combinedErrors = multierror.Append(combinedErrors, err)
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: err.Error(), CheckedAt: timestamppb.Now()}, combinedErrors
+	}
+
+	err := cl.RemoveObject(ctx, bucket.GetBucketName(), objectKey, minio.RemoveObjectOptions{})
+	if err != nil {
+		combinedErrors = multierror.Append(combinedErrors, fmt.Errorf("failed to remove object at %q: %w", objectKey, err))
+		return errorToPermission(err), combinedErrors
+	}
+
+	return &pb.Permission{State: pb.StateType_STATE_TYPE_OK, CheckedAt: timestamppb.Now()}, nil
+}
+
+func dryRun(ctx context.Context, cl *minio.Client, bucket *storagebuckets.StorageBucket) (*pb.Permissions, *multierror.Error) {
+	objectKey := path.Join(bucket.GetBucketPrefix(), uuid.New().String())
+	permissions := &pb.Permissions{
+		Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, CheckedAt: timestamppb.Now()},
+		Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, CheckedAt: timestamppb.Now()},
+		Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, CheckedAt: timestamppb.Now()},
+	}
+
+	var combinedErrors *multierror.Error
+
+	writePermission, writeErr := validateWritePermission(ctx, cl, bucket, objectKey)
+	combinedErrors = multierror.Append(combinedErrors, writeErr)
+	permissions.Write = writePermission
+
+	readPermission, readErr := validateReadPermission(ctx, cl, bucket, objectKey)
+	combinedErrors = multierror.Append(combinedErrors, readErr)
+	permissions.Read = readPermission
+
+	deletePermission, deleteErr := validateDeletePermission(ctx, cl, bucket, objectKey)
+	combinedErrors = multierror.Append(combinedErrors, deleteErr)
+	permissions.Delete = deletePermission
+
+	return permissions, combinedErrors
 }
 
 func newMadminClient(sa *StorageAttributes, sec *StorageSecrets) (*madmin.AdminClient, error) {
@@ -655,3 +784,82 @@ func ensureServiceAccount(ctx context.Context, sa *StorageAttributes, sec *Stora
 
 	return nil
 }
+
+// errorToPermission converts a MinIO error to a plugin permission.
+// errorToPermission only handles a subset of MinIO errors that have to do with
+// permissions.
+// If the MinIO error is not recognized, it returns a permission with an unknown state.
+func errorToPermission(err error) *pb.Permission {
+	if err == nil {
+		return &pb.Permission{}
+	}
+
+	errResponse := minio.ToErrorResponse(err)
+	switch errResponse.Code {
+	// https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html#ErrorCodeList
+	case "AccessDenied":
+		fallthrough
+	case "InvalidSecurity":
+		fallthrough
+	case "InvalidAccessKeyId":
+		fallthrough
+	case "NoSuchBucket":
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: errResponse.Error(), CheckedAt: timestamppb.Now()}
+	case "InvalidObjectState":
+		fallthrough
+	default:
+		return &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: errResponse.Error(), CheckedAt: timestamppb.Now()}
+	}
+}
+
+// errorToStatus converts an error to a gRPC status with the plugin permission
+// attached.
+// If the permission error is not recognized, a gRPC status will be returned
+// with an code and error message passed in.
+func errorToStatus(code codes.Code, msg string, err error, req any) *status.Status {
+	errResponse := minio.ToErrorResponse(err)
+	switch errResponse.Code {
+	// https://docs.aws.amazon.com/AmazonS3/latest/API/ErrorResponses.html#ErrorCodeList
+	case "RequestTimeout":
+		return status.New(codes.DeadlineExceeded, "timeout exceeded when communicating with storage service provider")
+	case "SlowDown":
+		return status.New(codes.Unavailable, "received a throttling response from storage service provider")
+	case "BadDigest":
+		return status.New(codes.Aborted, "storage checksum mismatch")
+	case "NoSuchBucket":
+		return status.New(codes.NotFound, "specified storage bucket does not exist")
+	case "NoSuchKey":
+		return status.New(codes.NotFound, "specified object does not exist")
+	}
+	if errResponse.StatusCode == 400 {
+		// catch-all for various different bad-request related errors
+		return status.New(codes.InvalidArgument, "received a bad request response from storage service provider")
+	}
+
+	// for all other errors, we can parse as permission state errors
+
+	st := status.New(code, msg)
+	if err == nil {
+		return st
+	}
+	permission := errorToPermission(err)
+	state := &pb.StorageBucketCredentialState{State: &pb.Permissions{}}
+
+	// we can use the request type to determine what kind of response we should return
+	switch req.(type) {
+	case pb.HeadObjectRequest:
+		state.State.Read = permission
+	case pb.GetObjectRequest:
+		state.State.Read = permission
+	case pb.PutObjectRequest:
+		state.State.Write = permission
+	case pb.DeleteObjectsRequest:
+		state.State.Delete = permission
+	}
+
+	st, err = st.WithDetails(state)
+	if err != nil {
+		st = status.New(code, msg)
+	}
+	return st
+}

plugin/service/storage/plugin_test.go

@@ -7,6 +7,7 @@ import (
 	"bufio"
 	"bytes"
 	"context"
+	"crypto/rand"
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
@@ -191,11 +192,12 @@ func TestOnCreateStorageBucket(t *testing.T) {
 	require.NoError(t, err)
 
 	tests := []struct {
-		name            string
-		req             *plugin.OnCreateStorageBucketRequest
-		expRsp          *plugin.OnCreateStorageBucketResponse
-		expCredRotation bool
-		expErrMsg       string
+		name                string
+		req                 *plugin.OnCreateStorageBucketRequest
+		expRsp              *plugin.OnCreateStorageBucketResponse
+		expCredRotation     bool
+		expErrMsg           string
+		expectedPermissions *pb.Permissions
 	}{
 		{
 			name:      "nilBucket",
@@ -234,26 +236,6 @@ func TestOnCreateStorageBucket(t *testing.T) {
 			},
 			expErrMsg: "empty secrets input",
 		},
-		{
-			name: "usingNonServiceAccount",
-			req: &plugin.OnCreateStorageBucketRequest{
-				Bucket: &storagebuckets.StorageBucket{
-					BucketName: bucketName,
-					Attributes: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
-						},
-					},
-					Secrets: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
-							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
-						},
-					},
-				},
-			},
-			expErrMsg: "The specified service account is not found (Specified service account does not exist)",
-		},
 		{
 			name: "dryRunFail",
 			req: &plugin.OnCreateStorageBucketRequest{
@@ -281,6 +263,11 @@ func TestOnCreateStorageBucket(t *testing.T) {
 				},
 			},
 			expErrMsg: "failed to verify provided minio environment: failed to put object",
+			expectedPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "Access Denied."},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: "The specified key does not exist."},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_OK, ErrorDetails: ""},
+			},
 		},
 		{
 			name: "credRotationFail",
@@ -323,6 +310,59 @@ func TestOnCreateStorageBucket(t *testing.T) {
 			}(),
 			expErrMsg: "failed to rotate minio credentials: failed to create new minio service account: Access Denied",
 		},
+		{
+			name: "credRotationFailNoSvcAcct",
+			req: func() *plugin.OnCreateStorageBucketRequest {
+				return &plugin.OnCreateStorageBucketRequest{
+					Bucket: &storagebuckets.StorageBucket{
+						BucketName: bucketName,
+						Attributes: &structpb.Struct{
+							Fields: map[string]*structpb.Value{
+								ConstEndpointUrl:               structpb.NewStringValue("http://" + server.ApiAddr),
+								ConstDisableCredentialRotation: structpb.NewBoolValue(false),
+							},
+						},
+						Secrets: &structpb.Struct{
+							Fields: map[string]*structpb.Value{
+								ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+								ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+							},
+						},
+					},
+				}
+			}(),
+			expErrMsg: "The specified service account is not found (Specified service account does not exist)",
+		},
+		{
+			name: "successNonServiceAccount",
+			req: &plugin.OnCreateStorageBucketRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl:               structpb.NewStringValue("http://" + server.ApiAddr),
+							ConstDisableCredentialRotation: structpb.NewBoolValue(true),
+						},
+					},
+					Secrets: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+						},
+					},
+				},
+			},
+			expRsp: &plugin.OnCreateStorageBucketResponse{
+				Persisted: &storagebuckets.StorageBucketPersisted{
+					Data: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "successNoRotation",
 			req: &plugin.OnCreateStorageBucketRequest{
@@ -397,6 +437,28 @@ func TestOnCreateStorageBucket(t *testing.T) {
 			if tt.expErrMsg != "" {
 				require.ErrorContains(t, err, tt.expErrMsg)
 				require.Nil(t, rsp)
+
+				// Check the permission error.
+				if tt.expectedPermissions != nil {
+					st, ok := status.FromError(err)
+					require.True(t, ok)
+					var permission *pb.StorageBucketCredentialState
+					for _, detail := range st.Details() {
+						if errDetail, ok := detail.(*pb.StorageBucketCredentialState); ok {
+							permission = errDetail
+							break
+						}
+					}
+					require.NotNil(t, permission)
+
+					assert.Equal(t, tt.expectedPermissions.GetWrite().GetState(), permission.GetState().GetWrite().GetState())
+					assert.Equal(t, tt.expectedPermissions.GetRead().GetState(), permission.GetState().GetRead().GetState())
+					assert.Equal(t, tt.expectedPermissions.GetDelete().GetState(), permission.GetState().GetDelete().GetState())
+
+					assert.Contains(t, tt.expectedPermissions.GetWrite().GetErrorDetails(), permission.GetState().GetWrite().GetErrorDetails())
+					assert.Contains(t, tt.expectedPermissions.GetRead().GetErrorDetails(), permission.GetState().GetRead().GetErrorDetails())
+					assert.Contains(t, tt.expectedPermissions.GetDelete().GetErrorDetails(), permission.GetState().GetDelete().GetErrorDetails())
+				}
 				return
 			}
 
@@ -447,11 +509,12 @@ func TestOnUpdateStorageBucket(t *testing.T) {
 	require.NoError(t, server.Client.MakeBucket(ctx, bucketName, minio.MakeBucketOptions{}))
 
 	tests := []struct {
-		name     string
-		req      *plugin.OnUpdateStorageBucketRequest
-		expected *plugin.OnUpdateStorageBucketResponse
-		err      string
-		errCode  codes.Code
+		name                string
+		req                 *plugin.OnUpdateStorageBucketRequest
+		expected            *plugin.OnUpdateStorageBucketResponse
+		err                 string
+		expectedPermissions *pb.Permissions
+		errCode             codes.Code
 	}{
 		{
 			name:    "nilNewBucket",
@@ -764,43 +827,6 @@ func TestOnUpdateStorageBucket(t *testing.T) {
 			err:     "cannot disable rotation for already-rotated credentials",
 			errCode: codes.InvalidArgument,
 		},
-		{
-			name: "usingNonServiceAccount",
-			req: &plugin.OnUpdateStorageBucketRequest{
-				NewBucket: &storagebuckets.StorageBucket{
-					BucketName: bucketName,
-					Attributes: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
-						},
-					},
-					Secrets: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
-							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
-						},
-					},
-				},
-				CurrentBucket: &storagebuckets.StorageBucket{
-					BucketName: bucketName,
-					Attributes: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
-						},
-					},
-				},
-				Persisted: &storagebuckets.StorageBucketPersisted{
-					Data: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
-							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
-						},
-					},
-				},
-			},
-			err:     "The specified service account is not found (Specified service account does not exist)",
-			errCode: codes.InvalidArgument,
-		},
 		{
 			name: "dryRunFail",
 			req: &plugin.OnUpdateStorageBucketRequest{
@@ -845,6 +871,11 @@ func TestOnUpdateStorageBucket(t *testing.T) {
 			},
 			err:     "failed to verify provided minio environment: failed to put object",
 			errCode: codes.InvalidArgument,
+			expectedPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "Access Denied."},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: "The specified key does not exist."},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_OK, ErrorDetails: ""},
+			},
 		},
 		{
 			name: "credRotationFailWithNewCreds",
@@ -939,6 +970,52 @@ func TestOnUpdateStorageBucket(t *testing.T) {
 			err:     "credential rotation enabled with no new credentials provided",
 			errCode: codes.InvalidArgument,
 		},
+		{
+			name: "successUsingNonServiceAccount",
+			req: &plugin.OnUpdateStorageBucketRequest{
+				NewBucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl:               structpb.NewStringValue("http://" + server.ApiAddr),
+							ConstDisableCredentialRotation: structpb.NewBoolValue(true),
+						},
+					},
+					Secrets: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+						},
+					},
+				},
+				CurrentBucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
+						},
+					},
+				},
+				Persisted: &storagebuckets.StorageBucketPersisted{
+					Data: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+						},
+					},
+				},
+			},
+			expected: &plugin.OnUpdateStorageBucketResponse{
+				Persisted: &storagebuckets.StorageBucketPersisted{
+					Data: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "successNoRotation",
 			req: &plugin.OnUpdateStorageBucketRequest{
@@ -996,6 +1073,29 @@ func TestOnUpdateStorageBucket(t *testing.T) {
 				require.ErrorContains(err, tt.err)
 				require.Equal(tt.errCode.String(), status.Code(err).String())
 				require.Nil(res)
+
+				// Check the permission error.
+				if tt.expectedPermissions != nil {
+					st, ok := status.FromError(err)
+					require.True(ok)
+					var permission *pb.StorageBucketCredentialState
+					for _, detail := range st.Details() {
+						if errDetail, ok := detail.(*pb.StorageBucketCredentialState); ok {
+							permission = errDetail
+							break
+						}
+					}
+					require.NotNil(t, permission)
+
+					assert.Equal(t, tt.expectedPermissions.GetWrite().GetState(), permission.GetState().GetWrite().GetState())
+					assert.Equal(t, tt.expectedPermissions.GetRead().GetState(), permission.GetState().GetRead().GetState())
+					assert.Equal(t, tt.expectedPermissions.GetDelete().GetState(), permission.GetState().GetDelete().GetState())
+
+					assert.Contains(t, tt.expectedPermissions.GetWrite().GetErrorDetails(), permission.GetState().GetWrite().GetErrorDetails())
+					assert.Contains(t, tt.expectedPermissions.GetRead().GetErrorDetails(), permission.GetState().GetRead().GetErrorDetails())
+					assert.Contains(t, tt.expectedPermissions.GetDelete().GetErrorDetails(), permission.GetState().GetDelete().GetErrorDetails())
+				}
+
 				return
 			}
 
@@ -1644,26 +1744,6 @@ func TestValidatePermissions(t *testing.T) {
 			},
 			expErrMsg: "empty secrets input",
 		},
-		{
-			name: "usingNonServiceAccount",
-			req: &plugin.ValidatePermissionsRequest{
-				Bucket: &storagebuckets.StorageBucket{
-					BucketName: bucketName,
-					Attributes: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
-						},
-					},
-					Secrets: &structpb.Struct{
-						Fields: map[string]*structpb.Value{
-							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
-							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
-						},
-					},
-				},
-			},
-			expErrMsg: "The specified service account is not found (Specified service account does not exist)",
-		},
 		{
 			name: "dryRunFail",
 			req: &plugin.ValidatePermissionsRequest{
@@ -1691,6 +1771,46 @@ func TestValidatePermissions(t *testing.T) {
 			},
 			expErrMsg: "failed to verify provided minio environment: failed to put object",
 		},
+		{
+			name: "rotatedNonServiceAccount",
+			req: &plugin.ValidatePermissionsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
+						},
+					},
+					Secrets: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+							ConstLastRotatedTime: structpb.NewStringValue(time.Now().Format(time.RFC3339Nano)),
+						},
+					},
+				},
+			},
+			expErrMsg: "The specified service account is not found (Specified service account does not exist)",
+		},
+		{
+			name: "successNonServiceAccount",
+			req: &plugin.ValidatePermissionsRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
+						},
+					},
+					Secrets: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.RootUsername),
+							ConstSecretAccessKey: structpb.NewStringValue(server.RootPassword),
+						},
+					},
+				},
+			},
+		},
 		{
 			name: "success",
 			req: &plugin.ValidatePermissionsRequest{
@@ -1848,7 +1968,7 @@ func TestHeadObject(t *testing.T) {
 				},
 				Key: "doesnt-exist",
 			},
-			expErrMsg: "failed to stat object: The specified key does not exist.",
+			expErrMsg: "specified object does not exist",
 		},
 		{
 			name: "success",
@@ -2320,7 +2440,7 @@ func TestPutObject(t *testing.T) {
 	cases := []struct {
 		name    string
 		req     *plugin.PutObjectRequest
-		setup   func(*plugin.PutObjectRequest) (string, error)
+		setup   func(*plugin.PutObjectRequest) ([]byte, error)
 		err     string
 		errCode codes.Code
 	}{
@@ -2425,13 +2545,13 @@ func TestPutObject(t *testing.T) {
 				Key:  "test-key",
 				Path: dirPath("empty-test-file"),
 			},
-			setup: func(req *plugin.PutObjectRequest) (string, error) {
+			setup: func(req *plugin.PutObjectRequest) ([]byte, error) {
 				file, err := os.Create(req.Path)
 				if err != nil {
-					return "", err
+					return nil, err
 				}
 				file.Close()
-				return "", nil
+				return []byte{}, nil
 			},
 			err:     "file is empty",
 			errCode: codes.InvalidArgument,
@@ -2479,17 +2599,17 @@ func TestPutObject(t *testing.T) {
 				Key:  "test-key",
 				Path: dirPath("test-file"),
 			},
-			setup: func(req *plugin.PutObjectRequest) (string, error) {
+			setup: func(req *plugin.PutObjectRequest) ([]byte, error) {
 				file, err := os.Create(req.Path)
 				if err != nil {
-					return "", err
+					return nil, err
 				}
 				defer file.Close()
 
-				data := "test file data"
+				data := []byte("test file data")
 
-				if _, err = file.WriteString(data); err != nil {
-					return "", err
+				if _, err = file.Write(data); err != nil {
+					return nil, err
 				}
 				return data, nil
 			},
@@ -2515,17 +2635,59 @@ func TestPutObject(t *testing.T) {
 				Key:  "test-key-with-prefix",
 				Path: dirPath("test-file-with-prefix"),
 			},
-			setup: func(req *plugin.PutObjectRequest) (string, error) {
+			setup: func(req *plugin.PutObjectRequest) ([]byte, error) {
 				file, err := os.Create(req.Path)
 				if err != nil {
-					return "", err
+					return nil, err
 				}
 				defer file.Close()
 
-				data := "test file data for bucket with prefix"
+				data := []byte("test file data for bucket with prefix")
 
-				if _, err = file.WriteString(data); err != nil {
-					return "", err
+				if _, err = file.Write(data); err != nil {
+					return nil, err
+				}
+				return data, nil
+			},
+		},
+		{
+			name: "successWithLargeFile",
+			req: &plugin.PutObjectRequest{
+				Bucket: &storagebuckets.StorageBucket{
+					BucketName: bucketName,
+					Attributes: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstEndpointUrl: structpb.NewStringValue("http://" + server.ApiAddr),
+						},
+					},
+					Secrets: &structpb.Struct{
+						Fields: map[string]*structpb.Value{
+							ConstAccessKeyId:     structpb.NewStringValue(server.ServiceAccountAccessKeyId),
+							ConstSecretAccessKey: structpb.NewStringValue(server.ServiceAccountSecretAccessKey),
+						},
+					},
+				},
+				Key:  "test-key-large-file",
+				Path: dirPath("test-file-large"),
+			},
+			setup: func(req *plugin.PutObjectRequest) ([]byte, error) {
+				file, err := os.Create(req.Path)
+				if err != nil {
+					return nil, err
+				}
+				defer file.Close()
+
+				data := make([]byte, 30e6)
+				n, err := rand.Reader.Read(data)
+				if err != nil {
+					return nil, err
+				}
+				if n != 30e6 {
+					return nil, fmt.Errorf("did not read 30 million bytes")
+				}
+
+				if _, err = file.Write(data); err != nil {
+					return nil, err
 				}
 				return data, nil
 			},
@@ -2537,13 +2699,13 @@ func TestPutObject(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			assert, require := assert.New(t), require.New(t)
 			var checksum []byte
-			var content string
+			var content []byte
 			if tt.setup != nil {
 				var err error
 				content, err = tt.setup(tt.req)
 				require.NoError(err)
 				hash := sha256.New()
-				_, err = hash.Write([]byte(content))
+				_, err = hash.Write(content)
 				require.NoError(err)
 				checksum = hash.Sum(nil)
 				require.NoError(err)
@@ -2572,14 +2734,16 @@ func TestPutObject(t *testing.T) {
 			require.NoError(err)
 			defer obj.Close()
 
-			contentLen := len([]byte(content))
+			fmt.Printf("path: %v\n", tt.req.Path)
+			contentLen := len(content)
 			reader := bufio.NewReader(obj)
 			buffer := make([]byte, contentLen)
-			n, err := reader.Read(buffer)
-			require.NoError(err)
+			n, _ := reader.Read(buffer)
+			// large files, for some reason, yield an EOF error when you read the full thing
+			// require.NoError(err)
 
 			assert.Equal(contentLen, n)
-			assert.Equal(content, string(buffer))
+			assert.Equal(content, buffer)
 		})
 	}
 }
@@ -2882,12 +3046,18 @@ func TestDryRun(t *testing.T) {
 		makeBucketFn func(t *testing.T, cl *minio.Client) string
 		// This test verifies that each message in the err slice is contained in
 		// the error string, effectively &&-ing the checks together.
-		expErrMsgs []string
+		expErrMsgs     []string
+		expPermissions *pb.Permissions
 	}{
 		{
 			name:         "bucketNotExists",
 			makeBucketFn: func(t *testing.T, cl *minio.Client) string { return "unknownbucket" },
 			expErrMsgs:   []string{"The specified bucket does not exist"},
+			expPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "The specified bucket does not exist"},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "The specified bucket does not exist"},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "The specified bucket does not exist"},
+			},
 		},
 		{
 			name: "putObjectFail",
@@ -2916,6 +3086,12 @@ func TestDryRun(t *testing.T) {
 				"failed to put object",
 				"Access Denied",
 			},
+			expPermissions: &pb.Permissions{
+				Write: &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "Access Denied."},
+				// The write permission is denied, so the any read operation will return with a no such key error.
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_UNKNOWN, ErrorDetails: "The specified key does not exist."},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+			},
 		},
 		{
 			name: "statObjectFail",
@@ -2944,6 +3120,11 @@ func TestDryRun(t *testing.T) {
 				"failed to stat object",
 				"Access Denied",
 			},
+			expPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "Access Denied."},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+			},
 		},
 		{
 			name: "listObjectsFail",
@@ -2972,6 +3153,11 @@ func TestDryRun(t *testing.T) {
 				"failed to list objects in bucket",
 				"Access Denied",
 			},
+			expPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "Access Denied."},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+			},
 		},
 		{
 			name: "removeObjectFail",
@@ -3000,6 +3186,37 @@ func TestDryRun(t *testing.T) {
 				"failed to remove object",
 				"Access Denied",
 			},
+			expPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "Access Denied."},
+			},
+		},
+		{
+			name: "Invalid Access Key Id",
+			minioClFn: func(t *testing.T, s *internaltest.MinioServer) *minio.Client {
+				cl, err := minio.New(s.ApiAddr, &minio.Options{
+					Creds:  credentials.NewStaticV4("invalidAccessKey", "invalidSecretKey", ""),
+					Secure: false,
+				})
+				require.NoError(t, err)
+
+				return cl
+			},
+			makeBucketFn: func(t *testing.T, cl *minio.Client) string {
+				bucketName := "testdryrun-invalidaccesskeyid"
+				_ = cl.MakeBucket(ctx, bucketName, minio.MakeBucketOptions{})
+
+				return bucketName
+			},
+			expErrMsgs: []string{
+				"The Access Key Id you provided does not exist in our records",
+			},
+			expPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "The Access Key Id you provided does not exist in our records."},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "The Access Key Id you provided does not exist in our records."},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_ERROR, ErrorDetails: "The Access Key Id you provided does not exist in our records."},
+			},
 		},
 		{
 			name: "success",
@@ -3009,6 +3226,11 @@ func TestDryRun(t *testing.T) {
 				require.NoError(t, err)
 				return bucketName
 			},
+			expPermissions: &pb.Permissions{
+				Write:  &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+				Read:   &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+				Delete: &pb.Permission{State: pb.StateType_STATE_TYPE_OK},
+			},
 		},
 	}
 
@@ -3021,17 +3243,27 @@ func TestDryRun(t *testing.T) {
 			}
 
 			bucketName := tt.makeBucketFn(t, minioClient)
-			err := dryRun(ctx, minioClient, &storagebuckets.StorageBucket{
+			per, err := dryRun(ctx, minioClient, &storagebuckets.StorageBucket{
 				BucketName: bucketName,
 			})
+			if tt.expPermissions != nil {
+				assert.Equal(t, tt.expPermissions.GetWrite().GetState(), per.GetWrite().GetState())
+				assert.Equal(t, tt.expPermissions.GetRead().GetState(), per.GetRead().GetState())
+				assert.Equal(t, tt.expPermissions.GetDelete().GetState(), per.GetDelete().GetState())
+
+				assert.Contains(t, tt.expPermissions.GetWrite().GetErrorDetails(), per.GetWrite().GetErrorDetails())
+				assert.Contains(t, tt.expPermissions.GetRead().GetErrorDetails(), per.GetRead().GetErrorDetails())
+				assert.Contains(t, tt.expPermissions.GetDelete().GetErrorDetails(), per.GetDelete().GetErrorDetails())
+			}
+			require.NotNil(t, per)
 			if len(tt.expErrMsgs) > 0 {
 				for _, errMsg := range tt.expErrMsgs {
-					require.ErrorContains(t, err, errMsg)
+					require.ErrorContains(t, err.Unwrap(), errMsg)
 				}
 				return
 			}
 
-			require.NoError(t, err)
+			require.NoError(t, err.Unwrap())
 		})
 	}
 }