""" 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 dataclasses import dataclass 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 . import sctp_fix, udp_buffer # noqa: F401 (patch aiortc and aioice on import) from .account import Account from .keys import ( LEGACY_MAGIC, BundleError, Identity, b64, bundle_is_current, open_bundle, open_recovery, opens_under, 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 NothingHere(CliError): """A node that holds no keys of this account: nothing to carry across.""" def __init__(self): super().__init__(NODE, "this account has no keys on this node") @dataclass class Rewrap: """ A connection made only to carry this account's copy on the node across a passphrase change or a reset: it never makes new keys, never asks for a device approval, and reads the recovery copy with `recovery` (the recovery key) where the passphrase copy does not open. """ recovery: bytes | None = None 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 # What the node keeps of this account's keys, fetched by a Rewrap # connection: `bundle_enc` and `bundle_enc_recovery`, or empty. self.held: dict = {} 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 def open_delta(self, msg: dict) -> dict: """A pushed `index_delta`, opened under the group key.""" try: return unseal(self.gek, PURPOSE_INDEX, "index_delta", self.group_id, msg) except Exception as e: raise CliError(NODE, "An index update does not open under the group key.") 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, *, rewrap: Rewrap | 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, rewrap) 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, rewrap: Rewrap | None = 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 held: dict = {} if rewrap is not None: held = await _fetch_held(ch) identity, source = await _rewrap_identity(account, node_pk, held, rewrap) bundle = held.get("bundle_enc") or "" else: identity, source, bundle = await _identity(account, ch, node_pk) if bundle is None and (account.browser_access or account.previous_masters): # 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" and rewrap is None: 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) if rewrap is None: await _settle_bundle(account, ch, identity, node_pk, bundle) session = GroupSession(pc, ch, group_id, node_id, node_pk, gek, config, identity, nonce_s, str(joined.get("role") or "")) session.held = held return session async def _fetch_held(ch: Channel) -> dict: """The node's copies of this account's keys: `bundle_enc`, sealed under `M`, and `bundle_enc_recovery`, under the recovery key — {} when it keeps none.""" found = await ch.request({"type": "keypair_bundle_fetch", "v": MNP_VERSION}, "keypair_bundle_resp") if not found.get("found"): return {} return {k: str(found[k]) for k in ("bundle_enc", "bundle_enc_recovery") if found.get(k)} async def _fetch_bundle(ch: Channel) -> str: """The node's copy of this account's keys, or "" when it keeps none.""" return (await _fetch_held(ch)).get("bundle_enc", "") def open_held(account: Account, held: dict, node_pk: str, recovery: bytes | None) -> dict: """The keys in the node's copy: under this session's `M` (or the legacy key), else under the recovery key. BundleError when none opens it.""" try: return open_bundle(held.get("bundle_enc", ""), master=account.master, legacy=account.legacy, user_id=account.user_id, node_pk=node_pk) except BundleError: if recovery is None or not held.get("bundle_enc_recovery"): raise return open_recovery(held["bundle_enc_recovery"], rkey=recovery, user_id=account.user_id, node_pk=node_pk) async def _rewrap_identity(account: Account, node_pk: str, held: dict, rewrap: Rewrap) -> tuple[Identity, str]: """The keys to join with on a Rewrap connection: kept here, or the node's copy opened — never new ones, and never a device request.""" kept = account.identity(node_pk) if kept is not None: return kept, "kept" if not held.get("bundle_enc") and not held.get("bundle_enc_recovery"): raise NothingHere() try: identity = Identity.from_pkcs8(open_held(account, held, node_pk, rewrap.recovery)) except BundleError as e: raise CliError(NODE, str(e)) from e await account.keep_identity(node_pk, identity) return identity, "bundle" 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 to leave on this node of the account's keys. A copy sealed under a passphrase this machine changed away from is sealed again, same keys, under the current `M` — the node was out of reach when the passphrase changed (docs/MESHBAY_DESIGN.md §3.6). Then what `browser-access` says (spec §2.12): `on`, a copy of this machine's keys where the node holds none, or only the older `MBK2`; `off`, the node's copy withdrawn; unset, nothing. A copy in `MBK3` that opens under no key this machine knows is left as it is: sealed under a passphrase set since, elsewhere, it is the one the browsers open — rewriting it under the old one would lock them out. A failure costs the browsers their copy, never this session. """ setting = account.browser_access if bundle is None or not account.master: return try: if bundle and setting != "off": earlier = next((keys for m in account.previous_masters if (keys := opens_under(bundle, m, user_id=account.user_id, node_pk=node_pk)) is not None), None) if earlier is not None: await _store(account, ch, earlier, node_pk) return if setting == "off" and bundle: await ch.request({"type": "keypair_bundle_delete", "v": MNP_VERSION}, "ack") elif setting == "on" and (not bundle or unb64(bundle)[:4] == LEGACY_MAGIC): await _store(account, ch, identity.to_pkcs8(), node_pk) elif setting == "on" and not bundle_is_current( bundle, master=account.master, user_id=account.user_id, node_pk=node_pk): log.warning("The node's copy of your keys is sealed under a passphrase this " "machine does not know — changed elsewhere? It is left as it is; " "run `meshbay-cli login` to sign in with the current one.") 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) async def _store(account: Account, ch: Channel, keys: dict, node_pk: str) -> None: sealed = seal_bundle(keys, 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") # 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)