MySQL & MariaDB in Depth Module 12 – Security and access control
Privileges, roles and least privilege in MySQL
Split MySQL privileges into static and dynamic, group them into roles, and keep admin accounts safe with partial revokes, SYSTEM_USER and definer views.
Free core, Pro depth: 7 of 20 minutes free
Free: Static privileges live in columns, dynamic ones in rows · Five accounts for five jobs, with only the grants of their work · A role groups privileges; the account holds the role. With a pass: 2 in-depth parts and 3 interview answers, about 13 minutes.
What you will learn
- Distinguish static and dynamic privileges and name the two tables that store them
- Design five accounts for five jobs on one server, each with least privilege
- Configure roles for accounts, set a default role and read SHOW GRANTS ... USING
- Limit a global grant with a partial revoke on one schema and protect admins with SYSTEM_USER
- Query information_schema and the mysql tables to find who can do what, and prevent orphaned definer views
Before you start
What you’ll learn · Before you start
What you will learn
- Distinguish static and dynamic privileges and name the two tables that store them
- Design five accounts for five jobs on one server, each with least privilege
- Configure roles for accounts, set a default role and read SHOW GRANTS ... USING
- Limit a global grant with a partial revoke on one schema and protect admins with SYSTEM_USER
- Query information_schema and the mysql tables to find who can do what, and prevent orphaned definer views
Before you start
On this page
MySQL protects data through privileges: fine-grained rights that attach to accounts. A new 8.4.11 server has
thirty-one static privileges as _priv columns of mysql.user, and root holds forty dynamic ones as rows of
mysql.global_grants, most registered by the server at startup. Least privilege in MySQL means each account
carries only the grants its one job needs, scoped to the hosts it logs in from. Five accounts, from checkout
to backup and the replication connection, each hold the privileges of one job.
A role: A named account in mysql.user with account_locked=Y and no password that groups privileges together; grant the role to accounts, activate it per session with SET ROLE or at login with a default role. is a locked account with no password that groups privileges together; granting the role
to a human or service keeps the account itself unchanged when the job changes. Three more features make
least privilege: The design rule that every account carries only the privileges it needs for its job, scoped to the hosts it logs in from, with wider grants replaced by schema- or table-level ones and risky dynamic privileges held only by named accounts. hold up on a real server: a
partial revoke: A REVOKE at the schema level that narrows one account's global grant, so that, for example, a global SELECT leaves out one schema. keeps a global SELECT out of one schema, the
SYSTEM_USER privilege: A dynamic privilege that marks an account as a system account, which an ordinary administrator without SYSTEM_USER cannot modify, so the holders of SYSTEM_USER form the server's smaller administrator set. privilege shields an admin account from an ordinary
administrator, and a SQL SECURITY DEFINER view exposes aggregates without the base rows.
Recorded runs of all three follow, with queries that find over-broad grants.
Static privileges live in columns, dynamic ones in rows
The first table of privileges_session.py counts the
static privilege: A privilege built into the MySQL server and always available to grant; its global and schema grants are Y/N columns of mysql.user and mysql.db, its table and column grants sets in mysql.tables_priv and mysql.columns_priv. MySQL 8.4 holds at the global level: thirty-one, one _priv
column each in mysql.user. The second counts the forty
dynamic privilege: A privilege registered at run time, most by the server at startup and the rest by the component or plugin that defines it; held per account as a row of mysql.global_grants and granted only at the global level, ON *.*. rows that root@localhost holds in mysql.global_grants,
each a named right such as BACKUP_ADMIN, FLUSH_TABLES or SYSTEM_USER. The split matters because the two
stores answer different audit queries and follow different rules for where a grant may apply, which the paragraph
after the figure makes precise.
privileges_session.py
"""Static and dynamic privileges on a throwaway MySQL 8.4.11 server.
privileges.sql counts the two kinds on a new server, then grants a dynamic privilege at a schema and at the global
level, grants one that no loaded plugin has registered, and grants the deprecated SUPER. The client runs on past the
two errors (--force), so every statement of the script prints what it did.
"""
import os
os.chdir(os.path.dirname(__file__) or ".")
from mysql_session import Server
with Server() as my:
my.up()
my.mysql("privileges.sql", errors=2)Output
$ mysql -t -vvv --skip-comments --force < privileges.sql
--------------
SELECT COUNT(*) AS static_global_columns
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = 'mysql' AND TABLE_NAME = 'user' AND COLUMN_NAME LIKE '%\_priv'
--------------
+-----------------------+
| static_global_columns |
+-----------------------+
| 31 |
+-----------------------+
1 row in set
--------------
SELECT COUNT(*) AS dynamic_held_by_root
FROM mysql.global_grants
WHERE USER = 'root' AND HOST = 'localhost'
--------------
+----------------------+
| dynamic_held_by_root |
+----------------------+
| 40 |
+----------------------+
1 row in set
--------------
SELECT PRIV
FROM mysql.global_grants
WHERE USER = 'root' AND HOST = 'localhost'
AND PRIV IN ('BACKUP_ADMIN', 'CONNECTION_ADMIN', 'FLUSH_TABLES', 'REPLICATION_APPLIER',
'SYSTEM_USER', 'SYSTEM_VARIABLES_ADMIN')
ORDER BY PRIV
--------------
+------------------------+
| PRIV |
+------------------------+
| BACKUP_ADMIN |
| CONNECTION_ADMIN |
| FLUSH_TABLES |
| REPLICATION_APPLIER |
| SYSTEM_USER |
| SYSTEM_VARIABLES_ADMIN |
+------------------------+
6 rows in set
--------------
CREATE DATABASE shop
--------------
Query OK, 1 row affected
--------------
CREATE USER 'backup_bot'@'localhost' ACCOUNT LOCK
--------------
Query OK, 0 rows affected
--------------
GRANT BACKUP_ADMIN ON shop.* TO 'backup_bot'@'localhost'
--------------
ERROR 3619 (HY000) at line 25: Illegal privilege level specified for BACKUP_ADMIN
--------------
GRANT BACKUP_ADMIN ON *.* TO 'backup_bot'@'localhost'
--------------
Query OK, 0 rows affected
--------------
GRANT FIREWALL_ADMIN ON *.* TO 'backup_bot'@'localhost'
--------------
ERROR 1149 (42000) at line 29: You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use
--------------
GRANT SUPER ON *.* TO 'backup_bot'@'localhost'
--------------
Query OK, 0 rows affected, 1 warning
--------------
SHOW WARNINGS
--------------
+---------+------+----------------------------------------------+
| Level | Code | Message |
+---------+------+----------------------------------------------+
| Warning | 1287 | The SUPER privilege identifier is deprecated |
+---------+------+----------------------------------------------+
1 row in set
--------------
SELECT PRIV FROM mysql.global_grants WHERE USER = 'backup_bot'
--------------
+--------------+
| PRIV |
+--------------+
| BACKUP_ADMIN |
+--------------+
1 row in set
--------------
SELECT Super_priv FROM mysql.user WHERE User = 'backup_bot'
--------------
+------------+
| Super_priv |
+------------+
| Y |
+------------+
1 row in set
--------------
SHOW GRANTS FOR 'backup_bot'@'localhost'
--------------
+-------------------------------------------------------+
| Grants for backup_bot@localhost |
+-------------------------------------------------------+
| GRANT SUPER ON *.* TO `backup_bot`@`localhost` |
| GRANT BACKUP_ADMIN ON *.* TO `backup_bot`@`localhost` |
+-------------------------------------------------------+
2 rows in set
Bye
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 privileges_session.py
privileges.sql
-- Run as root on a new MySQL 8.4.11 server.
-- Static privileges are built into the server: each one of the global level has its own column in mysql.user.
SELECT COUNT(*) AS static_global_columns
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = 'mysql' AND TABLE_NAME = 'user' AND COLUMN_NAME LIKE '%\_priv';
-- Dynamic privileges are registered while the server runs: root holds each one as a row of mysql.global_grants.
SELECT COUNT(*) AS dynamic_held_by_root
FROM mysql.global_grants
WHERE USER = 'root' AND HOST = 'localhost';
SELECT PRIV
FROM mysql.global_grants
WHERE USER = 'root' AND HOST = 'localhost'
AND PRIV IN ('BACKUP_ADMIN', 'CONNECTION_ADMIN', 'FLUSH_TABLES', 'REPLICATION_APPLIER',
'SYSTEM_USER', 'SYSTEM_VARIABLES_ADMIN')
ORDER BY PRIV;
CREATE DATABASE shop;
-- A locked account: nobody logs in as it in this session, so it needs no password.
CREATE USER 'backup_bot'@'localhost' ACCOUNT LOCK;
-- A dynamic privilege exists only at the global level, ON *.*.
GRANT BACKUP_ADMIN ON shop.* TO 'backup_bot'@'localhost';
GRANT BACKUP_ADMIN ON *.* TO 'backup_bot'@'localhost';
-- FIREWALL_ADMIN belongs to the Enterprise firewall plugin, which this server has not loaded.
GRANT FIREWALL_ADMIN ON *.* TO 'backup_bot'@'localhost';
-- SUPER still works, and the server says what it thinks of it.
GRANT SUPER ON *.* TO 'backup_bot'@'localhost';
SHOW WARNINGS;
-- Where the two grants that worked were stored, and how SHOW GRANTS prints them.
SELECT PRIV FROM mysql.global_grants WHERE USER = 'backup_bot';
SELECT Super_priv FROM mysql.user WHERE User = 'backup_bot';
SHOW GRANTS FOR 'backup_bot'@'localhost';mysql_session.py
"""Throwaway MySQL 8.4.11 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, and tarball(name) finds one.
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 and the pinned MySQL installations, 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)
def tarball(name):
"""The folder of an unpacked official tarball, such as mysql-9.7.2, for Server(basedir=...). It is the folder
<name> in a folder called toolchains, looked for in every folder above the one 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)
candidates = [parent / "toolchains" / name]
if (parent / "tmp").is_dir():
candidates += sorted((parent / "tmp").glob(f"*/toolchains/{name}"))
for found in candidates:
if (found / "bin").is_dir():
return found
sys.exit(f"no toolchains folder above this one holds {name}")
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().
"""
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 / "mysqld").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")})
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 / "mysqld"), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
"--socket=mysql.sock", "--mysqlx=OFF", *self.options, *[str(e) for e in extra]]
if show:
shown = ["mysqld", opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
"--port=$PORT", "--socket=mysql.sock", "--mysqlx=OFF", *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(LOG_PREFIX.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 _ping(self):
code, _ = self._call(["mysqladmin", 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(["mysqladmin", 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(LOG_PREFIX.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 = ["mysql", 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 = ["mysql", *flags, *(["--force"] if force else []), *([db] if db else [])]
args = ["mysql", 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 = ["mysql", *flags, *([db] if db else []), "-e", statement]
code, out = self._call(["mysql", 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(["mysql", 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"mysql> {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()The drawing shows the whole scene; its 4 steps are listed below.
With the drawing focused, the left and right arrow keys step, Home and End go to the first and last step, and Space or K plays and pauses.
Every step as text
- Two stores, one SHOW. SHOW PRIVILEGES lists every static privilege and every dynamic one registered so far, each with the Context it applies to, such as Databases, Tables or Server Admin. Where a grant of each kind is kept differs.
- Static lives in columns. A static privilege granted globally or on a schema is a Y or N column of mysql.user or mysql.db; its table and column grants are sets in mysql.tables_priv and mysql.columns_priv. The recorded run counts 31 such columns at the global level.
- Dynamic lives in rows. A dynamic privilege is a row of mysql.global_grants, with USER, HOST and PRIV columns. The recorded run counts 40 rows for root@localhost on 8.4.11 (BACKUP_ADMIN, SYSTEM_USER, FLUSH_TABLES among them).
- Dynamic grants only at ON *.*. GRANT BACKUP_ADMIN ON shop.* returns ERROR 3619, because mysql.global_grants has no schema column. The same GRANT ON . is accepted and recorded as one row.
Two differences follow from the shape. A dynamic privilege only grants at the global level, so GRANT BACKUP_ADMIN ON shop.* TO 'backup_bot'@'localhost' returns ERROR 3619 and GRANT BACKUP_ADMIN ON *.* ... is accepted. A dynamic
privilege that nothing has registered cannot be granted at all: GRANT FIREWALL_ADMIN ON *.* fails as a plain
syntax error (ERROR 1149), because the firewall that would register it ships only in the Enterprise build. The deprecated SUPER still works and
leaves a warning on SHOW WARNINGS: The SUPER privilege identifier is deprecated.
Checkpoint
Check: Static versus dynamic privileges
Five accounts for five jobs, with only the grants of their work
The second recorded run gives five accounts five different jobs on one database. checkout reads and writes the
rows of shop; migrator adds those four grants plus CREATE, ALTER, DROP, INDEX and REFERENCES for
schema changes; reporting holds SELECT only; backup reads every schema (SELECT, SHOW VIEW, EVENT,
TRIGGER on *.*) and the dynamic FLUSH_TABLES, which mysqldump --single-transaction needs on a server with
GTIDs on; the replication connection
holds exactly one grant, scoped to the subnet replicas log in from and refusing a plain-text login with
A per-account rule that refuses any login that does not use TLS, set with REQUIRE SSL on CREATE USER or ALTER USER..
accounts_session.py
"""Five accounts with five jobs on a throwaway MySQL 8.4.11 server.
The server runs with GTIDs on, as a replication source does. accounts.sql gives each account the privileges of its
job; then each account tries something that belongs to another job. The backup account runs the dump it was set up
for, and the same dump once more after FLUSH_TABLES is taken away. Every password is made up for the run and kept in
its own client option file (<name>.cnf), which is never printed.
"""
import os
import secrets
os.chdir(os.path.dirname(__file__) or ".")
from mysql_session import Server
def login(my, name, host="localhost"):
"""Creates the account <name>@<host> with a random password, and <name>.cnf to connect as it."""
password = secrets.token_urlsafe(18)
my.silent(f"CREATE USER '{name}'@'{host}' IDENTIFIED BY '{password}'")
cnf = my.base / f"{name}.cnf"
cnf.write_text(f"[client]\nuser={name}\npassword={password}\nhost=127.0.0.1\nport={my.port}\nprotocol=TCP\n")
cnf.chmod(0o600)
def as_user(my, name, statement, fails=False):
"""Runs one statement as that account, and prints the command and what it printed."""
my.shell("mysql", f"--defaults-file={name}.cnf", "-t", "-e", statement, status=1 if fails else 0)
with Server(options=["--gtid-mode=ON", "--enforce-gtid-consistency=ON"], label="lab") as my:
my.up()
for name in ("checkout", "migrator", "reporting", "backup"):
login(my, name)
login(my, "repl", host="10.0.0.%")
my.mysql("accounts.sql", flags=("-N",))
print("# Each account stays inside its job.")
as_user(my, "migrator", "ALTER TABLE shop.orders ADD COLUMN note VARCHAR(80)")
as_user(my, "checkout", "ALTER TABLE shop.orders ADD COLUMN note2 VARCHAR(80)", fails=True)
as_user(my, "reporting", "UPDATE shop.orders SET amount = 0 WHERE id = 1", fails=True)
as_user(my, "backup", "DELETE FROM shop.orders", fails=True)
print("# The backup account's dump works, and stops without FLUSH_TABLES.")
dump = ["mysqldump", "--defaults-file=backup.cnf", "--single-transaction", "--no-tablespaces",
"--compact", "--no-create-info", "shop"]
my.shell(*dump)
my.sql("REVOKE FLUSH_TABLES ON *.* FROM 'backup'@'localhost'")
my.shell(*dump, status=2)Output
$ mysql -N < accounts.sql GRANT USAGE ON *.* TO `checkout`@`localhost` GRANT SELECT, INSERT, UPDATE, DELETE ON `shop`.* TO `checkout`@`localhost` GRANT USAGE ON *.* TO `migrator`@`localhost` GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, REFERENCES, INDEX, ALTER ON `shop`.* TO `migrator`@`localhost` GRANT USAGE ON *.* TO `reporting`@`localhost` GRANT SELECT ON `shop`.* TO `reporting`@`localhost` GRANT SELECT, SHOW VIEW, EVENT, TRIGGER ON *.* TO `backup`@`localhost` GRANT FLUSH_TABLES ON *.* TO `backup`@`localhost` GRANT REPLICATION SLAVE ON *.* TO `repl`@`10.0.0.%` # Each account stays inside its job. $ mysql --defaults-file=migrator.cnf -t -e "ALTER TABLE shop.orders ADD COLUMN note VARCHAR(80)" $ mysql --defaults-file=checkout.cnf -t -e "ALTER TABLE shop.orders ADD COLUMN note2 VARCHAR(80)" ERROR 1142 (42000) at line 1: ALTER command denied to user 'checkout'@'localhost' for table 'orders' $ mysql --defaults-file=reporting.cnf -t -e "UPDATE shop.orders SET amount = 0 WHERE id = 1" ERROR 1142 (42000) at line 1: UPDATE command denied to user 'reporting'@'localhost' for table 'orders' $ mysql --defaults-file=backup.cnf -t -e "DELETE FROM shop.orders" ERROR 1142 (42000) at line 1: DELETE command denied to user 'backup'@'localhost' for table 'orders' # The backup account's dump works, and stops without FLUSH_TABLES. $ mysqldump --defaults-file=backup.cnf --single-transaction --no-tablespaces --compact --no-create-info shop Warning: A partial dump from a server that has GTIDs will by default include the GTIDs of all transactions, even those that changed suppressed parts of the database. If you don't want to restore GTIDs, pass --set-gtid-purged=OFF. To make a complete dump, pass --all-databases --triggers --routines --events. SET @MYSQLDUMP_TEMP_LOG_BIN = @@SESSION.SQL_LOG_BIN; SET @@SESSION.SQL_LOG_BIN= 0; SET @@GLOBAL.GTID_PURGED=/*!80000 '+'*/ '<lab-uuid>:1-16'; INSERT INTO `orders` VALUES (1,'asha',499.00,NULL),(2,'ravi',1250.50,NULL); SET @@SESSION.SQL_LOG_BIN = @MYSQLDUMP_TEMP_LOG_BIN; $ mysql -t -e "REVOKE FLUSH_TABLES ON *.* FROM 'backup'@'localhost'" $ mysqldump --defaults-file=backup.cnf --single-transaction --no-tablespaces --compact --no-create-info shop mysqldump: Couldn't execute 'FLUSH /*!40101 LOCAL */ TABLES': Access denied; you need (at least one of) the RELOAD or FLUSH_TABLES privilege(s) for this operation (1227)
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 accounts_session.py
accounts.sql
-- Run as root. The five accounts already exist, each with a password of its own; this script gives each one its job.
CREATE DATABASE shop;
CREATE TABLE shop.orders (
id INT AUTO_INCREMENT PRIMARY KEY,
customer VARCHAR(40) NOT NULL,
amount DECIMAL(10,2) NOT NULL
);
INSERT INTO shop.orders (customer, amount) VALUES ('asha', 499.00), ('ravi', 1250.50);
-- 1. The application: the rows of its own schema, and nothing else.
GRANT SELECT, INSERT, UPDATE, DELETE ON shop.* TO 'checkout'@'localhost';
-- 2. The migrator: the schema's structure, plus the rows its backfills change.
GRANT CREATE, ALTER, DROP, INDEX, REFERENCES, SELECT, INSERT, UPDATE, DELETE ON shop.* TO 'migrator'@'localhost';
-- 3. Reporting: read only.
GRANT SELECT ON shop.* TO 'reporting'@'localhost';
-- 4. Logical backups with mysqldump --single-transaction --no-tablespaces --routines --events:
-- read everything, and FLUSH_TABLES (narrower than RELOAD) for the GTID position.
GRANT SELECT, SHOW VIEW, TRIGGER, EVENT ON *.* TO 'backup'@'localhost';
GRANT FLUSH_TABLES ON *.* TO 'backup'@'localhost';
-- 5. Replication: one privilege, from the replicas' subnet, over TLS only.
-- learn-lint-allow inclusive: MySQL 8.4 keeps this SQL name for the replication privilege
GRANT REPLICATION SLAVE ON *.* TO 'repl'@'10.0.0.%';
ALTER USER 'repl'@'10.0.0.%' REQUIRE SSL;
SHOW GRANTS FOR 'checkout'@'localhost';
SHOW GRANTS FOR 'migrator'@'localhost';
SHOW GRANTS FOR 'reporting'@'localhost';
SHOW GRANTS FOR 'backup'@'localhost';
SHOW GRANTS FOR 'repl'@'10.0.0.%';mysql_session.py
"""Throwaway MySQL 8.4.11 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, and tarball(name) finds one.
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 and the pinned MySQL installations, 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)
def tarball(name):
"""The folder of an unpacked official tarball, such as mysql-9.7.2, for Server(basedir=...). It is the folder
<name> in a folder called toolchains, looked for in every folder above the one 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)
candidates = [parent / "toolchains" / name]
if (parent / "tmp").is_dir():
candidates += sorted((parent / "tmp").glob(f"*/toolchains/{name}"))
for found in candidates:
if (found / "bin").is_dir():
return found
sys.exit(f"no toolchains folder above this one holds {name}")
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().
"""
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 / "mysqld").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")})
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 / "mysqld"), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
"--socket=mysql.sock", "--mysqlx=OFF", *self.options, *[str(e) for e in extra]]
if show:
shown = ["mysqld", opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
"--port=$PORT", "--socket=mysql.sock", "--mysqlx=OFF", *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(LOG_PREFIX.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 _ping(self):
code, _ = self._call(["mysqladmin", 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(["mysqladmin", 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(LOG_PREFIX.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 = ["mysql", 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 = ["mysql", *flags, *(["--force"] if force else []), *([db] if db else [])]
args = ["mysql", 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 = ["mysql", *flags, *([db] if db else []), "-e", statement]
code, out = self._call(["mysql", 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(["mysql", 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"mysql> {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()The record then logs in as each account to prove the fences. migrator adds a column; checkout tries the same
ALTER and the server returns ERROR 1142, ALTER command denied; reporting is denied UPDATE, backup is
denied DELETE. The backup account’s dump works; after REVOKE FLUSH_TABLES, mysqldump fails right at
FLUSH /*!40101 LOCAL */ TABLES with ERROR 1227.
Key idea
Give each account only the rights of its one job, scope its host to the network it logs in from, and set
REQUIRE SSL on any account that reaches the server over a shared network. That is what least privilege looks
like on a running MySQL server.
The replication account’s one grant is the replication privilege, which GRANT and SHOW GRANTS still spell
with its A SQL privilege MySQL 8.4 still identifies by its older name, inside a GRANT or SHOW GRANTS statement, even though the related statements have been renamed., although the
statements around it moved to the REPLICA and SOURCE wording.
Checkpoint
Check: Five accounts for five jobs
A role groups privileges; the account holds the role
roles.sql creates three roles, app_rw, app_ddl and app_read, and grants them SELECT, INSERT, UPDATE, DELETE on shop.*, CREATE, ALTER, DROP, INDEX, REFERENCES on shop.* and SELECT on shop.*. Each account then
takes the roles its job needs: checkout holds app_rw, migrator holds app_ddl and app_rw, reporting and
oncall hold app_read.
roles_session.py
"""Roles for the application, the migrator and reporting on a throwaway MySQL 8.4.11 server.
roles.sql creates three roles and grants them to four accounts. The session reads the grants with and without
SHOW GRANTS ... USING, logs in as reporting before and after it has a default role, turns its role on by hand with
SET ROLE, and logs in as oncall while activate_all_roles_on_login is ON. Every password is made up for the run and
kept in its own client option file (<name>.cnf), which is never printed.
"""
import os
import secrets
os.chdir(os.path.dirname(__file__) or ".")
from mysql_session import Server
def login(my, name):
"""Creates the account <name>@localhost with a random password, and <name>.cnf to connect as it."""
password = secrets.token_urlsafe(18)
my.silent(f"CREATE USER '{name}'@'localhost' IDENTIFIED BY '{password}'")
cnf = my.base / f"{name}.cnf"
cnf.write_text(f"[client]\nuser={name}\npassword={password}\nhost=127.0.0.1\nport={my.port}\nprotocol=TCP\n")
cnf.chmod(0o600)
def as_user(my, name, statement, fails=False):
"""Logs in as that account, runs the statements, and prints the command and what it printed."""
my.shell("mysql", f"--defaults-file={name}.cnf", "-t", "-e", statement, status=1 if fails else 0)
with Server() as my:
my.up()
for name in ("checkout", "migrator", "reporting", "oncall"):
login(my, name)
my.mysql("roles.sql", flags=("-t",))
print("# 1. SHOW GRANTS names the roles; USING also lists what they grant.")
my.sql("SHOW GRANTS FOR 'migrator'@'localhost'")
my.sql("SHOW GRANTS FOR 'migrator'@'localhost' USING 'app_ddl', 'app_rw'")
print("# 2. reporting holds app_read, but no role is active at login.")
as_user(my, "reporting", "SELECT CURRENT_ROLE(); SELECT COUNT(*) FROM shop.orders", fails=True)
print("# 3. SET ROLE turns it on for this session only.")
as_user(my, "reporting", "SET ROLE 'app_read'; SELECT CURRENT_ROLE(); SELECT COUNT(*) FROM shop.orders")
print("# 4. A default role is active from every login on, and grants nothing more than the role holds.")
my.sql("SET DEFAULT ROLE 'app_read' TO 'reporting'@'localhost'")
as_user(my, "reporting", "SELECT CURRENT_ROLE(); SELECT COUNT(*) FROM shop.orders; DELETE FROM shop.orders",
fails=True)
print("# 5. activate_all_roles_on_login turns on every granted role, for every account.")
my.sql("SET GLOBAL activate_all_roles_on_login = ON")
as_user(my, "oncall", "SELECT CURRENT_ROLE()")
my.sql("SET GLOBAL activate_all_roles_on_login = OFF")
print("# 6. A role is an account that nobody can log in as.")
my.sql("SELECT User, Host, account_locked FROM mysql.user WHERE User LIKE 'app%' ORDER BY User")Output
$ mysql -t < roles.sql # 1. SHOW GRANTS names the roles; USING also lists what they grant. $ mysql -t -e "SHOW GRANTS FOR 'migrator'@'localhost'" +------------------------------------------------------------+ | Grants for migrator@localhost | +------------------------------------------------------------+ | GRANT USAGE ON *.* TO `migrator`@`localhost` | | GRANT `app_ddl`@`%`,`app_rw`@`%` TO `migrator`@`localhost` | +------------------------------------------------------------+ $ mysql -t -e "SHOW GRANTS FOR 'migrator'@'localhost' USING 'app_ddl', 'app_rw'" +--------------------------------------------------------------------------------------------------------------------+ | Grants for migrator@localhost | +--------------------------------------------------------------------------------------------------------------------+ | GRANT USAGE ON *.* TO `migrator`@`localhost` | | GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, REFERENCES, INDEX, ALTER ON `shop`.* TO `migrator`@`localhost` | | GRANT `app_ddl`@`%`,`app_rw`@`%` TO `migrator`@`localhost` | +--------------------------------------------------------------------------------------------------------------------+ # 2. reporting holds app_read, but no role is active at login. $ mysql --defaults-file=reporting.cnf -t -e "SELECT CURRENT_ROLE(); SELECT COUNT(*) FROM shop.orders" +----------------+ | CURRENT_ROLE() | +----------------+ | NONE | +----------------+ ERROR 1142 (42000) at line 1: SELECT command denied to user 'reporting'@'localhost' for table 'orders' # 3. SET ROLE turns it on for this session only. $ mysql --defaults-file=reporting.cnf -t -e "SET ROLE 'app_read'; SELECT CURRENT_ROLE(); SELECT COUNT(*) FROM shop.orders" +----------------+ | CURRENT_ROLE() | +----------------+ | `app_read`@`%` | +----------------+ +----------+ | COUNT(*) | +----------+ | 2 | +----------+ # 4. A default role is active from every login on, and grants nothing more than the role holds. $ mysql -t -e "SET DEFAULT ROLE 'app_read' TO 'reporting'@'localhost'" $ mysql --defaults-file=reporting.cnf -t -e "SELECT CURRENT_ROLE(); SELECT COUNT(*) FROM shop.orders; DELETE FROM shop.orders" +----------------+ | CURRENT_ROLE() | +----------------+ | `app_read`@`%` | +----------------+ +----------+ | COUNT(*) | +----------+ | 2 | +----------+ ERROR 1142 (42000) at line 1: DELETE command denied to user 'reporting'@'localhost' for table 'orders' # 5. activate_all_roles_on_login turns on every granted role, for every account. $ mysql -t -e "SET GLOBAL activate_all_roles_on_login = ON" $ mysql --defaults-file=oncall.cnf -t -e "SELECT CURRENT_ROLE()" +----------------+ | CURRENT_ROLE() | +----------------+ | `app_read`@`%` | +----------------+ $ mysql -t -e "SET GLOBAL activate_all_roles_on_login = OFF" # 6. A role is an account that nobody can log in as. $ mysql -t -e "SELECT User, Host, account_locked FROM mysql.user WHERE User LIKE 'app%' ORDER BY User" +----------+------+----------------+ | User | Host | account_locked | +----------+------+----------------+ | app_ddl | % | Y | | app_read | % | Y | | app_rw | % | Y | +----------+------+----------------+
Recorded with Python 3.14.8 on macOS 26 arm64. To run it yourself: mise exec python@3.14.8 -- python3 roles_session.py
roles.sql
-- Run as root. The accounts checkout, migrator, reporting and oncall already exist, each with a password of its own.
CREATE DATABASE shop;
CREATE TABLE shop.orders (
id INT AUTO_INCREMENT PRIMARY KEY,
customer VARCHAR(40) NOT NULL,
amount DECIMAL(10,2) NOT NULL
);
INSERT INTO shop.orders (customer, amount) VALUES ('asha', 499.00), ('ravi', 1250.50);
-- One role per kind of job. A role is created locked and without a password.
CREATE ROLE 'app_rw', 'app_ddl', 'app_read';
GRANT SELECT, INSERT, UPDATE, DELETE ON shop.* TO 'app_rw';
GRANT CREATE, ALTER, DROP, INDEX, REFERENCES ON shop.* TO 'app_ddl';
GRANT SELECT ON shop.* TO 'app_read';
-- Accounts get roles, not privileges.
GRANT 'app_rw' TO 'checkout'@'localhost';
GRANT 'app_ddl', 'app_rw' TO 'migrator'@'localhost';
GRANT 'app_read' TO 'reporting'@'localhost', 'oncall'@'localhost';
-- checkout and migrator start every session with their roles active.
-- reporting and oncall get no default role, so their role stays inactive at login.
SET DEFAULT ROLE ALL TO 'checkout'@'localhost', 'migrator'@'localhost';mysql_session.py
"""Throwaway MySQL 8.4.11 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, and tarball(name) finds one.
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 and the pinned MySQL installations, 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)
def tarball(name):
"""The folder of an unpacked official tarball, such as mysql-9.7.2, for Server(basedir=...). It is the folder
<name> in a folder called toolchains, looked for in every folder above the one 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)
candidates = [parent / "toolchains" / name]
if (parent / "tmp").is_dir():
candidates += sorted((parent / "tmp").glob(f"*/toolchains/{name}"))
for found in candidates:
if (found / "bin").is_dir():
return found
sys.exit(f"no toolchains folder above this one holds {name}")
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().
"""
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 / "mysqld").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")})
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 / "mysqld"), *opts, "--bind-address=127.0.0.1", f"--port={self.port}",
"--socket=mysql.sock", "--mysqlx=OFF", *self.options, *[str(e) for e in extra]]
if show:
shown = ["mysqld", opts[0].replace(str(self.base) + "/", ""), "--datadir=./data", "--bind-address=127.0.0.1",
"--port=$PORT", "--socket=mysql.sock", "--mysqlx=OFF", *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(LOG_PREFIX.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 _ping(self):
code, _ = self._call(["mysqladmin", 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(["mysqladmin", 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(LOG_PREFIX.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 = ["mysql", 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 = ["mysql", *flags, *(["--force"] if force else []), *([db] if db else [])]
args = ["mysql", 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 = ["mysql", *flags, *([db] if db else []), "-e", statement]
code, out = self._call(["mysql", 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(["mysql", 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"mysql> {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()The recorded run shows four facts in order. SHOW GRANTS FOR 'migrator'@'localhost' names the roles the account
holds; SHOW GRANTS ... USING 'app_ddl', 'app_rw' adds the lines the roles grant. reporting holds app_read yet
CURRENT_ROLE() on login is NONE, so a SELECT on shop.orders is denied until SET ROLE 'app_read' turns it
on for the session. SET DEFAULT ROLE 'app_read' TO 'reporting'@'localhost' then records a
default role: A role recorded for an account with SET DEFAULT ROLE, which MySQL activates each time the account logs in; without one, and with activate_all_roles_on_login off (the default), a granted role stays inactive until the session runs SET ROLE. that activates at every login, which the next session proves by
running the read and failing the DELETE. A role never grants more than
it holds. The global activate_all_roles_on_login flag turns every granted role on at every login for every
account; the recorded line shows oncall with app_read active after the switch.
Checkpoint
Check: Roles, default roles and SHOW GRANTS USING
Narrower grants with partial revokes, and SYSTEM_USER for admin accounts Pro: opens with a Pro, Premium or Ultimate pass
The lesson’s first half gave each account the rights of its job. Two features let a design hold up when the shape of the job
is wider than a single schema: a partial revoke keeps a global grant out of one schema, and SYSTEM_USER turns an
admin account into one that only another SYSTEM_USER can touch.
In this part:
- A partial revoke carves a schema out of a global grant
- How one privilege shields admin accounts from ordinary operators
- Key takeaways from this part
1 runnable example (mysql_session.py, partial_revokes.sql, partial_revokes_session.py, schemas.sql) · about 8 minutes
Opening this part with your pass…
This part opens with any MySmartCoPilot pass.
- Pro pass · 1 week₹79$2.99
- 30 days₹179$5.99
- 1 year₹1,699$45
One payment; a pass never renews. The price shown is the total you pay: no fees or charges are added at checkout.
Passes can’t be bought right now, so this part can’t be opened at the moment. The free part above stands on its own. If you have a pass, enter your licence key.
It also opens every other in-depth part in Learn and every Pro tool. Sales are final, except where the law gives you a right to withdraw, for example in the EU and the UK (refund policy).
Definer views without their owner, and queries that find over-broad grants Pro: opens with a Pro, Premium or Ultimate pass
A view or stored program runs its body as some account, and that account, not the caller, can decide which rows the caller reaches. Recorded runs compare a DEFINER view with an INVOKER view, follow a definer account the server refuses to drop, and turn seven queries into an audit of who can do what.
In this part:
- A SQL SECURITY DEFINER view exposes aggregates without the rows
- When a definer account goes: the refusal, the check and the fix
- An audit query set that finds over-broad grants
- Finish on the exercise: privilege puzzles
- Predict which SELECT each account can run from its grants
- Results of the sample tests
- Key takeaways on definer views and audits
3 runnable examples (audit_grants.sql, audit_grants_session.py, audit_setup.sql, definers.sql, definers_session.py, mysql_session.py, orphans.sql, overbroad.sql, overbroad_session.py, repoint.sql, schemas.sql) · 1 exercise with sample tests · about 15 minutes
Opening this part with your pass…Pro part. It opens with the same pass as the part above.
Practise: the privilege model, the five accounts and the roles
Each question comes back to the opening’s shapes, so a reader who has read it answers from the facts.
Practice that adapts to you
Practise this lesson
6 questions about 5 ideas of this lesson. Each answer is explained at once, and what you get right is kept in this browser.
Interview questions
The three important questions below extend the opening into a design conversation; the warm-up answers one plain question on the model.
Warm-up: what makes a privilege dynamic rather than static?
A static privilege is built into the server and is always available to grant; its global and schema grants are
Y/N columns of mysql.user and mysql.db, and its table and column grants are sets in mysql.tables_priv and
mysql.columns_priv. A dynamic privilege is registered at run time, most by the server at startup and the rest by
the component or plugin that defines it; it is held per account as a row of mysql.global_grants and grants only
at the global level: GRANT BACKUP_ADMIN ON shop.* returns ERROR 3619. The recorded run of privileges.sql
shows the counts on MySQL 8.4.11, 31 static global columns and 40 dynamic rows for root, and the refusal of
BACKUP_ADMIN at the schema level.
-
Mid to senior
A new team of five developers needs database accounts on a MySQL 8.4 server that holds application rows, a reporting replica of them and a
payrollschema. Design the privilege set each account carries, the roles that group the privileges, and the one line that keepspayrollout of the operators' globalSELECT. -
Senior to staff
A compliance review asks that no application account read
payrollschemas, even though a handful of accounts inherited a globalSELECTfrom a long-gone migration. You cannot drop and recreate every account during the week of the review. Plan the roll-out of partial revokes across one source and six replicas, and the proof you can hand to the auditor. -
Mid to senior
During an incident, an operator asks you which accounts can read
salaries. The server has twenty accounts, a handful of roles, a few partial revokes and three definer views onpayroll.salaries. Write the audit that answers the question completely, with the queries you run and the way you read them.
Flashcards · 7 cards
Recall it, then check
Say the answer to yourself before you look: recalling it is what makes it stick.
-
How does MySQL 8.4 store static and dynamic privileges?
Static privileges are columns, 31 global _priv columns of mysql.user on 8.4.11. Dynamic ones, registered at run time, are rows of mysql.global_grants (40 for root) and grant only globally: GRANT BACKUP_ADMIN ON shop.* returns ERROR 3619.
-
Why does GRANT FIREWALL_ADMIN ON *.* fail as a syntax error on a Community server?
A dynamic privilege nothing has registered cannot be granted, and the firewall that would register this one ships only in the Enterprise build, so the server answers ERROR 1149.
-
What does least privilege look like for MySQL accounts?
Each account holds only the grants of its one job, scoped to the hosts it logs in from, with REQUIRE SSL when it crosses a shared network: the application reads and writes its schema, the migrator adds schema changes, reporting reads, backup reads with FLUSH_TABLES, and replication holds one grant.
-
Why does a backup account need the FLUSH_TABLES privilege?
mysqldump --single-transaction on a server with GTIDs on runs a FLUSH TABLES; after REVOKE FLUSH_TABLES the dump failed at that step with ERROR 1227.
-
An account holds a role yet its SELECT is denied after login. Why?
A granted role is inactive until SET ROLE turns it on for the session, or SET DEFAULT ROLE (or activate_all_roles_on_login) activates it at login: CURRENT_ROLE() read NONE in the run until then.
-
What is a MySQL role, and how do you read what it grants?
A locked account with no password that groups privileges, granted to accounts so the accounts stay unchanged when a job changes. SHOW GRANTS FOR an account names its roles; adding USING with the role names lists what they grant.
-
Does the SUPER privilege still work on MySQL 8.4?
Yes, but it is deprecated: using it leaves a warning saying the SUPER privilege identifier is deprecated.
Key takeaways
- MySQL keeps static privileges in columns (31 global
_privcolumns ofmysql.useron 8.4.11) and dynamic privileges in rows ofmysql.global_grants(40 forroot); dynamic privileges only grant at the global level. - Five least-privilege account shapes keep each job separate: the application reads and writes its schema, the
migrator adds schema changes, reporting reads, backup reads with
FLUSH_TABLES, replication holds one grant andREQUIRE SSL. - A role is a locked account with no password that groups privileges; grant the role to accounts, set a default
role for the session to open with the right active, and read what a role carries with
SHOW GRANTS ... USING.
Terms in this lesson
- default role
- A role recorded for an account with SET DEFAULT ROLE, which MySQL activates each time the account logs in; without one, and with activate_all_roles_on_login off (the default), a granted role stays inactive until the session runs SET ROLE.
- dynamic privilege
- A privilege registered at run time, most by the server at startup and the rest by the component or plugin that defines it; held per account as a row of mysql.global_grants and granted only at the global level, ON *.*.
- least privilege
- The design rule that every account carries only the privileges it needs for its job, scoped to the hosts it logs in from, with wider grants replaced by schema- or table-level ones and risky dynamic privileges held only by named accounts.
- partial revoke
- A REVOKE at the schema level that narrows one account's global grant, so that, for example, a global SELECT leaves out one schema.
- role
- A named account in mysql.user with account_locked=Y and no password that groups privileges together; grant the role to accounts, activate it per session with SET ROLE or at login with a default role.
- static privilege
- A privilege built into the MySQL server and always available to grant; its global and schema grants are Y/N columns of mysql.user and mysql.db, its table and column grants sets in mysql.tables_priv and mysql.columns_priv.
- SYSTEM_USER privilege
- A dynamic privilege that marks an account as a system account, which an ordinary administrator without SYSTEM_USER cannot modify, so the holders of SYSTEM_USER form the server's smaller administrator set.
References
- MySQL 8.4 Reference Manual: 8.2.2 Privileges Provided by MySQL (Oracle Corporation)
- MySQL 8.4 Reference Manual: 8.2.3 Grant Tables (Oracle Corporation)
- MySQL 8.4 Reference Manual: 8.2.10 Using Roles (Oracle Corporation)
- MySQL 8.4 Reference Manual: 8.2.11 Account Categories (Oracle Corporation)
- MySQL 8.4 Reference Manual: 8.2.12 Privilege Restriction Using Partial Revokes (Oracle Corporation)
- MySQL 8.4 Reference Manual: 8.2.13 When Privilege Changes Take Effect (Oracle Corporation)
- MySQL 8.4 Reference Manual: 6.5.4 mysqldump, A Database Backup Program (Oracle Corporation)
- MySQL 8.4 Reference Manual: 15.1.23 CREATE VIEW Statement (Oracle Corporation)
- MySQL 8.4 Reference Manual: 15.7.1.6 GRANT Statement (Oracle Corporation)
- MySQL 8.4 Reference Manual: 15.7.1.8 REVOKE Statement (Oracle Corporation)
- MySQL 8.4 Reference Manual: 28.3.46 The INFORMATION_SCHEMA USER_PRIVILEGES Table (Oracle Corporation)
- MySQL 8.4 Reference Manual: 27.6 Stored Object Access Control (Oracle Corporation)
- MariaDB Server Documentation: GRANT (MariaDB plc)
- MariaDB Server Documentation: DENY (MariaDB plc)
Report a problem with this lesson
Kept only in this browser. Your Learn progress