diff options
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli/client/session.py')
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/client/session.py | 443 |
1 files changed, 443 insertions, 0 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/client/session.py b/packages/meshbay-cli/src/meshbay_cli/client/session.py new file mode 100644 index 0000000..1ea398a --- /dev/null +++ b/packages/meshbay-cli/src/meshbay_cli/client/session.py @@ -0,0 +1,443 @@ +""" +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) |