fix(storage): Static rotated creds being deleted on plugin err

On Storage Bucket update, when we determine if the credentials are to be
updated to new ones, we call `credState.ReplaceCreds`.

This call, if the old credentials were managed by us (rotated at create
time), will delete the old credentials and set the new ones onto
credState.

We called this function without validating if the incoming credentials
are valid. This could get the bucket in a state where where, while we'd
return an update error if the incoming credentials weren't working with
S3 (err in the dry run process that happens at the end of the update),
we'd also silently delete the old credentials the bucket used, rendering
the bucket completely broken.

internal/credential/attributes.go

@@ -79,7 +79,7 @@ func GetCredentialsConfig(secrets *structpb.Struct, attrs *CredentialAttributes,
 	case accessKey != "" && secretKey != "" && attrs.RoleArn != "":
 		badFields[fmt.Sprintf("secrets.%s", ConstAccessKeyId)] = "conflicts with role_arn value"
 		badFields[fmt.Sprintf("secrets.%s", ConstSecretAccessKey)] = "conflicts with role_arn value"
-		badFields[fmt.Sprintf("secrets.%s", ConstRoleArn)] = "conflicts with access_key_id and secret_access_key values"
+		badFields[fmt.Sprintf("attributes.%s", ConstRoleArn)] = "conflicts with access_key_id and secret_access_key values"
 	// static credential is missing it's secret_access_key
 	case accessKey != "" && secretKey == "":
 		badFields[fmt.Sprintf("secrets.%s", ConstSecretAccessKey)] = "missing required value"

internal/credential/attributes_test.go

@@ -244,7 +244,7 @@ func TestGetCredentialsConfig(t *testing.T) {
 					"foo": "bar",
 				},
 			},
-			expectedErrContains: "secrets.access_key_id: conflicts with role_arn value, secrets.role_arn: conflicts with access_key_id and secret_access_key values, secrets.secret_access_key: conflicts with role_arn value",
+			expectedErrContains: "attributes.role_arn: conflicts with access_key_id and secret_access_key values, secrets.access_key_id: conflicts with role_arn value, secrets.secret_access_key: conflicts with role_arn value",
 		},
 		{
 			name:    "with dynamic credentials and no disable credential rotation",

plugin/service/storage/plugin.go

@@ -161,6 +161,28 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
 	// 1. the new bucket has secrets for static credentials, in which case we need to delete the old static credentials
 	// 2. the new bucket has a different roleARN value for dynamic credentials
 	if newBucket.GetSecrets() != nil || newStorageAttributes.RoleArn != oldStorageAttributes.RoleArn {
+		// Ensure the incoming credentials are valid for interaction with the S3
+		// Bucket before replacing them.
+		newCredState, err := cred.NewAwsCredentialPersistedState(
+			append([]cred.AwsCredentialPersistedStateOption{
+				cred.WithCredentialsConfig(updatedCredentials),
+			}, p.testCredStateOpts...)...,
+		)
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "error setting up new credential persisted state: %s", err)
+		}
+		newStorageState, err := newAwsStoragePersistedState(
+			append([]awsStoragePersistedStateOption{
+				withCredentials(newCredState),
+			}, p.testStorageStateOpts...)...,
+		)
+		if err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "error loading persisted state: %s", err)
+		}
+		if err := dryRunValidation(ctx, newStorageState, newStorageAttributes, newBucket); err != nil {
+			return nil, status.Errorf(codes.InvalidArgument, "failed to validate incoming credentials: %s", err)
+		}
+
 		// Replace the existing credential state.
 		// This checks the timestamp on the last rotation time as well
 		// and deletes the credentials if we are managing them

plugin/service/storage/plugin_test.go

@@ -663,6 +663,16 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 					},
 				},
 			},
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
+						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
+						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+					),
+				),
+			},
 			credOpts: []credential.AwsCredentialPersistedStateOption{
 				credential.WithStateTestOpts([]awsutilv2.Option{
 					awsutilv2.WithIAMAPIFunc(
@@ -933,7 +943,7 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 				),
 			},
 			expectedErrContains: "head object failed oops",
-			expectedErrCode:     codes.Internal,
+			expectedErrCode:     codes.InvalidArgument,
 		},
 		{
 			name: "dynamic to dynamic credentials success",
@@ -1014,7 +1024,7 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 				),
 			},
 			expectedErrContains: "put object failed oops",
-			expectedErrCode:     codes.Internal,
+			expectedErrCode:     codes.InvalidArgument,
 		},
 		{
 			name: "success static non rotated to dynamic credentials",
@@ -1090,6 +1100,16 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 					},
 				},
 			},
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
+						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
+						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+					),
+				),
+			},
 			credOpts: []credential.AwsCredentialPersistedStateOption{
 				credential.WithStateTestOpts([]awsutilv2.Option{
 					awsutilv2.WithIAMAPIFunc(
@@ -1195,7 +1215,7 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 				),
 			},
 			expectedErrContains: "put object fail oops",
-			expectedErrCode:     codes.Internal,
+			expectedErrCode:     codes.InvalidArgument,
 		},
 		{
 			name: "success dynamic to static non-rotated credentials",
@@ -1275,6 +1295,16 @@ func TestStoragePlugin_OnUpdateStorageBucket(t *testing.T) {
 					Data: &structpb.Struct{Fields: map[string]*structpb.Value{}},
 				},
 			},
+			storageOpts: []awsStoragePersistedStateOption{
+				withTestS3APIFunc(
+					newTestMockS3(
+						nil,
+						testMockS3WithPutObjectOutput(&s3.PutObjectOutput{}),
+						testMockS3WithGetObjectOutput(&s3.GetObjectOutput{}),
+						testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{}),
+					),
+				),
+			},
 			credOpts: []credential.AwsCredentialPersistedStateOption{
 				credential.WithStateTestOpts([]awsutilv2.Option{
 					awsutilv2.WithSTSAPIFunc(