sql_drills: no root shell for learner SQL

Learner SQL piped into the mysql client could run `\! id` as container
root and read every DB. Now: --binary-mode (client commands error out),
a refused-backslash-line guard, learner exec as uid 65534, and a random
root password (0600 file, via env, adopted live on a legacy container).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WzxdWYGW5NZL5uVpnqmNKo

sql_drills/README.md

@@ -1,3 +1,4 @@
 SQL drill engine (MySQL 8.0.33 + PostgreSQL 18.6, one container each): `import sql_drills as sd`.
 Drills: `sd.run(id, sql, dialect)`. LeetCode examples: `sd.run_problem(slug_or_statement, sql, dialect)`.
 Data: `$SQL_DRILLS_DATA`. Install: `./install.sh`.
+Root password: random, `~/.config/sql_drills/root_password` (0600); learner SQL runs as a per-run user, client commands (`\!`) refused.

sql_drills/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "sql_drills"
-version = "0.2.1"
+version = "0.2.2"
 description = "SQL drills (MySQL, PostgreSQL): sandboxed grading, LeetCode statement examples"
 requires-python = ">=3.14"
 # Stdlib only on purpose: the tutor server imports it from the system python.

sql_drills/sql_drills/_engine.py

@@ -3,13 +3,16 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
+import os
 import shutil
 import subprocess
 import time
 from typing import TYPE_CHECKING
 
+from sql_drills import _secret
+
 if TYPE_CHECKING:
-    from collections.abc import Callable
+    from collections.abc import Callable, Mapping
 
 IMAGE = "mysql:8.0.33"
 CONTAINER = "sql-drills-mysql"
@@ -18,6 +21,10 @@ NETWORK_ALIAS = "db"
 READY_TIMEOUT = 45.0
 QUERY_TIMEOUT = 30.0
 _DOCKER_TIMEOUT = 50.0
+# Learner SQL runs as this OS user (nobody): no root shell, no datadir access.
+_LEARNER_UID = "65534"
+# ``mysql`` reads the root password from this variable, never from argv.
+_ROOT_LOGIN_VAR = "MYSQL_PWD"
 
 
 class EngineError(RuntimeError):
@@ -32,7 +39,10 @@ class SqlResult:
 
 
 def _docker(
-    *args: str, stdin: str | None = None, timeout: float = _DOCKER_TIMEOUT
+    *args: str,
+    stdin: str | None = None,
+    timeout: float = _DOCKER_TIMEOUT,
+    env: Mapping[str, str] | None = None,
 ) -> subprocess.CompletedProcess[str]:
     exe = shutil.which("docker")
     if exe is None:
@@ -46,6 +56,7 @@ def _docker(
             text=True,
             check=False,
             timeout=timeout,
+            env=env,
         )
     except subprocess.TimeoutExpired as exc:
         msg = f"docker {args[0]} timed out after {timeout:.0f}s"
@@ -55,6 +66,11 @@ def _docker(
 docker = _docker  # public alias for sibling engine modules
 
 
+def _root_env() -> dict[str, str]:
+    """Environment carrying the root password to ``docker`` (not via argv)."""
+    return {**os.environ, _ROOT_LOGIN_VAR: _secret.root_password()}
+
+
 def container_state() -> str:
     res = _docker("inspect", "-f", "{{.State.Status}}", CONTAINER)
     return res.stdout.strip() if res.returncode == 0 else "missing"
@@ -74,9 +90,10 @@ def _create() -> subprocess.CompletedProcess[str]:
         "run", "-d", "--name", CONTAINER,
         "--tmpfs", "/var/lib/mysql:rw,size=512m",
         "--memory", "1500m",
-        "-e", "MYSQL_ALLOW_EMPTY_PASSWORD=yes",
+        "-e", "MYSQL_ROOT_PASSWORD",
         *(shared if net.returncode == 0 else []),
         IMAGE,
+        env={**os.environ, "MYSQL_ROOT_PASSWORD": _secret.root_password()},
     )  # fmt: skip
 
 
@@ -118,16 +135,45 @@ def ensure_running(
     """Create or resume the container if needed, wait (bounded) for MySQL."""
     start_or_create(CONTAINER, _create)
     wait_ready(is_ready, "MySQL", timeout, sleep, clock)
+    secure_root()
+
+
+def secure_root() -> None:
+    """Make sure root has our password; adopt a legacy passwordless root once.
+
+    A container started before this existed (or restarted from its old
+    ``ALLOW_EMPTY_PASSWORD`` config) has an empty root password. Set it live,
+    without recreating the container: its tmpfs holds databases we must keep.
+    """
+    if run_sql("SELECT 1;").returncode == 0:
+        return
+    password = _secret.root_password()
+    alter = "".join(
+        f"ALTER USER IF EXISTS 'root'@'{host}' IDENTIFIED BY '{password}'; "
+        for host in ("localhost", "%")
+    )
+    res = exec_query(["exec", "-i", CONTAINER, "mysql", "-uroot"], alter, QUERY_TIMEOUT)
+    if res.returncode != 0:
+        msg = (
+            "MySQL root rejects the stored password and is not passwordless: "
+            f"{res.stderr.strip()}; remove {CONTAINER} to start over"
+        )
+        raise EngineError(msg)
 
 
 def stop() -> None:
     _docker("rm", "-f", CONTAINER)
 
 
-def exec_query(argv: list[str], stdin: str | None, timeout: float) -> SqlResult:
+def exec_query(
+    argv: list[str],
+    stdin: str | None,
+    timeout: float,
+    env: Mapping[str, str] | None = None,
+) -> SqlResult:
     """``docker <argv>`` as a query: a timeout becomes a clear EngineError."""
     try:
-        res = _docker(*argv, stdin=stdin, timeout=timeout)
+        res = _docker(*argv, stdin=stdin, timeout=timeout, env=env)
     except EngineError as exc:
         if "timed out" in str(exc):
             msg = f"query timed out after {timeout:.0f}s"
@@ -144,10 +190,19 @@ def run_sql(
     headers: bool = False,
     timeout: float = QUERY_TIMEOUT,
 ) -> SqlResult:
-    argv = ["exec", "-i", CONTAINER, "mysql", f"-u{user}"]
+    r"""Run ``sql`` through the ``mysql`` client as ``user``.
+
+    ``--binary-mode`` turns every client command (``\!`` shell escape,
+    ``source``, ``\.``) into an error for piped input. Only root gets the
+    password, and only the learner path drops to an unprivileged OS user.
+    """
+    root = user == "root"
+    argv = ["exec", "-i"]
+    argv += ["-e", _ROOT_LOGIN_VAR] if root else ["-u", _LEARNER_UID]
+    argv += [CONTAINER, "mysql", f"-u{user}", "--binary-mode"]
     argv += ["--default-character-set=utf8mb4", "-B"]
     if not headers:
         argv.append("-N")
     if database:
         argv.append(database)
-    return exec_query(argv, sql, timeout)
+    return exec_query(argv, sql, timeout, _root_env() if root else None)

sql_drills/sql_drills/_paths.py

@@ -6,6 +6,7 @@ import os
 from pathlib import Path
 
 ENV_VAR = "SQL_DRILLS_DATA"
+CONFIG_ENV_VAR = "SQL_DRILLS_CONFIG"
 
 
 def data_dir() -> Path:
@@ -21,3 +22,11 @@ def builtin_dir() -> Path:
 
 def generated_dir() -> Path:
     return data_dir() / "drills"
+
+
+def config_dir() -> Path:
+    """Where the engine's secret lives (never in git, never in the data dir)."""
+    override = os.environ.get(CONFIG_ENV_VAR, "")
+    if override:
+        return Path(override)
+    return Path.home() / ".config" / "sql_drills"

sql_drills/sql_drills/_sandbox.py

@@ -17,6 +17,8 @@ if TYPE_CHECKING:
 
 SNAPSHOT_ROWS = 200
 _MAX_EXECUTION_MS = 20000
+_NO_META = "mysql client commands (backslash) are not allowed"
+_META = re.compile(r"(?:^|\r)[\s\ufeff]*\\", re.MULTILINE)
 _CODE = re.compile(r"ERROR (\d+)")
 _COLUMNS_SQL = (
     "SELECT table_name, column_name FROM information_schema.columns "
@@ -43,12 +45,24 @@ def _outcome(res: _engine.SqlResult) -> Executed:
     return Executed(ok=True, error=None, code=None, rows=parse_rows(res.stdout))
 
 
+def has_meta_command(sql: str) -> bool:
+    r"""True if any line starts with a backslash (after a BOM or whitespace).
+
+    The client's ``\!`` runs a shell. ``--binary-mode`` already turns every
+    such command into an error; refusing here too keeps a stray one from ever
+    reaching the client, whatever precedes it (comments, other statements).
+    """
+    return _META.search(sql) is not None
+
+
 @dataclass(frozen=True)
 class Sandbox:
     db: str
     user: str
 
     def _run(self, sql: str, *, headers: bool) -> _engine.SqlResult:
+        if has_meta_command(sql):
+            return _engine.SqlResult(1, "", _NO_META)
         limit = f"SET SESSION max_execution_time={_MAX_EXECUTION_MS}; "
         return _engine.run_sql(limit + sql, self.db, self.user, headers=headers)
 

sql_drills/sql_drills/_secret.py

@@ -0,0 +1,38 @@
+"""The MySQL root password: random, 0600, readable only by this engine.
+
+The container's root used to be passwordless, so anything that reached a shell
+in it (a backslash-bang meta-command) owned every database. Now root has this
+password; learner SQL runs as a per-run user that never sees it.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import secrets
+
+from sql_drills import _paths
+
+_FILE = "root_password"
+
+
+def root_password() -> str:
+    """Read the password, creating it on first use.
+
+    The file appears complete or not at all: written to a private temp name,
+    then hard-linked into place (which fails if another process won the race).
+    """
+    path = _paths.config_dir() / _FILE
+    if path.exists():
+        return path.read_text(encoding="utf-8").strip()
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    tmp = path.with_name(f".{_FILE}.{secrets.token_hex(4)}")
+    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    with os.fdopen(fd, "w", encoding="utf-8") as handle:
+        handle.write(secrets.token_urlsafe(24) + "\n")
+    try:
+        with contextlib.suppress(FileExistsError):
+            os.link(tmp, path)
+    finally:
+        tmp.unlink()
+    return path.read_text(encoding="utf-8").strip()

sql_drills/sql_drills/tests/conftest.py

@@ -14,9 +14,17 @@ if TYPE_CHECKING:
 
 
 @pytest.fixture(autouse=True)
-def _redirect_data(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+def _redirect_data(
-    """Generated drills never land in the real ``~/.local/share/sql_drills``."""
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path, request: pytest.FixtureRequest
+) -> None:
+    """Generated drills never land in the real ``~/.local/share/sql_drills``.
+
+    The root-password file is redirected too, except for the ``docker`` live
+    tests: they talk to the real container, which only knows the real password.
+    """
     monkeypatch.setenv("SQL_DRILLS_DATA", str(tmp_path / "data"))
+    if request.node.get_closest_marker("docker") is None:
+        monkeypatch.setenv("SQL_DRILLS_CONFIG", str(tmp_path / "config"))
 
 
 @dataclass

sql_drills/sql_drills/tests/test_engine.py

@@ -9,7 +9,7 @@ from unittest.mock import MagicMock, patch
 
 import pytest
 
-from sql_drills import _engine
+from sql_drills import _engine, _secret
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -59,8 +59,8 @@ def test_is_ready_follows_ping_exit_code() -> None:
 
 
 def test_creates_container_on_shared_network_when_it_exists() -> None:
-    # inspect(missing), network inspect(ok), run, ping.
+    # inspect(missing), network inspect(ok), run, ping, root login check.
-    with _run(_cp(1), _cp(0), _cp(0, "id"), _cp(0)) as run:
+    with _run(_cp(1), _cp(0), _cp(0, "id"), _cp(0), _cp(0)) as run:
         _engine.ensure_running(timeout=5, sleep=_no_sleep)
     argv = run.call_args_list[2].args[0]
     assert argv[:3] == ["docker", "run", "-d"]
@@ -70,23 +70,60 @@ def test_creates_container_on_shared_network_when_it_exists() -> None:
     assert argv[argv.index("--network-alias") + 1] == _engine.NETWORK_ALIAS
 
 
+def test_new_container_gets_a_random_root_password_not_via_argv() -> None:
+    with _run(_cp(1), _cp(1), _cp(0, "id"), _cp(0), _cp(0)) as run:
+        _engine.ensure_running(timeout=5, sleep=_no_sleep)
+    argv = run.call_args_list[2].args[0]
+    password = run.call_args_list[2].kwargs["env"]["MYSQL_ROOT_PASSWORD"]
+    assert len(password) >= 24
+    assert "MYSQL_ALLOW_EMPTY_PASSWORD=yes" not in argv
+    assert not any(password in part for part in argv)
+    assert "MYSQL_ROOT_PASSWORD" in argv
+
+
 def test_creates_container_plain_when_network_absent() -> None:
-    with _run(_cp(1), _cp(1), _cp(0, "id"), _cp(0)) as run:
+    with _run(_cp(1), _cp(1), _cp(0, "id"), _cp(0), _cp(0)) as run:
         _engine.ensure_running(timeout=5, sleep=_no_sleep)
     assert "--network" not in run.call_args_list[2].args[0]
 
 
 def test_resumes_exited_container() -> None:
-    with _run(_cp(0, "exited\n"), _cp(0), _cp(0)) as run:
+    with _run(_cp(0, "exited\n"), _cp(0), _cp(0), _cp(0)) as run:
         _engine.ensure_running(timeout=5, sleep=_no_sleep)
     assert run.call_args_list[1].args[0][:2] == ["docker", "start"]
 
 
 def test_running_container_is_never_recreated() -> None:
-    with _run(_cp(0, "running\n"), _cp(0)) as run:
+    with _run(_cp(0, "running\n"), _cp(0), _cp(0)) as run:
         _engine.ensure_running(timeout=5, sleep=_no_sleep)
     verbs = [c.args[0][1] for c in run.call_args_list]
-    assert verbs == ["inspect", "exec"]
+    assert verbs == ["inspect", "exec", "exec"]
+
+
+def test_legacy_passwordless_root_gets_the_password_live() -> None:
+    # inspect, ping, root login refused, passwordless ALTER USER works.
+    denied = _cp(1, "", "ERROR 1045 (28000): Access denied")
+    with _run(_cp(0, "running\n"), _cp(0), denied, _cp(0)) as run:
+        _engine.ensure_running(timeout=5, sleep=_no_sleep)
+    alter = run.call_args_list[3]
+    password = _secret.root_password()
+    assert "env" not in alter.kwargs or alter.kwargs["env"] is None
+    assert (
+        f"ALTER USER IF EXISTS 'root'@'localhost' IDENTIFIED BY '{password}'"
+        in (alter.kwargs["input"])
+    )
+    assert "'root'@'%'" in alter.kwargs["input"]
+    assert not any(password in part for part in alter.args[0])
+    assert "-p" not in alter.args[0]
+
+
+def test_root_rejecting_our_password_is_a_clear_error() -> None:
+    denied = _cp(1, "", "ERROR 1045 (28000): Access denied")
+    with (
+        _run(_cp(0, "running\n"), _cp(0), denied, denied),
+        pytest.raises(_engine.EngineError, match="not passwordless"),
+    ):
+        _engine.ensure_running(timeout=5, sleep=_no_sleep)
 
 
 def test_create_failure_is_reported() -> None:
@@ -110,18 +147,38 @@ def test_stop_removes_container() -> None:
     assert run.call_args.args[0] == ["docker", "rm", "-f", _engine.CONTAINER]
 
 
-def test_run_sql_defaults_to_root_and_no_headers() -> None:
+def test_run_sql_defaults_to_root_with_password_from_env_only() -> None:
     with _run(_cp(0, "1\n", "")) as run:
         res = _engine.run_sql("SELECT 1;", database="d1")
     argv = run.call_args.args[0]
-    assert argv[:4] == ["docker", "exec", "-i", _engine.CONTAINER]
+    assert argv[:6] == ["docker", "exec", "-i", "-e", "MYSQL_PWD", _engine.CONTAINER]
     assert "-uroot" in argv
+    assert "-u" not in argv
+    password = _secret.root_password()
+    assert run.call_args.kwargs["env"]["MYSQL_PWD"] == password
+    assert not any(password in part for part in argv)
     assert "-N" in argv
     assert argv[-1] == "d1"
     assert run.call_args.kwargs["input"] == "SELECT 1;"
     assert (res.returncode, res.stdout) == (0, "1\n")
 
 
+def test_every_run_is_binary_mode_so_client_commands_cannot_run() -> None:
+    with _run(_cp(0), _cp(0)) as run:
+        _engine.run_sql("SELECT 1;")
+        _engine.run_sql("SELECT 1;", user="u_x")
+    assert all("--binary-mode" in c.args[0] for c in run.call_args_list)
+
+
+def test_learner_run_is_unprivileged_and_never_sees_the_password() -> None:
+    with _run(_cp(0)) as run:
+        _engine.run_sql("SELECT 1;", user="u_x")
+    argv = run.call_args.args[0]
+    assert argv[:6] == ["docker", "exec", "-i", "-u", "65534", _engine.CONTAINER]
+    assert "MYSQL_PWD" not in argv
+    assert run.call_args.kwargs["env"] is None
+
+
 def test_run_sql_headers_user_and_no_database() -> None:
     with _run(_cp(0)) as run:
         _engine.run_sql("SELECT 1;", user="u_x", headers=True)

sql_drills/sql_drills/tests/test_live.py

@@ -98,6 +98,64 @@ def test_user_sql_cannot_see_or_drop_other_databases() -> None:
         _engine.run_sql("DROP DATABASE IF EXISTS drill_canary_guard;")
 
 
+_ATTACKS = [
+    "\\! id",
+    "-- c\n\\! mysql -uroot -N -e 'show databases'",
+    "SELECT 1;\n\\! id",
+    "\ufeff\\! id",
+    "SELECT 1 -- \\! id",
+    "SELECT 1; \\! id",
+    "SELECT 1 \\! id",
+    "source /etc/passwd",
+]
+
+
+@pytest.mark.parametrize("attack", _ATTACKS)
+def test_client_commands_never_reach_a_shell(attack: str) -> None:
+    """Red-team: ``\\!`` used to run a root shell inside the container."""
+    drill = sd.get_drill("01_pivot_sum")
+    with _sandbox.sandbox(drill.schema_sql) as box:
+        done = box.query(attack)
+        # Even bypassing the text guard, the client itself must refuse.
+        raw = _engine.run_sql(attack, box.db, box.user)
+    assert "uid=" not in str(done.rows) + str(done.error)
+    assert "uid=" not in raw.stdout
+    assert "theory" not in raw.stdout
+
+
+def test_client_runs_without_a_shell_escape_or_root_os_user() -> None:
+    drill = sd.get_drill("01_pivot_sum")
+    with _sandbox.sandbox(drill.schema_sql) as box:
+        raw = _engine.run_sql("\\! id", box.db, box.user)
+    assert raw.returncode != 0
+    assert "Unknown command" in raw.stderr
+    # The unprivileged OS user is what ``run_sql`` execs as.
+    ident = _engine.docker("exec", "-u", "65534", _engine.CONTAINER, "id", "-u")
+    assert ident.stdout.strip() == "65534"
+
+
+def test_learner_cannot_see_theory_or_other_drill_databases() -> None:
+    drill = sd.get_drill("01_pivot_sum")
+    with _sandbox.sandbox(drill.schema_sql) as box, _sandbox.sandbox("") as other:
+        seen = {r[0] for r in box.execute("SHOW DATABASES;").rows}
+        assert seen <= {box.db, "information_schema", "performance_schema"}
+        assert "theory" not in seen
+        assert other.db not in seen
+        assert not box.execute("SELECT * FROM theory.anything;").ok
+        assert not box.execute("SELECT * FROM mysql.user;").ok
+
+
+def test_root_is_passwordless_no_more() -> None:
+    """Root without the engine's password is refused, locally and over TCP."""
+    for host in ("-hlocalhost", "-h127.0.0.1"):
+        res = _engine.docker(
+            "exec", _engine.CONTAINER, "mysql", "-uroot", host, "-N", "-e", "SELECT 1"
+        )
+        assert res.returncode != 0
+        assert "Access denied" in res.stderr
+    assert _engine.run_sql("SELECT 1;").stdout.strip() == "1"
+
+
 def test_ensure_running_does_not_recreate_a_running_container() -> None:
     before = _container_id()
     sd.ensure_running()

sql_drills/sql_drills/tests/test_paths.py

@@ -31,3 +31,13 @@ def test_empty_env_falls_back(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -
 
 def test_builtin_dir_ships_fixtures() -> None:
     assert (_paths.builtin_dir() / "01_pivot_sum" / "meta.json").is_file()
+
+
+def test_config_dir_env_and_default(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    monkeypatch.setenv("SQL_DRILLS_CONFIG", str(tmp_path / "c"))
+    assert _paths.config_dir() == tmp_path / "c"
+    monkeypatch.setenv("SQL_DRILLS_CONFIG", "")
+    monkeypatch.setenv("HOME", str(tmp_path))
+    assert _paths.config_dir() == tmp_path / ".config" / "sql_drills"

sql_drills/sql_drills/tests/test_sandbox.py

@@ -189,3 +189,38 @@ def test_query_error_has_no_columns(fake_mysql: FakeMysql) -> None:
         done = box.query("nope")
     assert not done.ok
     assert done.columns == []
+
+
+RED_TEAM = [
+    "\\! id",
+    "-- harmless comment\n\\! id",
+    "SELECT 1;\n\\! mysql -uroot -e 'show databases'",
+    "\\! id",
+    "  \t\\system id",
+    "SELECT 1;\r\\! id",
+    "/* c */\n\n   \\. /etc/passwd",
+]
+
+
+@pytest.mark.parametrize("payload", RED_TEAM)
+def test_meta_commands_are_refused_before_reaching_the_client(
+    fake_mysql: FakeMysql, payload: str
+) -> None:
+    with _sandbox.sandbox("") as box:
+        before = len(fake_mysql.calls)
+        done = box.query(payload)
+        also = box.execute(payload)
+    assert not done.ok
+    assert done.error == "mysql client commands (backslash) are not allowed"
+    assert not also.ok
+    assert all(payload not in str(c["sql"]) for c in fake_mysql.calls[before:]), (
+        "refused SQL must never be sent"
+    )
+
+
+@pytest.mark.parametrize(
+    "sql",
+    ["SELECT 1;", "SELECT 'a\\\\b';", "SELECT 1 -- \\! id", "SELECT 'x\\n';"],
+)
+def test_backslash_inside_a_line_is_ordinary_sql(sql: str) -> None:
+    assert not _sandbox.has_meta_command(sql)

sql_drills/sql_drills/tests/test_secret.py

@@ -0,0 +1,47 @@
+"""The root password is random, private and created exactly once."""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING
+
+from sql_drills import _paths, _secret
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from _typeshed import StrPath
+    import pytest
+
+
+def test_created_once_random_and_private() -> None:
+    first = _secret.root_password()
+    assert len(first) >= 24
+    assert _secret.root_password() == first
+    path = _paths.config_dir() / "root_password"
+    assert path.stat().st_mode & 0o777 == 0o600
+    assert path.parent.stat().st_mode & 0o777 == 0o700
+    assert [p.name for p in path.parent.iterdir()] == ["root_password"]
+
+
+def test_different_homes_get_different_passwords(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    one = _secret.root_password()
+    monkeypatch.setenv("SQL_DRILLS_CONFIG", str(tmp_path / "other"))
+    assert _secret.root_password() != one
+
+
+def test_losing_the_race_reads_the_winners_password(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    path = _paths.config_dir() / "root_password"
+    real_link = os.link
+
+    def winner_first(src: StrPath, dst: StrPath) -> None:
+        path.write_text("winner\n", encoding="utf-8")
+        real_link(src, dst)
+
+    monkeypatch.setattr(os, "link", winner_first)
+    assert _secret.root_password() == "winner"
+    assert [p.name for p in path.parent.iterdir()] == ["root_password"]