feat(encryption): add key manager

.github/workflows/lint_and_test_cpp.yaml

@@ -154,6 +154,7 @@ jobs:
           - pegasus_unit_test
           - recovery_test
           - restore_test
+          - security_test
           - throttle_test
     needs: build_Release
     runs-on: ubuntu-latest
@@ -235,6 +236,7 @@ jobs:
           - pegasus_unit_test
           - recovery_test
           - restore_test
+          - security_test
           # TODO(yingchun): Disable it because we find it's too flaky, we will re-enable it after
           # it has been optimized.
 #          - throttle_test
@@ -319,6 +321,7 @@ jobs:
 #          - pegasus_unit_test
 #          - recovery_test
 #          - restore_test
+#          - security_test
 #          - throttle_test
 #    needs: build_UBSAN
 #    runs-on: ubuntu-latest

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,38 @@
+// 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 kms_key_provider::DecryptEncryptionKey(const dsn::replication::kms_info &info,
+                                                    std::string *decrypted_key)
+{
+    return _client.DecryptEncryptionKey(info, decrypted_key);
+}
+
+dsn::error_s kms_key_provider::GenerateEncryptionKey(dsn::replication::kms_info *info)
+{
+    return _client.GenerateEncryptionKey(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 "security/kms_client.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace replication {
+struct 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 kms_key_provider
+{
+public:
+    ~kms_key_provider() {}
+
+    kms_key_provider(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::kms_info &info,
+                                      std::string *decrypted_key);
+
+    // Generate an encryption key from KMS.
+    dsn::error_s GenerateEncryptionKey(dsn::replication::kms_info *info);
+
+private:
+    kms_client _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>
@@ -54,23 +55,26 @@
 #include "mutation_log.h"
 #include "nfs/nfs_node.h"
 #include "nfs_types.h"
+#include "ranger/access_type.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"
 #include "runtime/rpc/serialization.h"
-#include "security/access_controller.h"
 #include "runtime/task/async_calls.h"
+#include "security/access_controller.h"
 #include "split/replica_split_manager.h"
 #include "utils/command_manager.h"
+#include "utils/env.h"
+#include "utils/errors.h"
 #include "utils/filesystem.h"
 #include "utils/fmt_logging.h"
+#include "utils/load_dump_object.h"
 #include "utils/ports.h"
 #include "utils/process_utils.h"
 #include "utils/rand.h"
@@ -214,7 +218,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,
@@ -283,12 +295,39 @@ 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_at_rest_with_kms_url, [](std::string &message) -> bool {
+#ifndef MOCK_TEST
+    if (FLAGS_encrypt_data_at_rest && utils::is_empty(FLAGS_hadoop_kms_url)) {
+        message = fmt::format("[security] hadoop_kms_url should not be empty when [pegasus.server] "
+                              "encrypt_data_at_rest is enabled.");
+        return false;
+    }
+#endif
+    return true;
+});
 
 bool replica_stub::s_not_exit_on_log_failure = false;
 
@@ -380,9 +419,51 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
         }
     }
 
+    const auto &kms_path =
+        utils::filesystem::path_combine(_options.data_dirs[0], kms_info::kKmsInfo);
+    // FLAGS_data_dirs may be empty when load configuration, use LOG_FATAL instead of group
+    // validator.
+    if (!FLAGS_encrypt_data_at_rest && utils::filesystem::path_exists(kms_path)) {
+        LOG_FATAL("The kms_info file exists at ({}), but [pegasus.server] "
+                  "encrypt_data_at_rest is enbale."
+                  "Encryption in Pegasus is irreversible after its initial activation.",
+                  kms_path);
+    }
+
+    dsn::replication::kms_info kms_info;
+    if (FLAGS_encrypt_data_at_rest && !utils::is_empty(FLAGS_hadoop_kms_url)) {
+        _key_provider.reset(new dsn::security::kms_key_provider(
+            ::absl::StrSplit(FLAGS_hadoop_kms_url, ",", ::absl::SkipEmpty()), FLAGS_cluster_name));
+        const auto &ec = dsn::utils::load_rjobj_from_file(
+            kms_path, dsn::utils::FileDataType::kNonSensitive, &kms_info);
+        if (ec != dsn::ERR_PATH_NOT_FOUND && ec != dsn::ERR_OK) {
+            CHECK_EQ_MSG(dsn::ERR_OK, ec, "Can't load kms key from kms-info file");
+        }
+        // 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.
+        if (ec == dsn::ERR_PATH_NOT_FOUND) {
+            LOG_WARNING("It's normal to encounter a temporary inability to open the kms-info file "
+                        "during the first process launch.");
+            CHECK_OK(_key_provider->GenerateEncryptionKey(&kms_info),
+                     "Generate encryption key from kms failed");
+        }
+        CHECK_OK(_key_provider->DecryptEncryptionKey(kms_info, &server_key),
+                 "Get decryption key failed from {}",
+                 kms_path);
+        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 && !utils::filesystem::path_exists(kms_path)) {
+        const auto &err = dsn::utils::dump_rjobj_to_file(
+            kms_info, dsn::utils::FileDataType::kNonSensitive, kms_path);
+        CHECK_EQ_MSG(dsn::ERR_OK, err, "Can't store kms key to kms-info file");
+    }
+
     // 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 kms_key_provider;
+} // 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::kms_key_provider> _key_provider;
 
     // command_handlers
     std::vector<std::unique_ptr<command_deregister>> _cmds;
@@ -490,6 +494,9 @@ class replica_stub : public serverlet<replica_stub>, public ref_counter
     // replica count executing emergency checkpoint concurrently
     std::atomic_int _manual_emergency_checkpointing_count;
 
+    // replica decrypted key for rocksdb
+    std::string server_key;
+
     bool _is_running;
 
     std::unique_ptr<dsn::security::access_controller> _access_controller;

src/replica/replication_app_base.cpp

@@ -69,6 +69,7 @@ namespace dsn {
 namespace replication {
 
 const std::string replica_init_info::kInitInfo = ".init-info";
+const std::string kms_info::kKmsInfo = ".kms-info";
 
 std::string replica_init_info::to_string()
 {

src/replica/replication_app_base.h

@@ -85,6 +85,24 @@ class replica_app_info
     error_code store(const std::string &fname);
 };
 
+// This class stores and loads EEK, IV, and KV from KMS as a JSON file.
+// To get the decrypted key, should POST EEK, IV, and KV to KMS.
+struct kms_info
+{
+    std::string encrypted_key;         // a.k.a encrypted encryption key
+    std::string initialization_vector; // a.k.a initialization vector
+    std::string key_version;           // a.k.a key version
+    DEFINE_JSON_SERIALIZATION(encrypted_key, initialization_vector, key_version)
+    static const std::string kKmsInfo; // json file name
+
+    kms_info(const std::string &e_key = "",
+             const std::string &i = "",
+             const std::string &k_version = "")
+        : encrypted_key(e_key), initialization_vector(i), key_version(k_version)
+    {
+    }
+};
+
 /// The store engine interface of Pegasus.
 /// Inherited by pegasus::pegasus_server_impl
 /// Inherited by apps::rrdb_service

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