1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
|
"""
The recovery-wrapped keypair copy (docs/MESHBAY_DESIGN.md §3.6, MNP 0.14).
`bundle_enc_recovery` is a second copy of the identity bundle wrapped under the
account's recovery key. The store has to add the column to a database that
predates it, and a plain re-backup that omits the recovery copy must not erase
one already there.
"""
import aiosqlite
import pytest
from meshbay_node.bundle_store import BundleStore
@pytest.mark.asyncio
async def test_round_trip_with_and_without_recovery(tmp_path):
store = BundleStore(db_path=tmp_path / "bundles.db")
await store.open()
await store.store_keypair("u1", "pass-wrapped-1")
row = await store.fetch_keypair("u1")
assert row == {"bundle_enc": "pass-wrapped-1", "bundle_enc_recovery": None}
await store.store_keypair("u2", "pass-wrapped-2", "recovery-wrapped-2")
row = await store.fetch_keypair("u2")
assert row["bundle_enc"] == "pass-wrapped-2"
assert row["bundle_enc_recovery"] == "recovery-wrapped-2"
assert await store.fetch_keypair("nobody") is None
await store.close()
@pytest.mark.asyncio
async def test_re_backup_without_recovery_keeps_the_existing_copy(tmp_path):
"""A passphrase-change re-wrap sends only bundle_enc; the recovery copy stays."""
store = BundleStore(db_path=tmp_path / "bundles.db")
await store.open()
await store.store_keypair("u1", "v1", "recovery-v1")
await store.store_keypair("u1", "v2") # no recovery arg
row = await store.fetch_keypair("u1")
assert row["bundle_enc"] == "v2"
assert row["bundle_enc_recovery"] == "recovery-v1"
# An explicit new recovery copy does replace it.
await store.store_keypair("u1", "v3", "recovery-v3")
row = await store.fetch_keypair("u1")
assert row == {"bundle_enc": "v3", "bundle_enc_recovery": "recovery-v3"}
await store.close()
@pytest.mark.asyncio
async def test_migration_adds_the_column_to_an_old_database(tmp_path):
db_path = tmp_path / "bundles.db"
# A keypair_bundles table as it looked before MNP 0.14.
async with aiosqlite.connect(str(db_path)) as db:
await db.execute(
"CREATE TABLE keypair_bundles ("
" user_id TEXT PRIMARY KEY,"
" bundle_enc TEXT NOT NULL,"
" stored_at TEXT NOT NULL DEFAULT (datetime('now')))")
await db.execute(
"INSERT INTO keypair_bundles (user_id, bundle_enc) VALUES ('old', 'legacy')")
await db.commit()
store = BundleStore(db_path=db_path)
await store.open() # runs the migration
row = await store.fetch_keypair("old")
assert row == {"bundle_enc": "legacy", "bundle_enc_recovery": None}
await store.store_keypair("old", "legacy", "recovery-now")
row = await store.fetch_keypair("old")
assert row["bundle_enc_recovery"] == "recovery-now"
await store.close()
|