[mesos] 05/06: Fixed a bug where OpenSSLSocketImpl accept loop can silently stop.

[bm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

bmahler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git

commit 5cd3f11733856d85605609cadf80a8a51d0bb4e0
Author: Benjamin Mahler <bm...@apache.org>
AuthorDate: Fri Apr 10 17:15:27 2020 -0400

    Fixed a bug where OpenSSLSocketImpl accept loop can silently stop.
    
    The accept loop was chaining the loop body Future on the result of
    io::poll/io::read, which meant that any failed poll/read would cause
    the loop body to return a failed future and the loop to stop running.
    
    This would lead to the server socket silently no longer accepting
    incoming connections.
    
    Review: https://reviews.apache.org/r/72352
---
 3rdparty/libprocess/src/ssl/openssl_socket.cpp | 29 +++++++++++++-------------
 3rdparty/libprocess/src/ssl/openssl_socket.hpp |  3 +--
 2 files changed, 15 insertions(+), 17 deletions(-)

diff --git a/3rdparty/libprocess/src/ssl/openssl_socket.cpp b/3rdparty/libprocess/src/ssl/openssl_socket.cpp
index f03a34f..31dc4a2 100644
--- a/3rdparty/libprocess/src/ssl/openssl_socket.cpp
+++ b/3rdparty/libprocess/src/ssl/openssl_socket.cpp
@@ -559,11 +559,11 @@ Future<std::shared_ptr<SocketImpl>> OpenSSLSocketImpl::accept()
         [weak_self]() -> Future<std::shared_ptr<SocketImpl>> {
           std::shared_ptr<OpenSSLSocketImpl> self(weak_self.lock());
 
-          if (self != nullptr) {
-            return self->PollSocketImpl::accept();
+          if (self == nullptr) {
+            return Failure("Socket destructed");
           }
 
-          return Failure("Socket destructed");
+          return self->PollSocketImpl::accept();
         },
         [weak_self](const std::shared_ptr<SocketImpl>& socket)
             -> Future<ControlFlow<Nothing>> {
@@ -577,12 +577,12 @@ Future<std::shared_ptr<SocketImpl>> OpenSSLSocketImpl::accept()
           // socket to become readable. We will then MSG_PEEK it to test
           // whether we want to dispatch as SSL or non-SSL.
           if (openssl::flags().support_downgrade) {
-            return io::poll(socket->get(), process::io::READ)
-              .then([weak_self, socket]() -> Future<ControlFlow<Nothing>> {
+            io::poll(socket->get(), process::io::READ)
+              .onReady([weak_self, socket]() {
                 std::shared_ptr<OpenSSLSocketImpl> self(weak_self.lock());
 
                 if (self == nullptr) {
-                  return Break();
+                  return;
                 }
 
                 char data[6];
@@ -624,15 +624,16 @@ Future<std::shared_ptr<SocketImpl>> OpenSSLSocketImpl::accept()
                 }
 
                 if (ssl) {
-                  return self->handle_accept_callback(socket);
+                  self->handle_accept_callback(socket);
                 } else {
                   self->accept_queue.put(socket);
-                  return Continue();
                 }
               });
+          } else {
+            self->handle_accept_callback(socket);
           }
 
-          return self->handle_accept_callback(socket);
+          return Continue();
         });
 
     accept_loop_started.done();
@@ -708,7 +709,7 @@ Try<Nothing, SocketError> OpenSSLSocketImpl::shutdown(int how)
 }
 
 
-Future<ControlFlow<Nothing>> OpenSSLSocketImpl::handle_accept_callback(
+void OpenSSLSocketImpl::handle_accept_callback(
     const std::shared_ptr<SocketImpl>& socket)
 {
   // Wrap this new socket up into our SSL wrapper class by releasing
@@ -720,7 +721,7 @@ Future<ControlFlow<Nothing>> OpenSSLSocketImpl::handle_accept_callback(
   SSL* accept_ssl = SSL_new(openssl::context());
   if (accept_ssl == nullptr) {
     accept_queue.put(Failure("Accept failed, SSL_new"));
-    return Continue();
+    return;
   }
 
   Try<Address> peer_address = network::peer(ssl_socket->get());
@@ -728,7 +729,7 @@ Future<ControlFlow<Nothing>> OpenSSLSocketImpl::handle_accept_callback(
     SSL_free(accept_ssl);
     accept_queue.put(
         Failure("Failed to determine peer IP: " + peer_address.error()));
-    return Continue();
+    return;
   }
 
   // NOTE: Right now, `openssl::configure_socket` does not do anything
@@ -742,7 +743,7 @@ Future<ControlFlow<Nothing>> OpenSSLSocketImpl::handle_accept_callback(
     accept_queue.put(
         Failure("Failed to openssl::configure_socket for " +
                 stringify(*peer_address) + ": " + configured.error()));
-    return Continue();
+    return;
   }
 
   // Set the SSL context in server mode.
@@ -807,8 +808,6 @@ Future<ControlFlow<Nothing>> OpenSSLSocketImpl::handle_accept_callback(
 
       self->accept_queue.put(ssl_socket);
     });
-
-  return Continue();
 }
 
 
diff --git a/3rdparty/libprocess/src/ssl/openssl_socket.hpp b/3rdparty/libprocess/src/ssl/openssl_socket.hpp
index 0528c03..2fe0f05 100644
--- a/3rdparty/libprocess/src/ssl/openssl_socket.hpp
+++ b/3rdparty/libprocess/src/ssl/openssl_socket.hpp
@@ -58,8 +58,7 @@ protected:
   // Verifies incoming sockets and initiates the SSL handshake.
   // Upon completion or failure of the SSL handshake, the peer socket
   // (or Failure object) will be enqueued on the server socket's accept queue.
-  Future<ControlFlow<Nothing>> handle_accept_callback(
-      const std::shared_ptr<SocketImpl>& socket);
+  void handle_accept_callback(const std::shared_ptr<SocketImpl>& socket);
 
   // Takes ownership of the given SSL object and performs an SSL handshake
   // with the context of the SSL object. Either `SSL_set_connect_state`