Wsrep TLS service

This commit defines a TLS service interface. If the implementation is
provided by the application when the provider is loaded, appropriate
hooks are probed from the provider and the provider side hooks are
initialized after the provider is loaded.

A sample implementation to demostrate the use of TLS interface
is provided in dbsim/db_tls.cpp.

Also contains a change to thread service interface: The
thread exit virtual method was changed to function pointer
to allow thread exit path which does not involve C++.

dbsim/CMakeLists.txt

@@ -13,6 +13,7 @@ add_executable(dbsim
   db_simulator.cpp
   db_storage_engine.cpp
   db_threads.cpp
+  db_tls.cpp
   dbsim.cpp
 )
 

dbsim/db_params.cpp

@@ -87,6 +87,11 @@ db::params db::parse_args(int argc, char** argv)
          po::value<bool>(&params.cond_checks),
          "Enable checks for correct condition variable use. "
          " Effective only if thread-instrumentation is enabled")
+        ("tls-service",
+         po::value<int>(&params.tls_service),
+         "Configure TLS service stubs.\n0 default disabled\n1 enabled\n"
+         "2 enabled with short read/write and renegotiation simulation\n"
+         "3 enabled with error simulation.")
         ;
     try
     {

dbsim/db_params.hpp

@@ -42,6 +42,7 @@ namespace db
         int fast_exit;
         int thread_instrumentation;
         bool cond_checks;
+        int tls_service;
         params()
             : n_servers(0)
             , n_clients(0)
@@ -58,6 +59,7 @@ namespace db
             , fast_exit(0)
             , thread_instrumentation()
             , cond_checks()
+            , tls_service()
         { }
     };
 

dbsim/db_simulator.cpp

@@ -20,6 +20,7 @@
 #include "db_simulator.hpp"
 #include "db_client.hpp"
 #include "db_threads.hpp"
+#include "db_tls.hpp"
 
 #include "wsrep/logger.hpp"
 
@@ -27,6 +28,7 @@
 #include <sstream>
 
 static db::ti thread_instrumentation;
+static db::tls tls_service;
 
 void db::simulator::run()
 {
@@ -36,6 +38,7 @@ void db::simulator::run()
     std::cout << "Results:\n";
     std::cout << stats() << std::endl;
     std::cout << db::ti::stats() << std::endl;
+    std::cout << db::tls::stats() << std::endl;
 }
 
 void db::simulator::sst(db::server& server,
@@ -114,6 +117,7 @@ void db::simulator::start()
 {
     thread_instrumentation.level(params_.thread_instrumentation);
     thread_instrumentation.cond_checks(params_.cond_checks);
+    tls_service.init(params_.tls_service);
     wsrep::log_info() << "Provider: " << params_.wsrep_provider;
 
     std::string cluster_address(build_cluster_address());
@@ -149,6 +153,9 @@ void db::simulator::start()
         services.thread_service = params_.thread_instrumentation
                                       ? &thread_instrumentation
                                       : nullptr;
+        services.tls_service = params_.tls_service
+            ? &tls_service
+            : nullptr;
         if (server.server_state().load_provider(params_.wsrep_provider,
                                                 server_options, services))
         {

dbsim/db_threads.cpp

@@ -35,6 +35,7 @@
 #include <unordered_map>
 #include <vector>
 
+extern "C" { static void* start_thread(void* args_ptr); }
 namespace
 {
     struct ti_obj
@@ -186,7 +187,6 @@ namespace
         }
     };
 
-    void* thread_start_fn(void* args_ptr);
 
     class ti_thread : public ti_obj
     {
@@ -209,7 +209,7 @@ namespace
         int run(void* (*fn)(void *), void* args)
         {
             auto ta(new thread_args{this, fn, args});
-            return pthread_create(&th_, nullptr, thread_start_fn, ta);
+            return pthread_create(&th_, nullptr, start_thread, ta);
         }
 
         int detach()
@@ -254,24 +254,6 @@ namespace
         bool detached_;
     };
 
-    void* thread_start_fn(void* args_ptr)
-    {
-        thread_args* ta(reinterpret_cast<thread_args*>(args_ptr));
-        ti_thread* thread = reinterpret_cast<ti_thread*>(ta->this_thread);
-        pthread_setspecific(this_thread_key, thread);
-        void* (*fn)(void*) = ta->fn;
-        void* args = ta->args;
-        delete ta;
-        void* ret((*fn)(args));
-        pthread_setspecific(this_thread_key, nullptr);
-
-        // If we end here the thread returned instead of calling
-        // pthread_exit()
-        if (thread->detached())
-            delete thread;
-        return ret;
-    }
-
     class ti_mutex : public ti_obj
     {
     public:
@@ -482,6 +464,42 @@ int db::ti::after_init()
 //                               Thread                                     //
 //////////////////////////////////////////////////////////////////////////////
 
+extern "C"
+{
+static void* start_thread(void* args_ptr)
+{
+    thread_args* ta(reinterpret_cast<thread_args*>(args_ptr));
+    ti_thread* thread = reinterpret_cast<ti_thread*>(ta->this_thread);
+    pthread_setspecific(this_thread_key, thread);
+    void* (*fn)(void*) = ta->fn;
+    void* args = ta->args;
+    delete ta;
+    void* ret = (*fn)(args);
+    pthread_setspecific(this_thread_key, nullptr);
+    // If we end here the thread returned instead of calling
+    // pthread_exit()
+    if (thread->detached())
+        delete thread;
+    return ret;
+}
+
+WSREP_NORETURN
+static void exit_thread(wsrep::thread_service::thread* thread, void* retval)
+{
+    pthread_setspecific(this_thread_key, nullptr);
+    ti_thread* th(reinterpret_cast<ti_thread*>(thread));
+    th->retval(retval);
+    if (th->detached())
+        delete th;
+    pthread_exit(retval);
+}
+} // extern "C"
+
+db::ti::ti()
+{
+    thread_service::exit = exit_thread;
+}
+
 const wsrep::thread_service::thread_key*
 db::ti::create_thread_key(const char* name) WSREP_NOEXCEPT
 {
@@ -513,15 +531,6 @@ int db::ti::detach(wsrep::thread_service::thread* thread) WSREP_NOEXCEPT
     return reinterpret_cast<ti_thread*>(thread)->detach();
 }
 
-void db::ti::exit(wsrep::thread_service::thread* thread, void* retval) WSREP_NOEXCEPT
-{
-    ti_thread* th(reinterpret_cast<ti_thread*>(thread));
-    th->retval(retval);
-    if (th->detached())
-        delete th;
-    pthread_exit(retval);
-}
-
 int db::ti::equal(wsrep::thread_service::thread* thread_1,
                    wsrep::thread_service::thread* thread_2) WSREP_NOEXCEPT
 {

dbsim/db_threads.hpp

@@ -28,7 +28,7 @@ namespace db
     class ti : public wsrep::thread_service
     {
     public:
-        ti() { }
+        ti();
         int before_init() override;
         int after_init() override;
 
@@ -41,7 +41,6 @@ namespace db
         int detach(wsrep::thread_service::thread*) WSREP_NOEXCEPT override;
         int equal(wsrep::thread_service::thread*,
                   wsrep::thread_service::thread*) WSREP_NOEXCEPT override;
-        void exit(wsrep::thread_service::thread*, void*) WSREP_NOEXCEPT override;
         int join(wsrep::thread_service::thread*, void**) WSREP_NOEXCEPT override;
         wsrep::thread_service::thread* self() WSREP_NOEXCEPT override;
         int setschedparam(wsrep::thread_service::thread*, int,

dbsim/db_tls.cpp

@@ -0,0 +1,451 @@
+/*
+ * Copyright (C) 2020 Codership Oy <info@codership.com>
+ *
+ * This file is part of wsrep-lib.
+ *
+ * Wsrep-lib is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * Wsrep-lib is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+/** @file db_tls.cpp
+ *
+ * This file demonstrates the use of TLS service. It does not implement
+ * real encryption, but may manipulate stream bytes for testing purposes.
+ */
+
+#include "db_tls.hpp"
+
+#include "wsrep/logger.hpp"
+
+#include <unistd.h> // read()
+#include <fcntl.h>
+#include <sys/types.h>
+#include <sys/socket.h> // send()
+#include <cassert>
+#include <cerrno>
+#include <cstring>
+
+#include <mutex>
+#include <string>
+
+namespace
+{
+    class db_stream : public wsrep::tls_stream
+    {
+    public:
+        db_stream(int fd, int mode)
+            : fd_(fd)
+            , state_(s_initialized)
+            , last_error_()
+            , mode_(mode)
+            , stats_()
+            , is_blocking_()
+        {
+            int val(fcntl(fd_, F_GETFL, 0));
+            is_blocking_ = not (val & O_NONBLOCK);
+        }
+        struct stats
+        {
+            size_t bytes_read{0};
+            size_t bytes_written{0};
+        };
+
+        /*
+         *    in            idle --|
+         *     |-> ch -|      ^    | -> want_read  --|
+         *     |-> sh -| ---- |--> | -> want_write --|
+         *                    |----------------------|
+         */
+        enum state
+        {
+            s_initialized,
+            s_client_handshake,
+            s_server_handshake,
+            s_idle,
+            s_want_read,
+            s_want_write
+        };
+
+        int get_error_number() const { return last_error_; }
+        const void* get_error_category() const
+        {
+            return reinterpret_cast<const void*>(1);
+        }
+
+        static char* get_error_message(int value, const void*)
+        {
+            return ::strerror(value);
+        }
+
+        enum wsrep::tls_service::status client_handshake();
+
+        enum wsrep::tls_service::status server_handshake();
+
+        wsrep::tls_service::op_result read(void*, size_t);
+
+        wsrep::tls_service::op_result write(const void*, size_t);
+
+        enum state state() const { return state_; }
+
+        int fd() const { return fd_; }
+        void inc_reads(size_t val) { stats_.bytes_read += val; }
+        void inc_writes(size_t val) { stats_.bytes_written += val; }
+        const stats& get_stats() const { return stats_; }
+    private:
+        enum wsrep::tls_service::status handle_handshake_read(const char* expect);
+        size_t determine_read_count(size_t max_count)
+        {
+            if (is_blocking_ || mode_ < 2) return max_count;
+            else if (::rand() % 100 == 0) return std::min(size_t(42), max_count);
+            else return max_count;
+        }
+        size_t determine_write_count(size_t count)
+        {
+            if (is_blocking_ || mode_ < 2) return count;
+            else if (::rand() % 100 == 0) return std::min(size_t(43), count);
+            else return count;
+        }
+
+        ssize_t do_read(void* buf, size_t max_count)
+        {
+            if (is_blocking_ || mode_ < 3 )
+                return ::read(fd_, buf, max_count);
+            else if (::rand() % 1000 == 0) return EINTR;
+            else return ::read(fd_, buf, max_count);
+        }
+
+        ssize_t do_write(const void* buf, size_t count)
+        {
+            if (is_blocking_ || mode_ < 3)
+                return ::send(fd_, buf, count, MSG_NOSIGNAL);
+            else if (::rand() % 1000 == 0) return EINTR;
+            else return ::send(fd_, buf, count, MSG_NOSIGNAL);
+        }
+
+        wsrep::tls_service::op_result map_success(ssize_t result)
+        {
+            if (is_blocking_ || mode_ < 2)
+            {
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::success, size_t(result)};
+            }
+            else if (::rand() % 1000 == 0)
+            {
+                wsrep::log_info() << "Success want extra read";
+                state_ = s_want_read;
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::want_read, size_t(result)};
+            }
+            else if (::rand() % 1000 == 0)
+            {
+                wsrep::log_info() << "Success want extra write";
+                state_ = s_want_write;
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::want_write, size_t(result)};
+            }
+            else
+            {
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::success, size_t(result)};
+            }
+        }
+
+        wsrep::tls_service::op_result map_result(ssize_t result)
+        {
+            if (result > 0)
+            {
+                return map_success(result);
+            }
+            else if (result == 0)
+            {
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::eof, 0};
+            }
+            else if (errno == EAGAIN || errno == EWOULDBLOCK)
+            {
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::want_read, 0};
+            }
+            else
+            {
+                last_error_ = errno;
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::error, 0};
+            }
+        }
+
+        void clear_error() { last_error_ = 0; }
+
+        int fd_;
+        enum state state_;
+        int last_error_;
+        // Operation mode:
+        // 1 - simulate handshake exchange
+        // 2 - simulate errors and short reads
+        int mode_;
+        stats stats_;
+        bool is_blocking_;
+    };
+
+    enum wsrep::tls_service::status db_stream::client_handshake()
+    {
+        clear_error();
+        enum wsrep::tls_service::status ret;
+        assert(state_ == s_initialized ||
+               state_ == s_client_handshake ||
+               state_ == s_want_write);
+        if (state_ == s_initialized)
+        {
+            (void)::send(fd_, "clie", 4, MSG_NOSIGNAL);
+            ret = wsrep::tls_service::want_read;
+            state_ = s_client_handshake;
+            wsrep::log_info() << this << " client handshake sent";
+            stats_.bytes_written += 4;
+            if (not is_blocking_) return ret;
+        }
+
+        if (state_ == s_client_handshake)
+        {
+            if ((ret = handle_handshake_read("serv")) ==
+                wsrep::tls_service::success)
+            {
+                state_ = s_want_write;
+                ret = wsrep::tls_service::want_write;
+            }
+            if (not is_blocking_) return ret;
+        }
+
+        if (state_ == s_want_write)
+        {
+            state_ = s_idle;
+            ret = wsrep::tls_service::success;
+            if (not is_blocking_) return ret;
+        }
+
+        if (not is_blocking_)
+        {
+            last_error_ = EPROTO;
+            ret = wsrep::tls_service::error;
+        }
+        return ret;
+    }
+
+
+
+    enum wsrep::tls_service::status db_stream::server_handshake()
+    {
+        enum wsrep::tls_service::status ret;
+        assert(state_ == s_initialized ||
+               state_ == s_server_handshake ||
+               state_ == s_want_write);
+
+        if (state_ == s_initialized)
+        {
+            ::send(fd_, "serv", 4, MSG_NOSIGNAL);
+            ret = wsrep::tls_service::want_read;
+            state_ = s_server_handshake;
+            stats_.bytes_written += 4;
+            if (not is_blocking_) return ret;
+        }
+
+        if (state_ == s_server_handshake)
+        {
+            if ((ret = handle_handshake_read("clie")) ==
+                wsrep::tls_service::success)
+            {
+                state_ = s_want_write;
+                ret = wsrep::tls_service::want_write;
+            }
+            if (not is_blocking_) return ret;
+        }
+
+        if (state_ == s_want_write)
+        {
+            state_ = s_idle;
+            ret = wsrep::tls_service::success;
+            if (not is_blocking_) return ret;
+        }
+
+        if (not is_blocking_)
+        {
+            last_error_ = EPROTO;
+            ret = wsrep::tls_service::error;
+        }
+        return ret;
+    }
+
+    enum wsrep::tls_service::status db_stream::handle_handshake_read(
+        const char* expect)
+    {
+        assert(::strlen(expect) >= 4);
+        char buf[4] = { };
+        ssize_t read_result(::read(fd_, buf, sizeof(buf)));
+        if (read_result > 0) stats_.bytes_read += size_t(read_result);
+        enum wsrep::tls_service::status ret;
+        if (read_result == -1 &&
+            (errno == EWOULDBLOCK || errno == EAGAIN))
+        {
+            ret = wsrep::tls_service::want_read;
+        }
+        else if (read_result == 0)
+        {
+            ret = wsrep::tls_service::eof;
+        }
+        else if (read_result != 4 || ::memcmp(buf, expect, 4))
+        {
+            last_error_ = EPROTO;
+            ret = wsrep::tls_service::error;
+        }
+        else
+        {
+            wsrep::log_info() << "Handshake success: " << std::string(buf, 4);
+            ret = wsrep::tls_service::success;
+        }
+        return ret;
+    }
+
+    wsrep::tls_service::op_result db_stream::read(void* buf, size_t max_count)
+    {
+        clear_error();
+        if (state_ == s_want_read)
+        {
+            state_ = s_idle;
+            if (max_count == 0)
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::success, 0};
+        }
+        max_count = determine_read_count(max_count);
+        ssize_t read_result(do_read(buf, max_count));
+        if (read_result > 0)
+        {
+            inc_reads(size_t(read_result));
+        }
+        return map_result(read_result);
+    }
+
+    wsrep::tls_service::op_result db_stream::write(
+        const void* buf, size_t count)
+    {
+        clear_error();
+        if (state_ == s_want_write)
+        {
+            state_ = s_idle;
+            if (count == 0)
+                return wsrep::tls_service::op_result{
+                    wsrep::tls_service::success, 0};
+        }
+        count = determine_write_count(count);
+        ssize_t write_result(do_write(buf, count));
+        if (write_result > 0)
+        {
+            inc_writes(size_t(write_result));
+        }
+        return map_result(write_result);
+    }
+}
+
+
+static db_stream::stats global_stats;
+std::mutex global_stats_lock;
+static int global_mode;
+
+static void merge_to_global_stats(const db_stream::stats& stats)
+{
+    std::lock_guard<std::mutex> lock(global_stats_lock);
+    global_stats.bytes_read += stats.bytes_read;
+    global_stats.bytes_written += stats.bytes_written;
+}
+
+wsrep::tls_stream* db::tls::create_tls_stream(int fd) WSREP_NOEXCEPT
+{
+    auto ret(new db_stream(fd, global_mode));
+    wsrep::log_debug() << "New DB stream: " << ret;
+    return ret;
+}
+
+void db::tls::destroy(wsrep::tls_stream* stream) WSREP_NOEXCEPT
+{
+    auto dbs(static_cast<db_stream*>(stream));
+    merge_to_global_stats(dbs->get_stats());
+    wsrep::log_debug() << "Stream destroy: " << dbs->get_stats().bytes_read
+                       << " " << dbs->get_stats().bytes_written;
+    wsrep::log_debug() << "Stream destroy" << dbs;
+    delete dbs;
+}
+
+int db::tls::get_error_number(const wsrep::tls_stream* stream)
+    const WSREP_NOEXCEPT
+{
+    return static_cast<const db_stream*>(stream)->get_error_number();
+}
+
+const void* db::tls::get_error_category(const wsrep::tls_stream* stream)
+    const WSREP_NOEXCEPT
+{
+    return static_cast<const db_stream*>(stream)->get_error_category();
+}
+
+const char* db::tls::get_error_message(const wsrep::tls_stream*,
+                                       int value, const void* category)
+    const WSREP_NOEXCEPT
+{
+    return db_stream::get_error_message(value, category);
+}
+
+enum wsrep::tls_service::status
+db::tls::client_handshake(wsrep::tls_stream* stream) WSREP_NOEXCEPT
+{
+    return static_cast<db_stream*>(stream)->client_handshake();
+}
+
+enum wsrep::tls_service::status
+db::tls::server_handshake(wsrep::tls_stream* stream) WSREP_NOEXCEPT
+{
+    return static_cast<db_stream*>(stream)->server_handshake();
+}
+
+wsrep::tls_service::op_result db::tls::read(
+    wsrep::tls_stream* stream,
+    void* buf, size_t max_count) WSREP_NOEXCEPT
+{
+    return static_cast<db_stream*>(stream)->read(buf, max_count);
+}
+
+wsrep::tls_service::op_result db::tls::write(
+    wsrep::tls_stream* stream,
+    const void* buf, size_t count) WSREP_NOEXCEPT
+{
+    return static_cast<db_stream*>(stream)->write(buf, count);
+}
+
+wsrep::tls_service::status
+db::tls::shutdown(wsrep::tls_stream*) WSREP_NOEXCEPT
+{
+    // @todo error simulation
+    return wsrep::tls_service::success;
+}
+
+
+void db::tls::init(int mode)
+{
+    global_mode = mode;
+}
+
+std::string db::tls::stats()
+{
+    std::ostringstream oss;
+    oss << "Transport stats:\n"
+        << "  bytes_read: " << global_stats.bytes_read << "\n"
+        << "  bytes_written: " << global_stats.bytes_written << "\n";
+    return oss.str();
+}

dbsim/db_tls.hpp

@@ -0,0 +1,62 @@
+/*
+ * Copyright (C) 2020 Codership Oy <info@codership.com>
+ *
+ * This file is part of wsrep-lib.
+ *
+ * Wsrep-lib is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * Wsrep-lib is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+#ifndef WSREP_DB_TLS_HPP
+#define WSREP_DB_TLS_HPP
+
+#include "wsrep/tls_service.hpp"
+
+#include <string>
+
+namespace db
+{
+    class tls : public wsrep::tls_service
+    {
+    public:
+        virtual wsrep::tls_stream* create_tls_stream(int)
+            WSREP_NOEXCEPT override;
+        virtual void destroy(wsrep::tls_stream*) WSREP_NOEXCEPT override;
+        virtual int get_error_number(const wsrep::tls_stream*) const
+            WSREP_NOEXCEPT override;
+        virtual const void* get_error_category(const wsrep::tls_stream*) const
+            WSREP_NOEXCEPT override;
+        virtual const char* get_error_message(const wsrep::tls_stream*,
+                                              int, const void*) const
+            WSREP_NOEXCEPT override;
+        virtual enum wsrep::tls_service::status
+        client_handshake(wsrep::tls_stream*)
+            WSREP_NOEXCEPT override;
+        virtual enum wsrep::tls_service::status
+        server_handshake(wsrep::tls_stream*)
+            WSREP_NOEXCEPT override;
+        virtual wsrep::tls_service::op_result
+        read(wsrep::tls_stream*, void* buf, size_t max_count)
+            WSREP_NOEXCEPT override;
+        virtual wsrep::tls_service::op_result
+        write(wsrep::tls_stream*, const void* buf, size_t count)
+            WSREP_NOEXCEPT override;
+        virtual wsrep::tls_service::status
+        shutdown(wsrep::tls_stream*) WSREP_NOEXCEPT override;
+
+        static void init(int mode);
+        static std::string stats();
+    };
+};
+
+#endif // WSREP_DB_TLS_HPP