mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-08-11 00:30:12 -04:00
9d549050f7
Signing in to Bambu Cloud with an authenticator-app account failed every time with "Invalid code", whatever the code was. Bambu Lab added double- submit CSRF protection to the bambulab.com web origin - which is where, and only where, this service posts the two-factor code. The endpoint refused the request with 403 "CSRF error: missing_cookie" before it ever evaluated the code, and Bambuddy reported that refusal as a bad code. Verified against the live endpoint with a deliberately invalid key: a bare POST returns missing_cookie; GET /api/csrf mints a bbl_csrf_token cookie; a POST carrying only the cookie returns missing_header; a POST carrying the cookie plus an x-bbl-csrf-token header reaches application logic. Landing on the sign-in page first - the intuitive fix - does not help, as that page sets only Cloudflare's __cf_bm. Of five header spellings tried, only x-bbl-csrf-token is accepted, so the tests pin it. verify_totp now performs that handshake against the same origin it will post to (bambulab.cn for the China region - a token minted by the global site is a cookie the .cn endpoint never issued), and declines to submit the code at all when no token can be obtained rather than burning the user's 30-second TOTP window on a request that is certain to be refused. A CSRF refusal now also says the code was never checked instead of masquerading as a wrong code, which is what sent the reporter chasing clock drift and leading-zero parsing. Only TOTP sign-ins were affected. Every other cloud call, the email-code two-factor path included, goes to api.bambulab.com, which is not gated, and existing stored tokens were unaffected throughout. The region-routing test's MockTransport needed teaching about the handshake: it returns one canned response for every request and set no cookie, so the fix correctly refused to POST and the test lost the URL it asserts on. It now mints a token for /api/csrf and additionally checks the handshake stays on the .cn origin.
630 lines
27 KiB
Python
630 lines
27 KiB
Python
"""Integration tests for per-user cloud credentials and cloud endpoint permissions.
|
|
|
|
Regression tests for:
|
|
- Per-user cloud token storage (when auth enabled)
|
|
- Global fallback (when auth disabled)
|
|
- Cloud endpoints use CLOUD_AUTH permission (not SETTINGS_READ)
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
class TestPerUserCloudCredentials:
|
|
"""Tests that cloud credentials are stored per-user when auth is enabled."""
|
|
|
|
@pytest.fixture
|
|
async def user_with_cloud_auth(self, db_session):
|
|
"""Create a user with CLOUD_AUTH permission via a group."""
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.group import Group
|
|
from backend.app.models.user import User
|
|
|
|
group = Group(
|
|
name="CloudUsers",
|
|
permissions=["cloud:auth", "filaments:read", "printers:read", "firmware:read"],
|
|
)
|
|
db_session.add(group)
|
|
await db_session.flush()
|
|
|
|
user = User(
|
|
username="clouduser",
|
|
password_hash=get_password_hash("testpass123"),
|
|
role="user",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
user.groups.append(group)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.fixture
|
|
async def second_user_with_cloud_auth(self, db_session):
|
|
"""Create a second user with CLOUD_AUTH permission."""
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.group import Group
|
|
from backend.app.models.user import User
|
|
|
|
result = await db_session.execute(select(Group).where(Group.name == "CloudUsers"))
|
|
group = result.scalar_one_or_none()
|
|
if not group:
|
|
group = Group(
|
|
name="CloudUsers2",
|
|
permissions=["cloud:auth", "filaments:read", "printers:read", "firmware:read"],
|
|
)
|
|
db_session.add(group)
|
|
await db_session.flush()
|
|
|
|
user = User(
|
|
username="clouduser2",
|
|
password_hash=get_password_hash("testpass456"),
|
|
role="user",
|
|
)
|
|
db_session.add(user)
|
|
await db_session.flush()
|
|
user.groups.append(group)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
return user
|
|
|
|
@pytest.fixture
|
|
async def cloud_auth_token(self, user_with_cloud_auth, async_client: AsyncClient):
|
|
"""Get auth token for user with cloud permissions."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "clouduser", "password": "testpass123"},
|
|
)
|
|
if response.status_code == 200:
|
|
return response.json().get("access_token")
|
|
return None
|
|
|
|
@pytest.fixture
|
|
async def second_auth_token(self, second_user_with_cloud_auth, async_client: AsyncClient):
|
|
"""Get auth token for second user."""
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "clouduser2", "password": "testpass456"},
|
|
)
|
|
if response.status_code == 200:
|
|
return response.json().get("access_token")
|
|
return None
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_status_returns_not_authenticated_by_default(self, async_client: AsyncClient):
|
|
"""Cloud status should show not authenticated when no token is stored."""
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
|
|
response = await async_client.get("/api/v1/cloud/status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["is_authenticated"] is False
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_status_accessible_when_auth_disabled(self, async_client: AsyncClient):
|
|
"""Cloud endpoints should work when auth is disabled (global fallback)."""
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
|
|
response = await async_client.get("/api/v1/cloud/status")
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_status_requires_auth_when_enabled(self, async_client: AsyncClient):
|
|
"""Cloud endpoints should require auth when auth is enabled."""
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
|
|
response = await async_client.get("/api/v1/cloud/status")
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestCloudEndpointPermissions:
|
|
"""Tests that cloud endpoints use CLOUD_AUTH permission, not SETTINGS_READ.
|
|
|
|
Uses JWT tokens created directly (not via login endpoint) to avoid
|
|
test infrastructure complexity with user creation across sessions.
|
|
"""
|
|
|
|
@pytest.fixture
|
|
async def settings_only_setup(self, async_client: AsyncClient):
|
|
"""Create user with settings:read but NOT cloud:auth, return JWT."""
|
|
from backend.app.core.auth import create_access_token, get_password_hash
|
|
from backend.app.core.database import async_session
|
|
from backend.app.models.group import Group
|
|
from backend.app.models.user import User
|
|
|
|
async with async_session() as db:
|
|
group = Group(name="SettingsReaders", permissions=["settings:read"])
|
|
db.add(group)
|
|
user = User(
|
|
username="settingsuser",
|
|
password_hash=get_password_hash("testpass123"),
|
|
role="user",
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(group)
|
|
await db.refresh(user)
|
|
|
|
from sqlalchemy import text
|
|
|
|
await db.execute(
|
|
text("INSERT INTO user_groups (user_id, group_id) VALUES (:uid, :gid)"),
|
|
{"uid": user.id, "gid": group.id},
|
|
)
|
|
await db.commit()
|
|
|
|
return create_access_token(data={"sub": "settingsuser"})
|
|
|
|
@pytest.fixture
|
|
async def cloud_only_setup(self, async_client: AsyncClient):
|
|
"""Create user with cloud:auth but NOT settings:read, return JWT."""
|
|
from backend.app.core.auth import create_access_token, get_password_hash
|
|
from backend.app.core.database import async_session
|
|
from backend.app.models.group import Group
|
|
from backend.app.models.user import User
|
|
|
|
async with async_session() as db:
|
|
group = Group(name="CloudOnly", permissions=["cloud:auth"])
|
|
db.add(group)
|
|
user = User(
|
|
username="cloudonly",
|
|
password_hash=get_password_hash("testpass123"),
|
|
role="user",
|
|
)
|
|
db.add(user)
|
|
await db.commit()
|
|
await db.refresh(group)
|
|
await db.refresh(user)
|
|
|
|
from sqlalchemy import text
|
|
|
|
await db.execute(
|
|
text("INSERT INTO user_groups (user_id, group_id) VALUES (:uid, :gid)"),
|
|
{"uid": user.id, "gid": group.id},
|
|
)
|
|
await db.commit()
|
|
|
|
return create_access_token(data={"sub": "cloudonly"})
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_settings_requires_cloud_auth_not_settings_read(
|
|
self, async_client: AsyncClient, settings_only_setup, cloud_only_setup
|
|
):
|
|
"""GET /cloud/settings should require CLOUD_AUTH, not SETTINGS_READ.
|
|
|
|
Regression test: previously used SETTINGS_READ which blocked users who
|
|
had cloud:auth permission but not settings:read.
|
|
"""
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
|
|
# User with only settings:read should be denied
|
|
response = await async_client.get(
|
|
"/api/v1/cloud/settings",
|
|
headers={"Authorization": f"Bearer {settings_only_setup}"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
# User with cloud:auth should be allowed (will get 401 since no cloud token,
|
|
# but NOT 403 — permission check passes)
|
|
response = await async_client.get(
|
|
"/api/v1/cloud/settings",
|
|
headers={"Authorization": f"Bearer {cloud_only_setup}"},
|
|
)
|
|
assert response.status_code == 401 # No cloud token, but permission OK
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_status_requires_cloud_auth(
|
|
self, async_client: AsyncClient, settings_only_setup, cloud_only_setup
|
|
):
|
|
"""GET /cloud/status should require CLOUD_AUTH."""
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
|
|
# settings:read only → 403
|
|
response = await async_client.get(
|
|
"/api/v1/cloud/status",
|
|
headers={"Authorization": f"Bearer {settings_only_setup}"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
# cloud:auth → 200
|
|
response = await async_client.get(
|
|
"/api/v1/cloud/status",
|
|
headers={"Authorization": f"Bearer {cloud_only_setup}"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_fields_requires_cloud_auth(
|
|
self, async_client: AsyncClient, settings_only_setup, cloud_only_setup
|
|
):
|
|
"""GET /cloud/fields should require CLOUD_AUTH, not SETTINGS_READ."""
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=True):
|
|
# settings:read only → 403
|
|
response = await async_client.get(
|
|
"/api/v1/cloud/fields",
|
|
headers={"Authorization": f"Bearer {settings_only_setup}"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
# cloud:auth → 200
|
|
response = await async_client.get(
|
|
"/api/v1/cloud/fields",
|
|
headers={"Authorization": f"Bearer {cloud_only_setup}"},
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
|
|
class TestCloudTokenStorage:
|
|
"""Unit-level tests for the token storage functions."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_stored_token_returns_none_when_no_user_no_global(self, db_session):
|
|
"""get_stored_token with user=None and no global token returns (None, None)."""
|
|
from backend.app.api.routes.cloud import get_stored_token
|
|
|
|
token, email, region = await get_stored_token(db_session, user=None)
|
|
assert token is None
|
|
assert email is None
|
|
assert region == "global" # default for missing rows
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_and_get_global_token(self, db_session):
|
|
"""store_token with user=None stores in global Settings table."""
|
|
from backend.app.api.routes.cloud import get_stored_token, store_token
|
|
|
|
await store_token(db_session, "test-token-123", "test@example.com", "global", user=None)
|
|
token, email, region = await get_stored_token(db_session, user=None)
|
|
assert token == "test-token-123"
|
|
assert email == "test@example.com"
|
|
assert region == "global"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_store_and_get_per_user_token(self, db_session):
|
|
"""store_token with user stores on the user record."""
|
|
from backend.app.api.routes.cloud import get_stored_token, store_token
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.user import User
|
|
|
|
user = User(username="tokentest", password_hash=get_password_hash("pass"), role="user")
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
await store_token(db_session, "user-token-abc", "user@example.com", "global", user=user)
|
|
|
|
# Re-fetch user to verify persistence
|
|
from sqlalchemy import select
|
|
|
|
result = await db_session.execute(select(User).where(User.id == user.id))
|
|
refreshed = result.scalar_one()
|
|
assert refreshed.cloud_token == "user-token-abc"
|
|
assert refreshed.cloud_email == "user@example.com"
|
|
assert refreshed.cloud_region == "global"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_per_user_token_does_not_affect_global(self, db_session):
|
|
"""Storing per-user token should not affect global Settings."""
|
|
from backend.app.api.routes.cloud import get_stored_token, store_token
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.user import User
|
|
|
|
user = User(username="isolationtest", password_hash=get_password_hash("pass"), role="user")
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
# Store per-user token
|
|
await store_token(db_session, "per-user-token", "per-user@test.com", "global", user=user)
|
|
|
|
# Global should still be empty
|
|
global_token, global_email, _ = await get_stored_token(db_session, user=None)
|
|
assert global_token is None
|
|
assert global_email is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_per_user_token(self, db_session):
|
|
"""clear_token with user clears only that user's credentials."""
|
|
from backend.app.api.routes.cloud import clear_token, store_token
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.user import User
|
|
|
|
user = User(username="cleartest", password_hash=get_password_hash("pass"), role="user")
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
await store_token(db_session, "to-clear", "clear@test.com", "china", user=user)
|
|
await clear_token(db_session, user=user)
|
|
|
|
from sqlalchemy import select
|
|
|
|
result = await db_session.execute(select(User).where(User.id == user.id))
|
|
refreshed = result.scalar_one()
|
|
assert refreshed.cloud_token is None
|
|
assert refreshed.cloud_email is None
|
|
assert refreshed.cloud_region is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clear_global_token(self, db_session):
|
|
"""clear_token with user=None clears from global Settings."""
|
|
from backend.app.api.routes.cloud import clear_token, get_stored_token, store_token
|
|
|
|
await store_token(db_session, "global-token", "global@test.com", "global", user=None)
|
|
await clear_token(db_session, user=None)
|
|
|
|
token, email, region = await get_stored_token(db_session, user=None)
|
|
assert token is None
|
|
assert email is None
|
|
assert region == "global" # normalised default
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_two_users_independent_tokens(self, db_session):
|
|
"""Two users should have completely independent cloud tokens and regions."""
|
|
from backend.app.api.routes.cloud import get_stored_token, store_token
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.user import User
|
|
|
|
user_a = User(username="user_a", password_hash=get_password_hash("pass"), role="user")
|
|
user_b = User(username="user_b", password_hash=get_password_hash("pass"), role="user")
|
|
db_session.add_all([user_a, user_b])
|
|
await db_session.commit()
|
|
await db_session.refresh(user_a)
|
|
await db_session.refresh(user_b)
|
|
|
|
# Different regions on purpose — a China user and a Global user must not
|
|
# bleed their region into each other's lookups.
|
|
await store_token(db_session, "token-a", "a@test.com", "china", user=user_a)
|
|
await store_token(db_session, "token-b", "b@test.com", "global", user=user_b)
|
|
|
|
# Verify each user reads their own token (re-fetch from DB)
|
|
from sqlalchemy import select
|
|
|
|
result_a = await db_session.execute(select(User).where(User.id == user_a.id))
|
|
result_b = await db_session.execute(select(User).where(User.id == user_b.id))
|
|
fresh_a = result_a.scalar_one()
|
|
fresh_b = result_b.scalar_one()
|
|
|
|
token_a, email_a, region_a = await get_stored_token(db_session, user=fresh_a)
|
|
token_b, email_b, region_b = await get_stored_token(db_session, user=fresh_b)
|
|
|
|
assert token_a == "token-a"
|
|
assert email_a == "a@test.com"
|
|
assert region_a == "china"
|
|
assert token_b == "token-b"
|
|
assert email_b == "b@test.com"
|
|
assert region_b == "global"
|
|
|
|
|
|
class TestCloudRegionPersistence:
|
|
"""Region must survive a DB round-trip so restarts don't silently flip users to api.bambulab.com."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_region_survives_roundtrip_per_user(self, db_session):
|
|
"""Stored China region is returned on subsequent get_stored_token calls."""
|
|
from backend.app.api.routes.cloud import get_stored_token, store_token
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.user import User
|
|
|
|
user = User(username="region-user", password_hash=get_password_hash("pass"), role="user")
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
await store_token(db_session, "cn-token", "token-auth", "china", user=user)
|
|
|
|
# Simulate "next request": re-fetch the user fresh from the DB.
|
|
from sqlalchemy import select
|
|
|
|
result = await db_session.execute(select(User).where(User.id == user.id))
|
|
refreshed = result.scalar_one()
|
|
|
|
_token, _email, region = await get_stored_token(db_session, user=refreshed)
|
|
assert region == "china"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_region_survives_roundtrip_global_fallback(self, db_session):
|
|
"""Stored China region in auth-disabled Settings fallback survives too."""
|
|
from backend.app.api.routes.cloud import get_stored_token, store_token
|
|
|
|
await store_token(db_session, "cn-token", "token-auth", "china", user=None)
|
|
_token, _email, region = await get_stored_token(db_session, user=None)
|
|
assert region == "china"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_region_is_normalised_to_global(self, db_session):
|
|
"""Unknown region values fall back to 'global' rather than mis-route."""
|
|
from backend.app.api.routes.cloud import get_stored_token, store_token
|
|
|
|
await store_token(db_session, "t", "x@test.com", "mars", user=None)
|
|
_token, _email, region = await get_stored_token(db_session, user=None)
|
|
assert region == "global"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_build_authenticated_cloud_uses_stored_region(self, db_session):
|
|
"""build_authenticated_cloud wires the stored region into the per-request service."""
|
|
from backend.app.api.routes.cloud import build_authenticated_cloud, store_token
|
|
from backend.app.core.auth import get_password_hash
|
|
from backend.app.models.user import User
|
|
|
|
user = User(username="cn-build", password_hash=get_password_hash("pass"), role="user")
|
|
db_session.add(user)
|
|
await db_session.commit()
|
|
await db_session.refresh(user)
|
|
|
|
await store_token(db_session, "cn-token", "token-auth", "china", user=user)
|
|
|
|
from sqlalchemy import select
|
|
|
|
result = await db_session.execute(select(User).where(User.id == user.id))
|
|
refreshed = result.scalar_one()
|
|
|
|
cloud = await build_authenticated_cloud(db_session, refreshed)
|
|
assert cloud is not None
|
|
try:
|
|
assert cloud.base_url == "https://api.bambulab.cn"
|
|
assert cloud.access_token == "cn-token"
|
|
finally:
|
|
await cloud.close()
|
|
|
|
|
|
class TestCloudRouteRegionPlumbing:
|
|
"""Route-level proof that region=china on the wire actually steers outbound
|
|
HTTP calls to api.bambulab.cn / bambulab.cn. This is the core bug the PR
|
|
fixes — unit tests prove the service does the right thing given the region,
|
|
storage tests prove the region persists, but only these tests prove the
|
|
route handlers plumb the region through end-to-end.
|
|
|
|
Auth is disabled (Settings-fallback path) to keep the fixture footprint
|
|
minimal; the region plumbing code path is identical for the per-user path.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _capturing_client(response_json: dict, status: int = 200):
|
|
"""Build an httpx.AsyncClient backed by MockTransport that records every
|
|
outbound request URL. Returns ``(client, captured_urls)``.
|
|
|
|
Using MockTransport (rather than ``patch.object(httpx.AsyncClient, ...)``)
|
|
is critical: class-level method patches also intercept the ASGI test
|
|
client's own requests, so the route handler never runs and the
|
|
assertions end up inspecting the test-client URL instead of the
|
|
backend's outbound URL. MockTransport only affects the client we
|
|
inject into the backend via ``set_shared_http_client``.
|
|
"""
|
|
import httpx
|
|
|
|
captured: list[str] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured.append(str(request.url))
|
|
# The TOTP path now performs a CSRF handshake first (#2696): it
|
|
# fetches /api/csrf and refuses to submit the code unless that call
|
|
# yields a bbl_csrf_token cookie. Mint one here so region-routing
|
|
# tests reach the TFA POST they are actually asserting on.
|
|
if request.url.path == "/api/csrf":
|
|
return httpx.Response(204, headers={"set-cookie": "bbl_csrf_token=csrf-test-token; Path=/"})
|
|
return httpx.Response(status, json=response_json)
|
|
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
return client, captured
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_set_token_route_with_china_region_hits_cn_endpoint(self, async_client: AsyncClient):
|
|
"""POST /cloud/token with region=china routes get_user_profile to api.bambulab.cn."""
|
|
from backend.app.services.bambu_cloud import set_shared_http_client
|
|
|
|
mock_client, captured_urls = self._capturing_client({"uid": "123", "email": "x"})
|
|
set_shared_http_client(mock_client)
|
|
try:
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
|
|
response = await async_client.post(
|
|
"/api/v1/cloud/token",
|
|
json={"access_token": "cn-token", "region": "china"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert any("api.bambulab.cn" in url for url in captured_urls), captured_urls
|
|
assert not any("api.bambulab.com" in url for url in captured_urls), captured_urls
|
|
finally:
|
|
set_shared_http_client(None)
|
|
await mock_client.aclose()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_login_route_with_china_region_hits_cn_endpoint(self, async_client: AsyncClient):
|
|
"""POST /cloud/login with region=china routes login_request to api.bambulab.cn."""
|
|
from backend.app.services.bambu_cloud import set_shared_http_client
|
|
|
|
mock_client, captured_urls = self._capturing_client({"loginType": "verifyCode"})
|
|
set_shared_http_client(mock_client)
|
|
try:
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
|
|
response = await async_client.post(
|
|
"/api/v1/cloud/login",
|
|
json={"email": "user@example.com", "password": "x", "region": "china"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert any("api.bambulab.cn" in url for url in captured_urls), captured_urls
|
|
assert not any("api.bambulab.com" in url for url in captured_urls), captured_urls
|
|
finally:
|
|
set_shared_http_client(None)
|
|
await mock_client.aclose()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_verify_route_with_china_region_hits_cn_tfa_endpoint(self, async_client: AsyncClient):
|
|
"""POST /cloud/verify with region=china + tfa_key routes TOTP to bambulab.cn."""
|
|
from backend.app.services.bambu_cloud import set_shared_http_client
|
|
|
|
mock_client, captured_urls = self._capturing_client({"token": "t"})
|
|
set_shared_http_client(mock_client)
|
|
try:
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
|
|
response = await async_client.post(
|
|
"/api/v1/cloud/verify",
|
|
json={
|
|
"email": "user@example.com",
|
|
"code": "123456",
|
|
"tfa_key": "tfa-xyz",
|
|
"region": "china",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
# TOTP endpoint lives on bambulab.cn (without the api. prefix),
|
|
# NOT bambulab.com — that's exactly the bug we just fixed.
|
|
assert any("bambulab.cn/api/sign-in/tfa" in url for url in captured_urls), captured_urls
|
|
# The CSRF handshake (#2696) must follow the same origin —
|
|
# fetching a token from the global site would hand the .cn
|
|
# endpoint a cookie it never issued.
|
|
assert any("bambulab.cn/api/csrf" in url for url in captured_urls), captured_urls
|
|
assert not any("bambulab.com" in url for url in captured_urls), captured_urls
|
|
finally:
|
|
set_shared_http_client(None)
|
|
await mock_client.aclose()
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_status_exposes_stored_region(self, async_client: AsyncClient):
|
|
"""GET /cloud/status returns the stored region so the UI can render
|
|
'Connected (China)' after a reload.
|
|
|
|
``validate_token`` is stubbed because the endpoint now asks Bambu whether
|
|
the stored token is still accepted rather than assuming it is — without
|
|
the stub this test would make a live call to api.bambulab.cn with a fake
|
|
token, get a 401, and correctly report the session as expired. Region
|
|
plumbing is what's under test here.
|
|
"""
|
|
from backend.app.api.routes.cloud import store_token
|
|
from backend.app.core.database import async_session
|
|
from backend.app.services.bambu_cloud import BambuCloudService
|
|
|
|
with (
|
|
patch("backend.app.core.auth.is_auth_enabled", return_value=False),
|
|
patch.object(BambuCloudService, "validate_token", AsyncMock(return_value=True)),
|
|
):
|
|
async with async_session() as db:
|
|
await store_token(db, "cn-token", "token-auth", "china", user=None)
|
|
|
|
response = await async_client.get("/api/v1/cloud/status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["is_authenticated"] is True
|
|
assert data["region"] == "china"
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.integration
|
|
async def test_cloud_status_region_is_null_when_unauthenticated(self, async_client: AsyncClient):
|
|
"""No stored token ⇒ no region in the status payload."""
|
|
with patch("backend.app.core.auth.is_auth_enabled", return_value=False):
|
|
response = await async_client.get("/api/v1/cloud/status")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["is_authenticated"] is False
|
|
assert data["region"] is None
|