ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Run the newline migration on the path upgrades actually take

Browse source 
ensure_default_admin() short-circuits to _load_bootstrap_password() once the
admin row exists, so the normalisation added in the previous commit sat on
generate_bootstrap_password(), which only fresh installs reach. An upgraded
install kept its newline-less file. Both readers now share
_read_persisted_bootstrap_password().

Make the write atomic while it is here: it can now rewrite a live file, and a
partial write would destroy the only plaintext copy of the recovery
credential. Same mkstemp plus os.replace shape the CLI writer already uses.

Tests cover the upgrade path through ensure_default_admin(), a well-formed
file not being rewritten on every start, a failing migration not blocking
startup, and the atomic replace.
This commit is contained in:
Daniel Han 2026-07-29 04:23:51 +00:00
commit d404cade59
2 changed files with 109 additions and 40 deletions
Download patch file Download diff file Expand all files Collapse all files

90
studio/backend/auth/storage.py
View file

@ -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
# "<secret>\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 "<secret>\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

59
studio/backend/tests/test_desktop_auth.py
View file

@ -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():