public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-19 07:33:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(firmware-check): bypass Cloudflare TLS-fingerprint gate via curl_cffi (#1666)

Browse Source 
Cloudflare on bambulab.com now serves cf-mitigated=challenge to plain
  Python TLS handshakes. Use curl_cffi.AsyncSession with impersonate="chrome"
  for the two bambulab.com fetches (index page + per-model JSON); wiki and
  CDN paths stay on httpx. HTTP User-Agent stays honest "Bambuddy/1.0" —
  only TLS-handshake bytes match Chrome, per the compliance commitment.

  Soft dependency — falls back to httpx with a startup warning when
  curl_cffi isn't importable; wiki-based version detection still works.
This commit is contained in:
maziggy
2026-06-06 13:30:13 +02:00
parent b362f159ac
commit 4bcab89eee
4 changed files with 230 additions and 99 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/firmware_check.py
+139 -70
View File
@@ -20,6 +20,26 @@ from backend.app.core.config import _data_dir
logger = logging.getLogger(__name__)
# Cloudflare on bambulab.com now gates the firmware-download page behind a
# JA3/TLS-fingerprint challenge (cf-mitigated=challenge) that plain Python
# TLS can't pass (#1666). curl_cffi replays Chrome's actual ClientHello
# bytes so the handshake clears CF; we override the HTTP User-Agent back to
# the honest Bambuddy/1.0 string so the application-layer identity stays
# truthful (TLS fingerprint matches Chrome because Python's TLS is the
# signal CF gates on; everything above TLS is still Bambuddy).
#
# Soft dependency — if curl_cffi isn't importable on the running platform,
# firmware_check degrades to httpx (which will likely 403) and wiki-based
# version detection continues to work for the badge; only the in-app
# firmware-download URL stops resolving.
try:
from curl_cffi.requests import AsyncSession as _CurlCffiAsyncSession
_CURL_CFFI_AVAILABLE = True
except ImportError: # pragma: no cover — exercised only on platforms without wheels
_CurlCffiAsyncSession = None # type: ignore[misc,assignment]
_CURL_CFFI_AVAILABLE = False
# Bambu Lab firmware download page (for download URLs)
BAMBU_FIRMWARE_BASE = "https://bambulab.com"
FIRMWARE_PAGE = "/en/support/firmware-download/all"
@@ -121,22 +141,71 @@ class FirmwareCheckService:
self._version_cache: dict[str, FirmwareVersion] = {}
self._versions_list_cache: dict[str, list[FirmwareVersion]] = {}
self._cache_time: float = 0
# Plain httpx client for the Bambu Lab wiki (no CF fingerprint check)
# and other endpoints. Honest UA throughout.
self._client = httpx.AsyncClient(
timeout=30.0,
headers={
# Identify honestly as Bambuddy when scraping the public Bambu
# Lab firmware wiki — verified 2026-05-12 that the wiki serves
# this UA identically to a Chrome UA (same HTML response shape).
# No browser impersonation needed for read-only public pages.
"User-Agent": "Bambuddy/1.0 (+https://github.com/maziggy/bambuddy)",
# Some Cloudflare bot rules on bambulab.com 403 requests with a
# bare UA but no browser-like Accept headers (seen on AU IPs in
# #1350). Sending normal Accept hints removes that signal while
# staying honestly identified via the UA above.
"Accept": "text/html,application/json,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
},
)
# curl_cffi session for bambulab.com (lazily initialised on first use).
# See module-level note on why this is needed.
self._bambulab_client: object | None = None
if not _CURL_CFFI_AVAILABLE:
logger.warning(
"curl_cffi not installed — bambulab.com firmware-download page "
"will likely return Cloudflare 403 (#1666). Wiki-based version "
"detection still works; install curl_cffi to also resolve "
"in-app firmware download URLs."
)
def _get_bambulab_client(self) -> object | None:
"""Lazy-init curl_cffi async session for bambulab.com.
Chrome TLS impersonation is required to pass Cloudflare's JA3
challenge. The HTTP `User-Agent` is overridden back to the honest
Bambuddy string so application-layer identity stays truthful.
Returns None when curl_cffi is unavailable.
"""
if not _CURL_CFFI_AVAILABLE:
return None
if self._bambulab_client is None:
assert _CurlCffiAsyncSession is not None # type-narrow for mypy
self._bambulab_client = _CurlCffiAsyncSession(
impersonate="chrome",
headers={
"User-Agent": "Bambuddy/1.0 (+https://github.com/maziggy/bambuddy)",
"Accept": "text/html,application/json,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
},
timeout=30,
)
return self._bambulab_client
async def _bambulab_get(self, url: str) -> tuple[int, str]:
"""GET against bambulab.com. Returns (status_code, body_text).
Routes through curl_cffi when available (passes the CF JA3 gate);
falls back to httpx otherwise (will likely 403 but worth the try
in case CF eases the check). status 0 indicates a transport error.
"""
client = self._get_bambulab_client()
if client is not None:
try:
response = await client.get(url) # type: ignore[attr-defined]
return response.status_code, response.text
except Exception as e:
logger.error("curl_cffi error fetching %s: %s", url, e)
return 0, ""
try:
response = await self._client.get(url)
return response.status_code, response.text
except Exception as e:
logger.error("httpx error fetching %s: %s", url, e)
return 0, ""
def _build_id_cache_path(self) -> Path:
cache_dir = _data_dir / "firmware"
@@ -188,28 +257,29 @@ class FirmwareCheckService:
self._build_id_time = disk_time
# 3. Live fetch
try:
response = await self._client.get(f"{BAMBU_FIRMWARE_BASE}{FIRMWARE_PAGE}")
if response.status_code == 200:
match = re.search(r'"buildId":"([^"]+)"', response.text)
if match:
new_build_id = match.group(1)
if new_build_id != self._build_id:
logger.info("Got Bambu Lab build ID: %s", new_build_id)
self._build_id = new_build_id
self._build_id_time = time.time()
self._download_page_unreachable = False
self._save_build_id_to_disk(new_build_id)
return self._build_id
else:
# 403/5xx — keep stale cached buildId if we have one (#1350).
logger.warning(
"Failed to get Bambu Lab page: %s (will try cached buildId if available)",
response.status_code,
)
self._download_page_unreachable = True
except Exception as e:
logger.error("Error fetching Bambu Lab build ID: %s", e)
status, body = await self._bambulab_get(f"{BAMBU_FIRMWARE_BASE}{FIRMWARE_PAGE}")
if status == 200:
match = re.search(r'"buildId":"([^"]+)"', body)
if match:
new_build_id = match.group(1)
if new_build_id != self._build_id:
logger.info("Got Bambu Lab build ID: %s", new_build_id)
self._build_id = new_build_id
self._build_id_time = time.time()
self._download_page_unreachable = False
self._save_build_id_to_disk(new_build_id)
return self._build_id
elif status == 0:
# Transport-level error already logged by _bambulab_get.
self._download_page_unreachable = True
else:
# 403/5xx — keep stale cached buildId if we have one (#1350).
# Without curl_cffi this is the expected outcome on Cloudflare
# JA3-gated zones (#1666).
logger.warning(
"Failed to get Bambu Lab page: %s (will try cached buildId if available)",
status,
)
self._download_page_unreachable = True
# Return whatever we have — even a stale buildId beats nothing.
@@ -302,50 +372,49 @@ class FirmwareCheckService:
return []
for attempt in range(2):
try:
url = f"{BAMBU_FIRMWARE_BASE}/_next/data/{build_id}/en/support/firmware-download/{api_key}.json"
response = await self._client.get(url)
url = f"{BAMBU_FIRMWARE_BASE}/_next/data/{build_id}/en/support/firmware-download/{api_key}.json"
status, body = await self._bambulab_get(url)
if response.status_code == 200:
data = response.json()
page_props = data.get("pageProps", {})
printer_map = page_props.get("printerMap", {})
printer_data = printer_map.get(api_key, {})
versions = printer_data.get("versions", [])
return [
FirmwareVersion(
version=v.get("version", ""),
download_url=v.get("url", ""),
release_notes=v.get("release_notes_en"),
release_time=v.get("release_time"),
)
for v in versions
if v.get("version")
]
if status == 200:
try:
data = json.loads(body)
except ValueError as e:
logger.debug("Download-page JSON for %s parse error: %s", api_key, e)
return []
page_props = data.get("pageProps", {})
printer_map = page_props.get("printerMap", {})
printer_data = printer_map.get(api_key, {})
versions = printer_data.get("versions", [])
return [
FirmwareVersion(
version=v.get("version", ""),
download_url=v.get("url", ""),
release_notes=v.get("release_notes_en"),
release_time=v.get("release_time"),
)
for v in versions
if v.get("version")
]
# 404 with cached buildId → Bambu rebuilt the page; invalidate
# and retry once. Other status codes (403, 5xx) are upstream
# blocks — don't churn.
if response.status_code == 404 and attempt == 0:
logger.info("Cached Bambu buildId stale (404), refreshing")
self._build_id = None
self._build_id_time = 0
build_id = await self._get_build_id()
if not build_id:
return []
continue
# 404 with cached buildId → Bambu rebuilt the page; invalidate
# and retry once. Other status codes (403, 5xx) are upstream
# blocks — don't churn.
if status == 404 and attempt == 0:
logger.info("Cached Bambu buildId stale (404), refreshing")
self._build_id = None
self._build_id_time = 0
build_id = await self._get_build_id()
if not build_id:
return []
continue
# 403 from the JSON endpoint is the same Cloudflare block
# signal as on the index page (#1350).
if response.status_code == 403:
self._download_page_unreachable = True
# 403 from the JSON endpoint is the same Cloudflare block
# signal as on the index page (#1350, #1666).
if status == 403:
self._download_page_unreachable = True
logger.debug("Download-page JSON for %s returned status %s", api_key, response.status_code)
return []
except Exception as e:
logger.debug("Error fetching download page firmware for %s: %s", api_key, e)
return []
logger.debug("Download-page JSON for %s returned status %s", api_key, status)
return []
return []
backend/tests/unit/test_firmware_versions.py
+77 -29
View File
@@ -164,9 +164,10 @@ async def test_get_available_versions_sorts_newest_first():
@pytest.mark.asyncio
async def test_client_headers_identify_honestly_and_send_browser_accept():
"""
The httpx client must identify as Bambuddy (no Chrome impersonation) and
must send Accept + Accept-Language so Cloudflare on bambulab.com doesn't
403 us for looking like a bare scraper (#1350).
The httpx client (used for the Bambu wiki and other non-CF-gated paths)
must identify as Bambuddy at the HTTP layer and must send Accept +
Accept-Language so Cloudflare on bambulab.com doesn't 403 us for
looking like a bare scraper (#1350).
"""
svc = FirmwareCheckService()
headers = svc._client.headers
@@ -176,17 +177,72 @@ async def test_client_headers_identify_honestly_and_send_browser_accept():
assert "Accept-Language" in headers
@pytest.mark.asyncio
async def test_bambulab_curl_cffi_session_keeps_honest_user_agent():
"""
The curl_cffi session impersonates Chrome at the TLS layer (required to
pass Cloudflare's JA3 challenge on bambulab.com per #1666), but the
HTTP-layer User-Agent MUST stay 'Bambuddy/...'. A future refactor that
drops the headers= override would silently revert to curl_cffi's
Chrome-default UA and break our compliance commitment to identify
honestly at the application layer.
Skipped when curl_cffi is not installed (constrained-platform fallback
path — covered by the import-guard test below).
"""
from backend.app.services import firmware_check as fc_module
if not fc_module._CURL_CFFI_AVAILABLE:
pytest.skip("curl_cffi not installed in this environment")
svc = FirmwareCheckService()
client = svc._get_bambulab_client()
assert client is not None
# curl_cffi's AsyncSession exposes the configured default headers on
# `.headers`. The exact attribute is part of curl_cffi's public API
# since 0.7.x — if a future upgrade renames it, this test will surface
# the rename rather than silently passing.
session_headers = client.headers # type: ignore[attr-defined]
ua = session_headers.get("User-Agent", "")
assert ua.startswith("Bambuddy/"), f"curl_cffi session UA leaked Chrome default: {ua!r}"
assert "Chrome" not in ua
assert "Mozilla" not in ua
@pytest.mark.asyncio
async def test_bambulab_get_falls_back_to_httpx_when_curl_cffi_missing(monkeypatch):
"""
When curl_cffi can't be imported (constrained platforms, alpine without
wheels, etc.), the service must still attempt the fetch via httpx so
wiki-based version detection isn't accidentally taken down with the
download-URL path. The httpx attempt will likely 403 against CF — that's
OK; the user still gets the version badge and a clear error in the UI.
"""
from backend.app.services import firmware_check as fc_module
monkeypatch.setattr(fc_module, "_CURL_CFFI_AVAILABLE", False)
monkeypatch.setattr(fc_module, "_CurlCffiAsyncSession", None)
svc = FirmwareCheckService()
mock_resp = MagicMock(status_code=403, text="<html>Forbidden</html>")
with patch.object(svc._client, "get", AsyncMock(return_value=mock_resp)):
status, body = await svc._bambulab_get("https://bambulab.com/whatever")
assert status == 403
assert "Forbidden" in body
@pytest.mark.asyncio
async def test_build_id_is_persisted_to_disk(tmp_path, monkeypatch):
"""Successful buildId fetch writes to disk so it survives restart (#1350)."""
monkeypatch.setattr("backend.app.services.firmware_check._data_dir", tmp_path)
svc = FirmwareCheckService()
mock_resp = MagicMock()
mock_resp.status_code = 200
mock_resp.text = 'window.__data = {"buildId":"abc123xyz","other":"stuff"}'
body = 'window.__data = {"buildId":"abc123xyz","other":"stuff"}'
with patch.object(svc._client, "get", AsyncMock(return_value=mock_resp)):
with patch.object(svc, "_bambulab_get", AsyncMock(return_value=(200, body))):
build_id = await svc._get_build_id()
assert build_id == "abc123xyz"
@@ -200,7 +256,7 @@ async def test_build_id_is_persisted_to_disk(tmp_path, monkeypatch):
@pytest.mark.asyncio
async def test_build_id_falls_back_to_disk_on_403(tmp_path, monkeypatch):
"""
When bambulab.com 403s (Cloudflare block reported in #1350) we must
When bambulab.com 403s (Cloudflare block reported in #1350, #1666) we must
fall back to the disk-cached buildId from the previous successful fetch.
Without this the user's screenshots happen: wiki version is detected
but the download URL stays empty forever.
@@ -213,11 +269,8 @@ async def test_build_id_falls_back_to_disk_on_403(tmp_path, monkeypatch):
(cache_dir / "build_id.json").write_text(json.dumps({"build_id": "cached_id_42", "fetched_at": 1000.0}))
svc = FirmwareCheckService()
mock_resp = MagicMock()
mock_resp.status_code = 403
mock_resp.text = "<html>Access denied</html>"
with patch.object(svc._client, "get", AsyncMock(return_value=mock_resp)):
with patch.object(svc, "_bambulab_get", AsyncMock(return_value=(403, "<html>Access denied</html>"))):
build_id = await svc._get_build_id()
assert build_id == "cached_id_42"
@@ -233,11 +286,7 @@ async def test_download_page_unreachable_flag_set_on_403_json(tmp_path, monkeypa
svc._build_id = "stale_id"
svc._build_id_time = 9999999999.0 # never expires for this test
mock_resp = MagicMock()
mock_resp.status_code = 403
mock_resp.text = "Forbidden"
with patch.object(svc._client, "get", AsyncMock(return_value=mock_resp)):
with patch.object(svc, "_bambulab_get", AsyncMock(return_value=(403, "Forbidden"))):
result = await svc._fetch_all_versions_from_download_page("x1")
assert result == []
@@ -256,12 +305,8 @@ async def test_download_page_retries_once_when_buildid_stale(tmp_path, monkeypat
svc._build_id = "stale_id"
svc._build_id_time = 9999999999.0
# First call (with stale buildId) → 404
# Second call (with fresh buildId after refresh) → 200 with versions
stale_resp = MagicMock(status_code=404, text="not found")
fresh_resp = MagicMock(status_code=200)
fresh_resp.json = MagicMock(
return_value={
fresh_json_body = json.dumps(
{
"pageProps": {
"printerMap": {
"x1": {
@@ -278,13 +323,16 @@ async def test_download_page_retries_once_when_buildid_stale(tmp_path, monkeypat
}
}
)
# Page fetch for the buildId refresh
page_resp = MagicMock(status_code=200)
page_resp.text = 'foo "buildId":"fresh_id" bar'
# Sequence: stale 404 → page refresh 200 → fresh 200
get = AsyncMock(side_effect=[stale_resp, page_resp, fresh_resp])
with patch.object(svc._client, "get", get):
# Sequence: stale JSON 404 → page refresh 200 (carries fresh buildId) → fresh JSON 200
get = AsyncMock(
side_effect=[
(404, "not found"),
(200, 'foo "buildId":"fresh_id" bar'),
(200, fresh_json_body),
]
)
with patch.object(svc, "_bambulab_get", get):
result = await svc._fetch_all_versions_from_download_page("x1")
assert len(result) == 1