diff --git a/.github/workflows/lint_and_test_cpp.yaml b/.github/workflows/lint_and_test_cpp.yaml
index 1a9cf2ff27..40c6655ae9 100644
--- a/.github/workflows/lint_and_test_cpp.yaml
+++ b/.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
diff --git a/src/replica/CMakeLists.txt b/src/replica/CMakeLists.txt
index 5bcd3f57bc..3bb64dbaf2 100644
--- a/src/replica/CMakeLists.txt
+++ b/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
diff --git a/src/replica/kms_key_provider.cpp b/src/replica/kms_key_provider.cpp
new file mode 100644
index 0000000000..8f8cd77c37
--- /dev/null
+++ b/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
diff --git a/src/replica/kms_key_provider.h b/src/replica/kms_key_provider.h
new file mode 100644
index 0000000000..280a85c94a
--- /dev/null
+++ b/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
diff --git a/src/replica/replica_stub.cpp b/src/replica/replica_stub.cpp
index 1a54b5281b..2f5a8e6f10 100644
--- a/src/replica/replica_stub.cpp
+++ b/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)) {
diff --git a/src/replica/replica_stub.h b/src/replica/replica_stub.h
index 08ae770a5b..a02921b596 100644
--- a/src/replica/replica_stub.h
+++ b/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;
diff --git a/src/replica/replication_app_base.cpp b/src/replica/replication_app_base.cpp
index bc25925d1b..c780fd2b9c 100644
--- a/src/replica/replication_app_base.cpp
+++ b/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()
 {
diff --git a/src/replica/replication_app_base.h b/src/replica/replication_app_base.h
index b8014003e0..8ccc17dc36 100644
--- a/src/replica/replication_app_base.h
+++ b/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
diff --git a/src/replica/storage/simple_kv/test/case-000.ini b/src/replica/storage/simple_kv/test/case-000.ini
index 8cf61e0679..34d18892fc 100644
--- a/src/replica/storage/simple_kv/test/case-000.ini
+++ b/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
diff --git a/src/replica/storage/simple_kv/test/case-001.ini b/src/replica/storage/simple_kv/test/case-001.ini
index ab71ae8472..edf1d69f46 100644
--- a/src/replica/storage/simple_kv/test/case-001.ini
+++ b/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
diff --git a/src/replica/storage/simple_kv/test/case-002.ini b/src/replica/storage/simple_kv/test/case-002.ini
index 91e4fa9641..d3012ab436 100644
--- a/src/replica/storage/simple_kv/test/case-002.ini
+++ b/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
diff --git a/src/replica/storage/simple_kv/test/case-003.ini b/src/replica/storage/simple_kv/test/case-003.ini
index fafc6c2a90..b0654769d8 100644
--- a/src/replica/storage/simple_kv/test/case-003.ini
+++ b/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
diff --git a/src/replica/storage/simple_kv/test/case-004.ini b/src/replica/storage/simple_kv/test/case-004.ini
index e25c587ee7..2a8a7f539f 100644
--- a/src/replica/storage/simple_kv/test/case-004.ini
+++ b/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
diff --git a/src/replica/storage/simple_kv/test/case-005.ini b/src/replica/storage/simple_kv/test/case-005.ini
index 72700862a5..8e28e7b894 100644
--- a/src/replica/storage/simple_kv/test/case-005.ini
+++ b/src/replica/storage/simple_kv/test/case-005.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
diff --git a/src/replica/storage/simple_kv/test/case-006.ini b/src/replica/storage/simple_kv/test/case-006.ini
index 9947d72d95..386433359b 100644
--- a/src/replica/storage/simple_kv/test/case-006.ini
+++ b/src/replica/storage/simple_kv/test/case-006.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-100.ini b/src/replica/storage/simple_kv/test/case-100.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-100.ini
+++ b/src/replica/storage/simple_kv/test/case-100.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-101.ini b/src/replica/storage/simple_kv/test/case-101.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-101.ini
+++ b/src/replica/storage/simple_kv/test/case-101.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-102.ini b/src/replica/storage/simple_kv/test/case-102.ini
index 2b02cfa4f3..b1a48eec26 100644
--- a/src/replica/storage/simple_kv/test/case-102.ini
+++ b/src/replica/storage/simple_kv/test/case-102.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-103.ini b/src/replica/storage/simple_kv/test/case-103.ini
index e269693b3e..10b312756d 100644
--- a/src/replica/storage/simple_kv/test/case-103.ini
+++ b/src/replica/storage/simple_kv/test/case-103.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-104.ini b/src/replica/storage/simple_kv/test/case-104.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-104.ini
+++ b/src/replica/storage/simple_kv/test/case-104.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-105.ini b/src/replica/storage/simple_kv/test/case-105.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-105.ini
+++ b/src/replica/storage/simple_kv/test/case-105.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-106.ini b/src/replica/storage/simple_kv/test/case-106.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-106.ini
+++ b/src/replica/storage/simple_kv/test/case-106.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-107.ini b/src/replica/storage/simple_kv/test/case-107.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-107.ini
+++ b/src/replica/storage/simple_kv/test/case-107.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-108.ini b/src/replica/storage/simple_kv/test/case-108.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-108.ini
+++ b/src/replica/storage/simple_kv/test/case-108.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-109.ini b/src/replica/storage/simple_kv/test/case-109.ini
index 92ac7d41ef..d489c02923 100644
--- a/src/replica/storage/simple_kv/test/case-109.ini
+++ b/src/replica/storage/simple_kv/test/case-109.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-200.ini b/src/replica/storage/simple_kv/test/case-200.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-200.ini
+++ b/src/replica/storage/simple_kv/test/case-200.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-201.ini b/src/replica/storage/simple_kv/test/case-201.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-201.ini
+++ b/src/replica/storage/simple_kv/test/case-201.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-202-0.ini b/src/replica/storage/simple_kv/test/case-202-0.ini
index 80ca078e8a..bcd050de80 100644
--- a/src/replica/storage/simple_kv/test/case-202-0.ini
+++ b/src/replica/storage/simple_kv/test/case-202-0.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-202-1.ini b/src/replica/storage/simple_kv/test/case-202-1.ini
index 80ca078e8a..bcd050de80 100644
--- a/src/replica/storage/simple_kv/test/case-202-1.ini
+++ b/src/replica/storage/simple_kv/test/case-202-1.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-203-0.ini b/src/replica/storage/simple_kv/test/case-203-0.ini
index fa5a344bc2..a35404aee8 100644
--- a/src/replica/storage/simple_kv/test/case-203-0.ini
+++ b/src/replica/storage/simple_kv/test/case-203-0.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-204.ini b/src/replica/storage/simple_kv/test/case-204.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-204.ini
+++ b/src/replica/storage/simple_kv/test/case-204.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-205.ini b/src/replica/storage/simple_kv/test/case-205.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-205.ini
+++ b/src/replica/storage/simple_kv/test/case-205.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-206.ini b/src/replica/storage/simple_kv/test/case-206.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-206.ini
+++ b/src/replica/storage/simple_kv/test/case-206.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-207.ini b/src/replica/storage/simple_kv/test/case-207.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-207.ini
+++ b/src/replica/storage/simple_kv/test/case-207.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-208.ini b/src/replica/storage/simple_kv/test/case-208.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-208.ini
+++ b/src/replica/storage/simple_kv/test/case-208.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-209.ini b/src/replica/storage/simple_kv/test/case-209.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-209.ini
+++ b/src/replica/storage/simple_kv/test/case-209.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-210.ini b/src/replica/storage/simple_kv/test/case-210.ini
index 7f7a444c2c..04ce80bab5 100644
--- a/src/replica/storage/simple_kv/test/case-210.ini
+++ b/src/replica/storage/simple_kv/test/case-210.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-211.ini b/src/replica/storage/simple_kv/test/case-211.ini
index 7f7a444c2c..04ce80bab5 100644
--- a/src/replica/storage/simple_kv/test/case-211.ini
+++ b/src/replica/storage/simple_kv/test/case-211.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-212.ini b/src/replica/storage/simple_kv/test/case-212.ini
index 5c4a32e2cb..0f79cb8af9 100644
--- a/src/replica/storage/simple_kv/test/case-212.ini
+++ b/src/replica/storage/simple_kv/test/case-212.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-213.ini b/src/replica/storage/simple_kv/test/case-213.ini
index bba13387ba..902312e9e5 100644
--- a/src/replica/storage/simple_kv/test/case-213.ini
+++ b/src/replica/storage/simple_kv/test/case-213.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-214.ini b/src/replica/storage/simple_kv/test/case-214.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-214.ini
+++ b/src/replica/storage/simple_kv/test/case-214.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-215.ini b/src/replica/storage/simple_kv/test/case-215.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-215.ini
+++ b/src/replica/storage/simple_kv/test/case-215.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-216.ini b/src/replica/storage/simple_kv/test/case-216.ini
index 2e7d5fb9b9..609a6e6af8 100644
--- a/src/replica/storage/simple_kv/test/case-216.ini
+++ b/src/replica/storage/simple_kv/test/case-216.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-300-0.ini b/src/replica/storage/simple_kv/test/case-300-0.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-300-0.ini
+++ b/src/replica/storage/simple_kv/test/case-300-0.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-300-1.ini b/src/replica/storage/simple_kv/test/case-300-1.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-300-1.ini
+++ b/src/replica/storage/simple_kv/test/case-300-1.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-300-2.ini b/src/replica/storage/simple_kv/test/case-300-2.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-300-2.ini
+++ b/src/replica/storage/simple_kv/test/case-300-2.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-301.ini b/src/replica/storage/simple_kv/test/case-301.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-301.ini
+++ b/src/replica/storage/simple_kv/test/case-301.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-302.ini b/src/replica/storage/simple_kv/test/case-302.ini
index b0d50662f6..fcab05e782 100644
--- a/src/replica/storage/simple_kv/test/case-302.ini
+++ b/src/replica/storage/simple_kv/test/case-302.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-303.ini b/src/replica/storage/simple_kv/test/case-303.ini
index b4fa46b9ad..e3fefc0fb3 100644
--- a/src/replica/storage/simple_kv/test/case-303.ini
+++ b/src/replica/storage/simple_kv/test/case-303.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-304.ini b/src/replica/storage/simple_kv/test/case-304.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-304.ini
+++ b/src/replica/storage/simple_kv/test/case-304.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-305.ini b/src/replica/storage/simple_kv/test/case-305.ini
index b4fa46b9ad..e3fefc0fb3 100644
--- a/src/replica/storage/simple_kv/test/case-305.ini
+++ b/src/replica/storage/simple_kv/test/case-305.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-306.ini b/src/replica/storage/simple_kv/test/case-306.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-306.ini
+++ b/src/replica/storage/simple_kv/test/case-306.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-307.ini b/src/replica/storage/simple_kv/test/case-307.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-307.ini
+++ b/src/replica/storage/simple_kv/test/case-307.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-400.ini b/src/replica/storage/simple_kv/test/case-400.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-400.ini
+++ b/src/replica/storage/simple_kv/test/case-400.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-401.ini b/src/replica/storage/simple_kv/test/case-401.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-401.ini
+++ b/src/replica/storage/simple_kv/test/case-401.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-402.ini b/src/replica/storage/simple_kv/test/case-402.ini
index 4258202b45..fb366b244e 100644
--- a/src/replica/storage/simple_kv/test/case-402.ini
+++ b/src/replica/storage/simple_kv/test/case-402.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-600.ini b/src/replica/storage/simple_kv/test/case-600.ini
index 9947d72d95..386433359b 100644
--- a/src/replica/storage/simple_kv/test/case-600.ini
+++ b/src/replica/storage/simple_kv/test/case-600.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-601.ini b/src/replica/storage/simple_kv/test/case-601.ini
index 37a1097690..f9e2f3b969 100644
--- a/src/replica/storage/simple_kv/test/case-601.ini
+++ b/src/replica/storage/simple_kv/test/case-601.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-602.ini b/src/replica/storage/simple_kv/test/case-602.ini
index 9947d72d95..386433359b 100644
--- a/src/replica/storage/simple_kv/test/case-602.ini
+++ b/src/replica/storage/simple_kv/test/case-602.ini
@@ -152,6 +152,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
diff --git a/src/replica/storage/simple_kv/test/case-603.ini b/src/replica/storage/simple_kv/test/case-603.ini
index 42aec9583b..dd29d532db 100644
--- a/src/replica/storage/simple_kv/test/case-603.ini
+++ b/src/replica/storage/simple_kv/test/case-603.ini
@@ -154,6 +154,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
diff --git a/src/replica/storage/simple_kv/test/run.sh b/src/replica/storage/simple_kv/test/run.sh
index f63fba48a8..9c8d07328b 100755
--- a/src/replica/storage/simple_kv/test/run.sh
+++ b/src/replica/storage/simple_kv/test/run.sh
@@ -118,7 +118,7 @@ if [ ! -z "${cases}" ]; then
         run_case ${id}
         echo
     done
-    TEST_OPTS="${OLD_TEST_OPTS};encrypt_data_at_rest=true"
+    TEST_OPTS="${OLD_TEST_OPTS};enable_acl=true;super_user=pegasus;encrypt_data_at_rest=true;meta_acl_rpc_allow_list=RPC_CM_CONFIG_SYNC,RPC_CM_DUPLICATION_SYNC,RPC_CM_PROPOSE_BALANCER,RPC_CM_QUERY_PARTITION_CONFIG_BY_INDEX,RPC_CM_UPDATE_PARTITION_CONFIGURATION,RPC_PREPARE,RPC_LEARN_ADD_LEARNER,RPC_LEARN_COMPLETION_NOTIFY,RPC_GROUP_CHECK,RPC_FD_FAILURE_DETECTOR_PING,RPC_CONFIG_PROPOSAL"
     for id in ${cases}; do
         run_case ${id}
         echo
diff --git a/src/replica/test/replica_http_service_test.cpp b/src/replica/test/replica_http_service_test.cpp
index e4567439f1..8de287862f 100644
--- a/src/replica/test/replica_http_service_test.cpp
+++ b/src/replica/test/replica_http_service_test.cpp
@@ -37,6 +37,9 @@ using std::map;
 using std::string;
 
 namespace dsn {
+namespace security {
+DSN_DECLARE_bool(enable_acl);
+} // namespace security
 namespace replication {
 DSN_DECLARE_bool(duplication_enabled);
 DSN_DECLARE_bool(fd_disabled);
@@ -50,6 +53,12 @@ class replica_http_service_test : public replica_test_base
         // Disable unnecessary works before starting stub.
         FLAGS_fd_disabled = true;
         FLAGS_duplication_enabled = false;
+        // Set FLAGS_enable_acl to true, ensuring the group validator's
+        // encrypt_data_at_rest_pre_check
+        // is successful when encrypt_data_at_rest is also true.
+        // TODO(jingwei): It's a trick for test, it should set together at class
+        // pegasus::encrypt_data_at_rest.
+        dsn::security::FLAGS_enable_acl = true;
         stub->initialize_start();
 
         http_call_registry::instance().clear_paths();
diff --git a/src/security/CMakeLists.txt b/src/security/CMakeLists.txt
index e996d2ebcb..2afa311a9b 100644
--- a/src/security/CMakeLists.txt
+++ b/src/security/CMakeLists.txt
@@ -23,9 +23,14 @@ thrift_generate_cpp(
 set(MY_PROJ_SRC ${SECURITY_THRIFT_SRCS})
 set(MY_SRC_SEARCH_MODE "GLOB")
 set(MY_PROJ_LIBS
+        absl::strings
+        curl
+        dsn_http
         dsn_meta_server
         dsn_replication_common
         dsn_runtime
-        dsn_utils)
+        dsn_utils
+        gssapi_krb5)
+
 dsn_add_object()
 add_subdirectory(test)
diff --git a/src/security/kms_client.cpp b/src/security/kms_client.cpp
new file mode 100644
index 0000000000..ffe154b5a5
--- /dev/null
+++ b/src/security/kms_client.cpp
@@ -0,0 +1,193 @@
+// 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 <initializer_list>
+#include <map>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+#include "absl/strings/escaping.h"
+#include "fmt/core.h"
+#include "http/http_client.h"
+#include "http/http_method.h"
+#include "http/http_status_code.h"
+#include "nlohmann/json.hpp"
+#include "nlohmann/json_fwd.hpp"
+#include "replica/replication_app_base.h"
+#include "security/kms_client.h"
+#include "utils/error_code.h"
+#include "utils/fmt_logging.h"
+
+namespace dsn {
+namespace security {
+
+dsn::error_s kms_client::DecryptEncryptionKey(const dsn::replication::kms_info &info,
+                                              std::string *decrypted_key)
+{
+    nlohmann::json payload;
+    payload["name"] = _cluster_key_name;
+    std::string iv_plain = ::absl::HexStringToBytes(info.initialization_vector);
+    std::string iv_b64;
+    ::absl::WebSafeBase64Escape(iv_plain, &iv_b64);
+    payload["iv"] = iv_b64;
+    std::string eek_plain = ::absl::HexStringToBytes(info.encrypted_key);
+    std::string eek_b64;
+    ::absl::WebSafeBase64Escape(eek_plain, &eek_b64);
+    payload["material"] = eek_b64;
+
+    http_client client;
+    RETURN_NOT_OK(client.init());
+    RETURN_NOT_OK(client.set_auth(http_auth_type::SPNEGO));
+
+    std::vector<std::string> urls;
+    urls.reserve(_kms_urls.size());
+    for (const auto &url : _kms_urls) {
+        urls.emplace_back(
+            fmt::format("{}/v1/keyversion/{}/_eek?eek_op=decrypt", url, info.key_version));
+    }
+    client.clear_header_fields();
+    client.set_content_type("application/json");
+    client.set_accept("*/*");
+
+    RETURN_NOT_OK(client.with_post_method(payload.dump()));
+
+    nlohmann::json j;
+    for (const auto &url : urls) {
+        RETURN_NOT_OK(client.set_url(url));
+        std::string resp;
+        auto err = client.exec_method(&resp);
+        if (err.code() == ERR_NETWORK_FAILURE || err.code() == ERR_TIMEOUT) {
+            continue;
+        }
+        RETURN_NOT_OK(err);
+        http_status_code http_status;
+        RETURN_NOT_OK(client.get_http_status(http_status));
+        if (http_status != http_status_code::kOk) {
+            LOG_WARNING("The http status is ({}), and url is ({})",
+                        get_http_status_message(http_status),
+                        url);
+            continue;
+        }
+        try {
+            j = nlohmann::json::parse(resp);
+            break;
+        } catch (nlohmann::json::exception &exp) {
+            LOG_ERROR("encode kms_info to json failed: {}, data = [{}]", exp.what(), resp);
+        }
+    }
+
+    std::string dek_b64;
+    RETURN_ERRS_NOT_TRUE(
+        j.contains("material"),
+        ERR_INVALID_DATA,
+        "Received null material in kms json data, network may have some problems.");
+    dek_b64 = j.at("material");
+
+    std::string dek_plain;
+    RETURN_ERRS_NOT_TRUE(::absl::WebSafeBase64Unescape(dek_b64, &dek_plain),
+                         ERR_INVALID_DATA,
+                         "Decryption key base64 decoding failed.");
+
+    *decrypted_key = ::absl::BytesToHexString(dek_plain);
+    return dsn::error_s::ok();
+}
+
+dsn::error_s kms_client::GenerateEncryptionKeyFromKMS(const std::string &key_name,
+                                                      dsn::replication::kms_info *info)
+{
+    http_client client;
+    RETURN_NOT_OK(client.init());
+    RETURN_NOT_OK(client.set_auth(http_auth_type::SPNEGO));
+
+    std::vector<std::string> urls;
+    urls.reserve(_kms_urls.size());
+    for (const auto &url : _kms_urls) {
+        urls.emplace_back(
+            fmt::format("{}/v1/key/{}/_eek?eek_op=generate&num_keys=1", url, key_name));
+    }
+
+    nlohmann::json j = nlohmann::json::object();
+    for (const auto &url : urls) {
+        RETURN_NOT_OK(client.set_url(url));
+        RETURN_NOT_OK(client.with_get_method());
+        std::string resp;
+        const auto &err = client.exec_method(&resp);
+        if (err.code() == ERR_NETWORK_FAILURE || err.code() == ERR_TIMEOUT) {
+            continue;
+        }
+        RETURN_NOT_OK(err);
+        http_status_code http_status;
+        RETURN_NOT_OK(client.get_http_status(http_status));
+        if (http_status != http_status_code::kOk) {
+            LOG_WARNING("The http status is ({}), and url is ({})",
+                        get_http_status_message(http_status),
+                        url);
+            continue;
+        }
+        try {
+            j = nlohmann::json::parse(resp).at(0);
+            break;
+        } catch (nlohmann::json::exception &exp) {
+            LOG_ERROR("encode kms_info to json failed: {}, data = [{}]", exp.what(), resp);
+        }
+    }
+
+    RETURN_ERRS_NOT_TRUE(
+        !j["versionName"].is_null(),
+        ERR_INVALID_DATA,
+        "Received null versionName in kms json data, network may have some problems.");
+    j["versionName"].get_to(info->key_version);
+
+    std::string iv_b64;
+    RETURN_ERRS_NOT_TRUE(!j["iv"].is_null(),
+                         ERR_INVALID_DATA,
+                         "Received null IV in kms json data, network may have some problems.");
+    j["iv"].get_to(iv_b64);
+
+    std::string iv_plain;
+    RETURN_ERRS_NOT_TRUE(::absl::WebSafeBase64Unescape(iv_b64, &iv_plain),
+                         ERR_INVALID_DATA,
+                         "IV base64 decoding failed.");
+    info->initialization_vector = ::absl::BytesToHexString(iv_plain);
+
+    std::string key_b64;
+    RETURN_ERRS_NOT_TRUE(
+        !j["encryptedKeyVersion"].is_null(),
+        ERR_INVALID_DATA,
+        "Received null encryptedKeyVersion in kms json data, network may have some problems.");
+    RETURN_ERRS_NOT_TRUE(!j["encryptedKeyVersion"]["material"].is_null(),
+                         ERR_INVALID_DATA,
+                         "Received null material of encryptedKeyVersion in kms json data, network "
+                         "may have some problems.");
+    j["encryptedKeyVersion"]["material"].get_to(key_b64);
+
+    std::string key_plain;
+    RETURN_ERRS_NOT_TRUE(::absl::WebSafeBase64Unescape(key_b64, &key_plain),
+                         ERR_INVALID_DATA,
+                         "Encryption key base64 decoding failed.");
+    info->encrypted_key = ::absl::BytesToHexString(key_plain);
+    return dsn::error_s::ok();
+}
+
+dsn::error_s kms_client::GenerateEncryptionKey(dsn::replication::kms_info *info)
+{
+    return GenerateEncryptionKeyFromKMS(_cluster_key_name, info);
+}
+
+} // namespace security
+} // namespace dsn
diff --git a/src/security/kms_client.h b/src/security/kms_client.h
new file mode 100644
index 0000000000..57d949ca82
--- /dev/null
+++ b/src/security/kms_client.h
@@ -0,0 +1,59 @@
+// 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 <algorithm>
+#include <string>
+#include <utility>
+#include <vector>
+
+#include "utils/errors.h"
+
+namespace dsn {
+namespace replication {
+struct kms_info;
+} // namespace replication
+
+namespace security {
+// A class designed to generate an encryption key from KMS for file writing,
+// implemented using an HTTP client.
+// This class is not thread-safe. Thus maintain one instance for each thread.
+class kms_client
+{
+public:
+    kms_client(const std::vector<std::string> &kms_url, std::string cluster_key_name)
+        : _kms_urls(kms_url), _cluster_key_name(std::move(cluster_key_name))
+    {
+    }
+
+    // Retrieve the Decrypted Encryption Key (DEK) from KMS after generating the EEK, IV, and KV.
+    dsn::error_s DecryptEncryptionKey(const dsn::replication::kms_info &info,
+                                      std::string *decrypted_key);
+
+    // Generated the EEK, IV, KV from KMS.
+    dsn::error_s GenerateEncryptionKey(dsn::replication::kms_info *info);
+
+private:
+    dsn::error_s GenerateEncryptionKeyFromKMS(const std::string &key_name,
+                                              dsn::replication::kms_info *info);
+
+    std::vector<std::string> _kms_urls;
+    std::string _cluster_key_name;
+};
+} // namespace security
+} // namespace dsn
diff --git a/src/security/test/CMakeLists.txt b/src/security/test/CMakeLists.txt
index b5725a5301..a7c4e6a5a9 100644
--- a/src/security/test/CMakeLists.txt
+++ b/src/security/test/CMakeLists.txt
@@ -19,12 +19,16 @@ set(MY_PROJ_NAME dsn_security_tests)
 set(MY_PROJ_SRC "")
 set(MY_SRC_SEARCH_MODE "GLOB")
 set(MY_PROJ_LIBS
+        absl::strings
+        dsn_http
+        curl
         dsn_security
         dsn_meta_server
         dsn_replication_common
         dsn_runtime
         dsn_utils
-        gtest)
+        gtest
+        rocksdb)
 set(MY_BINPLACES
         config.ini
         run.sh)
diff --git a/src/test/function_test/CMakeLists.txt b/src/test/function_test/CMakeLists.txt
index b3453846b3..053138e334 100644
--- a/src/test/function_test/CMakeLists.txt
+++ b/src/test/function_test/CMakeLists.txt
@@ -23,4 +23,5 @@ add_subdirectory(detect_hotspot)
 add_subdirectory(partition_split)
 add_subdirectory(recovery)
 add_subdirectory(restore)
+add_subdirectory(security)
 add_subdirectory(throttle)
diff --git a/src/test/function_test/security/CMakeLists.txt b/src/test/function_test/security/CMakeLists.txt
new file mode 100644
index 0000000000..dbe32de945
--- /dev/null
+++ b/src/test/function_test/security/CMakeLists.txt
@@ -0,0 +1,46 @@
+# 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.
+
+set(MY_PROJ_NAME "security_test")
+set(MY_PROJ_SRC "")
+
+set(MY_SRC_SEARCH_MODE "GLOB")
+
+set(MY_PROJ_LIBS
+        absl::strings
+        dsn_http
+        curl
+        dsn_meta_server
+        dsn_utils
+        dsn_security
+        dsn_ranger
+        dsn_replica_server
+        sasl2
+        gssapi_krb5
+        krb5
+        gtest
+        function_test_utils
+        rocksdb
+        test_utils)
+
+set(MY_BOOST_LIBS Boost::system Boost::filesystem)
+
+set(MY_BINPLACES
+        ./config.ini
+        ../run.sh)
+
+dsn_add_test()
diff --git a/src/test/function_test/security/config.ini b/src/test/function_test/security/config.ini
new file mode 100644
index 0000000000..de2b64edf5
--- /dev/null
+++ b/src/test/function_test/security/config.ini
@@ -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.
+
+[apps.mimic]
+name = mimic
+type = dsn.app.mimic
+pools = THREAD_POOL_DEFAULT,THREAD_POOL_META_SERVER
+
+[core]
+tool = nativerun
+logging_start_level = LOG_LEVEL_INFO
+
+[replication]
+cluster_name = pegasus_cluster_key
+
+[pegasus.clusters]
+onebox = 127.0.0.1:34601,127.0.0.1:34602,127.0.0.1:34603
+
+[pegasus.server]
+encrypt_data_at_rest = true
+hadoop_kms_url =
+
+[security]
+enable_acl = true
+super_users = pegasus
diff --git a/src/test/function_test/security/main.cpp b/src/test/function_test/security/main.cpp
new file mode 100644
index 0000000000..74e29e22a7
--- /dev/null
+++ b/src/test/function_test/security/main.cpp
@@ -0,0 +1,30 @@
+/*
+ * 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 <gtest/gtest.h>
+
+#include "runtime/app_model.h"
+
+GTEST_API_ int main(int argc, char **argv)
+{
+    testing::InitGoogleTest(&argc, argv);
+    dsn_run_config("config.ini", false);
+    int ret = RUN_ALL_TESTS();
+    dsn_exit(ret);
+}
diff --git a/src/test/function_test/security/test_kms_client.cpp b/src/test/function_test/security/test_kms_client.cpp
new file mode 100644
index 0000000000..33b2ac0ecc
--- /dev/null
+++ b/src/test/function_test/security/test_kms_client.cpp
@@ -0,0 +1,63 @@
+// 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 <absl/strings/str_split.h>
+#include <string.h>
+#include <memory>
+#include <string>
+
+#include "gtest/gtest.h"
+#include "replica/kms_key_provider.h"
+#include "replica/replication_app_base.h"
+#include "utils/error_code.h"
+#include "utils/errors.h"
+#include "utils/flags.h"
+
+namespace dsn {
+DSN_DECLARE_string(cluster_name);
+namespace security {
+DSN_DECLARE_bool(enable_acl);
+} // namespace security
+namespace replication {
+DSN_DECLARE_string(hadoop_kms_url);
+} // namespace replication
+} // namespace dsn
+
+class kms_client_test : public testing::Test
+{
+};
+
+TEST_F(kms_client_test, test_generate_and_decrypt_encryption_key)
+{
+    if (strlen(dsn::replication::FLAGS_hadoop_kms_url) == 0) {
+        GTEST_SKIP() << "Set a proper 'hadoop_kms_url' in config.ini to enable this test.";
+    }
+
+    auto _key_provider = std::make_unique<dsn::security::kms_key_provider>(
+        ::absl::StrSplit(dsn::replication::FLAGS_hadoop_kms_url, ",", ::absl::SkipEmpty()),
+        dsn::FLAGS_cluster_name);
+    dsn::replication::kms_info info;
+
+    // 1. generate encryption key.
+    ASSERT_EQ(dsn::ERR_OK, _key_provider->GenerateEncryptionKey(&info).code());
+
+    // 2. decrypt encryption key.
+    std::string server_key;
+    ASSERT_EQ(dsn::ERR_OK, _key_provider->DecryptEncryptionKey(info, &server_key).code());
+    ASSERT_EQ(server_key.size(), info.encrypted_key.length());
+    ASSERT_NE(server_key, info.encrypted_key);
+}
diff --git a/src/utils/env.cpp b/src/utils/env.cpp
index 444ac3784a..8c312d17b3 100644
--- a/src/utils/env.cpp
+++ b/src/utils/env.cpp
@@ -39,9 +39,9 @@ DSN_DEFINE_bool(pegasus.server,
                 "Whether the sensitive files should be encrypted on the file system.");
 
 DSN_DEFINE_string(pegasus.server,
-                  server_key_for_testing,
+                  server_key,
                   "0123456789ABCDEF0123456789ABCDEF",
-                  "The encrypted server key to use in the filesystem. NOTE: only for testing.");
+                  "The encrypted server key to use in the filesystem.");
 
 DSN_DEFINE_string(pegasus.server,
                   encryption_method,
@@ -62,10 +62,9 @@ rocksdb::Env *NewEncryptedEnv()
 {
     // Create an encryption provider.
     std::shared_ptr<rocksdb::EncryptionProvider> provider;
-    auto provider_id = fmt::format("id=AES;hex_instance_key={};method={}",
-                                   FLAGS_server_key_for_testing,
-                                   FLAGS_encryption_method);
-    auto s = rocksdb::EncryptionProvider::CreateFromString(
+    const auto &provider_id = fmt::format(
+        "id=AES;hex_instance_key={};method={}", FLAGS_server_key, FLAGS_encryption_method);
+    const auto &s = rocksdb::EncryptionProvider::CreateFromString(
         rocksdb::ConfigOptions(), provider_id, &provider);
     CHECK(s.ok(), "Failed to create encryption provider: {}", s.ToString());
 
diff --git a/src/utils/metrics.h b/src/utils/metrics.h
index 09225ac011..848697c902 100644
--- a/src/utils/metrics.h
+++ b/src/utils/metrics.h
@@ -38,6 +38,7 @@
 #include <utility>
 #include <vector>
 
+#include "absl/strings/string_view.h"
 #include "common/json_helper.h"
 #include "http/http_server.h"
 #include "utils/alloc.h"
@@ -50,7 +51,6 @@
 #include "utils/nth_element.h"
 #include "utils/ports.h"
 #include "utils/singleton.h"
-#include "absl/strings/string_view.h"
 #include "utils/synchronize.h"
 #include "utils/time_utils.h"