From b2a74de715d1c36de7e0b6b7a9416ce888f411f2 Mon Sep 17 00:00:00 2001 From: Vitaliy Filippov Date: Sun, 5 Apr 2026 19:43:18 +0300 Subject: [PATCH] Implement OSD TLS support --- src/client/http_client.cpp | 41 +- src/client/http_client.h | 12 + src/client/messenger.cpp | 106 ++++- src/client/messenger.h | 65 ++- src/client/msgr_encrypt.cpp | 113 +++--- src/client/msgr_op.h | 2 +- src/client/msgr_rdma.cpp | 34 +- src/client/msgr_rdma.h | 2 +- src/client/msgr_receive.cpp | 655 +++++++++++++++++++------------ src/client/msgr_send.cpp | 720 ++++++++++++++++++++++++++-------- src/client/msgr_stop.cpp | 25 +- src/osd/osd_primary.cpp | 23 +- src/osd/osd_primary_chain.cpp | 3 +- src/osd/osd_scrub.cpp | 2 +- tests/common.sh | 18 + 15 files changed, 1267 insertions(+), 554 deletions(-) diff --git a/src/client/http_client.cpp b/src/client/http_client.cpp index 7768c429..c9554e2a 100644 --- a/src/client/http_client.cpp +++ b/src/client/http_client.cpp @@ -164,30 +164,43 @@ void http_ares_cb(void *data, ares_socket_t socket_fd, int readable, int writabl } #ifdef WITH_OPENSSL +bool openssl_ctx_add_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem) +{ + std::string pem; + BIO *bio = NULL; + if (file_or_pem.substr(0, 5) != "-----") + { + pem = read_file(file_or_pem); + bio = BIO_new_mem_buf(pem.data(), pem.size()); + } + else + bio = BIO_new_mem_buf(file_or_pem.data(), file_or_pem.size()); + if (!bio) + return false; + X509 *x509 = PEM_read_bio_X509(bio, NULL, 0, NULL); + bool ok = !!x509; + if (x509) + { + X509_STORE *store = SSL_CTX_get_cert_store(ssl_ctx); + X509_STORE_add_cert(store, x509); + X509_free(x509); + } + BIO_free(bio); + return ok; +} + bool openssl_ctx_use_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem) { if (file_or_pem.substr(0, 5) == "-----") { - BIO *bio = BIO_new_mem_buf(file_or_pem.data(), file_or_pem.size()); - if (!bio) - return false; - X509 *x509 = PEM_read_bio_X509(bio, NULL, 0, NULL); - bool ok = !!x509; - if (x509) - { - X509_STORE *store = SSL_CTX_get_cert_store(ssl_ctx); - X509_STORE_add_cert(store, x509); - X509_free(x509); - } - BIO_free(bio); - return ok; + return openssl_ctx_add_ca(ssl_ctx, file_or_pem); } return file_or_pem.empty() ? !!SSL_CTX_set_default_verify_paths(ssl_ctx) : !!SSL_CTX_load_verify_locations(ssl_ctx, file_or_pem.c_str(), NULL); } -static std::string openssl_get_cn(X509 *x509) +std::string openssl_get_cn(X509 *x509) { X509_NAME* subj = X509_get_subject_name(x509); int pos = X509_NAME_get_index_by_NID(subj, NID_commonName, -1); diff --git a/src/client/http_client.h b/src/client/http_client.h index fc663698..464fb2c5 100644 --- a/src/client/http_client.h +++ b/src/client/http_client.h @@ -8,6 +8,10 @@ #include #include "json11/json11.hpp" +#ifdef WITH_OPENSSL +#include +#endif + #define WS_CONTINUATION 0 #define WS_TEXT 1 #define WS_BINARY 2 @@ -69,3 +73,11 @@ void http_close(http_co_t *co); void http_destroy(http_co_t *co); #pragma GCC visibility pop + +#ifdef WITH_OPENSSL +bool openssl_ctx_add_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem); +bool openssl_ctx_use_ca(SSL_CTX *ssl_ctx, const std::string & file_or_pem); +std::string openssl_get_cn(X509 *x509); +bool openssl_ctx_use_cert(SSL_CTX *ssl_ctx, const std::string & file_or_pem, std::string & common_name); +bool openssl_ctx_use_key(SSL_CTX *ssl_ctx, const std::string & file_or_pem); +#endif diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index a6056169..194ad2f8 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -14,9 +14,60 @@ #ifdef WITH_RDMA #include "msgr_rdma.h" #endif +#include "http_client.h" +#ifdef WITH_OPENSSL +#include +#include +#include +#include +#endif void osd_messenger_t::init() { + if (!tls_cert.empty() || !tls_key.empty() || !osd_tls_ca.empty() || !client_tls_ca.empty()) + { + // Initialize TLS context + // FIXME: require OpenSSL +#ifndef WITH_OPENSSL + fprintf(stderr, "Vitastor is built without OpenSSL support\n"); + exit(1); +#else + if (tls_cert.empty() || tls_key.empty() || osd_tls_ca.empty() || osd_num && client_tls_ca.empty()) + { + if (osd_num) + fprintf(stderr, "Vitastor OSD TLS requires osd_tls_cert, osd_tls_key, osd_tls_ca, client_tls_ca\n"); + else + fprintf(stderr, "Vitastor client TLS requires tls_cert, tls_key and osd_tls_ca\n"); + exit(1); + } + else + { + ssl_ctx = SSL_CTX_new(TLS_method()); + if (!ssl_ctx) + { +init_err: + fprintf(stderr, "OpenSSL initialization failed: %s\n", ERR_error_string(ERR_get_error(), NULL)); + exit(1); + } + SSL_CTX_set_verify(ssl_ctx, SSL_VERIFY_PEER, NULL); + bool ok = SSL_CTX_set_min_proto_version(ssl_ctx, TLS1_3_VERSION); + ok = ok && openssl_ctx_add_ca(ssl_ctx, osd_tls_ca); + if (osd_num) + { + // OSD uses 2 separate root certificates to distinguish between clients and peer OSDs + ok = ok && openssl_ctx_add_ca(ssl_ctx, client_tls_ca); + } + ok = ok && openssl_ctx_use_cert(ssl_ctx, tls_cert, tls_cn); + ok = ok && openssl_ctx_use_key(ssl_ctx, tls_key); + if (!ok) + { + SSL_CTX_free(ssl_ctx); + ssl_ctx = NULL; + goto init_err; + } + } +#endif + } #ifdef WITH_RDMACM if (use_rdmacm) { @@ -192,6 +243,13 @@ osd_messenger_t::~osd_messenger_t() { destroy_aes_xts_decrypt(decrypt_ctx); } +#ifdef WITH_OPENSSL + if (ssl_ctx) + { + SSL_CTX_free(ssl_ctx); + ssl_ctx = NULL; + } +#endif } void osd_messenger_t::parse_config(const json11::Json & config) @@ -237,6 +295,19 @@ void osd_messenger_t::parse_config(const json11::Json & config) 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["tls_cert"].string_value(); + tls_key = config["tls_key"].string_value(); + osd_tls_ca = config["osd_tls_ca"].string_value(); + } + else + { + tls_cert = config["osd_tls_cert"].string_value(); + tls_key = config["osd_tls_key"].string_value(); + osd_tls_ca = config["osd_tls_ca"].string_value(); + client_tls_ca = config["client_tls_ca"].string_value(); + } if (!osd_num) this->iothread_count = (uint32_t)config["client_iothread_count"].uint64_value(); else @@ -245,7 +316,7 @@ void osd_messenger_t::parse_config(const json11::Json & config) 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(); + 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(); @@ -467,6 +538,10 @@ void osd_messenger_t::handle_connect_epoll(int peer_fd) handle_peer_epoll(peer_fd, epoll_events); }); // Check OSD number + if (!tls_cert.empty()) + { + ssl_init(cl, false); + } check_peer_config(cl); } @@ -708,6 +783,10 @@ void osd_messenger_t::accept_connections(int listen_fd) cl->peer_fd = peer_fd; cl->peer_state = PEER_CONNECTED; cl->in_buf = (uint8_t*)malloc_or_die(receive_buffer_size); + if (!tls_cert.empty()) + { + ssl_init(cl, true); + } // Add FD to epoll tfd->set_fd_handler(peer_fd, false, [this](int peer_fd, int epoll_events) { @@ -722,6 +801,31 @@ void osd_messenger_t::accept_connections(int listen_fd) } } +void osd_messenger_t::ssl_init(osd_client_t *cl, bool server_mode) +{ +#ifdef WITH_OPENSSL + cl->write_to_ssl = BIO_new(BIO_s_mem()); + cl->read_from_ssl = BIO_new(BIO_s_mem()); + cl->ssl_cli = SSL_new(ssl_ctx); + if (!cl->ssl_cli) + { + fprintf(stderr, "OpenSSL initialization failed: %s\n", ERR_error_string(ERR_get_error(), NULL)); + exit(1); + } + if (server_mode) + { + SSL_set_accept_state(cl->ssl_cli); + } + else + { + SSL_set_connect_state(cl->ssl_cli); + } + SSL_set_bio(cl->ssl_cli, cl->write_to_ssl, cl->read_from_ssl); + bool ok = ssl_do_handshake(cl); + assert(ok); +#endif +} + #ifdef WITH_RDMA msgr_rdma_context_t* osd_messenger_t::choose_rdma_context(osd_client_t *cl) { diff --git a/src/client/messenger.h b/src/client/messenger.h index b69af5be..15f3d1f8 100644 --- a/src/client/messenger.h +++ b/src/client/messenger.h @@ -12,6 +12,10 @@ #include #include +#ifdef WITH_OPENSSL +#include +#endif + #include "../util/xxh_x86dispatch.h" #include "../util/robin_hood.h" #include "malloc_or_die.h" @@ -58,6 +62,12 @@ 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 __attribute__((__packed__)) msgr_tls_record_hdr_t +{ + uint8_t encrypted; + uint32_t size; +}; + struct osd_client_t { uint64_t client_id = 0; @@ -80,7 +90,20 @@ struct osd_client_t msgr_rdma_connection_t *rdma_conn = NULL; #endif +#ifdef WITH_OPENSSL + SSL *ssl_cli = NULL; + BIO *write_to_ssl = NULL; + // FIXME: use custom bio to avoid 1 more memory copy? + BIO *read_from_ssl = NULL; + uint8_t *ssl_out_buf = NULL; + size_t ssl_out_buf_size = 0, ssl_out_buf_cap = 0; + bool ssl_handshake_done = false; + msgr_tls_record_hdr_t ssl_read_record; + size_t ssl_read_record_size = 0; +#endif + // Read state + bool io_error = false; int read_ready = 0; osd_op_t *read_op = NULL; size_t read_op_size = 0; @@ -88,7 +111,7 @@ struct osd_client_t iovec read_iov = { 0 }; msghdr read_msg = { 0 }; std::vector recv_list; - size_t recv_list_size = 0; + std::vector recv_flags; uint64_t read_op_id = 1; bool check_sequencing = false; bool enable_pg_locks = false; @@ -161,9 +184,19 @@ struct osd_messenger_t; struct rdmacm_connecting_t; #endif +class msgr_op_reader_t; +class msgr_op_writer_t; + struct __attribute__((visibility("default"))) osd_messenger_t { protected: + friend class copy_op_reader_t; + friend class ssl_op_reader_t; + friend class get_op_reader_t; + friend class copy_op_writer_t; + friend class ssl_op_writer_t; + friend class get_op_writer_t; + int keepalive_timer_id = -1; uint32_t receive_buffer_size = 0; @@ -177,6 +210,11 @@ protected: int iothread_count = 0; int max_aes_xts_pool_size = 256; + std::string tls_cert; + std::string tls_key; + std::string osd_tls_ca; + std::string client_tls_ca; + #ifdef WITH_RDMA bool use_rdma = true; bool use_rdmacm = false; @@ -193,6 +231,14 @@ protected: robin_hood::unordered_flat_map rdmacm_connecting; #endif +#ifdef WITH_OPENSSL + SSL_CTX *ssl_ctx = NULL; + std::string tls_cn; + + void ssl_init(osd_client_t *cl, bool server_mode); + bool ssl_do_handshake(osd_client_t *cl); +#endif + std::vector iothreads; std::vector read_ready_clients; std::vector write_ready_clients; @@ -274,29 +320,30 @@ protected: bool try_send(osd_client_t *cl); void handle_send(int result, bool prev, bool more, osd_client_t *cl); - size_t op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len); - void op_get_write_buffers(osd_client_t *cl, std::vector & lst); + bool op_write_to(osd_client_t *cl, msgr_op_writer_t & wr); + void next_write_op(osd_client_t *cl); + bool op_write_buf(osd_client_t *cl, uint8_t *src, size_t src_len, uint8_t *dst, size_t dst_len, bool skip_csum, size_t & from, size_t & done); + bool op_copy_data_to(osd_client_t *cl, uint8_t *dst, size_t dst_len, size_t & from, size_t & done); + size_t copy_ops_to(osd_client_t *cl, uint8_t *dst, size_t dst_len); void handle_read(int result, osd_client_t *cl); bool handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size_t bufsize); bool handle_hdr(osd_client_t *cl); bool allocate_op_buffers(osd_client_t *cl); bool allocate_reply_buffers(osd_client_t *cl, osd_op_t *op); - bool op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done); - void op_get_read_buffers(osd_client_t *cl, std::vector & lst); - void op_alloc_temp_buffers(osd_op_t *op, int i); + bool op_read_from(osd_client_t *cl, msgr_op_reader_t & rdr); bool handle_finished_op(osd_client_t *cl); void handle_immediate_ops(); - bool op_encrypted_copy_data_to(osd_client_t* cl, uint8_t *buf, size_t len, size_t from, size_t & done); - bool op_decrypted_copy_data_from(osd_client_t* cl, uint8_t *buf, size_t len, size_t from, size_t & done); + void op_encrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc); + void op_encrypt_free(osd_client_t* cl); + void op_decrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc); void op_decrypt_start(osd_client_t* cl); void op_decrypt_inline(osd_client_t* cl); void op_decrypt_free(osd_client_t* cl); #ifdef WITH_RDMA void try_send_rdma(osd_client_t *cl); - int try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len); bool init_recv_rdma(osd_client_t *cl); void handle_rdma_events(msgr_rdma_context_t *rdma_context); msgr_rdma_context_t* choose_rdma_context(osd_client_t *cl); diff --git a/src/client/msgr_encrypt.cpp b/src/client/msgr_encrypt.cpp index f6a02677..f9e9b8de 100644 --- a/src/client/msgr_encrypt.cpp +++ b/src/client/msgr_encrypt.cpp @@ -325,86 +325,49 @@ void destroy_aes_xts_decrypt(op_aes_xts_decrypt_t *decrypt_ctx) delete decrypt_ctx; } -bool osd_messenger_t::op_encrypted_copy_data_to(osd_client_t* cl, uint8_t *enc_buf, size_t enc_len, size_t from, size_t & done) +void osd_messenger_t::op_encrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc) { - auto op = cl->write_op; - auto & op_pos = cl->write_op_pos; - assert(op->req.hdr.opcode == OSD_OP_WRITE); - if (!from) + if (!cl->encrypt_ctx) { - if (!cl->encrypt_ctx) + if (encrypt_ctx_pool.size()) { - if (encrypt_ctx_pool.size()) - { - cl->encrypt_ctx = encrypt_ctx_pool.back(); - encrypt_ctx_pool.pop_back(); - } - else - cl->encrypt_ctx = new op_aes_xts_encrypt_t(); + cl->encrypt_ctx = encrypt_ctx_pool.back(); + encrypt_ctx_pool.pop_back(); } - assert(op->enc->key_chain[0]); - cl->encrypt_ctx->start(op->enc->key_chain[0], op->req.rw.offset, op->enc->bitmap_granularity); - } - for (int i = 0; i < op->iov.count; i++) - { - uint8_t *plain = (uint8_t*)op->iov.buf[i].iov_base; - size_t plain_len = op->iov.buf[i].iov_len; - while (from < plain_len || cl->encrypt_ctx->has_buffered()) - { - if (done >= enc_len) - return false; - size_t done_in = 0; - size_t done_out = 0; - cl->encrypt_ctx->update(plain+from, plain_len-from, enc_buf+done, enc_len-done, done_in, done_out); - if (cl->write_csum_state && done_out > 0) - XXH3_64bits_update(cl->write_csum_state, enc_buf+done, done_out); - done += done_out; - op_pos += done_in; - from += done_in; - } - from -= plain_len; - } - if (cl->encrypt_ctx) - { - if (encrypt_ctx_pool.size() > max_aes_xts_pool_size) - delete cl->encrypt_ctx; else - encrypt_ctx_pool.push_back(cl->encrypt_ctx); - cl->encrypt_ctx = NULL; + cl->encrypt_ctx = new op_aes_xts_encrypt_t(); + assert(cl->write_op->enc->key_chain[0]); + cl->encrypt_ctx->start(cl->write_op->enc->key_chain[0], cl->write_op->req.rw.offset, cl->write_op->enc->bitmap_granularity); + } + while (done_enc < enc_len && (done_plain < plain_len || cl->encrypt_ctx->has_buffered())) + { + size_t done_in = 0; + size_t done_out = 0; + cl->encrypt_ctx->update(plain+done_plain, plain_len-done_plain, enc_buf+done_enc, enc_len-done_enc, done_in, done_out); + if (cl->write_csum_state && done_out > 0) + XXH3_64bits_update(cl->write_csum_state, enc_buf+done_enc, done_out); + done_enc += done_out; + cl->write_op_pos += done_in; + done_plain += done_in; } - return true; } -bool osd_messenger_t::op_decrypted_copy_data_from(osd_client_t* cl, uint8_t *enc_buf, size_t enc_len, size_t from, size_t & done) +void osd_messenger_t::op_decrypted_copy_buf(osd_client_t *cl, uint8_t *enc_buf, size_t enc_len, uint8_t *plain, size_t plain_len, size_t & done_plain, size_t & done_enc) { op_decrypt_start(cl); - auto op = cl->read_op; - assert(op->req.hdr.opcode == OSD_OP_READ); - for (int i = 0; i < op->iov.count; i++) + while (done_plain < plain_len && done_enc < enc_len) { - uint8_t *plain = (uint8_t*)op->iov.buf[i].iov_base; - size_t plain_len = op->iov.buf[i].iov_len; - while (from < plain_len) - { - if (done >= enc_len) - return false; - size_t done_in = 0; - size_t done_out = 0; - // plain == NULL means skip output - cl->decrypt_ctx->update(enc_buf+done, enc_len-done, plain ? plain+from : NULL, plain_len-from, done_in, done_out); - if (cl->read_csum_state && done_in > 0) - XXH3_64bits_update(cl->read_csum_state, enc_buf+done, done_in); - done += done_in; - cl->read_op_pos += done_out; - cl->read_op_inline_decrypt_in += done_in; - from += done_out; - if (!done_out) - return false; - } - from -= plain_len; + size_t done_in = 0; + size_t done_out = 0; + // plain == NULL means skip output + cl->decrypt_ctx->update(enc_buf+done_enc, enc_len-done_enc, plain ? plain+done_plain : NULL, plain_len-done_plain, done_in, done_out); + if (cl->read_csum_state && done_in > 0) + XXH3_64bits_update(cl->read_csum_state, enc_buf+done_enc, done_in); + done_enc += done_in; + cl->read_op_pos += done_out; + cl->read_op_inline_decrypt_in += done_in; + done_plain += done_out; } - op_decrypt_free(cl); - return true; } void osd_messenger_t::op_decrypt_start(osd_client_t* cl) @@ -419,6 +382,7 @@ void osd_messenger_t::op_decrypt_start(osd_client_t* cl) else cl->decrypt_ctx = new op_aes_xts_decrypt_t(); auto & enc = cl->read_op->enc; + assert(cl->read_op->req.hdr.opcode == OSD_OP_READ); cl->decrypt_ctx->start(enc->key_chain, enc->chain_size, (cl->read_op->req.rw.flags & OSD_OP_RETURN_CHAIN) ? (uint8_t*)cl->read_op->bitmap + enc->read_chain_bitmap_pos : 0, cl->read_op->req.rw.offset, enc->bitmap_granularity); @@ -468,7 +432,6 @@ void osd_messenger_t::op_decrypt_inline(osd_client_t* cl) from_out += done_out; } assert(j >= op->iov.count); - op_decrypt_free(cl); } void osd_messenger_t::op_decrypt_free(osd_client_t* cl) @@ -482,3 +445,15 @@ void osd_messenger_t::op_decrypt_free(osd_client_t* cl) cl->decrypt_ctx = NULL; } } + +void osd_messenger_t::op_encrypt_free(osd_client_t* cl) +{ + if (cl->encrypt_ctx) + { + if (encrypt_ctx_pool.size() > max_aes_xts_pool_size) + delete cl->encrypt_ctx; + else + encrypt_ctx_pool.push_back(cl->encrypt_ctx); + cl->encrypt_ctx = NULL; + } +} diff --git a/src/client/msgr_op.h b/src/client/msgr_op.h index 9eb66932..73d38a69 100644 --- a/src/client/msgr_op.h +++ b/src/client/msgr_op.h @@ -183,7 +183,7 @@ struct __attribute__((visibility("default"))) osd_op_t void *bitmap = NULL; unsigned bitmap_len = 0; size_t bmp_data = 0; - void *bitmap_buf = NULL; + uint8_t *bitmap_buf = NULL; void *rmw_buf = NULL; std::shared_ptr enc; uint8_t *enc_buf = NULL; diff --git a/src/client/msgr_rdma.cpp b/src/client/msgr_rdma.cpp index 21f27a3c..494f9aa4 100644 --- a/src/client/msgr_rdma.cpp +++ b/src/client/msgr_rdma.cpp @@ -571,33 +571,6 @@ static void try_send_rdma_wr(osd_client_t *cl, ibv_sge *sge, int op_sge) cl->rdma_conn->cur_send++; } -int osd_messenger_t::try_send_rdma_copy(osd_client_t *cl, uint8_t *dst, int dst_len) -{ - int total_dst_len = dst_len; - while (dst_len > 0 && (cl->write_op || cl->write_ops.size())) - { - if (!cl->write_op) - { - cl->write_op = cl->write_ops.front(); - cl->write_ops.pop_front(); - } - osd_op_t *op = cl->write_op; - size_t copied = op_copy_to(cl, dst, dst_len); - if (!copied) - { - break; - } - dst += copied; - dst_len -= copied; - if (!cl->write_op && op->op_type == OSD_OP_IN) - { - // this is a reply, free the op after sending it - cl->send_free_ops.push_back(op); - } - } - return total_dst_len-dst_len; -} - void osd_messenger_t::try_send_rdma(osd_client_t *cl) { auto rc = cl->rdma_conn; @@ -625,7 +598,12 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl) : rc->send_done_pos-rc->send_out_pos); if (dst_len > rc->max_msg) dst_len = rc->max_msg; - copied = try_send_rdma_copy(cl, dst, dst_len); + copied = copy_ops_to(cl, dst, dst_len); + if (cl->io_error) + { + stop_client(cl->client_id); + return; + } if (copied > 0) { rc->send_out_pos += copied; diff --git a/src/client/msgr_rdma.h b/src/client/msgr_rdma.h index 9bf06668..a3f7e714 100644 --- a/src/client/msgr_rdma.h +++ b/src/client/msgr_rdma.h @@ -79,7 +79,7 @@ struct msgr_rdma_connection_t msgr_rdma_buf_t recv_buf; std::deque send_sizes; msgr_rdma_buf_t send_out; - int send_out_pos = 0, send_done_pos = 0, send_out_size = 0; + size_t send_out_pos = 0, send_done_pos = 0, send_out_size = 0; bool send_out_full = false; ~msgr_rdma_connection_t(); diff --git a/src/client/msgr_receive.cpp b/src/client/msgr_receive.cpp index 6b396db5..78eb3f0a 100644 --- a/src/client/msgr_receive.cpp +++ b/src/client/msgr_receive.cpp @@ -6,6 +6,342 @@ #include "messenger.h" #include "msgr_iothread.h" +#ifdef WITH_OPENSSL +#include +#include +#include +#include +#endif + +#define RDR_TLS 1 +#define RDR_XTS 2 +#define RDR_NO_CSUM 4 + +class msgr_op_reader_t +{ +public: + virtual bool read(uint8_t *dst, size_t dst_len, int flags = 0) = 0; +}; + +class copy_op_reader_t: public msgr_op_reader_t +{ +protected: + osd_messenger_t* msgr; + osd_client_t* cl; + size_t from; + + uint8_t *curbuf; + size_t bufsize; + size_t done; + +public: + copy_op_reader_t(osd_messenger_t* msgr, osd_client_t* cl, uint8_t *curbuf, size_t bufsize): + msgr(msgr), cl(cl), from(cl->read_op_pos), curbuf(curbuf), bufsize(bufsize), done(0) + {} + + void reset() + { + from = cl->read_op_pos; + } + + bool read(uint8_t *dst, size_t dst_len, int flags = 0) override + { + if (from >= dst_len) + { + from -= dst_len; + return true; + } + if (flags & RDR_XTS) + { + msgr->op_decrypted_copy_buf(cl, curbuf, bufsize, dst, dst_len, from, done); + } + else + { + size_t n = dst_len-from; + if (n > bufsize-done) + n = bufsize-done; + if (!n) + return false; + if (cl->read_csum_state && !(flags & RDR_NO_CSUM)) + { + // data may be skipped if dst == NULL but checksum is still calculated + XXH3_64bits_update(cl->read_csum_state, curbuf+done, n); + } + // Here, dst == NULL is allowed + if (dst != NULL) + memcpy(dst+from, curbuf+done, n); + done += n; + cl->read_op_pos += n; + from += n; + } + if (from < dst_len) + return false; + from = 0; + return true; + } + + size_t get_done() + { + return done; + } +}; + +class ssl_op_reader_t: public msgr_op_reader_t +{ + osd_messenger_t* msgr; + osd_client_t* cl; + size_t from; + + uint8_t *curbuf; + size_t bufsize; + size_t done; + +public: + ssl_op_reader_t(osd_messenger_t* msgr, osd_client_t* cl, uint8_t *curbuf, size_t bufsize): + msgr(msgr), cl(cl), from(cl->read_op_pos), curbuf(curbuf), bufsize(bufsize), done(0) + { + } + + void reset() + { + from = cl->read_op_pos; + } + + void buffer_encrypted() + { + while (done < bufsize) + { + if (cl->ssl_read_record_size < sizeof(msgr_tls_record_hdr_t)) + { + size_t n = bufsize-done; + if (n > sizeof(msgr_tls_record_hdr_t)-cl->ssl_read_record_size) + n = sizeof(msgr_tls_record_hdr_t)-cl->ssl_read_record_size; + memcpy(((uint8_t*)&cl->ssl_read_record) + cl->ssl_read_record_size, curbuf+done, n); + done += n; + cl->ssl_read_record_size += n; + if (done >= bufsize) + return; + } + if (cl->ssl_read_record.encrypted) + { + size_t n = cl->ssl_read_record.size; + if (n > bufsize-done) + n = bufsize-done; + // Buffer all encrypted data + // FIXME Limit the amount of buffered data + int r = BIO_write(cl->write_to_ssl, curbuf+done, n); + assert(r == n); + done += n; + cl->ssl_read_record.size -= n; + if (!cl->ssl_read_record.size) + cl->ssl_read_record_size = 0; + } + else + { + // Unencrypted data + break; + } + } + } + + bool read(uint8_t *dst, size_t dst_len, int flags) override + { + if (from >= dst_len) + { + // Skip + from -= dst_len; + return true; + } + size_t n = dst_len-from; + if (!(flags & RDR_TLS) || !cl->ssl_cli) + { + if (cl->ssl_cli && (cl->ssl_read_record_size < sizeof(msgr_tls_record_hdr_t) || + cl->ssl_read_record.size < n || cl->ssl_read_record.encrypted)) + { + fprintf(stderr, "Client %ju non-TLS data is too short, disconnecting\n", cl->client_id); + cl->io_error = true; + return false; + } + if (n > bufsize-done) + n = bufsize-done; + if (flags & RDR_XTS) + { + size_t prev = done; + msgr->op_decrypted_copy_buf(cl, curbuf, bufsize, dst, dst_len, from, done); + n = 0; + cl->ssl_read_record.size -= (done-prev); + } + else + { + if (cl->read_csum_state && !(flags & RDR_NO_CSUM)) + { + // data may be skipped if dst == NULL but checksum is still calculated + XXH3_64bits_update(cl->read_csum_state, curbuf+done, n); + } + // Here, dst == NULL is allowed + if (dst != NULL) + memcpy(dst+from, curbuf+done, n); + done += n; + cl->ssl_read_record.size -= n; + } + if (!cl->ssl_read_record.size) + cl->ssl_read_record_size = 0; + } + else + { + // Here, dst == NULL is not allowed + assert(dst != NULL); + buffer_encrypted(); + if (!cl->ssl_handshake_done) + { + if (!msgr->ssl_do_handshake(cl)) + return false; + if (cl->write_state == 0) + { + // SSL_ERROR_WANT_WRITE is absolutely non-informative with memory BIO, it basically never happens + // So we have to check memory BIO for outstanding data + char *bio_buf = NULL; + size_t bio_sz = BIO_get_mem_data(cl->read_from_ssl, &bio_buf); + if (bio_sz > 0) + { + cl->write_state = CL_WRITE_READY; + msgr->write_ready_clients.push_back(cl->client_id); + } + } + } + int ok = SSL_read_ex(cl->ssl_cli, dst+from, n, &n); + if (!ok) + { + ok = SSL_get_error(cl->ssl_cli, ok); + if (ok == SSL_ERROR_ZERO_RETURN) + { + fprintf(stderr, "Client %ju TLS disconnected\n", cl->client_id); + cl->io_error = true; + } + else if (ok != 0 && ok != SSL_ERROR_WANT_READ && ok != SSL_ERROR_WANT_WRITE) + { + fprintf(stderr, "Client %ju TLS read error: %s. Disconnecting client\n", cl->client_id, ERR_error_string(ERR_get_error(), NULL)); + cl->io_error = true; + } + return false; + } + if (cl->read_csum_state && !(flags & RDR_NO_CSUM)) + { + XXH3_64bits_update(cl->read_csum_state, dst+from, n); + } + } + cl->read_op_pos += n; + from += n; + if (from < dst_len) + { + return false; + } + from = 0; + return true; + } + + size_t get_done() + { + return done; + } +}; + +class get_op_reader_t: public msgr_op_reader_t +{ + osd_client_t* cl; + size_t from; + bool mpos; + +public: + get_op_reader_t(osd_messenger_t* msgr, osd_client_t* cl): + cl(cl), from(cl->read_op_pos), mpos(false) + { + if (cl->read_op->op_type == OSD_OP_OUT && + cl->read_op->reply.hdr.opcode == OSD_OP_READ && + cl->read_op->reply.hdr.retval > 0) + { + // When we recvmsg directly into the operation without copying, + // we need some place for all buffers, so we allocate temporary + // buffers for all skipped parts + alloc_temp_buffers(cl->read_op); + } + } + + void alloc_temp_buffers(osd_op_t *op) + { + size_t total_skip = 0; + for (int j = 0; j < op->iov.count; j++) + { + if (!op->iov.buf[j].iov_base) + { + total_skip += op->iov.buf[j].iov_len; + } + } + if (!total_skip) + { + return; + } + assert(!op->rmw_buf); + op->rmw_buf = malloc_or_die(total_skip); + total_skip = 0; + for (int j = 0; j < op->iov.count; j++) + { + if (!op->iov.buf[j].iov_base) + { + op->iov.buf[j].iov_base = (uint8_t*)op->rmw_buf + total_skip; + total_skip += op->iov.buf[j].iov_len; + } + } + } + + bool read(uint8_t *dst, size_t dst_len, int flags) override + { + if (from >= dst_len) + { + // Skip + from -= dst_len; + return true; + } + if ((flags & RDR_TLS) && cl->ssl_cli) + { + // Can't inplace read TLS data + return false; + } + if (cl->recv_list.size() >= IOV_MAX) + { + return false; + } + if ((flags & RDR_XTS) && cl->read_op->enc && !mpos) + { + mpos = true; + cl->read_op_inline_decrypt_pos = cl->read_op_pos; + cl->read_op_pos = cl->read_op_inline_decrypt_in + OSD_PACKET_SIZE + cl->read_op->reply.rw.bitmap_len; + from = cl->read_op_inline_decrypt_in; + } + if (cl->ssl_cli) + { + if (cl->ssl_read_record_size < sizeof(msgr_tls_record_hdr_t)) + { + return false; + } + if (cl->ssl_read_record.size < dst_len-from || cl->ssl_read_record.encrypted) + { + fprintf(stderr, "Client %ju non-TLS data is too short, disconnecting\n", cl->client_id); + cl->io_error = true; + return false; + } + cl->ssl_read_record.size -= (dst_len-from); + if (!cl->ssl_read_record.size) + cl->ssl_read_record_size = 0; + } + cl->recv_list.push_back((iovec){ dst+from, dst_len-from }); + cl->recv_flags.push_back(flags); + cl->read_op_pos += dst_len-from; + from = 0; + return true; + } +}; + void osd_messenger_t::read_requests() { for (int i = 0; i < read_ready_clients.size(); i++) @@ -18,9 +354,17 @@ void osd_messenger_t::read_requests() continue; } auto cl = cl_it->second; - if (cl->read_op && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size) + if (cl->read_op && cl->read_op_pos >= OSD_PACKET_SIZE && cl->read_op_size-(cl->read_op_pos-OSD_PACKET_SIZE) >= receive_buffer_size) { - op_get_read_buffers(cl, cl->recv_list); + get_op_reader_t rdr(this, cl); + if (!op_read_from(cl, rdr)) + { + if (cl->io_error) + { + stop_client(cl->client_id); + continue; + } + } } if (!cl->recv_list.size()) { @@ -38,7 +382,7 @@ void osd_messenger_t::read_requests() } assert(!cl->read_op || cl->read_op_pos < OSD_PACKET_SIZE || cl->read_op_size >= (cl->read_op_pos-OSD_PACKET_SIZE)); cl->refs++; - if (ringloop && !use_sync_send_recv) + if (!use_sync_send_recv) { auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL; io_uring_sqe sqe_local; @@ -66,7 +410,7 @@ void osd_messenger_t::read_requests() } else { - int result = recvmsg(cl->peer_fd, &cl->read_msg, 0); + int result = recvmsg(cl->peer_fd, &cl->read_msg, cl->recv_list.size() ? MSG_WAITALL : 0); if (result < 0) { result = -errno; @@ -126,7 +470,7 @@ out_wakeup: while (i < cl->recv_list.size() && result >= cl->recv_list[i].iov_len) { if (cl->read_csum_state && cl->recv_list[i].iov_len > 0 && - i != cl->recv_list.size()-1) // skip the checksum itself + !(cl->recv_flags[i] & RDR_NO_CSUM)) { XXH3_64bits_update(cl->read_csum_state, cl->recv_list[i].iov_base, cl->recv_list[i].iov_len); } @@ -143,10 +487,10 @@ out_wakeup: full_read = true; } cl->recv_list.erase(cl->recv_list.begin(), cl->recv_list.begin()+i); - if (!cl->recv_list.size()) + cl->recv_flags.erase(cl->recv_flags.begin(), cl->recv_flags.begin()+i); + if (!handle_finished_op(cl)) { - if (!handle_finished_op(cl)) - goto out_wakeup; + goto out_wakeup; } } } @@ -192,8 +536,8 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size cl->ping_time_remaining = 0; cl->idle_time_remaining = osd_idle_timeout; // Compose operation(s) from the buffer - size_t done = 0; - while (done < bufsize) + ssl_op_reader_t rdr(this, cl, curbuf, bufsize); + while (true) { if (!cl->read_op) { @@ -204,34 +548,25 @@ bool osd_messenger_t::handle_read_buffer(osd_client_t *cl, uint8_t *curbuf, size cl->read_op_size = 0; cl->read_op_inline_decrypt_in = 0; cl->read_op_inline_decrypt_pos = (size_t)-1; - if (cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD) - { - if (!cl->read_csum_state) - cl->read_csum_state = XXH3_createState(); - XXH3_64bits_reset(cl->read_csum_state); - } + rdr.reset(); } - if (cl->read_op_pos < OSD_PACKET_SIZE) + if (!cl->read_op_pos && (cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD)) { - int len = OSD_PACKET_SIZE - cl->read_op_pos; - if (len > bufsize-done) - len = bufsize-done; - memcpy(cl->read_op->req.buf + cl->read_op_pos, curbuf+done, len); - done += len; - cl->read_op_pos += len; - if (cl->read_op_pos < OSD_PACKET_SIZE) - return true; - if (!handle_hdr(cl)) + if (!cl->read_csum_state) + cl->read_csum_state = XXH3_createState(); + XXH3_64bits_reset(cl->read_csum_state); + } + if (!op_read_from(cl, rdr) || !handle_finished_op(cl)) + { + if (cl->io_error) { stop_client(cl->client_id); return false; } - } - if (!op_copy_from(cl, curbuf, bufsize, done)) - { - return false; + break; } } + assert(rdr.get_done() == bufsize); return true; } @@ -459,281 +794,118 @@ bool osd_messenger_t::allocate_reply_buffers(osd_client_t *cl, osd_op_t *op) return true; } -bool osd_messenger_t::op_copy_from(osd_client_t *cl, uint8_t *src, size_t src_len, size_t & done) +bool osd_messenger_t::op_read_from(osd_client_t *cl, msgr_op_reader_t & rdr) { osd_op_t *op = cl->read_op; - size_t from = cl->read_op_pos-OSD_PACKET_SIZE; - auto op_read_buf = [&](uint8_t *dst, size_t dst_len, bool skip_csum = false) + bool hdr = (cl->read_op_pos < OSD_PACKET_SIZE); + if (hdr || op->op_type == OSD_OP_IN) { - if (from < dst_len) - { - size_t n = dst_len-from; - if (n > src_len-done) - n = src_len-done; - if (cl->read_csum_state && !skip_csum) - { - // it may be skipped if !dst but checksum is still calculated - XXH3_64bits_update(cl->read_csum_state, src+done, n); - } - if (dst) - memcpy(dst+from, src+done, n); - else - assert(!this->osd_num); // NULL buffers are only used by clients - done += n; - cl->read_op_pos += n; - from += n; - if (from < dst_len) - return false; - from = 0; - } - else - from -= dst_len; - return true; - }; - if (op->op_type == OSD_OP_IN) - { - if (op->req.hdr.opcode == OSD_OP_SEC_WRITE || - op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) - { - if (!op_read_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len)) - return true; - if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len)) - return true; - } - else if (op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || - op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK) - { - if (!op_read_buf((uint8_t*)op->buf, op->req.sec_stab.len)) - return true; - } - else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) - { - if (!op_read_buf((uint8_t*)op->buf, op->req.sec_read_bmp.len)) - return true; - } - else if (op->req.hdr.opcode == OSD_OP_WRITE) - { - if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len)) - return true; - } - else if (op->req.hdr.opcode == OSD_OP_SHOW_CONFIG) - { - if (!op_read_buf((uint8_t*)op->buf, op->req.show_conf.json_len)) - return true; - } - } - else - { - if (op->reply.hdr.opcode == OSD_OP_SEC_READ) - { - if (op->reply.sec_rw.attr_len > 0) - { - if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len)) - return true; - } - if (op->reply.hdr.retval > 0) - { - for (int i = 0; i < op->iov.count; i++) - if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) - return true; - } - } - else if (op->reply.hdr.opcode == OSD_OP_READ) - { - if (op->reply.rw.bitmap_len > 0) - { - if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len)) - return true; - } - if (op->reply.hdr.retval > 0) - { - if (op->enc) - { - if (!op_decrypted_copy_data_from(cl, src, src_len, from, done)) - return true; - } - else - { - for (int i = 0; i < op->iov.count; i++) - if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) - return true; - } - } - } - else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0) - { - if (!op_read_buf((uint8_t*)op->buf, sizeof(obj_ver_id) * op->reply.hdr.retval)) - return true; - } - else if ((op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP || - op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG) && op->reply.hdr.retval > 0) - { - if (!op_read_buf((uint8_t*)op->buf, op->reply.hdr.retval)) - return true; - } - else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0) - { - if (!op_read_buf((uint8_t*)op->buf, op->reply.describe.result_bytes)) - return true; - } - } - if (cl->proto_csum_status == MSGR_CSUM_FULL || - cl->read_op_size > 0 && cl->proto_csum_status == MSGR_CSUM_PAYLOAD) - { - if (!op_read_buf((uint8_t*)&op->csum, 8, true)) - return true; - } - return handle_finished_op(cl); -} - -void osd_messenger_t::op_get_read_buffers(osd_client_t *cl, std::vector & lst) -{ - osd_op_t *op = cl->read_op; - size_t from = cl->read_op_pos-OSD_PACKET_SIZE; - size_t done = 0; - auto op_read_buf = [&](uint8_t *dst, size_t dst_len) - { - if (lst.size() >= IOV_MAX) + if (!rdr.read(op->req.buf, OSD_PACKET_SIZE, RDR_TLS | (cl->proto_csum_status == MSGR_CSUM_PAYLOAD ? RDR_NO_CSUM : 0))) return false; - if (from < dst_len) + if (hdr) { - lst.push_back((iovec){ .iov_base = dst+from, .iov_len = dst_len-from }); - cl->read_op_pos += dst_len-from; - done += dst_len-from; - from = 0; + if (!handle_hdr(cl)) + return false; + op = cl->read_op; + if (op->op_type == OSD_OP_OUT) + goto switched_type; } - else - from -= dst_len; - return true; - }; - if (op->op_type == OSD_OP_IN) - { if (op->req.hdr.opcode == OSD_OP_SEC_WRITE || op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) { - if (!op_read_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len)) - return; - if (!op_read_buf((uint8_t*)op->buf, op->req.sec_rw.len)) - return; + if (!rdr.read((uint8_t*)op->bitmap, op->req.sec_rw.attr_len, RDR_TLS)) + return false; + if (!rdr.read((uint8_t*)op->buf, op->req.sec_rw.len, 0)) + return false; } else if (op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK) { - if (!op_read_buf((uint8_t*)op->buf, op->req.sec_stab.len)) - return; + if (!rdr.read((uint8_t*)op->buf, op->req.sec_stab.len, RDR_TLS)) + return false; } else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) { - if (!op_read_buf((uint8_t*)op->buf, op->req.sec_read_bmp.len)) - return; + if (!rdr.read((uint8_t*)op->buf, op->req.sec_read_bmp.len, RDR_TLS)) + return false; } else if (op->req.hdr.opcode == OSD_OP_WRITE) { - if (!op_read_buf((uint8_t*)op->buf, op->req.rw.len)) - return; + if (!rdr.read((uint8_t*)op->buf, op->req.rw.len, 0)) + return false; } else if (op->req.hdr.opcode == OSD_OP_SHOW_CONFIG) { - if (!op_read_buf((uint8_t*)op->buf, op->req.show_conf.json_len)) - return; + if (!rdr.read((uint8_t*)op->buf, op->req.show_conf.json_len, RDR_TLS)) + return false; } } else { + if (!rdr.read(op->reply.buf, OSD_PACKET_SIZE, RDR_TLS | (cl->proto_csum_status == MSGR_CSUM_PAYLOAD ? RDR_NO_CSUM : 0))) + return false; +switched_type: if (op->reply.hdr.opcode == OSD_OP_SEC_READ) { if (op->reply.sec_rw.attr_len > 0) { - if (!op_read_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len)) - return; + if (!rdr.read((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len, RDR_TLS)) + return false; } if (op->reply.hdr.retval > 0) { for (int i = 0; i < op->iov.count; i++) - if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) - return; + if (!rdr.read((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len, 0)) + return false; } } else if (op->reply.hdr.opcode == OSD_OP_READ) { if (op->reply.rw.bitmap_len > 0) { - if (!op_read_buf((uint8_t*)op->bitmap, op->reply.rw.bitmap_len)) - return; + if (!rdr.read((uint8_t*)op->bitmap, op->reply.rw.bitmap_len, RDR_TLS)) + return false; } if (op->reply.hdr.retval > 0) { - if (op->enc) - { - cl->read_op_inline_decrypt_pos = cl->read_op_pos; - cl->read_op_pos = cl->read_op_inline_decrypt_in + OSD_PACKET_SIZE + op->reply.rw.bitmap_len; - from = cl->read_op_inline_decrypt_in; - } for (int i = 0; i < op->iov.count; i++) - { - if (!op->iov.buf[i].iov_base) - { - // When we recvmsg directly into the operation without copying, - // we need some place for all buffers, so we allocate temporary - // buffers for all skipped parts - op_alloc_temp_buffers(op, i); - } - if (!op_read_buf((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len)) - return; - } + if (!rdr.read((uint8_t*)op->iov.buf[i].iov_base, op->iov.buf[i].iov_len, (op->enc ? RDR_XTS : 0))) + return false; } } else if (op->reply.hdr.opcode == OSD_OP_SEC_LIST && op->reply.hdr.retval > 0) { - if (!op_read_buf((uint8_t*)op->buf, sizeof(obj_ver_id) * op->reply.hdr.retval)) - return; + if (!rdr.read((uint8_t*)op->buf, sizeof(obj_ver_id) * op->reply.hdr.retval, RDR_TLS)) + return false; } else if ((op->reply.hdr.opcode == OSD_OP_SEC_READ_BMP || op->reply.hdr.opcode == OSD_OP_SHOW_CONFIG) && op->reply.hdr.retval > 0) { - if (!op_read_buf((uint8_t*)op->buf, op->reply.hdr.retval)) - return; + if (!rdr.read((uint8_t*)op->buf, op->reply.hdr.retval, RDR_TLS)) + return false; } else if (op->reply.hdr.opcode == OSD_OP_DESCRIBE && op->reply.describe.result_bytes > 0) { - if (!op_read_buf((uint8_t*)op->buf, op->reply.describe.result_bytes)) - return; + if (!rdr.read((uint8_t*)op->buf, op->reply.describe.result_bytes, RDR_TLS)) + return false; } } if (cl->proto_csum_status == MSGR_CSUM_FULL || cl->read_op_size > 0 && cl->proto_csum_status == MSGR_CSUM_PAYLOAD) { - if (!op_read_buf((uint8_t*)&op->csum, 8)) - return; - } -} - -void osd_messenger_t::op_alloc_temp_buffers(osd_op_t *op, int i) -{ - size_t total_skip = 0; - for (int j = i; j < op->iov.count; j++) - { - if (!op->iov.buf[j].iov_base) - { - total_skip += op->iov.buf[j].iov_len; - } - } - assert(total_skip); - assert(!op->rmw_buf); - op->rmw_buf = malloc_or_die(total_skip); - total_skip = 0; - for (int j = i; j < op->iov.count; j++) - { - if (!op->iov.buf[j].iov_base) - { - op->iov.buf[j].iov_base = (uint8_t*)op->rmw_buf + total_skip; - total_skip += op->iov.buf[j].iov_len; - } + if (!rdr.read((uint8_t*)&op->csum, 8, RDR_TLS|RDR_NO_CSUM)) + return false; } + assert(cl->read_op_pos == cl->read_op_size+OSD_PACKET_SIZE); + return true; } bool osd_messenger_t::handle_finished_op(osd_client_t *cl) { + if (cl->read_op_pos < cl->read_op_size+OSD_PACKET_SIZE) + { + return true; + } osd_op_t *op = cl->read_op; if (cl->proto_csum_status == MSGR_CSUM_FULL || cl->read_op_size > 0 && cl->proto_csum_status == MSGR_CSUM_PAYLOAD) @@ -743,7 +915,7 @@ bool osd_messenger_t::handle_finished_op(osd_client_t *cl) { fprintf(stderr, "Client %ju checksum mismatch for received data: expected %016jx, got %016jx, disconnecting client\n", cl->client_id, op->csum, real_csum); - stop_client(cl->client_id); + cl->io_error = true; return false; } } @@ -774,6 +946,7 @@ bool osd_messenger_t::handle_finished_op(osd_client_t *cl) (tv_end.tv_nsec - op->tv_begin.tv_nsec)/1000 ); } + op_decrypt_free(cl); set_immediate_ops.push_back(op); cl->read_op = NULL; return true; diff --git a/src/client/msgr_send.cpp b/src/client/msgr_send.cpp index c6b26601..b149bb50 100644 --- a/src/client/msgr_send.cpp +++ b/src/client/msgr_send.cpp @@ -8,6 +8,362 @@ #include "messenger.h" #include "msgr_iothread.h" +#ifdef WITH_OPENSSL +#include +#include +#include +#include +#endif + +#define WR_TLS 1 +#define WR_XTS 2 +#define WR_NO_CSUM 4 + +class msgr_op_writer_t +{ +public: + virtual bool write(uint8_t *src, size_t src_len, int flags = 0) = 0; +}; + +class copy_op_writer_t: public msgr_op_writer_t +{ +protected: + osd_messenger_t* msgr; + osd_client_t* cl; + size_t from; + + uint8_t *curbuf; + size_t bufsize; + size_t done; + +public: + copy_op_writer_t(osd_messenger_t* msgr, osd_client_t* cl, uint8_t *curbuf, size_t bufsize): + msgr(msgr), cl(cl), from(cl->write_op_pos), curbuf(curbuf), bufsize(bufsize), done(0) + {} + + void reset() + { + from = cl->write_op_pos; + } + + bool write(uint8_t *src, size_t src_len, int flags = 0) override + { + if (from >= src_len) + { + from -= src_len; + return true; + } + if (flags & WR_XTS) + { + msgr->op_encrypted_copy_buf(cl, curbuf, bufsize, src, src_len, from, done); + } + else + { + size_t n = src_len-from; + if (n > bufsize-done) + n = bufsize-done; + if (cl->write_csum_state && !(flags & WR_NO_CSUM)) + XXH3_64bits_update(cl->write_csum_state, src+from, n); + memcpy(curbuf+done, src+from, n); + done += n; + cl->write_op_pos += n; + from += n; + } + if (from < src_len) + return false; + from = 0; + return true; + } + + size_t get_done() + { + return done; + } +}; + +class ssl_op_writer_t: public msgr_op_writer_t +{ + osd_messenger_t* msgr; + osd_client_t* cl; + size_t from; + + uint8_t *curbuf; + size_t bufsize; + size_t done; + +public: + ssl_op_writer_t(osd_messenger_t* msgr, osd_client_t* cl, uint8_t *curbuf, size_t bufsize): + msgr(msgr), cl(cl), from(cl->write_op_pos), curbuf(curbuf), bufsize(bufsize), done(0) + { + } + + void reset() + { + from = cl->write_op_pos; + } + + static inline size_t ssl_copy_from_bio(BIO *bio, uint8_t *buf, size_t size) + { + if (size < sizeof(msgr_tls_record_hdr_t)) + return 0; + int r = BIO_read(bio, buf+sizeof(msgr_tls_record_hdr_t), size-sizeof(msgr_tls_record_hdr_t)); + if (r > 0) + { + msgr_tls_record_hdr_t *hdr = (msgr_tls_record_hdr_t*)buf; + hdr->encrypted = 1; + hdr->size = r; + return r+sizeof(msgr_tls_record_hdr_t); + } + return 0; + } + + void flush_ssl() + { + if (!cl->ssl_handshake_done) + { + if (!msgr->ssl_do_handshake(cl)) + return; + } + done += ssl_copy_from_bio(cl->read_from_ssl, curbuf+done, bufsize-done); + } + + static inline bool write_to_ssl(osd_client_t *cl, uint8_t *src, size_t src_len, int flags, size_t & from) + { + size_t n = src_len-from; + int ok = SSL_write_ex(cl->ssl_cli, src+from, n, &n); + if (ok) + { + if (cl->write_csum_state && !(flags & WR_NO_CSUM)) + XXH3_64bits_update(cl->write_csum_state, src+from, n); + cl->write_op_pos += n; + from += n; + } + else + { + ok = SSL_get_error(cl->ssl_cli, ok); + if (ok == SSL_ERROR_ZERO_RETURN) + { + fprintf(stderr, "Client %ju TLS disconnected\n", cl->client_id); + cl->io_error = true; + return false; + } + else if (ok != SSL_ERROR_WANT_READ && ok != SSL_ERROR_WANT_WRITE) + { + fprintf(stderr, "Client %ju TLS write error: %s. Disconnecting client\n", cl->client_id, ERR_error_string(ERR_get_error(), NULL)); + cl->io_error = true; + return false; + } + } + return true; + } + + bool write(uint8_t *src, size_t src_len, int flags) override + { + if (from >= src_len) + { + from -= src_len; + return true; + } + if (!(flags & WR_TLS) || !cl->ssl_cli) + { + if (cl->ssl_cli && from == 0 && done < bufsize-sizeof(msgr_tls_record_hdr_t)-1) + { + msgr_tls_record_hdr_t *hdr = (msgr_tls_record_hdr_t*)(curbuf+done); + hdr->encrypted = 0; + hdr->size = src_len; + done += sizeof(msgr_tls_record_hdr_t); + } + if (flags & WR_XTS) + { + msgr->op_encrypted_copy_buf(cl, curbuf, bufsize, src, src_len, from, done); + } + else + { + size_t n = src_len-from; + if (n > bufsize-done) + n = bufsize-done; + if (cl->write_csum_state && !(flags & WR_NO_CSUM)) + XXH3_64bits_update(cl->write_csum_state, src+from, n); + memcpy(curbuf+done, src+from, n); + done += n; + cl->write_op_pos += n; + from += n; + } + } + else + { + if (!cl->ssl_handshake_done) + { + if (!msgr->ssl_do_handshake(cl)) + return false; + } + if (cl->ssl_handshake_done) + { + if (!write_to_ssl(cl, src, src_len, flags, from)) + return false; + } + done += ssl_copy_from_bio(cl->read_from_ssl, curbuf+done, bufsize-done); + } + if (from < src_len) + return false; + from = 0; + return true; + } + + size_t get_done() + { + return done; + } +}; + +class get_op_writer_t: public msgr_op_writer_t +{ + osd_messenger_t* msgr; + osd_client_t* cl; + size_t from; + size_t enc_size; + size_t done_enc; + + void ssl_extend_buf() + { + size_t min_cap = cl->ssl_out_buf_size*2; + if (min_cap < 16384) + min_cap = 16384; + if (cl->ssl_out_buf_cap < min_cap) + { + uint8_t *old_buf = cl->ssl_out_buf; + uint8_t *old_end = old_buf + cl->ssl_out_buf_cap; + cl->ssl_out_buf = (uint8_t*)realloc_or_die(cl->ssl_out_buf, min_cap); + cl->ssl_out_buf_cap = min_cap; + for (auto & iov: cl->send_list) + { + if (iov.iov_base >= old_buf && iov.iov_base < old_end) + iov.iov_base = cl->ssl_out_buf + ((uint8_t*)iov.iov_base - old_buf); + } + } + } + + void copy_ssl() + { + size_t prev_size = cl->ssl_out_buf_size; + do + { + ssl_extend_buf(); + cl->ssl_out_buf_size += ssl_op_writer_t::ssl_copy_from_bio(cl->read_from_ssl, + cl->ssl_out_buf+cl->ssl_out_buf_size, cl->ssl_out_buf_cap-cl->ssl_out_buf_size); + } while (cl->ssl_out_buf_size >= cl->ssl_out_buf_cap); + if (cl->ssl_out_buf_size > prev_size) + { + cl->send_list.push_back((iovec){ .iov_base = cl->ssl_out_buf+prev_size, .iov_len = cl->ssl_out_buf_size-prev_size }); + } + } + +public: + get_op_writer_t(osd_messenger_t* msgr, osd_client_t* cl): + msgr(msgr), cl(cl), from(cl->write_op_pos), enc_size(0), done_enc(0) + { + } + + void reset() + { + from = cl->write_op_pos; + enc_size = 0; + done_enc = 0; + } + + void flush_ssl() + { + if (!cl->ssl_handshake_done) + { + if (!msgr->ssl_do_handshake(cl)) + return; + } + if (cl->send_list.size() >= IOV_MAX) + { + return; + } + copy_ssl(); + } + + bool write(uint8_t *src, size_t src_len, int flags) override + { + if (from >= src_len) + { + // Skip + from -= src_len; + return true; + } + if (cl->send_list.size() >= IOV_MAX) + { + return false; + } + if (cl->ssl_cli) + { + if (flags & WR_TLS) + { + if (!cl->ssl_handshake_done) + { + if (!msgr->ssl_do_handshake(cl)) + return false; + } + if (cl->ssl_handshake_done) + { + if (!ssl_op_writer_t::write_to_ssl(cl, src, src_len, flags, from)) + return false; + } + // Copy data to client's temporary SSL output buffer + copy_ssl(); + if (from < src_len) + return false; + from = 0; + return true; + } + else if (!from) + { + if (cl->send_list.size() >= IOV_MAX-1) + { + return false; + } + ssl_extend_buf(); + msgr_tls_record_hdr_t *hdr = (msgr_tls_record_hdr_t*)(cl->ssl_out_buf+cl->ssl_out_buf_size); + hdr->encrypted = 0; + hdr->size = src_len; + cl->send_list.push_back((iovec){ .iov_base = cl->ssl_out_buf+cl->ssl_out_buf_size, .iov_len = sizeof(msgr_tls_record_hdr_t) }); + cl->ssl_out_buf_size += sizeof(msgr_tls_record_hdr_t); + } + } + if (flags & WR_XTS) + { + if (!cl->write_op->enc_buf) + { + if (cl->send_list.size() >= IOV_MAX-1) + { + // Make sure that 1 encrypted buffer and 1 checksum fits + return false; + } + // No way except than to allocate a temporary buffer and encrypt data to it + assert(cl->write_op->req.hdr.opcode == OSD_OP_WRITE); + enc_size = cl->write_op->req.rw.len - from + (from % 16); + assert(enc_size > 0); + cl->write_op->enc_buf = (uint8_t*)malloc_or_die(enc_size); + cl->send_list.push_back((iovec){ .iov_base = cl->write_op->enc_buf, .iov_len = enc_size }); + } + assert(enc_size > 0); + msgr->op_encrypted_copy_buf(cl, cl->write_op->enc_buf, enc_size, src, src_len, from, done_enc); + assert(from == src_len); + } + else + { + if (cl->write_csum_state && !(flags & WR_NO_CSUM)) + XXH3_64bits_update(cl->write_csum_state, src+from, src_len-from); + cl->send_list.push_back((iovec){ src+from, src_len-from }); + cl->write_op_pos += src_len-from; + } + from = 0; + return true; + } +}; + void osd_messenger_t::outbox_push(osd_op_t *cur_op) { assert(cur_op->client_id); @@ -59,37 +415,83 @@ void osd_messenger_t::outbox_push(osd_op_t *cur_op) } else { - if ((cl->write_msg.msg_iovlen > 0 || !try_send(cl)) && (cl->write_state == 0)) + if (!try_send(cl) && cl->write_state == 0) { cl->write_state = CL_WRITE_READY; - write_ready_clients.push_back(cur_op->client_id); + write_ready_clients.push_back(cl->client_id); } ringloop->wakeup(); } } -bool osd_messenger_t::try_send(osd_client_t *cl) +bool osd_messenger_t::ssl_do_handshake(osd_client_t *cl) { - if (!cl->write_op && !cl->write_ops.size() || cl->write_msg.msg_iovlen > 0 || cl->peer_state == PEER_STOPPED || cl->peer_fd < 0) + if (cl->ssl_handshake_done) { return true; } + int r = SSL_do_handshake(cl->ssl_cli); + if (r > 0) + { + cl->ssl_handshake_done = true; + } + else + { + r = SSL_get_error(cl->ssl_cli, r); + if (r != 0 && r != SSL_ERROR_WANT_READ && r != SSL_ERROR_WANT_WRITE) + { + fprintf(stderr, "Client %ju TLS handshake error: %s, stopping client\n", cl->client_id, ERR_error_string(ERR_get_error(), NULL)); + cl->io_error = true; + return false; + } + } + return true; +} + +bool osd_messenger_t::try_send(osd_client_t *cl) +{ + if (cl->peer_state == PEER_STOPPED || cl->peer_fd < 0) + { + return true; + } + if (cl->write_msg.msg_iovlen > 0 || !ringloop->space_left() && !use_sync_send_recv) + { + return false; + } assert(cl->peer_state != PEER_RDMA); + get_op_writer_t wr(this, cl); while ((cl->write_op || cl->write_ops.size()) && cl->send_list.size() < IOV_MAX) { if (!cl->write_op) { - cl->write_op = cl->write_ops.front(); - cl->write_ops.pop_front(); + next_write_op(cl); + wr.reset(); } osd_op_t *op = cl->write_op; - op_get_write_buffers(cl, cl->send_list); + if (!op_write_to(cl, wr)) + { + if (cl->io_error) + { + stop_client(cl->client_id); + return true; + } + break; + } if (!cl->write_op && op->op_type == OSD_OP_IN) { cl->send_free_ops.push_back(op); } } - if (ringloop && !use_sync_send_recv) + if (!cl->send_list.size() && cl->ssl_cli) + { + wr.flush_ssl(); + } + if (!cl->send_list.size()) + { + cl->write_state = 0; + return true; + } + if (!use_sync_send_recv) { auto iothread = iothreads.size() ? iothreads[cl->peer_fd % iothreads.size()] : NULL; io_uring_sqe sqe_local; @@ -100,10 +502,7 @@ bool osd_messenger_t::try_send(osd_client_t *cl) sqe_local = { .user_data = (uint64_t)&data_local }; data_local = {}; } - if (!sqe) - { - return false; - } + assert(sqe); cl->send_list_size = 0; for (auto & iov: cl->send_list) { @@ -149,6 +548,48 @@ bool osd_messenger_t::try_send(osd_client_t *cl) return true; } +size_t osd_messenger_t::copy_ops_to(osd_client_t *cl, uint8_t *dst, size_t dst_len) +{ + ssl_op_writer_t wr(this, cl, dst, dst_len); + while ((cl->write_op || cl->write_ops.size()) && wr.get_done() < dst_len) + { + if (!cl->write_op) + { + next_write_op(cl); + wr.reset(); + } + osd_op_t *op = cl->write_op; + if (!op_write_to(cl, wr)) + { + if (cl->io_error) + return 0; + break; + } + if (!cl->write_op && op->op_type == OSD_OP_IN) + { + // this is a reply, free the op after sending it + cl->send_free_ops.push_back(op); + } + } + if (!wr.get_done() && cl->ssl_cli) + { + wr.flush_ssl(); + } + return wr.get_done(); +} + +void osd_messenger_t::next_write_op(osd_client_t *cl) +{ + cl->write_op = cl->write_ops.front(); + cl->write_ops.pop_front(); + if (cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD) + { + if (!cl->write_csum_state) + cl->write_csum_state = XXH3_createState(); + XXH3_64bits_reset(cl->write_csum_state); + } +} + void osd_messenger_t::send_replies() { for (int i = 0; i < write_ready_clients.size(); i++) @@ -197,7 +638,12 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t // Second notification - only free a batch of postponed ops int i = 0; for (; i < cl->zc_free_list.size() && cl->zc_free_list[i]; i++) - delete cl->zc_free_list[i]; + { + if (!((size_t)cl->zc_free_list[i] & 7)) + delete cl->zc_free_list[i]; + else + free((void*)((size_t)cl->zc_free_list[i] & ~(size_t)7)); + } if (i > 0) cl->zc_free_list.erase(cl->zc_free_list.begin(), cl->zc_free_list.begin()+i+1); return; @@ -212,14 +658,35 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t for (auto op: cl->send_free_ops) { if (more) + { + assert(!((size_t)op & 7)); cl->zc_free_list.push_back(op); + } else delete op; } if (more) + { + if (cl->ssl_out_buf_size) + { + cl->zc_free_list.push_back((osd_op_t*)((size_t)cl->ssl_out_buf | 1)); + cl->ssl_out_buf = NULL; + cl->ssl_out_buf_cap = 0; + } cl->zc_free_list.push_back(NULL); // end marker + } + cl->ssl_out_buf_size = 0; cl->send_free_ops.clear(); - cl->write_state = cl->write_op || cl->write_ops.size() ? CL_WRITE_READY : 0; + cl->write_state = 0; + if (cl->write_op || cl->write_ops.size()) + cl->write_state = CL_WRITE_READY; + else if (cl->ssl_cli) + { + char *bio_buf = NULL; + size_t bio_sz = BIO_get_mem_data(cl->read_from_ssl, &bio_buf); + if (bio_sz > 0) + cl->write_state = CL_WRITE_READY; + } if ((cl->proto_csum_status & MSGR_CSUM_NEG) && !cl->write_op && !cl->write_ops.size()) { // Checksums negotiated, enable @@ -245,192 +712,97 @@ void osd_messenger_t::handle_send(int result, bool prev, bool more, osd_client_t } } -static inline bool op_write_headers(osd_op_t *op, std::function op_write_buf, bool skip_hdr_csum) -{ - if (!op_write_buf((op->op_type == OSD_OP_IN ? op->reply.buf : op->req.buf), OSD_PACKET_SIZE, skip_hdr_csum)) - { - return false; - } - // Bitmap - if (op->op_type == OSD_OP_IN && - op->req.hdr.opcode == OSD_OP_SEC_READ && - op->reply.sec_rw.attr_len > 0) - { - if (!op_write_buf((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len, false)) - return false; - } - else if (op->op_type == OSD_OP_OUT && - (op->req.hdr.opcode == OSD_OP_SEC_WRITE || op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) && - op->req.sec_rw.attr_len > 0) - { - if (!op_write_buf((uint8_t*)op->bitmap, op->req.sec_rw.attr_len, false)) - return false; - } - if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP) - { - if (op->op_type == OSD_OP_IN && op->reply.hdr.retval > 0) - { - if (!op_write_buf((uint8_t*)op->buf, (size_t)op->reply.hdr.retval, false)) - return false; - } - else if (op->op_type == OSD_OP_OUT && op->req.sec_read_bmp.len > 0) - { - if (!op_write_buf((uint8_t*)op->buf, (size_t)op->req.sec_read_bmp.len, false)) - return false; - } - } - return true; -} - -static inline bool op_has_data(osd_op_t *op) +static inline bool op_has_data_for_ssl(osd_op_t *op) { return (op->op_type == OSD_OP_IN - ? (op->req.hdr.opcode == OSD_OP_READ || - op->req.hdr.opcode == OSD_OP_SEC_READ || - op->req.hdr.opcode == OSD_OP_SEC_LIST || + ? (op->req.hdr.opcode == OSD_OP_SEC_LIST || op->req.hdr.opcode == OSD_OP_SHOW_CONFIG || op->req.hdr.opcode == OSD_OP_DESCRIBE) - : (op->req.hdr.opcode == OSD_OP_WRITE || - op->req.hdr.opcode == OSD_OP_SEC_WRITE || - op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE || - op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || + : (op->req.hdr.opcode == OSD_OP_SEC_STABILIZE || op->req.hdr.opcode == OSD_OP_SEC_ROLLBACK || op->req.hdr.opcode == OSD_OP_SHOW_CONFIG)) && op->iov.count > 0; } -size_t osd_messenger_t::op_copy_to(osd_client_t *cl, uint8_t *dst, size_t dst_len) +static inline bool op_has_data_for_nonssl(osd_op_t *op) { - size_t done = 0; - size_t from = cl->write_op_pos; - auto op_write_buf = [&](uint8_t *src, size_t src_len, bool skip_csum) - { - if (from < src_len) - { - size_t n = src_len-from; - if (n > dst_len-done) - n = dst_len-done; - if (cl->write_csum_state && !skip_csum) - XXH3_64bits_update(cl->write_csum_state, src+from, n); - memcpy(dst+done, src+from, n); - done += n; - cl->write_op_pos += n; - from += n; - if (from < src_len) - return false; - from = 0; - } - else - from -= src_len; - return true; - }; - if ((cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD) && !from) - { - if (!cl->write_csum_state) - cl->write_csum_state = XXH3_createState(); - XXH3_64bits_reset(cl->write_csum_state); - } - // Header - if (!op_write_headers(cl->write_op, op_write_buf, cl->proto_csum_status != MSGR_CSUM_FULL)) - { - return done; - } - // Operation data - if (op_has_data(cl->write_op)) - { - if (cl->write_op->enc) - { - if (!op_encrypted_copy_data_to(cl, dst, dst_len, from, done)) - { - return done; - } - } - else - { - for (int i = 0; i < cl->write_op->iov.count; i++) - { - if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len, false)) - return done; - } - } - } - if (cl->proto_csum_status == MSGR_CSUM_FULL || - cl->proto_csum_status == MSGR_CSUM_PAYLOAD && cl->write_op_pos > OSD_PACKET_SIZE) - { - if (!from) - cl->write_op->csum = XXH3_64bits_digest(cl->write_csum_state); - if (!op_write_buf((uint8_t*)&cl->write_op->csum, 8, true)) - return done; - } - cl->write_op = NULL; - cl->write_op_pos = 0; - return done; + return (op->op_type == OSD_OP_IN + ? (op->req.hdr.opcode == OSD_OP_READ || + op->req.hdr.opcode == OSD_OP_SEC_READ) + : (op->req.hdr.opcode == OSD_OP_WRITE || + op->req.hdr.opcode == OSD_OP_SEC_WRITE || + op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE)) && op->iov.count > 0; } -void osd_messenger_t::op_get_write_buffers(osd_client_t *cl, std::vector & lst) +bool osd_messenger_t::op_write_to(osd_client_t *cl, msgr_op_writer_t & wr) { - size_t from = cl->write_op_pos; - auto op_write_buf = [&](uint8_t *src, size_t src_len, bool skip_csum) - { - if (lst.size() >= IOV_MAX) - return false; - if (from < src_len) - { - if (cl->write_csum_state && !skip_csum) - XXH3_64bits_update(cl->write_csum_state, src+from, src_len-from); - lst.push_back((iovec){ .iov_base = src+from, .iov_len = src_len-from }); - cl->write_op_pos += src_len-from; - from = 0; - } - else - from -= src_len; - return true; - }; - if ((cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD) && !from) - { - if (!cl->write_csum_state) - cl->write_csum_state = XXH3_createState(); - XXH3_64bits_reset(cl->write_csum_state); - } + osd_op_t *op = cl->write_op; // Header - if (!op_write_headers(cl->write_op, op_write_buf, cl->proto_csum_status != MSGR_CSUM_FULL)) + if (!wr.write((op->op_type == OSD_OP_IN ? op->reply.buf : op->req.buf), OSD_PACKET_SIZE, + WR_TLS | (cl->proto_csum_status == MSGR_CSUM_PAYLOAD ? WR_NO_CSUM : 0))) { - return; + return false; + } + // Bitmap + if (op->op_type == OSD_OP_IN) + { + if (op->req.hdr.opcode == OSD_OP_SEC_READ && op->reply.sec_rw.attr_len > 0) + { + if (!wr.write((uint8_t*)op->bitmap, op->reply.sec_rw.attr_len, WR_TLS)) + return false; + } + else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP && op->reply.hdr.retval > 0) + { + if (!wr.write((uint8_t*)op->buf, (size_t)op->reply.hdr.retval, WR_TLS)) + return false; + } + else if (op->req.hdr.opcode == OSD_OP_READ && op->reply.rw.bitmap_len > 0) + { + if (!wr.write((uint8_t*)op->bitmap, op->reply.rw.bitmap_len, WR_TLS)) + return false; + } + } + else if (op->op_type == OSD_OP_OUT) + { + if ((op->req.hdr.opcode == OSD_OP_SEC_WRITE || op->req.hdr.opcode == OSD_OP_SEC_WRITE_STABLE) && + op->req.sec_rw.attr_len > 0) + { + if (!wr.write((uint8_t*)op->bitmap, op->req.sec_rw.attr_len, WR_TLS)) + return false; + } + else if (op->req.hdr.opcode == OSD_OP_SEC_READ_BMP && op->req.sec_read_bmp.len > 0) + { + if (!wr.write((uint8_t*)op->buf, (size_t)op->req.sec_read_bmp.len, WR_TLS)) + return false; + } } // Operation data - if (op_has_data(cl->write_op)) + if (op_has_data_for_ssl(op)) { - if (cl->write_op->enc) + for (int i = 0; i < cl->write_op->iov.count; i++) { - if (lst.size() >= IOV_MAX) - return; - // No way except to allocate a temporary buffer and encrypt data to it - assert(cl->write_op->req.hdr.opcode == OSD_OP_WRITE); - size_t remsize = cl->write_op->req.rw.len - from + (from % 16); - assert(remsize > 0); - assert(!cl->write_op->enc_buf); - cl->write_op->enc_buf = (uint8_t*)malloc_or_die(remsize); - size_t done = 0; - bool end = op_encrypted_copy_data_to(cl, cl->write_op->enc_buf, remsize, from, done); - assert(end); - lst.push_back((iovec){ .iov_base = cl->write_op->enc_buf, .iov_len = remsize }); + auto & iov = cl->write_op->iov.buf[i]; + if (!wr.write((uint8_t*)iov.iov_base, iov.iov_len, WR_TLS)) + return false; } - else + } + else if (op_has_data_for_nonssl(op)) + { + for (int i = 0; i < cl->write_op->iov.count; i++) { - for (int i = 0; i < cl->write_op->iov.count; i++) - { - if (!op_write_buf((uint8_t*)cl->write_op->iov.buf[i].iov_base, cl->write_op->iov.buf[i].iov_len, false)) - return; - } + auto & iov = cl->write_op->iov.buf[i]; + if (!wr.write((uint8_t*)iov.iov_base, iov.iov_len, (op->enc ? WR_XTS : 0))) + return false; } } if (cl->proto_csum_status == MSGR_CSUM_FULL || cl->proto_csum_status == MSGR_CSUM_PAYLOAD && cl->write_op_pos > OSD_PACKET_SIZE) { - if (!from) - cl->write_op->csum = XXH3_64bits_digest(cl->write_csum_state); - if (!op_write_buf((uint8_t*)&cl->write_op->csum, 8, true)) - return; + cl->write_op->csum = XXH3_64bits_digest(cl->write_csum_state); + if (!wr.write((uint8_t*)&cl->write_op->csum, 8, WR_TLS|WR_NO_CSUM)) + return false; } + op_encrypt_free(cl); cl->write_op = NULL; cl->write_op_pos = 0; + return true; } diff --git a/src/client/msgr_stop.cpp b/src/client/msgr_stop.cpp index 753f8c06..46ee6e1d 100644 --- a/src/client/msgr_stop.cpp +++ b/src/client/msgr_stop.cpp @@ -6,6 +6,12 @@ #include "messenger.h" #include "../util/xxh_x86dispatch.h" +#ifdef WITH_OPENSSL +#include +#include +#include +#include +#endif void osd_client_t::cancel_ops() { @@ -207,7 +213,10 @@ osd_client_t::~osd_client_t() { if (op) { - delete op; + if (!((size_t)op & 7)) + delete op; + else + free((void*)((size_t)op & ~(size_t)7)); } } if (read_csum_state) @@ -220,4 +229,18 @@ osd_client_t::~osd_client_t() XXH3_freeState(write_csum_state); write_csum_state = NULL; } +#ifdef WITH_OPENSSL + if (ssl_cli) + { + SSL_free(ssl_cli); + ssl_cli = NULL; + write_to_ssl = NULL; + read_from_ssl = NULL; + } + if (ssl_out_buf) + { + free(ssl_out_buf); + ssl_out_buf = NULL; + } +#endif } diff --git a/src/osd/osd_primary.cpp b/src/osd/osd_primary.cpp index 65bf29da..f3b9edf2 100644 --- a/src/osd/osd_primary.cpp +++ b/src/osd/osd_primary.cpp @@ -130,9 +130,7 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op) stripe_count * clean_entry_bitmap_size + // - 'missing' flags for chained reads (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 0 : pg_it->second.pg_size) - ) + - // read chain info - chain_info_len + ) ); void *data_buf = (uint8_t*)op_data + sizeof(osd_primary_op_data_t); op_data->pg_num = pg_num; @@ -147,10 +145,17 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op) split_stripes(pg_data_size, bs_block_size, (uint32_t)(cur_op->req.rw.offset - oid.stripe), cur_op->req.rw.len, op_data->stripes); // Resulting bitmaps have to survive op_data and be freed with the op itself assert(!cur_op->bitmap_buf); - cur_op->bitmap_buf = calloc_or_die(1, clean_entry_bitmap_size * stripe_count); + cur_op->bitmap_buf = (uint8_t*)calloc_or_die(1, clean_entry_bitmap_size*stripe_count + chain_info_len); + uint8_t *buf = cur_op->bitmap_buf; for (int i = 0; i < stripe_count; i++) { - op_data->stripes[i].bmp_buf = (uint8_t*)cur_op->bitmap_buf + clean_entry_bitmap_size * i; + op_data->stripes[i].bmp_buf = buf; + buf += clean_entry_bitmap_size; + if (i == pg_data_size-1 && chain_info_len) + { + op_data->chain_info = buf; + buf += chain_info_len; + } } } op_data->chain_size = chain_size; @@ -164,11 +169,6 @@ bool osd_t::prepare_primary_rw(osd_op_t *cur_op) data_buf = (uint8_t*)data_buf + chain_size * stripe_count * clean_entry_bitmap_size; op_data->missing_flags = (uint8_t*)data_buf; data_buf = (uint8_t*)data_buf + chain_size * (pool_cfg.scheme == POOL_SCHEME_REPLICATED ? 0 : pg_it->second.pg_size); - if (chain_info_len) - { - op_data->chain_info = (uint8_t*)data_buf; - data_buf = (uint8_t*)data_buf + chain_info_len; - } // Copy chain int chain_num = 0; op_data->read_chain[chain_num] = cur_op->req.rw.inode; @@ -325,6 +325,7 @@ resume_2: } cur_op->reply.rw.version = op_data->fact_ver; cur_op->reply.rw.bitmap_len = (pg ? pg->pg_data_size : 1) * clean_entry_bitmap_size; + cur_op->bitmap = op_data->stripes[0].bmp_buf; if (op_data->degraded) { // Reconstruct missing stripes @@ -337,7 +338,6 @@ resume_2: { reconstruct_stripes_ec(stripes, pg->pg_size, pg->pg_data_size, clean_entry_bitmap_size); } - cur_op->iov.push_back(op_data->stripes[0].bmp_buf, cur_op->reply.rw.bitmap_len); for (int role = 0; role < pg->pg_size; role++) { if (stripes[role].req_end != 0) @@ -352,7 +352,6 @@ resume_2: } else { - cur_op->iov.push_back(op_data->stripes[0].bmp_buf, cur_op->reply.rw.bitmap_len); cur_op->iov.push_back(cur_op->buf, cur_op->req.rw.len); } finish_op(cur_op, cur_op->req.rw.len); diff --git a/src/osd/osd_primary_chain.cpp b/src/osd/osd_primary_chain.cpp index c2ca977a..a30eaf62 100644 --- a/src/osd/osd_primary_chain.cpp +++ b/src/osd/osd_primary_chain.cpp @@ -584,11 +584,10 @@ void osd_t::send_chained_read_results(pg_t *pg, osd_op_t *cur_op) } } // Send bitmap + cur_op->bitmap = op_data->stripes[0].bmp_buf; cur_op->reply.rw.bitmap_len = (pg ? pg->pg_data_size : 1) * clean_entry_bitmap_size; - cur_op->iov.push_back(op_data->stripes[0].bmp_buf, cur_op->reply.rw.bitmap_len); if (cur_op->req.rw.flags & OSD_OP_RETURN_CHAIN) { - cur_op->iov.push_back(op_data->chain_info, (cur_op->req.rw.len / bs_bitmap_granularity)); cur_op->reply.rw.bitmap_len += (cur_op->req.rw.len / bs_bitmap_granularity); } // And finally compose the result diff --git a/src/osd/osd_scrub.cpp b/src/osd/osd_scrub.cpp index 6f898de6..34d8af1b 100644 --- a/src/osd/osd_scrub.cpp +++ b/src/osd/osd_scrub.cpp @@ -415,7 +415,7 @@ void osd_t::submit_scrub_subops(osd_op_t *cur_op) } } assert(!cur_op->bitmap_buf); - cur_op->bitmap_buf = calloc_or_die(1, clean_entry_bitmap_size * op_data->stripe_count); + cur_op->bitmap_buf = (uint8_t*)calloc_or_die(1, clean_entry_bitmap_size * op_data->stripe_count); for (int i = 0; i < op_data->stripe_count; i++) { op_data->stripes[i].bmp_buf = (uint8_t*)cur_op->bitmap_buf + clean_entry_bitmap_size * i; diff --git a/tests/common.sh b/tests/common.sh index 3abec920..0ffba510 100644 --- a/tests/common.sh +++ b/tests/common.sh @@ -27,6 +27,7 @@ ETCD_COUNT=${ETCD_COUNT:-1} ANTIETCD=${ANTIETCD} USE_RAMDISK=${USE_RAMDISK} ETCD_SCHEME=${ETCD_SCHEME:-http} +OSD_TLS=${OSD_TLS} RAMDISK=/run/user/$(id -u) findmnt $RAMDISK >/dev/null || (sudo mkdir -p $RAMDISK && sudo mount -t tmpfs tmpfs $RAMDISK) @@ -124,6 +125,23 @@ VITASTOR_CFG='"etcd_address":"'$ETCD_URL'"'"$VITASTOR_CFG" if [[ "$ETCD_SCHEME" = "https" ]]; then VITASTOR_CFG="$VITASTOR_CFG"',"etcd_ca":"'$(pwd)'/testdata/etcd.crt"' fi +if [[ "$OSD_TLS" = "1" ]]; then + cd ./testdata + openssl req -days 3650 -x509 -new -newkey rsa:4096 -nodes -keyout client_ca.key -out client_ca.crt \ + -subj '/C=RU/ST=Russia/L=Moscow/O=VitastorClientCA' + openssl req -days 3650 -x509 -new -newkey rsa:4096 -nodes -keyout osd.key -out osd.crt \ + -subj '/C=RU/ST=Russia/L=Moscow/O=VitastorOSD' -addext "extendedKeyUsage = serverAuth, clientAuth" + openssl req -subj '/CN=test' -nodes -new -keyout cli.key -out cli.csr -addext "extendedKeyUsage = clientAuth" + openssl x509 -req -days 3650 -CA client_ca.crt -CAkey client_ca.key -CAcreateserial -in cli.csr -out cli.crt + rm cli.csr + cd $(dirname $0)/.. + VITASTOR_CFG="$VITASTOR_CFG"',"osd_tls_cert":"'$(pwd)'/testdata/osd.crt"' + VITASTOR_CFG="$VITASTOR_CFG"',"osd_tls_key":"'$(pwd)'/testdata/osd.key"' + VITASTOR_CFG="$VITASTOR_CFG"',"osd_tls_ca":"'$(pwd)'/testdata/osd.crt"' + VITASTOR_CFG="$VITASTOR_CFG"',"client_tls_ca":"'$(pwd)'/testdata/client_ca.crt"' + VITASTOR_CFG="$VITASTOR_CFG"',"tls_cert":"'$(pwd)'/testdata/cli.crt"' + VITASTOR_CFG="$VITASTOR_CFG"',"tls_key":"'$(pwd)'/testdata/cli.key"' +fi echo "{$VITASTOR_CFG}" > ./testdata/vitastor.conf VITASTOR_CFG=./testdata/vitastor.conf VITASTOR_CLI="build/src/cmd/vitastor-cli --config_path $VITASTOR_CFG"