mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 11:12:35 +02:00
244 lines
11 KiB
Python
244 lines
11 KiB
Python
"""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="<html><body>captcha required</body></html>")
|
|
assert bc.is_captcha_challenge(resp) is True
|
|
|
|
def test_a_non_json_body_without_markers_is_not(self):
|
|
resp = _response(418, None, text="<html><body>Service unavailable</body></html>")
|
|
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.model_providers.makerworld.errors import MakerWorldUnavailableError
|
|
from backend.app.services.model_providers.makerworld.service import MakerWorldService
|
|
|
|
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"
|