Land #19141, Apache RocketMQ & ActiveMQ fixes

lib/msf/core/auxiliary/rocketmq.rb

@@ -25,10 +25,14 @@ def send_version_request
       begin
         connect
         sock.send(header + data_length + data, 0)
-        res = sock.recv(1024)
+        res_length = sock.timed_read(4)&.unpack1('N')
+        return nil if res_length.nil?
+
+        res = sock.timed_read(res_length)
       rescue Rex::AddressInUse, ::Errno::ETIMEDOUT, Rex::HostUnreachable, Rex::ConnectionTimeout, Rex::ConnectionRefused, ::Timeout::Error, ::EOFError => e
         print_error("Unable to connect: #{e.class} #{e.message}\n#{e.backtrace * "\n"}")
-        elog("#{e.class} #{e.message}\n#{e.backtrace * "\n"}")
+        elog('Error sending the rocketmq version request', error: e)
+        return nil
       ensure
         disconnect
       end
@@ -64,7 +68,11 @@ def get_rocketmq_version(id)
     # @return [Hash] Hash including RocketMQ versions info and Broker info if found
     def parse_rocketmq_data(res)
       # remove a response header so we have json-ish data
-      res = res[8..]
+      res = res.split(/\x00_/)[1]
+      unless res.starts_with?("{")
+        print_error("Failed to successfully remove the response header and now cannot parse the response.")
+        return nil
+      end
 
       # we have 2 json objects appended to each other, so we now need to split that out and make it usable
       res = res.split('}{')
@@ -111,14 +119,21 @@ def get_broker_port(broker_datas, rhost, default_broker_port: 10911)
       # Example of brokerData:
       # [{"brokerAddrs"=>{"0"=>"172.16.199.135:10911"}, "brokerName"=>"DESKTOP-8ATHH6O", "cluster"=>"DefaultCluster"}]
 
+      if broker_datas['brokerDatas'].blank?
+        print_status("brokerDatas field is missing from the response, assuming default broker port of #{default_broker_port}")
+        return default_broker_port
+      end
       broker_datas['brokerDatas'].each do |broker_data|
+        if broker_data['brokerAddrs'].blank?
+          print_status("brokerAddrs field is missing from the response, assuming default broker port of #{default_broker_port}")
+          return default_broker_port
+        end
         broker_data['brokerAddrs'].values.each do |broker_endpoint|
           next unless broker_endpoint.start_with?("#{rhost}:")
           return broker_endpoint.match(/\A#{rhost}:(\d+)\z/)[1].to_i
         end
       end
 
-
       print_status("autodetection failed, assuming default port of #{default_broker_port}")
       default_broker_port
     end

modules/exploits/multi/http/apache_rocketmq_update_config.rb

@@ -72,6 +72,8 @@ def initialize(info = {})
 
   def check
     @version_request_response = send_version_request
+    return Exploit::CheckCode::Unknown('Unable to determine the version') unless @version_request_response
+
     @parsed_data = parse_rocketmq_data(@version_request_response)
     return Exploit::CheckCode::Unknown('RocketMQ did not respond to the request for version information') unless @parsed_data['version']
 

modules/exploits/multi/misc/apache_activemq_rce_cve_2023_46604.rb

@@ -80,15 +80,19 @@ def initialize(info = {})
   def check
     connect
 
-    res = sock.get_once
+    len = sock.timed_read(4)&.unpack1('N')
+
+    return CheckCode::Unknown if len.nil? || len > 0x2000 # upper limit in case the service isn't ActiveMQ
+
+    res = sock.timed_read(len)
 
     disconnect
 
     return CheckCode::Unknown unless res
 
-    len, _, magic = res.unpack('NCZ*')
+    _, magic = res.unpack('CZ*')
 
-    return CheckCode::Unknown unless res.length == len + 4
+    return CheckCode::Unknown unless res.length == len
 
     return CheckCode::Unknown unless magic == 'ActiveMQ'
 
@@ -110,6 +114,8 @@ def check
     end
 
     Exploit::CheckCode::Safe("Apache ActiveMQ #{version}")
+  rescue ::Timeout::Error
+    CheckCode::Unknown
   end
 
   def exploit

spec/lib/msf/core/auxiliary/rocketmq_spec.rb

@@ -13,7 +13,7 @@
   end
 
   let(:expected_name_server_response) do
-    "\x00\x00\x01a\x00\x00\x00_{\"code\":0,\"flag\":1,\"language\":\"JAVA\",\"opaque\":1,\"serializeTypeCurrentRPC\":\"JSON\",\"version\":403}{\"brokerDatas\":[{\"brokerAddrs\":{\"0\":\"172.16.199.135:10911\"},\"brokerName\":\"DESKTOP-8ATHH6O\",\"cluster\":\"DefaultCluster\"}],\"filterServerTable\":{},\"queueDatas\":[{\"brokerName\":\"DESKTOP-8ATHH6O\",\"perm\":7,\"readQueueNums\":8,\"topicSysFlag\":0,\"writeQueueNums\":8}]}".b
+    "\x00\x00\x00_{\"code\":0,\"flag\":1,\"language\":\"JAVA\",\"opaque\":1,\"serializeTypeCurrentRPC\":\"JSON\",\"version\":403}{\"brokerDatas\":[{\"brokerAddrs\":{\"0\":\"172.16.199.135:10911\"},\"brokerName\":\"DESKTOP-8ATHH6O\",\"cluster\":\"DefaultCluster\"}],\"filterServerTable\":{},\"queueDatas\":[{\"brokerName\":\"DESKTOP-8ATHH6O\",\"perm\":7,\"readQueueNums\":8,\"topicSysFlag\":0,\"writeQueueNums\":8}]}".b
   end
 
   let(:expected_parsed_data_response) do
@@ -55,6 +55,9 @@
   describe '#send_version_request' do
     it 'returns version info' do
       expect(mock_sock).to receive(:send).with(mock_name_server_response, 0)
+      expect(mock_sock).to receive(:timed_read).with(4).and_return([expected_name_server_response.length].pack('N'))
+      expect(mock_sock).to receive(:timed_read).with(expected_name_server_response.length).and_return(expected_name_server_response)
+
       expect(subject.send_version_request).to eq(expected_name_server_response)
     end
   end
@@ -74,4 +77,4 @@
       expect(subject.get_broker_port(expected_parsed_data_response, '172.16.199.1', default_broker_port: 10000)).to eq(10000)
     end
   end
-end
+end