Rework queue handling and error reporting, close rapid7#4249

Author: HD Moore

lib/msf/core/handler/reverse_tcp.rb

@@ -59,7 +59,6 @@ def initialize(info = {})
         OptBool.new('ReverseListenerThreaded', [ true, 'Handle every connection in a new thread (experimental)', false])
       ], Msf::Handler::ReverseTcp)
 
-    self.handler_queue = ::Queue.new
     self.conn_threads = []
   end
 
@@ -137,32 +136,41 @@ def cleanup_handler
   # Starts monitoring for an inbound connection.
   #
   def start_handler
+    queue = ::Queue.new
+
     local_port = bind_port
-    self.listener_thread = framework.threads.spawn("ReverseTcpHandlerListener-#{local_port}", false) {
-      client = nil
 
-      begin
+    self.listener_thread = framework.threads.spawn("ReverseTcpHandlerListener-#{local_port}", false, queue) { |lqueue|
+      loop do
         # Accept a client connection
         begin
           client = self.listener_sock.accept
-        rescue
-          wlog("Exception raised during listener accept: #{$!}\n\n#{$@.join("\n")}")
+          if ! client
+            wlog("ReverseTcpHandlerListener-#{local_port}: No client received in call to accept, exiting...")
+            break
+          end
+
+          self.pending_connections += 1
+          lqueue.push(client)
+        rescue ::Exception
+          wlog("ReverseTcpHandlerListener-#{local_port}: Exception raised during listener accept: #{$!}\n\n#{$@.join("\n")}")
           break
         end
-
-        # Increment the has connection counter
-        self.pending_connections += 1
-
-        self.handler_queue.push( client )
-      end while true
+      end
     }
 
-    self.handler_thread = framework.threads.spawn("ReverseTcpHandlerWorker-#{local_port}", false) {
-      while true
-        client = self.handler_queue.pop
+    self.handler_thread = framework.threads.spawn("ReverseTcpHandlerWorker-#{local_port}", false, queue) { |cqueue|
+      loop do
         begin
+          client = cqueue.pop
+
+          if ! client
+            elog("ReverseTcpHandlerWorker-#{local_port}: Queue returned an empty result, exiting...")
+            break
+          end
+
           if datastore['ReverseListenerThreaded']
-            self.conn_threads << framework.threads.spawn("ReverseTcpHandlerSession-#{local_port}-#{client.peerhost}", false, client) { | client_copy|
+            self.conn_threads << framework.threads.spawn("ReverseTcpHandlerSession-#{local_port}-#{client.peerhost}", false, client) { |client_copy|
               handle_connection(wrap_aes_socket(client_copy))
             }
           else
@@ -273,7 +281,6 @@ def bind_address
   attr_accessor :listener_sock # :nodoc:
   attr_accessor :listener_thread # :nodoc:
   attr_accessor :handler_thread # :nodoc:
-  attr_accessor :handler_queue # :nodoc:
   attr_accessor :conn_threads # :nodoc:
 end