Land rapid7#5851, meterpreter dispatcher queue

Author: egypt

lib/rex/post/meterpreter/packet_dispatcher.rb

@@ -253,7 +253,7 @@ def monitor_socket
 
     self.waiters = []
 
-    @pqueue = []
+    @pqueue = ::Queue.new
     @finish = false
     @last_recvd = Time.now
     @ping_sent = false
@@ -318,16 +318,12 @@ def monitor_socket
       # Whether we're finished or not is determined by the receiver
       # thread above.
       while(not @finish)
-        if(@pqueue.empty?)
-          ::IO.select(nil, nil, nil, 0.10)
-          next
-        end
-
         incomplete = []
         backlog    = []
 
+        backlog << @pqueue.pop
         while(@pqueue.length > 0)
-          backlog << @pqueue.shift
+          backlog << @pqueue.pop
         end
 
         #
@@ -393,11 +389,16 @@ def monitor_socket
           ::IO.select(nil, nil, nil, 0.10)
         end
 
-        @pqueue.unshift(*incomplete)
+        while incomplete.length > 0
+          @pqueue << incomplete.shift
+        end
 
         if(@pqueue.length > 100)
-          dlog("Backlog has grown to over 100 in monitor_socket, dropping older packets: #{@pqueue[0 .. 25].map{|x| x.inspect}.join(" - ")}", 'meterpreter', LEV_1)
-          @pqueue = @pqueue[25 .. 100]
+          removed = []
+          (1..25).each {
+            removed << @pqueue.pop
+          }
+          dlog("Backlog has grown to over 100 in monitor_socket, dropping older packets: #{removed.map{|x| x.inspect}.join(" - ")}", 'meterpreter', LEV_1)
         end
       end
       rescue ::Exception => e
@@ -454,15 +455,16 @@ def add_response_waiter(request, completion_routine = nil, completion_param = ni
   # if anyone.
   #
   def notify_response_waiter(response)
+    handled = false
     self.waiters.each() { |waiter|
       if (waiter.waiting_for?(response))
         waiter.notify(response)
-
         remove_response_waiter(waiter)
-
+        handled = true
         break
       end
     }
+    return handled
   end
 
   #

lib/rex/post/meterpreter/packet_response_waiter.rb

@@ -27,7 +27,8 @@ def initialize(rid, completion_routine = nil, completion_param = nil)
       self.completion_routine = completion_routine
       self.completion_param   = completion_param
     else
-      self.done  = false
+      self.mutex = Mutex.new
+      self.cond  = ConditionVariable.new
     end
   end
 
@@ -43,12 +44,14 @@ def waiting_for?(packet)
   # Notifies the waiter that the supplied response packet has arrived.
   #
   def notify(response)
-    self.response = response
-
     if (self.completion_routine)
+      self.response = response
       self.completion_routine.call(response, self.completion_param)
     else
-      self.done = true
+      self.mutex.synchronize do
+        self.response = response
+        self.cond.signal
+      end
     end
   end
 
@@ -57,25 +60,16 @@ def notify(response)
   # If the interval is -1 we can wait forever.
   #
   def wait(interval)
-    if( interval and interval == -1 )
-      while(not self.done)
-        ::IO.select(nil, nil, nil, 0.1)
-      end
-    else
-      begin
-        Timeout.timeout(interval) {
-          while(not self.done)
-            ::IO.select(nil, nil, nil, 0.1)
-          end
-        }
-      rescue Timeout::Error
-        self.response = nil
+    interval = nil if interval and interval == -1
+    self.mutex.synchronize do
+      if self.response.nil?
+        self.cond.wait(self.mutex, interval)
       end
     end
     return self.response
   end
 
-  attr_accessor :rid, :done, :response # :nodoc:
+  attr_accessor :rid, :mutex, :cond, :response # :nodoc:
   attr_accessor :completion_routine, :completion_param # :nodoc:
 end