diff --git a/studio/backend/auth/storage.py b/studio/backend/auth/storage.py index 3d4e6b182d..d7e890593c 100644 --- a/studio/backend/auth/storage.py +++ b/studio/backend/auth/storage.py @@ -9,6 +9,7 @@ import ipaddress import os import secrets import sqlite3 +import tempfile import threading from datetime import datetime, timezone from typing import Optional, Tuple @@ -41,12 +42,55 @@ def _bootstrap_file_bytes(password: str) -> bytes: def _persist_bootstrap_password(password: str) -> None: - """Write the bootstrap password 0600, with a trailing LF on every OS.""" - _BOOTSTRAP_PW_PATH.write_bytes(_bootstrap_file_bytes(password)) + """Write the bootstrap password 0600, with a trailing LF on every OS. + + Atomic: this can rewrite a live file, and a partial write would destroy the + only plaintext copy of the recovery credential. + """ + fd, tmp_name = tempfile.mkstemp( + prefix = f".{_BOOTSTRAP_PW_PATH.name}.", dir = _BOOTSTRAP_PW_PATH.parent + ) try: - os.chmod(_BOOTSTRAP_PW_PATH, 0o600) - except OSError: - pass + with os.fdopen(fd, "wb") as f: + f.write(_bootstrap_file_bytes(password)) + try: + os.chmod(tmp_name, 0o600) + except OSError: + pass + os.replace(tmp_name, _BOOTSTRAP_PW_PATH) + except BaseException: + try: + os.unlink(tmp_name) + except OSError: + pass + raise + + +def _read_persisted_bootstrap_password() -> Optional[str]: + """Read the persisted password, normalising the file if it is malformed.""" + if not _BOOTSTRAP_PW_PATH.is_file(): + return None + + # No caller handles a raise, so an unreadable file has to mean "no bootstrap + # password", not a dead backend. We write UTF-8, so bytes that will not + # decode are damage whose plaintext is worthless anyway. + try: + raw = _BOOTSTRAP_PW_PATH.read_bytes() + password = raw.decode("utf-8").strip() + except (OSError, UnicodeDecodeError): + return None + if not password: + return None + + # Older releases wrote no terminator, and text mode wrote CRLF on Windows, + # so upgrades kept the `cat` problem. Rewrite anything that isn't exactly + # "\n". Best-effort: a read-only auth dir must not fail startup. + if raw != _bootstrap_file_bytes(password): + try: + _persist_bootstrap_password(password) + except OSError: + pass + return password def generate_bootstrap_password() -> str: @@ -62,19 +106,10 @@ def generate_bootstrap_password() -> str: return _bootstrap_password # Persisted from a previous run? - if _BOOTSTRAP_PW_PATH.is_file(): - raw = _BOOTSTRAP_PW_PATH.read_bytes() - _bootstrap_password = raw.decode("utf-8").strip() - if _bootstrap_password: - # Older releases wrote no terminator, so upgrades kept the `cat` - # problem. Rewrite anything that isn't exactly "\n". - # Best-effort: a read-only auth dir must not fail startup. - if raw != _bootstrap_file_bytes(_bootstrap_password): - try: - _persist_bootstrap_password(_bootstrap_password) - except OSError: - pass - return _bootstrap_password + persisted = _read_persisted_bootstrap_password() + if persisted: + _bootstrap_password = persisted + return _bootstrap_password # First startup: generate a fresh passphrase. import diceware @@ -96,19 +131,14 @@ def get_bootstrap_password() -> Optional[str]: def _load_bootstrap_password() -> Optional[str]: - """Load an existing bootstrap password without creating one.""" + """Load an existing bootstrap password without creating one. + + This, not generate_bootstrap_password(), is the path an upgraded install + takes (ensure_default_admin short-circuits once the admin row exists), so + the normalisation has to happen here too. + """ global _bootstrap_password - _bootstrap_password = None - if _BOOTSTRAP_PW_PATH.is_file(): - # No caller handles a raise, so an unreadable file has to mean "no bootstrap - # password", not a dead backend. We write UTF-8, so bytes that will not - # decode are damage whose plaintext is worthless anyway. - try: - bootstrap_password = _BOOTSTRAP_PW_PATH.read_text(encoding = "utf-8").strip() - except (OSError, UnicodeDecodeError): - return _bootstrap_password - if bootstrap_password: - _bootstrap_password = bootstrap_password + _bootstrap_password = _read_persisted_bootstrap_password() return _bootstrap_password diff --git a/studio/backend/tests/test_desktop_auth.py b/studio/backend/tests/test_desktop_auth.py index 8c6a06ff75..5d53ad8898 100644 --- a/studio/backend/tests/test_desktop_auth.py +++ b/studio/backend/tests/test_desktop_auth.py @@ -153,30 +153,69 @@ def test_bootstrap_password_round_trips_across_a_restart_with_the_newline(): assert storage.generate_bootstrap_password() == original -def test_legacy_bootstrap_password_without_a_newline_is_migrated(): - # Upgrades kept the `cat` problem: the file is read, not rewritten. +@pytest.mark.parametrize("legacy", [ + b"legacy-bootstrap-secret", # written before the newline existed + b"legacy-bootstrap-secret\r\n", # written by text mode on Windows +]) +def test_upgrade_normalises_the_bootstrap_file(legacy): + # The upgrade path is ensure_default_admin() on an install that already has + # the admin row, which never reaches generate_bootstrap_password(). + seed_user() + storage._BOOTSTRAP_PW_PATH.write_bytes(legacy) + + storage.ensure_default_admin() + + assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"legacy-bootstrap-secret\n" + assert storage.get_bootstrap_password() == "legacy-bootstrap-secret" + + +def test_upgrade_normalises_when_the_admin_row_is_missing(): storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret") assert storage.generate_bootstrap_password() == "legacy-bootstrap-secret" assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"legacy-bootstrap-secret\n" -def test_legacy_bootstrap_password_with_crlf_is_migrated(): - storage._BOOTSTRAP_PW_PATH.write_bytes(b"crlf-bootstrap-secret\r\n") +def test_a_well_formed_bootstrap_file_is_not_rewritten(): + seed_user() + storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret\n") + mtime = storage._BOOTSTRAP_PW_PATH.stat().st_mtime_ns - assert storage.generate_bootstrap_password() == "crlf-bootstrap-secret" - assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"crlf-bootstrap-secret\n" + storage.ensure_default_admin() + + assert storage._BOOTSTRAP_PW_PATH.stat().st_mtime_ns == mtime -def test_migrating_a_legacy_bootstrap_password_survives_a_read_only_dir(monkeypatch): +def test_migration_failure_does_not_break_startup(monkeypatch): + seed_user() storage._BOOTSTRAP_PW_PATH.write_bytes(b"legacy-bootstrap-secret") def refuse(*args, **kwargs): - raise OSError("read-only auth dir") + raise PermissionError("read-only auth dir") - monkeypatch.setattr(Path, "write_bytes", refuse) + monkeypatch.setattr(storage.tempfile, "mkstemp", refuse) - assert storage.generate_bootstrap_password() == "legacy-bootstrap-secret" + storage.ensure_default_admin() + + assert storage.get_bootstrap_password() == "legacy-bootstrap-secret" + assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"legacy-bootstrap-secret" + + +def test_persisting_the_bootstrap_password_is_atomic(monkeypatch, tmp_path): + # A partial write would destroy the only plaintext recovery credential. + storage._persist_bootstrap_password("original-secret") + + def boom(src, dst): + raise OSError("crash before replace") + + monkeypatch.setattr(storage.os, "replace", boom) + with pytest.raises(OSError): + storage._persist_bootstrap_password("new-secret") + + assert storage._BOOTSTRAP_PW_PATH.read_bytes() == b"original-secret\n" + leftovers = [p.name for p in storage._BOOTSTRAP_PW_PATH.parent.iterdir() + if "bootstrap_password." in p.name] + assert leftovers == [] def test_ensure_default_admin_does_not_generate_for_empty_existing_bootstrap():