utils/crdt-sync/crdt_sync/tests/test_e2e_interop.py
Krzysztof kuhy Rudnicki 47f2d239c4
Some checks are pending
crdt_sync_dart tests / test (push) Waiting to run
crdt_sync_flutter tests / test (push) Waiting to run
crdt-sync pre-commit / pre-commit (push) Waiting to run
crdt-sync tests / test (push) Waiting to run
dependency freshness / dependencies on newest stable (push) Waiting to run
design_system tests / test (push) Waiting to run
file length / file-length (push) Waiting to run
md-naming / markdown naming convention (push) Waiting to run
palette drift / palette (push) Waiting to run
repo-contract / repo contract (push) Waiting to run
shared gate tests / gate unit tests (push) Waiting to run
sync_settings_ui tests / test (push) Waiting to run
crdt-sync 0.10.0: E2E port (Argon2id, XChaCha20)
Byte-compatible with crdt_sync_dart 0.14.0, proven by golden vectors in
both directions. EncryptingStore, -e2e layout, one-time copy, and
open_encrypted_client for the Python daemons. Dev pins bumped.

Todo: fda71ed1

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-12 09:47:33 +01:00

75 lines
2.0 KiB
Python

# Copyright (c) 2026 Krzysztof Rudnicki
"""Cross-language golden vectors: Python opens what Dart sealed."""
from __future__ import annotations
import json
from pathlib import Path
import pytest
from crdt_sync import (
E2eDecryptError,
E2eKey,
E2eWrongPassphraseError,
open_log_text,
seal_log_text,
unlock_e2e,
)
from crdt_sync.tests.conftest import FakeStore as MemoryRemote
WRONG = "nope"
VECTORS = json.loads(
(Path(__file__).parent / "fixtures" / "e2e_dart_vectors.json").read_text(
encoding="utf-8"
)
)
def _key() -> E2eKey:
remote = MemoryRemote({"vec/e2e": VECTORS["header"]})
return unlock_e2e(
remote,
path_prefix="vec/notes-e2e",
passphrase=str(VECTORS["passphrase"]),
e2e_path="vec/e2e",
)
def test_argon2id_and_hkdf_match_dart() -> None:
assert _key().master.hex() == VECTORS["master_hex"]
def test_keyed_revision_matches_dart() -> None:
assert _key().revision("abc") == VECTORS["revision_of_abc"]
def test_opens_a_dart_envelope() -> None:
opened = open_log_text(VECTORS["envelope"], _key().cipher(), path=VECTORS["path"])
assert json.loads(opened) == json.loads(VECTORS["log"])
def test_opens_a_dart_whole_blob() -> None:
opened = open_log_text(VECTORS["whole"], _key().cipher(), path=VECTORS["path"])
assert opened == "not a log"
def test_wrong_passphrase_is_refused() -> None:
remote = MemoryRemote({"vec/e2e": VECTORS["header"]})
with pytest.raises(E2eWrongPassphraseError):
unlock_e2e(remote, path_prefix="x", passphrase=WRONG, e2e_path="vec/e2e")
def test_a_moved_envelope_is_refused() -> None:
with pytest.raises(E2eDecryptError):
open_log_text(VECTORS["envelope"], _key().cipher(), path="vec/other")
def test_python_round_trip() -> None:
cipher = _key().cipher()
sealed = seal_log_text(VECTORS["log"], cipher, path=VECTORS["path"])
assert "buy milk" not in sealed
assert json.loads(open_log_text(sealed, cipher, path=VECTORS["path"])) == (
json.loads(VECTORS["log"])
)