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

Top of the crash stack:

    write_full_pages                storage/maria/ma_blockrec.c:2170
    write_block_record              storage/maria/ma_blockrec.c:3475
    allocate_and_write_block_record storage/maria/ma_blockrec.c:3580

Root cause: UNION ALL + global ORDER BY materializes the union result into an
Aria block-record temp table; a blob slot in the written record carries a
(length, pointer) pair that does not describe a real buffer, and
write_full_pages memcpys from the wild address.

Trigger requirements:
  - UNION + global ORDER BY (forces Aria temp-table materialization)
  - the big parameter value carried straight into a union output column
    (parameter index 4 / output column a2); both branches must scan real
    tables (with a single shared table or no table, the same root cause
    surfaces one frame away, in write_tail at ma_blockrec.c:2001)
  - value size: bytes * connection mbmaxlen > 65535 (utf8mb4: 16384 suffices;
    65536 used here); protocol-level type 0xfe (MYSQL_TYPE_STRING) for that
    parameter — SQL-level PREPARE/EXECUTE USING and text inlining cannot
    produce this type handler and do not reproduce the crash
  - the second branch's aggregate (count) and its column sources are needed
    to keep the record layout on the full-pages path
  - 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 `t3` (
  `c1` int NOT NULL, `c2` double DEFAULT NULL, `c3` text DEFAULT NULL,
  `c4` double DEFAULT NULL, `c5` tinyint DEFAULT NULL, `c6` int NOT NULL,
  `c7` double DEFAULT NULL, `c8` double DEFAULT NULL, `c9` double DEFAULT NULL,
  `c10` int NOT NULL,
  PRIMARY KEY (`c10`,`c6`), UNIQUE KEY `c8` (`c8`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4""",
    "INSERT INTO `t3` (`c1`, `c2`, `c3`, `c4`, `c5`, `c6`, `c7`, `c8`, `c9`, `c10`) VALUES ('0', '25.29', X'7E3063236A6230', NULL, '0', '0', '48.81', '1.200816770692068e-16', '4.185891831851989', '-1264390041')",
    "INSERT INTO `t3` (`c1`, `c2`, `c3`, `c4`, `c5`, `c6`, `c7`, `c8`, `c9`, `c10`) VALUES ('-1759736796', NULL, NULL, '57.29577951308232', NULL, '2147483647', '0.5514266812416906', '2.6', '1.859999999999999', '-1')",
    "CREATE TABLE `t4` (`c1` int DEFAULT NULL) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4",
]

PREPARE_SQL = (
    "select cast(? as signed) as a1, ? as a2, 1 as a3, 1 as a4, 1 as a5, "
    "cast(? as double) as a6 from t4 "
    "union all (select count(t.c3) - cast(? as signed) as a1, ? as a2, "
    "t.c9 as a3, t.c3 as a4, count(t.c4) as a5, t.c4 as a6 from t3 t) "
    "order by a5, a6"
)

# COM_STMT_EXECUTE: flags=0x00, iteration_count=1, null_bitmap=0x00,
# new_params_bind_flag=1; 5 params: LONG, LONG, DOUBLE, LONG, and param 4
# (index 4) = 0xfe STRING carrying the big value straight into output col a2.
PARAM_VALUE = b"D" * 65536


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 (t3 with 2 rows, t4 empty)")

    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 != 5:
        print("[-] expected 5 parameters")
        sys.exit(1)

    types = (b"\x03\x00" b"\x03\x00" b"\x05\x00" b"\x03\x00" b"\xfe\x00")
    values = (struct.pack("<I", 1) + struct.pack("<I", 1) +
              struct.pack("<d", 1.5) + struct.pack("<I", 1) +
              lenenc(PARAM_VALUE))
    payload = (b"\x17" + struct.pack("<I", stmt_id) + b"\x00" +
               struct.pack("<I", 1) + b"\x00" + b"\x01" + types + values)
    print(f"[*] sending EXECUTE: flags=0x00, payload_len={len(payload)}, "
          f"param4=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 write_full_pages@ma_blockrec.c:2170)")


if __name__ == "__main__":
    main()