fix(audit-g5c): refuse silent Fernet key regeneration on partial restore

Audit 2026-06-02: `_load_or_create_key` silently minted a new Fernet
key whenever the key file was missing — no log, no warning. The
failure mode the audit flagged: a partial disaster restore where the
DB was restored but `/images/secrets/` was lost would produce a
working-looking system in which every authenticated download fails
AUTH_ERROR until the operator re-uploads every credential by hand.

Two opt-ins now needed for auto-creation:
  1. Explicit `bootstrap_ok=True` kwarg (tests, scripts), OR
  2. `CURATOR_BOOTSTRAP_NEW_KEY=1` env var (operator first-time setup)

Otherwise the constructor raises `MissingCredentialKey` so the app
fails fast at startup and the operator can restore the key file
from backup before encrypted_blob rows go undecryptable.

Also: docstring path was wrong (said "images/data root" but actual
location is `/images/secrets/credential_key.b64`) — corrected.

Tests updated to pass `bootstrap_ok=True` explicitly, and two new
tests cover the safety behavior (missing-key-raises, env-var-bootstraps).

tests/test_credential_crypto.py

@@ -14,7 +14,7 @@ pytestmark = pytest.mark.integration
 
 def test_load_or_create_writes_key_with_mode_0600(tmp_path):
     key_path = tmp_path / "secrets" / "credential_key.b64"
-    CredentialCrypto(key_path)
+    CredentialCrypto(key_path, bootstrap_ok=True)
     assert key_path.exists()
     mode = stat.S_IMODE(os.stat(key_path).st_mode)
     assert mode == 0o600
@@ -25,9 +25,9 @@ def test_load_or_create_writes_key_with_mode_0600(tmp_path):
 
 def test_load_existing_key_is_idempotent(tmp_path):
     key_path = tmp_path / "credential_key.b64"
-    crypto1 = CredentialCrypto(key_path)
+    crypto1 = CredentialCrypto(key_path, bootstrap_ok=True)
     contents_after_first = key_path.read_bytes()
-    crypto2 = CredentialCrypto(key_path)
+    crypto2 = CredentialCrypto(key_path, bootstrap_ok=True)
     contents_after_second = key_path.read_bytes()
     assert contents_after_first == contents_after_second
     # And both crypto instances decrypt each other's ciphertext
@@ -36,7 +36,7 @@ def test_load_existing_key_is_idempotent(tmp_path):
 
 
 def test_encrypt_decrypt_round_trip(tmp_path):
-    crypto = CredentialCrypto(tmp_path / "k")
+    crypto = CredentialCrypto(tmp_path / "k", bootstrap_ok=True)
     plaintext = "domain.com\tTRUE\t/\tTRUE\t1700000000\tname\tvalue"
     ct = crypto.encrypt(plaintext)
     assert isinstance(ct, bytes)
@@ -45,8 +45,27 @@ def test_encrypt_decrypt_round_trip(tmp_path):
 
 
 def test_decrypt_with_wrong_key_raises(tmp_path):
-    crypto_a = CredentialCrypto(tmp_path / "a")
-    crypto_b = CredentialCrypto(tmp_path / "b")
+    crypto_a = CredentialCrypto(tmp_path / "a", bootstrap_ok=True)
+    crypto_b = CredentialCrypto(tmp_path / "b", bootstrap_ok=True)
     ct = crypto_a.encrypt("secret")
     with pytest.raises(InvalidCredentialBlob):
         crypto_b.decrypt(ct)
+
+
+def test_missing_key_without_bootstrap_raises(tmp_path, monkeypatch):
+    """Audit 2026-06-02: without explicit opt-in, a missing key file
+    is a fatal startup error — silent regeneration on partial restore
+    would make every existing Credential row undecryptable."""
+    from backend.app.services.credential_crypto import MissingCredentialKey
+    monkeypatch.delenv("CURATOR_BOOTSTRAP_NEW_KEY", raising=False)
+    with pytest.raises(MissingCredentialKey):
+        CredentialCrypto(tmp_path / "absent.b64")
+
+
+def test_missing_key_with_env_var_bootstraps(tmp_path, monkeypatch):
+    """The env var CURATOR_BOOTSTRAP_NEW_KEY=1 is the operator's
+    first-time-setup opt-in for auto-creating the key file."""
+    monkeypatch.setenv("CURATOR_BOOTSTRAP_NEW_KEY", "1")
+    key_path = tmp_path / "bootstrap.b64"
+    CredentialCrypto(key_path)  # no bootstrap_ok kwarg — relies on env
+    assert key_path.exists()

tests/test_credential_service.py

@@ -21,7 +21,7 @@ _NETSCAPE = (
 
 @pytest.fixture
 def crypto(tmp_path):
-    return CredentialCrypto(tmp_path / "k")
+    return CredentialCrypto(tmp_path / "k", bootstrap_ok=True)
 
 
 @pytest.mark.asyncio

tests/test_download_service.py

@@ -130,7 +130,7 @@ async def test_download_source_attaches_written_files(
         thumbnailer=Thumbnailer(images_root=images_root),
         settings=sync_settings,
     )
-    crypto = CredentialCrypto(tmp_path / "key.b64")
+    crypto = CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
     cred_service = CredentialService(db, crypto)
 
     svc = DownloadService(
@@ -405,7 +405,7 @@ async def test_backfill_decrements_after_run(
         session=db_sync, images_root=images_root, import_root=images_root,
         thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
     )
-    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
     svc = DownloadService(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
@@ -446,7 +446,7 @@ async def test_backfill_auto_resets_on_clean_zero_files(
         thumbnailer=Thumbnailer(images_root=tmp_path / "images"),
         settings=sync_settings,
     )
-    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
     svc = DownloadService(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
@@ -484,7 +484,7 @@ async def test_tick_mode_does_not_touch_backfill_counter(
         thumbnailer=Thumbnailer(images_root=tmp_path / "images"),
         settings=sync_settings,
     )
-    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
     svc = DownloadService(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
@@ -529,7 +529,7 @@ async def test_partial_error_type_maps_to_ok_status(
         session=db_sync, images_root=images_root, import_root=images_root,
         thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
     )
-    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
     svc = DownloadService(
         async_session=db, sync_session=db_sync,
         gdl=fake_gdl, importer=importer, cred_service=cred_service,
@@ -582,7 +582,7 @@ async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
         session=db_sync, images_root=images_root, import_root=images_root,
         thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
     )
-    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64"))
+    cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
 
     # Capture the IDs that the orchestrator hands off to each Celery task.
     # The .delay() shim runs inside DownloadService._phase3_persist (lazy