path: root/packages/meshbay-common/tests/test_handshake.py

"""
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"


def test_a_group_this_node_does_not_host_is_refused_with_a_code():
    """
    A client is handed every node the hub registered for a group, and only some
    of them may be able to serve it. Telling "try the next node" apart from
    "you, here, must do something first" is what this code is for: without it
    the client either stopped at the first refusal — which is how a group went
    dark on 2026-09-11 with its real host online — or had to match on wording.
    """
    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)

    group = "g" * 32
    token = _jwt.encode(
        {"sub": "u1", "jti": "j1", "scope": "user", "groups": [group]},
        pem_priv, algorithm="EdDSA")

    # A member of the group, on a node that does not host it.
    with pytest.raises(HandshakeError) as excinfo:
        authorize_token(token, pem_pub, group_id=group, hosted_groups={"other"})
    assert excinfo.value.code == "not_hosted"

    # And the node that does host it still lets them in.
    peer = authorize_token(token, pem_pub, group_id=group, hosted_groups={group})
    assert peer.group_id == group