diff options
Diffstat (limited to 'packages/meshbay-common/tests/test_handshake.py')
| -rw-r--r-- | packages/meshbay-common/tests/test_handshake.py | 225 |
1 files changed, 225 insertions, 0 deletions
diff --git a/packages/meshbay-common/tests/test_handshake.py b/packages/meshbay-common/tests/test_handshake.py new file mode 100644 index 0000000..8981db8 --- /dev/null +++ b/packages/meshbay-common/tests/test_handshake.py @@ -0,0 +1,225 @@ +""" +Unified handshake — properties every transport must inherit (11.5.4/5, C6, C3, L4). + +These test the shared module rather than any one transport. The point of the module +is that WebRTC and QUIC cannot drift apart again: the handshake existed three times +over and only the newest copy enforced the GEK proof. +""" + +import time + +import jwt +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.handshake import ( + HANDSHAKE_PREFIX, + NONCE_LEN, + ROLE_CLIENT, + ROLE_NODE, + AuthorizedPeer, + HandshakeError, + authorize_token, + handshake_transcript, + make_proof, + quic_binding, + verify_proof, + webrtc_binding, +) + +GEK = b"\x11" * 32 +GROUP = "g" * 32 +NONCE_C = b"\x01" * NONCE_LEN +NONCE_S = b"\x02" * NONCE_LEN +BINDING = webrtc_binding(b"\xaa" * 32, b"\xbb" * 32) + + +# ── Token authorization ─────────────────────────────────────────────────────── + +@pytest.fixture +def hub_key(): + sk = Ed25519PrivateKey.generate() + pem = sk.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + pub = sk.public_key().public_bytes( + serialization.Encoding.PEM, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + return pem, pub + + +def _token(sk_pem, **over): + now = int(time.time()) + payload = { + "iss": "test-hub", "sub": "user-1", "jti": "jti-1", + "iat": now, "exp": now + 3600, + "groups": [GROUP], "scope": "user", "pk_user": "pk", + } + payload.update(over) + return jwt.encode(payload, sk_pem, algorithm="EdDSA") + + +def test_valid_token_authorizes(hub_key): + sk_pem, pk_pem = hub_key + peer = authorize_token(_token(sk_pem), pk_pem, group_id=GROUP) + assert isinstance(peer, AuthorizedPeer) + assert peer.user_id == "user-1" + + +def test_group_id_is_mandatory(hub_key): + """ + M1: group_id used to be optional, and omitting it skipped the membership check + entirely while falling back to the node's first group. + """ + sk_pem, pk_pem = hub_key + with pytest.raises(HandshakeError, match="group_id"): + authorize_token(_token(sk_pem), pk_pem, group_id="") + + +def test_non_member_refused(hub_key): + sk_pem, pk_pem = hub_key + token = _token(sk_pem, groups=["other-group"]) + with pytest.raises(HandshakeError, match="Not a member"): + authorize_token(token, pk_pem, group_id=GROUP) + + +def test_node_scoped_token_refused_on_client_path(hub_key): + """M9: a daemon's node-scoped token must not be usable as a client token.""" + sk_pem, pk_pem = hub_key + token = _token(sk_pem, scope="node") + with pytest.raises(HandshakeError, match="scope"): + authorize_token(token, pk_pem, group_id=GROUP) + + +def test_unhosted_group_refused(hub_key): + sk_pem, pk_pem = hub_key + with pytest.raises(HandshakeError, match="not hosted"): + authorize_token(_token(sk_pem), pk_pem, group_id=GROUP, + hosted_groups={"some-other-group"}) + + +def test_denylisted_token_refused(hub_key): + sk_pem, pk_pem = hub_key + + class _Deny: + def is_denied(self, user_id, jti, group_id=""): + return group_id == GROUP + + with pytest.raises(HandshakeError, match="revoked"): + authorize_token(_token(sk_pem), pk_pem, group_id=GROUP, denylist=_Deny()) + + +def test_forged_token_refused(hub_key): + _, pk_pem = hub_key + other = Ed25519PrivateKey.generate().private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + with pytest.raises(HandshakeError, match="Invalid JWT"): + authorize_token(_token(other), pk_pem, group_id=GROUP) + + +# ── Proof transcript ────────────────────────────────────────────────────────── + +def test_transcript_is_domain_separated(): + assert handshake_transcript( + ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING + ).startswith(HANDSHAKE_PREFIX) + + +def test_client_proof_is_not_a_node_proof(): + """ + C3: the node proves itself with the same key over the same connection. Without + the role bound in, a client's proof would satisfy the node check and vice + versa, so an impersonating peer could simply echo it back. + """ + client = make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING) + assert not verify_proof(GEK, client, ROLE_NODE, GROUP, NONCE_C, NONCE_S, BINDING) + + node = make_proof(GEK, ROLE_NODE, GROUP, NONCE_C, NONCE_S, BINDING) + assert not verify_proof(GEK, node, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING) + assert client != node + + +@pytest.mark.parametrize("field,value", [ + ("group_id", "other-group"), + ("nonce_client", b"\x09" * NONCE_LEN), + ("nonce_node", b"\x09" * NONCE_LEN), + ("binding", webrtc_binding(b"\xcc" * 32, b"\xdd" * 32)), +]) +def test_proof_binds_every_field(field, value): + base = dict(role=ROLE_CLIENT, group_id=GROUP, nonce_client=NONCE_C, + nonce_node=NONCE_S, binding=BINDING) + proof = make_proof(GEK, **base) + altered = dict(base, **{field: value}) + assert not verify_proof(GEK, proof, **altered), ( + f"proof ignores {field} — replayable across connections") + + +def test_proof_requires_channel_binding(): + """ + L4/NS5: the old transcript was nonce ‖ offer_fp ‖ answer_fp, and a missing + fingerprint silently degraded it to nonce-only, dropping MitM detection. + """ + with pytest.raises(HandshakeError, match="binding"): + make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, b"") + + assert not verify_proof( + GEK, b"\x00" * 32, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, b"") + + +def test_proof_requires_gek(): + with pytest.raises(HandshakeError): + make_proof(b"", ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING) + + +def test_wrong_gek_fails(): + proof = make_proof(GEK, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING) + assert not verify_proof( + b"\x22" * 32, proof, ROLE_CLIENT, GROUP, NONCE_C, NONCE_S, BINDING) + + +def test_transcript_is_unambiguous(): + """ + L4: with bare concatenation, a crafted group id could impersonate the + following field and two different handshakes would produce identical bytes. + """ + a = handshake_transcript(ROLE_CLIENT, "gg", NONCE_C, NONCE_S, BINDING) + b = handshake_transcript(ROLE_CLIENT, "g", b"g" + NONCE_C, NONCE_S, BINDING) + assert a != b + + +def test_bindings_differ_by_transport(): + """A WebRTC proof must not be replayable on a QUIC connection.""" + assert webrtc_binding(b"\xaa" * 32, b"\xbb" * 32) != quic_binding(b"cert-der") + + +def test_membership_refusal_carries_a_code_a_client_can_act_on(): + """ + `groups` is baked into the token at login, so someone added to a group after + signing in is refused although they are a member. The client refreshes and + retries on this code — it must not have to match on the human wording, which + is exactly the kind of coupling that breaks when someone improves a message. + """ + import jwt as _jwt + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + from cryptography.hazmat.primitives import serialization + + sk = Ed25519PrivateKey.generate() + pem_priv = sk.private_bytes( + serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, + serialization.NoEncryption()) + pem_pub = sk.public_key().public_bytes( + serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo) + + token = _jwt.encode({"sub": "u1", "jti": "j1", "scope": "user", "groups": []}, + pem_priv, algorithm="EdDSA") + + with pytest.raises(HandshakeError) as excinfo: + authorize_token(token, pem_pub, group_id="g" * 32) + assert excinfo.value.code == "not_a_member" |