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 | 353 |
1 files changed, 327 insertions, 26 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 e692c80..13e90c8 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -24,7 +24,10 @@ Signaling flow (handled externally by the hub): import asyncio import base64 +import hashlib +import hmac import logging +import os import struct from pathlib import Path from typing import Any @@ -32,7 +35,10 @@ from typing import Any import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel -from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) from meshbay_common import MNP_VERSION from meshbay_common.crypto import pk_to_b64 @@ -46,6 +52,15 @@ CHUNK_SIZE = 1024 * 1024 MAX_MSG = 64 * 1024 * 1024 +def _extract_dtls_fingerprint(sdp: str) -> bytes: + """Extract the DTLS SHA-256 fingerprint from SDP as raw 32 bytes.""" + for line in sdp.splitlines(): + if line.startswith("a=fingerprint:sha-256 "): + hex_str = line.split(" ", 1)[1].replace(":", "") + return bytes.fromhex(hex_str) + return b"" + + STREAM_SEGMENT_SIZE = 256 * 1024 _H264_PROFILES = {"Baseline": "42", "Main": "4d", "High": "64", "High 10": "6e"} @@ -153,6 +168,9 @@ class WebRTCPeerSession: self._peer_id: str = peer_id self._remote_ip: str = "" self._username: str = "" + self._pk_user: str = "" + self._gek_challenge: bytes | None = None + self._admin_challenges: dict[str, bytes] = {} def _setup_channel(self, channel: RTCDataChannel) -> None: self._channel = channel @@ -171,6 +189,12 @@ class WebRTCPeerSession: try: if mtype == MNP.HANDSHAKE: self._do_handshake(msg) + elif mtype == MNP.HANDSHAKE_RESPONSE: + self._do_handshake_response(msg) + elif mtype == MNP.GEK_BUNDLE_FETCH and self._gek_challenge is not None: + asyncio.ensure_future(self._do_gek_bundle_fetch()) + elif mtype == MNP.KEYPAIR_BUNDLE_FETCH and self._gek_challenge is not None: + asyncio.ensure_future(self._do_keypair_bundle_fetch()) elif self._user_id is None: self._send({"type": "error", "detail": "Handshake required"}) elif mtype == MNP.INDEX_SYNC: @@ -179,8 +203,6 @@ class WebRTCPeerSession: self._do_file_request(msg) elif mtype == MNP.STREAM_SEGMENT: self._do_stream_segment(msg) - elif mtype == MNP.GEK_REQUEST: - self._do_gek_request() elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) elif mtype == MNP.CHAT_HISTORY: @@ -189,6 +211,12 @@ class WebRTCPeerSession: self._do_file_upload(msg) elif mtype == MNP.FILE_DELETE: self._do_file_delete(msg) + elif mtype == MNP.ADMIN_RESPONSE: + self._do_admin_response(msg) + elif mtype == MNP.GEK_BUNDLE_STORE: + asyncio.ensure_future(self._do_gek_bundle_store(msg)) + elif mtype == MNP.KEYPAIR_BUNDLE_STORE: + asyncio.ensure_future(self._do_keypair_bundle_store(msg)) elif mtype == MNP.STREAM_REQUEST: asyncio.ensure_future(self._stream_video(msg)) else: @@ -234,23 +262,235 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "Group not hosted on this node"}) return - self._user_id = decoded["sub"] - self._group_id = group_id - self._username = decoded.get("username", "") + # Store decoded JWT data but DO NOT set self._user_id yet — + # the user is not authenticated until they prove GEK possession. + self._pending_sub = decoded["sub"] + self._pending_group = group_id + self._pending_username = decoded.get("username", "") + self._pending_pk_user = decoded.get("pk_user", "") + + ctx = self._ctx + if "groups" in ctx and group_id: + gctx = ctx["groups"].get(group_id, ctx) + else: + gctx = ctx + gek = gctx.get("gek") + + nonce = os.urandom(32) + self._gek_challenge = nonce + challenge = { + "type": MNP.HANDSHAKE_CHALLENGE, + "v": MNP_VERSION, + "nonce": base64.b64encode(nonce).decode(), + } + if not gek: + self._send({ + "type": "error", + "detail": "Group encryption not initialized — contact node operator", + }) + return + self._send(challenge) + + def _do_handshake_response(self, msg: dict) -> None: + if not self._gek_challenge or not hasattr(self, "_pending_sub"): + self._send({"type": "error", "detail": "No pending handshake challenge"}) + return + + group_id = self._pending_group + ctx = self._ctx + if "groups" in ctx and group_id: + gctx = ctx["groups"].get(group_id, ctx) + else: + gctx = ctx + gek = gctx.get("gek") + + if not gek: + self._send({"type": "error", "detail": "Group encryption not initialized"}) + self._gek_challenge = None + return + + proof = msg.get("proof", "") + try: + proof_bytes = base64.b64decode(proof) + except Exception: + self._send({"type": "error", "detail": "Invalid proof encoding"}) + return + + offer_fp = b"" + answer_fp = b"" + if self._pc.remoteDescription: + offer_fp = _extract_dtls_fingerprint(self._pc.remoteDescription.sdp) + if self._pc.localDescription: + answer_fp = _extract_dtls_fingerprint(self._pc.localDescription.sdp) + + data = self._gek_challenge + offer_fp + answer_fp + expected = hmac.new(gek, data, hashlib.sha256).digest() + if not hmac.compare_digest(proof_bytes, expected): + self._send({"type": "error", "detail": "GEK proof failed"}) + self._gek_challenge = None + self._audit_auth_failed(group_id, "GEK HMAC mismatch") + return + + self._gek_challenge = None + self._complete_handshake() + + def _complete_handshake(self) -> None: + self._user_id = self._pending_sub + self._group_id = self._pending_group + self._username = self._pending_username + self._pk_user = self._pending_pk_user peers = self._ctx.get("_peers") if peers is not None: peers[self._user_id] = self + node_user_id = self._ctx.get("node_user_id") log.info("WebRTC handshake OK — user=%s group=%s", - self._user_id[:8], group_id[:8] if group_id else "none") - self._send({ + self._user_id[:8], + self._group_id[:8] if self._group_id else "none") + ack = { "type": MNP.HANDSHAKE_ACK, "v": MNP_VERSION, "node_pk": pk_to_b64(self._ctx["sk_node"].public_key()), - }) + "is_node_admin": bool(node_user_id and self._user_id == node_user_id), + } + if node_user_id: + ack["node_user_id"] = node_user_id + pk_x_b64 = self._ctx.get("pk_x25519_b64") + if pk_x_b64: + ack["node_pk_x25519"] = pk_x_b64 + self._send(ack) self._audit("handshake") + async def _do_gek_bundle_fetch(self) -> None: + """Serve the caller's wrapped GEK bundle during the handshake window.""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + return + + group_id = getattr(self, "_pending_group", "") + user_id = getattr(self, "_pending_sub", "") + if not group_id or not user_id: + self._send({"type": "error", "detail": "No pending handshake"}) + return + + bundle = await bundle_store.fetch(group_id, user_id) + if bundle: + self._send({ + "type": MNP.GEK_BUNDLE_RESP, + "v": MNP_VERSION, + "found": True, + "pk_eph_b64": bundle["pk_eph_b64"], + "nonce_b64": bundle["nonce_b64"], + "wrapped_b64": bundle["wrapped_b64"], + }) + else: + self._send({"type": MNP.GEK_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + + async def _do_gek_bundle_store(self, msg: dict) -> None: + """Store a wrapped GEK bundle for a target user (admin operation).""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": "error", "detail": "Bundle store not available"}) + return + + target_user_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"}) + return + + await bundle_store.store(group_id, target_user_id, pk_eph, nonce, wrapped) + log.info("GEK bundle stored: group=%s user=%s", group_id[:8], target_user_id[:8]) + self._audit("gek_bundle_store", f"target={target_user_id[:8]}") + + self._send({ + "type": "ack", "v": MNP_VERSION, + "detail": "gek_bundle_stored", + "user_id": target_user_id, + }) + + # Auto-activate GEK if the bundle is for the node operator + node_user_id = self._ctx.get("node_user_id") + if node_user_id and target_user_id == node_user_id and group_id: + await self._try_activate_gek(group_id, target_user_id) + + async def _try_activate_gek(self, group_id: str, user_id: str) -> None: + """Unwrap and activate GEK for the node when the operator's bundle arrives.""" + from meshbay_common.crypto import unwrap_gek_aes + + bundle_store = self._ctx.get("bundle_store") + sk_x_raw = self._ctx.get("sk_x25519_raw") + pk_x_raw = self._ctx.get("pk_x25519_raw") + if not bundle_store or not sk_x_raw or not pk_x_raw: + return + + bundle = await bundle_store.fetch(group_id, user_id) + if not bundle: + return + + try: + gek = unwrap_gek_aes(bundle, sk_x_raw, pk_x_raw) + except Exception as e: + log.warning("Failed to unwrap GEK for auto-activation: %s", e) + return + + groups = self._ctx.get("groups") + if groups and group_id in groups: + groups[group_id]["gek"] = gek + log.info("GEK auto-activated for group %s", group_id[:8]) + elif "gek" in self._ctx: + self._ctx["gek"] = gek + log.info("GEK auto-activated (single-group mode)") + + async def _do_keypair_bundle_fetch(self) -> None: + """Serve the caller's encrypted keypair bundle during the handshake window.""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + return + + user_id = getattr(self, "_pending_sub", "") + if not user_id: + self._send({"type": "error", "detail": "No pending handshake"}) + return + + bundle_enc = await bundle_store.fetch_keypair(user_id) + if bundle_enc: + self._send({ + "type": MNP.KEYPAIR_BUNDLE_RESP, + "v": MNP_VERSION, + "found": True, + "bundle_enc": bundle_enc, + }) + else: + self._send({"type": MNP.KEYPAIR_BUNDLE_RESP, "v": MNP_VERSION, "found": False}) + + async def _do_keypair_bundle_store(self, msg: dict) -> None: + """Store an encrypted keypair bundle (user backs up their own keys on node).""" + bundle_store = self._ctx.get("bundle_store") + if not bundle_store: + self._send({"type": "error", "detail": "Bundle store not available"}) + return + + bundle_enc = msg.get("bundle_enc", "") + if not bundle_enc: + self._send({"type": "error", "detail": "Missing bundle_enc"}) + return + + await bundle_store.store_keypair(self._user_id, bundle_enc) + log.info("Keypair bundle stored for user=%s", self._user_id[:8]) + self._audit("keypair_bundle_store") + self._send({ + "type": "ack", "v": MNP_VERSION, + "detail": "keypair_bundle_stored", + }) + def _audit_auth_failed(self, group_id: str, reason: str) -> None: audit = self._ctx.get("audit_store") if audit: @@ -275,6 +515,7 @@ class WebRTCPeerSession: { "id": e.id, "name": e.name, "path": e.path, "size": e.size, "type": e.type, "added_at": e.added_at, + "uploader_id": e.uploader_id, } for e in idx.entries ] @@ -286,18 +527,6 @@ class WebRTCPeerSession: "entries": entries, }) - def _do_gek_request(self) -> None: - ctx = self._group_ctx() - gek = ctx.get("gek") - if not gek: - self._send({"type": "error", "detail": "No GEK available"}) - return - self._send({ - "type": MNP.GEK_RESPONSE, - "v": MNP_VERSION, - "gek_b64": base64.b64encode(gek).decode(), - }) - def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg["file_id"] @@ -378,7 +607,7 @@ class WebRTCPeerSession: if chat_store: raw = payload.encode() if isinstance(payload, str) else payload asyncio.ensure_future(chat_store.save_message( - sender_id=msg.get("sender_id", self._user_id), + sender_id=self._user_id, iteration=msg.get("iteration", 0), payload=raw, thread_id=msg.get("thread_id"), @@ -389,7 +618,7 @@ class WebRTCPeerSession: broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, - "sender_id": msg.get("sender_id", self._user_id), + "sender_id": self._user_id, "sender_name": sender_name, "payload": payload, "thread_id": msg.get("thread_id"), @@ -493,6 +722,18 @@ class WebRTCPeerSession: tmp_path.rename(final_path) log.info("Upload complete: %s (%d chunks)", safe_name, total_chunks) self._audit("file_upload", safe_name) + self._register_uploader(ctx, safe_name) + + def _register_uploader(self, ctx: dict, filename: str) -> None: + """Tag the index entry with the uploader's user_id after upload completes.""" + idx = ctx.get("index") + if not idx: + return + for entry in idx.entries: + if entry.name == filename and entry.path == "": + entry.uploader_id = self._user_id + entry.uploader_pk = self._pk_user + return def _do_file_delete(self, msg: dict) -> None: ctx = self._group_ctx() @@ -501,16 +742,76 @@ class WebRTCPeerSession: self._send({"type": "error", "detail": "Missing file_id"}) return - node_user_id = self._ctx.get("node_user_id") - if node_user_id and self._user_id != node_user_id: - self._send({"type": "error", "detail": "Only node admin can delete files"}) + entry = ctx["index"].get_entry(file_id) + if not entry: + 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: + self._send({"type": "error", "detail": "No authorized key for deletion"}) + return + + challenge = os.urandom(32) + self._admin_challenges[file_id] = challenge + self._send({ + "type": MNP.ADMIN_CHALLENGE, + "v": MNP_VERSION, + "challenge": base64.b64encode(challenge).decode(), + "file_id": file_id, + }) + + def _do_admin_response(self, msg: dict) -> None: + file_id = msg.get("file_id", "") + sig_b64 = msg.get("signature", "") + + challenge = self._admin_challenges.pop(file_id, None) + if not challenge: + self._send({"type": "error", "detail": "No pending admin challenge"}) + return + + try: + sig_bytes = base64.b64decode(sig_b64) + except Exception: + self._send({"type": "error", "detail": "Invalid signature encoding"}) return + ctx = self._group_ctx() entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return + verified = False + + # Try admin key (locally pinned) + admin_pk = self._ctx.get("admin_pk_ed25519") + if admin_pk: + try: + admin_pk.verify(sig_bytes, challenge) + verified = True + except Exception: + pass + + # Try uploader key (stored at upload time) + if not verified and entry.uploader_pk: + try: + uploader_key = Ed25519PublicKey.from_public_bytes( + base64.b64decode(entry.uploader_pk)) + uploader_key.verify(sig_bytes, challenge) + verified = True + except Exception: + pass + + if not verified: + self._send({"type": "error", "detail": "Signature verification failed"}) + self._audit("admin_auth_failed", f"file_delete:{file_id[:16]}") + return + + self._exec_file_delete(ctx, file_id, entry) + + def _exec_file_delete(self, ctx: dict, file_id: str, entry) -> None: file_path = ctx["shared_root"] / entry.path / entry.name if file_path.exists(): file_path.unlink() |