[fix](vault) avoid encrypt twice when altering vault (#45156)

Author: gavinchou

cloud/src/meta-service/meta_service_resource.cpp

@@ -648,10 +648,19 @@ static int alter_s3_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Tran
         obj_info.has_provider()) {
         code = MetaServiceCode::INVALID_ARGUMENT;
         std::stringstream ss;
-        ss << "Only ak, sk can be altered";
+        ss << "Bucket, endpoint, prefix and provider can not be altered";
         msg = ss.str();
         return -1;
     }
+
+    if (obj_info.has_ak() ^ obj_info.has_sk()) {
+        code = MetaServiceCode::INVALID_ARGUMENT;
+        std::stringstream ss;
+        ss << "Accesskey and secretkey must be alter together";
+        msg = ss.str();
+        return -1;
+    }
+
     const auto& name = vault.name();
     // Here we try to get mutable iter since we might need to alter the vault name
     auto name_itr = std::find_if(instance.mutable_storage_vault_names()->begin(),
@@ -703,22 +712,25 @@ static int alter_s3_storage_vault(InstanceInfoPB& instance, std::unique_ptr<Tran
         *name_itr = vault.alter_name();
     }
     auto origin_vault_info = new_vault.DebugString();
-    AkSkPair pre {new_vault.obj_info().ak(), new_vault.obj_info().sk()};
-    const auto& plain_ak = obj_info.has_ak() ? obj_info.ak() : new_vault.obj_info().ak();
-    const auto& plain_sk = obj_info.has_ak() ? obj_info.sk() : new_vault.obj_info().sk();
-    AkSkPair plain_ak_sk_pair {plain_ak, plain_sk};
-    AkSkPair cipher_ak_sk_pair;
-    EncryptionInfoPB encryption_info;
-    auto ret = encrypt_ak_sk_helper(plain_ak, plain_sk, &encryption_info, &cipher_ak_sk_pair, code,
-                                    msg);
-    if (ret != 0) {
-        msg = "failed to encrypt";
-        code = MetaServiceCode::ERR_ENCRYPT;
-        LOG(WARNING) << msg;
-        return -1;
+
+    // For ak or sk is not altered.
+    EncryptionInfoPB encryption_info = new_vault.obj_info().encryption_info();
+    AkSkPair new_ak_sk_pair {new_vault.obj_info().ak(), new_vault.obj_info().sk()};
+
+    if (obj_info.has_ak()) {
+        // ak and sk must be altered together, there is check before.
+        auto ret = encrypt_ak_sk_helper(obj_info.ak(), obj_info.sk(), &encryption_info,
+                                        &new_ak_sk_pair, code, msg);
+        if (ret != 0) {
+            msg = "failed to encrypt";
+            code = MetaServiceCode::ERR_ENCRYPT;
+            LOG(WARNING) << msg;
+            return -1;
+        }
     }
-    new_vault.mutable_obj_info()->set_ak(cipher_ak_sk_pair.first);
-    new_vault.mutable_obj_info()->set_sk(cipher_ak_sk_pair.second);
+
+    new_vault.mutable_obj_info()->set_ak(new_ak_sk_pair.first);
+    new_vault.mutable_obj_info()->set_sk(new_ak_sk_pair.second);
     new_vault.mutable_obj_info()->mutable_encryption_info()->CopyFrom(encryption_info);
     if (obj_info.has_use_path_style()) {
         new_vault.mutable_obj_info()->set_use_path_style(obj_info.use_path_style());

regression-test/suites/vault_p0/alter/test_alter_s3_vault.groovy

@@ -62,6 +62,25 @@ suite("test_alter_s3_vault", "nonConcurrent") {
         """
     }, "Alter property")
 
+    expectExceptionLike({
+        sql """
+            ALTER STORAGE VAULT ${suiteName}
+            PROPERTIES (
+            "type"="S3",
+            "s3.access_key" = "new_ak"
+            );
+        """
+    }, "Alter property")
+
+    expectExceptionLike({
+        sql """
+            ALTER STORAGE VAULT ${suiteName}
+            PROPERTIES (
+            "type"="S3",
+            "s3.secret_key" = "new_sk"
+            );
+        """
+    }, "Alter property")
 
     def vaultName = suiteName
     String properties;
@@ -75,28 +94,105 @@ suite("test_alter_s3_vault", "nonConcurrent") {
         }
     }
 
-    def newVaultName = suiteName + "_new";
+    // alter ak sk
+    sql """
+        ALTER STORAGE VAULT ${vaultName}
+        PROPERTIES (
+            "type"="S3",
+            "s3.access_key" = "${getS3AK()}",
+            "s3.secret_key" = "${getS3SK()}"
+        );
+    """
+
+    vaultInfos = sql """SHOW STORAGE VAULT;"""
+
+    for (int i = 0; i < vaultInfos.size(); i++) {
+        def name = vaultInfos[i][0]
+        logger.info("name is ${name}, info ${vaultInfos[i]}")
+        if (name.equals(vaultName)) {
+            assert properties == newProperties, "Properties are not the same"
+        }
+    }
+
+    sql """insert into alter_s3_vault_tbl values("2", "2"); """
+
+
+    // rename
+    newVaultName = vaultName + "_new";
+
+    sql """
+        ALTER STORAGE VAULT ${vaultName}
+        PROPERTIES (
+            "type"="S3",
+            "VAULT_NAME" = "${newVaultName}"
+        );
+    """
+
+    vaultInfos = sql """SHOW STORAGE VAULT;"""
+    for (int i = 0; i < vaultInfos.size(); i++) {
+        def name = vaultInfos[i][0]
+        logger.info("name is ${name}, info ${vaultInfos[i]}")
+        if (name.equals(newVaultName)) {
+            assert properties == newProperties, "Properties are not the same"
+        }
+        if (name.equals(vaultName)) {
+            assertTrue(false);
+        }
+    }
+
+    sql """insert into alter_s3_vault_tbl values("2", "2"); """
+
+    // rename + aksk
+    vaultName = newVaultName
 
     sql """
         ALTER STORAGE VAULT ${vaultName}
         PROPERTIES (
             "type"="S3",
             "VAULT_NAME" = "${newVaultName}",
-            "s3.access_key" = "new_ak"
+            "s3.access_key" = "${getS3AK()}",
+            "s3.secret_key" = "${getS3SK()}"
         );
     """
 
+    vaultInfos = sql """SHOW STORAGE VAULT;"""
+    for (int i = 0; i < vaultInfos.size(); i++) {
+        def name = vaultInfos[i][0]
+        logger.info("name is ${name}, info ${vaultInfos[i]}")
+        if (name.equals(newVaultName)) {
+            assert properties == newProperties, "Properties are not the same"
+        }
+        if (name.equals(vaultName)) {
+            assertTrue(false);
+        }
+    }
+    sql """insert into alter_s3_vault_tbl values("2", "2"); """
+
+
+    vaultName = newVaultName;
+    newVaultName = vaultName + "_new";
+
     vaultInfos = sql """SHOW STORAGE VAULT;"""
     boolean exist = false
 
+    sql """
+        ALTER STORAGE VAULT ${vaultName}
+        PROPERTIES (
+            "type"="S3",
+            "VAULT_NAME" = "${newVaultName}",
+            "s3.access_key" = "new_ak_ak",
+            "s3.secret_key" = "sk"
+        );
+    """
+
     for (int i = 0; i < vaultInfos.size(); i++) {
         def name = vaultInfos[i][0]
         logger.info("name is ${name}, info ${vaultInfos[i]}")
         if (name.equals(vaultName)) {
             assertTrue(false);
         }
         if (name.equals(newVaultName)) {
-            assertTrue(vaultInfos[i][2].contains("new_ak"))
+            assertTrue(vaultInfos[i][2].contains("new_ak_ak"))
             exist = true
         }
     }