"""Tests for Bambu's anti-abuse CAPTCHA challenge on sign-in (#2790). Bambu's own anti-abuse layer -- not the Cloudflare edge -- answers a request it has flagged with ``HTTP 418`` and ``{"captchaId": ..., "error": "We need you to confirm you are not a robot"}``. It is keyed to the source IP, no credential will be accepted until it clears, and there is no server-side solve. That body is well-formed JSON, so the Cloudflare detector never fired on it and ``login_request`` fell through to its generic error path, which lifted Bambu's sentence out of ``error`` and returned it verbatim. The reporter got a bare toast reading "We need you to confirm you are not a robot" -- no challenge to answer, no explanation, nothing to click -- and filed it as a Bambuddy bug. These tests pin: the challenge is recognised by shape rather than by wording, all three sign-in calls report it as ``reason="captcha"`` with an explanation instead of Bambu's raw string, retries are held back per-origin so Bambuddy stops deepening the block, and the scanner names it in the next support bundle. """ from __future__ import annotations from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest from backend.app.services import bambu_cloud as bc from backend.app.services.bambu_cloud import BambuCloudService # Bambu's actual challenge body, as seen on both the login endpoint and the # design-service endpoints MakerWorld imports use. _CAPTCHA_BODY = { "captchaId": "3f2a9c1e64b04d7f", "error": "We need you to confirm you are not a robot", } @pytest.fixture(autouse=True) def _clear_captcha_cooloff(): """The cool-off map is module-level; don't leak it across tests.""" bc._captcha_blocked_until.clear() yield bc._captcha_blocked_until.clear() def _response(status_code: int, body: object | None = None, *, text: str | None = None): resp = MagicMock() resp.status_code = status_code if body is None and text is not None: resp.json = MagicMock(side_effect=ValueError("not json")) else: resp.json = MagicMock(return_value=body if body is not None else {}) resp.text = text if text is not None else "{}" return resp def _service(response) -> BambuCloudService: svc = BambuCloudService(client=MagicMock(spec=httpx.AsyncClient)) svc._client.post = AsyncMock(return_value=response) svc._client.get = AsyncMock(return_value=response) return svc class TestChallengeIsRecognisedByShape: def test_captcha_id_marks_the_challenge(self): assert bc.is_captcha_challenge(_response(418, _CAPTCHA_BODY)) is True def test_wording_alone_is_enough(self): """No captchaId, but the text says what it is. Bambu has shipped the challenge under more than one body shape.""" assert bc.is_captcha_challenge(_response(418, {"error": "please confirm you are not a robot"})) is True def test_a_418_without_a_marker_is_not_reported_as_a_captcha(self): """Telling a user to solve a CAPTCHA that was never offered is the exact confusion this issue is about -- don't invent one for any stray 418.""" assert bc.is_captcha_challenge(_response(418, {"error": "Too many requests"})) is False def test_status_alone_does_not_decide_it(self): """A captchaId on a 200 is not a refusal -- only the 418 is.""" assert bc.is_captcha_challenge(_response(200, _CAPTCHA_BODY)) is False def test_a_non_json_challenge_is_still_recognised(self): resp = _response(418, None, text="captcha required") assert bc.is_captcha_challenge(resp) is True def test_a_non_json_body_without_markers_is_not(self): resp = _response(418, None, text="Service unavailable") assert bc.is_captcha_challenge(resp) is False class TestSignInReportsTheChallenge: @pytest.mark.asyncio async def test_login_explains_instead_of_echoing_bambu(self): svc = _service(_response(418, _CAPTCHA_BODY)) result = await svc.login_request("user@example.com", "pw") assert result["success"] is False assert result["needs_verification"] is False assert result["reason"] == "captcha" # The regression in one line: this used to BE Bambu's sentence. assert result["message"] != _CAPTCHA_BODY["error"] assert "CAPTCHA" in result["message"] # The two things the reporter had no way to know. assert "password" in result["message"].lower() assert "access token" in result["message"].lower() @pytest.mark.asyncio async def test_email_code_verification_reports_it_too(self): svc = _service(_response(418, _CAPTCHA_BODY)) result = await svc.verify_code("user@example.com", "123456") assert result["reason"] == "captcha" assert result["message"] != _CAPTCHA_BODY["error"] @pytest.mark.asyncio async def test_totp_verification_reports_it_too(self): svc = _service(_response(418, _CAPTCHA_BODY)) with patch.object(svc, "_fetch_csrf_token", AsyncMock(return_value="csrf-token")): result = await svc.verify_totp("tfa-key", "123456") assert result["reason"] == "captcha" assert result["message"] != _CAPTCHA_BODY["error"] @pytest.mark.asyncio async def test_an_ordinary_rejection_is_unchanged(self): """Wrong password still says what Bambu said, and carries no reason -- the UI must keep toasting those rather than showing the CAPTCHA panel.""" svc = _service(_response(400, {"error": "Login failed"})) result = await svc.login_request("user@example.com", "wrong") assert result["message"] == "Login failed" assert result.get("reason") is None assert not bc.captcha_cooloff_active(svc.base_url) class TestRetriesAreHeldBack: @pytest.mark.asyncio async def test_a_second_attempt_is_not_sent_to_bambu(self): """The reporter's log shows four attempts in eighteen seconds. Every one of them is more evidence for the thing that flagged us.""" svc = _service(_response(418, _CAPTCHA_BODY)) await svc.login_request("user@example.com", "pw") assert svc._client.post.await_count == 1 result = await svc.login_request("user@example.com", "pw") assert svc._client.post.await_count == 1 assert result["reason"] == "captcha" @pytest.mark.asyncio async def test_the_cooloff_covers_a_fresh_service_instance(self): """Services are built per request, so the cool-off has to outlive one.""" await _service(_response(418, _CAPTCHA_BODY)).login_request("user@example.com", "pw") second = _service(_response(200, {"loginType": "verifyCode"})) result = await second.login_request("user@example.com", "pw") second._client.post.assert_not_awaited() assert result["reason"] == "captcha" @pytest.mark.asyncio async def test_the_cooloff_expires(self): svc = _service(_response(418, _CAPTCHA_BODY)) await svc.login_request("user@example.com", "pw") bc._captcha_blocked_until[svc.base_url] = bc.time.monotonic() - 1 svc._client.post = AsyncMock(return_value=_response(200, {"loginType": "verifyCode"})) result = await svc.login_request("user@example.com", "pw") assert result["needs_verification"] is True assert bc._captcha_blocked_until == {}, "the expired entry should be dropped on the way past" @pytest.mark.asyncio async def test_a_challenge_on_the_api_host_does_not_strand_a_totp_sign_in(self): """TOTP verification goes to bambulab.com, everything else to api.bambulab.com. Blocking one on the other's behalf would leave a user halfway through two-factor with no way forward.""" svc = _service(_response(418, _CAPTCHA_BODY)) await svc.login_request("user@example.com", "pw") svc._client.post = AsyncMock(return_value=_response(200, {"accessToken": "tok"})) with patch.object(svc, "_fetch_csrf_token", AsyncMock(return_value="csrf-token")): result = await svc.verify_totp("tfa-key", "123456") assert result["success"] is True @pytest.mark.asyncio async def test_the_china_region_is_tracked_separately(self): """The block lives at the edge in front of one region.""" await _service(_response(418, _CAPTCHA_BODY)).login_request("user@example.com", "pw") cn = BambuCloudService(region="china", client=MagicMock(spec=httpx.AsyncClient)) cn._client.post = AsyncMock(return_value=_response(200, {"loginType": "verifyCode"})) result = await cn.login_request("user@example.com", "pw") cn._client.post.assert_awaited_once() assert result["needs_verification"] is True class TestMakerWorldSharesTheDetector: @pytest.mark.asyncio async def test_a_challenge_worded_differently_is_still_named(self): """MakerWorld used to require the literal word "robot" in the error text and reported anything else as an unexplained block.""" from backend.app.services.makerworld import MakerWorldService, MakerWorldUnavailableError svc = MakerWorldService(client=MagicMock(spec=httpx.AsyncClient), auth_token="tok") svc._client.get = AsyncMock(return_value=_response(418, {"captchaId": "abc", "error": "verification required"})) with pytest.raises(MakerWorldUnavailableError) as exc: await svc._get_json("/design/1") assert "CAPTCHA" in str(exc.value) assert "Open on MakerWorld" in str(exc.value) class TestTheSupportBundleNamesIt: def test_the_warning_we_log_matches_the_signature(self, tmp_path, monkeypatch, caplog): """The reporter's bundle came back with zero log-health findings while the log was full of the failure -- tie the two ends together.""" from backend.app.core.config import settings as app_settings from backend.app.services.log_health import scan_logs svc = _service(_response(418, _CAPTCHA_BODY)) with caplog.at_level("WARNING", logger="backend.app.services.bambu_cloud"): svc._note_captcha(_response(418, _CAPTCHA_BODY)) logged = caplog.records[-1].getMessage() log_file = tmp_path / "bambuddy.log" log_file.write_text( f"2026-08-08 05:15:37,068 WARNING [backend.app.services.bambu_cloud] {logged}\n", encoding="utf-8", ) monkeypatch.setattr(app_settings, "log_dir", tmp_path) findings = scan_logs().findings assert [f.signature_id for f in findings] == ["bambu-cloud-captcha"] assert findings[0].wiki_anchor == "bambu-cloud-captcha"