feat(security): implement reset interface for replica access controll…

…er (#672)

include/dsn/dist/replication/replica_envs.h

@@ -54,6 +54,7 @@ class replica_envs
     static const std::string MANUAL_COMPACT_PERIODIC_TARGET_LEVEL;
     static const std::string MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION;
     static const std::string BUSINESS_INFO;
+    static const std::string REPLICA_ACCESS_CONTROLLER_ALLOWED_USERS;
 };
 
 } // namespace replication

src/common/replication_common.cpp

@@ -629,6 +629,8 @@ const std::string replica_envs::ROCKSDB_CHECKPOINT_RESERVE_TIME_SECONDS(
 const std::string replica_envs::ROCKSDB_ITERATION_THRESHOLD_TIME_MS(
     "replica.rocksdb_iteration_threshold_time_ms");
 const std::string replica_envs::BUSINESS_INFO("business.info");
+const std::string replica_envs::REPLICA_ACCESS_CONTROLLER_ALLOWED_USERS(
+    "replica_access_controller.allowed_users");
 
 const std::string bulk_load_constant::BULK_LOAD_INFO("bulk_load_info");
 const int32_t bulk_load_constant::BULK_LOAD_REQUEST_INTERVAL = 10;

src/meta/app_env_validator.cpp

@@ -167,7 +167,8 @@ void app_env_validator::register_all_validators()
         {replica_envs::MANUAL_COMPACT_ONCE_BOTTOMMOST_LEVEL_COMPACTION, nullptr},
         {replica_envs::MANUAL_COMPACT_PERIODIC_TRIGGER_TIME, nullptr},
         {replica_envs::MANUAL_COMPACT_PERIODIC_TARGET_LEVEL, nullptr},
-        {replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION, nullptr}};
+        {replica_envs::MANUAL_COMPACT_PERIODIC_BOTTOMMOST_LEVEL_COMPACTION, nullptr},
+        {replica_envs::REPLICA_ACCESS_CONTROLLER_ALLOWED_USERS, nullptr}};
 }
 
 } // namespace replication

src/replica/replica.h

@@ -127,6 +127,8 @@ class replica : public serverlet<replica>, public ref_counter, public replica_ba
     void update_throttle_env_internal(const std::map<std::string, std::string> &envs,
                                       const std::string &key,
                                       throttling_controller &cntl);
+    // update allowed users for access controller
+    void update_ac_allowed_users(const std::map<std::string, std::string> &envs);
 
     //
     //    messages and tools from/for meta server

src/replica/replica_config.cpp

@@ -39,6 +39,7 @@
 #include "mutation_log.h"
 #include "replica_stub.h"
 #include "bulk_load/replica_bulk_loader.h"
+#include "runtime/security/access_controller.h"
 #include "split/replica_split_manager.h"
 #include <dsn/dist/fmt_logging.h>
 #include <dsn/dist/replication/replication_app_base.h>
@@ -569,6 +570,19 @@ void replica::update_app_envs_internal(const std::map<std::string, std::string>
     }
 
     update_throttle_envs(envs);
+
+    update_ac_allowed_users(envs);
+}
+
+void replica::update_ac_allowed_users(const std::map<std::string, std::string> &envs)
+{
+    std::string allowed_users;
+    auto iter = envs.find(replica_envs::REPLICA_ACCESS_CONTROLLER_ALLOWED_USERS);
+    if (iter != envs.end()) {
+        allowed_users = iter->second;
+    }
+
+    _access_controller->update(allowed_users);
 }
 
 void replica::query_app_envs(/*out*/ std::map<std::string, std::string> &envs)

src/runtime/security/access_controller.h

@@ -31,10 +31,10 @@ class access_controller
     virtual ~access_controller() = 0;
 
     /**
-     * reset the access controller
-     *    acls - the new acls to reset
+     * update the access controller
+     *    acls - the new acls to update
      **/
-    virtual void reset(const std::string &acls){};
+    virtual void update(const std::string &acls){};
 
     /**
      * check if the message received is allowd to do something.

src/runtime/security/replica_access_controller.cpp

@@ -41,5 +41,25 @@ bool replica_access_controller::allowed(message_ex *msg)
         return true;
     }
 }
+
+void replica_access_controller::update(const std::string &users)
+{
+    {
+        // check to see whether we should update it or not.
+        utils::auto_read_lock l(_lock);
+        if (_env_users == users) {
+            return;
+        }
+    }
+
+    std::unordered_set<std::string> users_set;
+    utils::split_args(users.c_str(), users_set, ',');
+    {
+        utils::auto_write_lock l(_lock);
+        // This swap operation is in constant time
+        _users.swap(users_set);
+        _env_users = users;
+    }
+}
 } // namespace security
 } // namespace dsn

src/runtime/security/replica_access_controller.h

@@ -27,10 +27,12 @@ class replica_access_controller : public access_controller
 public:
     replica_access_controller(const std::string &name);
     bool allowed(message_ex *msg);
+    void update(const std::string &users);
 
 private:
     utils::rw_lock_nr _lock; // [
     std::unordered_set<std::string> _users;
+    std::string _env_users;
     // ]
     std::string _name;