ssl_handshake

contrib/epee/src/net_ssl.cpp

@@ -33,6 +33,8 @@
 #include <boost/cerrno.hpp>
 #include <boost/filesystem/operations.hpp>
 #include <boost/lambda/lambda.hpp>
+#include <boost/asio/strand.hpp>
+#include <condition_variable>
 #include <openssl/ssl.h>
 #include <openssl/pem.h>
 #include "misc_log_ex.h"
@@ -539,29 +541,90 @@ bool ssl_options_t::handshake(
     });
   }
 
-  auto& io_service = GET_IO_SERVICE(socket);
-  boost::asio::steady_timer deadline(io_service, timeout);
-  deadline.async_wait([&socket](const boost::system::error_code& error) {
-    if (error != boost::asio::error::operation_aborted)
+  auto start_handshake = [&]{
+    using ec_t = boost::system::error_code;
+    using timer_t = boost::asio::steady_timer;
+    using strand_t = boost::asio::io_service::strand;
+    using lock_t = std::mutex;
+    using lock_guard_t = std::lock_guard<lock_t>;
+    using condition_t = std::condition_variable_any;
+    using socket_t = boost::asio::ip::tcp::socket;
+
+    auto &io_context = GET_IO_SERVICE(socket);
+    if (io_context.stopped())
+      io_context.reset();
+    strand_t strand(io_context);
+    timer_t deadline(io_context, timeout);
+
+    struct state_t {
+      lock_t lock;
+      condition_t condition;
+      ec_t result;
+      bool wait_timer;
+      bool wait_handshake;
+      bool cancel_timer;
+      bool cancel_handshake;
+    };
+    state_t state{};
+
+    state.wait_timer = true;
+    auto on_timer = [&](const ec_t &ec){
+      lock_guard_t guard(state.lock);
+      state.wait_timer = false;
+      state.condition.notify_all();
+      if (not state.cancel_timer) {
+        state.cancel_handshake = true;
+        ec_t ec;
+        socket.next_layer().cancel(ec);
+      }
+    };
+
+    state.wait_handshake = true;
+    auto on_handshake = [&](const ec_t &ec, size_t bytes_transferred){
+      lock_guard_t guard(state.lock);
+      state.wait_handshake = false;
+      state.condition.notify_all();
+      state.result = ec;
+      if (not state.cancel_handshake) {
+        state.cancel_timer = true;
+        ec_t ec;
+        deadline.cancel(ec);
+      }
+    };
+
+    deadline.async_wait(on_timer);
+    strand.post(
+      [&]{
+        socket.async_handshake(
+          type,
+          boost::asio::buffer(buffer),
+          strand.wrap(on_handshake)
+        );
+      }
+    );
+
+    while (!io_context.stopped())
     {
-      socket.next_layer().close();
+      io_context.poll_one();
+      lock_guard_t guard(state.lock);
+      state.condition.wait_for(
+        state.lock,
+        std::chrono::milliseconds(30),
+        [&]{
+          return not state.wait_timer and not state.wait_handshake;
+        }
+      );
+      if (not state.wait_timer and not state.wait_handshake)
+        break;
     }
-  });
-
-  boost::system::error_code ec = boost::asio::error::would_block;
-  socket.async_handshake(type, boost::asio::buffer(buffer), boost::lambda::var(ec) = boost::lambda::_1);
-  if (io_service.stopped())
-  {
-    io_service.reset();
-  }
-  while (ec == boost::asio::error::would_block && !io_service.stopped())
-  {
-    // should poll_one(), can't run_one() because it can block if there is
-    // another worker thread executing io_service's tasks
-    // TODO: once we get Boost 1.66+, replace with run_one_for/run_until
-    std::this_thread::sleep_for(std::chrono::milliseconds(30));
-    io_service.poll_one();
-  }
+    if (state.result.value()) {
+      ec_t ec;
+      socket.next_layer().shutdown(socket_t::shutdown_both, ec);
+      socket.next_layer().close(ec);
+    }
+    return state.result;
+  };
+  const auto ec = start_handshake();
 
   if (ec)
   {

tests/unit_tests/epee_boosted_tcp_server.cpp

@@ -462,3 +462,50 @@ TEST(test_epee_connection, test_lifetime)
   server.timed_wait_server_stop(5 * 1000);
   server.deinit_server();
 }
+
+TEST(test_epee_connection, ssl_handshake)
+{
+  using io_context_t = boost::asio::io_service;
+  using work_t = boost::asio::io_service::work;
+  using work_ptr = std::shared_ptr<work_t>;
+  using workers_t = std::vector<std::thread>;
+  using socket_t = boost::asio::ip::tcp::socket;
+  using ssl_socket_t = boost::asio::ssl::stream<socket_t>;
+  using ssl_socket_ptr = std::unique_ptr<ssl_socket_t>;
+  using ssl_options_t = epee::net_utils::ssl_options_t;
+  io_context_t io_context;
+  work_ptr work(std::make_shared<work_t>(io_context));
+  workers_t workers;
+  auto constexpr N = 2;
+  while (workers.size() < N) {
+    workers.emplace_back([&io_context]{
+      io_context.run();
+    });
+  }
+  ssl_options_t ssl_options{{}};
+  auto ssl_context = ssl_options.create_context();
+  for (size_t i = 0; i < N * N * N; ++i) {
+    ssl_socket_ptr ssl_socket(new ssl_socket_t(io_context, ssl_context));
+    ssl_socket->next_layer().open(boost::asio::ip::tcp::v4());
+    for (size_t i = 0; i < N; ++i) {
+      io_context.post([]{
+        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+      });
+    }
+    EXPECT_EQ(
+      ssl_options.handshake(
+        *ssl_socket,
+        ssl_socket_t::server,
+        {},
+        {},
+        std::chrono::milliseconds(0)
+      ),
+      false
+    );
+    ssl_socket->next_layer().close();
+    ssl_socket.reset();
+  }
+  work.reset();
+  for (;workers.size(); workers.pop_back())
+    workers.back().join();
+}