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, opened
   from the bundle the node keeps for browsers, or new (an invitation's code,
   or device linking);
5. the group key, wrapped by the node for that identity (`join_request`, with
   the invitation's code when there is one);
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;
8. what `browser-access` says to leave on the node: a sealed copy, or none.

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,
    bundle_is_current,
    open_bundle,
    seal_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,
                 identity: Identity | None = None, nonce_node: bytes = b"",
                 role: str = ""):
        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
        # What a device message is signed with: this machine's key on the node,
        # and the nonce that ties a signature to this connection.
        self.identity = identity
        self.nonce_node = nonce_node
        # What the node's roster says this key is here: "operator" when it was
        # paired with the node — the authority to delete anyone's file. Not the
        # hub's word on who owns the group (docs/MESHBAY_DESIGN.md §3.3).
        self.role = role

    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,
                  join_code: str | None = None) -> GroupSession:
    """
    Open a session with `node` (a `/v1/groups/{id}/nodes` row) for `group_id`.
    `join_code`: an invitation's code, for a node that has not admitted this
    account yet; otherwise one kept for this group and node is sent.
    """
    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, join_code)
    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,
                     join_code: str | None) -> 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, bundle = await _identity(account, ch, node_pk)
    if bundle is None and account.browser_access:
        # Whether the node holds a copy, and a current one, decides what the
        # setting does after the proof. Asked here: the fetch is a pre-proof
        # message, and the window allows it.
        bundle = await _fetch_bundle(ch)

    code = join_code or account.join_code(group_id, node_pk) or ""
    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": 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 == "code_invalid":
            # Spent, cancelled or mistyped: it will not work on another try.
            await account.drop_join_code(group_id)
        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)
    elif source == "new":
        # Admitted with a code, or by an open group: these keys are now pinned
        # there, and an identity not kept would strand the account on the node.
        await account.keep_identity(node_pk, identity)
    if code:
        await account.drop_join_code(group_id)
    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)
    await _settle_bundle(account, ch, identity, node_pk, bundle)
    return GroupSession(pc, ch, group_id, node_id, node_pk, gek, config,
                        identity, nonce_s, str(joined.get("role") or ""))


async def _fetch_bundle(ch: Channel) -> str:
    """The node's copy of this account's keys, or "" when it keeps none."""
    found = await ch.request({"type": "keypair_bundle_fetch", "v": MNP_VERSION},
                             "keypair_bundle_resp")
    return str(found.get("bundle_enc") or "") if found.get("found") else ""


async def _identity(account: Account, ch: Channel,
                    node_pk: str) -> tuple[Identity, str, str | None]:
    """
    This account's keys on this node, where they came from, and the node's bundle
    when it was fetched (None: not asked). 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 an invitation's code, or for device linking.
    """
    kept = account.identity(node_pk)
    if kept is not None:
        return kept, "kept", None
    pending = account.pending_identity(node_pk)
    if pending is not None:
        return pending.identity, "pending", None
    bundle = await _fetch_bundle(ch)
    if not bundle:
        return (Identity(Ed25519PrivateKey.generate(), X25519PrivateKey.generate()),
                "new", bundle)
    try:
        keys = open_bundle(bundle, 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", bundle


async def _settle_bundle(account: Account, ch: Channel, identity: Identity, node_pk: str,
                         bundle: str | None) -> None:
    """
    What `browser-access` says to leave on this node (spec §2.12): `on`, a copy
    sealed under the current `M` when the node holds none or an outdated one —
    a current copy is left alone, whichever device sealed it, so two clients do
    not take turns rewriting it; `off`, the node's copy withdrawn; unset,
    nothing. A failure costs the browsers their copy, never this session.
    """
    setting = account.browser_access
    if not setting or bundle is None:
        return
    try:
        if setting == "off" and bundle:
            await ch.request({"type": "keypair_bundle_delete", "v": MNP_VERSION}, "ack")
        elif setting == "on" and account.master and not bundle_is_current(
                bundle, master=account.master, user_id=account.user_id, node_pk=node_pk):
            sealed = seal_bundle(identity.to_pkcs8(), master=account.master,
                                 pepper_version=account.pepper_version,
                                 user_id=account.user_id, node_pk=node_pk)
            await ch.request({"type": "keypair_bundle_store", "v": MNP_VERSION,
                              "bundle_enc": sealed}, "ack")
    except CliError as e:
        log.warning("Browser access is %s here, but the node's copy of your keys was "
                    "not %s: %s", setting, "withdrawn" if setting == "off" else "stored",
                    e.message)


# What a refused join means, for the person reading it.
JOIN_REFUSALS = {
    "code_required": "this node has not admitted you yet; it needs the code of your "
                     "invitation: meshbay-cli join GROUP --code CODE",
    "code_invalid": "the invitation's code was refused: already used, cancelled, or "
                    "mistyped. Ask the group's owner for a new one",
    "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)