Fix cloud profiles shared across all users (#665)

  Cloud credentials were stored globally — one Bambu Cloud account per
  Bambuddy instance. When auth was enabled, any user logging into Cloud
  overwrote everyone else's credentials. Credentials are now stored
  per-user: each user gets their own independent Cloud login.

  Also fixed cloud data endpoints (settings, fields, preset CRUD)
  requiring settings:read/settings:update permissions instead of
  cloud:auth — users who had "Cloud Auth" enabled but "Settings"
  disabled couldn't load profiles after logging in.

CHANGELOG.md

@@ -24,6 +24,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **Prometheus Build Info Metric** ([#633](https://github.com/maziggy/bambuddy/pull/633)) — Added a `bambuddy_build_info` gauge metric to the Prometheus metrics endpoint, exposing the application version, Python version, platform, and architecture as labels. Follows the standard Prometheus `_build_info` convention for dashboards and version-change alerting. Contributed by @sw1nn.
 
 ### Fixed
+- **Cloud Profiles Shared Across All Users** ([#665](https://github.com/maziggy/bambuddy/issues/665)) — When authentication was enabled, Bambu Cloud credentials were stored globally — one account per Bambuddy instance. If User A logged into Cloud, every other user saw User A's account and profiles. User B logging in would overwrite User A's credentials. Cloud credentials are now stored per-user: each user logs into their own Bambu Cloud account independently. When auth is disabled (single-user mode), behavior is unchanged. Also fixed cloud data endpoints (`/cloud/settings`, `/cloud/fields`, preset CRUD) requiring `settings:read` / `settings:update` permissions instead of `cloud:auth` — users who had "Cloud Auth" enabled but "Settings" disabled couldn't load profiles after logging in. Reported by @cadtoolbox.
 - **Local Profiles Not Shown in AMS Slot Configuration** — Imported local filament profiles were hidden in the AMS slot configure modal when a printer model was set. The `compatible_printers` filter parsed the stored JSON array as a semicolon-delimited string, so the matching always failed and every local preset was silently skipped. Removed the filter entirely — user-imported profiles should be available on any printer.
 - **Interface Aliases Not Shown in Virtual Printer Interface Select** — Interface aliases (e.g. `eth0:1`) added for multi-virtual-printer setups were invisible in the bind IP dropdown. The Docker image didn't include `iproute2`, so the `ip` command wasn't available and the code fell back to ioctl-based enumeration which can only return one IP per interface. Added `iproute2` to the Docker image.
 - **P2S Camera Stream Disconnects After a Few Seconds** ([#661](https://github.com/maziggy/bambuddy/issues/661)) — The P2S firmware drops RTSP sessions after a few seconds with an I/O error. The backend treated this as a fatal failure, ending the MJPEG stream and forcing the frontend through a full reconnection cycle (stop → start → brief connection → fail → repeat). Added transparent auto-reconnection: when ffmpeg's RTSP connection dies, it respawns immediately and continues streaming MJPEG frames to the browser without interruption. Reported by @ddetton, confirmed by @DMoenning.

README.md

@@ -203,6 +203,7 @@ Perfect for remote print farms, traveling makers, or accessing your home printer
 - Comprehensive API protection (200+ endpoints secured)
 - User management (create, edit, delete, groups)
 - User activity tracking (who uploaded archives, library files, queued prints, started prints)
+- **Per-user Bambu Cloud accounts** — Each user has their own independent Cloud login for profiles
 - **Advanced Auth via Email** — SMTP integration for automated user onboarding and self-service password resets
 - Admin creates users with email — system sends secure random password automatically
 - Users can reset their own password from the login screen (no admin needed)

backend/app/api/routes/cloud.py

@@ -50,15 +50,37 @@ CLOUD_TOKEN_KEY = "bambu_cloud_token"
 CLOUD_EMAIL_KEY = "bambu_cloud_email"
 
 
-async def get_stored_token(db: AsyncSession) -> tuple[str | None, str | None]:
-    """Get stored cloud token and email from database."""
+async def get_stored_token(db: AsyncSession, user: User | None = None) -> tuple[str | None, str | None]:
+    """Get stored cloud token and email.
+
+    When a user is provided (auth enabled), returns that user's per-user credentials.
+    When user is None (auth disabled), falls back to global Settings table.
+    """
+    if user is not None:
+        return user.cloud_token, user.cloud_email
+
+    # Fallback: global storage (auth disabled)
     result = await db.execute(select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY])))
     settings = {s.key: s.value for s in result.scalars().all()}
     return settings.get(CLOUD_TOKEN_KEY), settings.get(CLOUD_EMAIL_KEY)
 
 
-async def store_token(db: AsyncSession, token: str, email: str) -> None:
-    """Store cloud token and email in database."""
+async def store_token(db: AsyncSession, token: str, email: str, user: User | None = None) -> None:
+    """Store cloud token and email.
+
+    When a user is provided (auth enabled), stores on the user record.
+    When user is None (auth disabled), stores in global Settings table.
+    """
+    if user is not None:
+        # User object is from the auth dependency's session (detached),
+        # so use a direct UPDATE via the route's db session.
+        from sqlalchemy import update
+
+        await db.execute(update(User).where(User.id == user.id).values(cloud_token=token, cloud_email=email))
+        await db.commit()
+        return
+
+    # Fallback: global storage (auth disabled)
     for key, value in [(CLOUD_TOKEN_KEY, token), (CLOUD_EMAIL_KEY, email)]:
         result = await db.execute(select(Settings).where(Settings.key == key))
         setting = result.scalar_one_or_none()
@@ -69,8 +91,20 @@ async def store_token(db: AsyncSession, token: str, email: str) -> None:
     await db.commit()
 
 
-async def clear_token(db: AsyncSession) -> None:
-    """Clear stored cloud token and email."""
+async def clear_token(db: AsyncSession, user: User | None = None) -> None:
+    """Clear stored cloud token and email.
+
+    When a user is provided (auth enabled), clears that user's credentials.
+    When user is None (auth disabled), clears from global Settings table.
+    """
+    if user is not None:
+        from sqlalchemy import update
+
+        await db.execute(update(User).where(User.id == user.id).values(cloud_token=None, cloud_email=None))
+        await db.commit()
+        return
+
+    # Fallback: global storage (auth disabled)
     result = await db.execute(select(Settings).where(Settings.key.in_([CLOUD_TOKEN_KEY, CLOUD_EMAIL_KEY])))
     for setting in result.scalars().all():
         await db.delete(setting)
@@ -80,10 +114,10 @@ async def clear_token(db: AsyncSession) -> None:
 @router.get("/status", response_model=CloudAuthStatus)
 async def get_auth_status(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """Get current cloud authentication status."""
-    token, email = await get_stored_token(db)
+    token, email = await get_stored_token(db, current_user)
     cloud = get_cloud_service()
 
     if token:
@@ -99,7 +133,7 @@ async def get_auth_status(
 async def login(
     request: CloudLoginRequest,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Initiate login to Bambu Cloud.
@@ -113,15 +147,12 @@ async def login(
     """
     cloud = get_cloud_service()
 
-    # Store email temporarily for verification step
-    await store_token(db, "", request.email)
-
     try:
         result = await cloud.login_request(request.email, request.password)
 
         if result.get("success") and cloud.access_token:
             # Direct login succeeded (rare)
-            await store_token(db, cloud.access_token, request.email)
+            await store_token(db, cloud.access_token, request.email, current_user)
 
         return CloudLoginResponse(
             success=result.get("success", False),
@@ -140,7 +171,7 @@ async def login(
 async def verify_code(
     request: CloudVerifyRequest,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Complete login with verification code (email or TOTP).
@@ -163,7 +194,7 @@ async def verify_code(
             result = await cloud.verify_code(request.email, request.code)
 
         if result.get("success") and cloud.access_token:
-            await store_token(db, cloud.access_token, request.email)
+            await store_token(db, cloud.access_token, request.email, current_user)
 
         return CloudLoginResponse(
             success=result.get("success", False),
@@ -180,7 +211,7 @@ async def verify_code(
 async def set_token(
     request: CloudTokenRequest,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Set access token directly.
@@ -193,7 +224,7 @@ async def set_token(
     # Verify token works by trying to get profile
     try:
         await cloud.get_user_profile()
-        await store_token(db, request.access_token, "token-auth")
+        await store_token(db, request.access_token, "token-auth", current_user)
         return CloudAuthStatus(is_authenticated=True, email="token-auth")
     except BambuCloudError:
         cloud.logout()
@@ -203,12 +234,12 @@ async def set_token(
 @router.post("/logout")
 async def logout(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """Log out of Bambu Cloud."""
     cloud = get_cloud_service()
     cloud.logout()
-    await clear_token(db)
+    await clear_token(db, current_user)
     return {"success": True}
 
 
@@ -216,14 +247,14 @@ async def logout(
 async def get_slicer_settings(
     version: str = "02.04.00.70",
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Get all slicer settings (filament, printer, process presets).
 
     Requires authentication.
     """
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -281,7 +312,7 @@ async def get_slicer_settings(
 
         return result
     except BambuCloudAuthError:
-        await clear_token(db)
+        await clear_token(db, current_user)
         raise HTTPException(status_code=401, detail="Authentication expired")
     except BambuCloudError as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -291,14 +322,14 @@ async def get_slicer_settings(
 async def get_setting_detail(
     setting_id: str,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Get detailed information for a specific setting/preset.
 
     Returns the full preset configuration.
     """
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -312,7 +343,7 @@ async def get_setting_detail(
         data = await cloud.get_setting_detail(setting_id)
         return data
     except BambuCloudAuthError:
-        await clear_token(db)
+        await clear_token(db, current_user)
         raise HTTPException(status_code=401, detail="Authentication expired")
     except BambuCloudError as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -322,7 +353,7 @@ async def get_setting_detail(
 async def get_filament_presets(
     version: str = "02.04.00.70",
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """
     Get just filament presets (convenience endpoint).
@@ -330,7 +361,7 @@ async def get_filament_presets(
     Returns all filament presets with custom presets first.
     Uses the same cache as get_slicer_settings.
     """
-    settings = await get_slicer_settings(version=version, db=db)
+    settings = await get_slicer_settings(version=version, db=db, current_user=current_user)
     return settings.filament
 
 
@@ -512,7 +543,7 @@ _filament_id_to_setting_id = filament_id_to_setting_id
 async def get_filament_info(
     setting_ids: list[str] = Body(...),
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """
     Get filament preset info (name and K value) for multiple setting IDs.
@@ -545,7 +576,7 @@ async def get_filament_info(
 
     # Phase 2: Try cloud for uncached IDs
     if unresolved_ids:
-        token, _ = await get_stored_token(db)
+        token, _ = await get_stored_token(db, current_user)
         if token:
             cloud = get_cloud_service()
             cloud.set_token(token)
@@ -590,14 +621,14 @@ async def get_filament_info(
 @router.get("/devices", response_model=list[CloudDevice])
 async def get_devices(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
 ):
     """
     Get list of bound printer devices.
 
     Returns printers registered to the user's Bambu account.
     """
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -622,7 +653,7 @@ async def get_devices(
             for d in devices
         ]
     except BambuCloudAuthError:
-        await clear_token(db)
+        await clear_token(db, current_user)
         raise HTTPException(status_code=401, detail="Authentication expired")
     except BambuCloudError as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -631,7 +662,7 @@ async def get_devices(
 @router.get("/firmware-updates", response_model=FirmwareUpdatesResponse)
 async def get_firmware_updates(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.FIRMWARE_READ),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.FIRMWARE_READ),
 ):
     """
     Check for firmware updates for all bound devices.
@@ -644,7 +675,7 @@ async def get_firmware_updates(
 
     Requires cloud authentication.
     """
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -701,7 +732,7 @@ async def get_firmware_updates(
         return FirmwareUpdatesResponse(updates=updates, updates_available=updates_available)
 
     except BambuCloudAuthError:
-        await clear_token(db)
+        await clear_token(db, current_user)
         raise HTTPException(status_code=401, detail="Authentication expired")
     except BambuCloudError as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -711,7 +742,7 @@ async def get_firmware_updates(
 async def create_setting(
     request: SlicerSettingCreate,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Create a new slicer preset/setting.
@@ -721,7 +752,7 @@ async def create_setting(
 
     Type should be: 'filament', 'print', or 'printer'
     """
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -741,7 +772,7 @@ async def create_setting(
         )
         return data
     except BambuCloudAuthError:
-        await clear_token(db)
+        await clear_token(db, current_user)
         raise HTTPException(status_code=401, detail="Authentication expired")
     except BambuCloudError as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -752,14 +783,14 @@ async def update_setting(
     setting_id: str,
     request: SlicerSettingUpdate,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Update an existing slicer preset/setting.
 
     Updates the preset's name and/or settings on Bambu Cloud.
     """
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -777,7 +808,7 @@ async def update_setting(
         )
         return data
     except BambuCloudAuthError:
-        await clear_token(db)
+        await clear_token(db, current_user)
         raise HTTPException(status_code=401, detail="Authentication expired")
     except BambuCloudError as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -787,14 +818,14 @@ async def update_setting(
 async def delete_setting(
     setting_id: str,
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_UPDATE),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Delete a slicer preset/setting.
 
     Removes the preset from Bambu Cloud. This cannot be undone.
     """
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         raise HTTPException(status_code=401, detail="Not authenticated")
 
@@ -811,7 +842,7 @@ async def delete_setting(
             message=result.get("message", "Setting deleted"),
         )
     except BambuCloudAuthError:
-        await clear_token(db)
+        await clear_token(db, current_user)
         raise HTTPException(status_code=401, detail="Authentication expired")
     except BambuCloudError as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -874,7 +905,7 @@ _filament_id_name_cache_time: float = 0
 @router.get("/filament-id-map")
 async def get_filament_id_map(
     db: AsyncSession = Depends(get_db),
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
+    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.FILAMENTS_READ),
 ):
     """
     Get filament_id → name mapping for user cloud presets.
@@ -891,7 +922,7 @@ async def get_filament_id_map(
     if _filament_id_name_cache and time.time() - _filament_id_name_cache_time < FILAMENT_CACHE_TTL:
         return _filament_id_name_cache
 
-    token, _ = await get_stored_token(db)
+    token, _ = await get_stored_token(db, current_user)
     if not token:
         return _filament_id_name_cache or {}
 
@@ -930,7 +961,7 @@ async def get_filament_id_map(
 @router.get("/fields/{preset_type}")
 async def get_preset_fields(
     preset_type: Literal["filament", "print", "process", "printer"],
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Get field definitions for a preset type.
@@ -951,7 +982,7 @@ async def get_preset_fields(
 
 @router.get("/fields")
 async def get_all_preset_fields(
-    _: User | None = RequirePermissionIfAuthEnabled(Permission.SETTINGS_READ),
+    _: User | None = RequirePermissionIfAuthEnabled(Permission.CLOUD_AUTH),
 ):
     """
     Get all field definitions for all preset types.

backend/app/core/database.py

@@ -1407,6 +1407,16 @@ async def run_migrations(conn):
     except OperationalError:
         pass  # Already applied
 
+    # Migration: Add per-user Bambu Cloud credential columns
+    try:
+        await conn.execute(text("ALTER TABLE users ADD COLUMN cloud_token VARCHAR(500)"))
+    except OperationalError:
+        pass  # Already applied
+    try:
+        await conn.execute(text("ALTER TABLE users ADD COLUMN cloud_email VARCHAR(255)"))
+    except OperationalError:
+        pass  # Already applied
+
     # Cleanup: Remove obsolete settings keys that are no longer used
     obsolete_keys = ["slicer_binary_path"]
     for key in obsolete_keys:

backend/app/models/user.py

@@ -33,6 +33,10 @@ class User(Base):
     created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
     updated_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), onupdate=func.now())
 
+    # Per-user Bambu Cloud credentials (when auth is enabled, each user has their own)
+    cloud_token: Mapped[str | None] = mapped_column(String(500), nullable=True, default=None)
+    cloud_email: Mapped[str | None] = mapped_column(String(255), nullable=True, default=None)
+
     # Relationship to groups through association table
     groups: Mapped[list[Group]] = relationship(
         "Group",

backend/tests/integration/test_cloud_auth.py

@@ -0,0 +1,391 @@
+"""Integration tests for per-user cloud credentials and cloud endpoint permissions.
+
+Regression tests for:
+- Per-user cloud token storage (when auth enabled)
+- Global fallback (when auth disabled)
+- Cloud endpoints use CLOUD_AUTH permission (not SETTINGS_READ)
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+
+
+class TestPerUserCloudCredentials:
+    """Tests that cloud credentials are stored per-user when auth is enabled."""
+
+    @pytest.fixture
+    async def user_with_cloud_auth(self, db_session):
+        """Create a user with CLOUD_AUTH permission via a group."""
+        from backend.app.core.auth import get_password_hash
+        from backend.app.models.group import Group
+        from backend.app.models.user import User
+
+        group = Group(
+            name="CloudUsers",
+            permissions=["cloud:auth", "filaments:read", "printers:read", "firmware:read"],
+        )
+        db_session.add(group)
+        await db_session.flush()
+
+        user = User(
+            username="clouduser",
+            password_hash=get_password_hash("testpass123"),
+            role="user",
+        )
+        db_session.add(user)
+        await db_session.flush()
+        user.groups.append(group)
+        await db_session.commit()
+        await db_session.refresh(user)
+        return user
+
+    @pytest.fixture
+    async def second_user_with_cloud_auth(self, db_session):
+        """Create a second user with CLOUD_AUTH permission."""
+        from sqlalchemy import select
+
+        from backend.app.core.auth import get_password_hash
+        from backend.app.models.group import Group
+        from backend.app.models.user import User
+
+        result = await db_session.execute(select(Group).where(Group.name == "CloudUsers"))
+        group = result.scalar_one_or_none()
+        if not group:
+            group = Group(
+                name="CloudUsers2",
+                permissions=["cloud:auth", "filaments:read", "printers:read", "firmware:read"],
+            )
+            db_session.add(group)
+            await db_session.flush()
+
+        user = User(
+            username="clouduser2",
+            password_hash=get_password_hash("testpass456"),
+            role="user",
+        )
+        db_session.add(user)
+        await db_session.flush()
+        user.groups.append(group)
+        await db_session.commit()
+        await db_session.refresh(user)
+        return user
+
+    @pytest.fixture
+    async def cloud_auth_token(self, user_with_cloud_auth, async_client: AsyncClient):
+        """Get auth token for user with cloud permissions."""
+        response = await async_client.post(
+            "/api/v1/auth/login",
+            json={"username": "clouduser", "password": "testpass123"},
+        )
+        if response.status_code == 200:
+            return response.json().get("access_token")
+        return None
+
+    @pytest.fixture
+    async def second_auth_token(self, second_user_with_cloud_auth, async_client: AsyncClient):
+        """Get auth token for second user."""
+        response = await async_client.post(
+            "/api/v1/auth/login",
+            json={"username": "clouduser2", "password": "testpass456"},
+        )
+        if response.status_code == 200:
+            return response.json().get("access_token")
+        return None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cloud_status_returns_not_authenticated_by_default(self, async_client: AsyncClient):
+        """Cloud status should show not authenticated when no token is stored."""
+        with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
+            response = await async_client.get("/api/v1/cloud/status")
+            assert response.status_code == 200
+            data = response.json()
+            assert data["is_authenticated"] is False
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cloud_status_accessible_when_auth_disabled(self, async_client: AsyncClient):
+        """Cloud endpoints should work when auth is disabled (global fallback)."""
+        with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
+            response = await async_client.get("/api/v1/cloud/status")
+            assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cloud_status_requires_auth_when_enabled(self, async_client: AsyncClient):
+        """Cloud endpoints should require auth when auth is enabled."""
+        with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
+            response = await async_client.get("/api/v1/cloud/status")
+            assert response.status_code == 401
+
+
+class TestCloudEndpointPermissions:
+    """Tests that cloud endpoints use CLOUD_AUTH permission, not SETTINGS_READ.
+
+    Uses JWT tokens created directly (not via login endpoint) to avoid
+    test infrastructure complexity with user creation across sessions.
+    """
+
+    @pytest.fixture
+    async def settings_only_setup(self, async_client: AsyncClient):
+        """Create user with settings:read but NOT cloud:auth, return JWT."""
+        from backend.app.core.auth import create_access_token, get_password_hash
+        from backend.app.core.database import async_session
+        from backend.app.models.group import Group
+        from backend.app.models.user import User
+
+        async with async_session() as db:
+            group = Group(name="SettingsReaders", permissions=["settings:read"])
+            db.add(group)
+            user = User(
+                username="settingsuser",
+                password_hash=get_password_hash("testpass123"),
+                role="user",
+            )
+            db.add(user)
+            await db.commit()
+            await db.refresh(group)
+            await db.refresh(user)
+
+            from sqlalchemy import text
+
+            await db.execute(
+                text("INSERT INTO user_groups (user_id, group_id) VALUES (:uid, :gid)"),
+                {"uid": user.id, "gid": group.id},
+            )
+            await db.commit()
+
+        return create_access_token(data={"sub": "settingsuser"})
+
+    @pytest.fixture
+    async def cloud_only_setup(self, async_client: AsyncClient):
+        """Create user with cloud:auth but NOT settings:read, return JWT."""
+        from backend.app.core.auth import create_access_token, get_password_hash
+        from backend.app.core.database import async_session
+        from backend.app.models.group import Group
+        from backend.app.models.user import User
+
+        async with async_session() as db:
+            group = Group(name="CloudOnly", permissions=["cloud:auth"])
+            db.add(group)
+            user = User(
+                username="cloudonly",
+                password_hash=get_password_hash("testpass123"),
+                role="user",
+            )
+            db.add(user)
+            await db.commit()
+            await db.refresh(group)
+            await db.refresh(user)
+
+            from sqlalchemy import text
+
+            await db.execute(
+                text("INSERT INTO user_groups (user_id, group_id) VALUES (:uid, :gid)"),
+                {"uid": user.id, "gid": group.id},
+            )
+            await db.commit()
+
+        return create_access_token(data={"sub": "cloudonly"})
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cloud_settings_requires_cloud_auth_not_settings_read(
+        self, async_client: AsyncClient, settings_only_setup, cloud_only_setup
+    ):
+        """GET /cloud/settings should require CLOUD_AUTH, not SETTINGS_READ.
+
+        Regression test: previously used SETTINGS_READ which blocked users who
+        had cloud:auth permission but not settings:read.
+        """
+        with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
+            # User with only settings:read should be denied
+            response = await async_client.get(
+                "/api/v1/cloud/settings",
+                headers={"Authorization": f"Bearer {settings_only_setup}"},
+            )
+            assert response.status_code == 403
+
+            # User with cloud:auth should be allowed (will get 401 since no cloud token,
+            # but NOT 403 — permission check passes)
+            response = await async_client.get(
+                "/api/v1/cloud/settings",
+                headers={"Authorization": f"Bearer {cloud_only_setup}"},
+            )
+            assert response.status_code == 401  # No cloud token, but permission OK
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cloud_status_requires_cloud_auth(
+        self, async_client: AsyncClient, settings_only_setup, cloud_only_setup
+    ):
+        """GET /cloud/status should require CLOUD_AUTH."""
+        with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
+            # settings:read only → 403
+            response = await async_client.get(
+                "/api/v1/cloud/status",
+                headers={"Authorization": f"Bearer {settings_only_setup}"},
+            )
+            assert response.status_code == 403
+
+            # cloud:auth → 200
+            response = await async_client.get(
+                "/api/v1/cloud/status",
+                headers={"Authorization": f"Bearer {cloud_only_setup}"},
+            )
+            assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_cloud_fields_requires_cloud_auth(
+        self, async_client: AsyncClient, settings_only_setup, cloud_only_setup
+    ):
+        """GET /cloud/fields should require CLOUD_AUTH, not SETTINGS_READ."""
+        with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
+            # settings:read only → 403
+            response = await async_client.get(
+                "/api/v1/cloud/fields",
+                headers={"Authorization": f"Bearer {settings_only_setup}"},
+            )
+            assert response.status_code == 403
+
+            # cloud:auth → 200
+            response = await async_client.get(
+                "/api/v1/cloud/fields",
+                headers={"Authorization": f"Bearer {cloud_only_setup}"},
+            )
+            assert response.status_code == 200
+
+
+class TestCloudTokenStorage:
+    """Unit-level tests for the token storage functions."""
+
+    @pytest.mark.asyncio
+    async def test_get_stored_token_returns_none_when_no_user_no_global(self, db_session):
+        """get_stored_token with user=None and no global token returns (None, None)."""
+        from backend.app.api.routes.cloud import get_stored_token
+
+        token, email = await get_stored_token(db_session, user=None)
+        assert token is None
+        assert email is None
+
+    @pytest.mark.asyncio
+    async def test_store_and_get_global_token(self, db_session):
+        """store_token with user=None stores in global Settings table."""
+        from backend.app.api.routes.cloud import get_stored_token, store_token
+
+        await store_token(db_session, "test-token-123", "test@example.com", user=None)
+        token, email = await get_stored_token(db_session, user=None)
+        assert token == "test-token-123"
+        assert email == "test@example.com"
+
+    @pytest.mark.asyncio
+    async def test_store_and_get_per_user_token(self, db_session):
+        """store_token with user stores on the user record."""
+        from backend.app.api.routes.cloud import get_stored_token, store_token
+        from backend.app.core.auth import get_password_hash
+        from backend.app.models.user import User
+
+        user = User(username="tokentest", password_hash=get_password_hash("pass"), role="user")
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+
+        await store_token(db_session, "user-token-abc", "user@example.com", user=user)
+
+        # Re-fetch user to verify persistence
+        from sqlalchemy import select
+
+        result = await db_session.execute(select(User).where(User.id == user.id))
+        refreshed = result.scalar_one()
+        assert refreshed.cloud_token == "user-token-abc"
+        assert refreshed.cloud_email == "user@example.com"
+
+    @pytest.mark.asyncio
+    async def test_per_user_token_does_not_affect_global(self, db_session):
+        """Storing per-user token should not affect global Settings."""
+        from backend.app.api.routes.cloud import get_stored_token, store_token
+        from backend.app.core.auth import get_password_hash
+        from backend.app.models.user import User
+
+        user = User(username="isolationtest", password_hash=get_password_hash("pass"), role="user")
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+
+        # Store per-user token
+        await store_token(db_session, "per-user-token", "per-user@test.com", user=user)
+
+        # Global should still be empty
+        global_token, global_email = await get_stored_token(db_session, user=None)
+        assert global_token is None
+        assert global_email is None
+
+    @pytest.mark.asyncio
+    async def test_clear_per_user_token(self, db_session):
+        """clear_token with user clears only that user's credentials."""
+        from backend.app.api.routes.cloud import clear_token, get_stored_token, store_token
+        from backend.app.core.auth import get_password_hash
+        from backend.app.models.user import User
+
+        user = User(username="cleartest", password_hash=get_password_hash("pass"), role="user")
+        db_session.add(user)
+        await db_session.commit()
+        await db_session.refresh(user)
+
+        await store_token(db_session, "to-clear", "clear@test.com", user=user)
+        await clear_token(db_session, user=user)
+
+        from sqlalchemy import select
+
+        result = await db_session.execute(select(User).where(User.id == user.id))
+        refreshed = result.scalar_one()
+        assert refreshed.cloud_token is None
+        assert refreshed.cloud_email is None
+
+    @pytest.mark.asyncio
+    async def test_clear_global_token(self, db_session):
+        """clear_token with user=None clears from global Settings."""
+        from backend.app.api.routes.cloud import clear_token, get_stored_token, store_token
+
+        await store_token(db_session, "global-token", "global@test.com", user=None)
+        await clear_token(db_session, user=None)
+
+        token, email = await get_stored_token(db_session, user=None)
+        assert token is None
+        assert email is None
+
+    @pytest.mark.asyncio
+    async def test_two_users_independent_tokens(self, db_session):
+        """Two users should have completely independent cloud tokens."""
+        from backend.app.api.routes.cloud import get_stored_token, store_token
+        from backend.app.core.auth import get_password_hash
+        from backend.app.models.user import User
+
+        user_a = User(username="user_a", password_hash=get_password_hash("pass"), role="user")
+        user_b = User(username="user_b", password_hash=get_password_hash("pass"), role="user")
+        db_session.add_all([user_a, user_b])
+        await db_session.commit()
+        await db_session.refresh(user_a)
+        await db_session.refresh(user_b)
+
+        await store_token(db_session, "token-a", "a@test.com", user=user_a)
+        await store_token(db_session, "token-b", "b@test.com", user=user_b)
+
+        # Verify each user reads their own token (re-fetch from DB)
+        from sqlalchemy import select
+
+        result_a = await db_session.execute(select(User).where(User.id == user_a.id))
+        result_b = await db_session.execute(select(User).where(User.id == user_b.id))
+        fresh_a = result_a.scalar_one()
+        fresh_b = result_b.scalar_one()
+
+        token_a, email_a = await get_stored_token(db_session, user=fresh_a)
+        token_b, email_b = await get_stored_token(db_session, user=fresh_b)
+
+        assert token_a == "token-a"
+        assert email_a == "a@test.com"
+        assert token_b == "token-b"
+        assert email_b == "b@test.com"