diff --git a/backend/requirements.txt b/backend/requirements.txt index caafc0e7..c61a51b3 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -7,6 +7,10 @@ pydantic>=2.5.0 sqlalchemy>=2.0.0 alembic>=1.13.0 +# Cloud backup/sync E2E encryption (services/cloud_crypto.py) +pynacl>=1.5.0 +mnemonic>=0.21 + # ML models torch>=2.2.0 transformers>=4.36.0,<=4.57.6 diff --git a/backend/services/cloud_crypto.py b/backend/services/cloud_crypto.py new file mode 100644 index 00000000..31253fb4 --- /dev/null +++ b/backend/services/cloud_crypto.py @@ -0,0 +1,228 @@ +""" +Client-side cryptography for Voicebox Cloud backup & sync. + +This is the auditable half of the cloud's privacy promise: every byte that +leaves this machine is encrypted here first, and the server only ever stores +ciphertext plus routing metadata. The key hierarchy (cloud repo, +``docs/DESIGN.md``): + + Recovery phrase (BIP39) Device X25519 keypairs + | Argon2id(salt, params) | sealed box + v v + Recovery KEK --wrap--> Master Key (MK) <--wrapped to each device + | + | per-blob random Content Key (CK), + | wrapped by MK inside the envelope + v + XChaCha20-Poly1305(CK) over each record/asset blob + +Everything in this module is a pure function over bytes — no I/O, no storage, +no network. Key persistence (OS keychain) and the sync engine live elsewhere. + +Invariants this module enforces: +- The Master Key is random, never derived from anything the server holds + (API keys and wallet keys are auth/entitlement, never encryption roots). +- Each blob's AAD binds it to its logical slot (object, role, version), so a + server cannot substitute one blob for another without decryption failing. +- Content Keys travel only inside the envelope, wrapped by MK — the database + (local and cloud) stays free of key material. +""" + +import json +import secrets +from dataclasses import dataclass + +import nacl.bindings +import nacl.exceptions +import nacl.pwhash +import nacl.utils +from mnemonic import Mnemonic +from nacl.public import PrivateKey, PublicKey, SealedBox +from nacl.secret import SecretBox + +KEY_BYTES = 32 + +# Envelope framing: magic | alg | len(wrapped_ck) | wrapped_ck | nonce | ciphertext +ENVELOPE_MAGIC = b"VBX1" +ALG_XCHACHA20_POLY1305 = 1 +_WRAPPED_CK_LEN_BYTES = 2 +_NONCE_BYTES = nacl.bindings.crypto_aead_xchacha20poly1305_ietf_NPUBBYTES # 24 + +# Argon2id cost for the recovery-phrase KEK. MODERATE (~256 MiB, interactive +# latency) fits a desktop restore flow; the phrase itself already carries +# 128 bits of entropy, so the KDF is hardening, not the main defense. +_KDF_OPSLIMIT = nacl.pwhash.argon2id.OPSLIMIT_MODERATE +_KDF_MEMLIMIT = nacl.pwhash.argon2id.MEMLIMIT_MODERATE + +_mnemonic = Mnemonic("english") + + +class CloudCryptoError(Exception): + """Envelope malformed, key wrong, or ciphertext tampered with.""" + + +# --------------------------------------------------------------------------- +# Master Key + device keys + + +def generate_master_key() -> bytes: + """The 32-byte root of content secrecy. Generated once per account, client-side.""" + return secrets.token_bytes(KEY_BYTES) + + +def generate_device_keypair() -> tuple[bytes, bytes]: + """X25519 (private, public) for this install. The private key never leaves + the device; the public key is registered with the cloud so existing devices + can wrap MK to it.""" + private = PrivateKey.generate() + return bytes(private), bytes(private.public_key) + + +def wrap_master_key_for_device(master_key: bytes, device_public_key: bytes) -> bytes: + """Seal MK to another device's public key (run on an *existing* device when + provisioning a new one). Only the target device's private key can open it.""" + return SealedBox(PublicKey(device_public_key)).encrypt(master_key) + + +def unwrap_master_key_for_device(wrapped: bytes, device_private_key: bytes) -> bytes: + """Open a sealed MK with this device's private key.""" + try: + return SealedBox(PrivateKey(device_private_key)).decrypt(wrapped) + except nacl.exceptions.CryptoError as err: + raise CloudCryptoError("wrapped master key does not match this device key") from err + + +# --------------------------------------------------------------------------- +# Recovery phrase escrow + + +@dataclass(frozen=True) +class RecoveryWrap: + """What the cloud stores in ``account_key``: ciphertext + the KDF context + any client needs to re-derive the KEK from the phrase. Holds no secrets.""" + + wrapped_key: bytes + kdf_salt: bytes + kdf_params: str # JSON, e.g. {"m": ..., "t": ..., "p": 1} + + +def generate_recovery_phrase() -> str: + """12-word BIP39 mnemonic (128-bit entropy) — shown to the user exactly once.""" + return _mnemonic.generate(strength=128) + + +def validate_recovery_phrase(phrase: str) -> bool: + """Word-level checksum validation, for catching typos before an unwrap attempt.""" + return _mnemonic.check(_normalize_phrase(phrase)) + + +def _normalize_phrase(phrase: str) -> str: + return " ".join(phrase.lower().split()) + + +def _derive_kek(phrase: str, salt: bytes, opslimit: int, memlimit: int) -> bytes: + return nacl.pwhash.argon2id.kdf( + KEY_BYTES, + _normalize_phrase(phrase).encode("utf-8"), + salt, + opslimit=opslimit, + memlimit=memlimit, + ) + + +def wrap_master_key_with_phrase(master_key: bytes, phrase: str) -> RecoveryWrap: + """Argon2id-stretch the phrase into a KEK and wrap MK under it.""" + salt = nacl.utils.random(nacl.pwhash.argon2id.SALTBYTES) + kek = _derive_kek(phrase, salt, _KDF_OPSLIMIT, _KDF_MEMLIMIT) + return RecoveryWrap( + wrapped_key=SecretBox(kek).encrypt(master_key), + kdf_salt=salt, + kdf_params=json.dumps({"m": _KDF_MEMLIMIT, "t": _KDF_OPSLIMIT, "p": 1}), + ) + + +def unwrap_master_key_with_phrase(wrap: RecoveryWrap, phrase: str) -> bytes: + """Recover MK on a fresh device from the phrase + the stored KDF context.""" + try: + params = json.loads(wrap.kdf_params) + opslimit, memlimit = int(params["t"]), int(params["m"]) + except (ValueError, KeyError, TypeError) as err: + raise CloudCryptoError("malformed KDF parameters") from err + kek = _derive_kek(phrase, wrap.kdf_salt, opslimit, memlimit) + try: + return SecretBox(kek).decrypt(wrap.wrapped_key) + except nacl.exceptions.CryptoError as err: + raise CloudCryptoError("recovery phrase does not unlock this account key") from err + + +# --------------------------------------------------------------------------- +# Blob envelope + + +def _build_aad(object_id: str, role: str, version: int) -> bytes: + # Binds a blob to its logical slot. Unambiguous because object_id is a UUID + # and role is a fixed token — neither contains ":". + return f"{object_id}:{role}:{version}".encode() + + +def encrypt_blob(plaintext: bytes, master_key: bytes, *, object_id: str, role: str, version: int) -> bytes: + """Produce a self-describing VBX1 envelope: a fresh Content Key wrapped by + MK rides in the header, and the AAD pins the blob to (object, role, version).""" + content_key = secrets.token_bytes(KEY_BYTES) + wrapped_ck = SecretBox(master_key).encrypt(content_key) + nonce = nacl.utils.random(_NONCE_BYTES) + ciphertext = nacl.bindings.crypto_aead_xchacha20poly1305_ietf_encrypt( + plaintext, + _build_aad(object_id, role, version), + nonce, + content_key, + ) + return b"".join( + [ + ENVELOPE_MAGIC, + bytes([ALG_XCHACHA20_POLY1305]), + len(wrapped_ck).to_bytes(_WRAPPED_CK_LEN_BYTES, "big"), + wrapped_ck, + nonce, + ciphertext, + ] + ) + + +def decrypt_blob(envelope: bytes, master_key: bytes, *, object_id: str, role: str, version: int) -> bytes: + """Open a VBX1 envelope. Raises CloudCryptoError if the envelope is + malformed, the key is wrong, the ciphertext was modified, or the blob was + served for a different (object, role, version) slot.""" + offset = len(ENVELOPE_MAGIC) + if envelope[:offset] != ENVELOPE_MAGIC: + raise CloudCryptoError("not a VBX1 envelope") + if len(envelope) < offset + 1 + _WRAPPED_CK_LEN_BYTES: + raise CloudCryptoError("envelope truncated") + alg = envelope[offset] + if alg != ALG_XCHACHA20_POLY1305: + raise CloudCryptoError(f"unsupported envelope algorithm {alg}") + offset += 1 + + wrapped_len = int.from_bytes(envelope[offset : offset + _WRAPPED_CK_LEN_BYTES], "big") + offset += _WRAPPED_CK_LEN_BYTES + wrapped_ck = envelope[offset : offset + wrapped_len] + offset += wrapped_len + nonce = envelope[offset : offset + _NONCE_BYTES] + offset += _NONCE_BYTES + ciphertext = envelope[offset:] + if len(wrapped_ck) != wrapped_len or len(nonce) != _NONCE_BYTES or not ciphertext: + raise CloudCryptoError("envelope truncated") + + try: + content_key = SecretBox(master_key).decrypt(wrapped_ck) + except nacl.exceptions.CryptoError as err: + raise CloudCryptoError("master key does not unwrap this blob's content key") from err + try: + return nacl.bindings.crypto_aead_xchacha20poly1305_ietf_decrypt( + ciphertext, + _build_aad(object_id, role, version), + nonce, + content_key, + ) + except nacl.exceptions.CryptoError as err: + raise CloudCryptoError("blob failed authentication (tampered, or served for the wrong slot)") from err diff --git a/backend/tests/test_cloud_crypto.py b/backend/tests/test_cloud_crypto.py new file mode 100644 index 00000000..1b239ad8 --- /dev/null +++ b/backend/tests/test_cloud_crypto.py @@ -0,0 +1,137 @@ +"""Tests for the cloud E2E crypto primitives (services/cloud_crypto.py).""" + +import pytest + +from backend.services.cloud_crypto import ( + ALG_XCHACHA20_POLY1305, + ENVELOPE_MAGIC, + CloudCryptoError, + RecoveryWrap, + decrypt_blob, + encrypt_blob, + generate_device_keypair, + generate_master_key, + generate_recovery_phrase, + unwrap_master_key_for_device, + unwrap_master_key_with_phrase, + validate_recovery_phrase, + wrap_master_key_for_device, + wrap_master_key_with_phrase, +) + +SLOT = {"object_id": "0c8f6f4e-9f5a-4a2f-8f6a-1d2e3f4a5b6c", "role": "audio", "version": 3} + + +class TestMasterKey: + def test_master_keys_are_random_32_bytes(self): + a, b = generate_master_key(), generate_master_key() + assert len(a) == 32 + assert a != b + + +class TestDeviceWrap: + def test_round_trip(self): + mk = generate_master_key() + private, public = generate_device_keypair() + assert unwrap_master_key_for_device(wrap_master_key_for_device(mk, public), private) == mk + + def test_wrong_device_key_fails(self): + mk = generate_master_key() + _, public = generate_device_keypair() + other_private, _ = generate_device_keypair() + with pytest.raises(CloudCryptoError): + unwrap_master_key_for_device(wrap_master_key_for_device(mk, public), other_private) + + +class TestRecoveryPhrase: + def test_phrase_is_valid_bip39(self): + phrase = generate_recovery_phrase() + assert len(phrase.split()) == 12 + assert validate_recovery_phrase(phrase) + + def test_typo_fails_checksum(self): + words = generate_recovery_phrase().split() + words[0] = "abandon" if words[0] != "abandon" else "ability" + assert not validate_recovery_phrase(" ".join(words)) + + def test_round_trip(self): + mk = generate_master_key() + phrase = generate_recovery_phrase() + assert unwrap_master_key_with_phrase(wrap_master_key_with_phrase(mk, phrase), phrase) == mk + + def test_normalization_tolerates_case_and_whitespace(self): + mk = generate_master_key() + phrase = generate_recovery_phrase() + wrap = wrap_master_key_with_phrase(mk, phrase) + sloppy = f" {phrase.upper().replace(' ', ' ')} \n" + assert unwrap_master_key_with_phrase(wrap, sloppy) == mk + + def test_wrong_phrase_fails(self): + wrap = wrap_master_key_with_phrase(generate_master_key(), generate_recovery_phrase()) + with pytest.raises(CloudCryptoError): + unwrap_master_key_with_phrase(wrap, generate_recovery_phrase()) + + def test_malformed_kdf_params_fail(self): + wrap = wrap_master_key_with_phrase(generate_master_key(), generate_recovery_phrase()) + broken = RecoveryWrap(wrapped_key=wrap.wrapped_key, kdf_salt=wrap.kdf_salt, kdf_params="not json") + with pytest.raises(CloudCryptoError): + unwrap_master_key_with_phrase(broken, generate_recovery_phrase()) + + +class TestEnvelope: + def test_round_trip(self): + mk = generate_master_key() + plaintext = b"capture transcript \xf0\x9f\x8e\x99 and some audio bytes" * 100 + envelope = encrypt_blob(plaintext, mk, **SLOT) + assert envelope[:4] == ENVELOPE_MAGIC + assert envelope[4] == ALG_XCHACHA20_POLY1305 + assert decrypt_blob(envelope, mk, **SLOT) == plaintext + + def test_fresh_content_key_per_blob(self): + mk = generate_master_key() + assert encrypt_blob(b"same", mk, **SLOT) != encrypt_blob(b"same", mk, **SLOT) + + def test_wrong_master_key_fails(self): + envelope = encrypt_blob(b"secret", generate_master_key(), **SLOT) + with pytest.raises(CloudCryptoError): + decrypt_blob(envelope, generate_master_key(), **SLOT) + + @pytest.mark.parametrize( + "slot", + [ + {**SLOT, "object_id": "11111111-2222-3333-4444-555555555555"}, + {**SLOT, "role": "avatar"}, + {**SLOT, "version": 4}, + ], + ) + def test_wrong_slot_fails_aad(self, slot): + mk = generate_master_key() + envelope = encrypt_blob(b"secret", mk, **SLOT) + with pytest.raises(CloudCryptoError): + decrypt_blob(envelope, mk, **slot) + + def test_tampered_ciphertext_fails(self): + mk = generate_master_key() + envelope = bytearray(encrypt_blob(b"secret", mk, **SLOT)) + envelope[-1] ^= 0x01 + with pytest.raises(CloudCryptoError): + decrypt_blob(bytes(envelope), mk, **SLOT) + + def test_bad_magic_and_truncation_fail(self): + mk = generate_master_key() + envelope = encrypt_blob(b"secret", mk, **SLOT) + with pytest.raises(CloudCryptoError): + decrypt_blob(b"NOPE" + envelope[4:], mk, **SLOT) + with pytest.raises(CloudCryptoError): + decrypt_blob(envelope[:20], mk, **SLOT) + + def test_unknown_algorithm_fails(self): + mk = generate_master_key() + envelope = bytearray(encrypt_blob(b"secret", mk, **SLOT)) + envelope[4] = 99 + with pytest.raises(CloudCryptoError): + decrypt_blob(bytes(envelope), mk, **SLOT) + + def test_empty_plaintext_round_trips(self): + mk = generate_master_key() + assert decrypt_blob(encrypt_blob(b"", mk, **SLOT), mk, **SLOT) == b""