MySQL connections: drivers, pools and proxies

connect_session.py

"""Three drivers, one lab server, verified TLS. Python, Node.js and Java each open a
connection to a throwaway MySQL 8.4.11 server on 127.0.0.1, verify the server
certificate against the CA the server generated, run the same parameterised SELECT and
INSERT, and print what they got. The PyMySQL program also tries verifying the server's
host name (VERIFY_IDENTITY) and shows that MySQL's auto-generated certificate, whose
SAN does not list 127.0.0.1, is refused: production uses a certificate that names the
host and the check then passes."""

import subprocess
import sys
from pathlib import Path

from mysql_session import Server, toolchain

INSTALLS = Path(sys.executable).resolve().parents[3]
KEYTOOL = INSTALLS / "java" / "temurin-25.0.4+101.0.LTS" / "bin" / "keytool"

with Server() as my:
    my.up()
    # A tiny schema the three programs share: one user to look up, one target table for
    # the INSERT. Nothing else of the server is involved.
    my.sql("CREATE DATABASE shop")
    my.sql("CREATE TABLE shop.app_user (id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(40) NOT NULL)")
    my.sql("INSERT INTO shop.app_user (name) VALUES ('asha'), ('ravi'), ('kiran')")
    my.sql("CREATE TABLE shop.order_item (id INT PRIMARY KEY AUTO_INCREMENT, order_id INT, sku VARCHAR(10), qty INT)",
           flags=("-t", "--skip-comments"))

    # Connector/J needs a Java trust store. The session builds one from the server's own
    # ca.pem with keytool before the Java program runs.
    data = my.base / "data"
    truststore = my.base / "truststore.jks"
    print("$ keytool -importcert -alias mysql-ca -file data/ca.pem -keystore truststore.jks "
          "-storepass changeit -noprompt")
    subprocess.run(
        [str(KEYTOOL), "-importcert", "-alias", "mysql-ca", "-file", str(data / "ca.pem"),
         "-keystore", str(truststore), "-storepass", "changeit", "-noprompt"],
        check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )
    print("(certificate added to a new keystore)\n")

    # One value changes between runs: the TLS cipher name differs across the three
    # drivers' TLS libraries, so the session masks it. Everything else is stable --
    # each program inserts one row, in this order, so the ids read 1, 2 and 3.
    mask = ((r"TLS cipher negotiated: TLS_[A-Z0-9_]+", "TLS cipher negotiated: <TLS cipher>"),)

    my.app("python3", "py_connect.py", mask=mask)
    my.app("node", "node_connect.mjs", mask=mask)
    my.app("java", "JavaConnect.java", mask=mask)

$ mysql -t -e "CREATE DATABASE shop"

$ mysql -t -e "CREATE TABLE shop.app_user (id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(40) NOT NULL)"

$ mysql -t -e "INSERT INTO shop.app_user (name) VALUES ('asha'), ('ravi'), ('kiran')"

$ mysql -t --skip-comments -e "CREATE TABLE shop.order_item (id INT PRIMARY KEY AUTO_INCREMENT, order_id INT, sku VARCHAR(10), qty INT)"

$ keytool -importcert -alias mysql-ca -file data/ca.pem -keystore truststore.jks -storepass changeit -noprompt
(certificate added to a new keystore)

$ python3 py_connect.py
server version: 8.4.11
TLS cipher negotiated: <TLS cipher>
parameterised SELECT: (1, 'asha')
parameterised INSERT ok; last id = 1
pymysql version: 1.2.3
VERIFY_IDENTITY: refused - IP address mismatch

$ node node_connect.mjs
server version: 8.4.11
TLS cipher negotiated: <TLS cipher>
parameterised SELECT: { id: 1, name: 'asha' }
parameterised INSERT ok; last id = 2

$ java JavaConnect.java
server version: 8.4.11
TLS cipher negotiated: <TLS cipher>
parameterised SELECT: id=1, name=asha
parameterised INSERT ok; last id = 3
Connector/J version: mysql-connector-j-26.7.0

py_connect.py

"""One parameterised query from PyMySQL 1.2.3 against the lab server, over TLS verified
against the server's CA. The second attempt asks for VERIFY_IDENTITY (default
check_hostname): MySQL's auto-generated server certificate does not name 127.0.0.1 and
PyMySQL refuses it."""

import os
import pymysql

CA = os.environ["LAB_CA"]


def connect(check_hostname):
    return pymysql.connect(
        host=os.environ["LAB_HOST"], port=int(os.environ["LAB_PORT"]),
        user=os.environ["LAB_USER"], password=os.environ["LAB_PASSWORD"],
        database="shop",
        ssl={"ca": CA, "check_hostname": check_hostname},
    )


# VERIFY_CA: trust the server certificate because it chains to the CA we gave the driver.
conn = connect(check_hostname=False)
with conn.cursor() as cur:
    cur.execute("SELECT @@version")
    print("server version:", cur.fetchone()[0])
    cur.execute("SHOW SESSION STATUS LIKE 'Ssl_cipher'")
    print("TLS cipher negotiated:", cur.fetchone()[1])
    cur.execute("SELECT id, name FROM app_user WHERE name = %s", ("asha",))
    print("parameterised SELECT:", cur.fetchone())
    cur.execute("INSERT INTO order_item (order_id, sku, qty) VALUES (%s, %s, %s)", (1, "T-01", 2))
    conn.commit()
    print("parameterised INSERT ok; last id =", cur.lastrowid)
conn.close()
print("pymysql version:", pymysql.VERSION_STRING)

# VERIFY_IDENTITY: also check the server's host name. The auto-generated certificate has
# no 127.0.0.1 in its SAN, so verification fails. In production the certificate names the
# host and this check passes.
try:
    conn = connect(check_hostname=True)
    conn.close()
    print("VERIFY_IDENTITY: unexpectedly succeeded")
except Exception as e:
    text = str(e)
    reason = "IP address mismatch" if "IP address mismatch" in text else text[:80]
    print("VERIFY_IDENTITY: refused -", reason)

node_connect.mjs

// One parameterised query from mysql2 3.24.5 against the lab server, over TLS verified
// against the server's CA. The driver reads the CA pem and keeps rejectUnauthorized
// true; checkServerIdentity is overridden because MySQL's auto-generated certificate
// does not name 127.0.0.1 (CA verification stays on).

import { readFileSync } from 'node:fs';
import mysql from 'mysql2/promise';

const conn = await mysql.createConnection({
  host: process.env.LAB_HOST,
  port: Number(process.env.LAB_PORT),
  user: process.env.LAB_USER,
  password: process.env.LAB_PASSWORD,
  database: 'shop',
  ssl: {
    ca: readFileSync(process.env.LAB_CA),
    rejectUnauthorized: true,
    checkServerIdentity: () => undefined,
  },
});

const [[v]] = await conn.query('SELECT @@version AS version');
console.log('server version:', v.version);

const [[c]] = await conn.query("SHOW SESSION STATUS LIKE 'Ssl_cipher'");
console.log('TLS cipher negotiated:', c.Value);

const [[row]] = await conn.execute(
  'SELECT id, name FROM app_user WHERE name = ?',
  ['asha'],
);
console.log('parameterised SELECT:', row);

const [res] = await conn.execute(
  'INSERT INTO order_item (order_id, sku, qty) VALUES (?, ?, ?)',
  [1, 'T-01', 2],
);
console.log('parameterised INSERT ok; last id =', res.insertId);

await conn.end();

JavaConnect.java

// One parameterised query from MySQL Connector/J 26.7.0 against the lab server, over
// TLS verified against the server's CA. The session hands the CA to keytool as a Java
// trust store before this program runs; the driver then verifies the chain
// (sslMode=VERIFY_CA). Java 25's single-file source launcher compiles and runs this
// file in one step: `java JavaConnect.java`.

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Properties;

public class JavaConnect {
    public static void main(String[] args) throws Exception {
        String url = "jdbc:mysql://" + System.getenv("LAB_HOST") + ":"
                + System.getenv("LAB_PORT") + "/shop";
        Properties p = new Properties();
        p.setProperty("user", System.getenv("LAB_USER"));
        p.setProperty("password", System.getenv("LAB_PASSWORD"));
        p.setProperty("sslMode", "VERIFY_CA");
        p.setProperty("trustCertificateKeyStoreUrl", "file:truststore.jks");
        p.setProperty("trustCertificateKeyStorePassword", "changeit");

        try (Connection c = DriverManager.getConnection(url, p)) {
            try (Statement s = c.createStatement();
                 ResultSet rs = s.executeQuery("SELECT @@version")) {
                while (rs.next()) System.out.println("server version: " + rs.getString(1));
            }
            try (Statement s = c.createStatement();
                 ResultSet rs = s.executeQuery("SHOW SESSION STATUS LIKE 'Ssl_cipher'")) {
                while (rs.next()) System.out.println("TLS cipher negotiated: " + rs.getString(2));
            }
            try (PreparedStatement ps = c.prepareStatement(
                     "SELECT id, name FROM app_user WHERE name = ?")) {
                ps.setString(1, "asha");
                try (ResultSet rs = ps.executeQuery()) {
                    while (rs.next()) {
                        System.out.println("parameterised SELECT: id=" + rs.getInt(1)
                                + ", name=" + rs.getString(2));
                    }
                }
            }
            try (PreparedStatement ps = c.prepareStatement(
                     "INSERT INTO order_item (order_id, sku, qty) VALUES (?, ?, ?)",
                     Statement.RETURN_GENERATED_KEYS)) {
                ps.setInt(1, 1);
                ps.setString(2, "T-01");
                ps.setInt(3, 2);
                ps.executeUpdate();
                try (ResultSet keys = ps.getGeneratedKeys()) {
                    if (keys.next()) {
                        System.out.println("parameterised INSERT ok; last id = " + keys.getLong(1));
                    }
                }
            }
            System.out.println("Connector/J version: "
                    + c.getMetaData().getDriverVersion().split(" \\(")[0]);
        }
    }
}

mysql_session.py

"""Throwaway MySQL 8.4.11 and MariaDB servers for the examples of this lesson.

Every session creates a server of its own: `mysqld --initialize-insecure` makes a new data directory inside a new
folder in the examples folder, the server starts on it, the session runs `mysql` client scripts and MySQL's programs
against it and prints what they printed, and then the server is stopped and the folder deleted, also when something
fails. The server is the pinned MySQL 8.4.11 from the mise installations next to the Python that runs this file;
Server(basedir=...) runs the programs of another unpacked MySQL tarball instead (dump_defaults.py uses it).
MariaDB(version="12.3") (or "11.8") does the same with a MariaDB server: `mariadb-install-db` makes the data
directory, `mariadbd` serves it, and the session runs MariaDB's own programs (`mariadb`, `mariadb-admin`,
`mariadb-dump`, `mariadb-binlog`, `mariadb-backup`), which is how the printed commands name them. They are the
Homebrew macOS builds of MariaDB 12.3.3 and 11.8.9 (MariaDB publishes no macOS build of its own), unpacked in a
toolchains folder above this one (toolchain()). A session may run a MySQL and a MariaDB server side by side.

Safety: the server listens only on 127.0.0.1, an address no other computer can reach, on a port the operating system
reports as unused; the X Protocol listener is off (--mysqlx=OFF); its Unix socket, temporary files, error log and
import/export folder all stay inside the run folder. Right after the first start the session gives root a password made
up for the run (never printed) and connects only with a client option file that holds it, so no password ever appears
on a command line. Every program starts with --no-defaults or --defaults-file, so no option file of the computer
(/etc/my.cnf, ~/.my.cnf) can change a result, and HOME points at the run folder, so ~/.mylogin.cnf is not read either.
A printed mysqld command shows the options that matter to the lesson; the session also passes, on the command line,
the folders it keeps inside the run folder (--basedir, --tmpdir, --log-error, --pid-file, --secure-file-priv) and
the port, which is why performance_schema.variables_info reports those variables as COMMAND_LINE.
The sessions read only this folder, the MySQL installation and the unpacked toolchains, and they connect to nothing
but 127.0.0.1.

The output is exactly what the programs printed, with these substitutions so that a record never shows the computer it
ran on and is the same on every run: the run folder is shown as ".", the folder of MySQL's programs is left out of
their paths and MySQL's own folder is shown as <basedir>, the computer's name is shown as "lab-host" and the operating-system user's name as "learner", the time a
statement took (the client's "(0.01 sec)") is left out, the time stamp SET PERSIST writes into mysqld-auto.cnf is
shown as <when it was set>, error-log lines lose their timestamp and thread number, and a session can name more values
that change between runs with my.mask(pattern, replacement), which then prints in their place. A client table whose cells
change is drawn again so that its borders still line up.
The sessions never print the password, and the run's port prints as <port>. A session stops with status 1 when a command that should have worked
fails, or when a script reports a different number of errors than the session expects, so a run that went wrong is
never recorded as if it had worked.
"""

import os
import pwd
import re
import secrets
import shlex
import shutil
import socket
import subprocess
import sys
import tempfile
import textwrap
import time
from pathlib import Path

MYSQL_VERSION = "8.4.11"
# .../installs/python/3.14.8/bin/python3 -> .../installs
INSTALLS = Path(sys.executable).resolve().parents[3]
PINNED = INSTALLS / "mysql" / MYSQL_VERSION
OS_USER = pwd.getpwuid(os.geteuid()).pw_name
HOST = socket.gethostname().split(".")[0]
# 2026-10-17T16:50:01.123456Z 0 [ERROR] ... -> [ERROR] ...
LOG_PREFIX = re.compile(r"^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d+Z \d+ ", re.M)
# MariaDB: "2026-10-20  0:56:56 0 [ERROR] ..." -> "[ERROR] ..."
MARIADB_LOG_PREFIX = re.compile(r"^\d{4}-\d\d-\d\d +\d{1,2}:\d\d:\d\d \d+ ", re.M)
# The unpacked toolchains the sessions use (toolchain()): the MariaDB builds and the three drivers.
TOOLCHAINS = ("mariadb-12.3.3", "pymysql-1.2.3", "node-mysql2-3.24.5", "mysql-connector-j-26.7.0")


def toolchain(name):
    """The unpacked toolchain <name>, one of TOOLCHAINS. They all come from one folder: the first folder called
    toolchains that holds every one of TOOLCHAINS, looked for above the folder the session runs in and above this
    file, directly or inside one of the folders of a tmp folder there. The search stops below the home folder."""
    home = Path(pwd.getpwuid(os.geteuid()).pw_dir).resolve()
    seen = set()
    for start in (Path.cwd().resolve(), Path(__file__).resolve().parent):
        for parent in (start, *start.parents):
            if parent in seen or parent == home or parent == Path(parent.anchor):
                break
            seen.add(parent)
            scratch = parent / "tmp"
            for folder in (parent, *(sorted(scratch.iterdir()) if scratch.is_dir() else ())):
                found = folder / "toolchains"
                if all((found / each).is_dir() for each in TOOLCHAINS):
                    return found / name
    sys.exit(f"no toolchains folder holds all of {', '.join(TOOLCHAINS)}")


def free_port():
    """A TCP port on 127.0.0.1 that nothing listens on now."""
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


class Server:
    """One throwaway server: initialise, start; stopped and deleted when the with block ends.

    options: extra mysqld options for every start (for example "--sql-mode=" or "--character-set-server=latin1").
    basedir: the folder of an unpacked MySQL tarball to run instead of the pinned 8.4.11.
    masks: (pattern, replacement) pairs, as for mask().
    """

    SERVER, CLIENT, ADMIN = "mysqld", "mysql", "mysqladmin"   # the programs; MariaDB names its own
    LOG_RE = LOG_PREFIX

    def __init__(self, options=(), redo=None, basedir=None, masks=(), label=None, port_label=None):
        self.options = [str(o) for o in options]
        self.redo = redo
        self.basedir = Path(basedir).resolve() if basedir else PINNED
        self.bin = self.basedir / "bin"
        self.label = label                       # a stable name for this server in the record ("source", "replica")
        self.port_label = port_label             # the text that shows in place of this server's port
        self.uuid = None
        self.masks = []
        for pattern, replacement in masks:
            self.mask(pattern, replacement)
        self.proc = None

    def mask(self, pattern, replacement):
        """From now on every printed line shows text matching the regular expression as the replacement: for a value
        that changes from run to run (an uptime, a time stamp), so that the recorded output stays the same."""
        self.masks.append((re.compile(pattern), replacement))

    # ---- the folder ------------------------------------------------------------------------------------------------
    def __enter__(self):
        if not (self.bin / self.SERVER).exists():
            sys.exit(f"no MySQL server program in {self.basedir}" if self.basedir != PINNED
                     else f"MySQL {MYSQL_VERSION} was not found next to this Python")
        self.base = Path(tempfile.mkdtemp(prefix="my", dir=".")).resolve()
        for sub in ("tmp", "files"):
            (self.base / sub).mkdir(mode=0o700)
        self.port = free_port()
        self.password = secrets.token_urlsafe(24)
        self.client_cnf = self.base / "client.cnf"
        self._write_client_cnf(password=None)
        self.env = {key: value for key, value in os.environ.items()
                    if not key.startswith("MYSQL") and key not in ("TZ", "HOME", "LANG", "LC_ALL")}
        self.env.update({"HOME": str(self.base), "TZ": "UTC", "LANG": "en_US.UTF-8",
                         "MYSQL_TEST_LOGIN_FILE": str(self.base / "no-login-paths.cnf")})
        # mariadb-install-db (a shell script) calls chown, which lives in /usr/sbin; a run's PATH may hold only
        # python's bin, /usr/bin and /bin, so add the system sbin folders when they are missing. These hold
        # standard OS tools, nothing of this computer's own, and the session never prints the PATH.
        parts = self.env.get("PATH", "/usr/bin:/bin").split(os.pathsep)
        for d in ("/usr/sbin", "/sbin"):
            if d not in parts:
                parts.append(d)
        self.env["PATH"] = os.pathsep.join(parts)
        for pattern in ("*.sql", "*.cnf", "*.csv", "*.txt"):
            for f in Path(".").glob(pattern):
                shutil.copy(f, self.base / f.name)
        return self

    def __exit__(self, *exc):
        if self.proc and self.proc.poll() is None:
            self.stop(quiet=True)
        shutil.rmtree(self.base, ignore_errors=True)

    def _write_client_cnf(self, password):
        lines = ["[client]", "user=root", "host=127.0.0.1", f"port={self.port}", "protocol=TCP"]
        if password:
            lines.append(f"password={password}")
        self.client_cnf.write_text("\n".join(lines) + "\n")
        self.client_cnf.chmod(0o600)

    def server_options(self):
        b = self.base
        return ["--no-defaults", f"--basedir={self.basedir}", f"--datadir={b / 'data'}", f"--tmpdir={b / 'tmp'}",
                f"--log-error={b / 'mysqld.err'}", f"--pid-file={b / 'mysqld.pid'}",
                f"--secure-file-priv={b / 'files'}",
                *([f"--innodb-redo-log-capacity={self.redo}"] if self.redo else [])]

    # ---- the server ------------------------------------------------------------------------------------------------
    def initialize(self, show=False):
        """mysqld --initialize-insecure: a new data directory whose root account has no password yet."""
        args = ["mysqld", *self.server_options(), "--initialize-insecure"]
        code, out = self._call(args)
        if show:
            _show(["mysqld", "--no-defaults", "--initialize-insecure", "--datadir=./data"], out)
        if code != 0:
            print(self.log())
            sys.exit(f"mysqld --initialize-insecure ended with status {code}")

    def start(self, extra=(), defaults_file=None, expect_fail=False, show=False):
        """Starts mysqld on 127.0.0.1 and waits until it answers. With expect_fail, prints the error-log lines of a
        start that must fail, and returns them."""
        log = self.base / "mysqld.err"
        before = log.read_text() if log.exists() else ""
        opts = self.server_options()
        if defaults_file:
            opts[0] = f"--defaults-file={self.base / defaults_file}"
        args = [str(self.bin / self.SERVER), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
                *self.listener_options(), *self.options, *[str(e) for e in extra]]
        if show:
            shown = [self.SERVER, opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
                     "--port=$PORT", *self.listener_options(), *self.options, *extra]
            print("$ " + " ".join(_arg(a) if a != "--port=$PORT" else a for a in shown) + " &")
            print()
        with open(self.base / "mysqld.out", "a") as out:
            self.proc = subprocess.Popen(args, stdout=out, stderr=subprocess.STDOUT, cwd=self.base, env=self.env)
        for _ in range(600):
            if self.proc.poll() is not None:
                break
            if self._ping():
                break
            time.sleep(0.1)
        if self.proc.poll() is not None:
            new = (log.read_text() if log.exists() else "")[len(before):]
            lines = [l for l in self.clean(self.LOG_RE.sub("", new)).splitlines() if "[ERROR]" in l]
            if expect_fail:
                print("\n".join(lines))
                print(f"(mysqld ended with status {self.proc.returncode})")
                print()
                self.proc = None
                return lines
            print("\n".join(lines))
            sys.exit(f"mysqld ended with status {self.proc.returncode} while starting")
        if expect_fail:
            sys.exit("mysqld started, but the session expected it to fail")
        if not self._ping():
            sys.exit("mysqld did not answer within 60 seconds")
        return []

    def listener_options(self):
        """The socket inside the run folder and, for MySQL, the X Protocol listener switched off."""
        return ["--socket=mysql.sock", "--mysqlx=OFF"]

    def app(self, program, script, *args, status=0, env=None, port=None, mask=(), note=None, timeout=100, show=True,
            during=None):
        """Runs an application program of the examples folder against this server and prints it as a shell would.
        program: "python3" (PyMySQL 1.2.3), "node" (mysql2 3.24.5) or "java" (Connector/J 26.7.0, the single-file
        source launcher: `java Demo.java`). The program finds the server in the environment variables LAB_HOST
        (127.0.0.1), LAB_PORT (printed as <port>), LAB_USER (root unless env says otherwise), LAB_PASSWORD (the run's
        password: a program never prints it, and the record would show <password> if it did) and LAB_CA (the server's
        auto-generated ca.pem, for verified TLS); env adds or replaces variables, port points it at another server of
        the run (a proxy, a replica). It runs in the run folder, beside the session's files; its output is cleaned like
        every program's, with mask=((pattern, replacement), ...) for values that change between runs. during, a
        function of no arguments, runs unprinted while the program runs."""
        tools = {"python3": None, "node": toolchain("node-mysql2-3.24.5"), "java": toolchain("mysql-connector-j-26.7.0")}
        if program not in tools:
            sys.exit(f"app() does not run {program}")
        source = Path(script)
        if not source.exists():
            sys.exit(f"{script} is not in the examples folder")
        shutil.copy(source, self.base / source.name)
        ca = self.base / "data" / "ca.pem"
        merged = {**self.env, "LAB_HOST": "127.0.0.1", "LAB_PORT": str(port or self.port), "LAB_USER": "root",
                  "LAB_PASSWORD": self.password, "LAB_CA": str(ca) if ca.exists() else "", **dict(env or {})}
        if program == "python3":
            command = [sys.executable, source.name]
            merged.update({"PYTHONPATH": str(toolchain("pymysql-1.2.3") / "site"), "PYTHONNOUSERSITE": "1",
                           "PYTHONDONTWRITEBYTECODE": "1", "PYTHONUNBUFFERED": "1"})
        elif program == "node":
            node = INSTALLS / "node" / "24.21.0" / "bin" / "node"
            if not node.exists():
                sys.exit("Node.js 24.21.0 was not found next to this Python")
            # Node resolves import and require from a node_modules folder beside the program (NODE_PATH does not
            # reach ES modules), so the run folder gets a link to the installed driver.
            link = self.base / "node_modules"
            if not link.exists():
                link.symlink_to(tools["node"] / "node_modules", target_is_directory=True)
            command = [str(node), source.name]
        else:
            java = INSTALLS / "java" / "temurin-25.0.4+101.0.LTS" / "bin" / "java"
            if not java.exists():
                sys.exit("Java 25 (temurin-25.0.4+101.0.LTS) was not found next to this Python")
            # No performance-data file and no temporary file outside the run folder (macOS gives Java its own
            # temporary folder otherwise), and the home folder is the run folder.
            command = [str(java), "-XX:-UsePerfData", f"-Djava.io.tmpdir={self.base / 'tmp'}",
                       f"-Duser.home={self.base}", "-cp", str(tools["java"] / "mysql-connector-j-26.7.0.jar"),
                       source.name]
        command += [str(a) for a in args]
        if note:
            print(f"-- {note}")
        proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=self.base,
                                env=merged)
        try:
            if during is not None:
                during()
            output, _ = proc.communicate(timeout=timeout)
        except subprocess.TimeoutExpired:
            proc.kill()
            proc.communicate()
            sys.exit(f"{script} did not finish in {timeout} s")
        except BaseException:
            proc.kill()
            proc.communicate()
            raise
        # A program's traceback names the files it ran: the drivers' folder and the language's installation are shown
        # as <toolchains> and <installs>, so a record never shows this computer's folders.
        raw = output.replace(self.password, "<password>")
        raw = raw.replace(str(tools["java"].parent if tools["java"] else toolchain("pymysql-1.2.3").parent),
                          "<toolchains>").replace(str(INSTALLS), "<installs>")
        out = self.clean(raw)
        for pattern, replacement in mask:
            out = re.sub(pattern, replacement, out)
        if show:
            _show([program, source.name, *args], out)
        if proc.returncode != status:
            if not show:
                print(out)
            sys.exit(f"{script} ended with status {proc.returncode}, not {status}")
        return out

    def _ping(self):
        code, _ = self._call([self.ADMIN, f"--defaults-file={self.client_cnf}", "--connect-timeout=2", "ping"])
        return code == 0

    def up(self, extra=(), defaults_file=None, show=False):
        """Initialise, start (with start()'s options), and give root the run's password, quietly."""
        self.initialize()
        self.start(extra=extra, defaults_file=defaults_file, show=show)
        return self.login()

    def login(self):
        """Gives root the run's password (never printed) on a server started from a new data directory, and makes
        every later client of the session use it. If this server has a label, its random server_uuid is masked to a
        stable value so GTID sets read the same on every run. A server started with --super-read-only=ON (a replica)
        blocks even root's ALTER USER, so super_read_only is turned off for the one password statement and restored.
        The statement runs with sql_log_bin off, for two reasons: the binary log a lesson decodes then holds only the
        lesson's own statements, and the replica keeps its own root password (each server of a run has a different one,
        so a replicated ALTER USER would lock the session out of the replica)."""
        sro = self.query("SELECT @@global.super_read_only") == "1"
        if sro:
            self.query("SET GLOBAL super_read_only=OFF")
        self.silent("SET SESSION sql_log_bin = 0;\n"
                    f"ALTER USER 'root'@'localhost' IDENTIFIED BY '{self.password}'")
        self._write_client_cnf(self.password)
        if sro:
            self.query("SET GLOBAL super_read_only=ON")
        if self.label:
            self.uuid = self.query("SELECT @@server_uuid")
            self.mask(re.escape(self.uuid), f"<{self.label}-uuid>")
        return self

    def stop(self, quiet=False):
        """A clean shutdown (mysqladmin shutdown), then waits until the process has ended."""
        code, out = self._call([self.ADMIN, f"--defaults-file={self.client_cnf}", "shutdown"])
        if code != 0 and not quiet:
            sys.exit(f"mysqladmin shutdown failed: {out}")
        try:
            self.proc.wait(timeout=120)
        except subprocess.TimeoutExpired:
            self.proc.terminate()
            self.proc.wait(timeout=60)
        self.proc = None

    def restart(self, extra=(), defaults_file=None, show=False):
        self.stop()
        self.start(extra=extra, defaults_file=defaults_file, show=show)

    # ---- running things --------------------------------------------------------------------------------------------
    def _call(self, args, stdin=None, merge=True, clean=True):
        """Runs one of MySQL's programs in the run folder: (status, cleaned output). merge=False keeps only what it
        wrote to standard output (its warnings on standard error are dropped). clean=False returns the output exactly
        as the program wrote it, for a value the session feeds back into SQL (a GTID set, a file name): a masked value
        is text for the record only and MySQL would reject it."""
        args = [str(a) for a in args]
        done = subprocess.run([str(self.bin / args[0]), *args[1:]], stdout=subprocess.PIPE,
                              stderr=subprocess.STDOUT if merge else subprocess.DEVNULL,
                              text=True, cwd=self.base, env=self.env, input=stdin)
        return done.returncode, (self.clean(done.stdout) if clean else done.stdout)

    def output(self, *args, connect=False, status=0):
        """Runs one of MySQL's programs without printing anything and returns its standard output, for example the
        SQL that mysql_tzinfo_to_sql writes, which the session saves with write_file(..., show=False) and runs next."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run, merge=False)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def clean(self, text):
        """The substitutions of the docstring. A table of the client whose cells change is drawn again with the
        new column widths, so its borders still line up."""
        lines, out, i = text.split("\n"), [], 0
        while i < len(lines):
            if lines[i].startswith("+-"):
                j = i
                while j < len(lines) and lines[j][:1] in ("+", "|"):
                    j += 1
                block = lines[i:j]
                done = [self._clean(l) for l in block]
                out.extend(done if done == block else _redraw(block, self._clean))
                i = j
            else:
                out.append(self._clean(lines[i]))
                i += 1
        return "\n".join(out)

    def _clean(self, text):
        text = text.replace(str(self.bin) + "/", "").replace(str(self.basedir), "<basedir>")
        text = text.replace(str(self.base) + "/", "./").replace(str(self.base), ".")
        text = re.sub(r" \(\d+\.\d{2,3} sec\)", "", text)  # 8.4 prints 2 decimals, 9.7 and 26.7 print 3
        text = re.sub(r'"Timestamp": \d+', '"Timestamp": <when it was set>', text)
        text = re.sub(rf"\b{re.escape(HOST)}\b", "lab-host", text)
        text = re.sub(rf"\b{re.escape(OS_USER)}\b", "learner", text)
        text = re.sub(rf"(?<![\d.]){self.port}(?![\d.])", self.port_label or "<port>", text)
        for pattern, replacement in self.masks:
            text = pattern.sub(replacement, text)
        return text

    def log(self, pattern=None):
        """The error log without timestamps and thread numbers, optionally only the lines matching a pattern."""
        text = self.clean(self.LOG_RE.sub("", (self.base / "mysqld.err").read_text()))
        if pattern:
            text = "\n".join(l for l in text.splitlines() if re.search(pattern, l))
        return text

    def show_log(self, pattern, note=None):
        """Prints the error-log lines that match a pattern, as grep would."""
        print(f"$ grep -E {_arg(pattern)} mysqld.err" + (f"   # {note}" if note else ""))
        print(self.log(pattern))
        print()

    def query(self, sql, db=None):
        """One value from one query, unprinted and UNMASKED (the session uses it to wait or to decide, and often
        feeds it back into SQL: a masked value would be a syntax error). Use gtid_executed() or clean() for a value
        that goes into the record."""
        args = [self.CLIENT, f"--defaults-file={self.client_cnf}", "-N", "-B", *([db] if db else [])]
        code, out = self._call(args, stdin=sql, clean=False)
        if code != 0:
            sys.exit(f"the query failed: {self.clean(out)}")
        return out.strip()

    def silent(self, sql):
        """Runs a statement without printing it (used only for the run's password)."""
        self.query(sql)

    def mysql(self, script, errors=0, status=None, db=None, flags=("-t", "-vvv", "--skip-comments"), note=None, force=None):
        """Runs a client script from standard input and prints the command and its output. `errors` is how many
        ERROR lines the script must print; with errors > 0 the client runs on past an error (--force) and then ends
        with status 0, without --force it stops at the first error with status 1. The default flags print each
        statement (-vvv) as a table (-t) without the script's comments (--skip-comments)."""
        force = errors > 0 if force is None else force
        shown = [self.CLIENT, *flags, *(["--force"] if force else []), *([db] if db else [])]
        args = [self.CLIENT, f"--defaults-file={self.client_cnf}", *shown[1:]]
        code, out = self._call(args, stdin=(self.base / script).read_text())
        if note:
            print(f"-- {note}")
        print("$ " + " ".join(_arg(a) for a in shown) + f" < {script}")
        if out.strip():
            print(out.rstrip("\n"))
        print()
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        expected_status = (1 if errors and not force else 0) if status is None else status
        if found != errors or code != expected_status:
            sys.exit(f"{script}: {found} errors and status {code}; the session expects {errors} and {expected_status}")
        return out

    def shell(self, *args, status=0, shown=None, connect=False):
        """Runs one of MySQL's programs and prints the command and its output. connect=True gives a client program
        (mysqladmin, mysqlshow, mysql) the run's client option file, which the printed command leaves out."""
        args = [str(a) for a in args]
        run = [args[0], f"--defaults-file={self.client_cnf}", *args[1:]] if connect else args
        code, out = self._call(run)
        _show(shown or args, out)
        if code != status:
            sys.exit(f"{args[0]} ended with status {code}, not {status}")
        return out

    def sql(self, statement, flags=("-t",), errors=0, db=None, note=None):
        """Runs one statement with mysql -e and prints the command and its output."""
        shown = [self.CLIENT, *flags, *([db] if db else []), "-e", statement]
        code, out = self._call([self.CLIENT, f"--defaults-file={self.client_cnf}", *shown[1:]])
        if note:
            print(f"-- {note}")
        _show(shown, out)
        found = sum(1 for line in out.splitlines() if line.startswith("ERROR "))
        if found != errors or (code != 0) != (errors > 0):
            sys.exit(f"{statement}: {found} errors and status {code}; the session expects {errors}")
        return out

    def write_file(self, path, text, show=True):
        """Writes a text file into the run folder (an option file, a script) and prints it, as cat would."""
        (self.base / path).write_text(textwrap.dedent(text).lstrip("\n"))
        if show:
            self.show_file(path)

    def show_file(self, path):
        print(f"$ cat {path}")
        print(self.clean((self.base / path).read_text()).rstrip("\n"))
        print()

    def listing(self, folder, prefix=""):
        """Prints the names in a folder of the run (sorted, folders with a slash), as `ls -p` would list them."""
        path = self.base / folder
        names = sorted(n + ("/" if (path / n).is_dir() else "") for n in os.listdir(path) if n.startswith(prefix))
        print(f"$ ls -p {folder}" + (f" | grep '^{prefix}'" if prefix else ""))
        print(textwrap.fill("  ".join(names), width=78, break_on_hyphens=False))
        print()
        return names


    # ---- replication (two or three servers on 127.0.0.1) -----------------------------------------------------------
    def gtid_executed(self, raw=False):
        """This server's @@global.gtid_executed as one line (no newlines). The value is masked for the record (the
        server_uuid reads as <label-uuid>); raw=True gives the real set, which is what MySQL accepts in a statement."""
        value = self.query("SELECT @@global.gtid_executed").replace("\n", "")
        return value if raw else self.clean(value)

    def track(self, other):
        """Teach this server's record to mask another server's port and server_uuid, so a statement that names the
        other server (CHANGE REPLICATION SOURCE TO, SHOW REPLICA STATUS) reads the same on every run."""
        if other.port:
            self.mask(rf"(?<![\d.]){other.port}(?![\d.])", other.port_label or "<port>")
        if other.uuid:
            self.mask(re.escape(other.uuid), f"<{other.label}-uuid>")
        self.mask(re.escape(str(other.base)), ".")

    def change_source(self, source, user="repl", password="replpass", auto_position=True, ssl=False,
                      verify_ca=False, channel=None, delay=None, extra=(), show=True, run=True):
        """Builds and (run=True) runs CHANGE REPLICATION SOURCE TO on this replica, pointing at `source`. With
        verify_ca the source's auto-generated ca.pem is copied into the replica's data directory and the replica
        verifies the server certificate's issuer chains to it (OpenSSL's VERIFY_CA mode).
        A relative SOURCE_SSL_CA is resolved relative to the replica's data directory, so the file is copied there,
        not into the run folder.
        VERIFY_IDENTITY (SOURCE_SSL_VERIFY_SERVER_CERT=1, which also checks the server certificate's SAN against the
        host) is not wired up here: MySQL's --initialize-insecure auto-generates self-signed certificates whose
        subject is "MySQL_Server_<version>_Auto_Generated_Server_Certificate" and whose SAN does not include
        127.0.0.1, so VERIFY_IDENTITY would fail with "certificate verify failed". A production setup issues server
        certificates whose SAN includes the host and then enables SOURCE_SSL_VERIFY_SERVER_CERT=1."""
        self.track(source)
        opts = ["SOURCE_HOST='127.0.0.1'", f"SOURCE_PORT={source.port}", f"SOURCE_USER='{user}'",
                f"SOURCE_PASSWORD='{password}'"]
        if auto_position:
            opts.append("SOURCE_AUTO_POSITION=1")
        if ssl:
            opts.append("SOURCE_SSL=1")
        if verify_ca:
            shutil.copy(source.base / "data" / "ca.pem", self.base / "data" / "source-ca.pem")
            opts.append("SOURCE_SSL_CA='source-ca.pem'")
        if delay is not None:
            opts.append(f"SOURCE_DELAY={delay}")
        opts += [str(e) for e in extra]
        stmt = "CHANGE REPLICATION SOURCE TO\n  " + ",\n  ".join(opts)
        if channel:
            stmt += f"\n  FOR CHANNEL '{channel}'"
        stmt += ";"
        self.write_file("_change_source.sql", stmt + "\n", show=False)
        if run:
            if show:
                self.mysql("_change_source.sql")
            else:
                self.query(stmt)
        return stmt

    def start_replica(self, user=None, password=None, channel=None, show=True):
        """START REPLICA, optionally with USER/PASSWORD given at start time (not stored in the metadata repository)."""
        parts = ["START REPLICA"]
        if user:
            parts.append(f"USER='{user}' PASSWORD='{password}'")
        if channel:
            parts.append(f"FOR CHANNEL '{channel}'")
        stmt = " ".join(parts) + ";"
        if show:
            self.write_file("_start_replica.sql", stmt + "\n", show=False)
            self.mysql("_start_replica.sql")
        else:
            self.query(stmt)

    def stop_replica(self, channel=None, show=False):
        stmt = "STOP REPLICA" + (f" FOR CHANNEL '{channel}'" if channel else "")
        if show:
            self.sql(stmt)
        else:
            self.query(stmt)

    def wait_replica_running(self, channel=None, timeout=30):
        """Waits until both replication threads report ON in performance_schema (deterministic, no fixed sleep)."""
        where = f" WHERE CHANNEL_NAME='{channel}'" if channel else ""
        for _ in range(timeout * 10):
            io = self.query(f"SELECT SERVICE_STATE FROM performance_schema.replication_connection_status{where}")
            applier = self.query(f"SELECT SERVICE_STATE FROM performance_schema.replication_applier_status{where}")
            if io == "ON" and applier == "ON":
                return
            time.sleep(0.1)
        print(self.sql("SHOW REPLICA STATUS", flags=("-E",)))
        sys.exit("the replica did not reach the running state in time")

    def wait_caught_up(self, source, timeout=30):
        """Blocks until this replica has executed every transaction the source had committed when called."""
        gset = source.gtid_executed(raw=True)
        if not gset:
            return
        result = self.query(f"SELECT WAIT_FOR_EXECUTED_GTID_SET('{gset}', {timeout})")
        if result != "0":
            sys.exit(f"the replica did not catch up (WAIT_FOR_EXECUTED_GTID_SET returned {result})")

    def replica_status(self, fields, channel=None, note=None, label="SHOW REPLICA STATUS"):
        """Prints only the named SHOW REPLICA STATUS fields, in the order given, vertically (phone-friendly)."""
        stmt = "SHOW REPLICA STATUS" + (f" FOR CHANNEL '{channel}'" if channel else "")
        code, out = self._call([self.CLIENT, f"--defaults-file={self.client_cnf}", "-E", "-e", stmt])
        if code != 0:
            sys.exit(f"SHOW REPLICA STATUS failed: {out}")
        values = {}
        for line in out.splitlines():
            m = re.match(r"\s*([A-Za-z_]+):\s?(.*)$", line)
            if m:
                values[m.group(1)] = m.group(2)
        if note:
            print(f"-- {note}")
        print(f"{self.CLIENT}> {label}\\G")
        for field in fields:
            if field in values:
                print(f"  {field}: {values[field]}")
        print()
        return values


def _redraw(block, clean):
    """Draws a client table again after its cells were cleaned. The columns come from the first border line; the row
    after it is the header (left-aligned); a column whose data cells were padded on the left (numbers) stays
    right-aligned (an empty cell says nothing about its column)."""
    cuts = [k for k, ch in enumerate(block[0]) if ch == "+"]
    rows = [None if line.startswith("+") else [line[x + 1:y] for x, y in zip(cuts, cuts[1:])] for line in block]
    header = next(k for k, r in enumerate(rows) if r is not None)
    ncol = len(cuts) - 1
    right = [any(r is not None and k != header and r[c].strip() and r[c].startswith("  ") for k, r in enumerate(rows))
             for c in range(ncol)]
    cells = [None if r is None else [clean(x.strip()) for x in r] for r in rows]
    widths = [max(len(r[c]) for r in cells if r is not None) for c in range(ncol)]
    drawn = []
    for k, r in enumerate(cells):
        if r is None:
            drawn.append("+" + "+".join("-" * (w + 2) for w in widths) + "+")
        else:
            drawn.append("|" + "|".join(" " + (t.rjust(w) if right[c] and k != header else t.ljust(w)) + " "
                                        for c, (t, w) in enumerate(zip(r, widths))) + "|")
    return drawn


def _arg(a):
    """A command-line argument as a reader would type it: quotes only where the shell needs them."""
    a = str(a)
    if re.fullmatch(r"[\w@%+=:,./-]+", a):
        return a
    if not re.search(r"[\"$`\\!]", a):
        return f'"{a}"'
    return shlex.quote(a)


def _show(args, out):
    print("$ " + " ".join(_arg(a) for a in args))
    if out.strip():
        print(out.rstrip("\n"))
    print()


class MariaDB(Server):
    """One throwaway MariaDB server (12.3.3 by default, or version="11.8" for 11.8.9), used as Server is.

    What differs from MySQL, and how the class handles it: mariadb-install-db makes the data directory, and root has no
    password until login() gives it the run's (install-db's own messages name this computer's folders and are not
    printed); the programs are mariadbd, mariadb and mariadb-admin (the mysql* names are links that MariaDB calls
    deprecated); there is no X Protocol, no super_read_only and no server_uuid; the error log's time stamps have
    another shape; and the server is told where its plugins, messages and character sets are, because the Homebrew
    build would otherwise look for them under /opt/homebrew. Its socket is mariadb.sock in the data directory (a
    relative path: MariaDB refuses a socket path longer than 103 characters), and clients connect over 127.0.0.1."""

    SERVER, CLIENT, ADMIN = "mariadbd", "mariadb", "mariadb-admin"
    LOG_RE = MARIADB_LOG_PREFIX

    def __init__(self, version="12.3", options=(), masks=(), label=None, port_label=None):
        super().__init__(options=options, basedir=toolchain("mariadb-12.3.3") / "opt" / f"mariadb@{version}",
                         masks=masks, label=label, port_label=port_label)
        self.version = version

    def server_options(self):
        b, home = self.base, self.basedir
        return ["--no-defaults", f"--basedir={home}", f"--datadir={b / 'data'}", f"--tmpdir={b / 'tmp'}",
                f"--log-error={b / 'mysqld.err'}", f"--pid-file={b / 'mysqld.pid'}",
                f"--secure-file-priv={b / 'files'}", f"--plugin-dir={home / 'lib' / 'plugin'}",
                f"--lc-messages-dir={home / 'share' / 'mysql'}",
                f"--character-sets-dir={home / 'share' / 'mysql' / 'charsets'}"]

    def listener_options(self):
        return ["--socket=mariadb.sock"]

    def initialize(self, show=False):
        """mariadb-install-db: a new data directory whose root@localhost account has no password yet."""
        args = ["mariadb-install-db", *self.server_options(), "--auth-root-authentication-method=normal",
                "--skip-test-db", f"--user={OS_USER}"]
        code, out = self._call(args)
        if show:
            _show(["mariadb-install-db", "--no-defaults", "--datadir=./data", "--auth-root-authentication-method=normal",
                   "--skip-test-db"], "")
        if code != 0:
            print(out)
            sys.exit(f"mariadb-install-db ended with status {code}")

    def login(self):
        """Gives root the run's password (never printed), with sql_log_bin off for that one statement, as
        Server.login() does and for the same reasons; a labelled server masks nothing here (MariaDB has no server_uuid:
        its GTIDs name the domain and the server_id instead)."""
        self.silent("SET SESSION sql_log_bin = 0;\n"
                    f"ALTER USER 'root'@'localhost' IDENTIFIED BY '{self.password}'")
        self._write_client_cnf(self.password)
        return self

    # ---- MariaDB GTIDs and replication (two or three servers on 127.0.0.1) -------------------------------------------
    def gtid_pos(self, which="gtid_binlog_pos"):
        """@@global.gtid_binlog_pos (what this server wrote to its binary log), gtid_slave_pos (what it applied as a
        replica) or gtid_current_pos, as one line."""
        return self.query(f"SELECT @@global.{which}").replace("\n", "")

    def change_master(self, source, user="repl", password="replpass", use_gtid="slave_pos", extra=(), show=True,
                      connection=None):
        """Builds and runs MariaDB's change-source statement on this replica, pointing at `source` on 127.0.0.1.
        MariaDB keeps the statement's legacy name (in the code below; the lessons say source and replica in prose).
        use_gtid: slave_pos, current_pos, or None for file-and-position replication (then give MASTER_LOG_FILE and
        MASTER_LOG_POS in extra). connection names a multi-source connection."""
        self.track(source)
        opts = ["MASTER_HOST='127.0.0.1'", f"MASTER_PORT={source.port}", f"MASTER_USER='{user}'",
                f"MASTER_PASSWORD='{password}'"]
        if use_gtid:
            opts.append(f"MASTER_USE_GTID={use_gtid}")
        opts += [str(e) for e in extra]
        head = "CHANGE MASTER " + (f"'{connection}' " if connection else "") + "TO"  # learn-lint-allow inclusive: MariaDB's statement name
        stmt = head + "\n  " + ",\n  ".join(opts) + ";"
        self.write_file("_change_source.sql", stmt + "\n", show=False)
        if show:
            self.mysql("_change_source.sql")
        else:
            self.query(stmt)
        return stmt

    def wait_replica_running(self, channel=None, timeout=30):
        """Waits until both replication threads say Yes in SHOW REPLICA STATUS (performance_schema is off by default
        on MariaDB, so its replication tables are not used)."""
        stmt = "SHOW REPLICA STATUS" if not channel else f"SHOW REPLICA '{channel}' STATUS"
        for _ in range(timeout * 10):
            code, out = self._call([self.CLIENT, f"--defaults-file={self.client_cnf}", "-E", "-e", stmt], clean=False)
            io = re.search(r"(?:Slave|Replica)_IO_Running: (\w+)", out)  # learn-lint-allow inclusive: MariaDB's own column names (this line and the next)
            sql_ = re.search(r"(?:Slave|Replica)_SQL_Running: (\w+)", out)
            if io and sql_ and io.group(1) == "Yes" and sql_.group(1) == "Yes":
                return
            time.sleep(0.1)
        print(self.clean(out))
        sys.exit("the replica did not reach the running state in time")

    def wait_caught_up(self, source, timeout=30):
        """Blocks until this replica has applied everything the source had written to its binary log when called
        (MASTER_GTID_WAIT on the source's gtid_binlog_pos)."""
        pos = source.gtid_pos("gtid_binlog_pos")
        if not pos:
            return
        result = self.query(f"SELECT MASTER_GTID_WAIT('{pos}', {timeout})")
        if result != "0":
            sys.exit(f"the replica did not catch up (MASTER_GTID_WAIT returned {result})")

inject_session.py

"""Parameters, not strings: the same hostile value put into two tiny apps.

The weak app (inject_bad.py) builds the WHERE clause by appending the name to a SQL
string, so a value with a quote in it closes the string literal early and the rest is
read as SQL. The fix (inject_safe.py) passes the value as a parameter: PyMySQL escapes
it into one quoted string literal, so the server compares it as a VARCHAR whatever its
contents.

The second half of the session is a check: a function imported from each module must
match zero rows for the hostile value and exactly one row for 'asha'. It passes against
inject_safe and fails against inject_bad with status 1 (the OWASP SQL Injection
Prevention Cheat Sheet's defensive framing: catch the weak version with a check that
also proves the fix works)."""

from mysql_session import Server

HOSTILE = "' OR '1'='1"

with Server() as my:
    my.up()
    my.sql("CREATE DATABASE shop")
    my.sql("CREATE TABLE shop.app_user (id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(40) NOT NULL)")
    my.sql("INSERT INTO shop.app_user (name) VALUES ('asha'), ('ravi'), ('kiran')")

    # One hostile call against each program.
    print(f"-- hostile input: {HOSTILE!r}")
    my.app("python3", "inject_bad.py", HOSTILE,
           mask=((r"rows returned: 3", "rows returned: 3  <- WRONG: every row came back for a name nobody has"),))
    my.app("python3", "inject_safe.py", HOSTILE,
           mask=((r"rows returned: 0", "rows returned: 0  <- the parameter stays a value"),))

    # The check: PASS on the fix, FAIL on the weak sample.
    my.app("python3", "check_find_user.py", "inject_safe", status=0)
    my.app("python3", "check_find_user.py", "inject_bad", status=1)

$ mysql -t -e "CREATE DATABASE shop"

$ mysql -t -e "CREATE TABLE shop.app_user (id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(40) NOT NULL)"

$ mysql -t -e "INSERT INTO shop.app_user (name) VALUES ('asha'), ('ravi'), ('kiran')"

-- hostile input: "' OR '1'='1"
$ python3 inject_bad.py "' OR '1'='1"
rows returned: 3  <- WRONG: every row came back for a name nobody has

$ python3 inject_safe.py "' OR '1'='1"
rows returned: 0  <- the parameter stays a value

$ python3 check_find_user.py inject_safe
PASS [inject_safe]: hostile value matched 0 rows and 'asha' matched 1 row

$ python3 check_find_user.py inject_bad
FAIL [inject_bad]: a hostile value matched 3 rows (expected 0)

inject_bad.py

"""The weak sample. It builds SQL from a string, so a value containing a quote can close
the quoted literal and change the shape of the query. Fix it by passing the value as a
parameter (see inject_safe.py)."""

import os

import pymysql


def connect():
    return pymysql.connect(
        host=os.environ["LAB_HOST"], port=int(os.environ["LAB_PORT"]),
        user=os.environ["LAB_USER"], password=os.environ["LAB_PASSWORD"],
        database="shop",
        ssl={"ca": os.environ["LAB_CA"], "check_hostname": False},
    )


def find_user(name):
    with connect() as conn, conn.cursor() as cur:
        sql = f"SELECT id, name FROM app_user WHERE name = '{name}'"
        cur.execute(sql)
        return cur.fetchall()


if __name__ == "__main__":
    import sys
    rows = find_user(sys.argv[1])
    print(f"rows returned: {len(rows)}")

inject_safe.py

"""The fix. The value is passed as a parameter, and PyMySQL escapes it into one quoted string literal, so it
never becomes SQL: a hostile value matches no row rather than changing the query."""

import os

import pymysql


def connect():
    return pymysql.connect(
        host=os.environ["LAB_HOST"], port=int(os.environ["LAB_PORT"]),
        user=os.environ["LAB_USER"], password=os.environ["LAB_PASSWORD"],
        database="shop",
        ssl={"ca": os.environ["LAB_CA"], "check_hostname": False},
    )


def find_user(name):
    with connect() as conn, conn.cursor() as cur:
        cur.execute("SELECT id, name FROM app_user WHERE name = %s", (name,))
        return cur.fetchall()


if __name__ == "__main__":
    import sys
    rows = find_user(sys.argv[1])
    print(f"rows returned: {len(rows)}")

check_find_user.py

"""A check that two conditions hold against a find_user(name) function imported from
sys.argv[1]: a value crafted to end a SQL string literal must match no row, and the
plain user name 'asha' must match exactly one. On inject_safe it passes; on inject_bad
it fails with the exit status the pre-commit runner reads."""

import importlib
import sys

mod = importlib.import_module(sys.argv[1])

HOSTILE = "' OR '1'='1"
PLAIN = "asha"

got_hostile = len(mod.find_user(HOSTILE))
got_plain = len(mod.find_user(PLAIN))

problems = []
if got_hostile != 0:
    problems.append(f"a hostile value matched {got_hostile} rows (expected 0)")
if got_plain != 1:
    problems.append(f"'asha' matched {got_plain} rows (expected 1)")

if problems:
    print(f"FAIL [{sys.argv[1]}]: " + "; ".join(problems))
    sys.exit(1)
print(f"PASS [{sys.argv[1]}]: hostile value matched 0 rows and 'asha' matched 1 row")