diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
| -rw-r--r-- | packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py | 411 |
1 files changed, 356 insertions, 55 deletions
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 34a96bd..fe3ee2e 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -57,10 +57,16 @@ from meshbay_common.handshake import ( from meshbay_common.adminop import ( ADMIN_CHALLENGE_TTL, OP_FILE_DELETE, - OP_GEK_BUNDLE_STORE, + OP_INVITE_CREATE, admin_transcript, ) -from meshbay_common.crypto import pk_to_b64 +from meshbay_common.crypto import pk_to_b64, wrap_gek_aes +from meshbay_common.join import ( + JOIN_TTL, + ROLE_MEMBER, + ROLE_OPERATOR, + join_transcript, +) from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex @@ -85,6 +91,14 @@ MAX_CONCURRENT_TRANSCODES = 2 # 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 UPLOAD_DIR_NAME = ".uploads" # Conservative allowlist: also what keeps markup out of filenames, which the node admin # UI used to render unescaped (finding H2). @@ -218,6 +232,11 @@ class WebRTCPeerSession: self._username: str = "" self._pk_user: str = "" self._gek_challenge: bytes | None = None + # Same value as the GEK challenge, but kept for the life of the connection: + # a join_request is signed over it, and it must stay verifiable after the + # handshake clears the challenge (an operator pairs while already connected). + self._nonce_node: bytes = b"" + self._join_attempts = 0 self._nonce_client: bytes = b"" self._admin_ops: dict[str, dict] = {} # op_id → pending admin operation self._uploads: dict[str, dict] = {} # filename → {next_index, bytes} @@ -259,6 +278,12 @@ class WebRTCPeerSession: asyncio.ensure_future(self._do_gek_bundle_fetch()) else: asyncio.ensure_future(self._do_keypair_bundle_fetch()) + elif mtype == MNP.JOIN_REQUEST and self._nonce_node: + # Valid both before the GEK proof (a new member has no GEK to prove + # with) and after it (an operator pairing a browser is already + # connected). Authority comes from the pairing code and the + # signature, never from the session state. + asyncio.ensure_future(self._do_join_request(msg)) elif self._user_id is None: self._send({"type": "error", "detail": "Handshake required"}) elif mtype == MNP.INDEX_SYNC: @@ -277,8 +302,8 @@ class WebRTCPeerSession: self._do_file_delete(msg) elif mtype == MNP.ADMIN_RESPONSE: self._do_admin_response(msg) - elif mtype == MNP.GEK_BUNDLE_STORE: - self._do_gek_bundle_store(msg) + elif mtype == MNP.INVITE_CREATE: + self._do_invite_create(msg) elif mtype == MNP.KEYPAIR_BUNDLE_STORE: asyncio.ensure_future(self._do_keypair_bundle_store(msg)) elif mtype == MNP.STREAM_REQUEST: @@ -359,6 +384,7 @@ class WebRTCPeerSession: return self._gek_challenge = os.urandom(NONCE_LEN) + self._nonce_node = self._gek_challenge self._send({ "type": MNP.HANDSHAKE_CHALLENGE, "v": MNP_VERSION, @@ -472,49 +498,40 @@ class WebRTCPeerSession: else: self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) - def _do_gek_bundle_store(self, msg: dict) -> None: + def _do_invite_create(self, msg: dict) -> None: """ - Request to store a wrapped GEK bundle for a target user. - - Finding C5b: this used to write whatever any authenticated member sent, with - INSERT OR REPLACE semantics, and then auto-activate the bundle if it was - addressed to the node operator. Since the operator's X25519 public key is - public — the node even hands it out in handshake_ack — any member could wrap - a GEK of their own choosing for the operator and make the node adopt it, - locking every legitimate member out of the group and taking over the key. + Issue a one-time pairing code for someone the operator wants to admit. - Storing a bundle is now a node-operator operation gated by an Ed25519 - challenge, and nothing arriving over MNP can activate a GEK: activation - happens only through the local admin UI or the CLI. + 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. """ - bundle_store = self._ctx.get("bundle_store") - if not bundle_store: - self._send({"type": "error", "detail": "Bundle store not available"}) + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) return - target_user_id = msg.get("user_id", "") + invitee_id = msg.get("user_id", "") group_id = msg.get("group_id") or self._group_id - pk_eph = msg.get("pk_eph_b64", "") - nonce = msg.get("nonce_b64", "") - wrapped = msg.get("wrapped_b64", "") - - if not target_user_id or not pk_eph or not nonce or not wrapped or not group_id: - self._send({"type": "error", "detail": "Missing bundle fields"}) + 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._ctx.get("admin_pk_ed25519"): + if not self._has_admin_authority(): self._send({ "type": "error", - "detail": "No admin key pinned — bundle storage refused", + "detail": "No operator paired — run `meshbay-node operator pair`", }) return - self._issue_admin_challenge(OP_GEK_BUNDLE_STORE, target_user_id, { + self._issue_admin_challenge(OP_INVITE_CREATE, invitee_id, { "group_id": group_id, - "user_id": target_user_id, - "pk_eph_b64": pk_eph, - "nonce_b64": nonce, - "wrapped_b64": wrapped, + "user_id": invitee_id, + "username": str(msg.get("username", ""))[:64], }) async def _do_keypair_bundle_fetch(self) -> None: @@ -560,6 +577,240 @@ class WebRTCPeerSession: "detail": "keypair_bundle_stored", }) + # ── 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) + asyncio.ensure_future(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 + + known = await roster.get_identity(user_id) + if known: + if known["pk_ed25519"] != pk_ed_b64 or known["pk_x25519"] != pk_x_b64: + # The blocking warning, raised where it matters: whoever this is + # holds a different key than the person the operator paired. + self._join_refuse( + "key_changed", + f"pinned={known['pk_ed25519'][:16]} presented={pk_ed_b64[:16]}") + return + member = await roster.get_member(group_id, user_id) + 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) + if not invite: + self._join_refuse("code_invalid") + return + + await self._pin_and_admit( + roster, user_id, username, pk_ed_b64, pk_x_b64, + group_id=invite["group_id"], role=invite["role"], + approved_by=invite["created_by"], via="code") + await self._join_ok(user_id, pk_x_raw, invite["group_id"], + role=invite["role"], recognised=False) + + 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 (§3.4 of docs/invite-pairing-v1.md). + """ + 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 _audit_pre_proof_fetch(self, mtype: str) -> None: """Record bundle access made before the GEK proof (C4).""" audit = self._ctx.get("audit_store") @@ -903,9 +1154,8 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "File not found"}) return - admin_pk = self._ctx.get("admin_pk_ed25519") has_uploader_pk = bool(entry.uploader_pk) - if not admin_pk and not has_uploader_pk: + if not self._has_admin_authority() and not has_uploader_pk: self._send({"type": "error", "detail": "No authorized key for deletion"}) return @@ -956,6 +1206,43 @@ class WebRTCPeerSession: except Exception: return False + def _has_admin_authority(self) -> bool: + """ + Cheap synchronous pre-check: is there anyone who could authorize this? + + Only decides whether to issue a challenge at all — the gate is + `_verify_admin_sig`. The flag is set at startup and refreshed in-process + when an operator pairs. + """ + return bool(self._ctx.get("admin_pk_ed25519") + or self._ctx.get("has_admin_authority")) + + async def _verify_admin_sig(self, transcript: bytes, sig: bytes) -> bool: + """ + Check a signature against every key holding node-operator authority. + + Read from the roster on each call rather than cached: revoking a paired + browser must take effect immediately, and admin operations are rare enough + that a SQLite read costs nothing. `admin_pk_ed25519` in node.toml is still + honoured so an existing deployment keeps working until its operator pairs + (M3) — it is the legacy form of the same statement. + """ + legacy = self._ctx.get("admin_pk_ed25519") + if self._verify_sig(legacy, transcript, sig): + return True + + roster = self._ctx.get("roster") + if roster is None: + return False + for pk_b64 in await roster.operator_pks(): + try: + pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(pk_b64)) + except Exception: + continue + if self._verify_sig(pk, transcript, sig): + return True + return False + def _do_admin_response(self, msg: dict) -> None: op_id = msg.get("op_id", "") sig_b64 = msg.get("signature", "") @@ -985,14 +1272,15 @@ class WebRTCPeerSession: ) if pending["op"] == OP_FILE_DELETE: - self._admin_exec_file_delete(pending, transcript, sig_bytes) - elif pending["op"] == OP_GEK_BUNDLE_STORE: asyncio.ensure_future( - self._admin_exec_bundle_store(pending, transcript, sig_bytes)) + self._admin_exec_file_delete(pending, transcript, sig_bytes)) + elif pending["op"] == OP_INVITE_CREATE: + asyncio.ensure_future( + self._admin_exec_invite_create(pending, transcript, sig_bytes)) else: self._send({"type": "error", "detail": "Unknown admin operation"}) - def _admin_exec_file_delete( + async def _admin_exec_file_delete( self, pending: dict, transcript: bytes, sig: bytes, ) -> None: file_id = pending["subject"] @@ -1012,7 +1300,7 @@ class WebRTCPeerSession: # Node operator, or the user who uploaded this file — verified by the key # recorded at upload time, never by a JWT claim (the hub controls those). - if not (self._verify_sig(self._ctx.get("admin_pk_ed25519"), transcript, sig) + if not (await self._verify_admin_sig(transcript, sig) or self._verify_sig(uploader_pk, transcript, sig)): self._send({"type": "error", "detail": "Signature verification failed"}) self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") @@ -1020,33 +1308,46 @@ class WebRTCPeerSession: self._exec_file_delete(ctx, file_id, entry) - async def _admin_exec_bundle_store( + 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 what this node stores (draft-v4 §4.2.x, deny by default). - if not self._verify_sig(self._ctx.get("admin_pk_ed25519"), transcript, sig): + # authority over who this node admits (deny by default). Delegation is + # designed but deferred — see §6.2 of docs/invite-pairing-v1.md. + if not await self._verify_admin_sig(transcript, sig): self._send({"type": "error", "detail": "Signature verification failed"}) - self._audit("admin_auth_failed", f"gek_bundle_store:{pending['subject'][:16]}") + self._audit("admin_auth_failed", f"invite_create:{pending['subject'][:16]}") return - payload = pending["payload"] - bundle_store = self._ctx.get("bundle_store") - if not bundle_store: - self._send({"type": "error", "detail": "Bundle store not available"}) + roster = self._ctx.get("roster") + if roster is None: + self._send({"type": "error", "detail": "Roster not available"}) return - await bundle_store.store( - payload["group_id"], payload["user_id"], - payload["pk_eph_b64"], payload["nonce_b64"], payload["wrapped_b64"], + payload = pending["payload"] + code = await roster.create_invite( + group_id=payload["group_id"], + user_id=payload["user_id"], + role=ROLE_MEMBER, + created_by=self._user_id or "", ) - log.info("GEK bundle stored: group=%s user=%s", + invites = await roster.list_invites() + expires = next( + (i["expires_at"] for i in invites + if i["user_id"] == payload["user_id"] + and i["group_id"] == payload["group_id"]), "") + + log.info("Invite created: group=%s user=%s", payload["group_id"][:8], payload["user_id"][:8]) - self._audit("gek_bundle_store", f"target={payload['user_id'][:8]}") + self._audit("invite_create", f"target={payload['user_id'][:8]}") + # The code exists in the clear exactly here and in the operator's hands. self._send({ - "type": "ack", "v": MNP_VERSION, - "detail": "gek_bundle_stored", + "type": MNP.INVITE_RESULT, + "v": MNP_VERSION, + "code": code, + "expires_at": expires, "user_id": payload["user_id"], + "username": payload.get("username", ""), }) def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: |