From 25602555b18eba00cf8cce20b6385c5143040e6a Mon Sep 17 00:00:00 2001 From: Jamie Pine Date: Wed, 1 Jul 2026 15:16:04 -0700 Subject: [PATCH] Add the cloud sync engine Walks the local store and maps it onto the cloud object model per the design doc: captures, generations (+ version wavs), profiles (+ samples, avatar), and the settings singletons. Records carry the full row (paths storage-relative so restores re-anchor under the destination data dir); audio travels as per-asset ciphertext blobs. Envelopes are randomized, so a new cloud_sync_state table remembers both the plaintext fingerprint (did content actually change?) and the ciphertext hash the server holds (re-declare unchanged blobs without re-encrypting or re-uploading). updated_at is excluded from fingerprints - its onupdate trigger fires as a side effect of applying pulls and would bounce synced objects otherwise. Push runs before pull; conflicts are last-writer-wins per object. Own pushes echoing back through the feed are recognized by ciphertext hash and skipped. Tests simulate two machines (own DB, data dir, keychain) syncing through an in-process fake cloud: full backup/restore with byte-identical audio, incremental edit (only the record blob re-uploads), delete propagation, LWW convergence, and the blindness check that server storage only ever holds VBX1 envelopes. --- backend/database/__init__.py | 2 + backend/database/migrations.py | 7 +- backend/database/models.py | 34 +- backend/services/cloud_sync.py | 535 ++++++++++++++++++++++++++++ backend/tests/fake_cloud.py | 188 ++++++++++ backend/tests/test_cloud_account.py | 54 +-- backend/tests/test_cloud_sync.py | 278 +++++++++++++++ 7 files changed, 1042 insertions(+), 56 deletions(-) create mode 100644 backend/services/cloud_sync.py create mode 100644 backend/tests/fake_cloud.py create mode 100644 backend/tests/test_cloud_sync.py diff --git a/backend/database/__init__.py b/backend/database/__init__.py index fd1252bf..7abe7cdb 100644 --- a/backend/database/__init__.py +++ b/backend/database/__init__.py @@ -12,6 +12,7 @@ from .models import ( CaptureSettings, ChannelDeviceMapping, CloudSettings, + CloudSyncState, EffectPreset, Generation, GenerationSettings, @@ -34,6 +35,7 @@ __all__ = [ "CaptureSettings", "ChannelDeviceMapping", "CloudSettings", + "CloudSyncState", "EffectPreset", "Generation", "GenerationSettings", diff --git a/backend/database/migrations.py b/backend/database/migrations.py index e589b673..0e1abcb7 100644 --- a/backend/database/migrations.py +++ b/backend/database/migrations.py @@ -285,13 +285,16 @@ def _migrate_mcp_bindings(engine, inspector, tables: set[str]) -> None: def _migrate_cloud_settings(engine, inspector, tables: set[str]) -> None: - """Add ``sync_device_id`` — the server-assigned id from registering this - install as an encryption-capable sync device.""" + """Add the cloud sync columns: ``sync_device_id`` (the server-assigned id + from registering this install as an encryption-capable sync device) and + ``sync_cursor`` (highest applied seq from the sync feed).""" if "cloud_settings" not in tables: return columns = _get_columns(inspector, "cloud_settings") if "sync_device_id" not in columns: _add_column(engine, "cloud_settings", "sync_device_id VARCHAR", "sync_device_id") + if "sync_cursor" not in columns: + _add_column(engine, "cloud_settings", "sync_cursor INTEGER NOT NULL DEFAULT 0", "sync_cursor") def _supports_drop_column(engine) -> bool: diff --git a/backend/database/models.py b/backend/database/models.py index e726d6f4..4623a528 100644 --- a/backend/database/models.py +++ b/backend/database/models.py @@ -3,7 +3,7 @@ from datetime import datetime import uuid -from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Boolean, JSON +from sqlalchemy import Column, String, Integer, Float, DateTime, Text, ForeignKey, Boolean, JSON, UniqueConstraint from sqlalchemy.ext.declarative import declarative_base from ..utils.capture_chords import ( @@ -257,9 +257,41 @@ class CloudSettings(Base): # this install registers as an encryption-capable device; the matching # X25519 private key lives in the OS keychain (services/cloud_keys.py). sync_device_id = Column(String, nullable=True) + # Highest server seq this install has applied from the sync feed. + sync_cursor = Column(Integer, nullable=False, default=0) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) +class CloudSyncState(Base): + """Per-entity sync bookkeeping for cloud backup (one row per synced object). + + Envelopes are randomized (fresh content key + nonce per encryption), so the + ciphertext hash changes on every re-encrypt even when the content didn't. + To keep the server's changed-hash dedup working, this table remembers what + was last uploaded: the plaintext fingerprints (to detect real local edits) + and the ciphertext hashes the server currently holds (to re-declare + unchanged blobs without re-encrypting or re-uploading them). + """ + + __tablename__ = "cloud_sync_state" + __table_args__ = (UniqueConstraint("kind", "client_id", name="uq_cloud_sync_state_kind_client"),) + + id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4())) + kind = Column(String, nullable=False) # capture | generation | profile | settings + client_id = Column(String, nullable=False) # the local entity id + server_object_id = Column(String, nullable=True) + # Client-bumped LWW version (object.version on the server). + version = Column(Integer, nullable=False, default=1) + # SHA-256 of the canonical plaintext record JSON at last push/pull. + record_fingerprint = Column(String, nullable=True) + # SHA-256 + size of the record ciphertext the server currently holds. + record_hash = Column(String, nullable=True) + record_size = Column(Integer, nullable=False, default=0) + # Per-asset bookkeeping, JSON: {clientAssetId: {role, fingerprint, hash, size}} + assets_json = Column(Text, nullable=False, default="{}") + last_synced_at = Column(DateTime, nullable=True) + + class MCPClientBinding(Base): """Per-MCP-client settings (voice profile, engine, personality default). diff --git a/backend/services/cloud_sync.py b/backend/services/cloud_sync.py new file mode 100644 index 00000000..3b0cf4a5 --- /dev/null +++ b/backend/services/cloud_sync.py @@ -0,0 +1,535 @@ +""" +The cloud sync engine: encrypted backup + multi-device restore. + +Walks the local store (SQLite rows + audio files), maps each entity onto the +cloud's object model, and drives the push/pull loop against the blind server. +Everything crosses the wire as VBX1 ciphertext (``cloud_crypto``); the server +only ever learns kinds, ids, sizes, and hashes. + +Mapping (cloud repo ``docs/DESIGN.md`` §5): + +| local entity | kind | record (encrypted JSON) | assets | +| ------------------------------------- | ---------- | ------------------------- | ----------------- | +| ``captures`` row + wav | capture | the row | the capture audio | +| ``generations`` row + version wavs | generation | the row + version rows | each version wav | +| ``profiles`` row + samples + avatar | profile | the row + sample rows | sample wavs, avatar | +| ``capture_settings`` / ``generation_settings`` | settings | the row | — | + +Path columns are stored storage-relative inside the (encrypted) record, so a +restore re-anchors them under the destination machine's data dir. + +Change detection: envelopes are randomized, so ``CloudSyncState`` keeps the +plaintext fingerprint (did the content actually change?) alongside the +ciphertext hash the server holds (re-declare unchanged blobs without +re-encrypting). Conflicts are last-writer-wins per object, matching §6 — +push runs before pull, so local edits are declared before remote state lands. + +AAD binding: records are bound to ``(clientId, "record", version)`` and +re-encrypted on every version bump. Asset blobs are bound to +``(clientId, "asset:", 1)`` — assets are content-addressed and +practically immutable (audio never changes in place), so their slot binding +doesn't chase the object version. +""" + +import hashlib +import json +import logging +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path + +from sqlalchemy.orm import Session + +from .. import config +from ..database import ( + Capture, + CaptureSettings, + CloudSettings as DBCloudSettings, + CloudSyncState, + Generation, + GenerationSettings, + GenerationVersion, + ProfileSample, + VoiceProfile, +) +from . import cloud_account, cloud_crypto +from .cloud_api import CloudApiClient + +logger = logging.getLogger(__name__) + +_ASSET_AAD_VERSION = 1 + + +class CloudSyncError(Exception): + """Sync could not run or an object failed to round-trip.""" + + +# --------------------------------------------------------------------------- +# Local object collection (push side) + + +@dataclass(frozen=True) +class LocalAsset: + client_asset_id: str + role: str # audio | version | sample | avatar + path: Path + + +@dataclass(frozen=True) +class LocalObject: + kind: str + client_id: str + record: dict + assets: list[LocalAsset] = field(default_factory=list) + + +_PATH_COLUMNS = {"audio_path", "avatar_path"} + + +def _row_to_record(row) -> dict: + """All mapped columns as JSON-safe values; paths storage-relative, + datetimes ISO-8601.""" + record: dict = {} + for column in row.__mapper__.columns: + value = getattr(row, column.key) + if value is None: + record[column.key] = None + elif column.key in _PATH_COLUMNS: + # Rows normally hold data-dir-relative paths already; only rebase + # absolute ones. (to_storage_path on a relative value would resolve + # it against the CWD and corrupt it.) + record[column.key] = config.to_storage_path(value) if Path(value).is_absolute() else value + elif isinstance(value, datetime): + record[column.key] = value.isoformat() + else: + record[column.key] = value + return record + + +def _is_datetime_column(column) -> bool: + try: + return column.type.python_type is datetime + except NotImplementedError: # e.g. JSON columns don't declare a python_type + return False + + +def _record_to_row(model, record: dict, existing=None): + """Build or update a model instance from a record dict.""" + row = existing if existing is not None else model() + for column in row.__mapper__.columns: + if column.key not in record: + continue + value = record[column.key] + if isinstance(value, str) and _is_datetime_column(column): + value = datetime.fromisoformat(value) + setattr(row, column.key, value) + return row + + +def _existing_path(value: str | None) -> Path | None: + resolved = config.resolve_storage_path(value) + return resolved if resolved is not None and resolved.exists() else None + + +def _collect_captures(db: Session) -> list[LocalObject]: + objects = [] + for row in db.query(Capture).all(): + assets = [] + if (path := _existing_path(row.audio_path)) is not None: + assets.append(LocalAsset(client_asset_id=row.id, role="audio", path=path)) + objects.append(LocalObject(kind="capture", client_id=row.id, record=_row_to_record(row), assets=assets)) + return objects + + +def _collect_generations(db: Session) -> list[LocalObject]: + objects = [] + for row in db.query(Generation).filter(Generation.status == "completed").all(): + record = _row_to_record(row) + assets = [] + if (path := _existing_path(row.audio_path)) is not None: + assets.append(LocalAsset(client_asset_id=row.id, role="audio", path=path)) + versions = db.query(GenerationVersion).filter(GenerationVersion.generation_id == row.id).all() + record["versions"] = [_row_to_record(v) for v in versions] + for version in versions: + if (path := _existing_path(version.audio_path)) is not None: + assets.append(LocalAsset(client_asset_id=version.id, role="version", path=path)) + objects.append(LocalObject(kind="generation", client_id=row.id, record=record, assets=assets)) + return objects + + +def _collect_profiles(db: Session) -> list[LocalObject]: + objects = [] + for row in db.query(VoiceProfile).all(): + record = _row_to_record(row) + assets = [] + if (path := _existing_path(row.avatar_path)) is not None: + assets.append(LocalAsset(client_asset_id=f"{row.id}-avatar", role="avatar", path=path)) + samples = db.query(ProfileSample).filter(ProfileSample.profile_id == row.id).all() + record["samples"] = [_row_to_record(s) for s in samples] + for sample in samples: + if (path := _existing_path(sample.audio_path)) is not None: + assets.append(LocalAsset(client_asset_id=sample.id, role="sample", path=path)) + objects.append(LocalObject(kind="profile", client_id=row.id, record=record, assets=assets)) + return objects + + +def _collect_settings(db: Session) -> list[LocalObject]: + objects = [] + for client_id, model in (("capture_settings", CaptureSettings), ("generation_settings", GenerationSettings)): + row = db.query(model).first() + if row is not None: + objects.append(LocalObject(kind="settings", client_id=client_id, record=_row_to_record(row))) + return objects + + +def collect_local_objects(db: Session) -> list[LocalObject]: + return _collect_captures(db) + _collect_generations(db) + _collect_profiles(db) + _collect_settings(db) + + +# --------------------------------------------------------------------------- +# Applying pulled records (pull side) + + +def _write_asset(record_path_value: str | None, data: bytes) -> None: + resolved = config.resolve_storage_path(record_path_value) + if resolved is None: + return + resolved.parent.mkdir(parents=True, exist_ok=True) + resolved.write_bytes(data) + + +def _apply_children(db: Session, model, parent_filter, child_records: list[dict], blobs: dict[str, bytes]) -> None: + """Upsert child rows (versions/samples) by id; drop local children the + record no longer contains; write any pulled audio next to them.""" + wanted = {child["id"] for child in child_records} + for stale in db.query(model).filter(parent_filter).all(): + if stale.id not in wanted: + db.delete(stale) + for child in child_records: + existing = db.query(model).filter(model.id == child["id"]).first() + row = _record_to_row(model, child, existing) + if existing is None: + db.add(row) + if child["id"] in blobs: + _write_asset(child.get("audio_path"), blobs[child["id"]]) + + +def _apply_capture(db: Session, client_id: str, record: dict, blobs: dict[str, bytes]) -> None: + existing = db.query(Capture).filter(Capture.id == client_id).first() + row = _record_to_row(Capture, record, existing) + if existing is None: + db.add(row) + if client_id in blobs: + _write_asset(record.get("audio_path"), blobs[client_id]) + + +def _apply_generation(db: Session, client_id: str, record: dict, blobs: dict[str, bytes]) -> None: + record = dict(record) + versions = record.pop("versions", []) + existing = db.query(Generation).filter(Generation.id == client_id).first() + row = _record_to_row(Generation, record, existing) + if existing is None: + db.add(row) + if client_id in blobs: + _write_asset(record.get("audio_path"), blobs[client_id]) + _apply_children(db, GenerationVersion, GenerationVersion.generation_id == client_id, versions, blobs) + + +def _apply_profile(db: Session, client_id: str, record: dict, blobs: dict[str, bytes]) -> None: + record = dict(record) + samples = record.pop("samples", []) + existing = db.query(VoiceProfile).filter(VoiceProfile.id == client_id).first() + row = _record_to_row(VoiceProfile, record, existing) + if existing is None: + db.add(row) + if f"{client_id}-avatar" in blobs: + _write_asset(record.get("avatar_path"), blobs[f"{client_id}-avatar"]) + _apply_children(db, ProfileSample, ProfileSample.profile_id == client_id, samples, blobs) + + +def _apply_settings(db: Session, client_id: str, record: dict) -> None: + model = CaptureSettings if client_id == "capture_settings" else GenerationSettings + existing = db.query(model).first() + row = _record_to_row(model, record, existing) + if existing is None: + db.add(row) + + +def _apply_record(db: Session, kind: str, client_id: str, record: dict, blobs: dict[str, bytes]) -> None: + if kind == "capture": + _apply_capture(db, client_id, record, blobs) + elif kind == "generation": + _apply_generation(db, client_id, record, blobs) + elif kind == "profile": + _apply_profile(db, client_id, record, blobs) + elif kind == "settings": + _apply_settings(db, client_id, record) + else: + raise CloudSyncError(f"unknown object kind {kind!r}") + + +def _delete_local(db: Session, kind: str, client_id: str) -> None: + if kind == "capture": + db.query(Capture).filter(Capture.id == client_id).delete() + elif kind == "generation": + db.query(GenerationVersion).filter(GenerationVersion.generation_id == client_id).delete() + db.query(Generation).filter(Generation.id == client_id).delete() + elif kind == "profile": + db.query(ProfileSample).filter(ProfileSample.profile_id == client_id).delete() + db.query(VoiceProfile).filter(VoiceProfile.id == client_id).delete() + # settings singletons are never deleted + + +# --------------------------------------------------------------------------- +# The engine + + +@dataclass +class SyncReport: + pushed: int = 0 + pushed_deletes: int = 0 + pulled: int = 0 + pulled_deletes: int = 0 + cursor: int = 0 + + +def _canonical(record: dict) -> bytes: + """Canonical bytes for change detection. ``updated_at`` is excluded (at the + top level and in embedded child rows): its ``onupdate`` trigger can bump it + as a side effect of *applying* a pulled record, and letting that feed back + into the fingerprint would bounce an already-synced object back and forth. + The field still syncs — it just doesn't count as a change by itself.""" + stripped = {k: v for k, v in record.items() if k != "updated_at"} + for key, value in stripped.items(): + if isinstance(value, list): + stripped[key] = [ + {k: v for k, v in item.items() if k != "updated_at"} if isinstance(item, dict) else item + for item in value + ] + return json.dumps(stripped, sort_keys=True, separators=(",", ":")).encode() + + +def _fingerprint(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _get_state(db: Session, kind: str, client_id: str) -> CloudSyncState | None: + return db.query(CloudSyncState).filter(CloudSyncState.kind == kind, CloudSyncState.client_id == client_id).first() + + +def _settings_row(db: Session) -> DBCloudSettings: + row = db.query(DBCloudSettings).filter(DBCloudSettings.id == 1).first() + if row is None: + raise CloudSyncError("not connected to Voicebox Cloud") + return row + + +async def _push_object( + client: CloudApiClient, + db: Session, + master_key: bytes, + obj: LocalObject, + state: CloudSyncState | None, +) -> bool: + """Push one object if it changed. Returns True when a push happened.""" + record_payload = json.dumps(obj.record, sort_keys=True, separators=(",", ":")).encode() + record_fp = _fingerprint(_canonical(obj.record)) + known_assets: dict = json.loads(state.assets_json) if state else {} + + asset_plain: dict[str, bytes] = {} + asset_fps: dict[str, str] = {} + for asset in obj.assets: + data = asset.path.read_bytes() + asset_plain[asset.client_asset_id] = data + asset_fps[asset.client_asset_id] = _fingerprint(data) + + unchanged = ( + state is not None + and state.server_object_id is not None + and state.record_fingerprint == record_fp + and {k: v["fingerprint"] for k, v in known_assets.items()} == asset_fps + ) + if unchanged: + return False + + version = (state.version + 1) if state is not None else 1 + record_env = cloud_crypto.encrypt_blob( + record_payload, master_key, object_id=obj.client_id, role="record", version=version + ) + + descriptors = [] + envelopes: dict[str, bytes] = {} + next_assets: dict[str, dict] = {} + for asset in obj.assets: + caid = asset.client_asset_id + known = known_assets.get(caid) + if known and known["fingerprint"] == asset_fps[caid]: + # Content unchanged: re-declare the ciphertext the server holds. + entry = {"role": asset.role, "fingerprint": asset_fps[caid], "hash": known["hash"], "size": known["size"]} + else: + envelope = cloud_crypto.encrypt_blob( + asset_plain[caid], + master_key, + object_id=obj.client_id, + role=f"asset:{caid}", + version=_ASSET_AAD_VERSION, + ) + envelopes[caid] = envelope + entry = { + "role": asset.role, + "fingerprint": asset_fps[caid], + "hash": _fingerprint(envelope), + "size": len(envelope), + } + next_assets[caid] = entry + descriptors.append({"role": asset.role, "clientAssetId": caid, "hash": entry["hash"], "size": entry["size"]}) + + pushed = await client.push_object( + kind=obj.kind, + client_id=obj.client_id, + version=version, + record={"hash": _fingerprint(record_env), "size": len(record_env)}, + assets=descriptors, + ) + uploads = {u["for"]: u["url"] for u in pushed["uploads"]} + if "record" in uploads: + await client.upload_blob(uploads["record"], record_env) + for caid, envelope in envelopes.items(): + url = uploads.get(f"asset:{caid}") + if url: + await client.upload_blob(url, envelope) + await client.commit_object(pushed["objectId"]) + + if state is None: + state = CloudSyncState(kind=obj.kind, client_id=obj.client_id) + db.add(state) + state.server_object_id = pushed["objectId"] + state.version = version + state.record_fingerprint = record_fp + state.record_hash = _fingerprint(record_env) + state.record_size = len(record_env) + state.assets_json = json.dumps(next_assets) + state.last_synced_at = datetime.utcnow() + return True + + +async def _push_all(client: CloudApiClient, db: Session, master_key: bytes, report: SyncReport) -> None: + local = collect_local_objects(db) + local_ids = {(o.kind, o.client_id) for o in local} + + for obj in local: + if await _push_object(client, db, master_key, obj, _get_state(db, obj.kind, obj.client_id)): + report.pushed += 1 + db.commit() + + # Local deletions: state rows whose entity no longer exists → tombstone. + for state in db.query(CloudSyncState).all(): + if (state.kind, state.client_id) not in local_ids and state.kind != "settings": + if state.server_object_id: + await client.delete_object(state.server_object_id) + db.delete(state) + report.pushed_deletes += 1 + db.commit() + + +async def _pull_changes(client: CloudApiClient, db: Session, master_key: bytes, report: SyncReport) -> None: + settings = _settings_row(db) + cursor = settings.sync_cursor or 0 + + while True: + page = await client.get_changes(since=cursor) + for change in page["changes"]: + kind, client_id = change["kind"], change["clientId"] + state = _get_state(db, kind, client_id) + + if change["deleted"]: + if state is not None: + _delete_local(db, kind, client_id) + db.delete(state) + report.pulled_deletes += 1 + # Our own pushes echo back through the feed; the stored ciphertext + # hash identifies them as already applied. + elif change["record"] and (state is None or state.record_hash != change["record"]["hash"]): + await _apply_change(client, db, master_key, change, state) + report.pulled += 1 + + cursor = change["seq"] + + settings.sync_cursor = cursor + db.commit() + report.cursor = cursor + if not page["hasMore"]: + break + + +async def _apply_change( + client: CloudApiClient, + db: Session, + master_key: bytes, + change: dict, + state: CloudSyncState | None, +) -> None: + kind, client_id = change["kind"], change["clientId"] + record_cipher = await client.download_blob(change["record"]["url"]) + record = json.loads( + cloud_crypto.decrypt_blob( + record_cipher, master_key, object_id=client_id, role="record", version=change["version"] + ) + ) + + known_assets: dict = json.loads(state.assets_json) if state else {} + blobs: dict[str, bytes] = {} + next_assets: dict[str, dict] = {} + for asset in change["assets"]: + caid = asset["clientAssetId"] + known = known_assets.get(caid) + if known and known["hash"] == asset["hash"]: + next_assets[caid] = known + continue # ciphertext we already hold locally + if not asset["url"]: + continue # declared but never uploaded; skip until it lands + cipher = await client.download_blob(asset["url"]) + plain = cloud_crypto.decrypt_blob( + cipher, master_key, object_id=client_id, role=f"asset:{caid}", version=_ASSET_AAD_VERSION + ) + blobs[caid] = plain + next_assets[caid] = { + "role": asset["role"], + "fingerprint": _fingerprint(plain), + "hash": asset["hash"], + "size": asset["size"], + } + + _apply_record(db, kind, client_id, record, blobs) + + if state is None: + state = CloudSyncState(kind=kind, client_id=client_id) + db.add(state) + state.server_object_id = change["id"] + state.version = change["version"] + state.record_fingerprint = _fingerprint(_canonical(record)) + state.record_hash = change["record"]["hash"] + state.record_size = change["record"]["size"] + state.assets_json = json.dumps(next_assets) + state.last_synced_at = datetime.utcnow() + + +async def run_sync(db: Session) -> SyncReport: + """One full sync: push local changes, then pull and apply remote ones.""" + settings = _settings_row(db) + master_key = cloud_account.load_master_key(db) + report = SyncReport() + + async with CloudApiClient(config.get_cloud_api_url(), settings.api_key) as client: + await _push_all(client, db, master_key, report) + await _pull_changes(client, db, master_key, report) + + logger.info( + "cloud sync: pushed %d (+%d deletes), pulled %d (+%d deletes), cursor %d", + report.pushed, + report.pushed_deletes, + report.pulled, + report.pulled_deletes, + report.cursor, + ) + return report diff --git a/backend/tests/fake_cloud.py b/backend/tests/fake_cloud.py new file mode 100644 index 00000000..b890bd86 --- /dev/null +++ b/backend/tests/fake_cloud.py @@ -0,0 +1,188 @@ +"""An in-process fake of the voicebox-cloud API for tests. + +Implements the surface the desktop sync client uses — devices, the account-key +escrow, the encrypted object store, the sync feed, and blob storage — behind an +httpx.MockTransport, mirroring apps/api in the voicebox-cloud repo. Every +request body is kept in ``seen_bodies`` so tests can assert what the server was +shown (never key material, never plaintext). +""" + +import json + +import httpx + +STORAGE_HOST = "http://cloud.test/__storage/" + + +class FakeCloud: + def __init__(self): + self.devices: dict[str, dict] = {} + self.account_key: dict | None = None + self.objects: dict[str, dict] = {} # objectId -> row (incl. assets dict) + self.storage: dict[str, bytes] = {} # key -> ciphertext + self.seen_bodies: list[bytes] = [] + self._next_device = 0 + self._next_object = 0 + self._seq = 0 + + def transport(self) -> httpx.MockTransport: + return httpx.MockTransport(self.handle) + + # -- helpers -------------------------------------------------------------- + + @staticmethod + def _ok(data, status=200): + return httpx.Response(status, json={"ok": True, "data": data}) + + def _seq_next(self) -> int: + self._seq += 1 + return self._seq + + def _find_object(self, kind: str, client_id: str) -> dict | None: + return next((o for o in self.objects.values() if o["kind"] == kind and o["clientId"] == client_id), None) + + # -- request routing -------------------------------------------------------- + + 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.startswith("/__storage/"): + key = path[len("/__storage/") :] + if method == "PUT": + self.storage[key] = request.content + return httpx.Response(200) + data = self.storage.get(key) + return httpx.Response(200, content=data) if data is not None else httpx.Response(404) + + if path == "/v1/devices" and method == "POST": + return self._register_device(json.loads(request.content)) + 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.startswith("/v1/devices/") and path.endswith("/wrapped-key"): + device_id = path.split("/")[3] + if method == "POST": + self.devices[device_id]["wrappedMasterKey"] = json.loads(request.content)["wrappedMasterKey"] + return self._ok(None) + return self._ok({"wrappedMasterKey": self.devices[device_id]["wrappedMasterKey"]}) + + if path == "/v1/objects" and method == "POST": + return self._upsert_object(json.loads(request.content)) + if path.startswith("/v1/objects/") and path.endswith("/commit"): + return self._commit(path.split("/")[3]) + if path.startswith("/v1/objects/") and method == "DELETE": + obj = self.objects.get(path.split("/")[3]) + if obj: + obj["deleted"] = True + obj["seq"] = self._seq_next() + return self._ok(None) + + if path == "/v1/sync/changes" and method == "GET": + return self._changes(request.url.params) + + return httpx.Response(404, json={"ok": False, "error": {"message": f"unhandled {method} {path}"}}) + + # -- endpoint implementations ---------------------------------------------- + + def _register_device(self, body: dict) -> httpx.Response: + self._next_device += 1 + device_id = f"dev-{self._next_device}" + 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) + + def _upsert_object(self, body: dict) -> httpx.Response: + obj = self._find_object(body["kind"], body["clientId"]) + if obj is None: + self._next_object += 1 + obj = { + "id": f"obj-{self._next_object}", + "kind": body["kind"], + "clientId": body["clientId"], + "version": 0, + "deleted": False, + "record": None, + "assets": {}, + } + self.objects[obj["id"]] = obj + + obj["version"] = max(obj["version"], body["version"]) + obj["seq"] = self._seq_next() + obj["deleted"] = False + + uploads = [] + record = body.get("record") + if record and (obj["record"] is None or obj["record"]["hash"] != record["hash"]): + key = f"o/{obj['id']}/record" + obj["record"] = {**record, "key": key} + uploads.append({"for": "record", "key": key, "url": STORAGE_HOST + key}) + for asset in body.get("assets", []): + caid = asset["clientAssetId"] + existing = obj["assets"].get(caid) + if existing is None or existing["hash"] != asset["hash"]: + key = f"o/{obj['id']}/a/{caid}" + obj["assets"][caid] = {**asset, "key": key} + uploads.append({"for": f"asset:{caid}", "key": key, "url": STORAGE_HOST + key}) + return self._ok({"objectId": obj["id"], "seq": obj["seq"], "uploads": uploads}, 201) + + def _commit(self, object_id: str) -> httpx.Response: + obj = self.objects.get(object_id) + if obj is None: + return httpx.Response(404, json={"ok": False, "error": {"message": "object not found"}}) + missing = [] + if obj["record"] and obj["record"]["key"] not in self.storage: + missing.append("record") + for caid, asset in obj["assets"].items(): + if asset["key"] not in self.storage: + missing.append(f"asset:{caid}") + if missing: + return httpx.Response(409, json={"ok": False, "error": {"message": f"uploads missing: {missing}"}}) + return self._ok({"objectId": object_id, "committed": True}) + + def _changes(self, params) -> httpx.Response: + since = int(params.get("since", 0)) + limit = int(params.get("limit", 200)) + rows = sorted((o for o in self.objects.values() if o["seq"] > since), key=lambda o: o["seq"])[:limit] + changes = [ + { + "id": o["id"], + "kind": o["kind"], + "clientId": o["clientId"], + "version": o["version"], + "seq": o["seq"], + "deleted": o["deleted"], + "record": ( + { + "hash": o["record"]["hash"], + "size": o["record"]["size"], + "url": STORAGE_HOST + o["record"]["key"], + } + if o["record"] + else None + ), + "assets": [ + { + "clientAssetId": caid, + "role": a["role"], + "hash": a["hash"], + "size": a["size"], + "url": STORAGE_HOST + a["key"] if a["key"] in self.storage else None, + } + for caid, a in o["assets"].items() + ], + } + for o in rows + ] + cursor = rows[-1]["seq"] if rows else since + return self._ok({"changes": changes, "cursor": cursor, "hasMore": len(rows) == limit}) diff --git a/backend/tests/test_cloud_account.py b/backend/tests/test_cloud_account.py index 796bdf60..fe581450 100644 --- a/backend/tests/test_cloud_account.py +++ b/backend/tests/test_cloud_account.py @@ -6,10 +6,8 @@ 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 @@ -20,6 +18,7 @@ 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 +from backend.tests.fake_cloud import FakeCloud USER_A = "user-a" @@ -41,57 +40,6 @@ class InMemoryKeyring(keyring.backend.KeyringBackend): 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() diff --git a/backend/tests/test_cloud_sync.py b/backend/tests/test_cloud_sync.py new file mode 100644 index 00000000..d7c9bccc --- /dev/null +++ b/backend/tests/test_cloud_sync.py @@ -0,0 +1,278 @@ +"""Tests for the cloud sync engine (services/cloud_sync.py). + +Simulates two installs ("machines") of the desktop app — each with its own +SQLite database, data directory, and keychain — syncing through the in-process +FakeCloud. Covers the full backup → restore path, incremental pushes, deletes, +and the blindness invariant (nothing plaintext ever lands in server storage). +""" + +from datetime import datetime + +import keyring +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from backend import config +from backend.database.models import ( + Base, + Capture, + CaptureSettings, + CloudSettings, + Generation, + GenerationVersion, + ProfileSample, + VoiceProfile, +) +from backend.services import cloud_account, cloud_sync +from backend.services.cloud_api import CloudApiClient +from backend.tests.fake_cloud import FakeCloud + +AUDIO_A = b"RIFFfake-capture-audio" + b"\x11" * 4000 +AUDIO_B = b"RIFFfake-generation-audio" + b"\x22" * 4000 +AUDIO_C = b"RIFFfake-version-audio" + b"\x33" * 4000 +AUDIO_D = b"RIFFfake-sample-audio" + b"\x44" * 4000 +AVATAR = b"\x89PNGfake-avatar" + b"\x55" * 500 + + +class Install: + """One simulated machine: its own DB, data dir, and keychain store.""" + + def __init__(self, root, name: str): + self.data_dir = root / name + self.data_dir.mkdir() + self.keychain: dict[tuple[str, str], str] = {} + engine = create_engine("sqlite://") + Base.metadata.create_all(engine) + self.db = sessionmaker(bind=engine)() + self.db.add( + CloudSettings( + id=1, + api_key=f"voicebox_{name}", + device_name=name, + account_user_id="user-1", + connected_at=datetime(2026, 7, 1), + ) + ) + self.db.commit() + + def activate(self, monkeypatch): + """Point global config + keychain at this machine.""" + config.set_data_dir(self.data_dir) + store = self.keychain + monkeypatch.setattr(keyring, "get_password", lambda s, u: store.get((s, u))) + monkeypatch.setattr(keyring, "set_password", lambda s, u, p: store.__setitem__((s, u), p)) + monkeypatch.setattr(keyring, "delete_password", lambda s, u: store.pop((s, u), None)) + + def write_file(self, relative: str, data: bytes) -> str: + path = self.data_dir / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(data) + return str(path) + + +@pytest.fixture +def cloud(monkeypatch): + fake = FakeCloud() + monkeypatch.setattr( + cloud_sync, + "CloudApiClient", + lambda url, key: CloudApiClient(url, key, transport=fake.transport()), + ) + monkeypatch.setattr( + cloud_account, + "_client", + lambda row: CloudApiClient("http://cloud.test", row.api_key, transport=fake.transport()), + ) + return fake + + +def seed_content(install: Install) -> None: + db = install.db + profile = VoiceProfile( + id="prof-1", + name="Morgan", + description="test voice", + language="en", + avatar_path=install.write_file("profiles/prof-1/avatar.png", AVATAR), + personality="dry wit", + ) + db.add(profile) + db.add( + ProfileSample( + id="samp-1", + profile_id="prof-1", + audio_path=install.write_file("profiles/prof-1/samples/samp-1.wav", AUDIO_D), + reference_text="hello there", + ) + ) + db.add( + Capture( + id="cap-1", + audio_path=install.write_file("captures/cap-1.wav", AUDIO_A), + source="dictation", + language="en", + transcript_raw="the raw transcript", + transcript_refined="the refined transcript", + ) + ) + db.add( + Generation( + id="gen-1", + profile_id="prof-1", + text="hello world", + audio_path=install.write_file("generations/gen-1.wav", AUDIO_B), + status="completed", + source="manual", + ) + ) + db.add( + GenerationVersion( + id="ver-1", + generation_id="gen-1", + label="Take 2", + audio_path=install.write_file("generations/ver-1.wav", AUDIO_C), + ) + ) + db.add(CaptureSettings(id=1, stt_model="turbo", language="auto")) + db.commit() + + +async def connect(install: Install, monkeypatch, phrase: str | None = None) -> str | None: + install.activate(monkeypatch) + result = await cloud_account.setup_device(install.db) + if phrase is not None: + await cloud_account.restore_with_phrase(install.db, phrase) + return result + + +class TestSyncEngine: + async def test_backup_then_restore_on_second_machine(self, cloud, tmp_path, monkeypatch): + a = Install(tmp_path, "machine-a") + phrase = await connect(a, monkeypatch) + seed_content(a) + + report = await cloud_sync.run_sync(a.db) + assert report.pushed == 4 # capture, generation, profile, capture_settings + assert report.pulled == 0 # own echoes are recognized by ciphertext hash + + # Server blindness: every stored blob is a VBX1 envelope, no plaintext. + assert cloud.storage + for blob in cloud.storage.values(): + assert blob[:4] == b"VBX1" + assert b"transcript" not in blob + assert AUDIO_A not in blob + for body in cloud.seen_bodies: + assert b"the raw transcript" not in body + assert b"Morgan" not in body + + # Fresh machine: restore identity from phrase, then pull everything. + b = Install(tmp_path, "machine-b") + await connect(b, monkeypatch, phrase=phrase) + report_b = await cloud_sync.run_sync(b.db) + assert report_b.pulled == report.pushed + + cap = b.db.query(Capture).one() + assert cap.id == "cap-1" + assert cap.transcript_raw == "the raw transcript" + assert (b.data_dir / "captures/cap-1.wav").read_bytes() == AUDIO_A + + prof = b.db.query(VoiceProfile).one() + assert prof.name == "Morgan" + assert prof.personality == "dry wit" + assert (b.data_dir / "profiles/prof-1/avatar.png").read_bytes() == AVATAR + samp = b.db.query(ProfileSample).one() + assert samp.reference_text == "hello there" + assert (b.data_dir / "profiles/prof-1/samples/samp-1.wav").read_bytes() == AUDIO_D + + gen = b.db.query(Generation).one() + assert gen.text == "hello world" + assert (b.data_dir / "generations/gen-1.wav").read_bytes() == AUDIO_B + ver = b.db.query(GenerationVersion).one() + assert ver.label == "Take 2" + assert (b.data_dir / "generations/ver-1.wav").read_bytes() == AUDIO_C + + settings = b.db.query(CaptureSettings).one() + assert settings.stt_model == "turbo" + + # Second sync on B is a no-op in both directions. + report_b2 = await cloud_sync.run_sync(b.db) + assert (report_b2.pushed, report_b2.pulled) == (0, 0) + + async def test_incremental_edit_propagates(self, cloud, tmp_path, monkeypatch): + a = Install(tmp_path, "machine-a") + phrase = await connect(a, monkeypatch) + seed_content(a) + await cloud_sync.run_sync(a.db) + + b = Install(tmp_path, "machine-b") + await connect(b, monkeypatch, phrase=phrase) + await cloud_sync.run_sync(b.db) + + # Edit on A: only the capture should push, and only its record blob + # should re-upload (the audio is unchanged). + a.activate(monkeypatch) + blobs_before = dict(cloud.storage) + cap = a.db.query(Capture).one() + cap.transcript_refined = "edited on machine A" + a.db.commit() + report_a = await cloud_sync.run_sync(a.db) + assert report_a.pushed == 1 + changed_keys = [k for k, v in cloud.storage.items() if blobs_before.get(k) != v] + assert changed_keys == [k for k in changed_keys if k.endswith("/record")] + + b.activate(monkeypatch) + report_b = await cloud_sync.run_sync(b.db) + assert report_b.pulled == 1 + assert b.db.query(Capture).one().transcript_refined == "edited on machine A" + + async def test_delete_propagates(self, cloud, tmp_path, monkeypatch): + a = Install(tmp_path, "machine-a") + phrase = await connect(a, monkeypatch) + seed_content(a) + await cloud_sync.run_sync(a.db) + + b = Install(tmp_path, "machine-b") + await connect(b, monkeypatch, phrase=phrase) + await cloud_sync.run_sync(b.db) + + a.activate(monkeypatch) + cap = a.db.query(Capture).one() + a.db.delete(cap) + a.db.commit() + report_a = await cloud_sync.run_sync(a.db) + assert report_a.pushed_deletes == 1 + + b.activate(monkeypatch) + report_b = await cloud_sync.run_sync(b.db) + assert report_b.pulled_deletes == 1 + assert b.db.query(Capture).count() == 0 + + async def test_last_writer_wins_on_conflict(self, cloud, tmp_path, monkeypatch): + a = Install(tmp_path, "machine-a") + phrase = await connect(a, monkeypatch) + seed_content(a) + await cloud_sync.run_sync(a.db) + + b = Install(tmp_path, "machine-b") + await connect(b, monkeypatch, phrase=phrase) + await cloud_sync.run_sync(b.db) + + # Concurrent edits to the same capture on both machines. + a.activate(monkeypatch) + a.db.query(Capture).one().transcript_refined = "A's edit" + a.db.commit() + await cloud_sync.run_sync(a.db) + + b.activate(monkeypatch) + b.db.query(Capture).one().transcript_refined = "B's edit" + b.db.commit() + await cloud_sync.run_sync(b.db) # B pushes after A: B is the last writer + + a.activate(monkeypatch) + await cloud_sync.run_sync(a.db) + assert a.db.query(Capture).one().transcript_refined == "B's edit" + + b.activate(monkeypatch) + await cloud_sync.run_sync(b.db) + assert b.db.query(Capture).one().transcript_refined == "B's edit"