Implement OSD-side authorization for operations

src/client/cluster_client.cpp

@@ -51,7 +51,7 @@ cluster_client_t::cluster_client_t(ring_loop_t *ringloop, timerfd_manager_t *tfd
         msgr.stop_client(op->client_id);
         delete op;
     };
-    msgr.parse_config(config);
+    msgr.parse_config(config, true);
 
     st_cli.tfd = tfd;
     st_cli.on_load_config_hook = [this](json11::Json::object & cfg) { on_load_config_hook(cfg); };
@@ -481,7 +481,7 @@ void cluster_client_t::on_load_config_hook(json11::Json::object & etcd_global_co
     }
     // vault
     vault_parse_config();
-    msgr.parse_config(config);
+    msgr.parse_config(config, false);
     st_cli.parse_config(config);
     st_cli.load_pgs();
 }
@@ -1600,6 +1600,9 @@ static inline void mem_or(void *res, const void *r2, unsigned int len)
     }
 }
 
+// Error priority: others > EPERM > EIO > ENOSPC > ETIMEDOUT > EPIPE
+#define ERR_PRIO(e) (((e) == -EPERM ? 5 : ((e) == -EIO ? 4 : ((e) == -ENOSPC ? 3 : ((e) == -ETIMEDOUT ? 2 : ((e) == -EPIPE ? 1 : (!(e) ? 0 : 10)))))))
+
 void cluster_client_t::handle_op_part(cluster_op_part_t *part)
 {
     cluster_op_t *op = part->parent;
@@ -1608,15 +1611,10 @@ void cluster_client_t::handle_op_part(cluster_op_part_t *part)
     {
         // Operation failed, retry
         part->flags |= PART_ERROR;
-        if (!op->retval || op->retval == -EPIPE ||
-            part->op.reply.hdr.retval == -ENOSPC && op->retval == -ETIMEDOUT ||
-            part->op.reply.hdr.retval == -EIO)
-        {
-            // Error priority: EIO > ENOSPC > ETIMEDOUT > EPIPE
+        if (ERR_PRIO(part->op.reply.hdr.retval) > ERR_PRIO(op->retval))
             op->retval = part->op.reply.hdr.retval;
-        }
         uint64_t stop_client_id = 0;
-        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC)
+        if (op->retval != -EINTR && op->retval != -EIO && op->retval != -ENOSPC && op->retval != -EPERM)
         {
             stop_client_id = part->op.client_id;
             if (op->retval != -EPIPE || log_level > 0)

src/client/etcd_state_client.cpp

@@ -12,6 +12,7 @@
 #include "http_client.h"
 #endif
 #include "str_util.h"
+#include "json_util.h"
 
 etcd_state_client_t::~etcd_state_client_t()
 {
@@ -80,6 +81,46 @@ std::vector<std::string> etcd_state_client_t::get_addresses()
     return addrs;
 }
 
+std::shared_ptr<user_info_t> etcd_state_client_t::get_user(const std::string & username)
+{
+    auto user_it = user_info.find(username);
+    if (user_it != user_info.end())
+    {
+        return user_it->second;
+    }
+    auto inf = std::make_shared<user_info_t>();
+    inf->name = username;
+    return inf;
+}
+
+bool etcd_state_client_t::check_image_perm(const std::shared_ptr<user_info_t> & user_info, inode_t inode_num, bool write)
+{
+    if (user_info->type == user_type_t::ADMIN)
+    {
+        return true;
+    }
+    auto cache_it = user_info->perm_cache.find(inode_num);
+    if (cache_it != user_info->perm_cache.end() &&
+        cache_it->second.mod_revision == user_perm_cache_revision)
+    {
+        return write ? (cache_it->second.perm == user_perm_t::OWNER) : (cache_it->second.perm != user_perm_t::DENY);
+    }
+    auto inode_it = inode_config.find(inode_num);
+    if (inode_it == inode_config.end())
+    {
+        return false;
+    }
+    // FIXME Implement cache reset after reworking etcd interaction to not keep everything in memory
+    auto & perm_item = user_info->perm_cache[inode_num];
+    perm_item.mod_revision = user_perm_cache_revision;
+    perm_item.perm = (user_info->name == inode_it->second.owner || inode_it->second.owner_group != "" &&
+        user_info->groups.find(inode_it->second.owner_group) != user_info->groups.end()
+        ? user_perm_t::OWNER : (inode_it->second.reader_group != "" &&
+            user_info->groups.find(inode_it->second.reader_group) != user_info->groups.end()
+            ? user_perm_t::READER : user_perm_t::DENY));
+    return write ? (perm_item.perm == user_perm_t::OWNER) : (perm_item.perm != user_perm_t::DENY);
+}
+
 http_context_t *etcd_state_client_t::get_http_ctx()
 {
     if (!http_ctx)
@@ -252,17 +293,21 @@ void etcd_state_client_t::parse_config(const json11::Json & config)
     }
     if (this->osd_num)
     {
-        this->etcd_client_cert = config["osd_etcd_client_cert"].string_value();
-        this->etcd_client_key = config["osd_etcd_client_key"].string_value();
+        this->etcd_client_cert = config["osd_cert"].string_value();
+        this->etcd_client_key = config["osd_pkey"].string_value();
     }
     else
+    {
+        this->etcd_client_cert = config["cert"].string_value();
+        this->etcd_client_key = config["pkey"].string_value();
+    }
+    if (this->etcd_client_cert == "")
     {
         this->etcd_client_cert = config["etcd_client_cert"].string_value();
         this->etcd_client_key = config["etcd_client_key"].string_value();
     }
     this->etcd_ca = config["etcd_ca"].string_value();
     this->etcd_prefix = config["etcd_prefix"].string_value();
-    this->use_auth = config["use_auth"].bool_value();
     if (this->etcd_prefix == "")
     {
         this->etcd_prefix = "/vitastor";
@@ -1339,6 +1384,10 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
                 {
                     on_inode_change_hook(inode_num, true);
                 }
+                if (this->inode_config.find(inode_num) != this->inode_config.end())
+                {
+                    user_perm_cache_revision = kv.mod_revision;
+                }
                 this->inode_config.erase(inode_num);
             }
             else
@@ -1354,13 +1403,39 @@ void etcd_state_client_t::parse_state(const etcd_kv_t & kv)
         if (on_change_node_placement_hook)
             on_change_node_placement_hook();
     }
-    else if (use_auth && key.substr(0, etcd_prefix.length()+13) == etcd_prefix+"/config/user/")
+    else if (key.substr(0, etcd_prefix.length()+13) == etcd_prefix+"/config/user/")
     {
         // <etcd_prefix>/config/user/<username>
+        auto name = key.substr(etcd_prefix.length()+13);
+        auto & inf = user_info[name];
         if (!value.is_object())
-            user_info.erase(key.substr(etcd_prefix.length()+13));
+        {
+            if (inf)
+            {
+                inf->type = user_type_t::CLIENT;
+                inf->groups.clear();
+                inf->perm_cache.clear();
+            }
+            user_info.erase(name);
+        }
         else
-            user_info[key.substr(etcd_prefix.length()+13)] = value;
+        {
+            if (!inf)
+            {
+                inf = std::make_shared<user_info_t>();
+                inf->name = name;
+            }
+            inf->type = value["type"] == "admin" ? user_type_t::ADMIN :
+                (value["type"] == "mon" ? user_type_t::MON :
+                (value["type"] == "osd" ? user_type_t::OSD : user_type_t::CLIENT));
+            inf->groups.clear();
+            for (auto & group: value["groups"].array_items())
+            {
+                if (group.string_value() != "")
+                    inf->groups.insert(group.string_value());
+            }
+            inf->perm_cache.clear();
+        }
     }
 }
 
@@ -1384,7 +1459,12 @@ uint32_t etcd_state_client_t::parse_scheme(const std::string & scheme)
 
 void etcd_state_client_t::insert_inode_config(const inode_config_t & cfg)
 {
-    this->inode_config[cfg.num] = cfg;
+    auto & cfg_ref = this->inode_config[cfg.num];
+    if (cfg_ref.mod_revision != cfg.mod_revision)
+    {
+        user_perm_cache_revision = cfg.mod_revision;
+    }
+    cfg_ref = cfg;
     if (cfg.name != "")
     {
         this->inode_by_name[cfg.name] = cfg.num;

src/client/etcd_state_client.h

@@ -9,6 +9,7 @@
 #include "json11/json11.hpp"
 #include "object_id.h"
 #include "timerfd_manager.h"
+#include "../util/robin_hood.h"
 
 #define ETCD_CONFIG_WATCH_ID 1
 #define ETCD_OSD_STATE_WATCH_ID 2
@@ -110,6 +111,32 @@ struct http_url_t
     std::string path;
 };
 
+enum class user_type_t
+{
+    CLIENT = 0,
+    ADMIN = 1,
+    MON = 2,
+    OSD = 3,
+};
+
+struct user_perm_t
+{
+    enum class perm_type_t: uint8_t;
+    constexpr static perm_type_t DENY = (perm_type_t)0;
+    constexpr static perm_type_t READER = (perm_type_t)1;
+    constexpr static perm_type_t OWNER = (perm_type_t)2;
+    uint64_t mod_revision = 0;
+    perm_type_t perm = DENY;
+};
+
+struct user_info_t
+{
+    std::string name;
+    user_type_t type;
+    robin_hood::unordered_flat_set<std::string> groups;
+    robin_hood::unordered_flat_map<inode_t, user_perm_t> perm_cache;
+};
+
 struct http_co_t;
 struct http_context_t;
 
@@ -145,7 +172,6 @@ public:
     int etcd_slow_timeout = 5000;
     int etcd_min_reload_interval = 1000;
     bool infinite_start = true;
-    bool use_auth = false;
     uint64_t global_block_size = DEFAULT_BLOCK_SIZE;
     uint32_t global_bitmap_granularity = DEFAULT_BITMAP_GRANULARITY;
     uint32_t global_immediate_commit = IMMEDIATE_NONE;
@@ -171,7 +197,8 @@ public:
     std::set<osd_num_t> seen_peers;
     std::map<inode_t, inode_config_t> inode_config;
     std::map<std::string, inode_t> inode_by_name;
-    std::map<std::string, json11::Json> user_info;
+    robin_hood::unordered_flat_map<std::string, std::shared_ptr<user_info_t>> user_info;
+    uint64_t user_perm_cache_revision = 0;
     json11::Json node_placement;
 
     std::function<void(std::map<std::string, etcd_kv_t> &)> on_change_hook;
@@ -193,6 +220,8 @@ public:
     inode_config_t deserialize_inode_cfg(uint64_t inode_num, json11::Json value, uint64_t mod_revision);
     etcd_kv_t parse_etcd_kv(const json11::Json & kv_json);
     std::vector<std::string> get_addresses();
+    std::shared_ptr<user_info_t> get_user(const std::string & username);
+    bool check_image_perm(const std::shared_ptr<user_info_t> & user_info, inode_t inode_num, bool write);
     http_context_t *get_http_ctx();
     void etcd_call_oneshot(const std::string & etcd_address, const std::string & api, json11::Json payload, int timeout, std::function<void(std::string, json11::Json)> callback);
     void etcd_call(const std::string & api, json11::Json payload, int timeout, int retries, int interval, std::function<void(std::string, json11::Json)> callback);

src/client/messenger.cpp

@@ -308,8 +308,41 @@ osd_messenger_t::~osd_messenger_t()
     destroy_tls();
 }
 
-void osd_messenger_t::parse_config(const json11::Json & config)
+void osd_messenger_t::parse_config(const json11::Json & config, bool init)
 {
+    this->max_cipher_pool_size = config["max_cipher_pool_size"].uint64_value();
+    if (!this->max_cipher_pool_size)
+        this->max_cipher_pool_size = 256;
+    if (config["proto_checksums"].is_null())
+        this->use_proto_checksums = MSGR_CSUM_PAYLOAD;
+    else if (config["proto_checksums"].is_bool())
+        this->use_proto_checksums = config["proto_checksums"].bool_value() ? MSGR_CSUM_FULL : 0;
+    else if (config["proto_checksums"].string_value() != "")
+        this->use_proto_checksums = config["proto_checksums"].string_value() == "full" ? MSGR_CSUM_FULL : MSGR_CSUM_PAYLOAD;
+    else
+        this->use_proto_checksums = 0;
+    this->receive_buffer_size = (uint32_t)config["tcp_header_buffer_size"].uint64_value();
+    if (!this->receive_buffer_size || this->receive_buffer_size > 1024*1024*1024)
+        this->receive_buffer_size = 65536;
+    this->min_zerocopy_send_size = config["min_zerocopy_send_size"].is_null()
+        ? DEFAULT_MIN_ZEROCOPY_SEND_SIZE
+        : (int)config["min_zerocopy_send_size"].int64_value();
+    this->peer_connect_interval = config["peer_connect_interval"].uint64_value();
+    if (!this->peer_connect_interval)
+        this->peer_connect_interval = 5;
+    this->peer_connect_timeout = config["peer_connect_timeout"].uint64_value();
+    if (!this->peer_connect_timeout)
+        this->peer_connect_timeout = 5;
+    this->osd_idle_timeout = config["osd_idle_timeout"].uint64_value();
+    if (!this->osd_idle_timeout)
+        this->osd_idle_timeout = 5;
+    this->osd_ping_timeout = config["osd_ping_timeout"].uint64_value();
+    if (!this->osd_ping_timeout)
+        this->osd_ping_timeout = 5;
+    this->log_level = config["log_level"].uint64_value();
+    // All other parameters are only set on init
+    if (!init)
+        return;
 #ifdef WITH_RDMA
     if (!config["use_rdma"].is_null())
     {
@@ -340,17 +373,6 @@ void osd_messenger_t::parse_config(const json11::Json & config)
     if (!this->rdma_max_msg || this->rdma_max_msg > 128*1024*1024)
         this->rdma_max_msg = 129*1024;
 #endif
-    this->max_cipher_pool_size = config["max_cipher_pool_size"].uint64_value();
-    if (!this->max_cipher_pool_size)
-        this->max_cipher_pool_size = 256;
-    if (config["proto_checksums"].is_null())
-        this->use_proto_checksums = MSGR_CSUM_PAYLOAD;
-    else if (config["proto_checksums"].is_bool())
-        this->use_proto_checksums = config["proto_checksums"].bool_value() ? MSGR_CSUM_FULL : 0;
-    else if (config["proto_checksums"].string_value() != "")
-        this->use_proto_checksums = config["proto_checksums"].string_value() == "full" ? MSGR_CSUM_FULL : MSGR_CSUM_PAYLOAD;
-    else
-        this->use_proto_checksums = 0;
     if (!osd_num)
     {
         tls_cert = config["cert"].string_value();
@@ -368,27 +390,8 @@ void osd_messenger_t::parse_config(const json11::Json & config)
         this->iothread_count = (uint32_t)config["client_iothread_count"].uint64_value();
     else
         this->iothread_count = (uint32_t)config["osd_iothread_count"].uint64_value();
-    this->receive_buffer_size = (uint32_t)config["tcp_header_buffer_size"].uint64_value();
-    if (!this->receive_buffer_size || this->receive_buffer_size > 1024*1024*1024)
-        this->receive_buffer_size = 65536;
     this->use_sync_send_recv = config["use_sync_send_recv"].bool_value() ||
         config["use_sync_send_recv"].uint64_value() || !ringloop;
-    this->min_zerocopy_send_size = config["min_zerocopy_send_size"].is_null()
-        ? DEFAULT_MIN_ZEROCOPY_SEND_SIZE
-        : (int)config["min_zerocopy_send_size"].int64_value();
-    this->peer_connect_interval = config["peer_connect_interval"].uint64_value();
-    if (!this->peer_connect_interval)
-        this->peer_connect_interval = 5;
-    this->peer_connect_timeout = config["peer_connect_timeout"].uint64_value();
-    if (!this->peer_connect_timeout)
-        this->peer_connect_timeout = 5;
-    this->osd_idle_timeout = config["osd_idle_timeout"].uint64_value();
-    if (!this->osd_idle_timeout)
-        this->osd_idle_timeout = 5;
-    this->osd_ping_timeout = config["osd_ping_timeout"].uint64_value();
-    if (!this->osd_ping_timeout)
-        this->osd_ping_timeout = 5;
-    this->log_level = config["log_level"].uint64_value();
     // OSD public & cluster networks
     this->osd_networks.clear();
     if (config["osd_network"].is_string())
@@ -729,6 +732,11 @@ void osd_messenger_t::check_peer_config(osd_client_t *cl)
             err = true;
             fprintf(stderr, "Failed to get config from OSD %ju (retval=%jd), disconnecting peer\n", cl->osd_num, op->reply.hdr.retval);
         }
+        else if (cl->gcm_enabled && !cl->hs_result.peer_is_osd)
+        {
+            err = true;
+            fprintf(stderr, "Client %ju is not authenticated as an OSD, disconnecting peer\n", cl->client_id);
+        }
         else
         {
             config = json11::Json::parse(std::string((char*)op->buf), json_err);
@@ -879,6 +887,11 @@ bool osd_messenger_t::is_use_rdmacm()
 }
 #endif
 
+bool osd_messenger_t::is_encryption_enabled()
+{
+    return tls_cert != "" || tls_key != "" || osd_tls_ca != "";
+}
+
 json11::Json::object osd_messenger_t::read_config(const json11::Json & config)
 {
     json11::Json::object file_config;

src/client/messenger.h

@@ -66,6 +66,8 @@ struct op_aes_xts_decrypt_t;
 void destroy_aes_xts_encrypt(op_aes_xts_encrypt_t *encrypt_ctx);
 void destroy_aes_xts_decrypt(op_aes_xts_decrypt_t *decrypt_ctx);
 
+struct user_info_t;
+
 struct osd_client_t
 {
     uint64_t client_id = 0;
@@ -94,6 +96,7 @@ struct osd_client_t
     bool gcm_enabled = false;
     msgr_handshake_i *hs = NULL;
     msgr_handshake_result_t hs_result;
+    std::shared_ptr<user_info_t> user_info;
     std::vector<uint8_t> my_key, peer_key;
     uint64_t my_iv_ctr = 0, peer_iv_ctr = 0;
 #ifdef WITH_ISAL_CRYPTO
@@ -322,7 +325,7 @@ public:
     osd_op_stats_t stats, recovery_stats;
 
     void init();
-    void parse_config(const json11::Json & config);
+    void parse_config(const json11::Json & config, bool init);
     void connect_peer(uint64_t osd_num, json11::Json peer_state);
     void stop_client(uint64_t client_id, bool force_delete = false);
     void destroy_client(osd_client_t *cl);
@@ -331,6 +334,7 @@ public:
     std::function<void(osd_num_t)> repeer_pgs;
     std::function<void(osd_num_t)> break_pg_locks;
     std::function<bool(osd_client_t*, json11::Json)> check_config_hook;
+    std::function<void(osd_client_t*)> handshake_hook;
     void read_requests();
     void send_replies();
     void accept_connections(int listen_fd);
@@ -351,6 +355,7 @@ public:
     rdma_cm_id *rdmacm_listen(const std::string & bind_address, int rdmacm_port, int *bound_port, int log_level);
     void rdmacm_destroy_listener(rdma_cm_id *listener);
 #endif
+    bool is_encryption_enabled();
 
     void inc_op_stats(osd_op_stats_t & stats, uint64_t opcode, timespec & tv_begin, timespec & tv_end, uint64_t len);
     void measure_exec(osd_op_t *cur_op);

src/client/msgr_encrypt.cpp

@@ -491,6 +491,10 @@ bool osd_messenger_t::derive_aes_keys(osd_client_t *cl, bool update_my, bool upd
     if (!cl->hs_result.shared_secret.size())
     {
         cl->hs_result = cl->hs->get_result();
+        if (handshake_hook)
+        {
+            handshake_hook(cl);
+        }
     }
     std::vector<uint8_t> old_my = cl->my_key, old_peer = cl->peer_key;
     // Both keys include AES key and iv + xxhash3 secret