Metasploit meterpreter transport test module

meterpreter_transport.rb

require 'msf/core'
require 'rex'

lib = File.join(Msf::Config.install_root, "test", "lib")
$:.push(lib) unless $:.include?(lib)
require 'module_test'

class Metasploit4 < Msf::Post

  include Msf::ModuleTest::PostTest

  def initialize(info={})
    super( update_info( info,
        'Name'          => 'Testing Meterpreter Transport Stuff',
        'Description'   => %q{ This module will test meterpreter transport API methods },
        'License'       => MSF_LICENSE,
        'Author'        => [ 'Spencer McIntyre'],
        'Platform'      => %w{ java linux python windows },
        'SessionTypes'  => [ 'meterpreter' ]
      ))
  end

  def test_add_and_remove
    vprint_status("Starting transport addition and deletion tests")

    new_transport = {
      :lhost     => "10.%d.%d.%d" % [rand(255), rand(255), rand(255)],
      :lport     => rand(50000) + 1024,
      :transport => 'reverse_tcp'
    }
    new_url = "tcp://#{new_transport[:lhost]}:#{new_transport[:lport]}"

    it "should accept new transports" do
      ret = true

      org_transports = client.core.transport_list[:transports]
      ret &&= client.core.transport_add(new_transport)
      transports = client.core.transport_list[:transports]
      ret &&= (transports.length == org_transports.length + 1)
      if ret
        found = false
        transports.each do |trans|
          if trans[:url] == new_url
            found = true
            break
          end
        end
        ret &&= found
      end
      ret
    end

    it "should remove transports" do
      ret = true

      org_transports = client.core.transport_list[:transports]
      ret &&= client.core.transport_remove(new_transport)
      transports = client.core.transport_list[:transports]
      ret &&= (transports.length == org_transports.length - 1)
      ret
    end
  end

  def test_list
    vprint_status("Starting transport timeout tests")

    it "should have a non-empty list of transports" do
      transports = client.core.transport_list[:transports]
      !transports.empty?
    end

    it "should include all required transport fields" do
      ret = true
      transports = client.core.transport_list[:transports]
      transports.each do |trans|
        [:url, :comm_timeout, :retry_total, :retry_wait].each do |field|
          ret &&= !trans[field].nil?
        end
      end
      ret
    end
  end

  def test_set_timeouts
    vprint_status("Starting transport timeout tests")

    it "should return timeout information" do
      ret = true
      timeouts = client.core.set_transport_timeouts
      ret &&= !timeouts[:comm_timeout].nil?
      ret &&= !timeouts[:retry_total].nil?
      ret &&= !timeouts[:retry_wait].nil?
    end

    it "should set timeout information" do
      ret = true
      org_timeouts = client.core.set_transport_timeouts
      new_timeouts = {
        :session_exp  => rand(1000...5000),
        :comm_timeout => rand(1000...5000),
        :retry_total  => rand(1000...5000),
        :retry_wait   => rand(1000...5000)
      }

      resp_timeouts = client.core.set_transport_timeouts(new_timeouts)
      new_timeouts.each do |name, value|
        next if name == :session_exp
        ret &&= resp_timeouts[name] == value
      end

      client.core.set_transport_timeouts(org_timeouts)
      ret
    end
  end
end