fix(vault): harden the auto-generated master key file

Three fixes to how a node provisions its own master key when the operator
sets no ROUTSTR_SECRET_KEY:

- Publish the key atomically. It is written to a same-directory temp file,
  fsynced, then os.link-ed into place and the directory fsynced. os.link
  publishes the complete file in one step, so a crash mid-write can no longer
  strand an empty key at the final path that a later boot would read as corrupt
  and then fail to decrypt every secret under. os.link also refuses to clobber,
  so a racing worker that generated first keeps ownership and the loser adopts
  its key.
- Tighten loose permissions on read. A key file that is group/other-readable is
  repaired to 0600 rather than trusted, keeping an upgrading node booting.
- Stop printing the key value. The one-time notice names the file to back up and
  shouts the backup imperative, but no longer echoes the key itself, which would
  leak it into captured stdout / aggregated container logs; the durable 0600 file
  is the recovery path.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

routstr/core/vault.py

@@ -27,6 +27,7 @@ import hashlib
 import hmac
 import os
 import secrets
+import tempfile
 from pathlib import Path
 
 from cryptography.fernet import Fernet
@@ -99,7 +100,27 @@ def _read_key_file(path: Path) -> str | None:
         stored = path.read_text().strip()
     except OSError:
         return None
-    return stored or None
+    if not stored:
+        return None
+    _repair_key_file_perms(path)
+    return stored
+
+
+def _repair_key_file_perms(path: Path) -> None:
+    # A master key must never be group/other-readable. On POSIX, tighten loose
+    # permissions to owner-only (0600) rather than trust — or hard-fail on — a
+    # world-readable key; a friendlier repair keeps an upgrading node booting.
+    if os.name != "posix":
+        return
+    try:
+        mode = path.stat().st_mode
+    except OSError:
+        return
+    if mode & 0o077:
+        try:
+            os.chmod(path, 0o600)
+        except OSError:
+            pass
 
 
 def _load_secret_key() -> str | None:
@@ -107,7 +128,7 @@ def _load_secret_key() -> str | None:
     return os.environ.get("ROUTSTR_SECRET_KEY") or _read_key_file(_key_file_path())
 
 
-def _warn_generated_key(path: Path, key: str) -> None:
+def _warn_generated_key(path: Path) -> None:
     # stdout, not the logger: the operator must see this once (e.g. in
     # ``docker compose logs``), but it must never be persisted into the on-disk
     # log files the logger also writes. Mirrors the generated-admin-password
@@ -117,31 +138,66 @@ def _warn_generated_key(path: Path, key: str) -> None:
         f"rest and saved it to {path}.\n"
         "!! BACK UP THIS FILE. If it is lost, the encrypted secrets cannot be "
         "recovered and will have to be re-entered.\n"
-        "To manage the key yourself (e.g. from a secrets manager) set it in the "
-        f"environment instead:\n    ROUTSTR_SECRET_KEY={key}",
+        "To manage the key yourself (e.g. from a secrets manager) set "
+        "ROUTSTR_SECRET_KEY in the environment instead; the value is in the file "
+        "above.",
         flush=True,
     )
 
 
 def _generate_and_persist_key(path: Path) -> str:
-    """Generate a Fernet key, persist it owner-only, and warn once."""
+    """Generate a Fernet key, persist it owner-only and atomically, warn once.
+
+    The key is written to a temp file in the same directory, flushed durably,
+    then ``os.link``-ed into place. ``os.link`` publishes the complete file in a
+    single atomic step — a crash mid-write leaves only the temp file (which is
+    removed), never a half-written or empty key at the final path that a later
+    boot would read as corrupt. It also refuses to overwrite an existing key, so
+    a racing worker that generated first keeps ownership (secrets may already be
+    encrypted under its key); the loser adopts that key instead of clobbering it.
+    """
     key = Fernet.generate_key().decode()
     path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(
+        dir=path.parent, prefix=".routstr_secret.", suffix=".tmp"
+    )
+    tmp = Path(tmp_name)
     try:
-        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
-    except FileExistsError:
-        # Another worker won the race and wrote the key first; adopt theirs
-        # rather than clobber a key that secrets may already be encrypted under.
-        existing = _read_key_file(path)
-        if existing:
-            return existing
-        raise
-    with os.fdopen(fd, "w") as handle:
-        handle.write(key)
-    _warn_generated_key(path, key)
+        with os.fdopen(fd, "w") as handle:
+            handle.write(key)  # mkstemp already created it 0600
+            handle.flush()
+            os.fsync(handle.fileno())
+        try:
+            os.link(tmp, path)
+        except FileExistsError:
+            # A concurrent worker linked its key in first; adopt theirs rather
+            # than clobber a key that secrets may already be encrypted under.
+            existing = _read_key_file(path)
+            if existing:
+                return existing
+            raise
+        _fsync_dir(path.parent)
+    finally:
+        tmp.unlink(missing_ok=True)
+    _warn_generated_key(path)
     return key
 
 
+def _fsync_dir(directory: Path) -> None:
+    # Persist the new directory entry so the linked key survives a crash right
+    # after publish. Best-effort: not every platform lets you fsync a directory.
+    try:
+        dir_fd = os.open(directory, os.O_RDONLY)
+    except OSError:
+        return
+    try:
+        os.fsync(dir_fd)
+    except OSError:
+        pass
+    finally:
+        os.close(dir_fd)
+
+
 def ensure_secret_key() -> str:
     """Return the master key, provisioning one if the operator supplied none.
 

tests/integration/test_secret_bootstrap.py

@@ -202,10 +202,11 @@ async def test_legacy_nsec_without_secret_key_generates_and_encrypts(
     assert settings.nsec == NSEC_HEX
     assert settings.npub == derive_npub_from_nsec(NSEC_HEX)
     # ...and the operator is loudly told a key was generated and must be backed up
-    # (path + value shown) so an upgrade cannot silently create an unbacked key.
+    # (path shown, but never the key value) so an upgrade cannot silently create
+    # an unbacked key nor leak the key into captured stdout / aggregated logs.
     out = capsys.readouterr().out
     assert str(key_file) in out
-    assert key_file.read_text().strip() in out
+    assert key_file.read_text().strip() not in out
     assert "BACK UP" in out.upper()
 
 

tests/unit/test_vault.py

@@ -197,19 +197,20 @@ def test_encrypt_without_key_generates_and_persists_key_file(
     assert vault.decrypt(token) == "nsec1secret"
 
 
-def test_generated_key_warns_operator_with_path_and_value(
+def test_generated_key_warns_operator_with_path_not_value(
     generated_key_file: Path, capsys: pytest.CaptureFixture[str]
 ) -> None:
-    # The whole point of auto-generating is that an upgrading operator MUST NOT
-    # miss it: the notice names the file to back up, prints the key so it can be
-    # promoted into a secrets manager, and shouts the back-up imperative.
+    # The notice names the file to back up and shouts the back-up imperative so an
+    # upgrading operator cannot miss it, but it MUST NOT echo the key value: the
+    # secret lives in the 0600 file, and printing it would leak it into captured
+    # stdout / aggregated container logs.
     key_file = generated_key_file
     vault.encrypt("x")
 
     out = capsys.readouterr().out
     assert "ROUTSTR_SECRET_KEY" in out
     assert str(key_file) in out
-    assert key_file.read_text().strip() in out
+    assert key_file.read_text().strip() not in out
     assert "BACK UP" in out.upper()
 
 
@@ -229,6 +230,74 @@ def test_existing_key_file_is_reused_and_warns_only_once(
     assert capsys.readouterr().out == ""
 
 
+def test_generated_key_is_published_atomically(
+    generated_key_file: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # The key must appear at its final path only as a complete file: it is written
+    # to a temp file and atomically linked into place. If the publish (link) step
+    # fails — e.g. the process crashes — the destination must be absent, never a
+    # half-written or empty file that a later boot would read as a corrupt key and
+    # then refuse to decrypt every secret. No temp debris is left behind.
+    key_file = generated_key_file
+
+    def boom(src: object, dst: object) -> None:
+        raise OSError("crash during atomic publish")
+
+    monkeypatch.setattr(vault.os, "link", boom)
+
+    with pytest.raises(OSError):
+        vault.encrypt("x")
+
+    assert not key_file.exists()
+    assert list(key_file.parent.iterdir()) == []
+
+
+def test_racing_worker_adopts_winners_key_without_clobber(
+    generated_key_file: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # Two workers auto-generate a key concurrently. The first to link "wins" and
+    # its key is the one on disk; a worker that loses the link race must adopt the
+    # winner's key — secrets may already be encrypted under it — rather than
+    # clobber it or crash. Force the race deterministically: the winner publishes
+    # its key at the final path just as this worker tries to link, so os.link
+    # raises FileExistsError and the loser reads the winner's key back.
+    key_file = generated_key_file
+
+    def winner_links_first(src: object, dst: object) -> None:
+        key_file.write_text(KEY_A)  # the winner's already-published key
+        raise FileExistsError
+
+    monkeypatch.setattr(vault.os, "link", winner_links_first)
+
+    token = vault.encrypt("secret")
+
+    # The winner's key stays put and the loser encrypted under it, so the value
+    # round-trips under KEY_A even though this worker had generated its own key.
+    assert key_file.read_text().strip() == KEY_A
+    assert vault.decrypt(token) == "secret"
+    # The losing worker left no temp debris behind.
+    assert [p.name for p in key_file.parent.iterdir()] == [key_file.name]
+
+
+def test_loose_key_file_perms_are_tightened_on_read(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    # A key file left group/other-readable (e.g. written under a loose umask
+    # before this hardening, or by a careless operator) is a leaked-secret risk.
+    # Reading it repairs the permissions to owner-only rather than trusting a
+    # world-readable master key, while still using the key so boot is not broken.
+    monkeypatch.delenv("ROUTSTR_SECRET_KEY", raising=False)
+    key_file = tmp_path / "routstr_secret.key"
+    key_file.write_text(KEY_A)
+    key_file.chmod(0o644)
+    monkeypatch.setenv("ROUTSTR_SECRET_KEY_FILE", str(key_file))
+
+    token = vault.encrypt("secret")  # reads the loose file, repairs its perms
+
+    assert key_file.stat().st_mode & 0o077 == 0
+    assert vault.decrypt(token) == "secret"
+
+
 def test_env_key_takes_precedence_over_key_file(
     monkeypatch: pytest.MonkeyPatch, tmp_path: Path
 ) -> None: