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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
|
"""
One account, several devices, on one node.
Device linking (2026-08-18) made `identities` a table keyed by
`(user_id, pk_ed25519)`, so a person legitimately holds several keys here. The
chat path never followed: the peer registry was keyed by `user_id`, so the
second connection of one account **evicted the first**, and the broadcast loop
skipped recipients by account, so a person's own other devices never received
what they said.
Neither shows up as an error anywhere. The first is a message that silently
reaches nobody after a second device connects and disconnects; the second is a
phone that never shows what was typed on the laptop. Both are
finding F7 (`docs/MESHBAY_DESIGN.md` §13.6), and both are the same "keyed by
account where it should be keyed by connection" mistake as `pin_identity`'s
old INSERT OR REPLACE.
"""
import base64
import hashlib
from pathlib import Path
from aiortc import RTCPeerConnection
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.transport.webrtc_server import WebRTCPeerSession
from conftest import one_root
def _sealed(device_raw: bytes, text: bytes = b"ciphertext") -> dict:
"""A well-formed sealed envelope.
The bytes are not a real ciphertext and do not need to be: the node never
opens one. What it *does* check is the envelope's shape and that the device
is the connection's own, and going through the real `_do_chat_message`
rather than around it is the point — these tests are about delivery, and
delivery now runs after that check.
"""
return {"format": 1, "epoch": 1, "device": device_raw, "ct": text,
"nonce": b"\x02" * 12, "sig": b"\x03" * 64}
def _session(ctx: dict, user_id: str, group_id: str) -> WebRTCPeerSession:
"""A peer session with only what the chat path touches wired up.
Built through the real `__init__` — with a bare RTCPeerConnection, which
costs under a millisecond and opens no socket — so `_registry_key` is the
one production assigns. Constructing it in the test instead would make
these tests agree with the fix by construction, which is precisely the
trap the repo's own notes record.
"""
session = WebRTCPeerSession(pc=RTCPeerConnection(), node_ctx=ctx)
session._group_id = group_id
session._user_id = user_id
session._username = user_id
# Each connection is a distinct device of that account — which is the whole
# subject here, and what `_check_chat_envelope` compares a message against.
session._pinned_pk = base64.b64encode(_device_raw(session)).decode()
session._device_confirmed = True
session.sent = []
session._send = session.sent.append
session._audit = lambda *a, **k: None
session._spawn = lambda coro: coro.close()
return session
def _ctx(tmp_path: Path, group_id: str) -> dict:
index = GroupIndex(group_id=group_id, sk_node=Ed25519PrivateKey.generate())
root = tmp_path / "shared"
root.mkdir(exist_ok=True)
return {"groups": {group_id: {
"index": index, "roots": one_root(root), "chat_store": None,
}}}
GROUP = "g" * 32
def _device_raw(session) -> bytes:
"""A stable 32-byte stand-in for this connection's device key.
Derived from the registry key, so two sessions of one account get two
devices — which is exactly the situation being tested, and a shared one
would make the envelope check pass for the wrong reason.
"""
return hashlib.sha256(session._registry_key.encode()).digest()
def test_two_devices_of_one_account_both_stay_registered(tmp_path):
"""
F7: keyed by `user_id`, the second device overwrote the first, and closing
either then removed the other's entry — so one person's two devices could
never both be reachable.
"""
ctx = _ctx(tmp_path, GROUP)
laptop = _session(ctx, "alice", GROUP)
phone = _session(ctx, "alice", GROUP)
laptop._register_peer()
phone._register_peer()
registry = laptop._peer_registry()
assert len(registry) == 2, (
"one account's two devices must both be in the registry; keyed by "
"user_id the second silently replaced the first")
assert set(registry.values()) == {laptop, phone}
def test_a_message_reaches_the_senders_other_device(tmp_path):
"""
F7, the visible half: the broadcast excluded recipients whose `user_id`
matched the sender's, so everything typed on the laptop was missing from
the phone — with no error and no way to notice but to hold both.
"""
ctx = _ctx(tmp_path, GROUP)
laptop = _session(ctx, "alice", GROUP)
phone = _session(ctx, "alice", GROUP)
bob = _session(ctx, "bob", GROUP)
for s in (laptop, phone, bob):
s._register_peer()
laptop._user_names = lambda: ctx["groups"][GROUP].setdefault("_names", {})
laptop._do_chat_message(_sealed(_device_raw(laptop), b"hello-ciphertext"))
def chats(session):
return [m for m in session.sent if m.get("type") == "chat_msg"]
assert len(chats(phone)) == 1, (
"the sender's other device is an ordinary recipient — it composed "
"nothing and has no local echo to fall back on")
assert chats(phone)[0]["ct"] == b"hello-ciphertext"
assert len(chats(bob)) == 1
assert chats(laptop) == [], "the composing connection must not echo to itself"
def test_closing_one_device_leaves_the_other_connected(tmp_path):
"""
The teardown half. `close()` popped `self._user_id`, so the phone
disconnecting unregistered the laptop, which then received nothing for the
rest of its session while still reading as connected.
"""
ctx = _ctx(tmp_path, GROUP)
laptop = _session(ctx, "alice", GROUP)
phone = _session(ctx, "alice", GROUP)
for s in (laptop, phone):
s._register_peer()
phone._unregister_peer()
assert laptop._registry_key in laptop._peer_registry()
assert len(laptop._peer_registry()) == 1
def test_registry_key_is_per_connection_not_per_account(tmp_path):
"""The property the two tests above depend on, asserted directly."""
ctx = _ctx(tmp_path, GROUP)
a = _session(ctx, "alice", GROUP)
b = _session(ctx, "alice", GROUP)
assert a._registry_key != b._registry_key
|