"""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}) # ── 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 ` 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 ─────────────────────────────────────────────────────── # # A person may hold several devices on one node. The authority admitting a # new one is a key the node already pinned — never the hub, which has stored # 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]}") 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"]})