Worker storage bucket credential state (#18)

* feat: parse minio errors and plugin permission

- Evaluate minio errors. If the error is a permission related error, convert the error to a `Permission` proto and attach it to the gRPC status
- Update `dryRun` method to join all errors before returning response. This ensures all the permissions are validated before returning

* refactor errorToStatus to return storage states

* update oncreate and validate errorcodes on failure

* update go.mod for new sdk version

* update tests to parse state instead of permissions

---------

Co-authored-by: Elim Tsiagbey <elim.tsiagbey@hashicorp.com>

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"
@@ -79,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())
@@ -209,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())
@@ -300,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
@@ -338,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{
@@ -383,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()
 
@@ -472,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()
@@ -493,7 +514,7 @@ func (sp *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest
 		},
 	})
 	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()
 	}
 
 	resChecksum := res.ChecksumSHA256
@@ -547,7 +568,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())
@@ -559,7 +580,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),
@@ -572,7 +593,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)
 	}
@@ -589,7 +610,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++
 	}
@@ -599,37 +620,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) {
@@ -654,3 +783,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

@@ -191,11 +191,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",
@@ -261,6 +262,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",
@@ -430,6 +436,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
 			}
 
@@ -480,11 +508,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",
@@ -841,6 +870,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",
@@ -1038,6 +1072,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
 			}
 
@@ -1910,7 +1967,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",
@@ -2944,12 +3001,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",
@@ -2978,6 +3041,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",
@@ -3006,6 +3075,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",
@@ -3034,6 +3108,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",
@@ -3062,6 +3141,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",
@@ -3071,6 +3181,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},
+			},
 		},
 	}
 
@@ -3083,17 +3198,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())
 		})
 	}
 }