diff options
3 files changed, 865 insertions, 848 deletions
diff --git a/docs/MESHBAY_NODE_PROTOCOL.md b/docs/MESHBAY_NODE_PROTOCOL.md index 90b0b41..35366f0 100644 --- a/docs/MESHBAY_NODE_PROTOCOL.md +++ b/docs/MESHBAY_NODE_PROTOCOL.md @@ -2286,7 +2286,7 @@ LP(x) = uint32be(len(x)) || x every field, no exceptions | Pairing / device code | 40 bits, Crockford base32, single use | `roster.py`, `device.py` | | Code lifetimes (default, settable) | invitation 7 d, operator pairing 24 h, device request 1 h | `roster.py` | | `MAX_PRE_PROOF_FETCHES` | 4 per connection | `webrtc_server.py` | -| `MAX_JOIN_ATTEMPTS` | 5 per connection | ” | +| `MAX_JOIN_ATTEMPTS` | 5 per connection | `webrtc/admission.py` | | `MAX_JOIN_FAILURES_WINDOW` / `JOIN_FAILURE_WINDOW` | 20 / 600 s, node-wide | ” | | Device attempts | 5 per connection | ” | | `MAX_DEVICES_PER_USER` | 5 | `roster.py` | diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/admission.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admission.py new file mode 100644 index 0000000..6a714cb --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/admission.py @@ -0,0 +1,861 @@ +"""Who gets in: invitations and invitation links, the join request and its +pairing code, and device linking.""" + +import base64 +import logging +import re +import time + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey +from meshbay_common import MNP_VERSION +from meshbay_common.adminop import OP_INVITE_CANCEL, OP_INVITE_CREATE, OP_INVITE_LINK_CREATE +from meshbay_common.crypto import wrap_gek_aes +from meshbay_common.device import ( + DEVICE_TTL, + device_add_transcript, + device_hello_transcript, + device_request_transcript, +) +from meshbay_common.join import JOIN_TTL, ROLE_MEMBER, ROLE_OPERATOR, join_transcript +from meshbay_common.protocol import MNP + +from meshbay_node import ops +from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK +from meshbay_node.transport.webrtc.channel import _get_remote_ip + +log = logging.getLogger("meshbay_node.transport.webrtc_server") + + +# An invitation link's handle, as `roster.create_link_invite` mints it. +_INVITE_ID_RE = re.compile(r"[0-9a-f]{32}") + + +# Pairing codes carry 40 bits and are single-use, but a connection must not be +# allowed to sit there guessing. Failures are audited, so a grind is visible. +MAX_JOIN_ATTEMPTS = 5 +# Per-connection limits alone would not bind an attacker who can open connections +# at will — and the adversary who can mint tokens for any account is the hub. So +# failed pairings are also counted node-wide over a window. +MAX_JOIN_FAILURES_WINDOW = 20 +JOIN_FAILURE_WINDOW = 600 # seconds + + +class AdmissionMixin: + def _do_invite_create(self, msg: dict) -> None: + """ + Issue a one-time pairing code for someone the operator wants to admit. + + Replaces the old invite path, where the inviter fetched the invitee's + public key from the hub and wrapped the group key for whatever came back + (H3). The node now needs nothing but a name: it will wrap the key itself, + later, for a key the invitee proves they hold. + """ + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) + return + + invitee_id = msg.get("user_id", "") + group_id = msg.get("group_id") or self._group_id + if not invitee_id or not group_id: + self._send({"type": "error", "detail": "Missing user_id or group_id"}) + return + if group_id != self._group_id: + self._send({"type": "error", "detail": "Wrong group for this session"}) + return + + if not self._has_admin_authority(): + self._send({ + "type": "error", + "detail": "No operator paired — run `meshbay-node operator pair`", + }) + return + + self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, { + "group_id": group_id, + "user_id": invitee_id, + "username": str(msg.get("username", ""))[:64], + }) + + def _do_invite_link_create(self, msg: dict) -> None: + """ + Issue a code bound to no account, for an invitation link — into the + group this connection authenticated to, and no other: a link names its + group, so the operator signs for exactly that one (docs/MESHBAY_DESIGN.md §3.4). + """ + group_id = self._group_id or "" + if not group_id: + self._send({"type": "error", "detail": "No group on this connection"}) + return + if msg.get("group_id") and msg["group_id"] != group_id: + self._send({"type": "error", "detail": "Wrong group for this session"}) + return + if not self._has_admin_authority(): + self._send({ + "type": "error", + "detail": "No operator paired — run `meshbay-node operator pair`", + }) + return + self._issue_admin_challenge( + OP_INVITE_LINK_CREATE, f"link:{group_id}", {"group_id": group_id}) + + def _do_invite_cancel(self, msg: dict) -> None: + """Take back an unredeemed link of this group, by its handle.""" + group_id = self._group_id or "" + invite_id = str(msg.get("invite_id", "")) + if not group_id: + self._send({"type": "error", "detail": "No group on this connection"}) + return + if not _INVITE_ID_RE.fullmatch(invite_id): + self._send({"type": "error", "detail": "Not an invitation id"}) + return + if not self._has_admin_authority(): + self._send({"type": "error", "detail": "No authorized key for this"}) + return + self._issue_admin_challenge( + OP_INVITE_CANCEL, invite_id, {"group_id": group_id, "invite_id": invite_id}) + + def _join_refuse(self, reason: str, audit_detail: str = "") -> None: + self._join_attempts += 1 + # Node-wide window, shared across connections: reconnecting must not reset + # the budget. + now = time.time() + failures = [t for t in self._ctx.get("join_failures", []) + if now - t < JOIN_FAILURE_WINDOW] + failures.append(now) + self._ctx["join_failures"] = failures + self._audit_join("join_refused", audit_detail or reason) + self._send({ + "type": MNP.JOIN_RESULT, + "v": MNP_VERSION, + "ok": False, + "reason": reason, + }) + + def _audit_join(self, event: str, detail: str) -> None: + audit = self._ctx.get("audit_store") + if not audit: + return + self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) + self._spawn(audit.log_event( + user_id=self._user_id or getattr(self, "_pending_sub", "unknown"), + event=event, + ip=self._remote_ip, + username=self._username or getattr(self, "_pending_username", ""), + group_id=self._group_id or getattr(self, "_pending_group", "") or "", + detail=detail, + )) + + async def _do_join_request(self, msg: dict) -> None: + """ + Pin an identity, or recognise one already pinned. + + The client signs its own Ed25519 and X25519 keys together with the node's + nonce, so the identity key vouches for the encryption key — that is what + will make it safe for the node to wrap the GEK for a key that arrived over + the wire instead of one fetched from the hub's directory (H3). + + A first pairing needs a one-time code, which the hub never sees. Afterwards + the pin is the credential and a changed key is refused outright, the same + rule the client applies to `pk_node` (11.5.8). + """ + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) + return + + if self._join_attempts >= MAX_JOIN_ATTEMPTS: + self._send({"type": "error", "detail": "Too many attempts"}) + return + + now = time.time() + recent = [t for t in self._ctx.get("join_failures", []) + if now - t < JOIN_FAILURE_WINDOW] + if len(recent) >= MAX_JOIN_FAILURES_WINDOW: + self._audit_join("join_throttled", f"{len(recent)} failures in window") + self._send({"type": "error", "detail": "Pairing temporarily locked"}) + return + + user_id = self._user_id or getattr(self, "_pending_sub", "") + username = self._username or getattr(self, "_pending_username", "") + if not user_id: + self._send({"type": "error", "detail": "Handshake required"}) + return + + pk_ed_b64 = msg.get("pk_ed25519", "") + pk_x_b64 = msg.get("pk_x25519", "") + code = msg.get("code", "") + ts = msg.get("ts", 0) + + try: + pk_ed_raw = base64.b64decode(pk_ed_b64) + pk_x_raw = base64.b64decode(pk_x_b64) + if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32: + raise ValueError + pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw) + except Exception: + self._join_refuse("invalid_keys") + return + + if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL: + self._join_refuse("stale_request") + return + + # An empty group_id means operator pairing, which is node-wide. Anything + # else must be the group this connection authenticated to — a signature + # obtained for one group must not name another. + group_id = msg.get("group_id", "") or "" + session_group = self._group_id or getattr(self, "_pending_group", "") or "" + if group_id and group_id != session_group: + self._join_refuse("group_mismatch") + return + + transcript = join_transcript( + node_pk_b64=self._node_pk_b64(), + group_id=group_id, + user_id=user_id, + pk_ed25519_b64=pk_ed_b64, + pk_x25519_b64=pk_x_b64, + nonce_node=self._nonce_node, + ts=ts, + ) + try: + sig = base64.b64decode(msg.get("sig", "")) + except Exception: + self._join_refuse("invalid_signature_encoding") + return + if not self._verify_sig(pk_ed, transcript, sig): + self._join_refuse("signature_invalid") + return + + # One person may hold several devices here — a browser and a desktop + # client are two keys on one account. So the question is not "is this + # THE key" but "is this ONE OF this account's live devices". + device = await roster.find_device(user_id, pk_ed_b64) + if device and device["pk_x25519"] != pk_x_b64: + # The Ed25519 key is pinned but arrives with a different encryption + # key. The join transcript signs both together, so this is either a + # client that regenerated half its identity or something splicing + # two messages; either way the pair is not the one admitted. + self._join_refuse( + "key_changed", + f"pinned x25519={device['pk_x25519'][:16]} presented={pk_x_b64[:16]}") + return + known = device + if not known and await roster.list_devices(user_id): + # The account is known here but this key is not one of its devices. + # Not an error to shout about: it is a second browser or a new + # client, and the way in is a device-add approved by a device that + # is already trusted — no operator, no new invitation code. + self._join_refuse( + "unknown_device", + f"presented={pk_ed_b64[:16]} — approve it from a device already " + f"paired with this node") + return + if known: + # This group's own row first; then the join message's group_id (empty + # on the node-wide first connect); then the operator's node-wide row, + # which is where an operator opening any group finds their authority. + member = (await roster.get_member(session_group, user_id) + or await roster.get_member(group_id, user_id) + or await roster.get_member("", user_id)) + if not member and self._group_join_policy(session_group) == "open": + await roster.set_member( + group_id=session_group, user_id=user_id, role=ROLE_MEMBER, + status="active", approved_by="open-join", + ) + member = await roster.get_member(session_group, user_id) + + # A pending invite means the operator explicitly re-invited this + # person — require the code even if they already have a member + # row (e.g. they left and were re-invited, or were revoked then + # re-invited). Without this gate a stale roster row lets them + # back in without proving they received the new code. + pending_invite = any( + i["kind"] == KIND_ACCOUNT + and i["user_id"] == user_id + and i["group_id"] in (session_group, "") + for i in await roster.list_invites()) + # Or they bring a link for this group: somebody already pinned here + # through another group, which is the ordinary case for a link, or + # somebody removed from it and invited back. Only when they are not + # an active member — a member opening the group leaves the link for + # whoever it was meant for. + active = bool(member) and member.get("status") == "active" + if pending_invite or (code and not active): + if not code: + self._join_refuse("code_required") + return + invite = await roster.consume_invite(code, user_id, session_group) + if not invite: + self._join_refuse("code_invalid") + return + await roster.set_member( + group_id=invite["group_id"], user_id=user_id, + role=invite["role"], status="active", + approved_by=invite["created_by"], + ) + self._audit_join( + "join_pinned", + f"group={invite['group_id'][:8]} role={invite['role']} " + f"via={'link' if invite['kind'] == KIND_LINK else 'code'} " + "(device already known)") + member = (await roster.get_member(session_group, user_id) + or await roster.get_member(invite["group_id"], user_id)) + + if not member: + self._join_refuse("not_authorized_for_group") + return + await self._join_ok( + user_id, pk_x_raw, session_group, + role=member["role"] if member else "", + recognised=True, + ) + return + + if not code: + if self._group_join_policy(session_group) == "open": + # An open-join group admits anyone the hub calls a member, so a + # code would protect nothing — the hub can walk in through the + # front door. Pin what turns up and say so in the audit log. + await self._pin_and_admit( + roster, user_id, username, pk_ed_b64, pk_x_b64, + group_id=session_group, role=ROLE_MEMBER, + approved_by="open-join", via="tofu") + await self._join_ok(user_id, pk_x_raw, session_group, + role=ROLE_MEMBER, recognised=False) + return + self._join_refuse("code_required") + return + + invite = await roster.consume_invite(code, user_id, session_group) + if not invite: + self._join_refuse("code_invalid") + return + + await self._pin_and_admit( + # The name comes from the invitation, not from the token: the hub does + # not put a username claim in a JWT, so pinning from the session alone + # left the roster nameless and `member revoke <name>` unable to match. + roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64, + group_id=invite["group_id"], role=invite["role"], + approved_by=invite["created_by"], + via="link" if invite["kind"] == KIND_LINK else "code") + # The roster row comes from the invitation; the key comes from the + # connection. An operator pairing is node-wide (empty group), but they + # redeemed the code while opening a group and expect to read it — and + # is_authorized() already grants an operator every group on this node. + await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"], + role=invite["role"], recognised=False) + + async def _do_device_request(self, msg: dict) -> None: + """ + A new device files itself as pending, bound to a code it displays. + + Served in the pre-proof window: by construction the caller holds no key + this node knows, so there is nothing yet to prove. Filing is inert — + nothing is admitted until an existing device countersigns. + """ + roster = self._ctx.get("roster") + if roster is None or not self._user_id or not self._nonce_node: + self._send({"type": "error", "detail": "Not ready for a device request"}) + return + + if not self._spend_device_attempt(): + return + + pk_ed_b64 = str(msg.get("pk_ed25519", "")) + pk_x_b64 = str(msg.get("pk_x25519", "")) + code_hash = str(msg.get("code_hash", "")) + if not (pk_ed_b64 and pk_x_b64 and code_hash): + self._send({"type": "error", "detail": "Missing device keys or code"}) + return + + # The account must already be known here. Anti-spam rather than a + # security boundary: the filing key is unpinned by construction, so this + # bounds the table, not the trust. + existing = await roster.list_devices(self._user_id) + if not existing: + self._send({"type": "error", + "detail": "This account has no device on this node yet — " + "an invitation code is what admits the first"}) + return + if len(existing) >= roster.MAX_DEVICES_PER_USER: + self._send({"type": "error", + "detail": f"Already {len(existing)} devices, which is the " + f"limit. Revoke one first."}) + return + + ts = int(msg.get("ts", 0)) + if abs(time.time() - ts) > DEVICE_TTL: + self._send({"type": "error", "detail": "Device request expired"}) + return + + transcript = device_request_transcript( + node_pk_b64=self._node_pk_b64(), user_id=self._user_id, + pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, + code_hash=code_hash, nonce_node=self._nonce_node, ts=ts) + try: + pk_ed = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64)) + sig = base64.b64decode(msg.get("sig", "")) + except Exception: + self._send({"type": "error", "detail": "Invalid device key encoding"}) + return + if not self._verify_sig(pk_ed, transcript, sig): + # Proof of possession, and nothing more: this says the caller holds + # the keys, never that they belong to this account. + self._send({"type": "error", "detail": "Device signature invalid"}) + return + + ttl = int(self._ctx.get("device_request_ttl") or 3600) + expires = await roster.file_device_request( + user_id=self._user_id, username=self._username or "", + pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, + code_hash=code_hash, ttl=ttl) + self._audit("device_request", f"{pk_ed_b64[:16]}") + log.info("Device request filed for %s (%s)", self._user_id[:8], + pk_ed_b64[:16]) + self._send({"type": MNP.DEVICE_REQUEST_ACK, "v": MNP_VERSION, + "expires_at": expires}) + + async def _do_device_lookup(self, msg: dict) -> None: + """ + List this account's pending device requests, each with its code hash. + + **The node never learns the code**, which is what makes it unable to + substitute a key. It answers with candidates; the approver recomputes + `sha256(code ‖ keys)` for each and keeps the one that matches. A node + offering fabricated keys would have to produce a hash matching + `sha256(code ‖ fabricated)` — and it does not know the code. + + An earlier version of this took the hash from the client and looked the + request up by it. That is circular: the client cannot compute the hash + without already knowing the keys it is asking about. + """ + roster = self._ctx.get("roster") + if roster is None or not self._user_id: + self._send({"type": "error", "detail": "Roster not available"}) + return + + pending = await roster.list_device_requests(self._user_id) + self._send({ + "type": MNP.DEVICE_LOOKUP_RESULT, "v": MNP_VERSION, + "requests": [ + {"pk_ed25519": r["pk_ed25519"], "pk_x25519": r["pk_x25519"], + "code_hash": r["code_hash"], "created_at": r["created_at"]} + for r in pending + ], + }) + + async def _do_device_add(self, msg: dict) -> None: + """ + Admit a device, countersigned by one this node already pinned. + + The whole control is in `_verify_device_signer`: the signature must + verify against a **live device of this same account**. The hub holds no + user keys and so cannot produce one. + """ + roster = self._ctx.get("roster") + if roster is None or not self._user_id or not self._nonce_node: + self._send({"type": "error", "detail": "Not ready to add a device"}) + return + if not self._spend_device_attempt(): + return + + pk_ed_b64 = str(msg.get("pk_ed25519", "")) + pk_x_b64 = str(msg.get("pk_x25519", "")) + ts = int(msg.get("ts", 0)) + if not (pk_ed_b64 and pk_x_b64): + self._send({"type": "error", "detail": "Missing device keys"}) + return + if abs(time.time() - ts) > DEVICE_TTL: + self._send({"type": "error", "detail": "Approval expired"}) + return + + transcript = device_add_transcript( + node_pk_b64=self._node_pk_b64(), user_id=self._user_id, + pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, + nonce_node=self._nonce_node, ts=ts) + signer = await self._verify_device_signer(roster, transcript, + msg.get("sig", "")) + if signer is None: + self._audit("device_add_refused", pk_ed_b64[:16]) + self._send({"type": "error", + "detail": "Not signed by a device already paired here"}) + return + + devices = await roster.list_devices(self._user_id) + if len(devices) >= roster.MAX_DEVICES_PER_USER: + self._send({"type": "error", "detail": "Device limit reached"}) + return + + # Spend the request. Single use: an approval cannot be replayed, and a + # code that was used is gone whatever else happens next. + code_hash = str(msg.get("code_hash", "")) + if code_hash and not await roster.take_device_request( + code_hash, self._user_id): + self._send({"type": "error", + "detail": "That request is no longer pending"}) + return + + # The countersignature is **kept**, with the two fields needed to rebuild + # what it signed. Until 2026-09-07 it was verified here and thrown away, + # leaving only `added_by_pk` — which says *which* key approved and + # proves nothing to anyone else. `device_add_transcript` binds + # `nonce_node`, this connection's handshake nonce, so a stored signature + # without it is still unverifiable; that is why all three go in. + # + # This is what lets another member check for themselves that this device + # belongs to an account whose earlier device they have already pinned, + # instead of taking the node's word (Tier 2, docs/MESHBAY_DESIGN.md §3.3). + await roster.pin_identity( + user_id=self._user_id, username=self._username or "", + pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via="device", + label=str(msg.get("label", ""))[:64], added_by_pk=signer, + add_sig=str(msg.get("sig", "")), + add_nonce=base64.b64encode(self._nonce_node).decode(), + add_ts=ts) + self._audit("device_added", f"{pk_ed_b64[:16]} by {signer[:16]}") + log.info("Device added for %s: %s (approved by %s)", + self._user_id[:8], pk_ed_b64[:16], signer[:16]) + self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, + "pk_ed25519": pk_ed_b64}) + + async def _do_device_hello(self, msg: dict) -> None: + """ + Learn which of this account's devices is on this connection. + + The handshake authenticates a *group membership* (the GEK-HMAC) and an + *account* (the hub's token). It has never authenticated a device, and + while one account meant one key that was the same statement. It stopped + being so on 2026-08-18, and `_load_pinned_pk` — which resolves the + account's oldest live device — has been standing in for the real answer + ever since, including as the recorded uploader of every file. + + What is checked, in order: the key is a live device *of this account* in + the node's own roster (never a token claim — that is + `docs/MESHBAY_DESIGN.md` §3.2's rule), the timestamp is fresh, and the + signature verifies over a transcript naming this node, this group and + this connection's nonce. A key that is merely well-formed proves nothing. + + Idempotent for the same key, refused for a different one: a connection + does not get to change device half way through, which would let one + session's uploads be attributed to two. + """ + roster = self._ctx.get("roster") + if roster is None or not self._user_id: + self._send({"type": "error", "detail": "Roster not available"}) + return + if not self._spend_device_attempt(): + return + + pk_ed_b64 = str(msg.get("pk_ed25519", "")) + ts = int(msg.get("ts", 0) or 0) + if not pk_ed_b64: + self._send({"type": "error", "detail": "Missing device key"}) + return + if self._device_confirmed and pk_ed_b64 != self._pinned_pk: + self._send({"type": "error", + "detail": "This connection is already another device"}) + return + if abs(time.time() - ts) > DEVICE_TTL: + self._send({"type": "error", "detail": "Stale device_hello"}) + return + + device = await roster.find_device(self._user_id, pk_ed_b64) + if device is None: + self._audit("device_hello_refused", pk_ed_b64[:16]) + self._send({"type": "error", + "detail": "Not a device paired here"}) + return + + transcript = device_hello_transcript( + node_pk_b64=self._node_pk_b64(), group_id=self._group_id or "", + user_id=self._user_id, pk_ed25519_b64=pk_ed_b64, + nonce_node=self._nonce_node, ts=ts) + try: + pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64)) + except Exception: + self._send({"type": "error", "detail": "Unreadable device key"}) + return + try: + sig = base64.b64decode(msg.get("sig", "")) + except Exception: + sig = b"" + if not self._verify_sig(pk, transcript, sig): + self._audit("device_hello_refused", pk_ed_b64[:16]) + self._send({"type": "error", "detail": "Signature verification failed"}) + return + + self._pinned_pk = pk_ed_b64 + self._device_confirmed = True + log.info("Device identified on connection: user=%s device=%s", + self._user_id[:8], pk_ed_b64[:16]) + self._send({"type": MNP.DEVICE_HELLO_ACK, "v": MNP_VERSION, + "pk_ed25519": pk_ed_b64}) + + async def _do_device_list(self, msg: dict) -> None: + """This account's devices. Anyone may read their own, nobody else's.""" + roster = self._ctx.get("roster") + if roster is None or not self._user_id: + self._send({"type": "error", "detail": "Roster not available"}) + return + devices = await roster.list_devices(self._user_id) + pending = await roster.pending_device_requests(self._user_id) + self._send({ + "type": MNP.DEVICE_LIST_RESULT, "v": MNP_VERSION, + "pending": pending, + "devices": [ + {"pk_ed25519": d["pk_ed25519"], "label": d.get("label", ""), + "pinned_at": d["pinned_at"], "pinned_via": d["pinned_via"], + "added_by_pk": d.get("added_by_pk", ""), + "is_this_one": d["pk_ed25519"] == self._pinned_pk} + for d in devices + ], + }) + + async def _do_device_revoke(self, msg: dict) -> None: + """ + Retire one of this account's devices — a lost laptop. + + Countersigned like an addition, by a live device of the same account. + The last one cannot go: an account with no device on this node can only + return through an operator's invitation code, and doing that to yourself + by accident is not a mistake worth allowing. + """ + roster = self._ctx.get("roster") + if roster is None or not self._user_id or not self._nonce_node: + self._send({"type": "error", "detail": "Not ready"}) + return + if not self._spend_device_attempt(): + return + + target = str(msg.get("pk_ed25519", "")) + ts = int(msg.get("ts", 0)) + if not target: + self._send({"type": "error", "detail": "Missing device key"}) + return + if abs(time.time() - ts) > DEVICE_TTL: + self._send({"type": "error", "detail": "Request expired"}) + return + + victim = await roster.find_device(self._user_id, target) + if victim is None: + self._send({"type": "error", "detail": "No such device"}) + return + + transcript = device_add_transcript( + node_pk_b64=self._node_pk_b64(), user_id=self._user_id, + pk_ed25519_b64=target, pk_x25519_b64=victim["pk_x25519"], + nonce_node=self._nonce_node, ts=ts) + signer = await self._verify_device_signer(roster, transcript, + msg.get("sig", "")) + if signer is None: + self._send({"type": "error", + "detail": "Not signed by a device already paired here"}) + return + + if len(await roster.list_devices(self._user_id)) <= 1: + self._send({"type": "error", + "detail": "This is your only device here — removing it " + "would need an operator code to come back"}) + return + + await roster.revoke_device(self._user_id, target) + # A revoked device holds every chat key it ever received — a lost laptop + # reads the group's chat until the epoch moves. + await self._new_chat_epoch(self._group_id or "", "device_revoke") + self._audit("device_revoked", f"{target[:16]} by {signer[:16]}") + log.info("Device revoked for %s: %s", self._user_id[:8], target[:16]) + self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, + "revoked": target}) + + async def _verify_device_signer(self, roster, transcript: bytes, + sig_b64: str) -> str | None: + """ + The pinned key that signed this, or None. + + Every live device of the account is tried, because any of them may + approve. A revoked one is not in the list — that is the point of marking + rather than deleting: a lost laptop must stop being able to admit its + replacement. + """ + try: + sig = base64.b64decode(sig_b64) + except Exception: + return None + for device in await roster.list_devices(self._user_id): + try: + pk = Ed25519PublicKey.from_public_bytes( + base64.b64decode(device["pk_ed25519"])) + except Exception: + continue + if self._verify_sig(pk, transcript, sig): + return device["pk_ed25519"] + return None + + def _spend_device_attempt(self) -> bool: + """ + Bound guessing on this connection, as the join path does. + + A code is 40 bits, single use and bound to the keys it names, so this is + depth rather than the control — but an unbounded loop over the lookup is + still a free oracle, and a burst of failures belongs in the audit log. + """ + self._device_attempts = getattr(self, "_device_attempts", 0) + 1 + if self._device_attempts > 5: + self._audit("device_attempts_exceeded", str(self._device_attempts)) + self._send({"type": "error", + "detail": "Too many device attempts on this connection"}) + return False + return True + + def _group_join_policy(self, group_id: str) -> str: + """ + Admission policy for a group, read from the node's own configuration. + + Never from the hub: a hub that could declare a group open would be handed + the key to it (docs/MESHBAY_DESIGN.md §3.4). + """ + gctx = (self._ctx.get("groups") or {}).get(group_id) or {} + return gctx.get("join_policy", "invite") + + async def _pin_and_admit( + self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str, + *, group_id: str, role: str, approved_by: str, via: str, + ) -> None: + await roster.pin_identity( + user_id=user_id, username=username, + pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via, + ) + await roster.set_member( + group_id=group_id, user_id=user_id, role=role, + status="active", approved_by=approved_by, + ) + if role == ROLE_OPERATOR: + self._ctx["has_admin_authority"] = True + + log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role) + self._audit_join("join_pinned", f"role={role} via={via}") + + async def _join_ok( + self, user_id: str, pk_x_raw: bytes, group_id: str, + *, role: str, recognised: bool, + ) -> None: + """ + Answer a join, wrapping the group key for the key the caller just proved. + + This is the H3 fix. The inviter used to fetch the invitee's public key from + the hub and wrap the GEK for whatever came back, so a hub that answered + with its own key was handed the group key by an honest member following the + protocol exactly. The node now wraps for a key that arrived from its owner + over an authenticated channel, bound to a pinned identity. + """ + reply = { + "type": MNP.JOIN_RESULT, + "v": MNP_VERSION, + "ok": True, + "recognised": recognised, + "role": role, + } + + roster = self._ctx["roster"] + if group_id and not await roster.is_authorized(group_id, user_id): + # Pinned on this node, but not admitted to this group. Hub membership + # alone must not produce a key. + reply["gek"] = False + reply["reason"] = "not_authorized_for_group" + self._send(reply) + self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized") + return + + gctx = (self._ctx.get("groups") or {}).get(group_id) or {} + gek = gctx.get("gek") + if not gek: + reply["gek"] = False + reply["reason"] = "no_gek" + self._send(reply) + return + + bundle = wrap_gek_aes(gek, pk_x_raw) + reply["gek"] = True + reply["pk_eph_b64"] = bundle["pk_eph_b64"] + reply["nonce_b64"] = bundle["nonce_b64"] + reply["wrapped_b64"] = bundle["wrapped_b64"] + self._send(reply) + self._audit_join("gek_wrapped", f"group={group_id[:8]}") + + async def _admin_exec_invite_create( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + # Node operator only. A group admin who does not run the node has no + # authority over who this node admits (deny by default). Delegation is + # designed but deferred — see docs/MESHBAY_DESIGN.md §3.4. + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}") + return + + payload = pending["payload"] + try: + result = await self._run_op( + ops.create_invite, + payload["group_id"], + payload.get("username", ""), + user_id=payload["user_id"], + created_by=self._user_id or "", + ) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + + self._audit("invite_create", f"target={payload['user_id'][:8]}") + self._send({ + "type": MNP.INVITE_RESULT, + "v": MNP_VERSION, + "code": result["code"], + "expires_at": result["expires_at"], + "user_id": result["user_id"], + "username": result.get("username", ""), + }) + + async def _admin_exec_invite_link_create( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", "invite_link_create") + return + try: + result = await self._run_op( + ops.create_link_invite, pending["payload"]["group_id"], + created_by=self._user_id or "") + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("invite_link_create", f"invite={result['invite_id'][:8]}") + self._send({ + "type": MNP.INVITE_LINK_RESULT, + "v": MNP_VERSION, + "code": result["code"], + "invite_id": result["invite_id"], + "expires_at": result["expires_at"], + "group_id": result["group_id"], + }) + + async def _admin_exec_invite_cancel( + self, pending: dict, transcript: bytes, sig: bytes, + ) -> None: + if not await self._verify_admin_sig(transcript, sig): + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"invite_cancel:{pending['subject'][:8]}") + return + payload = pending["payload"] + try: + await self._run_op(ops.cancel_invite, payload["group_id"], payload["invite_id"]) + except ops.OpError as e: + self._send({"type": "error", "detail": e.message}) + return + self._audit("invite_cancel", f"invite={payload['invite_id'][:8]}") + self._send({"type": "ack", "v": MNP_VERSION, "detail": "invite_cancelled", + "invite_id": payload["invite_id"]}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 9264b60..73fefd8 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -26,7 +26,6 @@ import asyncio import base64 import logging import os -import re import time import uuid from typing import Any @@ -69,13 +68,7 @@ from meshbay_common.adminop import ( OP_TRANSFER_LIMITS, admin_transcript, ) -from meshbay_common.crypto import pk_to_b64, wrap_gek_aes -from meshbay_common.device import ( - DEVICE_TTL, - device_add_transcript, - device_hello_transcript, - device_request_transcript, -) +from meshbay_common.crypto import pk_to_b64 from meshbay_common.groupbox import ( PURPOSE_ACK, PURPOSE_ROSTER, @@ -95,12 +88,6 @@ from meshbay_common.handshake import ( verify_proof, webrtc_binding, ) -from meshbay_common.join import ( - JOIN_TTL, - ROLE_MEMBER, - ROLE_OPERATOR, - join_transcript, -) from meshbay_common.protocol import ( MNP, ) @@ -117,7 +104,7 @@ from meshbay_node.indexer.indexer import DirectoryIndexer from meshbay_node.roots import ( RootSet, ) -from meshbay_node.roster import KIND_ACCOUNT, KIND_LINK +from meshbay_node.transport.webrtc.admission import AdmissionMixin from meshbay_node.transport.webrtc.apps.music import MusicMixin from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin @@ -138,9 +125,6 @@ from meshbay_node.transport.webrtc.upload_handlers import UploadMixin log = logging.getLogger(__name__) -# An invitation link's handle, as `roster.create_link_invite` mints it. -_INVITE_ID_RE = re.compile(r"[0-9a-f]{32}") - # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch, # nowhere near enough to be a memory-exhaustion primitive (H6). @@ -160,21 +144,12 @@ UNAUTHENTICATED_SESSION_TIMEOUT = 60 # seconds # Bundle fetches are served in the pre-proof window (C4). Bounded and audited # until the native client removes remote keypair bundles entirely. MAX_PRE_PROOF_FETCHES = 4 -# Pairing codes carry 40 bits and are single-use, but a connection must not be -# allowed to sit there guessing. Failures are audited, so a grind is visible. -MAX_JOIN_ATTEMPTS = 5 -# Per-connection limits alone would not bind an attacker who can open connections -# at will — and the adversary who can mint tokens for any account is the hub. So -# failed pairings are also counted node-wide over a window. -MAX_JOIN_FAILURES_WINDOW = 20 -JOIN_FAILURE_WINDOW = 600 # seconds # Everything a member sends lands here: files from the Files panel and # attachments from the chat alike. One visible directory the operator can look # into, back up or empty — rather than a hidden tree of per-user uuids that # nobody could read, or files scattered wherever someone happened to be looking. - # Opt-in, off by default: a per-session heartbeat log (message count, time # since the last message, ICE state) and ICE-state-change logging, on top of # the connectionstatechange logging that already runs unconditionally. Added @@ -190,7 +165,7 @@ _WEBRTC_TRACE_INTERVAL_S = 30.0 class WebRTCPeerSession( - BlobsMixin, ChatMixin, FilesMixin, TransferMixin, UploadMixin, + AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, TransferMixin, UploadMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" @@ -856,80 +831,6 @@ class WebRTCPeerSession( if note_activity: note_activity() - def _do_invite_create(self, msg: dict) -> None: - """ - Issue a one-time pairing code for someone the operator wants to admit. - - Replaces the old invite path, where the inviter fetched the invitee's - public key from the hub and wrapped the group key for whatever came back - (H3). The node now needs nothing but a name: it will wrap the key itself, - later, for a key the invitee proves they hold. - """ - roster = self._ctx.get("roster") - if roster is None: - self._send({"type": "error", "detail": "Roster not available"}) - return - - invitee_id = msg.get("user_id", "") - group_id = msg.get("group_id") or self._group_id - if not invitee_id or not group_id: - self._send({"type": "error", "detail": "Missing user_id or group_id"}) - return - if group_id != self._group_id: - self._send({"type": "error", "detail": "Wrong group for this session"}) - return - - if not self._has_admin_authority(): - self._send({ - "type": "error", - "detail": "No operator paired — run `meshbay-node operator pair`", - }) - return - - self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, { - "group_id": group_id, - "user_id": invitee_id, - "username": str(msg.get("username", ""))[:64], - }) - - def _do_invite_link_create(self, msg: dict) -> None: - """ - Issue a code bound to no account, for an invitation link — into the - group this connection authenticated to, and no other: a link names its - group, so the operator signs for exactly that one (docs/MESHBAY_DESIGN.md §3.4). - """ - group_id = self._group_id or "" - if not group_id: - self._send({"type": "error", "detail": "No group on this connection"}) - return - if msg.get("group_id") and msg["group_id"] != group_id: - self._send({"type": "error", "detail": "Wrong group for this session"}) - return - if not self._has_admin_authority(): - self._send({ - "type": "error", - "detail": "No operator paired — run `meshbay-node operator pair`", - }) - return - self._issue_admin_challenge( - OP_INVITE_LINK_CREATE, f"link:{group_id}", {"group_id": group_id}) - - def _do_invite_cancel(self, msg: dict) -> None: - """Take back an unredeemed link of this group, by its handle.""" - group_id = self._group_id or "" - invite_id = str(msg.get("invite_id", "")) - if not group_id: - self._send({"type": "error", "detail": "No group on this connection"}) - return - if not _INVITE_ID_RE.fullmatch(invite_id): - self._send({"type": "error", "detail": "Not an invitation id"}) - return - if not self._has_admin_authority(): - self._send({"type": "error", "detail": "No authorized key for this"}) - return - self._issue_admin_challenge( - OP_INVITE_CANCEL, invite_id, {"group_id": group_id, "invite_id": invite_id}) - # ── Per-account blobs (playlists, docs/playlists.md §8.2) ──────────────── # # The node stores bytes it cannot read and hands them back. `self._user_id` @@ -939,239 +840,6 @@ class WebRTCPeerSession( # ── Pairing and join (H3, M3) ──────────────────────────────────────────── - def _join_refuse(self, reason: str, audit_detail: str = "") -> None: - self._join_attempts += 1 - # Node-wide window, shared across connections: reconnecting must not reset - # the budget. - now = time.time() - failures = [t for t in self._ctx.get("join_failures", []) - if now - t < JOIN_FAILURE_WINDOW] - failures.append(now) - self._ctx["join_failures"] = failures - self._audit_join("join_refused", audit_detail or reason) - self._send({ - "type": MNP.JOIN_RESULT, - "v": MNP_VERSION, - "ok": False, - "reason": reason, - }) - - def _audit_join(self, event: str, detail: str) -> None: - audit = self._ctx.get("audit_store") - if not audit: - return - self._remote_ip = self._remote_ip or _get_remote_ip(self._pc) - self._spawn(audit.log_event( - user_id=self._user_id or getattr(self, "_pending_sub", "unknown"), - event=event, - ip=self._remote_ip, - username=self._username or getattr(self, "_pending_username", ""), - group_id=self._group_id or getattr(self, "_pending_group", "") or "", - detail=detail, - )) - - async def _do_join_request(self, msg: dict) -> None: - """ - Pin an identity, or recognise one already pinned. - - The client signs its own Ed25519 and X25519 keys together with the node's - nonce, so the identity key vouches for the encryption key — that is what - will make it safe for the node to wrap the GEK for a key that arrived over - the wire instead of one fetched from the hub's directory (H3). - - A first pairing needs a one-time code, which the hub never sees. Afterwards - the pin is the credential and a changed key is refused outright, the same - rule the client applies to `pk_node` (11.5.8). - """ - roster = self._ctx.get("roster") - if roster is None: - self._send({"type": "error", "detail": "Roster not available"}) - return - - if self._join_attempts >= MAX_JOIN_ATTEMPTS: - self._send({"type": "error", "detail": "Too many attempts"}) - return - - now = time.time() - recent = [t for t in self._ctx.get("join_failures", []) - if now - t < JOIN_FAILURE_WINDOW] - if len(recent) >= MAX_JOIN_FAILURES_WINDOW: - self._audit_join("join_throttled", f"{len(recent)} failures in window") - self._send({"type": "error", "detail": "Pairing temporarily locked"}) - return - - user_id = self._user_id or getattr(self, "_pending_sub", "") - username = self._username or getattr(self, "_pending_username", "") - if not user_id: - self._send({"type": "error", "detail": "Handshake required"}) - return - - pk_ed_b64 = msg.get("pk_ed25519", "") - pk_x_b64 = msg.get("pk_x25519", "") - code = msg.get("code", "") - ts = msg.get("ts", 0) - - try: - pk_ed_raw = base64.b64decode(pk_ed_b64) - pk_x_raw = base64.b64decode(pk_x_b64) - if len(pk_ed_raw) != 32 or len(pk_x_raw) != 32: - raise ValueError - pk_ed = Ed25519PublicKey.from_public_bytes(pk_ed_raw) - except Exception: - self._join_refuse("invalid_keys") - return - - if not isinstance(ts, int) or abs(time.time() - ts) > JOIN_TTL: - self._join_refuse("stale_request") - return - - # An empty group_id means operator pairing, which is node-wide. Anything - # else must be the group this connection authenticated to — a signature - # obtained for one group must not name another. - group_id = msg.get("group_id", "") or "" - session_group = self._group_id or getattr(self, "_pending_group", "") or "" - if group_id and group_id != session_group: - self._join_refuse("group_mismatch") - return - - transcript = join_transcript( - node_pk_b64=self._node_pk_b64(), - group_id=group_id, - user_id=user_id, - pk_ed25519_b64=pk_ed_b64, - pk_x25519_b64=pk_x_b64, - nonce_node=self._nonce_node, - ts=ts, - ) - try: - sig = base64.b64decode(msg.get("sig", "")) - except Exception: - self._join_refuse("invalid_signature_encoding") - return - if not self._verify_sig(pk_ed, transcript, sig): - self._join_refuse("signature_invalid") - return - - # One person may hold several devices here — a browser and a desktop - # client are two keys on one account. So the question is not "is this - # THE key" but "is this ONE OF this account's live devices". - device = await roster.find_device(user_id, pk_ed_b64) - if device and device["pk_x25519"] != pk_x_b64: - # The Ed25519 key is pinned but arrives with a different encryption - # key. The join transcript signs both together, so this is either a - # client that regenerated half its identity or something splicing - # two messages; either way the pair is not the one admitted. - self._join_refuse( - "key_changed", - f"pinned x25519={device['pk_x25519'][:16]} presented={pk_x_b64[:16]}") - return - known = device - if not known and await roster.list_devices(user_id): - # The account is known here but this key is not one of its devices. - # Not an error to shout about: it is a second browser or a new - # client, and the way in is a device-add approved by a device that - # is already trusted — no operator, no new invitation code. - self._join_refuse( - "unknown_device", - f"presented={pk_ed_b64[:16]} — approve it from a device already " - f"paired with this node") - return - if known: - # This group's own row first; then the join message's group_id (empty - # on the node-wide first connect); then the operator's node-wide row, - # which is where an operator opening any group finds their authority. - member = (await roster.get_member(session_group, user_id) - or await roster.get_member(group_id, user_id) - or await roster.get_member("", user_id)) - if not member and self._group_join_policy(session_group) == "open": - await roster.set_member( - group_id=session_group, user_id=user_id, role=ROLE_MEMBER, - status="active", approved_by="open-join", - ) - member = await roster.get_member(session_group, user_id) - - # A pending invite means the operator explicitly re-invited this - # person — require the code even if they already have a member - # row (e.g. they left and were re-invited, or were revoked then - # re-invited). Without this gate a stale roster row lets them - # back in without proving they received the new code. - pending_invite = any( - i["kind"] == KIND_ACCOUNT - and i["user_id"] == user_id - and i["group_id"] in (session_group, "") - for i in await roster.list_invites()) - # Or they bring a link for this group: somebody already pinned here - # through another group, which is the ordinary case for a link, or - # somebody removed from it and invited back. Only when they are not - # an active member — a member opening the group leaves the link for - # whoever it was meant for. - active = bool(member) and member.get("status") == "active" - if pending_invite or (code and not active): - if not code: - self._join_refuse("code_required") - return - invite = await roster.consume_invite(code, user_id, session_group) - if not invite: - self._join_refuse("code_invalid") - return - await roster.set_member( - group_id=invite["group_id"], user_id=user_id, - role=invite["role"], status="active", - approved_by=invite["created_by"], - ) - self._audit_join( - "join_pinned", - f"group={invite['group_id'][:8]} role={invite['role']} " - f"via={'link' if invite['kind'] == KIND_LINK else 'code'} " - "(device already known)") - member = (await roster.get_member(session_group, user_id) - or await roster.get_member(invite["group_id"], user_id)) - - if not member: - self._join_refuse("not_authorized_for_group") - return - await self._join_ok( - user_id, pk_x_raw, session_group, - role=member["role"] if member else "", - recognised=True, - ) - return - - if not code: - if self._group_join_policy(session_group) == "open": - # An open-join group admits anyone the hub calls a member, so a - # code would protect nothing — the hub can walk in through the - # front door. Pin what turns up and say so in the audit log. - await self._pin_and_admit( - roster, user_id, username, pk_ed_b64, pk_x_b64, - group_id=session_group, role=ROLE_MEMBER, - approved_by="open-join", via="tofu") - await self._join_ok(user_id, pk_x_raw, session_group, - role=ROLE_MEMBER, recognised=False) - return - self._join_refuse("code_required") - return - - invite = await roster.consume_invite(code, user_id, session_group) - if not invite: - self._join_refuse("code_invalid") - return - - await self._pin_and_admit( - # The name comes from the invitation, not from the token: the hub does - # not put a username claim in a JWT, so pinning from the session alone - # left the roster nameless and `member revoke <name>` unable to match. - roster, user_id, invite["username"] or username, pk_ed_b64, pk_x_b64, - group_id=invite["group_id"], role=invite["role"], - approved_by=invite["created_by"], - via="link" if invite["kind"] == KIND_LINK else "code") - # The roster row comes from the invitation; the key comes from the - # connection. An operator pairing is node-wide (empty group), but they - # redeemed the code while opening a group and expect to read it — and - # is_authorized() already grants an operator every group on this node. - await self._join_ok(user_id, pk_x_raw, session_group or invite["group_id"], - role=invite["role"], recognised=False) - # ── Device linking ─────────────────────────────────────────────────────── # @@ -1180,443 +848,6 @@ class WebRTCPeerSession( # no user keys since 2026-08-14 and therefore cannot countersign anything. # See docs/MESHBAY_DESIGN.md §3.3. - async def _do_device_request(self, msg: dict) -> None: - """ - A new device files itself as pending, bound to a code it displays. - - Served in the pre-proof window: by construction the caller holds no key - this node knows, so there is nothing yet to prove. Filing is inert — - nothing is admitted until an existing device countersigns. - """ - roster = self._ctx.get("roster") - if roster is None or not self._user_id or not self._nonce_node: - self._send({"type": "error", "detail": "Not ready for a device request"}) - return - - if not self._spend_device_attempt(): - return - - pk_ed_b64 = str(msg.get("pk_ed25519", "")) - pk_x_b64 = str(msg.get("pk_x25519", "")) - code_hash = str(msg.get("code_hash", "")) - if not (pk_ed_b64 and pk_x_b64 and code_hash): - self._send({"type": "error", "detail": "Missing device keys or code"}) - return - - # The account must already be known here. Anti-spam rather than a - # security boundary: the filing key is unpinned by construction, so this - # bounds the table, not the trust. - existing = await roster.list_devices(self._user_id) - if not existing: - self._send({"type": "error", - "detail": "This account has no device on this node yet — " - "an invitation code is what admits the first"}) - return - if len(existing) >= roster.MAX_DEVICES_PER_USER: - self._send({"type": "error", - "detail": f"Already {len(existing)} devices, which is the " - f"limit. Revoke one first."}) - return - - ts = int(msg.get("ts", 0)) - if abs(time.time() - ts) > DEVICE_TTL: - self._send({"type": "error", "detail": "Device request expired"}) - return - - transcript = device_request_transcript( - node_pk_b64=self._node_pk_b64(), user_id=self._user_id, - pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, - code_hash=code_hash, nonce_node=self._nonce_node, ts=ts) - try: - pk_ed = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64)) - sig = base64.b64decode(msg.get("sig", "")) - except Exception: - self._send({"type": "error", "detail": "Invalid device key encoding"}) - return - if not self._verify_sig(pk_ed, transcript, sig): - # Proof of possession, and nothing more: this says the caller holds - # the keys, never that they belong to this account. - self._send({"type": "error", "detail": "Device signature invalid"}) - return - - ttl = int(self._ctx.get("device_request_ttl") or 3600) - expires = await roster.file_device_request( - user_id=self._user_id, username=self._username or "", - pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, - code_hash=code_hash, ttl=ttl) - self._audit("device_request", f"{pk_ed_b64[:16]}") - log.info("Device request filed for %s (%s)", self._user_id[:8], - pk_ed_b64[:16]) - self._send({"type": MNP.DEVICE_REQUEST_ACK, "v": MNP_VERSION, - "expires_at": expires}) - - async def _do_device_lookup(self, msg: dict) -> None: - """ - List this account's pending device requests, each with its code hash. - - **The node never learns the code**, which is what makes it unable to - substitute a key. It answers with candidates; the approver recomputes - `sha256(code ‖ keys)` for each and keeps the one that matches. A node - offering fabricated keys would have to produce a hash matching - `sha256(code ‖ fabricated)` — and it does not know the code. - - An earlier version of this took the hash from the client and looked the - request up by it. That is circular: the client cannot compute the hash - without already knowing the keys it is asking about. - """ - roster = self._ctx.get("roster") - if roster is None or not self._user_id: - self._send({"type": "error", "detail": "Roster not available"}) - return - - pending = await roster.list_device_requests(self._user_id) - self._send({ - "type": MNP.DEVICE_LOOKUP_RESULT, "v": MNP_VERSION, - "requests": [ - {"pk_ed25519": r["pk_ed25519"], "pk_x25519": r["pk_x25519"], - "code_hash": r["code_hash"], "created_at": r["created_at"]} - for r in pending - ], - }) - - async def _do_device_add(self, msg: dict) -> None: - """ - Admit a device, countersigned by one this node already pinned. - - The whole control is in `_verify_device_signer`: the signature must - verify against a **live device of this same account**. The hub holds no - user keys and so cannot produce one. - """ - roster = self._ctx.get("roster") - if roster is None or not self._user_id or not self._nonce_node: - self._send({"type": "error", "detail": "Not ready to add a device"}) - return - if not self._spend_device_attempt(): - return - - pk_ed_b64 = str(msg.get("pk_ed25519", "")) - pk_x_b64 = str(msg.get("pk_x25519", "")) - ts = int(msg.get("ts", 0)) - if not (pk_ed_b64 and pk_x_b64): - self._send({"type": "error", "detail": "Missing device keys"}) - return - if abs(time.time() - ts) > DEVICE_TTL: - self._send({"type": "error", "detail": "Approval expired"}) - return - - transcript = device_add_transcript( - node_pk_b64=self._node_pk_b64(), user_id=self._user_id, - pk_ed25519_b64=pk_ed_b64, pk_x25519_b64=pk_x_b64, - nonce_node=self._nonce_node, ts=ts) - signer = await self._verify_device_signer(roster, transcript, - msg.get("sig", "")) - if signer is None: - self._audit("device_add_refused", pk_ed_b64[:16]) - self._send({"type": "error", - "detail": "Not signed by a device already paired here"}) - return - - devices = await roster.list_devices(self._user_id) - if len(devices) >= roster.MAX_DEVICES_PER_USER: - self._send({"type": "error", "detail": "Device limit reached"}) - return - - # Spend the request. Single use: an approval cannot be replayed, and a - # code that was used is gone whatever else happens next. - code_hash = str(msg.get("code_hash", "")) - if code_hash and not await roster.take_device_request( - code_hash, self._user_id): - self._send({"type": "error", - "detail": "That request is no longer pending"}) - return - - # The countersignature is **kept**, with the two fields needed to rebuild - # what it signed. Until 2026-09-07 it was verified here and thrown away, - # leaving only `added_by_pk` — which says *which* key approved and - # proves nothing to anyone else. `device_add_transcript` binds - # `nonce_node`, this connection's handshake nonce, so a stored signature - # without it is still unverifiable; that is why all three go in. - # - # This is what lets another member check for themselves that this device - # belongs to an account whose earlier device they have already pinned, - # instead of taking the node's word (Tier 2, docs/MESHBAY_DESIGN.md §3.3). - await roster.pin_identity( - user_id=self._user_id, username=self._username or "", - pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via="device", - label=str(msg.get("label", ""))[:64], added_by_pk=signer, - add_sig=str(msg.get("sig", "")), - add_nonce=base64.b64encode(self._nonce_node).decode(), - add_ts=ts) - self._audit("device_added", f"{pk_ed_b64[:16]} by {signer[:16]}") - log.info("Device added for %s: %s (approved by %s)", - self._user_id[:8], pk_ed_b64[:16], signer[:16]) - self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, - "pk_ed25519": pk_ed_b64}) - - async def _do_device_hello(self, msg: dict) -> None: - """ - Learn which of this account's devices is on this connection. - - The handshake authenticates a *group membership* (the GEK-HMAC) and an - *account* (the hub's token). It has never authenticated a device, and - while one account meant one key that was the same statement. It stopped - being so on 2026-08-18, and `_load_pinned_pk` — which resolves the - account's oldest live device — has been standing in for the real answer - ever since, including as the recorded uploader of every file. - - What is checked, in order: the key is a live device *of this account* in - the node's own roster (never a token claim — that is - `docs/MESHBAY_DESIGN.md` §3.2's rule), the timestamp is fresh, and the - signature verifies over a transcript naming this node, this group and - this connection's nonce. A key that is merely well-formed proves nothing. - - Idempotent for the same key, refused for a different one: a connection - does not get to change device half way through, which would let one - session's uploads be attributed to two. - """ - roster = self._ctx.get("roster") - if roster is None or not self._user_id: - self._send({"type": "error", "detail": "Roster not available"}) - return - if not self._spend_device_attempt(): - return - - pk_ed_b64 = str(msg.get("pk_ed25519", "")) - ts = int(msg.get("ts", 0) or 0) - if not pk_ed_b64: - self._send({"type": "error", "detail": "Missing device key"}) - return - if self._device_confirmed and pk_ed_b64 != self._pinned_pk: - self._send({"type": "error", - "detail": "This connection is already another device"}) - return - if abs(time.time() - ts) > DEVICE_TTL: - self._send({"type": "error", "detail": "Stale device_hello"}) - return - - device = await roster.find_device(self._user_id, pk_ed_b64) - if device is None: - self._audit("device_hello_refused", pk_ed_b64[:16]) - self._send({"type": "error", - "detail": "Not a device paired here"}) - return - - transcript = device_hello_transcript( - node_pk_b64=self._node_pk_b64(), group_id=self._group_id or "", - user_id=self._user_id, pk_ed25519_b64=pk_ed_b64, - nonce_node=self._nonce_node, ts=ts) - try: - pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_ed_b64)) - except Exception: - self._send({"type": "error", "detail": "Unreadable device key"}) - return - try: - sig = base64.b64decode(msg.get("sig", "")) - except Exception: - sig = b"" - if not self._verify_sig(pk, transcript, sig): - self._audit("device_hello_refused", pk_ed_b64[:16]) - self._send({"type": "error", "detail": "Signature verification failed"}) - return - - self._pinned_pk = pk_ed_b64 - self._device_confirmed = True - log.info("Device identified on connection: user=%s device=%s", - self._user_id[:8], pk_ed_b64[:16]) - self._send({"type": MNP.DEVICE_HELLO_ACK, "v": MNP_VERSION, - "pk_ed25519": pk_ed_b64}) - - async def _do_device_list(self, msg: dict) -> None: - """This account's devices. Anyone may read their own, nobody else's.""" - roster = self._ctx.get("roster") - if roster is None or not self._user_id: - self._send({"type": "error", "detail": "Roster not available"}) - return - devices = await roster.list_devices(self._user_id) - pending = await roster.pending_device_requests(self._user_id) - self._send({ - "type": MNP.DEVICE_LIST_RESULT, "v": MNP_VERSION, - "pending": pending, - "devices": [ - {"pk_ed25519": d["pk_ed25519"], "label": d.get("label", ""), - "pinned_at": d["pinned_at"], "pinned_via": d["pinned_via"], - "added_by_pk": d.get("added_by_pk", ""), - "is_this_one": d["pk_ed25519"] == self._pinned_pk} - for d in devices - ], - }) - - async def _do_device_revoke(self, msg: dict) -> None: - """ - Retire one of this account's devices — a lost laptop. - - Countersigned like an addition, by a live device of the same account. - The last one cannot go: an account with no device on this node can only - return through an operator's invitation code, and doing that to yourself - by accident is not a mistake worth allowing. - """ - roster = self._ctx.get("roster") - if roster is None or not self._user_id or not self._nonce_node: - self._send({"type": "error", "detail": "Not ready"}) - return - if not self._spend_device_attempt(): - return - - target = str(msg.get("pk_ed25519", "")) - ts = int(msg.get("ts", 0)) - if not target: - self._send({"type": "error", "detail": "Missing device key"}) - return - if abs(time.time() - ts) > DEVICE_TTL: - self._send({"type": "error", "detail": "Request expired"}) - return - - victim = await roster.find_device(self._user_id, target) - if victim is None: - self._send({"type": "error", "detail": "No such device"}) - return - - transcript = device_add_transcript( - node_pk_b64=self._node_pk_b64(), user_id=self._user_id, - pk_ed25519_b64=target, pk_x25519_b64=victim["pk_x25519"], - nonce_node=self._nonce_node, ts=ts) - signer = await self._verify_device_signer(roster, transcript, - msg.get("sig", "")) - if signer is None: - self._send({"type": "error", - "detail": "Not signed by a device already paired here"}) - return - - if len(await roster.list_devices(self._user_id)) <= 1: - self._send({"type": "error", - "detail": "This is your only device here — removing it " - "would need an operator code to come back"}) - return - - await roster.revoke_device(self._user_id, target) - # A revoked device holds every chat key it ever received — a lost laptop - # reads the group's chat until the epoch moves. - await self._new_chat_epoch(self._group_id or "", "device_revoke") - self._audit("device_revoked", f"{target[:16]} by {signer[:16]}") - log.info("Device revoked for %s: %s", self._user_id[:8], target[:16]) - self._send({"type": MNP.DEVICE_ADD_ACK, "v": MNP_VERSION, - "revoked": target}) - - async def _verify_device_signer(self, roster, transcript: bytes, - sig_b64: str) -> str | None: - """ - The pinned key that signed this, or None. - - Every live device of the account is tried, because any of them may - approve. A revoked one is not in the list — that is the point of marking - rather than deleting: a lost laptop must stop being able to admit its - replacement. - """ - try: - sig = base64.b64decode(sig_b64) - except Exception: - return None - for device in await roster.list_devices(self._user_id): - try: - pk = Ed25519PublicKey.from_public_bytes( - base64.b64decode(device["pk_ed25519"])) - except Exception: - continue - if self._verify_sig(pk, transcript, sig): - return device["pk_ed25519"] - return None - - def _spend_device_attempt(self) -> bool: - """ - Bound guessing on this connection, as the join path does. - - A code is 40 bits, single use and bound to the keys it names, so this is - depth rather than the control — but an unbounded loop over the lookup is - still a free oracle, and a burst of failures belongs in the audit log. - """ - self._device_attempts = getattr(self, "_device_attempts", 0) + 1 - if self._device_attempts > 5: - self._audit("device_attempts_exceeded", str(self._device_attempts)) - self._send({"type": "error", - "detail": "Too many device attempts on this connection"}) - return False - return True - - def _group_join_policy(self, group_id: str) -> str: - """ - Admission policy for a group, read from the node's own configuration. - - Never from the hub: a hub that could declare a group open would be handed - the key to it (docs/MESHBAY_DESIGN.md §3.4). - """ - gctx = (self._ctx.get("groups") or {}).get(group_id) or {} - return gctx.get("join_policy", "invite") - - async def _pin_and_admit( - self, roster, user_id: str, username: str, pk_ed_b64: str, pk_x_b64: str, - *, group_id: str, role: str, approved_by: str, via: str, - ) -> None: - await roster.pin_identity( - user_id=user_id, username=username, - pk_ed25519=pk_ed_b64, pk_x25519=pk_x_b64, via=via, - ) - await roster.set_member( - group_id=group_id, user_id=user_id, role=role, - status="active", approved_by=approved_by, - ) - if role == ROLE_OPERATOR: - self._ctx["has_admin_authority"] = True - - log.info("Identity pinned (%s): user=%s role=%s", via, user_id[:8], role) - self._audit_join("join_pinned", f"role={role} via={via}") - - async def _join_ok( - self, user_id: str, pk_x_raw: bytes, group_id: str, - *, role: str, recognised: bool, - ) -> None: - """ - Answer a join, wrapping the group key for the key the caller just proved. - - This is the H3 fix. The inviter used to fetch the invitee's public key from - the hub and wrap the GEK for whatever came back, so a hub that answered - with its own key was handed the group key by an honest member following the - protocol exactly. The node now wraps for a key that arrived from its owner - over an authenticated channel, bound to a pinned identity. - """ - reply = { - "type": MNP.JOIN_RESULT, - "v": MNP_VERSION, - "ok": True, - "recognised": recognised, - "role": role, - } - - roster = self._ctx["roster"] - if group_id and not await roster.is_authorized(group_id, user_id): - # Pinned on this node, but not admitted to this group. Hub membership - # alone must not produce a key. - reply["gek"] = False - reply["reason"] = "not_authorized_for_group" - self._send(reply) - self._audit_join("join_no_gek", f"group={group_id[:8]} not authorized") - return - - gctx = (self._ctx.get("groups") or {}).get(group_id) or {} - gek = gctx.get("gek") - if not gek: - reply["gek"] = False - reply["reason"] = "no_gek" - self._send(reply) - return - - bundle = wrap_gek_aes(gek, pk_x_raw) - reply["gek"] = True - reply["pk_eph_b64"] = bundle["pk_eph_b64"] - reply["nonce_b64"] = bundle["nonce_b64"] - reply["wrapped_b64"] = bundle["wrapped_b64"] - self._send(reply) - self._audit_join("gek_wrapped", f"group={group_id[:8]}") - def _do_member_revoke(self, msg: dict) -> None: """ Stop serving the group key to someone, at the operator's request. @@ -2969,81 +2200,6 @@ class WebRTCPeerSession( else: self._send({"type": "error", "detail": "Unknown admin operation"}) - async def _admin_exec_invite_create( - self, pending: dict, transcript: bytes, sig: bytes, - ) -> None: - # Node operator only. A group admin who does not run the node has no - # authority over who this node admits (deny by default). Delegation is - # designed but deferred — see docs/MESHBAY_DESIGN.md §3.4. - if not await self._verify_admin_sig(transcript, sig): - self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}") - return - - payload = pending["payload"] - try: - result = await self._run_op( - ops.create_invite, - payload["group_id"], - payload.get("username", ""), - user_id=payload["user_id"], - created_by=self._user_id or "", - ) - except ops.OpError as e: - self._send({"type": "error", "detail": e.message}) - return - - self._audit("invite_create", f"target={payload['user_id'][:8]}") - self._send({ - "type": MNP.INVITE_RESULT, - "v": MNP_VERSION, - "code": result["code"], - "expires_at": result["expires_at"], - "user_id": result["user_id"], - "username": result.get("username", ""), - }) - - async def _admin_exec_invite_link_create( - self, pending: dict, transcript: bytes, sig: bytes, - ) -> None: - if not await self._verify_admin_sig(transcript, sig): - self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", "invite_link_create") - return - try: - result = await self._run_op( - ops.create_link_invite, pending["payload"]["group_id"], - created_by=self._user_id or "") - except ops.OpError as e: - self._send({"type": "error", "detail": e.message}) - return - self._audit("invite_link_create", f"invite={result['invite_id'][:8]}") - self._send({ - "type": MNP.INVITE_LINK_RESULT, - "v": MNP_VERSION, - "code": result["code"], - "invite_id": result["invite_id"], - "expires_at": result["expires_at"], - "group_id": result["group_id"], - }) - - async def _admin_exec_invite_cancel( - self, pending: dict, transcript: bytes, sig: bytes, - ) -> None: - if not await self._verify_admin_sig(transcript, sig): - self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", f"invite_cancel:{pending['subject'][:8]}") - return - payload = pending["payload"] - try: - await self._run_op(ops.cancel_invite, payload["group_id"], payload["invite_id"]) - except ops.OpError as e: - self._send({"type": "error", "detail": e.message}) - return - self._audit("invite_cancel", f"invite={payload['invite_id'][:8]}") - self._send({"type": "ack", "v": MNP_VERSION, "detail": "invite_cancelled", - "invite_id": payload["invite_id"]}) - def _send(self, obj: dict) -> None: # Stamp the reply with the id of the request being answered, so the # caller never has to guess. Only for this session's own replies: a |