From de7d5d17eb01839d60d01816caa5bad244f994f5 Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Mon, 2 Nov 2026 16:05:59 +0100 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01WzxdWYGW5NZL5uVpnqmNKo --- sql_drills/README.md | 1 + sql_drills/pyproject.toml | 2 +- sql_drills/sql_drills/_engine.py | 69 +++++++++++++++++-- sql_drills/sql_drills/_paths.py | 9 +++ sql_drills/sql_drills/_sandbox.py | 14 ++++ sql_drills/sql_drills/_secret.py | 38 +++++++++++ sql_drills/sql_drills/tests/conftest.py | 12 +++- sql_drills/sql_drills/tests/test_engine.py | 75 ++++++++++++++++++--- sql_drills/sql_drills/tests/test_live.py | 58 ++++++++++++++++ sql_drills/sql_drills/tests/test_paths.py | 10 +++ sql_drills/sql_drills/tests/test_sandbox.py | 35 ++++++++++ sql_drills/sql_drills/tests/test_secret.py | 47 +++++++++++++ 12 files changed, 351 insertions(+), 19 deletions(-) create mode 100644 sql_drills/sql_drills/_secret.py create mode 100644 sql_drills/sql_drills/tests/test_secret.py diff --git a/sql_drills/README.md b/sql_drills/README.md index 6c43f0f..9b4e5e4 100644 --- a/sql_drills/README.md +++ b/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. diff --git a/sql_drills/pyproject.toml b/sql_drills/pyproject.toml index 8426301..a195f64 100644 --- a/sql_drills/pyproject.toml +++ b/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. diff --git a/sql_drills/sql_drills/_engine.py b/sql_drills/sql_drills/_engine.py index becce17..1535542 100644 --- a/sql_drills/sql_drills/_engine.py +++ b/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 `` 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) diff --git a/sql_drills/sql_drills/_paths.py b/sql_drills/sql_drills/_paths.py index b10d1a7..bce81f7 100644 --- a/sql_drills/sql_drills/_paths.py +++ b/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" diff --git a/sql_drills/sql_drills/_sandbox.py b/sql_drills/sql_drills/_sandbox.py index bbec754..13818eb 100644 --- a/sql_drills/sql_drills/_sandbox.py +++ b/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) diff --git a/sql_drills/sql_drills/_secret.py b/sql_drills/sql_drills/_secret.py new file mode 100644 index 0000000..9baa9c5 --- /dev/null +++ b/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() diff --git a/sql_drills/sql_drills/tests/conftest.py b/sql_drills/sql_drills/tests/conftest.py index befbb6e..7176ca0 100644 --- a/sql_drills/sql_drills/tests/conftest.py +++ b/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: - """Generated drills never land in the real ``~/.local/share/sql_drills``.""" +def _redirect_data( + 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 diff --git a/sql_drills/sql_drills/tests/test_engine.py b/sql_drills/sql_drills/tests/test_engine.py index 4470eb1..a3e2561 100644 --- a/sql_drills/sql_drills/tests/test_engine.py +++ b/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. - with _run(_cp(1), _cp(0), _cp(0, "id"), _cp(0)) as run: + # inspect(missing), network inspect(ok), run, ping, root login check. + 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) diff --git a/sql_drills/sql_drills/tests/test_live.py b/sql_drills/sql_drills/tests/test_live.py index 4d0ba7b..f94ff30 100644 --- a/sql_drills/sql_drills/tests/test_live.py +++ b/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() diff --git a/sql_drills/sql_drills/tests/test_paths.py b/sql_drills/sql_drills/tests/test_paths.py index fa11df8..a96a3e4 100644 --- a/sql_drills/sql_drills/tests/test_paths.py +++ b/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" diff --git a/sql_drills/sql_drills/tests/test_sandbox.py b/sql_drills/sql_drills/tests/test_sandbox.py index f16eaec..359f6fb 100644 --- a/sql_drills/sql_drills/tests/test_sandbox.py +++ b/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) diff --git a/sql_drills/sql_drills/tests/test_secret.py b/sql_drills/sql_drills/tests/test_secret.py new file mode 100644 index 0000000..ad2a9ab --- /dev/null +++ b/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"]