path: root/packages/meshbay-node/tests/test_device_linking.py

"""
One person, several devices on one node.

Identity keys are per node, so a browser and a desktop client are two keys on
one account. Admitting the second must not need an operator — that friction is
what would make "the native client must not prevent web use" fail — and must not
be something the hub or the node itself can do.

The controls, and the tests that hold them:

  * **A key the node already pinned countersigns.** The hub has stored no user
    keys since 2026-08-14, so it cannot produce that signature.
  * **The code is hashed together with the requesting keys**, so the node cannot
    answer an approver with a substituted key: the approver recomputes the hash
    and finds nothing.
  * **Nothing rests on a human comparing digits.** Phase 12.1 dropped that
    ritual as "correct, unusable as the default"; it must not come back here.

Everything below is written as "this does not work".
"""

import base64
import time
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from conftest import one_root
from meshbay_common.crypto import pk_to_b64
from meshbay_common.device import (
    device_add_transcript,
    device_code_hash,
    device_request_transcript,
)
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 generate_code, normalize_code, 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())


async def _session(tmp_path: Path, roster, user_id: str = "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 = {
        "roots": one_root(shared), "index": index, "sk_node": index.sk_node,
        "roster": roster, "device_request_ttl": 3600,
        "groups": {GROUP: {"gek": b"\x01" * 32, "index": index,
                           "roots": one_root(shared), "join_policy": "invite"}},
    }
    session._group_id = GROUP
    session._user_id = user_id
    session._username = user_id
    session._pk_user = ""
    session._pinned_pk = ""
    session._uploads = {}
    session._nonce_node = NONCE
    session._remote_ip = ""
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


async def test_device_messages_need_an_authenticated_session(tmp_path, roster):
    """
    Filing a device is not a pre-proof message, and must not become one.

    The pre-proof window exists for what a peer needs *in order to* prove
    possession of the group key, and adding a device is not that: the request is
    countersigned later by a device already pinned, so nothing is lost by
    requiring the caller to finish its own handshake first. Left in the window it
    would be a second thing a hub that forges a JWT could reach.

    Driven through the real dispatcher, because this is a property of the order
    of its branches and of nothing else.
    """
    session = await _session(tmp_path, roster)
    session._user_id = None            # challenged, has not proved anything yet
    _sk, pk_ed, pk_x = _keys()

    for mtype in ("device_add_request", "device_hello", "device_lookup",
                  "device_add", "device_list", "device_revoke"):
        session.sent.clear()
        session._dispatch_message({"type": mtype, "pk_ed25519": pk_ed,
                                   "pk_x25519": pk_x})
        assert _last(session) == {"type": "error",
                                  "detail": "Handshake required"}, mtype


def _last(session):
    return session.sent[-1] if session.sent else {}


async def _file_request(session, sk_new, pk_ed, pk_x, code):
    """A new device asks to be added, signing over its own keys."""
    code_hash = device_code_hash(normalize_code(code), pk_ed, pk_x)
    ts = int(time.time())
    transcript = device_request_transcript(
        node_pk_b64=session._node_pk_b64(), user_id=session._user_id,
        pk_ed25519_b64=pk_ed, pk_x25519_b64=pk_x, code_hash=code_hash,
        nonce_node=NONCE, ts=ts)
    await session._do_device_request({
        "pk_ed25519": pk_ed, "pk_x25519": pk_x, "code_hash": code_hash,
        "ts": ts, "sig": base64.b64encode(sk_new.sign(transcript)).decode(),
    })
    return code_hash


async def _approve(session, sk_signer, pk_ed, pk_x, code_hash=""):
    ts = int(time.time())
    transcript = device_add_transcript(
        node_pk_b64=session._node_pk_b64(), user_id=session._user_id,
        pk_ed25519_b64=pk_ed, pk_x25519_b64=pk_x, nonce_node=NONCE, ts=ts)
    await session._do_device_add({
        "pk_ed25519": pk_ed, "pk_x25519": pk_x, "ts": ts,
        "code_hash": code_hash,
        "sig": base64.b64encode(sk_signer.sign(transcript)).decode(),
    })


async def _match_by_code(session, code):
    """
    What an approving client does: list what is pending and recompute.

    The code never reaches the node. The client hashes it against each
    candidate's keys and keeps the row that matches — so a node offering
    fabricated keys produces no match, having no way to compute a hash over a
    code it does not know.
    """
    await session._do_device_lookup({})
    listed = _last(session)
    if listed.get("type") != MNP.DEVICE_LOOKUP_RESULT:
        return None
    for req in listed.get("requests", []):
        expect = device_code_hash(normalize_code(code), req["pk_ed25519"],
                                  req["pk_x25519"])
        if expect == req["code_hash"]:
            return req
    return None


# ── The happy path, so the refusals mean something ───────────────────────────

async def test_an_existing_device_admits_a_new_one(tmp_path, roster):
    sk_old, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_new, pk_new_ed, pk_new_x = _keys()

    session = await _session(tmp_path, roster)
    code = generate_code()
    await _file_request(session, sk_new, pk_new_ed, pk_new_x, code)
    assert _last(session)["type"] == MNP.DEVICE_REQUEST_ACK

    match = await _match_by_code(session, code)
    assert match is not None, "the approver could not find the pending request"
    assert match["pk_ed25519"] == pk_new_ed

    await _approve(session, sk_old, pk_new_ed, pk_new_x,
                   code_hash=match["code_hash"])

    assert _last(session)["type"] == MNP.DEVICE_ADD_ACK
    devices = await roster.list_devices("alice")
    assert {d["pk_ed25519"] for d in devices} == {pk_old_ed, pk_new_ed}
    added = next(d for d in devices if d["pk_ed25519"] == pk_new_ed)
    assert added["added_by_pk"] == pk_old_ed, "provenance is not recorded"


async def test_both_devices_then_open_the_group(tmp_path, roster):
    """The point of the whole exercise: web and native at the same time."""
    sk_old, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    await roster.set_member(GROUP, "alice", ROLE_MEMBER, "active", "grenet")
    _, pk_new_ed, pk_new_x = _keys()
    await roster.pin_identity("alice", "alice", pk_new_ed, pk_new_x, "device",
                              added_by_pk=pk_old_ed)

    for pk in (pk_old_ed, pk_new_ed):
        assert await roster.find_device("alice", pk) is not None
    assert await roster.is_authorized(GROUP, "alice")


# ── What must not work ───────────────────────────────────────────────────────

async def test_the_request_alone_admits_nothing(tmp_path, roster):
    """Filing is inert. A node that pinned here would let anyone with a hub
    token join any account that has ever used it."""
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_new, pk_new_ed, pk_new_x = _keys()

    session = await _session(tmp_path, roster)
    await _file_request(session, sk_new, pk_new_ed, pk_new_x, generate_code())

    assert await roster.find_device("alice", pk_new_ed) is None
    assert [d["pk_ed25519"] for d in await roster.list_devices("alice")] == \
        [pk_old_ed]


async def test_the_new_device_cannot_approve_itself(tmp_path, roster):
    """
    Otherwise anyone the hub can mint a token for walks in: the request is
    self-signed by construction, so self-approval would be no control at all.
    """
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_new, pk_new_ed, pk_new_x = _keys()

    session = await _session(tmp_path, roster)
    await _file_request(session, sk_new, pk_new_ed, pk_new_x, generate_code())
    await _approve(session, sk_new, pk_new_ed, pk_new_x)

    assert _last(session)["type"] == "error"
    assert await roster.find_device("alice", pk_new_ed) is None


async def test_a_stranger_cannot_approve(tmp_path, roster):
    """A key belonging to somebody else, or to nobody, is not this account's."""
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_bob, pk_bob_ed, pk_bob_x = _keys()
    await roster.pin_identity("bob", "bob", pk_bob_ed, pk_bob_x, "code")
    _, pk_new_ed, pk_new_x = _keys()

    session = await _session(tmp_path, roster)
    await _approve(session, sk_bob, pk_new_ed, pk_new_x)

    assert _last(session)["type"] == "error"
    assert await roster.find_device("alice", pk_new_ed) is None


async def test_a_revoked_device_cannot_admit_its_replacement(tmp_path, roster):
    """
    The lost laptop. Marking rather than deleting is what makes this hold: a
    deleted row is a key the node would pin again on the next device-add.
    """
    sk_lost, pk_lost_ed, pk_lost_x = _keys()
    _, pk_keep_ed, pk_keep_x = _keys()
    await roster.pin_identity("alice", "alice", pk_lost_ed, pk_lost_x, "code")
    await roster.pin_identity("alice", "alice", pk_keep_ed, pk_keep_x, "device")
    await roster.revoke_device("alice", pk_lost_ed)

    _, pk_new_ed, pk_new_x = _keys()
    session = await _session(tmp_path, roster)
    await _approve(session, sk_lost, pk_new_ed, pk_new_x)

    assert _last(session)["type"] == "error"
    assert await roster.find_device("alice", pk_new_ed) is None


async def test_an_account_with_no_device_here_cannot_file(tmp_path, roster):
    """The first device is admitted by an operator's invitation code. Letting
    this path serve that purpose would bypass the roster entirely."""
    sk_new, pk_new_ed, pk_new_x = _keys()
    session = await _session(tmp_path, roster, user_id="nobody")

    await _file_request(session, sk_new, pk_new_ed, pk_new_x, generate_code())

    assert _last(session)["type"] == "error"
    assert "invitation code" in _last(session)["detail"]


async def test_a_signature_by_the_wrong_key_is_not_a_request(tmp_path, roster):
    """Proof of possession: the request must be signed by the keys it presents."""
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_other, _, _ = _keys()
    _, pk_new_ed, pk_new_x = _keys()

    session = await _session(tmp_path, roster)
    await _file_request(session, sk_other, pk_new_ed, pk_new_x, generate_code())

    assert _last(session)["type"] == "error"
    assert "signature" in _last(session)["detail"].lower()


# ── The code binds the keys ──────────────────────────────────────────────────

async def test_the_node_cannot_substitute_the_keys(tmp_path, roster):
    """
    The load-bearing property. The hash covers the code **and** the requesting
    keys, so an approver looking a request up with different keys finds nothing
    — and never signs. This is what replaces "compare these digits".
    """
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_new, pk_new_ed, pk_new_x = _keys()
    _, pk_evil_ed, pk_evil_x = _keys()

    session = await _session(tmp_path, roster)
    code = generate_code()
    await _file_request(session, sk_new, pk_new_ed, pk_new_x, code)

    # A node that answered with keys of its own choosing would have to produce a
    # hash matching sha256(code ‖ those keys) — over a code it never receives.
    forged = device_code_hash(normalize_code(code), pk_evil_ed, pk_evil_x)
    real = (await _match_by_code(session, code))["code_hash"]

    assert forged != real, "substituted keys produced a matching hash"
    # And the client's own matching would reject the substitution outright.
    await session._do_device_lookup({})
    offered = _last(session)["requests"]
    assert all(r["pk_ed25519"] != pk_evil_ed for r in offered)


async def test_a_wrong_code_finds_nothing(tmp_path, roster):
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_new, pk_new_ed, pk_new_x = _keys()

    session = await _session(tmp_path, roster)
    await _file_request(session, sk_new, pk_new_ed, pk_new_x, generate_code())

    assert await _match_by_code(session, generate_code()) is None


async def test_a_code_is_spent_once(tmp_path, roster):
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    sk_new, pk_new_ed, pk_new_x = _keys()

    sk_old_signer, pk_old_ed2, pk_old_x2 = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed2, pk_old_x2, "device")
    session = await _session(tmp_path, roster)
    code = generate_code()
    await _file_request(session, sk_new, pk_new_ed, pk_new_x, code)

    match = await _match_by_code(session, code)
    await _approve(session, sk_old_signer, pk_new_ed, pk_new_x,
                   code_hash=match["code_hash"])
    assert _last(session)["type"] == MNP.DEVICE_ADD_ACK

    # Spent: the same approval cannot be replayed.
    await _approve(session, sk_old_signer, pk_new_ed, pk_new_x,
                   code_hash=match["code_hash"])
    assert _last(session)["type"] == "error"


async def test_another_account_cannot_redeem_your_code(tmp_path, roster):
    """Scoped to the account as well as to the keys."""
    _, pk_a_ed, pk_a_x = _keys()
    await roster.pin_identity("alice", "alice", pk_a_ed, pk_a_x, "code")
    _, pk_b_ed, pk_b_x = _keys()
    await roster.pin_identity("bob", "bob", pk_b_ed, pk_b_x, "code")
    sk_new, pk_new_ed, pk_new_x = _keys()

    alice = await _session(tmp_path, roster, user_id="alice")
    code = generate_code()
    await _file_request(alice, sk_new, pk_new_ed, pk_new_x, code)

    bob = await _session(tmp_path, roster, user_id="bob")

    # Scoped to the account: Bob is not offered Alice's pending request at all,
    # so the code buys him nothing even if he has it.
    assert await _match_by_code(bob, code) is None


async def test_guessing_is_bounded_on_a_connection(tmp_path, roster):
    _, pk_old_ed, pk_old_x = _keys()
    await roster.pin_identity("alice", "alice", pk_old_ed, pk_old_x, "code")
    session = await _session(tmp_path, roster)

    sk_new, pk_new_ed, pk_new_x = _keys()
    for _ in range(8):
        await _file_request(session, sk_new, pk_new_ed, pk_new_x, generate_code())

    assert "Too many device attempts" in _last(session)["detail"]


# ── Limits and revocation ────────────────────────────────────────────────────

async def test_the_device_ceiling_holds(tmp_path, roster):
    """
    A chain of devices inherits the weakness of its weakest ancestor, so the
    answer to "how many" is a ceiling and visibility, not cryptography.
    """
    sk_first, pk_first_ed, pk_first_x = _keys()
    await roster.pin_identity("alice", "alice", pk_first_ed, pk_first_x, "code")
    for _ in range(roster.MAX_DEVICES_PER_USER - 1):
        _, pk_ed, pk_x = _keys()
        await roster.pin_identity("alice", "alice", pk_ed, pk_x, "device")

    session = await _session(tmp_path, roster)
    sk_new, pk_new_ed, pk_new_x = _keys()
    await _file_request(session, sk_new, pk_new_ed, pk_new_x, generate_code())

    assert _last(session)["type"] == "error"
    assert "limit" in _last(session)["detail"]


async def test_your_last_device_cannot_be_revoked(tmp_path, roster):
    """Removing it would need an operator's code to come back, and doing that
    to yourself by accident is not a mistake worth allowing."""
    sk_only, pk_only_ed, pk_only_x = _keys()
    await roster.pin_identity("alice", "alice", pk_only_ed, pk_only_x, "code")
    session = await _session(tmp_path, roster)

    ts = int(time.time())
    transcript = device_add_transcript(
        node_pk_b64=session._node_pk_b64(), user_id="alice",
        pk_ed25519_b64=pk_only_ed, pk_x25519_b64=pk_only_x,
        nonce_node=NONCE, ts=ts)
    await session._do_device_revoke({
        "pk_ed25519": pk_only_ed, "ts": ts,
        "sig": base64.b64encode(sk_only.sign(transcript)).decode()})

    assert _last(session)["type"] == "error"
    assert await roster.find_device("alice", pk_only_ed) is not None


async def test_unpinning_an_account_takes_every_device(tmp_path, roster):
    """`member unpin` is what an operator runs when someone must start over.
    Leaving one device would let them walk back in with a forgotten key."""
    for _ in range(3):
        _, pk_ed, pk_x = _keys()
        await roster.pin_identity("alice", "alice", pk_ed, pk_x, "device")

    assert len(await roster.list_devices("alice")) == 3
    await roster.unpin("alice")
    assert await roster.list_devices("alice") == []