fix(cloud): carry Bambu Cloud credentials across the auth on/off boundary

get_stored_token() reads the global Settings rows when auth is disabled and
User.cloud_token when it is enabled, so completing /auth/setup switched which
store the /cloud/* routes consult without moving the token. An account linked
before enabling auth was stranded: build_authenticated_cloud() returned None,
get_filament_info() skipped its cloud phase and answered 200 from local
fallbacks, and /cloud/devices began returning 401 -- all silently.

setup_auth() now migrates the global token onto the owning admin and deletes
the global rows; disable_auth() mirrors the hand-off back. Neither guesses:
setup migrates only when it creates the admin or exactly one exists, disable
declines to overwrite an existing global token. Region survives both hops.

Instances that already crossed the transition must re-link once.

backend/app/api/routes/auth.py

@@ -295,6 +295,38 @@ async def setup_auth(request: SetupRequest, db: AsyncSession = Depends(get_db)):
                         detail="Failed to create admin user",
                     )
 
+        if request.auth_enabled:
+            # Enabling auth flips cloud-credential storage from the global
+            # Settings rows to User.cloud_token. Carry any token linked while
+            # auth was off across to the owning admin, or /cloud/* silently
+            # degrades to local presets with no indication anything broke
+            # (#2530). Only migrate when there is exactly one obvious owner:
+            # handing another admin's session a Bambu credential is not a
+            # guess worth making.
+            from backend.app.api.routes.cloud import (
+                get_stored_token,
+                migrate_global_cloud_token_to_user,
+            )
+
+            if admin_created:
+                cloud_owner = admin_user
+            elif len(existing_admin_users) == 1:
+                cloud_owner = existing_admin_users[0]
+            else:
+                cloud_owner = None
+
+            if cloud_owner is not None:
+                if await migrate_global_cloud_token_to_user(db, cloud_owner):
+                    logger.info("Migrated global Bambu Cloud credentials to admin '%s'", cloud_owner.username)
+            else:
+                global_token, _, _ = await get_stored_token(db, None)
+                if global_token:
+                    logger.warning(
+                        "A Bambu Cloud account is linked globally but %s admins exist; "
+                        "leaving it unassigned. Re-link the account from Settings after login.",
+                        len(existing_admin_users),
+                    )
+
         # Set auth enabled and mark setup as completed
         await set_auth_enabled(db, request.auth_enabled)
         await set_setup_completed(db, True)
@@ -349,6 +381,14 @@ async def disable_auth(
         )
 
     try:
+        # Mirror of the migration in setup_auth: with auth off the cloud routes
+        # read the global Settings rows and never look at User.cloud_token, so
+        # hand this admin's credential over rather than stranding it (#2530).
+        from backend.app.api.routes.cloud import migrate_user_cloud_token_to_global
+
+        if await migrate_user_cloud_token_to_global(db, user):
+            logger.info("Migrated Bambu Cloud credentials from admin '%s' to global storage", user.username)
+
         await set_auth_enabled(db, False)
         await db.commit()
         logger.info("Authentication disabled by admin user: %s", user.username)

backend/app/api/routes/cloud.py

@@ -250,6 +250,69 @@ async def clear_token(db: AsyncSession, user: User | None = None) -> None:
     await db.commit()
 
 
+async def migrate_global_cloud_token_to_user(db: AsyncSession, user: User) -> bool:
+    """Move a globally-stored cloud token onto ``user`` (auth being enabled).
+
+    ``get_stored_token`` reads the global ``Settings`` rows when auth is off and
+    ``User.cloud_token`` when it's on. Enabling auth therefore switches which
+    column the cloud routes consult — without this migration the token linked
+    before setup is stranded in ``Settings``, ``build_authenticated_cloud``
+    returns ``None``, and every ``/cloud/*`` route silently degrades (#2530).
+
+    The global rows are deleted after the copy so the credential isn't left at
+    rest in a table nothing reads any more. Does **not** commit — the caller
+    owns the transaction. Returns True when a token was actually migrated.
+    """
+    token, email, region = await get_stored_token(db, None)
+    if not token:
+        return False
+
+    user.cloud_token = token
+    user.cloud_email = email
+    user.cloud_region = _normalise_region(region)
+
+    result = await db.execute(
+        select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY, CLOUD_REGION_KEY]))
+    )
+    for setting in result.scalars().all():
+        await db.delete(setting)
+    return True
+
+
+async def migrate_user_cloud_token_to_global(db: AsyncSession, user: User) -> bool:
+    """Move ``user``'s cloud token into global storage (auth being disabled).
+
+    The mirror of :func:`migrate_global_cloud_token_to_user`: once auth is off,
+    ``get_stored_token`` stops consulting ``User.cloud_token`` entirely, so the
+    admin who turns auth off would otherwise lose their own cloud link.
+
+    Refuses to overwrite an existing global token — a stale row from a previous
+    no-auth stint is still someone's credential, and clobbering it silently is
+    worse than leaving this admin to re-link. Does **not** commit. Returns True
+    when a token was actually migrated.
+    """
+    if not user.cloud_token:
+        return False
+
+    existing, _, _ = await get_stored_token(db, None)
+    if existing:
+        return False
+
+    for key, value in [
+        (CLOUD_TOKEN_KEY, user.cloud_token),
+        (CLOUD_EMAIL_KEY, user.cloud_email),
+        (CLOUD_REGION_KEY, _normalise_region(user.cloud_region)),
+    ]:
+        if value is None:
+            continue
+        db.add(Settings(key=key, value=value))
+
+    user.cloud_token = None
+    user.cloud_email = None
+    user.cloud_region = None
+    return True
+
+
 def _assert_api_key_can_access_cloud(api_key: APIKey) -> None:
     """Reject API keys that aren't authorised to read cloud data.
 

backend/tests/integration/test_cloud_token_auth_migration.py

@@ -0,0 +1,201 @@
+"""Cloud-credential migration across the auth on/off boundary (#2530).
+
+``get_stored_token`` reads global ``Settings`` rows when auth is disabled and
+``User.cloud_token`` when it's enabled. Toggling auth therefore switches which
+store the ``/cloud/*`` routes consult. Without an explicit hand-off the token
+is stranded in the store nobody reads: ``build_authenticated_cloud`` returns
+``None``, Phase 2 of ``get_filament_info`` is skipped entirely, and the caller
+sees a ``200`` full of local-preset fallbacks with no sign the cloud was never
+contacted. That silent degradation is what #2530 actually reported.
+
+These tests pin the hand-off in both directions, and — just as importantly —
+pin the two cases where Bambuddy must refuse to guess who owns a credential.
+"""
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from backend.app.api.routes.cloud import (
+    CLOUD_EMAIL_KEY,
+    CLOUD_REGION_KEY,
+    CLOUD_TOKEN_KEY,
+    get_stored_token,
+)
+from backend.app.core.auth import get_password_hash
+from backend.app.models.settings import Settings
+from backend.app.models.user import User
+
+
+async def _seed_global_token(db: AsyncSession, token: str = "tok-global", region: str = "china") -> None:
+    db.add(Settings(key=CLOUD_TOKEN_KEY, value=token))
+    db.add(Settings(key=CLOUD_EMAIL_KEY, value="owner@example.com"))
+    db.add(Settings(key=CLOUD_REGION_KEY, value=region))
+    await db.commit()
+
+
+async def _global_rows(db: AsyncSession) -> dict[str, str]:
+    rows = (
+        (
+            await db.execute(
+                select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY, CLOUD_REGION_KEY]))
+            )
+        )
+        .scalars()
+        .all()
+    )
+    return {r.key: r.value for r in rows}
+
+
+async def _make_admin(db: AsyncSession, username: str) -> User:
+    user = User(
+        username=username,
+        password_hash=get_password_hash("AdminPass1!"),
+        role="admin",
+        is_active=True,
+    )
+    db.add(user)
+    await db.commit()
+    await db.refresh(user)
+    return user
+
+
+# ---------------------------------------------------------------------------
+# auth OFF -> ON
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_setup_migrates_global_token_to_created_admin(async_client: AsyncClient, db_session: AsyncSession):
+    """The reporter's exact path: link cloud with auth off, then enable auth."""
+    await _seed_global_token(db_session)
+
+    resp = await async_client.post(
+        "/api/v1/auth/setup",
+        json={"auth_enabled": True, "admin_username": "admin", "admin_password": "AdminPass1!"},
+    )
+    assert resp.status_code == 200, resp.text
+
+    admin = (await db_session.execute(select(User).where(User.role == "admin"))).scalar_one()
+    token, email, region = await get_stored_token(db_session, admin)
+    assert token == "tok-global"
+    assert email == "owner@example.com"
+    assert region == "china", "region must survive the hop, not silently reset to global"
+
+    # Credential must not be left at rest in a table nothing reads any more.
+    assert await _global_rows(db_session) == {}
+
+
+@pytest.mark.asyncio
+async def test_setup_migrates_to_sole_pre_existing_admin(async_client: AsyncClient, db_session: AsyncSession):
+    """Re-enabling auth when exactly one admin already exists has one obvious owner."""
+    admin = await _make_admin(db_session, "solo")
+    await _seed_global_token(db_session, token="tok-solo")
+
+    resp = await async_client.post("/api/v1/auth/setup", json={"auth_enabled": True})
+    assert resp.status_code == 200, resp.text
+    assert resp.json()["admin_created"] is False
+
+    await db_session.refresh(admin)
+    token, _, _ = await get_stored_token(db_session, admin)
+    assert token == "tok-solo"
+    assert await _global_rows(db_session) == {}
+
+
+@pytest.mark.asyncio
+async def test_setup_refuses_to_guess_owner_when_multiple_admins(async_client: AsyncClient, db_session: AsyncSession):
+    """Two admins, one credential: handing it to either is a security decision we don't make."""
+    a = await _make_admin(db_session, "admin_a")
+    b = await _make_admin(db_session, "admin_b")
+    await _seed_global_token(db_session, token="tok-ambiguous")
+
+    resp = await async_client.post("/api/v1/auth/setup", json={"auth_enabled": True})
+    assert resp.status_code == 200, resp.text
+
+    await db_session.refresh(a)
+    await db_session.refresh(b)
+    assert a.cloud_token is None
+    assert b.cloud_token is None
+    # Left intact so the operator can re-link rather than lose it.
+    assert (await _global_rows(db_session))[CLOUD_TOKEN_KEY] == "tok-ambiguous"
+
+
+@pytest.mark.asyncio
+async def test_setup_with_auth_disabled_leaves_global_token_untouched(
+    async_client: AsyncClient, db_session: AsyncSession
+):
+    """Completing setup while declining auth must not move anything."""
+    await _seed_global_token(db_session, token="tok-stay")
+
+    resp = await async_client.post("/api/v1/auth/setup", json={"auth_enabled": False})
+    assert resp.status_code == 200, resp.text
+
+    assert (await _global_rows(db_session))[CLOUD_TOKEN_KEY] == "tok-stay"
+
+
+# ---------------------------------------------------------------------------
+# auth ON -> OFF
+# ---------------------------------------------------------------------------
+
+
+async def _admin_bearer(async_client: AsyncClient, username: str = "admin") -> str:
+    await async_client.post(
+        "/api/v1/auth/setup",
+        json={"auth_enabled": True, "admin_username": username, "admin_password": "AdminPass1!"},
+    )
+    login = await async_client.post(
+        "/api/v1/auth/login",
+        json={"username": username, "password": "AdminPass1!"},
+    )
+    return login.json()["access_token"]
+
+
+@pytest.mark.asyncio
+async def test_disable_auth_migrates_admin_token_to_global(async_client: AsyncClient, db_session: AsyncSession):
+    bearer = await _admin_bearer(async_client)
+    admin = (await db_session.execute(select(User).where(User.role == "admin"))).scalar_one()
+    admin.cloud_token = "tok-user"
+    admin.cloud_email = "user@example.com"
+    admin.cloud_region = "china"
+    await db_session.commit()
+
+    resp = await async_client.post("/api/v1/auth/disable", headers={"Authorization": f"Bearer {bearer}"})
+    assert resp.status_code == 200, resp.text
+
+    rows = await _global_rows(db_session)
+    assert rows[CLOUD_TOKEN_KEY] == "tok-user"
+    assert rows[CLOUD_REGION_KEY] == "china"
+
+    await db_session.refresh(admin)
+    assert admin.cloud_token is None, "credential must not be duplicated across both stores"
+
+    # And the no-auth read path now finds it.
+    token, _, _ = await get_stored_token(db_session, None)
+    assert token == "tok-user"
+
+
+@pytest.mark.asyncio
+async def test_disable_auth_does_not_clobber_existing_global_token(async_client: AsyncClient, db_session: AsyncSession):
+    """A stale global row is still somebody's credential — refuse rather than overwrite."""
+    bearer = await _admin_bearer(async_client)
+    admin = (await db_session.execute(select(User).where(User.role == "admin"))).scalar_one()
+    admin.cloud_token = "tok-user"
+    await db_session.commit()
+    await _seed_global_token(db_session, token="tok-preexisting")
+
+    resp = await async_client.post("/api/v1/auth/disable", headers={"Authorization": f"Bearer {bearer}"})
+    assert resp.status_code == 200, resp.text
+
+    assert (await _global_rows(db_session))[CLOUD_TOKEN_KEY] == "tok-preexisting"
+    await db_session.refresh(admin)
+    assert admin.cloud_token == "tok-user", "admin keeps their token when we decline to migrate"
+
+
+@pytest.mark.asyncio
+async def test_disable_auth_with_no_cloud_token_is_a_noop(async_client: AsyncClient, db_session: AsyncSession):
+    bearer = await _admin_bearer(async_client)
+
+    resp = await async_client.post("/api/v1/auth/disable", headers={"Authorization": f"Bearer {bearer}"})
+    assert resp.status_code == 200, resp.text
+    assert await _global_rows(db_session) == {}