path: root/packages/meshbay-cli/src/meshbay_cli/client/session.py

"""
A member's connection to one node, for one group.

The hub relays the WebRTC offer and the answer and nothing after; everything
from the handshake on goes straight to the node, sealed under the group key
(docs/MESHBAY_DESIGN.md §5.1, §5.2). The steps, in order:

1. a token for this node and this group only (`/v1/nodes/mnp-token`, E10);
2. the DataChannel, and the channel binding from both DTLS fingerprints;
3. the handshake: the node signs its challenge with the key it announces, and
   that key must be the one pinned for this node, if any (§5.2, TOFU);
4. this account's identity on the node — kept from an earlier run, or opened
   from the bundle the node keeps for browsers;
5. the group key, wrapped by the node for that identity (`join_request`);
6. the mutual proof, the ack verified and unsealed — a payload that does not
   open ends the session, never reads as an empty group (§4.4);
7. `device_hello`, so the node knows which of the account's devices this is.

Every request carries a `req_id` and its reply is matched on it (§5.3): matching
by arrival order is a guess that fails silently, and against the unrelated
request.
"""

import asyncio
import contextlib
import itertools
import logging
import os
import struct
import time
from datetime import UTC, datetime

import msgpack
from aiortc import RTCConfiguration, RTCIceServer, RTCPeerConnection, RTCSessionDescription
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import (
    Ed25519PrivateKey,
    Ed25519PublicKey,
)
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
from meshbay_common import MNP_VERSION
from meshbay_common.crypto import unwrap_gek_aes
from meshbay_common.device import (
    device_code_hash,
    device_hello_transcript,
    device_request_transcript,
)
from meshbay_common.groupbox import PURPOSE_ACK, PURPOSE_INDEX, unseal
from meshbay_common.handshake import (
    MNP_MIN_SUPPORTED,
    HandshakeError,
    challenge_transcript,
    check_version,
    handshake_transcript,
    make_proof,
    verify_proof,
    webrtc_binding,
)
from meshbay_common.join import join_transcript

from ..errors import NODE, NODE_KEY, VERSION, CliError
from .account import Account
from .keys import BundleError, Identity, b64, open_bundle, unb64

log = logging.getLogger(__name__)

# The browser's list (transport.js), two providers deep: one slow server must
# not add its whole timeout to every connection.
STUN_SERVERS = [
    "stun:stun.l.google.com:19302",
    "stun:stun1.l.google.com:19302",
    "stun:stun.cloudflare.com:3478",
]

OPEN_TIMEOUT = 30      # ICE + DTLS + SCTP, measured at 0.6–7 s
REPLY_TIMEOUT = 60


class NodeRefusal(CliError):
    def __init__(self, detail: str, code: str = ""):
        super().__init__(NODE, f"The node refused: {detail}")
        self.detail = detail
        self.refusal_code = code


class NodeKeyChanged(CliError):
    def __init__(self, node_id: str, pinned: str, now: str):
        super().__init__(
            NODE_KEY,
            f"The node {node_id} presents a different key than the one pinned at "
            f"first contact.\n  pinned: {pinned}\n  now:    {now}\n"
            "If its operator announced a new key, run `meshbay-cli trust reset "
            f"{node_id}`; otherwise do not connect.")


class Channel:
    """MNP frames over a DataChannel: a 4-byte big-endian length, then msgpack."""

    def __init__(self, dc):
        self.dc = dc
        self._buf = bytearray()
        self._ids = itertools.count(1)
        self._pending: dict[str, asyncio.Future] = {}
        self._push_handlers: dict[str, list] = {}
        self.closed = asyncio.Event()
        dc.on("message", self._on_message)
        dc.on("close", self._on_close)

    def on_push(self, msg_type: str, handler) -> None:
        """Call `handler(msg)` for every unsolicited message of this type."""
        self._push_handlers.setdefault(msg_type, []).append(handler)

    def off_push(self, msg_type: str, handler) -> None:
        with contextlib.suppress(KeyError, ValueError):
            self._push_handlers[msg_type].remove(handler)

    def _on_close(self) -> None:
        self.closed.set()
        for fut in self._pending.values():
            if not fut.done():
                fut.set_exception(CliError(NODE, "The connection to the node closed."))

    def _on_message(self, data) -> None:
        self._buf.extend(data if isinstance(data, bytes) else data.encode())
        while len(self._buf) >= 4:
            size = struct.unpack(">I", self._buf[:4])[0]
            if len(self._buf) < 4 + size:
                break
            msg = msgpack.unpackb(bytes(self._buf[4:4 + size]), raw=False)
            del self._buf[:4 + size]
            self._route(msg)

    def _route(self, msg: dict) -> None:
        # Type and id only: payloads carry sealed content and key material.
        log.debug("← %s req_id=%s", msg.get("type"), msg.get("req_id"))
        fut = self._pending.pop(str(msg.get("req_id", "")), None)
        if fut is not None and not fut.done():
            fut.set_result(msg)
            return
        handlers = self._push_handlers.get(str(msg.get("type")), [])
        if not handlers and msg.get("type") == "error":
            # A refusal that names no request answers none of ours: logged,
            # never handed to whichever request happens to be waiting.
            log.debug("unattributed error from the node: %s", msg.get("detail"))
        for handler in list(handlers):
            try:
                handler(msg)
            except Exception:
                log.exception("push handler for %s failed", msg.get("type"))

    def send(self, msg: dict) -> None:
        log.debug("→ %s req_id=%s", msg.get("type"), msg.get("req_id"))
        data = msgpack.packb(msg, use_bin_type=True)
        self.dc.send(struct.pack(">I", len(data)) + data)

    async def request(self, msg: dict, *expect: str, timeout: float = REPLY_TIMEOUT) -> dict:
        req_id = str(next(self._ids))
        fut = asyncio.get_running_loop().create_future()
        self._pending[req_id] = fut
        self.send({**msg, "req_id": req_id})
        try:
            reply = await asyncio.wait_for(fut, timeout)
        except TimeoutError:
            raise CliError(NODE, f"The node did not answer {msg.get('type')} "
                                 f"within {timeout:.0f} s.") from None
        finally:
            self._pending.pop(req_id, None)
        if reply.get("type") == "error":
            raise NodeRefusal(str(reply.get("detail") or "no reason given"),
                              str(reply.get("code") or ""))
        if expect and reply.get("type") not in expect:
            raise CliError(NODE, f"The node answered {msg.get('type')} with "
                                 f"{reply.get('type')}.")
        return reply


def _dtls_fingerprint(sdp: str) -> bytes:
    for line in sdp.splitlines():
        if line.startswith("a=fingerprint:sha-256 "):
            return bytes.fromhex(line.split(" ", 1)[1].strip().replace(":", ""))
    raise CliError(NODE, "No DTLS fingerprint in the session description.")


class GroupSession:
    """An authenticated connection to one node, for one group."""

    def __init__(self, pc: RTCPeerConnection, ch: Channel, group_id: str,
                 node_id: str, node_pk: str, gek: bytes, ack: dict):
        self.pc = pc
        self.ch = ch
        self.group_id = group_id
        self.node_id = node_id
        self.node_pk = node_pk
        self.gek = gek
        self.ack = ack

    async def index(self) -> dict:
        """`{version, entries, dirs, roots}`, opened under the group key."""
        msg = await self.ch.request({"type": "index_sync", "v": MNP_VERSION}, "index_sync")
        try:
            return unseal(self.gek, PURPOSE_INDEX, "index_sync", self.group_id, msg)
        except Exception as e:
            raise CliError(NODE, "The group's index does not open under the group key; "
                                 "the session is ended.") from e

    async def close(self) -> None:
        with contextlib.suppress(Exception):
            await self.pc.close()


async def connect(account: Account, group_id: str, node: dict) -> GroupSession:
    """Open a session with `node` (a `/v1/groups/{id}/nodes` row) for `group_id`."""
    node_id, listed_pk = node["node_id"], node["pk_node"]
    pinned = account.pin(node_id)
    if pinned and pinned != listed_pk:
        raise NodeKeyChanged(node_id, pinned, listed_pk)

    mnp_token = await account.call(account.hub.mnp_token, listed_pk, group_id)

    pc = RTCPeerConnection(RTCConfiguration(
        iceServers=[RTCIceServer(urls=u) for u in STUN_SERVERS]))
    try:
        dc = pc.createDataChannel("mnp", ordered=True)
        ch = Channel(dc)
        opened = asyncio.Event()
        dc.on("open", opened.set)
        await pc.setLocalDescription(await pc.createOffer())
        offer_sdp = pc.localDescription.sdp
        answer_sdp = await account.call(account.hub.offer, node_id, offer_sdp)
        await pc.setRemoteDescription(RTCSessionDescription(sdp=answer_sdp, type="answer"))
        try:
            await asyncio.wait_for(opened.wait(), OPEN_TIMEOUT)
        except TimeoutError:
            raise CliError(NODE, f"No connection to the node within {OPEN_TIMEOUT} s "
                                 "(it is online at the hub, but not reachable from "
                                 "here).") from None
        binding = webrtc_binding(_dtls_fingerprint(offer_sdp), _dtls_fingerprint(answer_sdp))
        return await _handshake(account, pc, ch, group_id, node_id, listed_pk,
                                mnp_token, binding)
    except BaseException:
        with contextlib.suppress(Exception):
            await pc.close()
        raise


async def _handshake(account: Account, pc, ch: Channel, group_id: str, node_id: str,
                     listed_pk: str, mnp_token: str, binding: bytes) -> GroupSession:
    nonce_c = os.urandom(32)
    challenge = await ch.request(
        {"type": "handshake", "v": MNP_VERSION, "v_min": MNP_MIN_SUPPORTED,
         "token": mnp_token, "group_id": group_id, "nonce": b64(nonce_c)},
        "handshake_challenge")
    try:
        check_version(challenge["v"], challenge.get("v_min", ""))
    except HandshakeError as e:
        raise CliError(VERSION, f"Protocol mismatch with the node: {e}") from e
    nonce_s = unb64(challenge["nonce"])
    node_pk = challenge["node_pk"]
    if node_pk != listed_pk:
        raise CliError(NODE, "The node answering is not the one the hub listed.")
    try:
        Ed25519PublicKey.from_public_bytes(unb64(node_pk)).verify(
            unb64(challenge.get("sig", "")),
            challenge_transcript(group_id, nonce_c, nonce_s, binding))
    except (InvalidSignature, ValueError) as e:
        raise CliError(NODE, "The node's challenge is not signed by the key it "
                             "announces.") from e

    identity, source = await _identity(account, ch, node_pk)

    ts = int(time.time())
    joined = await ch.request({
        "type": "join_request", "v": MNP_VERSION, "group_id": group_id,
        "pk_ed25519": identity.pk_ed, "pk_x25519": identity.pk_x, "code": "", "ts": ts,
        "sig": b64(identity.sk_ed.sign(join_transcript(
            node_pk_b64=node_pk, group_id=group_id, user_id=account.user_id,
            pk_ed25519_b64=identity.pk_ed, pk_x25519_b64=identity.pk_x,
            nonce_node=nonce_s, ts=ts)))}, "join_result")
    if not joined.get("ok") or not joined.get("gek"):
        reason = str(joined.get("reason") or "")
        if reason == "unknown_device" and source != "bundle":
            await _link(account, ch, node_pk, nonce_s, identity, source)
        raise NodeRefusal(JOIN_REFUSALS.get(reason, reason or "the group key was not given"),
                          reason)
    if source == "pending":
        await account.adopt_pending(node_pk)
    gek = unwrap_gek_aes(joined, identity.sk_x.private_bytes_raw(),
                         identity.sk_x.public_key().public_bytes_raw())

    ack = await ch.request({"type": "handshake_response", "v": MNP_VERSION,
                            "proof": b64(make_proof(gek, "client", group_id, nonce_c,
                                                    nonce_s, binding))},
                           "handshake_ack")
    if not verify_proof(gek, unb64(ack["proof"]), "node", group_id, nonce_c, nonce_s,
                        binding):
        raise CliError(NODE, "The node does not hold the group key.")
    if ack.get("node_pk") != node_pk:
        raise CliError(NODE, "The node changed identity during the handshake.")
    try:
        Ed25519PublicKey.from_public_bytes(unb64(node_pk)).verify(
            unb64(ack["sig"]), handshake_transcript("node", group_id, nonce_c, nonce_s,
                                                    binding))
    except (InvalidSignature, ValueError, KeyError) as e:
        raise CliError(NODE, "The node's acknowledgement is not signed by its key.") from e
    try:
        config = unseal(gek, PURPOSE_ACK, "handshake_ack", group_id, ack)
    except Exception as e:
        raise CliError(NODE, "The node's configuration does not open under the group "
                             "key; the session is ended.") from e

    if account.pin(node_id) is None:
        await account.set_pin(node_id, node_pk)

    await _device_hello(account, ch, identity, node_pk, group_id, nonce_s)
    return GroupSession(pc, ch, group_id, node_id, node_pk, gek, config)


async def _identity(account: Account, ch: Channel, node_pk: str) -> tuple[Identity, str]:
    """
    This account's keys on this node, and where they came from: kept from an
    earlier run, waiting for a device approval, opened from the copy the node
    keeps for browsers — or, failing all three, new, for device linking.
    """
    kept = account.identity(node_pk)
    if kept is not None:
        return kept, "kept"
    pending = account.pending_identity(node_pk)
    if pending is not None:
        return pending.identity, "pending"
    found = await ch.request({"type": "keypair_bundle_fetch", "v": MNP_VERSION},
                             "keypair_bundle_resp")
    if not found.get("found"):
        return Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()), "new"
    try:
        keys = open_bundle(found["bundle_enc"], master=account.master,
                           legacy=account.legacy, user_id=account.user_id, node_pk=node_pk)
        identity = Identity.from_pkcs8(keys)
    except BundleError as e:
        raise CliError(NODE, str(e)) from e
    await account.keep_identity(node_pk, identity)
    return identity, "bundle"


# What a refused join means, for the person reading it.
JOIN_REFUSALS = {
    "code_required": "this node has not admitted you yet. Accept the invitation in the "
                     "MeshBay application and enter its code there; the command line "
                     "does not join groups",
    "unknown_device": "the key this node holds for you is not one of your devices there "
                      "any more; ask its operator, or approve this machine from a device "
                      "already linked",
    "key_changed": "your identity on this node no longer matches the one it pinned",
    "not_authorized_for_group": "this node does not count you a member of the group",
}

# The alphabet of every MeshBay code (Crockford base32, no I L O U), so the one
# this machine shows is typed like an invitation code, and needs no normalising.
CODE_ALPHABET = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"


class DeviceLinkPending(CliError):
    """This machine asked to be a device of the account on a node; it waits for an
    approval from one already linked there."""

    def __init__(self, link_code: str, expires_at: float):
        super().__init__(NODE, f"Waiting for this machine to be approved, code {link_code}.")
        # Not `code`: on a CliError that is the exit status.
        self.link_code = link_code
        self.expires_at = expires_at


async def _link(account: Account, ch: Channel, node_pk: str, nonce_s: bytes,
                identity: Identity, source: str) -> None:
    """
    File a device request bound to a code this machine shows (design §3.3), or
    show again the one already filed. Always raises `DeviceLinkPending`.

    The code is hashed together with this machine's keys, so the device that
    approves cannot be handed a substituted key; the code itself never reaches
    the node.
    """
    if not account.remembered:
        raise CliError(NODE, "This machine is not yet a device of yours on this node. "
                             "Run `meshbay-cli login` first: linking it only makes sense "
                             "on a machine that keeps the identity it links.")
    if source == "pending":
        pending = account.pending_identity(node_pk)
        raise DeviceLinkPending(pending.code, pending.expires_at)
    if source == "kept":
        # Revoked or unpinned on the node since this machine kept it.
        await account.forget_identity(node_pk)
        identity = Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate())
    raw = "".join(CODE_ALPHABET[b % 32] for b in os.urandom(8))
    code = f"{raw[:4]}-{raw[4:]}"
    code_hash = device_code_hash(raw, identity.pk_ed, identity.pk_x)
    ts = int(time.time())
    sig = identity.sk_ed.sign(device_request_transcript(
        node_pk_b64=node_pk, user_id=account.user_id, pk_ed25519_b64=identity.pk_ed,
        pk_x25519_b64=identity.pk_x, code_hash=code_hash, nonce_node=nonce_s, ts=ts))
    try:
        ack = await ch.request({"type": "device_add_request", "v": MNP_VERSION,
                                "pk_ed25519": identity.pk_ed, "pk_x25519": identity.pk_x,
                                "code_hash": code_hash, "ts": ts, "sig": b64(sig)},
                               "device_add_request_ack")
    except NodeRefusal as e:
        if e.detail != "Handshake required":
            raise
        # The node answers a device request only on a session that has proved
        # the group key, which a machine with no identity there cannot do.
        raise CliError(NODE, "This machine is not yet a device of yours on this node, and "
                             "the node does not accept a device request from a machine "
                             "that holds no key there. Turn on Browser access in the "
                             "MeshBay application's settings, then run the command "
                             "again.") from e
    expires = _parse_time(ack.get("expires_at")) or time.time() + 3600
    await account.keep_pending(node_pk, identity, code, expires)
    raise DeviceLinkPending(code, expires)


def _parse_time(text) -> float | None:
    try:
        moment = datetime.fromisoformat(str(text))
    except ValueError:
        return None
    if moment.tzinfo is None:
        moment = moment.replace(tzinfo=UTC)
    return moment.timestamp()


async def _device_hello(account: Account, ch: Channel, identity: Identity, node_pk: str,
                        group_id: str, nonce_s: bytes) -> None:
    ts = int(time.time())
    sig = identity.sk_ed.sign(device_hello_transcript(
        node_pk_b64=node_pk, group_id=group_id, user_id=account.user_id,
        pk_ed25519_b64=identity.pk_ed, nonce_node=nonce_s, ts=ts))
    try:
        await ch.request({"type": "device_hello", "v": MNP_VERSION,
                          "pk_ed25519": identity.pk_ed, "ts": ts, "sig": b64(sig)},
                         "device_hello_ack")
    except NodeRefusal as e:
        # Listing does not depend on it; what is lost is the attribution of an
        # upload to this device rather than to the account's oldest one.
        log.warning("The node did not accept this device's hello: %s", e.detail)