diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_roster_pairing.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_roster_pairing.py | 88 |
1 files changed, 86 insertions, 2 deletions
diff --git a/packages/meshbay-node/tests/test_roster_pairing.py b/packages/meshbay-node/tests/test_roster_pairing.py index 7eb436c..e1b8a38 100644 --- a/packages/meshbay-node/tests/test_roster_pairing.py +++ b/packages/meshbay-node/tests/test_roster_pairing.py @@ -56,6 +56,14 @@ def _keypair(): return sk_ed, pk_ed_b64, pk_x_b64 +def _x_raw(sk_x, pk_x_b64): + """(sk_x_raw, pk_x_raw) — what unwrap_gek_aes wants.""" + return (sk_x.private_bytes(serialization.Encoding.Raw, + serialization.PrivateFormat.Raw, + serialization.NoEncryption()), + base64.b64decode(pk_x_b64)) + + def _session(tmp_path: Path, roster, user_id: str = "grenet", group_id: str | None = None, gek: bytes | None = None, join_policy: str = "invite") -> WebRTCPeerSession: @@ -383,7 +391,8 @@ async def test_hub_membership_alone_yields_no_key(tmp_path, roster): session = _session(tmp_path, roster, user_id="eve", group_id=GROUP, gek=gek) sk_ed, pk_ed_b64, pk_x_b64 = _keypair() - # Pinned on this node (say, for another group) but never admitted to this one. + # Pinned on this node (say, for another group) but never admitted to this + # one, and with no invite waiting for it here. await roster.pin_identity("eve", "eve", pk_ed_b64, pk_x_b64, "code") await session._do_join_request( @@ -391,7 +400,8 @@ async def test_hub_membership_alone_yields_no_key(tmp_path, roster): user_id="eve", group_id=GROUP)) reply = _last(session) - assert reply.get("gek") is False + assert reply.get("ok") is False + assert not reply.get("gek") assert reply.get("reason") == "not_authorized_for_group" assert "wrapped_b64" not in reply @@ -428,6 +438,80 @@ async def test_invite_only_group_still_demands_a_code(tmp_path, roster): assert await roster.get_identity("newcomer") is None +async def test_known_device_can_still_be_invited_to_another_group(tmp_path, roster): + """ + The regression: grenet's device is pinned here as a member of group A. cbesson + invites grenet to invite-only group B; grenet opens B. This used to answer + `not_authorized_for_group` with no way back — the device-linking `known` + fast-path dropped straight into `_join_ok` and never looked at a code. With a + real invite waiting it must ask for the code, and a valid one must admit. + """ + GROUP_A = "a" * 32 + GROUP_B = "b" * 32 + gek_b = generate_gek() + sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full() + + await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") + await roster.set_member(GROUP_A, "grenet", ROLE_MEMBER, "active", "cbesson") + code = await roster.create_invite(GROUP_B, "grenet", ROLE_MEMBER, "cbesson") + + session = _session(tmp_path, roster, user_id="grenet", group_id=GROUP_B, + gek=gek_b, join_policy="invite") + + # First connect (node-wide group_id=""), no code: prompt for it, do not refuse. + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="grenet", group_id="")) + assert _last(session).get("reason") == "code_required" + assert await roster.get_member(GROUP_B, "grenet") is None + + # grenet enters the code cbesson sent through another channel. + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + code=code, user_id="grenet", group_id=GROUP_B)) + + reply = _last(session) + assert reply["ok"] is True and reply["gek"] is True + assert unwrap_gek_aes(reply, *_x_raw(sk_x, pk_x_b64)) == gek_b + member = await roster.get_member(GROUP_B, "grenet") + assert member and member["role"] == ROLE_MEMBER and member["status"] == "active" + + +async def test_known_pin_with_no_invite_still_gets_the_flat_refusal(tmp_path, roster): + """H3 guard: a pinned identity with no membership and no invite waiting for + it here gets `not_authorized_for_group`, not a code prompt.""" + GROUP_B = "b" * 32 + sk_ed, pk_ed_b64, pk_x_b64 = _keypair() + await roster.pin_identity("eve", "eve", pk_ed_b64, pk_x_b64, "code") + await roster.set_member("a" * 32, "eve", ROLE_MEMBER, "active", "cbesson") + + session = _session(tmp_path, roster, user_id="eve", group_id=GROUP_B, + gek=generate_gek(), join_policy="invite") + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="eve", group_id="")) + assert _last(session).get("reason") == "not_authorized_for_group" + + +async def test_known_device_already_a_member_still_needs_no_code(tmp_path, roster): + """Guard for the fix above: an existing member of B joins B with no code.""" + GROUP_B = "b" * 32 + gek_b = generate_gek() + sk_ed, pk_ed_b64, pk_x_b64, sk_x = _keypair_full() + await roster.pin_identity("grenet", "grenet", pk_ed_b64, pk_x_b64, "code") + await roster.set_member(GROUP_B, "grenet", ROLE_MEMBER, "active", "cbesson") + + session = _session(tmp_path, roster, user_id="grenet", group_id=GROUP_B, + gek=gek_b, join_policy="invite") + await session._do_join_request( + _join_msg(session, sk_ed, pk_ed_b64, pk_x_b64, + user_id="grenet", group_id="")) + + reply = _last(session) + assert reply["ok"] is True and reply["gek"] is True + assert unwrap_gek_aes(reply, *_x_raw(sk_x, pk_x_b64)) == gek_b + + async def test_unknown_group_is_invite_only(tmp_path, roster): """ Fail closed: a group whose policy the node cannot read is treated as |