"""
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 (
    CHALLENGE_PREFIX,
    HANDSHAKE_PREFIX,
    NONCE_LEN,
    ROLE_CLIENT,
    ROLE_NODE,
    AuthorizedPeer,
    HandshakeError,
    authorize_token,
    challenge_transcript,
    handshake_transcript,
    make_proof,
    quic_binding,
    verify_proof,
    webrtc_binding,
)
from meshbay_common.tokens import HUB_API_AUD, MNP_AUD

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",
        # A member presents an MNP-audience token to a node. A hub session token
        # (aud=HUB_API_AUD) is refused here — see test_a_hub_session_token_is_refused.
        "aud": MNP_AUD,
    }
    payload.update(over)
    # A None override omits the claim entirely (e.g. aud=None → no audience),
    # rather than encoding a null value.
    payload = {k: v for k, v in payload.items() if v is not None}
    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_a_hub_session_token_is_refused_by_a_node(hub_key):
    """The core of the audience split: a member hands whatever token it presents
    to the node operator, so it must not be the hub session token (aud=HUB_API_AUD),
    which opens the hub API. Only the MNP-audience token is accepted here."""
    sk_pem, pk_pem = hub_key
    session_token = _token(sk_pem, aud=HUB_API_AUD)
    with pytest.raises(HandshakeError):
        authorize_token(session_token, pk_pem, group_id=GROUP)


def test_a_token_with_no_audience_is_refused(hub_key):
    sk_pem, pk_pem = hub_key
    no_aud = _token(sk_pem, aud=None)
    with pytest.raises(HandshakeError):
        authorize_token(no_aud, pk_pem, group_id=GROUP)


def test_a_token_bound_to_another_node_is_refused(hub_key):
    """E10: a token names the node it is for. A member's token captured by one
    node's operator cannot be replayed to a second node the member also uses."""
    sk_pem, pk_pem = hub_key
    token_for_A = _token(sk_pem, node="node-A-pk")
    # Node B (its own key is 'node-B-pk') refuses it.
    with pytest.raises(HandshakeError, match="this node"):
        authorize_token(token_for_A, pk_pem, group_id=GROUP, node_pk_b64="node-B-pk")
    # Node A accepts it.
    peer = authorize_token(token_for_A, pk_pem, group_id=GROUP, node_pk_b64="node-A-pk")
    assert peer.user_id == "user-1"


def test_a_token_naming_no_node_is_still_accepted(hub_key):
    """Lenient by design: the hub mints an unbound token only for the requester,
    so it grants nothing across accounts, and older callers keep working."""
    sk_pem, pk_pem = hub_key
    peer = authorize_token(_token(sk_pem), pk_pem, group_id=GROUP, node_pk_b64="node-B-pk")
    assert peer.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_the_challenge_transcript_is_its_own_domain():
    """
    MNP 3.4: the node signs the challenge with the same key that signs the ack.
    The two must never be interchangeable — a challenge signature passed off as
    an ack signature would authenticate a node that never proved the GEK.
    """
    challenge = challenge_transcript(GROUP, NONCE_C, NONCE_S, BINDING)
    assert challenge.startswith(CHALLENGE_PREFIX)
    assert challenge != handshake_transcript(ROLE_NODE, GROUP, NONCE_C, NONCE_S, BINDING)
    for other in (
        challenge_transcript("other", NONCE_C, NONCE_S, BINDING),
        challenge_transcript(GROUP, b"x" * 32, NONCE_S, BINDING),
        challenge_transcript(GROUP, NONCE_C, b"y" * 32, BINDING),
        challenge_transcript(GROUP, NONCE_C, NONCE_S, BINDING + b"z"),
    ):
        assert other != challenge


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 import serialization
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

    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": [],
                         "exp": int(time.time()) + 3600, "aud": MNP_AUD},
                        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 import serialization
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

    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],
         "exp": int(time.time()) + 3600, "aud": MNP_AUD},
        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