feat(encryption): add kms key management

src/replica/CMakeLists.txt

@@ -58,6 +58,8 @@ set(MY_PROJ_SRC
 set(MY_SRC_SEARCH_MODE "GLOB")
 
 set(MY_PROJ_LIBS 
+    absl::strings
+    curl
     dsn_replication_common
     dsn.failure_detector
     dsn.block_service
@@ -71,6 +73,7 @@ set(MY_PROJ_LIBS
     dsn_aio
     dsn_meta_server
     galaxy-fds-sdk-cpp
+    gssapi_krb5
     PocoNet
     PocoFoundation
     PocoNetSSL

src/utils/kms_client.cpp → src/replica/kms_client.cpp

@@ -23,7 +23,7 @@
 #include "http/http_client.h"
 #include "http/http_method.h"
 #include "nlohmann/json.hpp"
-#include "utils/kms_client.h"
+#include "replica/kms_client.h"
 
 namespace dsn {
 namespace security {
@@ -46,15 +46,9 @@ dsn::error_s PegasusKMSClient::DecryptEncryptionKey(const std::string &encryptio
 
     http_client client;
     auto err = client.init();
-    if(!err.is_ok()){
-        return dsn::error_s::make(ERR_CURL_FAILED, "Start http client failed");
-    }
-
+    CHECK_EQ_MSG(err, ERR_OK, "Start http client failed");
     err = client.set_auth(http_auth_type::SPNEGO);
-    if(!err.is_ok()){
-        return dsn::error_s::make(ERR_CURL_FAILED, "http client set auth type failed");
-    }
-
+    CHECK_EQ_MSG(err, ERR_OK, "http client set auth type failed");
     std::vector<std::string> urls;
     urls.reserve(kms_urls_.size());
     for (const auto &url : kms_urls_) {
@@ -65,23 +59,16 @@ dsn::error_s PegasusKMSClient::DecryptEncryptionKey(const std::string &encryptio
     client.clear_header_fields();
     client.set_content_type("application/json");
     client.set_accept("*/*");
-
     err = client.with_post_method(post.dump());
-    if(!err.is_ok()){
-        return dsn::error_s::make(ERR_CURL_FAILED, "http client set method failed");
-    }
+    CHECK_EQ_MSG(err, ERR_OK, "http client set method failed");
 
     nlohmann::json j;
     for (const auto &url : urls) {
         err = client.set_url(url);
-        if(!err.is_ok()){
-            return dsn::error_s::make(ERR_CURL_FAILED, "http clientt set url failed");
-        }
+        CHECK_EQ_MSG(err, ERR_OK, "http clientt set url failed");
         std::string resp;
         err = client.exec_method(&resp);
-        if(!err.is_ok()){
-            return dsn::error_s::make(ERR_CURL_FAILED, "http client exec post method failed");
-        }
+        CHECK_EQ_MSG(err, ERR_OK, "http client exec post method failed");
         long http_status;
         client.get_http_status(http_status);
         LOG_INFO("http status = ({})", http_status);
@@ -112,13 +99,10 @@ dsn::error_s PegasusKMSClient::GenerateEncryptionKeyFromKMS(const std::string &k
 {
     http_client client;
     auto err = client.init();
-    if(!err.is_ok()){
-        return dsn::error_s::make(ERR_CURL_FAILED, "Start http client failed");
-    }
+    CHECK_EQ_MSG(err, ERR_OK, "Start http client failed");
     err = client.set_auth(http_auth_type::SPNEGO);
-    if(!err.is_ok()){
-        return dsn::error_s::make(ERR_CURL_FAILED, "http client set auth type failed");
-    }
+    CHECK_EQ_MSG(err, ERR_OK, "http client set auth type failed");
+
     std::vector<std::string> urls;
     urls.reserve(kms_urls_.size());
     for (const auto &url : kms_urls_) {
@@ -129,21 +113,12 @@ dsn::error_s PegasusKMSClient::GenerateEncryptionKeyFromKMS(const std::string &k
     nlohmann::json j = nlohmann::json::object();
     for (const auto &url : urls) {
         err = client.set_url(url);
-        if(!err.is_ok()){
-            return dsn::error_s::make(ERR_CURL_FAILED, "http client set url failed");
-        }
-
+        CHECK_EQ_MSG(err, ERR_OK, "http client set url failed");
         err = client.with_get_method();
-        if(!err.is_ok()){
-            return dsn::error_s::make(ERR_CURL_FAILED, "http client set get method failed");
-        }
-
+        CHECK_EQ_MSG(err, ERR_OK, "http client set get method failed");
         std::string resp;
         err = client.exec_method(&resp);
-        if(!err.is_ok()){
-            return dsn::error_s::make(ERR_CURL_FAILED, "http client exec get method failed");
-        }
-
+        CHECK_EQ_MSG(err, ERR_OK, "http client exec get method failed");
         long http_status;
         client.get_http_status(http_status);
         LOG_INFO("http status = ({})", http_status);

src/server/pegasus_kms_key_provider.cpp → src/replica/pegasus_kms_key_provider.cpp

@@ -17,7 +17,7 @@
 
 #include <string>
 
-#include "server/pegasus_kms_key_provider.h"
+#include "replica/pegasus_kms_key_provider.h"
 #include "utils/errors.h"
 
 namespace dsn {

src/utils/pegasus_kms_key_provider.h → src/replica/pegasus_kms_key_provider.h

@@ -20,8 +20,8 @@
 #include <string>
 #include <utility>
 
-#include "utils/kms_client.h"
-#include "utils/key_provider.h"
+#include "replica/kms_client.h"
+#include "replica/key_provider.h"
 #include "utils/errors.h"
 
 namespace dsn {

src/replica/replica_stub.cpp

@@ -37,6 +37,7 @@
 // IWYU pragma: no_include <ext/alloc_traits.h>
 #include <fmt/core.h>
 #include <fmt/format.h>
+#include <fstream>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
@@ -50,6 +51,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"
@@ -71,15 +73,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"
@@ -187,7 +190,21 @@ 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_DEFINE_string(pegasus.server,
+                  server_key,
+                  "server_key",
+                  "The encrypted server key to use in the filesystem.");
+
+// DSN_DECLARE_string(data_dirs);
+DSN_DECLARE_bool(encrypt_data_at_rest);
 DSN_DECLARE_bool(duplication_enabled);
 DSN_DECLARE_int32(fd_beacon_interval_seconds);
 DSN_DECLARE_int32(fd_check_interval_seconds);
@@ -224,6 +241,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(); }
@@ -582,6 +604,21 @@ void replica_stub::initialize(bool clear /* = false*/)
     _access_controller = std::make_unique<dsn::security::access_controller>();
 }
 
+dsn::error_s store_kms_key(std::string server_key){
+    // std::string file_name = FLAGS_data_dirs;
+    std::string file_name = "/home/yujingwei";
+    std::string path = "/replica_encrypted_key";
+    file_name = absl::StrCat(file_name, path);
+    std::ofstream in(file_name, std::ios::app);
+    if(in.is_open()){
+        in << server_key << std::endl;
+    } else{
+        return dsn::error_s::make(ERR_FILE_OPERATION_FAILED, "Can't open replica_encrypted_key file to write");
+    }
+
+    return dsn::error_s::ok();
+}
+
 void replica_stub::initialize(const replication_options &opts, bool clear /* = false*/)
 {
     _primary_address = dsn_primary_address();
@@ -609,6 +646,34 @@ void replica_stub::initialize(const replication_options &opts, bool clear /* = f
         }
     }
 
+    //get and store eek from kms
+    if (key_provider) {
+        CHECK(key_provider, "invalid kms url ({})", FLAGS_hadoop_kms_url);
+        std::string encryption_key;
+        std::string iv;
+        std::string key_version;
+        std::string server_key;
+        auto err = key_provider->GenerateEncryptionKey(&encryption_key, &iv, &key_version);
+        if(!err.is_ok()){
+            fmt::print(stderr, "get encryption key failed, error={} ", err.description());
+        }
+        // LOG_INFO("data dir is = ({})",FLAGS_data_dirs);
+        err = store_kms_key(encryption_key);
+        if(!err.is_ok()){
+            fmt::print(stderr, "strore decryption key failed, error={} ", err.description());
+        }
+        LOG_INFO("encryption key is ({})", encryption_key);
+        LOG_INFO("iv is ({})", iv);
+        LOG_INFO("key version is ({})", key_version);
+        // 用 flag 传
+        // 暂时看下能不能用
+        err = key_provider->DecryptEncryptionKey(encryption_key, iv, key_version, &server_key);
+        if(!err.is_ok()){
+            fmt::print(stderr, "get decryption key failed, error={} ", err.description());
+        }
+        LOG_INFO("server key is {}", server_key);
+    }
+
     // Initialize the file system manager.
     _fs_manager.initialize(_options.data_dirs, _options.data_dir_tags);
 

src/replica/replica_stub.h

@@ -70,6 +70,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"
@@ -493,6 +494,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/runtime/CMakeLists.txt

@@ -53,6 +53,6 @@ add_library(dsn_runtime STATIC
         tracer.cpp
         zlocks.cpp
         )
-target_link_libraries(dsn_runtime PRIVATE absl_strings absl_raw_logging_internal dsn_utils sasl2 gssapi_krb5 krb5)
+target_link_libraries(dsn_runtime PRIVATE dsn_utils sasl2 gssapi_krb5 krb5)
 define_file_basename_for_sources(dsn_runtime)
 install(TARGETS dsn_runtime DESTINATION "lib")

src/utils/CMakeLists.txt

@@ -31,7 +31,8 @@ set(MY_SRC_SEARCH_MODE "GLOB")
 
 set(MY_BOOST_LIBS Boost::system Boost::filesystem Boost::regex)
 
-set(MY_PROJ_LIBS dsn_replication_common absl::strings dsn_http curl gssapi_krb5 rocksdb)
+# set(MY_PROJ_LIBS absl::strings dsn_http curl gssapi_krb5 rocksdb)
+set(MY_PROJ_LIBS dsn_http rocksdb)
 
 # Extra files that will be installed
 set(MY_BINPLACES "")
@@ -42,6 +43,5 @@ else()
     dsn_add_shared_library()
 endif()
 
-target_link_libraries(dsn_replication_common)
 add_subdirectory(long_adder_bench)
 add_subdirectory(test)