feat(encryption): add kms key management

src/replica/CMakeLists.txt

@@ -58,6 +58,7 @@ set(MY_PROJ_SRC
 set(MY_SRC_SEARCH_MODE "GLOB")
 
 set(MY_PROJ_LIBS
+        absl::strings
         dsn_replication_common
         dsn.failure_detector
         dsn.block_service
@@ -66,9 +67,11 @@ set(MY_PROJ_LIBS
         dsn_nfs
         dsn_dist_cmd
         dsn_http
+        curl
         dsn_runtime
         dsn_aio
         dsn_meta_server
+        dsn.security
         rocksdb
         lz4
         zstd

src/replica/kms_key_provider.cpp

@@ -0,0 +1,39 @@
+// 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
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// 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.
+
+#include <string>
+
+#include "replica/kms_key_provider.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace security {
+
+dsn::error_s
+KMSKeyProvider::DecryptEncryptionKey(const dsn::replication::replica_kms_info &kms_info,
+                                     std::string *decrypted_key)
+{
+    return client_.DecryptEncryptionKey(kms_info, decrypted_key);
+}
+
+dsn::error_s KMSKeyProvider::GenerateEncryptionKey(dsn::replication::replica_kms_info *kms_info)
+{
+    return client_.GenerateEncryptionKey(kms_info);
+}
+
+} // namespace security
+} // namespace dsn

src/replica/kms_key_provider.h

@@ -0,0 +1,56 @@
+// 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
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// 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
+
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "runtime/security/kms_client.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace replication {
+class replica_kms_info;
+} // namespace replication
+
+namespace security {
+// This class generates EEK IV KV from KMS (a.k.a Key Management Service) and retrieves DEK from
+// KMS.
+class KMSKeyProvider
+{
+public:
+    ~KMSKeyProvider() {}
+
+    KMSKeyProvider(const std::vector<std::string> &kms_url, std::string cluster_key_name)
+        : client_(kms_url, std::move(cluster_key_name))
+    {
+    }
+
+    // Decrypt the encryption key in 'kms_info' via KMS. The 'decrypted_key' will be a hex string.
+    dsn::error_s DecryptEncryptionKey(const dsn::replication::replica_kms_info &kms_info,
+                                      std::string *decrypted_key);
+
+    // Generate an encryption key from KMS.
+    dsn::error_s GenerateEncryptionKey(dsn::replication::replica_kms_info *kms_info);
+
+private:
+    KMSClient client_;
+};
+} // namespace security
+} // namespace dsn

src/replica/replica_stub.cpp

@@ -24,6 +24,7 @@
  * THE SOFTWARE.
  */
 
+#include <absl/strings/str_split.h>
 #include <boost/algorithm/string/replace.hpp>
 // IWYU pragma: no_include <ext/alloc_traits.h>
 #include <fmt/core.h>
@@ -56,11 +57,11 @@
 #include "nfs_types.h"
 #include "replica.h"
 #include "replica/duplication/replica_follower.h"
+#include "replica/kms_key_provider.h"
 #include "replica/replica_context.h"
 #include "replica/replica_stub.h"
 #include "replica/replication_app_base.h"
 #include "replica_disk_migrator.h"
-#include "replica_stub.h"
 #include "runtime/api_layer1.h"
 #include "ranger/access_type.h"
 #include "runtime/rpc/rpc_message.h"
@@ -214,7 +215,15 @@ METRIC_DEFINE_gauge_int64(server,
                           dsn::metric_unit::kBytes,
                           "The max size of copied files among all splitting replicas");
 
+DSN_DECLARE_bool(encrypt_data_at_rest);
+DSN_DECLARE_string(server_key);
+
 namespace dsn {
+DSN_DECLARE_string(cluster_name);
+
+namespace security {
+DSN_DECLARE_bool(enable_acl);
+}
 namespace replication {
 DSN_DEFINE_bool(replication,
                 deny_client_on_start,
@@ -292,12 +301,50 @@ DSN_DEFINE_int32(
     "if tcmalloc reserved but not-used memory exceed this percentage of application allocated "
     "memory, replica server will release the exceeding memory back to operating system");
 
+DSN_DEFINE_string(
+    pegasus.server,
+    hadoop_kms_url,
+    "",
+    "Provide the comma-separated list of URLs from which to retrieve the "
+    "file system's server key. Example format: 'hostname1:1234/kms,hostname2:1234/kms'.");
+
 DSN_DECLARE_bool(duplication_enabled);
 DSN_DECLARE_int32(fd_beacon_interval_seconds);
 DSN_DECLARE_int32(fd_check_interval_seconds);
 DSN_DECLARE_int32(fd_grace_seconds);
 DSN_DECLARE_int32(fd_lease_seconds);
 DSN_DECLARE_int32(gc_interval_ms);
+DSN_DECLARE_string(data_dirs);
+DSN_DEFINE_group_validator(encrypt_data_at_rest_pre_check, [](std::string &message) -> bool {
+    if (!dsn::security::FLAGS_enable_acl && FLAGS_encrypt_data_at_rest) {
+        message = fmt::format("[pegasus.server] encrypt_data_at_rest should be enabled only if "
+                              "[security] enable_acl is enabled.");
+        return false;
+    }
+    return true;
+});
+
+DSN_DEFINE_group_validator(encrypt_data_not_support_close, [](std::string &message) -> bool {
+    std::vector<std::string> dirs;
+    std::string data_dirs;
+    // In some unit test FLAGS_data_dirs may not set.
+    if (!dsn::utils::is_empty(FLAGS_data_dirs)) {
+        data_dirs = FLAGS_data_dirs;
+    } else {
+        return true;
+    }
+    ::absl::StrSplit(data_dirs.c_str(), dirs, ',');
+    std::string kms_path = utils::filesystem::path_combine(dirs[0], replica_kms_info::kKmsInfo);
+    if (!FLAGS_encrypt_data_at_rest && utils::filesystem::path_exists(kms_path)) {
+        message = fmt::format("The kms_info file exists at ({}), but [pegasus.server] "
+                              "encrypt_data_at_rest is set to ({})."
+                              "Encryption in Pegasus is irreversible after its initial activation.",
+                              kms_path,
+                              FLAGS_encrypt_data_at_rest);
+        return false;
+    }
+    return true;
+});
 
 bool replica_stub::s_not_exit_on_log_failure = false;
 
@@ -389,9 +436,45 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
         }
     }
 
+    std::string server_key;
+    dsn::replication::replica_kms_info kms_info;
+#ifndef MOCK_TEST
+    if (FLAGS_encrypt_data_at_rest && utils::is_empty(FLAGS_hadoop_kms_url)) {
+        CHECK(FLAGS_hadoop_kms_url, "hadoop_kms_url should not empty");
+    }
+#endif
+    if (FLAGS_encrypt_data_at_rest && !utils::is_empty(FLAGS_hadoop_kms_url)) {
+        key_provider.reset(new dsn::security::KMSKeyProvider(
+            ::absl::StrSplit(FLAGS_hadoop_kms_url, ",", ::absl::SkipEmpty()), FLAGS_cluster_name));
+        auto error_code = kms_info.load(_options.data_dirs[0]);
+        if (error_code != dsn::ERR_OK) {
+            LOG_WARNING("It's normal to encounter a temporary inability to open the kms-info file "
+                        "during the first process launch. error_code = {}",
+                        error_code);
+        }
+        // Upon the first launch, the encryption key should be empty. The process will then retrieve
+        // EEK, IV, and KV from KMS.
+        // After the first launch, the encryption key, obtained from the kms-info file, should not
+        // be empty. The process will then acquire the DEK from KMS.
+        std::string kms_path =
+            utils::filesystem::path_combine(_options.data_dirs[0], replica_kms_info::kKmsInfo);
+        if (!utils::filesystem::path_exists(kms_path)) {
+            auto err = key_provider->GenerateEncryptionKey(&kms_info);
+            CHECK(err, "get encryption key failed, err = {}", err);
+        }
+        auto err = key_provider->DecryptEncryptionKey(kms_info, &server_key);
+        CHECK(err, "get decryption key failed, err = {}", err);
+        FLAGS_server_key = server_key.c_str();
+    }
+
     // Initialize the file system manager.
     _fs_manager.initialize(_options.data_dirs, _options.data_dir_tags);
 
+    if (key_provider) {
+        auto err = kms_info.store(_options.data_dirs[0]);
+        CHECK_EQ_MSG(dsn::ERR_OK, err, "Can't store kms key to kms-info file, err = {}", err);
+    }
+
     // Check slog is not exist.
     auto full_slog_path = fmt::format("{}/replica/slog/", _options.slog_dir);
     if (utils::filesystem::directory_exists(full_slog_path)) {

src/replica/replica_stub.h

@@ -72,6 +72,9 @@ namespace dsn {
 class command_deregister;
 class message_ex;
 class nfs_node;
+namespace security {
+class KMSKeyProvider;
+} // namespace security
 
 namespace service {
 class copy_request;
@@ -459,6 +462,7 @@ class replica_stub : public serverlet<replica_stub>, public ref_counter
 
     std::unique_ptr<duplication_sync_timer> _duplication_sync_timer;
     std::unique_ptr<replica_backup_server> _backup_server;
+    std::unique_ptr<dsn::security::KMSKeyProvider> key_provider;
 
     // command_handlers
     std::vector<std::unique_ptr<command_deregister>> _cmds;

src/replica/replication_app_base.h

@@ -30,6 +30,7 @@
 #include <cstdint>
 #include <map>
 #include <string>
+#include <utility>
 
 #include "bulk_load_types.h"
 #include "common/json_helper.h"
@@ -44,14 +45,55 @@
 
 namespace dsn {
 class app_info;
-class blob;
 class message_ex;
 
 namespace replication {
 class learn_state;
 class mutation;
 class replica;
 
+<<<<<<< HEAD
+=======
+namespace {
+template <class T>
+error_code write_blob_to_file(const std::string &fname,
+                              const T &data,
+                              const dsn::utils::FileDataType &fileDataType)
+{
+    std::string tmp_fname = fname + ".tmp";
+    auto cleanup = defer([tmp_fname]() { utils::filesystem::remove_path(tmp_fname); });
+    auto s = rocksdb::WriteStringToFile(dsn::utils::PegasusEnv(fileDataType),
+                                        rocksdb::Slice(data.data(), data.length()),
+                                        tmp_fname,
+                                        /* should_sync */ true);
+    LOG_AND_RETURN_NOT_TRUE(
+        ERROR, s.ok(), ERR_FILE_OPERATION_FAILED, "write file {} failed", tmp_fname);
+    LOG_AND_RETURN_NOT_TRUE(ERROR,
+                            utils::filesystem::rename_path(tmp_fname, fname),
+                            ERR_FILE_OPERATION_FAILED,
+                            "move file from {} to {} failed",
+                            tmp_fname,
+                            fname);
+    return ERR_OK;
+}
+
+template <typename T, dsn::utils::FileDataType FileType>
+error_code load_json_generic(const std::string &fname, T &object)
+{
+    std::string data;
+    auto s = rocksdb::ReadFileToString(dsn::utils::PegasusEnv(FileType), fname, &data);
+    LOG_AND_RETURN_NOT_TRUE(ERROR, s.ok(), ERR_FILE_OPERATION_FAILED, "read file {} failed", fname);
+    LOG_AND_RETURN_NOT_TRUE(
+        ERROR,
+        json::json_forwarder<T>::decode(blob::create_from_bytes(std::move(data)), object),
+        ERR_FILE_OPERATION_FAILED,
+        "decode json from file {} failed",
+        fname);
+    return ERR_OK;
+}
+} // namespace
+
+>>>>>>> 4bd797d87 (feat(encryption): add kms key management)
 class replica_init_info
 {
 public:

src/replica/storage/simple_kv/test/case-000.ini

@@ -151,6 +151,11 @@ server_list = localhost:34601
 [pegasus.server]
 encrypt_data_at_rest = false
 
+[security]
+enable_acl =
+super_users =
+meta_acl_rpc_allow_list =
+
 [replication.app]
 app_name = simple_kv.instance0
 app_type = simple_kv

src/replica/storage/simple_kv/test/case-001.ini

@@ -151,6 +151,11 @@ server_list = localhost:34601
 [pegasus.server]
 encrypt_data_at_rest = false
 
+[security]
+enable_acl =
+super_users =
+meta_acl_rpc_allow_list =
+
 [replication.app]
 app_name = simple_kv.instance0
 app_type = simple_kv

src/replica/storage/simple_kv/test/case-002.ini

@@ -151,6 +151,11 @@ server_list = localhost:34601
 [pegasus.server]
 encrypt_data_at_rest = false
 
+[security]
+enable_acl =
+super_users =
+meta_acl_rpc_allow_list =
+
 [replication.app]
 app_name = simple_kv.instance0
 app_type = simple_kv

src/replica/storage/simple_kv/test/case-003.ini

@@ -151,6 +151,11 @@ server_list = localhost:34601
 [pegasus.server]
 encrypt_data_at_rest = false
 
+[security]
+enable_acl =
+super_users =
+meta_acl_rpc_allow_list =
+
 [replication.app]
 app_name = simple_kv.instance0
 app_type = simple_kv

src/replica/storage/simple_kv/test/case-004.ini

@@ -151,6 +151,11 @@ server_list = localhost:34601
 [pegasus.server]
 encrypt_data_at_rest = false
 
+[security]
+enable_acl =
+super_users =
+meta_acl_rpc_allow_list =
+
 [replication.app]
 app_name = simple_kv.instance0
 app_type = simple_kv