bambuddy/backend/app/services/bambu_cloud.py

"""
Bambu Lab Cloud API Service

Handles authentication and profile management with Bambu Lab's cloud services.
"""

import hashlib
import logging
import time
from collections.abc import Awaitable, Callable
from datetime import datetime, timezone

import httpx

logger = logging.getLogger(__name__)

BAMBU_API_BASE = "https://api.bambulab.com"
BAMBU_API_BASE_CN = "https://api.bambulab.cn"

# How long a "Bambu still accepts this token" answer is trusted before we ask
# again. ``/cloud/status`` is polled by several components, so validating on
# every call would put a Bambu round-trip behind every settings render; a token
# does not expire on a five-minute boundary, so caching that long is free.
_VALIDATION_TTL_SECONDS = 300

# token digest -> (monotonic deadline, accepted?). Keyed by digest so a token
# never sits in a process-wide dict in the clear.
_validation_cache: dict[str, tuple[float, bool]] = {}


def _token_digest(token: str) -> str:
    return hashlib.sha256(token.encode("utf-8")).hexdigest()


def is_expiry_401(response: httpx.Response) -> bool:
    """Whether a 401 is Bambu's genuine "token expired" signal.

    Bambu answers an expired/revoked token with ``{"code":4,"error":"Please
    login.","message":""}``. Not every 401 means that: individual endpoints
    return 401 for resource-, region- or scope-specific reasons, and a working
    token still draws the occasional transient 401 (Cloudflare edge, a brief
    backend blip). Treating *any* 401 as a dead credential signs the user out on
    a single stray rejection — the #2562 follow-up regression. We trust only the
    documented expiry body, so a benign 401 no longer nukes the whole cloud
    integration. An unparseable / unsigned 401 is deliberately NOT expiry.

    Shared by the Bambu Cloud and MakerWorld services — both carry the same
    token and see the same expiry body.
    """
    try:
        body = response.json()
    except Exception:
        return False
    if not isinstance(body, dict):
        return False
    if body.get("code") == 4:
        return True
    text = f"{body.get('error', '')} {body.get('message', '')}".lower()
    return "please login" in text


def invalidate_validation_cache(token: str | None = None) -> None:
    """Drop cached validation verdicts.

    Called on login/logout so a fresh token isn't judged by the previous one's
    cached verdict, and so a re-login clears a cached rejection immediately
    rather than leaving the user staring at "sign-in expired" for five minutes.
    """
    if token is None:
        _validation_cache.clear()
    else:
        _validation_cache.pop(_token_digest(token), None)


# Client identity sent to Bambu Lab's cloud services. We identify honestly as
# Bambuddy — the URL in parens makes the source unambiguous so Bambu can
# distinguish our traffic from impersonators. This is the opposite of what the
# OrcaSlicer fork was called out for in the May 2026 Bambu Lab blog post
# ("Setting the record straight on cloud access and community"): we do not
# introduce ourselves as official Bambu Studio.
_USER_AGENT = "Bambuddy/1.0 (+https://github.com/maziggy/bambuddy)"

# Cloudflare protection on Bambu Lab's edge intermittently returns interstitials /
# challenges instead of the JSON the API normally produces (issue #1575). The
# parse error that results is opaque — these helpers detect the CF markers so
# we can surface an actionable message instead of "Invalid response from Bambu Cloud".
_CF_INTERSTITIAL_USER_MESSAGE = (
    "Bambu Cloud is temporarily blocking automated requests from your network. "
    "This is a Cloudflare protection on Bambu Lab's side, not a Bambuddy issue. "
    "Please wait a few minutes and try again. If it persists, signing in to "
    "bambulab.com once from a browser on the same network usually clears the "
    "challenge."
)


def _detect_cloudflare_challenge(response) -> str | None:
    """Return a user-actionable message when the response is a Cloudflare
    challenge / mitigation page instead of the JSON the API normally returns.

    Triggers on any of:
      - body contains "Just a moment..." (CF interactive challenge title)
      - body contains "challenges.cloudflare.com" (CF turnstile widget src)
      - HTTP 403 with a "cf-mitigated" response header (CF blocked)
      - HTTP 503 with a "cf-ray" response header (CF Under Attack mode)

    Returns None when the response doesn't look like a CF challenge — callers
    fall through to their existing error path.
    """
    try:
        body = response.text or ""
    except Exception:
        body = ""
    if "Just a moment..." in body or "challenges.cloudflare.com" in body:
        return _CF_INTERSTITIAL_USER_MESSAGE
    try:
        status = int(getattr(response, "status_code", 0) or 0)
    except (TypeError, ValueError):
        status = 0
    headers = getattr(response, "headers", {}) or {}
    if status == 403 and "cf-mitigated" in headers:
        return _CF_INTERSTITIAL_USER_MESSAGE
    if status == 503 and "cf-ray" in headers:
        return _CF_INTERSTITIAL_USER_MESSAGE
    return None


# Bambu's own anti-abuse layer — distinct from the Cloudflare edge above —
# answers a request it has flagged with HTTP 418 and a challenge body:
#
#     {"captchaId": "...", "error": "We need you to confirm you are not a robot"}
#
# The flag is keyed to the source IP and covers api.bambulab.com as a whole:
# the same 418 turns up on the login endpoint and on the design-service
# endpoints MakerWorld imports use. It clears on its own after a few hours of
# quiet traffic, and there is no server-side solve — a CAPTCHA is designed to be
# unanswerable without a real browser, and the challenge id is of no use to us
# because we have nowhere to render the widget.
#
# It reaches ``login_request`` as a perfectly well-formed JSON body, so
# ``_detect_cloudflare_challenge`` above never fires on it. Before #2790 the
# generic error path then lifted Bambu's sentence out of ``error`` and showed it
# as a bare toast: the reporter saw "We need you to confirm you are not a robot"
# with no challenge, no explanation and nothing to click, and filed it as a
# Bambuddy bug.
_CAPTCHA_HTTP_STATUS = 418

# Markers that identify a 418 as the CAPTCHA challenge rather than some other
# refusal. ``captchaId`` is the reliable one; the wording is matched too because
# Bambu has shipped the challenge under more than one phrasing.
_CAPTCHA_BODY_MARKERS = ("captchaid", "captcha", "robot")

CAPTCHA_USER_MESSAGE = (
    "Bambu Cloud is challenging this network with a CAPTCHA before it will accept a sign-in, "
    "and there is no way to answer it from Bambuddy. Your email and password are not the "
    "problem. The block is tied to your public IP address and normally clears by itself within "
    "a few hours — retrying repeatedly extends it. To sign in now, use 'Use access token "
    "instead' and paste a token taken from a browser session."
)

# How long to stop sending sign-in requests to a Bambu region after it answered
# with a CAPTCHA challenge. The reporter's log shows four attempts in eighteen
# seconds, which is exactly the traffic pattern that deepens the block: every
# extra request is more evidence for the thing that flagged us. Five minutes is
# short against the hours the block itself lasts — the point is not to wait it
# out here, only to stop Bambuddy from making it worse while the user reads the
# explanation.
_CAPTCHA_COOLOFF_SECONDS = 300.0

# API base URL -> monotonic time its cool-off expires. Keyed by base URL because
# the block lives at the edge in front of one region: being challenged on
# api.bambulab.com says nothing about api.bambulab.cn.
_captcha_blocked_until: dict[str, float] = {}


def is_captcha_challenge(response) -> bool:
    """Whether Bambu answered with an anti-abuse CAPTCHA challenge.

    Requires the 418 status *and* a challenge marker in the body, so an
    unrelated 418 is not reported to the user as "solve a CAPTCHA" — that would
    send them looking for a widget that was never there, which is the exact
    confusion #2790 is about. Callers that want to say something about a bare
    418 must handle it themselves.

    Shared by the Bambu Cloud and MakerWorld services: same edge, same body.
    """
    try:
        status = int(getattr(response, "status_code", 0) or 0)
    except (TypeError, ValueError):
        return False
    if status != _CAPTCHA_HTTP_STATUS:
        return False
    try:
        data = response.json()
    except Exception:
        data = None
    if isinstance(data, dict):
        # Field *names* count as well as their text: the challenge is
        # identified by carrying a ``captchaId`` at all, whatever it says.
        parts = [str(key) for key in data]
        parts += [str(data[key]) for key in ("captchaId", "error", "message", "detail") if data.get(key)]
        haystack = " ".join(parts).lower()
    else:
        # Not JSON (or not an object) — fall back to the raw body so a
        # challenge served as HTML is still recognised rather than reported as
        # an unexplained failure.
        try:
            haystack = (response.text or "").lower()
        except Exception:
            return False
    return any(marker in haystack for marker in _CAPTCHA_BODY_MARKERS)


def captcha_cooloff_active(base_url: str) -> bool:
    """Whether sign-in requests to ``base_url`` are still held back after a
    CAPTCHA challenge. Expired entries are dropped on the way past, so the dict
    cannot grow past one entry per region."""
    deadline = _captcha_blocked_until.get(base_url)
    if deadline is None:
        return False
    if time.monotonic() >= deadline:
        del _captcha_blocked_until[base_url]
        return False
    return True


def note_captcha_challenge(base_url: str) -> None:
    """Start the cool-off for ``base_url`` after a challenge was seen."""
    _captcha_blocked_until[base_url] = time.monotonic() + _CAPTCHA_COOLOFF_SECONDS


# The `/v1/iot-service/api/slicer/setting` endpoint subtree — the plural GET
# for the list, the singular GET/DELETE for a specific preset by setting_id, and
# the POST for create — requires a `version` query parameter in the XX.YY.ZZ.WW
# format Bambu Studio releases use. Without it the API returns HTTP 400
# "field 'version' is not set"; non-matching formats like "bambuddy-1.0" return
# HTTP 422 "Invalid input parameters". However, Bambu's server accepts ANY value
# within that format — it doesn't validate against a release manifest. We
# therefore use a neutral "1.0.0.0" placeholder that does not impersonate any
# real Bambu Studio release. Our client identity is in the User-Agent header.
_SLICER_API_VERSION = "1.0.0.0"


class BambuCloudError(Exception):
    """Base exception for Bambu Cloud errors.

    ``status_code`` carries the upstream HTTP status when the failure came from
    a response rather than from the transport, so callers can tell an expected
    "this preset isn't in the catalog" 400 apart from an expired token or a
    cloud outage. It stays ``None`` for connection-level failures.
    """

    def __init__(self, message: str, *, status_code: int | None = None):
        super().__init__(message)
        self.status_code = status_code


class BambuCloudAuthError(BambuCloudError):
    """Authentication related errors."""

    pass


def _cloud_json(response: httpx.Response):
    """The response's JSON body, or a BambuCloudError when it is not JSON (a
    Cloudflare challenge or an edge error page): an answer from Bambu Cloud
    that it could not use, like any other failed request."""
    try:
        return response.json()
    except ValueError:
        raise BambuCloudError(
            f"Unexpected response from Bambu Cloud (HTTP {response.status_code})", status_code=response.status_code
        ) from None


_shared_http_client: httpx.AsyncClient | None = None


def set_shared_http_client(client: httpx.AsyncClient | None) -> None:
    """Register an app-scoped ``httpx.AsyncClient`` so per-request
    ``BambuCloudService`` instances can reuse its connection pool.

    Pass ``None`` during shutdown to unregister. The service only holds a
    reference (never closes a client it does not own), so region + token
    state still stays per-request — this only shares the transport pool.
    """
    global _shared_http_client
    _shared_http_client = client


class BambuCloudService:
    """Service for interacting with Bambu Lab Cloud API."""

    def __init__(
        self,
        region: str = "global",
        client: httpx.AsyncClient | None = None,
        on_auth_failure: Callable[[], Awaitable[None]] | None = None,
    ):
        self.base_url = BAMBU_API_BASE if region == "global" else BAMBU_API_BASE_CN
        self.access_token: str | None = None
        self.refresh_token: str | None = None
        self.token_expiry: datetime | None = None
        # Fired once when Bambu answers 401 to a call we made with a stored
        # token — the credential is dead and the caller wants to record that.
        # ``build_authenticated_cloud`` wires this to the persisted flag, so
        # every route that builds a service through it gets invalidation for
        # free rather than each one having to notice 401s for itself.
        self._on_auth_failure = on_auth_failure
        self._auth_failure_reported = False
        # Prefer an explicitly-injected client (tests), else fall back to the
        # app-scoped shared client (production), and finally create our own so
        # scripts / tests that skip the lifespan still get a working service.
        if client is not None:
            self._client = client
            self._owns_client = False
        elif _shared_http_client is not None:
            self._client = _shared_http_client
            self._owns_client = False
        else:
            self._client = httpx.AsyncClient(timeout=30.0)
            self._owns_client = True

    @property
    def is_authenticated(self) -> bool:
        """Whether a credential is *loaded* — NOT whether Bambu accepts it.

        Bambu's access token is opaque (no JWT claims to read an expiry out
        of), so the only authority on whether it still works is Bambu. This
        used to pretend otherwise: ``set_token`` stamped ``token_expiry =
        now + 30 days`` every time a stored token was loaded, which made the
        expiry check reset on every request and this property incapable of
        ever returning False. The UI reported "connected" indefinitely while
        every cloud call 401'd (#2562 follow-up).

        ``token_expiry`` is now only set when we genuinely know it. Callers
        that need to know the token still *works* must ask Bambu — see
        :meth:`validate_token` — or react to the 401 that surfaces.
        """
        if not self.access_token:
            return False
        return not (self.token_expiry and datetime.now(timezone.utc) > self.token_expiry)

    async def _note_response(self, response: httpx.Response) -> bool:
        """Record Bambu's genuine token-expiry 401 as "this credential is dead".

        Returns ``True`` only for the real expiry signal (see
        :meth:`_is_expiry_401`); a plain/transient 401 returns ``False`` and is
        left alone so it can't durably sign the user out. The durable flag is
        written at most once per service instance so a route making several
        calls doesn't write it repeatedly.
        """
        if response.status_code != 401:
            return False
        if not is_expiry_401(response):
            logger.info(
                "Bambu Cloud returned 401 without the expiry signature — treating as transient, "
                "not signing the stored token out"
            )
            return False
        if self._on_auth_failure is None or self._auth_failure_reported:
            return True
        self._auth_failure_reported = True
        if self.access_token:
            _validation_cache[_token_digest(self.access_token)] = (
                time.monotonic() + _VALIDATION_TTL_SECONDS,
                False,
            )
        try:
            await self._on_auth_failure()
        except Exception:
            # Recording the failure is best-effort — the caller still needs the
            # real error (a 401) rather than a bookkeeping exception on top.
            logger.exception("Failed to record Bambu Cloud auth failure")
        return True

    async def validate_token(self) -> bool | None:
        """Ask Bambu whether the loaded token is still accepted.

        ``True`` accepted, ``False`` rejected (401), ``None`` unknown — Bambu
        was unreachable or answered 5xx.

        ``None`` must never be treated as "invalid": a Bambu outage or a
        Cloudflare interstitial would otherwise sign every user out of a
        perfectly good session. Callers report their last known state instead.
        """
        if not self.access_token:
            return False

        digest = _token_digest(self.access_token)
        cached = _validation_cache.get(digest)
        if cached and cached[0] > time.monotonic():
            return cached[1]

        try:
            response = await self._client.get(
                f"{self.base_url}/v1/design-user-service/my/preference",
                headers=self._get_headers(),
                timeout=15.0,
            )
        except httpx.HTTPError as exc:
            logger.info("Could not reach Bambu Cloud to validate the stored token: %s", exc)
            return None

        if response.status_code == 401:
            # Only a 401 carrying Bambu's expiry signature is a real sign-out.
            # A signature-less 401 here is transient/edge noise — report unknown
            # (last-known state) rather than expiring a working session.
            expired = await self._note_response(response)
            return False if expired else None
        if response.status_code >= 500:
            logger.info(
                "Bambu Cloud returned %s while validating the token — treating as unknown", response.status_code
            )
            return None
        if response.status_code != 200:
            # 4xx that isn't 401 (403, 418 Cloudflare challenge, 429): the token
            # itself was not rejected, so don't declare it dead.
            logger.info(
                "Bambu Cloud returned %s while validating the token — treating as unknown", response.status_code
            )
            return None

        _validation_cache[digest] = (time.monotonic() + _VALIDATION_TTL_SECONDS, True)
        return True

    def _get_headers(self) -> dict:
        """Get headers for authenticated requests."""
        headers = {
            "Content-Type": "application/json",
            "User-Agent": _USER_AGENT,
        }
        if self.access_token:
            headers["Authorization"] = f"Bearer {self.access_token}"
        return headers

    def _captcha_refusal(self) -> dict:
        """The result every sign-in call returns while Bambu is challenging us.

        ``reason`` is what lets the UI tell this apart from a wrong password and
        render the explanation next to the access-token route, instead of
        flashing Bambu's own one-liner as a toast that then disappears (#2790).
        """
        return {
            "success": False,
            "needs_verification": False,
            "reason": "captcha",
            "message": CAPTCHA_USER_MESSAGE,
        }

    def _captcha_cooloff_holds(self, origin: str | None = None) -> bool:
        """Whether to refuse a sign-in locally because Bambu just challenged us.

        Keyed by the origin the call actually goes to. The TOTP step talks to
        ``bambulab.com`` while everything else talks to ``api.bambulab.com``, and
        a challenge seen on one must not strand a user halfway through a
        two-factor sign-in on the other.
        """
        origin = origin or self.base_url
        if not captcha_cooloff_active(origin):
            return False
        logger.warning(
            "Bambu Cloud is challenging this network with a CAPTCHA — not sending the sign-in to %s. "
            "The challenge cannot be answered from Bambuddy and normally clears within a few hours.",
            origin,
        )
        return True

    def _note_captcha(self, response, origin: str | None = None) -> bool:
        """Record and log a CAPTCHA challenge. Returns whether it was one."""
        if not is_captcha_challenge(response):
            return False
        origin = origin or self.base_url
        logger.warning(
            "Bambu Cloud is challenging this network with a CAPTCHA (HTTP %s from %s). Sign-in cannot "
            "complete until the challenge clears; pausing sign-in requests for %.0fs so retries do not "
            "extend the block.",
            response.status_code,
            origin,
            _CAPTCHA_COOLOFF_SECONDS,
        )
        note_captcha_challenge(origin)
        return True

    async def login_request(self, email: str, password: str) -> dict:
        """
        Initiate login - this will trigger either email verification or TOTP prompt.

        Returns dict with login status, verification type, and tfaKey if needed.
        """
        if self._captcha_cooloff_holds():
            return self._captcha_refusal()
        try:
            response = await self._client.post(
                f"{self.base_url}/v1/user-service/user/login",
                headers={"Content-Type": "application/json"},
                json={
                    "account": email,
                    "password": password,
                },
            )

            if self._note_captcha(response):
                return self._captcha_refusal()

            try:
                data = response.json()
            except Exception as json_err:
                logger.error("Failed to parse login response: %s, body: %s", json_err, response.text[:500])
                cf_message = _detect_cloudflare_challenge(response)
                return {
                    "success": False,
                    "needs_verification": False,
                    "message": cf_message or "Invalid response from Bambu Cloud",
                }
            logger.debug(
                f"Login response: status={response.status_code}, loginType={data.get('loginType')}, hasTfaKey={'tfaKey' in data}"
            )

            if response.status_code == 200:
                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:
                    self._set_tokens(data)
                    return {"success": True, "needs_verification": False, "message": "Login successful"}

            # Handle specific error codes
            error_msg = data.get("message") or data.get("error") or "Login failed"
            return {"success": False, "needs_verification": False, "message": error_msg}

        except Exception as e:
            logger.error("Login request failed: %s", e)
            # The request never got an answer: not a refused login.
            raise BambuCloudError(f"Login request failed: {e}")

    async def verify_code(self, email: str, code: str) -> dict:
        """
        Complete login with email verification code.
        """
        if self._captcha_cooloff_holds():
            return self._captcha_refusal()
        try:
            response = await self._client.post(
                f"{self.base_url}/v1/user-service/user/login",
                headers={"Content-Type": "application/json"},
                json={
                    "account": email,
                    "code": code,
                },
            )

            if self._note_captcha(response):
                return self._captcha_refusal()

            try:
                data = response.json()
            except Exception as json_err:
                logger.error("Failed to parse email-verify response: %s, body: %s", json_err, response.text[:500])
                cf_message = _detect_cloudflare_challenge(response)
                return {"success": False, "message": cf_message or "Invalid response from Bambu Cloud"}
            logger.debug("Email verify response: status=%s, hasToken=%s", response.status_code, "accessToken" in data)

            if response.status_code == 200 and "accessToken" in data:
                self._set_tokens(data)
                return {"success": True, "message": "Login successful"}

            return {"success": False, "message": data.get("message", "Verification failed")}

        except Exception as e:
            logger.error("Email verification failed: %s", e)
            # The request never got an answer: not a refused code.
            raise BambuCloudError(f"Verification failed: {e}")

    async def _fetch_csrf_token(self, web_origin: str) -> str | None:
        """Seed the ``bbl_csrf_token`` cookie and return its value (#2696).

        Bambu added double-submit CSRF protection to the ``bambulab.com`` web
        origin. A POST without the cookie is rejected ``403 {"error": "CSRF
        error: missing_cookie"}`` before the request body is looked at; with the
        cookie but no matching header it becomes ``missing_header``. Only
        ``GET /api/csrf`` mints one — the sign-in *page* sets nothing but
        Cloudflare's ``__cf_bm``, so landing there first does not help.

        The token is re-fetched per verification rather than cached: the client
        is process-wide and long-lived, so a stale cookie could otherwise
        disagree with the header we send.
        """
        try:
            response = await self._client.get(
                f"{web_origin}/api/csrf",
                headers={"User-Agent": _USER_AGENT, "Accept": "application/json"},
            )
        except Exception as e:
            logger.warning("Failed to fetch Bambu Cloud CSRF token: %s", e)
            return None
        # httpx stores the Set-Cookie on the shared jar, which is also what makes
        # the cookie ride along on the POST below — we only need the value here
        # to echo it back in the header.
        try:
            token = self._client.cookies.get("bbl_csrf_token")
        except Exception:  # multiple cookies of the same name across domains
            token = None
        if not token:
            logger.warning(
                "Bambu Cloud CSRF endpoint returned no bbl_csrf_token (status %s)",
                response.status_code,
            )
        return token

    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.
            # We previously sent a Chrome User-Agent plus Origin/Referer headers
            # under the assumption Cloudflare would block bot-identified
            # requests. Verified 2026-05-12 via curl that the endpoint accepts
            # honest "Bambuddy/X.Y.Z" identification cleanly (HTTP 400 with the
            # expected application-level "Login failed" JSON, no Cloudflare
            # interstitial). Browser-impersonation removed to stay clearly on
            # the right side of Bambu Lab's "no falsified client identity" line.
            web_origin = "https://bambulab.cn" if "bambulab.cn" in self.base_url else "https://bambulab.com"
            tfa_url = f"{web_origin}/api/sign-in/tfa"

            if self._captcha_cooloff_holds(web_origin):
                return self._captcha_refusal()

            # #2696: the web origin is CSRF-protected (double submit). Without
            # both halves the endpoint 403s before it ever evaluates the code,
            # which surfaced to users as a permanent, misleading "Invalid code".
            # api.bambulab.com — where every other call in this service goes,
            # including the email-code 2FA path — is not gated, which is why
            # only TOTP sign-ins broke.
            csrf_token = await self._fetch_csrf_token(web_origin)
            if not csrf_token:
                return {
                    "success": False,
                    "message": (
                        "Could not obtain a security token from Bambu Cloud. "
                        "Check the server's internet access and try again."
                    ),
                }

            response = await self._client.post(
                tfa_url,
                headers={
                    "Content-Type": "application/json",
                    "User-Agent": _USER_AGENT,
                    "Accept": "application/json",
                    # Echo of the bbl_csrf_token cookie httpx just stored. Both
                    # halves are required; the cookie alone yields
                    # "missing_header".
                    "x-bbl-csrf-token": csrf_token,
                },
                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)'}"
            )

            if self._note_captcha(response, web_origin):
                return self._captcha_refusal()

            # Handle empty response
            if not response.text or not response.text.strip():
                logger.warning("TOTP verification returned empty response (status %s)", 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("Failed to parse TOTP response: %s, body: %s", json_err, response.text[:500])
                cf_message = _detect_cloudflare_challenge(response)
                return {"success": False, "message": cf_message or "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")
                # Expiry left unset: Bambu does not tell us when the token dies
                # and the token is opaque, so any value here would be invented.
                self.token_expiry = None
                invalidate_validation_cache(access_token)
                return {"success": True, "message": "Login successful"}

            # Provide helpful error message
            error_msg = data.get("message", "")

            # A CSRF rejection means the code was never evaluated (#2696). It
            # used to fall through to the generic path below and read as
            # "Invalid code", which sent the reporter chasing clock drift and
            # leading-zero parsing for a request Bambu had already refused.
            csrf_error = data.get("error", "") if isinstance(data.get("error"), str) else ""
            if "csrf" in csrf_error.lower() or data.get("reason") in ("missing_cookie", "missing_header"):
                logger.error("Bambu Cloud rejected the TOTP request on CSRF grounds: %s", response.text[:200])
                return {
                    "success": False,
                    "message": (
                        "Bambu Cloud rejected the sign-in request before checking your code "
                        "(security-token error). Your code is fine — please try again."
                    ),
                }

            if "expired" in error_msg.lower():
                return {"success": False, "message": "TOTP session expired. Please try logging in again."}
            if not error_msg:
                error_msg = data.get("error") or f"TOTP verification failed (status {response.status_code})"

            return {"success": False, "message": error_msg}

        except Exception as e:
            logger.error("TOTP verification failed: %s", 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 a login response.

        No expiry is recorded. Bambu's login response carries no expiry, and
        the access token is opaque, so the old ``now + 30 days`` was a guess
        that outlived its own accuracy — see :attr:`is_authenticated`.
        """
        self.access_token = data.get("accessToken")
        self.refresh_token = data.get("refreshToken")
        self.token_expiry = None
        if self.access_token:
            invalidate_validation_cache(self.access_token)

    def set_token(self, access_token: str):
        """Load a stored access token.

        This used to stamp ``token_expiry = now + 30 days`` — re-derived from
        *now* on every request, for a token of entirely unknown age. That made
        ``is_authenticated`` a permanent True and is why Bambuddy went on
        reporting "connected" long after Bambu had stopped accepting the token.
        A stored token's remaining life is unknowable from the token alone, so
        we record no expiry and let Bambu be the authority.
        """
        self.access_token = access_token
        self.token_expiry = None

    def logout(self):
        """Clear authentication state."""
        self.access_token = None
        self.refresh_token = None
        self.token_expiry = None

    async def get_user_profile(self) -> dict:
        """Get user profile information."""
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            response = await self._client.get(
                f"{self.base_url}/v1/design-user-service/my/preference", headers=self._get_headers()
            )

            if response.status_code == 200:
                return _cloud_json(response)

            # Only Bambu's own answer judges the token: JSON, or no body at all
            # (how it refuses a wrong token here). A Cloudflare block or an
            # edge or proxy page says nothing about it, so it carries no
            # status to judge by.
            cf_message = _detect_cloudflare_challenge(response)
            if cf_message:
                raise BambuCloudError(cf_message)
            if response.content:
                try:
                    response.json()
                except ValueError:
                    raise BambuCloudError(
                        f"Unexpected response from Bambu Cloud (HTTP {response.status_code})"
                    ) from None
            raise BambuCloudError(f"Failed to get profile: {response.status_code}", status_code=response.status_code)

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def get_slicer_settings(self, version: str = _SLICER_API_VERSION) -> dict:
        """
        Get all slicer settings (filament, printer, process presets).

        Args:
            version: Slicer version string. Bambu's API requires the XX.YY.ZZ.WW
                format but does not validate against a release manifest — we
                default to the neutral _SLICER_API_VERSION placeholder so we
                never claim to be a specific Bambu Studio build. Callers should
                normally use the default.
        """
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            response = await self._client.get(
                f"{self.base_url}/v1/iot-service/api/slicer/setting",
                headers=self._get_headers(),
                params={"version": version},
            )

            data = _cloud_json(response)

            if await self._note_response(response):
                raise BambuCloudAuthError("Bambu Cloud session expired")
            if response.status_code == 200:
                return data

            raise BambuCloudError(f"Failed to get settings: {response.status_code}")

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def get_setting_detail(self, setting_id: str) -> dict:
        """Get detailed information for a specific setting/preset."""
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            response = await self._client.get(
                f"{self.base_url}/v1/iot-service/api/slicer/setting/{setting_id}",
                headers=self._get_headers(),
                params={"version": _SLICER_API_VERSION},
            )

            if await self._note_response(response):
                raise BambuCloudAuthError("Bambu Cloud session expired")
            if response.status_code == 200:
                return _cloud_json(response)

            # Include body so a future contract change is self-diagnostic from logs.
            body = (response.text or "")[:200]
            raise BambuCloudError(
                f"Failed to get setting detail: {response.status_code} {body}",
                status_code=response.status_code,
            )

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def create_setting(
        self, preset_type: str, name: str, base_id: str, setting: dict, version: str = "2.0.0.0"
    ) -> dict:
        """
        Create a new slicer preset/setting.

        Args:
            preset_type: Type of preset - "filament", "print", or "printer"
            name: Display name for the preset
            base_id: Base preset ID to inherit from (e.g., "GFSA00")
            setting: Dict of setting key-value pairs (only modified values from base)
            version: Version string for the preset (default: "2.0.0.0")

        Returns:
            Created preset data including the new setting_id
        """
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            # Add timestamp if not present
            import time

            if "updated_time" not in setting:
                setting["updated_time"] = str(int(time.time()))

            payload = {
                "type": preset_type,
                "name": name,
                "version": version,
                "base_id": base_id,
                "setting": setting,
            }

            response = await self._client.post(
                f"{self.base_url}/v1/iot-service/api/slicer/setting", headers=self._get_headers(), json=payload
            )

            data = _cloud_json(response)

            if await self._note_response(response):
                raise BambuCloudAuthError("Bambu Cloud session expired")
            if response.status_code in (200, 201):
                return data

            error_msg = data.get("message") or data.get("error") or f"HTTP {response.status_code}"
            raise BambuCloudError(f"Failed to create setting: {error_msg}")

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def update_setting(self, setting_id: str, name: str | None = None, setting: dict | None = None) -> dict:
        """
        Update an existing slicer preset/setting.

        Note: Bambu Cloud API doesn't support true updates. Instead, we:
        1. Fetch the current setting metadata (type, base_id, version)
        2. Use the provided settings as the new complete settings (NOT merged)
        3. Delete the old setting first (to avoid name conflicts)
        4. Create a new setting via POST

        Args:
            setting_id: ID of the preset to update
            name: New display name (optional)
            setting: Dict of setting key-value pairs - this REPLACES the old settings entirely

        Returns:
            Updated preset data with new setting_id
        """
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            # Fetch current setting to get metadata (type, base_id, version)
            current = await self.get_setting_detail(setting_id)
            preset_type = current.get("type", "filament")

            # Use provided settings directly (complete replacement, not merge)
            # This allows the frontend to edit the full settings JSON
            if setting is not None:
                updated_setting = setting.copy()
            else:
                updated_setting = current.get("setting", {}).copy()

            # Extract name from settings_id field in the JSON, or use provided name, or fall back to current
            # The settings_id field contains the name in quotes, e.g., '"My Preset Name"'
            settings_id_key = {
                "filament": "filament_settings_id",
                "print": "print_settings_id",
                "printer": "printer_settings_id",
            }.get(preset_type, "filament_settings_id")

            settings_id_value = updated_setting.get(settings_id_key, "")
            if settings_id_value:
                # Remove surrounding quotes if present (e.g., '"foo"' -> 'foo')
                updated_name = settings_id_value.strip('"')
            elif name is not None:
                updated_name = name
            else:
                updated_name = current.get("name", "Untitled")

            # Update the timestamp
            import time

            updated_setting["updated_time"] = str(int(time.time()))

            # Ensure settings_id field matches the name
            updated_setting[settings_id_key] = f'"{updated_name}"'

            # Delete the old setting FIRST to avoid name conflicts
            await self.delete_setting(setting_id)

            # Create new setting via POST
            payload = {
                "type": preset_type,
                "name": updated_name,
                "version": current.get("version", "2.0.0.0"),
                "base_id": current.get("base_id", ""),
                "setting": updated_setting,
            }

            response = await self._client.post(
                f"{self.base_url}/v1/iot-service/api/slicer/setting", headers=self._get_headers(), json=payload
            )

            data = _cloud_json(response)

            if await self._note_response(response):
                raise BambuCloudAuthError("Bambu Cloud session expired")
            if response.status_code == 200:
                return data

            error_msg = data.get("message") or data.get("error") or f"HTTP {response.status_code}"
            raise BambuCloudError(f"Failed to update setting: {error_msg}")

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def delete_setting(self, setting_id: str) -> dict:
        """
        Delete a slicer preset/setting.

        Args:
            setting_id: ID of the preset to delete

        Returns:
            Deletion confirmation
        """
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            response = await self._client.delete(
                f"{self.base_url}/v1/iot-service/api/slicer/setting/{setting_id}",
                headers=self._get_headers(),
                params={"version": _SLICER_API_VERSION},
            )

            if await self._note_response(response):
                raise BambuCloudAuthError("Bambu Cloud session expired")
            if response.status_code in (200, 204):
                return {"success": True, "message": "Setting deleted"}

            data = _cloud_json(response) if response.content else {}
            error_msg = data.get("message") or data.get("error") or f"HTTP {response.status_code}"
            raise BambuCloudError(f"Failed to delete setting: {error_msg}")

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def get_devices(self) -> dict:
        """Get list of bound devices."""
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            response = await self._client.get(
                f"{self.base_url}/v1/iot-service/api/user/bind", headers=self._get_headers()
            )

            if await self._note_response(response):
                raise BambuCloudAuthError("Bambu Cloud session expired")
            if response.status_code == 200:
                return _cloud_json(response)

            raise BambuCloudError(f"Failed to get devices: {response.status_code}")

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def get_firmware_version(self, device_id: str) -> dict:
        """
        Get firmware version info for a device.

        Returns dict with:
        - current_version: Installed firmware version
        - latest_version: Latest available firmware version
        - update_available: Boolean indicating if update is available
        - release_notes: Release notes for latest version
        """
        if not self.is_authenticated:
            raise BambuCloudAuthError("Not authenticated")

        try:
            response = await self._client.get(
                f"{self.base_url}/v1/iot-service/api/user/device/version",
                headers=self._get_headers(),
                params={"device_id": device_id},
            )

            if await self._note_response(response):
                raise BambuCloudAuthError("Bambu Cloud session expired")
            if response.status_code == 200:
                data = _cloud_json(response)
                # API wraps response in 'data' field
                return data.get("data", data)

            raise BambuCloudError(f"Failed to get firmware version: {response.status_code}")

        except httpx.RequestError as e:
            raise BambuCloudError(f"Request failed: {e}")

    async def close(self):
        """Close the HTTP client we own. No-op when sharing an app-scoped client."""
        if self._owns_client:
            await self._client.aclose()


# Previously this module exposed a process-wide ``_cloud_service`` singleton
# via ``get_cloud_service()`` / ``reset_cloud_service()``. That pattern leaked
# region and token state across users (a China-region login would pin the
# singleton to api.bambulab.cn until the next explicit reset), so the singleton
# has been removed. Callers should construct a per-request
# ``BambuCloudService(region=...)`` from the stored region and ``await
# cloud.close()`` it when done. See ``routes.cloud.build_authenticated_cloud``
# for the standard pattern.