thebestvibedev/tromcho.net
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement OSD TLS support

Browse Source
This commit is contained in:
Vitaliy Filippov
2026-04-22 01:41:29 +03:00
parent a9c12d9858
commit a21e4f9503
15 changed files with 1267 additions and 554 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/client/http_client.cpp
+27 -14
View File
@@ -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);
src/client/http_client.h
+12
View File
@@ -8,6 +8,10 @@
#include <functional>
#include "json11/json11.hpp"
#ifdef WITH_OPENSSL
#include <openssl/types.h>
#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
src/client/messenger.cpp
+105 -1
View File
@@ -14,6 +14,13 @@
#ifdef WITH_RDMA
#include "msgr_rdma.h"
#endif
#include "http_client.h"
#ifdef WITH_OPENSSL
#include <openssl/bio.h>
#include <openssl/err.h>
#include <openssl/pem.h>
#include <openssl/ssl.h>
#endif
#include <sys/poll.h>
@@ -117,6 +124,50 @@ void msgr_iothread_t::run()
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)
{
@@ -303,6 +354,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)
@@ -348,6 +406,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
@@ -356,7 +427,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();
@@ -578,6 +649,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);
}
@@ -819,6 +894,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)
{
@@ -833,6 +912,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)
{
src/client/messenger.h
+56 -9
View File
@@ -12,6 +12,10 @@
#include <deque>
#include <vector>
#ifdef WITH_OPENSSL
#include <openssl/types.h>
#endif
#include "../util/xxh_x86dispatch.h"
#include "../util/robin_hood.h"
#include "malloc_or_die.h"
@@ -56,6 +60,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;
@@ -78,7 +88,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;
@@ -86,7 +109,7 @@ struct osd_client_t
iovec read_iov = { 0 };
msghdr read_msg = { 0 };
std::vector<iovec> recv_list;
size_t recv_list_size = 0;
std::vector<int> recv_flags;
uint64_t read_op_id = 1;
bool check_sequencing = false;
bool enable_pg_locks = false;
@@ -195,9 +218,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;
@@ -211,6 +244,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;
@@ -227,6 +265,14 @@ protected:
robin_hood::unordered_flat_map<rdma_cm_id*, rdmacm_connecting_t*> 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<msgr_iothread_t*> iothreads;
std::vector<uint64_t> read_ready_clients;
std::vector<uint64_t> write_ready_clients;
@@ -303,29 +349,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<iovec> & 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<iovec> & 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);
src/client/msgr_encrypt.cpp
+44 -69
View File
@@ -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;
}
}
src/client/msgr_op.h
+1 -1
View File
@@ -182,7 +182,7 @@ struct __attribute__((visibility("default"))) osd_op_t
void *bitmap = NULL;
unsigned bitmap_len = 0;
unsigned bmp_data = 0;
void *bitmap_buf = NULL;
uint8_t *bitmap_buf = NULL;
void *rmw_buf = NULL;
std::shared_ptr<osd_op_enc_t> enc;
uint8_t *enc_buf = NULL;
src/client/msgr_rdma.cpp
+6 -28
View File
@@ -568,33 +568,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;
@@ -620,7 +593,12 @@ void osd_messenger_t::try_send_rdma(osd_client_t *cl)
dst_len = (rc->send_out_pos < rc->send_out_size ? rc->send_out_size-rc->send_out_pos : 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;
src/client/msgr_rdma.h
+1 -1
View File
@@ -79,7 +79,7 @@ struct msgr_rdma_connection_t
msgr_rdma_buf_t recv_buf;
std::deque<uint64_t> 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();
src/client/msgr_receive.cpp
+414 -241
View File
@@ -5,6 +5,342 @@
#include <limits.h>
#include "messenger.h"
#ifdef WITH_OPENSSL
#include <openssl/bio.h>
#include <openssl/err.h>
#include <openssl/pem.h>
#include <openssl/ssl.h>
#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++)
@@ -17,9 +353,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())
{
@@ -37,7 +381,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;
@@ -65,7 +409,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;
@@ -125,7 +469,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);
}
@@ -142,10 +486,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;
}
}
}
@@ -191,8 +535,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)
{
@@ -203,34 +547,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;
}
@@ -404,281 +739,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<iovec> & 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)
@@ -688,7 +860,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;
}
}
@@ -719,6 +891,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;
src/client/msgr_send.cpp
+546 -174
View File
@@ -7,6 +7,362 @@
#include "messenger.h"
#ifdef WITH_OPENSSL
#include <openssl/bio.h>
#include <openssl/err.h>
#include <openssl/pem.h>
#include <openssl/ssl.h>
#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);
@@ -58,10 +414,10 @@ 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();
}
@@ -116,28 +472,74 @@ void osd_messenger_t::measure_exec(osd_op_t *cur_op)
}
}
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;
@@ -148,10 +550,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)
{
@@ -197,6 +596,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++)
@@ -245,7 +686,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;
@@ -260,14 +706,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
@@ -293,192 +760,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<bool(uint8_t*, size_t, bool)> 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<iovec> & 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;
}
src/client/msgr_stop.cpp
+24 -1
View File
@@ -9,6 +9,12 @@
#ifdef WITH_RDMA
#include "msgr_rdma.h"
#endif
#ifdef WITH_OPENSSL
#include <openssl/bio.h>
#include <openssl/err.h>
#include <openssl/pem.h>
#include <openssl/ssl.h>
#endif
void osd_client_t::cancel_ops()
{
@@ -217,7 +223,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));
}
}
#ifndef __MOCK__
@@ -239,4 +248,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
}
src/osd/osd_primary.cpp
+11 -12
View File
@@ -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;
@@ -319,6 +319,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
@@ -331,7 +332,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)
@@ -346,7 +346,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);
src/osd/osd_primary_chain.cpp
+1 -2
View File
@@ -582,11 +582,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
src/osd/osd_scrub.cpp
+1 -1
View File
@@ -413,7 +413,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;
tests/common.sh
+18
View File
@@ -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"