feat(encryption): add kms key management

src/replica/pegasus_kms_key_provider.h

@@ -31,7 +31,7 @@ class PegasusKMSKeyProvider : public KeyProvider
 public:
     ~PegasusKMSKeyProvider() override {}
 
-    PegasusKMSKeyProvider(const std::string &kms_url, std::string cluster_key_name)
+    PegasusKMSKeyProvider(const std::vector<std::string> &kms_url, std::string cluster_key_name)
         : client_(kms_url, std::move(cluster_key_name))
     {
     }

src/replica/replica_stub.cpp

@@ -220,7 +220,7 @@ METRIC_DEFINE_gauge_int64(server,
                           "The max size of copied files among all splitting replicas");
 
 DSN_DECLARE_bool(encrypt_data_at_rest);
-DSN_DECLARE_string(server_encrypted_key);
+DSN_DECLARE_string(server_key);
 
 namespace dsn {
 
@@ -282,16 +282,6 @@ DSN_DEFINE_uint32(
     "The interval milliseconds of replica server to syncs replica configuration with meta server");
 DSN_TAG_VARIABLE(config_sync_interval_ms, FT_MUTABLE);
 DSN_DEFINE_validator(config_sync_interval_ms, [](uint32_t value) -> bool { return value > 0; });
-DSN_DEFINE_group_validator(encrypt_data_at_rest, [](std::string &message) -> bool {
-    if (!dsn::security::FLAGS_enable_acl || !FLAGS_encrypt_data_at_rest) {
-        message = fmt::format("FLAGS_enable_acl = ({}) FLAGS_encrypt_data_at_rest = ({}),should be "
-                              "true at the same time",
-                              dsn::security::FLAGS_enable_acl,
-                              FLAGS_encrypt_data_at_rest);
-        return false;
-    }
-    return true;
-});
 
 DSN_DEFINE_int32(replication,
                  disk_stat_interval_seconds,
@@ -333,7 +323,7 @@ DSN_DEFINE_string(pegasus.server,
 DSN_DEFINE_string(pegasus.server,
                   hadoop_kms_url,
                   "",
-                  "Where the server encrypted key of file system can get from.");
+                  "Where the server key of file system can get from.");
 
 DSN_DECLARE_bool(duplication_enabled);
 DSN_DECLARE_int32(fd_beacon_interval_seconds);
@@ -399,7 +389,8 @@ replica_stub::replica_stub(replica_state_subscriber subscriber /*= nullptr*/,
     _primary_address_str[0] = '\0';
     if (FLAGS_encrypt_data_at_rest) {
         key_provider.reset(new dsn::security::PegasusKMSKeyProvider(
-            FLAGS_hadoop_kms_url, FLAGS_encryption_cluster_key_name));
+            ::absl::StrSplit(FLAGS_hadoop_kms_url, ",", ::absl::SkipEmpty()),
+            FLAGS_encryption_cluster_key_name));
     }
 }
 
@@ -414,9 +405,9 @@ void replica_stub::initialize(bool clear /* = false*/)
 }
 
 dsn::error_s store_kms_key(const std::string &data_dir,
-                           std::string encryption_key,
-                           std::string iv,
-                           std::string key_version)
+                           const std::string &encryption_key,
+                           const std::string &iv,
+                           const std::string &key_version)
 {
     replica_kms_info kms_info(encryption_key, iv, key_version);
     auto err = kms_info.store(data_dir);
@@ -488,7 +479,7 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
         }
         CHECK(key_provider->DecryptEncryptionKey(encryption_key, iv, key_version, &server_key),
               "get decryption key failed");
-        FLAGS_server_encrypted_key = server_key.c_str();
+        FLAGS_server_key = server_key.c_str();
     }
 
     // Initialize the file system manager.

src/runtime/security/kms_client.h

@@ -27,7 +27,7 @@
 namespace dsn {
 namespace security {
 // A class to generate encryption_key from kms for writing file which implemented based on http
-// client
+// client.
 // This class is not thread-safe. Thus maintain one instance for each thread.
 //
 // Example of using Kms client:
@@ -45,17 +45,19 @@ namespace security {
 class KMSClient
 {
 public:
-    KMSClient(const std::string kms_url, std::string cluster_key_name)
-        : kms_urls_(::absl::StrSplit(kms_url, ",", ::absl::SkipEmpty())),
+    KMSClient(const std::vector<std::string> &kms_url, std::string cluster_key_name)
+        : kms_urls_(kms_url),
           cluster_key_name_(std::move(cluster_key_name))
     {
     }
 
+    // Decrypt 
     dsn::error_s DecryptEncryptionKey(const std::string &encryption_key,
                                       const std::string &iv,
                                       const std::string &key_version,
                                       std::string *decrypted_key);
 
+    // GenerateEncryption key from kms
     dsn::error_s
     GenerateEncryptionKey(std::string *encryption_key, std::string *iv, std::string *key_version);
 

src/runtime/security/replica_kms_info.h

@@ -1,27 +1,20 @@
 /*
- * The MIT License (MIT)
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
  *
- * Copyright (c) 2015 Microsoft Corporation
+ *   http://www.apache.org/licenses/LICENSE-2.0
  *
- * -=- Robust Distributed System Nucleus (rDSN) -=-
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy
- * of this software and associated documentation files (the "Software"), to deal
- * in the Software without restriction, including without limitation the rights
- * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
- * copies of the Software, and to permit persons to whom the Software is
- * furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in
- * all copies or substantial portions of the Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
- * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
- * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
- * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
- * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
- * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
- * THE SOFTWARE.
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
  */
 
 #pragma once
@@ -60,4 +53,4 @@ class replica_kms_info
 };
 
 } // namespace replication
-} // namespace dsn
+} // namespace dsn

src/utils/env.cpp

@@ -39,7 +39,7 @@ DSN_DEFINE_bool(pegasus.server,
                 "Whether the sensitive files should be encrypted on the file system.");
 
 DSN_DEFINE_string(pegasus.server,
-                  server_encrypted_key,
+                  server_key,
                   "0123456789ABCDEF0123456789ABCDEF",
                   "The encrypted server key to use in the filesystem.");
 
@@ -54,6 +54,17 @@ DSN_DEFINE_bool(replication,
                 false,
                 "Whether to enable direct I/O when download files");
 DSN_TAG_VARIABLE(enable_direct_io, FT_MUTABLE);
+DSN_TAG_VARIABLE(encrypt_data_at_rest);
+DSN_DEFINE_group_validator(encrypt_data_at_rest, [](std::string &message) -> bool {
+    if (!dsn::security::FLAGS_enable_acl && FLAGS_encrypt_data_at_rest) {
+        message = fmt::format("[security] enable_acl = ({}) [pegasus.server] encrypt_data_at_rest = ({}),should be "
+                              "true at the same time",
+                              dsn::security::FLAGS_enable_acl,
+                              FLAGS_encrypt_data_at_rest);
+        return false;
+    }
+    return true;
+});
 
 namespace dsn {
 namespace utils {
@@ -63,7 +74,7 @@ rocksdb::Env *NewEncryptedEnv()
     // Create an encryption provider.
     std::shared_ptr<rocksdb::EncryptionProvider> provider;
     auto provider_id = fmt::format("id=AES;hex_instance_key={};method={}",
-                                   FLAGS_server_encrypted_key,
+                                   FLAGS_server_key,
                                    FLAGS_encryption_method);
     auto s = rocksdb::EncryptionProvider::CreateFromString(
         rocksdb::ConfigOptions(), provider_id, &provider);