Implement OSD-side authorization for operations

src/osd/osd.cpp

@@ -15,6 +15,7 @@
 #include "http_client.h"
 #include "str_util.h"
 #include "json_util.h"
+#include "openssl_util.h"
 
 osd_t::osd_t(const json11::Json & config, ring_loop_i *ringloop, timerfd_manager_t *tfd,
     std::unique_ptr<etcd_state_client_t> st_cli_ptr, std::function<blockstore_i*(blockstore_config_t & config)> bs_factory)
@@ -71,6 +72,11 @@ void osd_t::start()
     msgr.repeer_pgs = [this](osd_num_t peer_osd) { repeer_pgs(peer_osd); };
     msgr.break_pg_locks = [this](osd_num_t peer_osd) { break_pg_locks(peer_osd); };
     msgr.check_config_hook = [this](osd_client_t *cl, json11::Json conf) { return check_peer_config(cl, conf); };
+    msgr.handshake_hook = [this](osd_client_t *cl)
+    {
+        if (!cl->hs_result.peer_is_osd)
+            cl->user_info = st_cli->get_user(openssl_get_cn(cl->hs_result.peer_cert));
+    };
     msgr.init();
 
     init_cluster();
@@ -182,9 +188,13 @@ void osd_t::parse_config(bool init)
         bs->parse_config(bs_cfg);
     }
     st_cli->parse_config(config);
-    msgr.parse_config(config);
+    msgr.parse_config(config, init);
     if (init)
     {
+        // use_auth is enabled by default when encryption is enabled
+        use_auth = (config["use_auth"].is_null()
+            ? msgr.is_encryption_enabled()
+            : json_is_true(config["use_auth"]));
         // Vital Blockstore parameters
         bs_block_size = config["block_size"].uint64_value();
         if (!bs_block_size)

src/osd/osd.h

@@ -29,6 +29,8 @@
 #define OSD_RECOVERING 0x10
 #define OSD_SCRUBBING 0x20
 
+#define SELF_CLIENT 0
+
 #define MAX_AUTOSYNC_INTERVAL 3600
 #define DEFAULT_AUTOSYNC_INTERVAL 5
 #define DEFAULT_AUTOSYNC_WRITES 128
@@ -159,6 +161,7 @@ class osd_t
     std::unique_ptr<etcd_state_client_t> st_cli;
     std::function<blockstore_i*(blockstore_config_t & config)> bs_factory;
     osd_messenger_t msgr;
+    bool use_auth = false;
     int etcd_failed_attempts = 0;
     std::string etcd_lease_id;
     json11::Json self_state;

src/osd/osd_flush.cpp

@@ -4,7 +4,6 @@
 #include "osd.h"
 
 #define FLUSH_BATCH 512
-#define SELF_CLIENT 0
 
 static_assert(FLUSH_BATCH <= MAX_SIMPLE_PAYLOAD_SIZE / sizeof(obj_ver_id));
 

src/osd/osd_peering.cpp

@@ -9,8 +9,6 @@
 #include "str_util.h"
 #include "osd.h"
 
-#define SELF_CLIENT 0
-
 // Peering loop
 void osd_t::handle_peers()
 {

src/osd/osd_primary.cpp

@@ -71,6 +71,21 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op)
         finish_op(cur_op, -EINVAL);
         return false;
     }
+    if (use_auth && cur_op->client_id != SELF_CLIENT)
+    {
+        osd_client_t *cl = msgr.clients.at(cur_op->client_id);
+        if (cl->hs_result.peer_is_osd)
+        {
+            // OSDs are not allowed to execute "primary" operations
+            finish_op(cur_op, -EPERM);
+            return false;
+        }
+        if (!st_cli->check_image_perm(cl->user_info, cur_op->req.rw.inode, (cur_op->req.hdr.opcode != OSD_OP_READ)))
+        {
+            finish_op(cur_op, -EPERM);
+            return false;
+        }
+    }
     int stripe_count = (cur_op->req.hdr.opcode == OSD_OP_SCRUB ? 0 :
         (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 1 : pg_it->second.pg_size));
     int chain_size = 0;

src/osd/osd_primary_describe.cpp

@@ -91,6 +91,21 @@ static void scan_lists(std::vector<unclean_list_t> & lists, uint64_t limit, desc
 // Describe unclean objects
 void osd_t::continue_primary_describe(osd_op_t *cur_op)
 {
+    if (use_auth)
+    {
+        osd_client_t *cl = msgr.clients.at(cur_op->client_id);
+        if (cl->hs_result.peer_is_osd)
+        {
+            // OSDs are not allowed to execute "primary" operations
+            finish_op(cur_op, -EPERM);
+            return;
+        }
+        if (cl->user_info->type != user_type_t::ADMIN)
+        {
+            finish_op(cur_op, -EPERM);
+            return;
+        }
+    }
     auto & desc = cur_op->req.describe;
     if (!desc.object_state)
         desc.object_state = ~desc.object_state;

src/osd/osd_primary_subops.cpp

@@ -3,8 +3,6 @@
 
 #include "osd_primary.h"
 
-#define SELF_CLIENT 0
-
 void osd_t::autosync()
 {
     if (immediate_commit != IMMEDIATE_ALL && !autosync_op)

src/osd/osd_primary_sync.cpp

@@ -8,6 +8,16 @@ void osd_t::continue_primary_sync(osd_op_t *cur_op)
 {
     if (!cur_op->op_data)
     {
+        if (use_auth && cur_op->client_id != SELF_CLIENT)
+        {
+            osd_client_t *cl = msgr.clients.at(cur_op->client_id);
+            if (cl->hs_result.peer_is_osd)
+            {
+                // OSDs are not allowed to execute "primary" operations
+                finish_op(cur_op, -EPERM);
+                return;
+            }
+        }
         cur_op->op_data = (osd_primary_op_data_t*)calloc_or_die(1, sizeof(osd_primary_op_data_t));
     }
     osd_primary_op_data_t *op_data = cur_op->op_data;

src/osd/osd_scrub.cpp

@@ -3,8 +3,6 @@
 
 #include "osd_primary.h"
 
-#define SELF_CLIENT 0
-
 void osd_t::scrub_list(pool_pg_num_t pg_id, osd_num_t role_osd, object_id min_oid)
 {
     pool_id_t pool_id = pg_id.pool_id;

src/osd/osd_secondary.cpp

@@ -1,9 +1,14 @@
 // Copyright (c) Vitaliy Filippov, 2019+
 // License: VNPL-1.1 (see README.md for details)
 
+#include "json11/json11.hpp"
+
 #include "osd.h"
 
-#include "json11/json11.hpp"
+#ifdef WITH_RDMA
+#include "msgr_rdma.h"
+#endif
+#include "openssl_util.h"
 
 void osd_t::secondary_op_callback(osd_op_t *op)
 {
@@ -109,6 +114,31 @@ bool osd_t::sec_check_pg_lock(osd_num_t primary_osd, const object_id &oid, uint3
 
 void osd_t::exec_secondary_real(osd_op_t *cur_op)
 {
+    osd_client_t *cl = msgr.clients.at(cur_op->client_id);
+    if (use_auth && !cl->hs_result.peer_is_osd)
+    {
+        // Non-OSDs are not allowed to execute "secondary" operations except LIST
+        bool allowed = false;
+        if (cur_op->req.hdr.opcode == OSD_OP_SEC_LIST)
+        {
+            if (cl->user_info->type == user_type_t::ADMIN)
+            {
+                // Admin is allowed to execute arbitrary listings
+                allowed = true;
+            }
+            else if (cl->user_info->type == user_type_t::CLIENT && cur_op->req.sec_list.min_inode &&
+                cur_op->req.sec_list.min_inode == cur_op->req.sec_list.max_inode)
+            {
+                // Clients are only allowed to execute listings for readable inodes
+                allowed = st_cli->check_image_perm(cl->user_info, cur_op->req.sec_list.min_inode, false);
+            }
+        }
+        if (!allowed)
+        {
+            finish_op(cur_op, -EPERM);
+            return;
+        }
+    }
     if (cur_op->req.hdr.opcode == OSD_OP_SEC_LIST &&
         (cur_op->req.sec_list.flags & OSD_LIST_PRIMARY))
     {
@@ -125,7 +155,6 @@ void osd_t::exec_secondary_real(osd_op_t *cur_op)
         exec_sec_lock(cur_op);
         return;
     }
-    osd_client_t *cl = msgr.clients.at(cur_op->client_id);
     cur_op->bs_op = new blockstore_op_t();
     cur_op->bs_op->callback = [this, cur_op](blockstore_op_t* bs_op) { secondary_op_callback(cur_op); };
     cur_op->bs_op->opcode = (cur_op->req.hdr.opcode == OSD_OP_SEC_READ ? BS_OP_READ