feat(encryption): add key manager

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 "replica/key_provider.h"
+#include "runtime/security/kms_client.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/string_view.h"
 #include "backup/replica_backup_server.h"
 #include "bulk_load/replica_bulk_loader.h"
 #include "common/backup_common.h"
@@ -57,12 +58,13 @@
 #include "perf_counter/perf_counter.h"
 #include "replica.h"
 #include "replica/duplication/replica_follower.h"
+#include "replica/key_provider.h"
 #include "replica/log_file.h"
+#include "replica/pegasus_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 "runtime/ranger/access_type.h"
 #include "runtime/rpc/rpc_message.h"
@@ -71,13 +73,13 @@
 #include "runtime/task/async_calls.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"
 #include "utils/process_utils.h"
 #include "utils/rand.h"
 #include "utils/string_conv.h"
-#include "absl/strings/string_view.h"
 #include "utils/strings.h"
 #include "utils/synchronize.h"
 #ifdef DSN_ENABLE_GPERF
@@ -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,30 @@ 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 && !utils::is_empty(FLAGS_hadoop_kms_url)) {
+        get_kms_key(_options.data_dirs[0], &encryption_key, &iv, &key_version);
+        if (encryption_key.empty()) {
+            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 && !utils::is_empty(FLAGS_hadoop_kms_url)) {
+        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

@@ -75,6 +75,10 @@ class command_deregister;
 class message_ex;
 class nfs_node;
 
+namespace security {
+class KeyProvider;
+}  // namespace security
+
 namespace service {
 class copy_request;
 class copy_response;
@@ -489,6 +493,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;