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)
 
 set(MY_BOOST_LIBS Boost::filesystem)

src/replica/key_provider.h

@@ -0,0 +1,45 @@
+// 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 "utils/errors.h"
+
+namespace dsn {
+namespace security {
+
+// An interface for encrypting and decrypting Pegasus's encryption keys.
+class KeyProvider
+{
+public:
+    virtual ~KeyProvider() = default;
+
+    // Decrypts the encryption key.
+    virtual dsn::error_s DecryptEncryptionKey(const std::string &encryption_key,
+                                              const std::string &iv,
+                                              const std::string &key_version,
+                                              std::string *decrypted_key) = 0;
+
+    // Generates an encryption key (the generated key is encrypted).
+    virtual dsn::error_s GenerateEncryptionKey(std::string *encryption_key,
+                                               std::string *iv,
+                                               std::string *key_version) = 0;
+};
+} // namespace security
+} // namespace dsn

src/replica/pegasus_kms_key_provider.cpp

@@ -0,0 +1,42 @@
+// 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/pegasus_kms_key_provider.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace security {
+
+dsn::error_s PegasusKMSKeyProvider::DecryptEncryptionKey(const std::string &encryption_key,
+                                                         const std::string &iv,
+                                                         const std::string &key_version,
+                                                         std::string *decrypted_key)
+{
+    return client_.DecryptEncryptionKey(encryption_key, iv, key_version, decrypted_key);
+}
+
+dsn::error_s PegasusKMSKeyProvider::GenerateEncryptionKey(std::string *encryption_key,
+                                                          std::string *iv,
+                                                          std::string *key_version)
+{
+    return client_.GenerateEncryptionKey(encryption_key, iv, key_version);
+}
+
+} // namespace security
+} // namespace dsn

src/replica/pegasus_kms_key_provider.h

@@ -0,0 +1,54 @@
+// 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 "runtime/security/kms_client.h"
+#include "replica/key_provider.h"
+#include "utils/errors.h"
+
+namespace dsn {
+namespace security {
+class PegasusKMSKeyProvider : public KeyProvider
+{
+public:
+    ~PegasusKMSKeyProvider() override {}
+
+    PegasusKMSKeyProvider(const std::string &kms_url, std::string cluster_key_name)
+        : client_(kms_url, std::move(cluster_key_name))
+    {
+    }
+
+    // Decrypts the encryption key.
+    dsn::error_s DecryptEncryptionKey(const std::string &encryption_key,
+                                      const std::string &iv,
+                                      const std::string &key_version,
+                                      std::string *decrypted_key) override;
+
+    // Generates an encryption key (the generated key is encrypted).
+    dsn::error_s GenerateEncryptionKey(std::string *encryption_key,
+                                       std::string *iv,
+                                       std::string *key_version) override;
+
+private:
+    PegasusKMSClient client_;
+};
+} // namespace security
+} // namespace dsn

src/replica/replica_stub.cpp

@@ -41,6 +41,7 @@
 #include <set>
 #include <vector>
 
+#include "absl/strings/str_cat.h"
 #include "backup/replica_backup_server.h"
 #include "bulk_load/replica_bulk_loader.h"
 #include "common/backup_common.h"
@@ -62,15 +63,16 @@
 #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 "runtime/ranger/access_type.h"
 #include "runtime/rpc/rpc_message.h"
 #include "runtime/rpc/serialization.h"
 #include "runtime/security/access_controller.h"
 #include "runtime/task/async_calls.h"
+#include "replica/pegasus_kms_key_provider.h"
 #include "split/replica_split_manager.h"
 #include "utils/command_manager.h"
+#include "utils/errors.h"
 #include "utils/filesystem.h"
 #include "utils/fmt_logging.h"
 #include "utils/ports.h"
@@ -89,6 +91,8 @@
 #include "remote_cmd/remote_command.h"
 #include "utils/fail_point.h"
 
+DSN_DECLARE_bool(encrypt_data_at_rest);
+DSN_DECLARE_string(server_key);
 namespace dsn {
 namespace replication {
 DSN_DEFINE_bool(replication,
@@ -178,6 +182,14 @@ DSN_DEFINE_int32(
     10,
     "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,
+                  encryption_cluster_key_name,
+                  "pegasus",
+                  "The cluster name of encrypted server which use to get server key from kms.");
+DSN_DEFINE_string(pegasus.server,
+                  hadoop_kms_url,
+                  "",
+                  "Where the server encrypted key of file system can get from.");
 
 DSN_DECLARE_bool(duplication_enabled);
 DSN_DECLARE_int32(fd_beacon_interval_seconds);
@@ -215,6 +227,11 @@ replica_stub::replica_stub(replica_state_subscriber subscriber /*= nullptr*/,
     _log = nullptr;
     _primary_address_str[0] = '\0';
     install_perf_counters();
+    if (FLAGS_encrypt_data_at_rest) {
+        // TODO: check enable_acl whether be true
+        key_provider.reset(new dsn::security::PegasusKMSKeyProvider(
+            FLAGS_hadoop_kms_url, FLAGS_encryption_cluster_key_name));
+    }
 }
 
 replica_stub::~replica_stub(void) { close(); }
@@ -573,6 +590,35 @@ void replica_stub::initialize(bool clear /* = false*/)
     _access_controller = std::make_unique<dsn::security::access_controller>();
 }
 
+dsn::error_s store_kms_key(std::string data_dir,
+                           std::string encryption_key,
+                           std::string iv,
+                           std::string key_version)
+{
+    replica_kms_info kms_info(encryption_key, iv, key_version);
+    auto err = kms_info.store(data_dir);
+    if (dsn::ERR_OK == err) {
+        return dsn::error_s::ok();
+    } else {
+        return dsn::error_s::make(err, "Can't open replica_encrypted_key file to write");
+    }
+}
+
+void get_kms_key(std::string data_dir,
+                 std::string *encryption_key,
+                 std::string *iv,
+                 std::string *key_version)
+{
+    replica_kms_info kms_info;
+    auto err = kms_info.load(data_dir);
+    *encryption_key = kms_info.encryption_key;
+    *iv = kms_info.iv;
+    *key_version = kms_info.key_version;
+    if (dsn::ERR_OK != err) {
+        CHECK(err, "Can't open replica_encrypted_key file to read");
+    }
+}
+
 void replica_stub::initialize(const replication_options &opts, bool clear /* = false*/)
 {
     _primary_address = dsn_primary_address();
@@ -600,9 +646,31 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
         }
     }
 
+    std::string encryption_key;
+    std::string iv;
+    std::string key_version;
+    std::string server_key;
+    // get and store eek from kms
+    if (key_provider) {
+        get_kms_key(_options.data_dirs[0], &encryption_key, &iv, &key_version);
+        if (encryption_key.empty()) {
+            CHECK(key_provider, "invalid kms url ({})", FLAGS_hadoop_kms_url);
+            CHECK(key_provider->GenerateEncryptionKey(&encryption_key, &iv, &key_version),
+                  "get encryption key failed");
+        }
+        CHECK(key_provider->DecryptEncryptionKey(encryption_key, iv, key_version, &server_key),
+              "get decryption key failed");
+        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) {
+        CHECK(store_kms_key(_options.data_dirs[0], encryption_key, iv, key_version),
+              "Cant store kms key");
+    }
+
     // TODO(yingchun): remove the slog related code.
     // Create slog directory if it does not exist.
     std::string cdir;

src/replica/replica_stub.h

@@ -64,6 +64,7 @@
 #include "runtime/task/task.h"
 #include "runtime/task/task_code.h"
 #include "runtime/task/task_tracker.h"
+#include "replica/pegasus_kms_key_provider.h"
 #include "utils/autoref_ptr.h"
 #include "utils/error_code.h"
 #include "utils/flags.h"
@@ -489,6 +490,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::KeyProvider> key_provider;
 
     // command_handlers
     std::vector<std::unique_ptr<command_deregister>> _cmds;

src/replica/replication_app_base.cpp

@@ -65,6 +65,7 @@ namespace dsn {
 namespace replication {
 
 const std::string replica_init_info::kInitInfo = ".init-info";
+const std::string replica_kms_info::kFileName = "/replica_encrypted_key";
 
 namespace {
 error_code write_blob_to_file(const std::string &fname, const blob &data)
@@ -149,6 +150,70 @@ std::string replica_init_info::to_string()
     return oss.str();
 }
 
+error_code replica_kms_info::load(const std::string &dir)
+{
+    std::string info_path = utils::filesystem::path_combine(dir, kFileName);
+    LOG_AND_RETURN_NOT_TRUE(ERROR,
+                            utils::filesystem::path_exists(info_path),
+                            ERR_PATH_NOT_FOUND,
+                            "file({}) not exist",
+                            info_path);
+    LOG_AND_RETURN_NOT_OK(
+        ERROR, load_json(info_path), "load replica_kms_info from {} failed", info_path);
+    LOG_INFO("load replica_kms_info from {} ", info_path);
+    return ERR_OK;
+}
+
+error_code replica_kms_info::store(const std::string &dir)
+{
+    uint64_t start = dsn_now_ns();
+    std::string info_path = utils::filesystem::path_combine(dir, kFileName);
+    LOG_AND_RETURN_NOT_OK(ERROR,
+                          store_json(info_path),
+                          "store replica_kms_info to {} failed, time_used_ns = {}",
+                          info_path,
+                          dsn_now_ns() - start);
+    LOG_INFO(
+        "store replica_kms_info to {} succeed, time_used_ns = {}", info_path, dsn_now_ns() - start);
+    return ERR_OK;
+}
+
+error_code replica_kms_info::load_json(const std::string &fname)
+{
+    std::string data;
+    auto s = rocksdb::ReadFileToString(
+        dsn::utils::PegasusEnv(dsn::utils::FileDataType::kNonSensitive), 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<replica_kms_info>::decode(
+                                blob::create_from_bytes(std::move(data)), *this),
+                            ERR_FILE_OPERATION_FAILED,
+                            "decode json from file {} failed",
+                            fname);
+    return ERR_OK;
+}
+
+error_code replica_kms_info::store_json(const std::string &fname)
+{
+    const blob &data = json::json_forwarder<replica_kms_info>::encode(*this);
+    std::string tmp_fname = fname + ".tmp";
+    auto cleanup = defer([tmp_fname]() { utils::filesystem::remove_path(tmp_fname); });
+    auto s =
+        rocksdb::WriteStringToFile(dsn::utils::PegasusEnv(dsn::utils::FileDataType::kNonSensitive),
+                                   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;
+}
+
 error_code replica_app_info::load(const std::string &fname)
 {
     std::string data;