Add TOTP authenticator support for Bambu Cloud login (fixes #182)

TOTP (Two-Factor Authentication):
- Detect TOTP vs email verification from Bambu API loginType response
- Use dedicated TFA endpoint on bambulab.com (not api.bambulab.com)
- Include browser-like headers to bypass Cloudflare protection
- Extract token from JSON response or cookies
- Frontend shows appropriate messages for each verification type
- Added i18n translations for TOTP UI (en, de, ja)

Closes #182

backend/app/api/routes/cloud.py

@@ -93,8 +93,12 @@ async def login(request: CloudLoginRequest, db: AsyncSession = Depends(get_db)):
     """
     Initiate login to Bambu Cloud.
 
-    This will typically trigger a verification code to be sent to the user's email.
-    After receiving the code, call /cloud/verify to complete the login.
+    This will trigger either:
+    - Email verification: A code is sent to the user's email
+    - TOTP verification: User enters code from their authenticator app
+
+    After receiving/generating the code, call /cloud/verify to complete the login.
+    For TOTP, include the tfa_key from this response in the verify request.
     """
     cloud = get_cloud_service()
 
@@ -112,6 +116,8 @@ async def login(request: CloudLoginRequest, db: AsyncSession = Depends(get_db)):
             success=result.get("success", False),
             needs_verification=result.get("needs_verification", False),
             message=result.get("message", "Unknown error"),
+            verification_type=result.get("verification_type"),
+            tfa_key=result.get("tfa_key"),
         )
     except BambuCloudAuthError as e:
         raise HTTPException(status_code=401, detail=str(e))
@@ -122,15 +128,24 @@ async def login(request: CloudLoginRequest, db: AsyncSession = Depends(get_db)):
 @router.post("/verify", response_model=CloudLoginResponse)
 async def verify_code(request: CloudVerifyRequest, db: AsyncSession = Depends(get_db)):
     """
-    Complete login with verification code.
+    Complete login with verification code (email or TOTP).
 
-    After calling /cloud/login, the user will receive an email with a 6-digit code.
-    Submit that code here to complete authentication.
+    For email verification:
+    - After calling /cloud/login, the user receives an email with a 6-digit code
+    - Submit the code with email address
+
+    For TOTP verification:
+    - The user enters the 6-digit code from their authenticator app
+    - Include the tfa_key from the /cloud/login response
     """
     cloud = get_cloud_service()
 
     try:
-        result = await cloud.verify_code(request.email, request.code)
+        # Use TOTP verification if tfa_key is provided
+        if request.tfa_key:
+            result = await cloud.verify_totp(request.tfa_key, request.code)
+        else:
+            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)

backend/app/schemas/cloud.py

@@ -10,10 +10,11 @@ class CloudLoginRequest(BaseModel):
 
 
 class CloudVerifyRequest(BaseModel):
-    """Request to verify login with 2FA code."""
+    """Request to verify login with 2FA code (email or TOTP)."""
 
     email: str = Field(..., description="Bambu Lab account email")
-    code: str = Field(..., description="6-digit verification code from email")
+    code: str = Field(..., description="6-digit verification code")
+    tfa_key: str | None = Field(None, description="TFA key for TOTP verification (from login response)")
 
 
 class CloudLoginResponse(BaseModel):
@@ -22,6 +23,8 @@ class CloudLoginResponse(BaseModel):
     success: bool
     needs_verification: bool = False
     message: str
+    verification_type: str | None = None  # "email" or "totp"
+    tfa_key: str | None = None  # Key needed for TOTP verification
 
 
 class CloudAuthStatus(BaseModel):

backend/app/services/bambu_cloud.py

@@ -56,9 +56,9 @@ class BambuCloudService:
 
     async def login_request(self, email: str, password: str) -> dict:
         """
-        Initiate login - this will trigger a verification code email.
+        Initiate login - this will trigger either email verification or TOTP prompt.
 
-        Returns dict with login status and whether verification is needed.
+        Returns dict with login status, verification type, and tfaKey if needed.
         """
         try:
             response = await self._client.post(
@@ -71,12 +71,33 @@ class BambuCloudService:
             )
 
             data = response.json()
+            logger.debug(
+                f"Login response: status={response.status_code}, loginType={data.get('loginType')}, hasTfaKey={'tfaKey' in data}"
+            )
 
             if response.status_code == 200:
-                # Check if we need verification code
-                # Bambu API returns loginType or may require tfaKey
-                if data.get("loginType") == "verifyCode" or "tfaKey" in data:
-                    return {"success": False, "needs_verification": True, "message": "Verification code sent to email"}
+                login_type = data.get("loginType")
+                tfa_key = data.get("tfaKey")
+
+                # TOTP authentication required
+                if login_type == "tfa" or (tfa_key and login_type != "verifyCode"):
+                    return {
+                        "success": False,
+                        "needs_verification": True,
+                        "verification_type": "totp",
+                        "tfa_key": tfa_key,
+                        "message": "Enter the code from your authenticator app",
+                    }
+
+                # Email verification required
+                if login_type == "verifyCode":
+                    return {
+                        "success": False,
+                        "needs_verification": True,
+                        "verification_type": "email",
+                        "tfa_key": None,
+                        "message": "Verification code sent to email",
+                    }
 
                 # Direct login success (rare, usually needs 2FA)
                 if "accessToken" in data:
@@ -93,7 +114,7 @@ class BambuCloudService:
 
     async def verify_code(self, email: str, code: str) -> dict:
         """
-        Complete login with verification code.
+        Complete login with email verification code.
         """
         try:
             response = await self._client.post(
@@ -106,6 +127,7 @@ class BambuCloudService:
             )
 
             data = response.json()
+            logger.debug(f"Email verify response: status={response.status_code}, hasToken={'accessToken' in data}")
 
             if response.status_code == 200 and "accessToken" in data:
                 self._set_tokens(data)
@@ -114,9 +136,89 @@ class BambuCloudService:
             return {"success": False, "message": data.get("message", "Verification failed")}
 
         except Exception as e:
-            logger.error(f"Verification failed: {e}")
+            logger.error(f"Email verification failed: {e}")
             raise BambuCloudAuthError(f"Verification failed: {e}")
 
+    async def verify_totp(self, tfa_key: str, code: str) -> dict:
+        """
+        Complete login with TOTP code from authenticator app.
+
+        Args:
+            tfa_key: The tfaKey returned from initial login request
+            code: 6-digit TOTP code from authenticator app
+        """
+        try:
+            # TFA endpoint is on bambulab.com, NOT api.bambulab.com
+            # Requires browser-like headers to bypass Cloudflare
+            tfa_url = "https://bambulab.com/api/sign-in/tfa"
+            if "bambulab.cn" in self.base_url:
+                tfa_url = "https://bambulab.cn/api/sign-in/tfa"
+
+            browser_headers = {
+                "Content-Type": "application/json",
+                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+                "Accept": "application/json, text/plain, */*",
+                "Accept-Language": "en-US,en;q=0.9",
+                "Origin": "https://bambulab.com",
+                "Referer": "https://bambulab.com/",
+            }
+
+            response = await self._client.post(
+                tfa_url,
+                headers=browser_headers,
+                json={
+                    "tfaKey": tfa_key,
+                    "tfaCode": code,
+                },
+            )
+
+            logger.debug(
+                f"TOTP verify response: status={response.status_code}, body={response.text[:200] if response.text else '(empty)'}"
+            )
+
+            # Handle empty response
+            if not response.text or not response.text.strip():
+                logger.warning(f"TOTP verification returned empty response (status {response.status_code})")
+                return {"success": False, "message": "Bambu Cloud returned empty response. Please try again."}
+
+            try:
+                data = response.json()
+            except Exception as json_err:
+                logger.error(f"Failed to parse TOTP response: {json_err}, body: {response.text[:500]}")
+                return {"success": False, "message": "Invalid response from Bambu Cloud"}
+
+            # Token might be in accessToken, token field, or cookies
+            access_token = data.get("accessToken") or data.get("token")
+
+            # Also check cookies for token
+            if not access_token:
+                for cookie in response.cookies:
+                    if "token" in cookie.lower():
+                        access_token = response.cookies.get(cookie)
+                        break
+
+            if response.status_code == 200 and access_token:
+                self.access_token = access_token
+                self.refresh_token = data.get("refreshToken")
+                from datetime import datetime, timedelta
+
+                self.token_expiry = datetime.now() + timedelta(days=30)
+                return {"success": True, "message": "Login successful"}
+
+            # Provide helpful error message
+            error_msg = data.get("message", "")
+            if "expired" in error_msg.lower():
+                return {"success": False, "message": "TOTP session expired. Please try logging in again."}
+            if not error_msg:
+                error_msg = f"TOTP verification failed (status {response.status_code})"
+
+            return {"success": False, "message": error_msg}
+
+        except Exception as e:
+            logger.error(f"TOTP verification failed: {e}")
+            # Return error instead of raising - don't trigger 401/500
+            return {"success": False, "message": f"TOTP verification error: {e}"}
+
     def _set_tokens(self, data: dict):
         """Set tokens from login response."""
         self.access_token = data.get("accessToken")

backend/tests/unit/services/test_bambu_cloud.py

@@ -0,0 +1,238 @@
+"""Tests for Bambu Cloud service - TOTP and email verification flows."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services.bambu_cloud import BambuCloudService
+
+
+class TestBambuCloudLogin:
+    """Test login flow detection (email vs TOTP)."""
+
+    @pytest.fixture
+    def cloud_service(self):
+        """Create a BambuCloudService instance."""
+        return BambuCloudService()
+
+    @pytest.mark.asyncio
+    async def test_login_detects_email_verification(self, cloud_service):
+        """When loginType is verifyCode, should return email verification type."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "loginType": "verifyCode",
+        }
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.login_request("test@example.com", "password")
+
+            assert result["success"] is False
+            assert result["needs_verification"] is True
+            assert result["verification_type"] == "email"
+            assert result["tfa_key"] is None
+            assert "email" in result["message"].lower()
+
+    @pytest.mark.asyncio
+    async def test_login_detects_totp(self, cloud_service):
+        """When loginType is tfa, should return TOTP verification type with tfaKey."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "loginType": "tfa",
+            "tfaKey": "test-tfa-key-123",
+        }
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.login_request("test@example.com", "password")
+
+            assert result["success"] is False
+            assert result["needs_verification"] is True
+            assert result["verification_type"] == "totp"
+            assert result["tfa_key"] == "test-tfa-key-123"
+            assert "authenticator" in result["message"].lower()
+
+    @pytest.mark.asyncio
+    async def test_login_direct_success(self, cloud_service):
+        """When accessToken is returned directly, should succeed without verification."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "accessToken": "test-access-token",
+            "refreshToken": "test-refresh-token",
+        }
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.login_request("test@example.com", "password")
+
+            assert result["success"] is True
+            assert result["needs_verification"] is False
+            assert cloud_service.access_token == "test-access-token"
+
+    @pytest.mark.asyncio
+    async def test_login_failure(self, cloud_service):
+        """When login fails, should return error message."""
+        mock_response = MagicMock()
+        mock_response.status_code = 401
+        mock_response.json.return_value = {
+            "message": "Invalid credentials",
+        }
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.login_request("test@example.com", "wrong-password")
+
+            assert result["success"] is False
+            assert result["needs_verification"] is False
+            assert "Invalid credentials" in result["message"]
+
+
+class TestBambuCloudEmailVerification:
+    """Test email verification flow."""
+
+    @pytest.fixture
+    def cloud_service(self):
+        """Create a BambuCloudService instance."""
+        return BambuCloudService()
+
+    @pytest.mark.asyncio
+    async def test_verify_code_success(self, cloud_service):
+        """When email code is correct, should return success with token."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "accessToken": "test-access-token",
+            "refreshToken": "test-refresh-token",
+        }
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.verify_code("test@example.com", "123456")
+
+            assert result["success"] is True
+            assert cloud_service.access_token == "test-access-token"
+
+    @pytest.mark.asyncio
+    async def test_verify_code_failure(self, cloud_service):
+        """When email code is incorrect, should return failure."""
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_response.json.return_value = {
+            "message": "Invalid verification code",
+        }
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.verify_code("test@example.com", "000000")
+
+            assert result["success"] is False
+            assert "Invalid" in result["message"] or "Verification failed" in result["message"]
+
+
+class TestBambuCloudTOTPVerification:
+    """Test TOTP verification flow."""
+
+    @pytest.fixture
+    def cloud_service(self):
+        """Create a BambuCloudService instance."""
+        return BambuCloudService()
+
+    @pytest.mark.asyncio
+    async def test_verify_totp_success(self, cloud_service):
+        """When TOTP code is correct, should return success with token."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = '{"token": "test-access-token"}'
+        mock_response.json.return_value = {
+            "token": "test-access-token",
+        }
+        mock_response.cookies = {}
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.verify_totp("test-tfa-key", "123456")
+
+            assert result["success"] is True
+            assert cloud_service.access_token == "test-access-token"
+
+    @pytest.mark.asyncio
+    async def test_verify_totp_uses_correct_endpoint(self, cloud_service):
+        """TOTP verification should use bambulab.com, not api.bambulab.com."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = '{"token": "test-token"}'
+        mock_response.json.return_value = {"token": "test-token"}
+        mock_response.cookies = {}
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            await cloud_service.verify_totp("test-tfa-key", "123456")
+
+            # Check the URL used
+            call_args = mock_post.call_args
+            url = call_args[0][0]
+            assert "bambulab.com/api/sign-in/tfa" in url
+            assert "api.bambulab.com" not in url
+
+    @pytest.mark.asyncio
+    async def test_verify_totp_empty_response(self, cloud_service):
+        """When TOTP returns empty response, should handle gracefully."""
+        mock_response = MagicMock()
+        mock_response.status_code = 400
+        mock_response.text = ""
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.verify_totp("test-tfa-key", "123456")
+
+            assert result["success"] is False
+            assert "empty response" in result["message"].lower()
+
+    @pytest.mark.asyncio
+    async def test_verify_totp_cloudflare_blocked(self, cloud_service):
+        """When Cloudflare blocks request, should handle gracefully."""
+        mock_response = MagicMock()
+        mock_response.status_code = 403
+        mock_response.text = "<!DOCTYPE html><html><head><title>Just a moment...</title>"
+        # json() raises an error when response is HTML
+        mock_response.json.side_effect = ValueError("No JSON")
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            result = await cloud_service.verify_totp("test-tfa-key", "123456")
+
+            assert result["success"] is False
+            assert "Invalid response" in result["message"]
+
+    @pytest.mark.asyncio
+    async def test_verify_totp_includes_browser_headers(self, cloud_service):
+        """TOTP verification should include browser-like headers to bypass Cloudflare."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = '{"token": "test-token"}'
+        mock_response.json.return_value = {"token": "test-token"}
+        mock_response.cookies = {}
+
+        with patch.object(cloud_service._client, "post", new_callable=AsyncMock) as mock_post:
+            mock_post.return_value = mock_response
+
+            await cloud_service.verify_totp("test-tfa-key", "123456")
+
+            # Check headers include User-Agent
+            call_args = mock_post.call_args
+            headers = call_args[1]["headers"]
+            assert "User-Agent" in headers
+            assert "Mozilla" in headers["User-Agent"]