""" Tier 2: a member verifies another member's device for themselves. `docs/desktop-client-v1.md` §4.8, and `docs/chat-sender-keys.md` §13, which recorded why it could not ship with the encryption: **the evidence was not being kept.** `_do_device_add` verified the countersignature and stored only `added_by_pk` — *which* key approved, never the proof — and the transcript binds `nonce_node`, the approving connection's handshake nonce, so even a stored signature was unverifiable by anyone who was not on that connection. So the node half is two things: keep `(sig, nonce, ts)` beside the pin, and relay them to any member of the group who asks. The node deliberately decides nothing here — it hands over evidence, and the client walks the chain. A node that lies is caught by a client that has seen the account before, which is the property, and it is why trust is not something the node is asked to assert. What this does **not** claim, per the convention: nothing is gained at first sight. A member who has never seen Alice has nothing to compare against. """ import base64 import time import pytest from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from conftest import one_root from meshbay_common.crypto import generate_gek, pk_to_b64 from meshbay_common.device import device_add_transcript from meshbay_common.groupbox import PURPOSE_ROSTER, unseal from meshbay_common.join import ROLE_MEMBER from meshbay_common.protocol import MNP from meshbay_node.indexer.group_index import GroupIndex from meshbay_node.roster import open_roster from meshbay_node.transport.webrtc_server import WebRTCPeerSession GROUP = "g" * 32 NONCE = b"\x11" * 32 @pytest.fixture async def roster(tmp_path): r = await open_roster(tmp_path) yield r await r.close() def _keys(): sk_ed = Ed25519PrivateKey.generate() sk_x = Ed25519PrivateKey.generate() # stand-in; only its b64 is used return sk_ed, pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()) def _session(tmp_path, roster, gek, user_id="alice"): shared = tmp_path / "shared" shared.mkdir(exist_ok=True) index = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()) session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = { "roster": roster, "sk_node": index.sk_node, "groups": {GROUP: {"gek": gek, "index": index, "roots": one_root(shared)}}, } session._group_id = GROUP session._user_id = user_id session._username = user_id session._nonce_node = NONCE session._pinned_pk = "" session._device_confirmed = False session.sent = [] session._send = session.sent.append session._audit = lambda *a, **k: None return session async def _add_device(session, roster, approver_sk, approver_pk, new_pk, new_px, user_id="alice"): """Run the real device-add path, so the evidence is stored the real way.""" ts = int(time.time()) transcript = device_add_transcript( node_pk_b64=session._node_pk_b64(), user_id=user_id, pk_ed25519_b64=new_pk, pk_x25519_b64=new_px, nonce_node=NONCE, ts=ts) session._user_id = user_id await session._do_device_add({ "pk_ed25519": new_pk, "pk_x25519": new_px, "ts": ts, "sig": base64.b64encode(approver_sk.sign(transcript)).decode(), }) return ts # ── the evidence is kept ───────────────────────────────────────────────────── async def test_the_countersignature_is_stored_not_discarded(tmp_path, roster): """ The finding that blocked Tier 2. Before this, `add_sig` did not exist and `added_by_pk` was all that survived — which proves nothing to a third party. """ sk_a, pk_a, px_a = _keys() _sk_b, pk_b, px_b = _keys() await roster.pin_identity("alice", "alice", pk_a, px_a, via="code") await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER, status="active", approved_by="op") session = _session(tmp_path, roster, generate_gek()) ts = await _add_device(session, roster, sk_a, pk_a, pk_b, px_b) devices = {d["pk_ed25519"]: d for d in await roster.group_devices(GROUP)} added = devices[pk_b] assert added["added_by_pk"] == pk_a assert added["add_sig"], "the countersignature was thrown away again" assert added["add_ts"] == ts assert base64.b64decode(added["add_nonce"]) == NONCE, ( "without the nonce the stored signature is unverifiable — the " "transcript binds it") async def test_the_stored_evidence_actually_verifies(tmp_path, roster): """ The point of storing it. A third party rebuilds the transcript from the roster alone and checks the signature — no access to the connection that approved it, which is the whole difficulty. """ sk_a, pk_a, px_a = _keys() _sk_b, pk_b, px_b = _keys() await roster.pin_identity("alice", "alice", pk_a, px_a, via="code") await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER, status="active", approved_by="op") session = _session(tmp_path, roster, generate_gek()) await _add_device(session, roster, sk_a, pk_a, pk_b, px_b) devices = {d["pk_ed25519"]: d for d in await roster.group_devices(GROUP)} d = devices[pk_b] transcript = device_add_transcript( node_pk_b64=session._node_pk_b64(), user_id="alice", pk_ed25519_b64=d["pk_ed25519"], pk_x25519_b64=d["pk_x25519"], nonce_node=base64.b64decode(d["add_nonce"]), ts=d["add_ts"]) from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PublicKey, ) pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(d["added_by_pk"])) pk.verify(base64.b64decode(d["add_sig"]), transcript) # raises if wrong async def test_a_first_device_has_no_evidence_and_says_so(tmp_path, roster): """ An operator code admitted it; there is no countersignature and there cannot be. It must read as trust-on-first-use rather than as verified — a client that treated an absent signature as a valid one would verify anything. """ _sk_a, pk_a, px_a = _keys() await roster.pin_identity("alice", "alice", pk_a, px_a, via="code") await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER, status="active", approved_by="op") (d,) = await roster.group_devices(GROUP) assert d["added_by_pk"] == "" and d["add_sig"] == "" # ── the relay ──────────────────────────────────────────────────────────────── async def test_a_member_is_served_the_roster_sealed(tmp_path, roster): """ Any member, not only the operator — that is the point. Sealed under a GEK-derived subkey for the same reason the index is. """ _sk_a, pk_a, px_a = _keys() await roster.pin_identity("alice", "alice", pk_a, px_a, via="code") await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER, status="active", approved_by="op") await roster.set_member(group_id=GROUP, user_id="bob", role=ROLE_MEMBER, status="active", approved_by="op") _sk_b, pk_b, px_b = _keys() await roster.pin_identity("bob", "bob", pk_b, px_b, via="code") gek = generate_gek() session = _session(tmp_path, roster, gek, user_id="bob") await session._do_group_roster_req({}) resp = session.sent[-1] assert resp["type"] == MNP.GROUP_ROSTER_RESP assert "devices" not in resp, "the roster must not travel in clear" payload = unseal(gek, PURPOSE_ROSTER, MNP.GROUP_ROSTER_RESP, GROUP, resp) assert {d["user_id"] for d in payload["devices"]} == {"alice", "bob"} assert payload["node_pk"], "the transcript needs the node key to rebuild" async def test_a_revoked_device_is_not_relayed(tmp_path, roster): """A retired laptop must stop being offered as one of the account's keys.""" sk_a, pk_a, px_a = _keys() _sk_b, pk_b, px_b = _keys() await roster.pin_identity("alice", "alice", pk_a, px_a, via="code") await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER, status="active", approved_by="op") session = _session(tmp_path, roster, generate_gek()) await _add_device(session, roster, sk_a, pk_a, pk_b, px_b) await roster.revoke_device("alice", pk_b) keys = {d["pk_ed25519"] for d in await roster.group_devices(GROUP)} assert keys == {pk_a} async def test_another_groups_members_are_not_disclosed(tmp_path, roster): """ Scoped to this group. A person in two groups on one node is not revealed to the second by being in the first — the roster is member-visible, so its scope *is* the privacy boundary. """ _sk_a, pk_a, px_a = _keys() _sk_c, pk_c, px_c = _keys() await roster.pin_identity("alice", "alice", pk_a, px_a, via="code") await roster.pin_identity("carol", "carol", pk_c, px_c, via="code") await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER, status="active", approved_by="op") await roster.set_member(group_id="h" * 32, user_id="carol", role=ROLE_MEMBER, status="active", approved_by="op") users = {d["user_id"] for d in await roster.group_devices(GROUP)} assert users == {"alice"} async def test_a_substituted_key_carries_no_evidence(tmp_path, roster): """ The attack Tier 2 exists to detect, from the node's side of it. A node that invents a device for an account can put it in the roster — it writes the roster. What it cannot do is produce a countersignature from a key it does not hold, so the fabricated device arrives with `add_sig` empty and no chain reaches it. The client is what refuses to walk to it; this asserts the node cannot manufacture the evidence. """ _sk_a, pk_a, px_a = _keys() _sk_evil, pk_evil, px_evil = _keys() await roster.pin_identity("alice", "alice", pk_a, px_a, via="code") await roster.set_member(group_id=GROUP, user_id="alice", role=ROLE_MEMBER, status="active", approved_by="op") # The node simply writes a second device for Alice, as a malicious one would. await roster.pin_identity("alice", "alice", pk_evil, px_evil, via="device") devices = {d["pk_ed25519"]: d for d in await roster.group_devices(GROUP)} assert devices[pk_evil]["add_sig"] == "", ( "a fabricated device cannot come with a countersignature — if this ever " "holds evidence, the node has been handed a way to mint trust")