Land rapid7#8250, Fix packet ordering issue with reverse_tcp sessions

Author: Brent Cook

lib/rex/post/meterpreter/packet_dispatcher.rb

@@ -284,20 +284,6 @@ def keepalive
   # Reception
   #
   ##
-
-  #
-  # Simple class to track packets and if they are in-progress or complete.
-  #
-  class QueuedPacket
-    attr_reader :packet
-    attr_reader :in_progress
-
-    def initialize(packet, in_progress)
-      @packet = packet
-      @in_progress = in_progress
-    end
-  end
-
   #
   # Monitors the PacketDispatcher's sock for data in its own
   # thread context and parsers all inbound packets.
@@ -320,8 +306,8 @@ def monitor_socket
         begin
           rv = Rex::ThreadSafe.select([ self.sock.fd ], nil, nil, PING_TIME)
           if rv
-            packet, in_progress = receive_packet
-            @pqueue << QueuedPacket.new(packet, in_progress)
+            packet = receive_packet
+            @pqueue << packet if packet
           elsif self.send_keepalives && @pqueue.empty?
             keepalive
           end
@@ -356,11 +342,11 @@ def monitor_socket
         tmp_channel = []
         tmp_close   = []
         backlog.each do |pkt|
-          if(pkt.packet.response?)
+          if(pkt.response?)
             tmp_command << pkt
             next
           end
-          if(pkt.packet.method == "core_channel_close")
+          if(pkt.method == "core_channel_close")
             tmp_close << pkt
             next
           end
@@ -379,23 +365,21 @@ def monitor_socket
         backlog.each do |pkt|
 
           begin
-          if ! dispatch_inbound_packet(pkt.packet, pkt.in_progress)
+          if ! dispatch_inbound_packet(pkt)
             # Keep Packets in the receive queue until a handler is registered
             # for them. Packets will live in the receive queue for up to
             # PACKET_TIMEOUT seconds, after which they will be dropped.
             #
             # A common reason why there would not immediately be a handler for
             # a received Packet is in channels, where a connection may
             # open and receive data before anything has asked to read.
-            #
-            # Also, don't bother saving incomplete packets if we have no handler.
-            if (!pkt.in_progress and ::Time.now.to_i - pkt.packet.created_at.to_i < PACKET_TIMEOUT)
+            if (::Time.now.to_i - pkt.created_at.to_i < PACKET_TIMEOUT)
               incomplete << pkt
             end
           end
 
           rescue ::Exception => e
-            dlog("Dispatching exception with packet #{pkt.packet}: #{e} #{e.backtrace}", 'meterpreter', LEV_1)
+            dlog("Dispatching exception with packet #{pkt}: #{e} #{e.backtrace}", 'meterpreter', LEV_1)
           end
         end
 
@@ -475,16 +459,12 @@ def add_response_waiter(request, completion_routine = nil, completion_param = ni
   # Notifies a whomever is waiting for a the supplied response,
   # if anyone.
   #
-  # For not-yet-complete responses, we might not be able to determine
-  # the response ID, in that case just let all waiters know that some
-  # responses are trickling in.
-  #
-  def notify_response_waiter(response, in_progress=false)
+  def notify_response_waiter(response)
     handled = false
     self.waiters.each() { |waiter|
-      if (in_progress || waiter.waiting_for?(response))
-        waiter.notify(response, in_progress)
-        remove_response_waiter(waiter) unless in_progress
+      if (waiter.waiting_for?(response))
+        waiter.notify(response)
+        remove_response_waiter(waiter)
         handled = true
         break
       end
@@ -518,7 +498,7 @@ def initialize_inbound_handlers
   # Otherwise, the packet is passed onto any registered dispatch
   # handlers until one returns success.
   #
-  def dispatch_inbound_packet(packet, in_progress=false)
+  def dispatch_inbound_packet(packet)
     handled = false
 
     # Update our last reply time
@@ -527,7 +507,7 @@ def dispatch_inbound_packet(packet, in_progress=false)
     # If the packet is a response, try to notify any potential
     # waiters
     if packet.response?
-      if (notify_response_waiter(packet, in_progress))
+      if (notify_response_waiter(packet))
         return true
       end
     end

lib/rex/post/meterpreter/packet_parser.rb

@@ -75,27 +75,22 @@ def recv(sock)
       end
     end
 
-    in_progress = true
-
-    # TODO: cipher decryption
-    if (cipher)
-    end
-
-    # Deserialize the packet from the raw buffer
-    packet.from_r(self.raw)
-
     # If we've finished reading the entire packet
     if ((self.hdr_length_left == 0) &&
         (self.payload_length_left == 0))
 
+      # TODO: cipher decryption
+      if (cipher)
+      end
+
+      # Deserialize the packet from the raw buffer
+      packet.from_r(self.raw)
+
       # Reset our state
       reset
 
-      # packet is complete!
-      in_progress = false
+      return packet
     end
-
-    return packet, in_progress
   end
 
 protected

lib/rex/post/meterpreter/packet_response_waiter.rb

@@ -39,17 +39,13 @@ class PacketResponseWaiter
   # @return [Integer] request ID to wait for
   attr_accessor :rid
 
-  # @return [Boolean] indicates if part of the response has been received
-  attr_accessor :in_progress
-
   #
   # Initializes a response waiter instance for the supplied request
   # identifier.
   #
   def initialize(rid, completion_routine = nil, completion_param = nil)
     self.rid      = rid.dup
     self.response = nil
-    self.in_progress = false
 
     if (completion_routine)
       self.completion_routine = completion_routine
@@ -73,21 +69,14 @@ def waiting_for?(packet)
   #
   # @param response [Packet]
   # @return [void]
-  def notify(response, in_progress = false)
+  def notify(response)
     if (self.completion_routine)
-      self.in_progress = in_progress
-      unless in_progress
-        self.response = response
-        self.completion_routine.call(response, self.completion_param)
-      end
+      self.response = response
+      self.completion_routine.call(response, self.completion_param)
     else
       self.mutex.synchronize do
-        self.in_progress = in_progress
-        unless in_progress
-          # complete packet, ready for processing...
-          self.response = response
-          self.cond.signal
-        end
+        self.response = response
+        self.cond.signal
       end
     end
   end
@@ -103,11 +92,7 @@ def wait(interval)
     interval = nil if interval and interval == -1
     self.mutex.synchronize do
       if self.response.nil?
-        loop do
-          self.cond.wait(self.mutex, interval)
-          break unless self.in_progress
-          self.in_progress = false
-        end
+        self.cond.wait(self.mutex, interval)
       end
     end
     return self.response

spec/lib/rex/post/meterpreter/packet_parser_spec.rb

@@ -26,12 +26,11 @@
 
   it "should parse valid raw data into a packet object" do
     while @raw.length >0
-      parsed_packet, in_progress = parser.recv(@sock)
+      parsed_packet = parser.recv(@sock)
     end
     expect(parsed_packet).to be_a Rex::Post::Meterpreter::Packet
     expect(parsed_packet.type).to eq Rex::Post::Meterpreter::PACKET_TYPE_REQUEST
     expect(parsed_packet.method?("test_method")).to eq true
-    expect(in_progress).to eq false
   end
 
 end