"""Tests for the cloud sync identity flows (services/cloud_account.py). Runs the real flows against a fake in-process cloud (httpx.MockTransport) and a fake in-memory keyring — no network, no OS keychain. The central assertion: the master key and recovery phrase never appear in anything sent to the server. """ import base64 import json from datetime import datetime import httpx import keyring import keyring.backend import pytest from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from backend.database.models import Base, CloudSettings from backend.services import cloud_account, cloud_crypto, cloud_keys from backend.services.cloud_account import CloudAccountError from backend.services.cloud_api import CloudApiClient USER_A = "user-a" class InMemoryKeyring(keyring.backend.KeyringBackend): priority = 1 def __init__(self): super().__init__() self.store: dict[tuple[str, str], str] = {} def get_password(self, service, username): return self.store.get((service, username)) def set_password(self, service, username, password): self.store[(service, username)] = password def delete_password(self, service, username): self.store.pop((service, username), None) class FakeCloud: """Just enough of the cloud API for the identity flows, plus a transcript of every request body so tests can assert what the server was shown.""" def __init__(self): self.devices: dict[str, dict] = {} self.account_key: dict | None = None self.seen_bodies: list[bytes] = [] self._next_id = 0 def transport(self) -> httpx.MockTransport: return httpx.MockTransport(self.handle) def handle(self, request: httpx.Request) -> httpx.Response: if request.content: self.seen_bodies.append(request.content) path, method = request.url.path, request.method if path == "/v1/devices" and method == "POST": body = json.loads(request.content) self._next_id += 1 device_id = f"dev-{self._next_id}" self.devices[device_id] = { "id": device_id, "name": body["name"], "publicKey": body["publicKey"], "wrappedMasterKey": None, "revokedAt": None, } return self._ok({"deviceId": device_id, "accountHasKey": self.account_key is not None}, 201) if path == "/v1/devices" and method == "GET": return self._ok(list(self.devices.values())) if path == "/v1/devices/account-key" and method == "PUT": self.account_key = json.loads(request.content) return self._ok(None) if path == "/v1/devices/account-key" and method == "GET": return self._ok(self.account_key) if path.endswith("/wrapped-key") and method == "POST": device_id = path.split("/")[3] self.devices[device_id]["wrappedMasterKey"] = json.loads(request.content)["wrappedMasterKey"] return self._ok(None) if path.endswith("/wrapped-key") and method == "GET": device_id = path.split("/")[3] return self._ok({"wrappedMasterKey": self.devices[device_id]["wrappedMasterKey"]}) return httpx.Response(404, json={"ok": False, "error": {"message": f"unhandled {method} {path}"}}) @staticmethod def _ok(data, status=200): return httpx.Response(status, json={"ok": True, "data": data}) @pytest.fixture def fake_keyring(monkeypatch): backend = InMemoryKeyring() monkeypatch.setattr(keyring, "get_password", backend.get_password) monkeypatch.setattr(keyring, "set_password", backend.set_password) monkeypatch.setattr(keyring, "delete_password", backend.delete_password) return backend @pytest.fixture def cloud(): return FakeCloud() def make_db(account_user_id=USER_A): engine = create_engine("sqlite://") Base.metadata.create_all(engine) db = sessionmaker(bind=engine)() db.add( CloudSettings( id=1, api_key="voicebox_test", device_name="Test Mac", account_user_id=account_user_id, connected_at=datetime(2026, 7, 1), ) ) db.commit() return db @pytest.fixture def patched_client(monkeypatch, cloud): def _client(row): return CloudApiClient("http://cloud.test", row.api_key, transport=cloud.transport()) monkeypatch.setattr(cloud_account, "_client", _client) @pytest.mark.usefixtures("patched_client", "fake_keyring") class TestIdentityFlows: async def test_first_device_setup(self, cloud): db = make_db() phrase = await cloud_account.setup_device(db) assert phrase is not None assert cloud_crypto.validate_recovery_phrase(phrase) assert cloud_account.identity_status(db).status == "ready" assert cloud.account_key is not None # Registered + provisioned to itself. (device,) = cloud.devices.values() assert device["wrappedMasterKey"] # The invariant: neither MK nor the phrase ever crossed the wire. mk = cloud_account.load_master_key(db) for body in cloud.seen_bodies: assert mk not in body assert base64.b64encode(mk) not in body assert phrase.encode() not in body async def test_second_device_via_provisioning(self, cloud): db_a = make_db() await cloud_account.setup_device(db_a) mk_a = cloud_account.load_master_key(db_a) # Second install: same account, its own DB + keychain namespace. Reuse # the same fake keyring but a distinct account row would collide, so # simulate the second device with a separate account_user_id-scoped # keychain by clearing MK after capturing device state. db_b = make_db(account_user_id="user-a-second-install") assert await cloud_account.setup_device(db_b) is None # account already has key material assert cloud_account.identity_status(db_b).status == "awaiting_provision" assert await cloud_account.adopt_wrapped_key(db_b) is False # nothing provisioned yet target_id = db_b.query(CloudSettings).one().sync_device_id await cloud_account.provision_device(db_a, target_id) assert await cloud_account.adopt_wrapped_key(db_b) is True assert cloud_account.load_master_key(db_b) == mk_a async def test_restore_with_phrase(self, cloud): db_a = make_db() phrase = await cloud_account.setup_device(db_a) mk_a = cloud_account.load_master_key(db_a) db_b = make_db(account_user_id="user-a-fresh-machine") assert await cloud_account.setup_device(db_b) is None await cloud_account.restore_with_phrase(db_b, phrase) assert cloud_account.load_master_key(db_b) == mk_a assert cloud_account.identity_status(db_b).status == "ready" async def test_restore_rejects_wrong_phrase(self, cloud): db_a = make_db() await cloud_account.setup_device(db_a) db_b = make_db(account_user_id="user-a-fresh-machine") await cloud_account.setup_device(db_b) with pytest.raises(cloud_crypto.CloudCryptoError): await cloud_account.restore_with_phrase(db_b, cloud_crypto.generate_recovery_phrase()) async def test_restore_rejects_invalid_phrase_early(self, cloud): db = make_db() await cloud_account.setup_device(db) with pytest.raises(CloudAccountError, match="valid recovery phrase"): await cloud_account.restore_with_phrase( db, "not a real phrase at all twelve words missing checksum here ok" ) async def test_double_registration_rejected(self, cloud): db = make_db() await cloud_account.setup_device(db) with pytest.raises(CloudAccountError, match="already registered"): await cloud_account.setup_device(db) async def test_requires_login(self, cloud): engine = create_engine("sqlite://") Base.metadata.create_all(engine) db = sessionmaker(bind=engine)() with pytest.raises(CloudAccountError, match="log in"): await cloud_account.setup_device(db) @pytest.mark.usefixtures("fake_keyring") class TestKeyStore: def test_round_trip_and_clear(self): cloud_keys.store_secret(USER_A, cloud_keys.MASTER_KEY, b"\x01" * 32) assert cloud_keys.load_secret(USER_A, cloud_keys.MASTER_KEY) == b"\x01" * 32 assert cloud_keys.load_secret("other-user", cloud_keys.MASTER_KEY) is None cloud_keys.clear(USER_A) assert cloud_keys.load_secret(USER_A, cloud_keys.MASTER_KEY) is None def test_delete_absent_is_noop(self): cloud_keys.delete_secret(USER_A, cloud_keys.DEVICE_PRIVATE_KEY)