""" MeshBay Node — WebRTC DataChannel server for browser clients. Browsers cannot use QUIC for NAT traversal (WebTransport doesn't allow choosing the UDP source port — Port-Restricted Cone NAT requires exact port matching). WebRTC DataChannel with ICE/STUN handles this automatically. The MNP protocol (handshake, file_request, file_chunk, chat, etc.) runs identically over WebRTC DataChannel as over QUIC streams. Same E2E encryption, same message types, same msgpack wire format. Wire format on the DataChannel: - Each message is length-prefixed msgpack (4-byte big-endian + msgpack payload) - Same as QUIC streams and TCP+TLS - DataChannel is ordered and reliable (SCTP over DTLS) Signaling flow (handled externally by the hub): Browser → Hub : POST /v1/nodes/{id}/webrtc/offer {sdp, ice_candidates} Hub → Node : WS push {type: "webrtc_offer", sdp, ice_candidates, peer_id} Node → Hub : WS push {type: "webrtc_answer", sdp, ice_candidates, peer_id} Hub → Browser : SSE/response {sdp, ice_candidates} After signaling, DataChannel is P2P — hub is out of the loop. """ import asyncio import base64 import hashlib import hmac import logging import os import struct from pathlib import Path from typing import Any import jwt import msgpack from aiortc import RTCPeerConnection, RTCSessionDescription, RTCDataChannel from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, Ed25519PublicKey, ) from meshbay_common import MNP_VERSION from meshbay_common.crypto import pk_to_b64 from meshbay_common.webcrypto import chunk_key_aes, encrypt_chunk_aes from meshbay_common.protocol import MNP from meshbay_node.indexer import GroupIndex log = logging.getLogger(__name__) 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"} async def _probe_video(path: str) -> tuple[str | None, float]: """Probe video file with ffprobe, return (MSE codec string, duration).""" import json as _json proc = await asyncio.create_subprocess_exec( "ffprobe", "-v", "error", "-show_entries", "stream=codec_name,profile,level,codec_type", "-show_entries", "format=duration", "-of", "json", path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, _ = await proc.communicate() info = _json.loads(stdout) duration = float(info.get("format", {}).get("duration", 0)) v_codec = a_codec = "" for s in info.get("streams", []): if s.get("codec_type") == "video" and not v_codec: cn = s.get("codec_name", "") if cn == "h264": p = _H264_PROFILES.get(s.get("profile", "High"), "64") lvl = int(s.get("level", 40)) v_codec = f"avc1.{p}00{lvl:02x}" elif cn == "hevc": v_codec = "hev1.1.6.L93.B0" elif cn == "vp9": v_codec = "vp09.00.10.08" elif cn == "av1": v_codec = "av01.0.01M.08" elif s.get("codec_type") == "audio" and not a_codec: cn = s.get("codec_name", "") if cn == "aac": a_codec = "mp4a.40.2" elif cn in ("mp3", "mp2"): a_codec = "mp4a.6b" elif cn == "opus": a_codec = "opus" elif cn == "ac3": a_codec = "ac-3" elif cn == "flac": a_codec = "flac" if not v_codec: return None, duration codec = f"{v_codec},{a_codec}" if a_codec else v_codec return codec, duration def _pack(obj: dict) -> bytes: data = msgpack.packb(obj, use_bin_type=True) return struct.pack(">I", len(data)) + data class _DataChannelBuffer: """Accumulate DataChannel messages and extract length-prefixed msgpack.""" def __init__(self): self._buf = bytearray() def feed(self, data: bytes): self._buf.extend(data) def messages(self): while len(self._buf) >= 4: length = struct.unpack(">I", self._buf[:4])[0] if length > MAX_MSG: raise ValueError(f"Message too large: {length}") if len(self._buf) < 4 + length: break msg_bytes = bytes(self._buf[4:4 + length]) del self._buf[:4 + length] yield msgpack.unpackb(msg_bytes, raw=False) def _get_remote_ip(pc: RTCPeerConnection) -> str: """Best-effort extraction of the remote peer IP from the ICE transport.""" try: dtls = pc.sctp and pc.sctp.transport ice = dtls and dtls.transport conn = ice and ice._connection if conn and hasattr(conn, '_nominated') and conn._nominated: for pair in conn._nominated.values(): return pair.remote_candidate.host if conn and conn.remote_candidates: return conn.remote_candidates[0].host except Exception: pass return "" class WebRTCPeerSession: """One WebRTC peer connection, handling MNP over a DataChannel.""" def __init__(self, pc: RTCPeerConnection, node_ctx: dict, peer_id: str = ""): self._pc = pc self._ctx = node_ctx self._channel: RTCDataChannel | None = None self._buffer = _DataChannelBuffer() self._user_id: str | None = None self._group_id: str | None = None 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 @channel.on("message") def on_message(message): if isinstance(message, str): message = message.encode() self._buffer.feed(message) for msg in self._buffer.messages(): self._handle_message(msg) def _handle_message(self, msg: dict) -> None: mtype = msg.get("type") log.debug("WebRTC recv: %s", mtype) 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: self._do_index_sync() elif mtype == MNP.FILE_REQUEST: self._do_file_request(msg) elif mtype == MNP.STREAM_SEGMENT: self._do_stream_segment(msg) elif mtype == MNP.CHAT_MESSAGE: self._do_chat_message(msg) elif mtype == MNP.CHAT_HISTORY: self._do_chat_history(msg) elif mtype == MNP.FILE_UPLOAD: 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: log.warning("Unknown MNP message type on DataChannel: %s", mtype) except Exception as e: log.error("Error handling %s on DataChannel: %s", mtype, e) self._send({"type": "error", "detail": str(e)}) def _audit(self, event: str, detail: str = "") -> None: audit = self._ctx.get("audit_store") if audit and self._user_id: if not self._remote_ip: self._remote_ip = _get_remote_ip(self._pc) asyncio.ensure_future(audit.log_event( user_id=self._user_id, event=event, ip=self._remote_ip, username=self._username, group_id=self._group_id or "", detail=detail, )) def _do_handshake(self, msg: dict) -> None: token = msg.get("token", "") group_id = msg.get("group_id", "") try: decoded = jwt.decode(token, self._ctx["hub_pk_pem"], algorithms=["EdDSA"]) except Exception as e: self._send({"type": "error", "detail": f"Invalid JWT: {e}"}) self._audit_auth_failed(group_id, str(e)) return denylist = self._ctx.get("denylist") if denylist and denylist.is_denied(decoded.get("sub", ""), decoded.get("jti", "")): self._send({"type": "error", "detail": "Token revoked"}) return if group_id and group_id not in decoded.get("groups", []): self._send({"type": "error", "detail": "Not a member of this group"}) return if group_id and "groups" in self._ctx and group_id not in self._ctx["groups"]: self._send({"type": "error", "detail": "Group not hosted on this node"}) return # 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], 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: self._remote_ip = _get_remote_ip(self._pc) asyncio.ensure_future(audit.log_event( user_id="unknown", event="auth_failed", ip=self._remote_ip, group_id=group_id, detail=reason, )) def _group_ctx(self) -> dict: if "groups" in self._ctx and self._group_id: return self._ctx["groups"][self._group_id] return self._ctx def _do_index_sync(self) -> None: ctx = self._group_ctx() idx = ctx["index"] entries = [ { "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 ] self._send({ "type": MNP.INDEX_SYNC, "v": MNP_VERSION, "group_id": idx.group_id, "version": idx.version, "entries": entries, }) def _do_file_request(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg["file_id"] chunk_index = msg["chunk_index"] entry = ctx["index"].get_entry(file_id) if not entry: log.warning("File not found: %s", file_id[:16]) self._send({"type": "error", "detail": "File not found"}) return file_path = ctx["shared_root"] / entry.path / entry.name if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return file_hash = bytes.fromhex(entry.id) chunk_data = _read_and_encrypt( self._ctx["sk_node"], ctx["gek"], file_path, chunk_index, file_hash, ) self._send(chunk_data) if chunk_index == 0: self._audit("file_download", entry.name) def _do_stream_segment(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg["file_id"] segment_index = msg["segment_index"] segment_duration = msg.get("segment_duration", 4) entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return file_path = ctx["shared_root"] / entry.path / entry.name if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return import subprocess try: result = subprocess.run( ["ffmpeg", "-hide_banner", "-loglevel", "error", "-ss", str(segment_index * segment_duration), "-i", str(file_path), "-t", str(segment_duration), "-c:v", "copy", "-c:a", "copy", "-f", "mpegts", "pipe:1"], capture_output=True, timeout=30, ) if result.returncode != 0 or not result.stdout: self._send({"type": "error", "detail": "Segment extraction failed"}) return segment_data = result.stdout except Exception: self._send({"type": "error", "detail": "Segment extraction failed"}) return self._send({ "type": MNP.STREAM_SEGMENT, "v": MNP_VERSION, "file_id": file_id, "segment_index": segment_index, "data_b64": base64.b64encode(segment_data).decode(), "size": len(segment_data), }) def _do_chat_message(self, msg: dict) -> None: chat_store = self._ctx.get("chat_store") payload = msg.get("payload", "") sender_name = msg.get("sender_name", "") if sender_name: self._ctx.setdefault("_user_names", {})[self._user_id] = sender_name if chat_store: raw = payload.encode() if isinstance(payload, str) else payload asyncio.ensure_future(chat_store.save_message( sender_id=self._user_id, iteration=msg.get("iteration", 0), payload=raw, thread_id=msg.get("thread_id"), sender_name=sender_name, )) peers = self._ctx.get("_peers", {}) broadcast = { "type": MNP.CHAT_MESSAGE, "v": MNP_VERSION, "sender_id": self._user_id, "sender_name": sender_name, "payload": payload, "thread_id": msg.get("thread_id"), "timestamp": __import__("time").time(), } for uid, session in list(peers.items()): if uid != self._user_id and session is not self: try: session._send(broadcast) except Exception: pass hub_ws = self._ctx.get("hub_ws") if hub_ws and self._group_id: try: import json as _json asyncio.ensure_future(hub_ws.send(_json.dumps({ "type": "chat_notify", "group_id": self._group_id, "sender_name": sender_name, }))) except Exception: pass self._send({"type": "ack", "v": MNP_VERSION}) self._audit("chat_message") def _do_chat_history(self, msg: dict) -> None: chat_store = self._ctx.get("chat_store") if not chat_store: self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "messages": [], }) return since = msg.get("since", 0) limit = msg.get("limit", 100) asyncio.ensure_future(self._send_chat_history(chat_store, since, limit)) async def _send_chat_history(self, chat_store, since: float, limit: int) -> None: msgs = await chat_store.get_messages(since=since, limit=limit) names = self._ctx.get("_user_names", {}) self._send({ "type": MNP.CHAT_HISTORY_RESPONSE, "v": MNP_VERSION, "messages": [ { "id": m.id, "sender_id": m.sender_id, "sender_name": m.sender_name or names.get(m.sender_id, ""), "payload": m.payload.decode("utf-8", errors="replace") if isinstance(m.payload, bytes) else m.payload, "timestamp": m.timestamp, "thread_id": m.thread_id, } for m in msgs ], }) def _do_file_upload(self, msg: dict) -> None: ctx = self._group_ctx() filename = msg.get("filename", "") chunk_index = msg.get("chunk_index", 0) total_chunks = msg.get("total_chunks", 1) data = msg.get("data") if not filename or data is None: self._send({"type": "error", "detail": "Missing filename or data"}) return shared_root = ctx.get("shared_root") if not shared_root: self._send({"type": "error", "detail": "No shared directory"}) return upload_dir = shared_root / ".uploads" upload_dir.mkdir(exist_ok=True) safe_name = filename.replace("/", "_").replace("\\", "_").replace("..", "_") tmp_path = upload_dir / f"{safe_name}.part" if isinstance(data, str): chunk_bytes = base64.b64decode(data) else: chunk_bytes = bytes(data) mode = "ab" if chunk_index > 0 else "wb" with open(tmp_path, mode) as f: f.write(chunk_bytes) self._send({ "type": MNP.FILE_UPLOAD_ACK, "v": MNP_VERSION, "chunk_index": chunk_index, "filename": filename, }) if chunk_index + 1 >= total_chunks: final_path = shared_root / safe_name 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() file_id = msg.get("file_id", "") if not file_id: self._send({"type": "error", "detail": "Missing file_id"}) return 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() log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) ctx["index"].remove_entry(file_id) self._send({ "type": MNP.FILE_DELETE_ACK, "v": MNP_VERSION, "file_id": file_id, }) async def _stream_video(self, msg: dict) -> None: """Stream a video file as fMP4 segments via MSE-compatible output.""" ctx = self._group_ctx() file_id = msg.get("file_id", "") entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) return file_path = ctx["shared_root"] / entry.path / entry.name if not file_path.exists(): self._send({"type": "error", "detail": "File not on disk"}) return gek = ctx.get("gek") file_hash = bytes.fromhex(entry.id) try: codec_str, duration = await _probe_video(str(file_path)) except Exception as e: self._send({"type": "error", "detail": f"Probe failed: {e}"}) return if not codec_str: self._send({"type": "error", "detail": "Unsupported video codec"}) return proc = await asyncio.create_subprocess_exec( "ffmpeg", "-hide_banner", "-loglevel", "error", "-i", str(file_path), "-c", "copy", "-movflags", "frag_keyframe+empty_moov+default_base_moof", "-f", "mp4", "pipe:1", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) self._send({ "type": MNP.STREAM_INIT, "v": MNP_VERSION, "file_id": file_id, "codec": codec_str, "duration": duration, }) index = 0 try: while True: data = await proc.stdout.read(STREAM_SEGMENT_SIZE) if not data: break ckey = chunk_key_aes(gek, file_hash, index) nonce, ct = encrypt_chunk_aes(ckey, data) self._send({ "type": MNP.STREAM_DATA, "v": MNP_VERSION, "file_id": file_id, "segment_index": index, "nonce": nonce, "ct": ct, "plaintext_size": len(data), }) index += 1 await asyncio.sleep(0) except Exception as e: log.error("Stream error: %s", e) finally: try: proc.kill() except ProcessLookupError: pass await proc.wait() self._send({ "type": MNP.STREAM_END, "v": MNP_VERSION, "file_id": file_id, }) log.info("Streamed %s: %d segments", entry.name, index) self._audit("stream_video", entry.name) def _send(self, obj: dict) -> None: if self._channel and self._channel.readyState == "open": self._channel.send(_pack(obj)) else: log.warning("WebRTC send skipped: channel=%s", self._channel.readyState if self._channel else "none") async def close(self) -> None: self._audit("disconnect") peers = self._ctx.get("_peers") if peers and self._user_id: peers.pop(self._user_id, None) await self._pc.close() def _read_and_encrypt( sk_node: Ed25519PrivateKey, gek: bytes, file_path: Path, chunk_index: int, file_hash: bytes, ) -> dict: with open(file_path, "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) ckey = chunk_key_aes(gek, file_hash, chunk_index) nonce, ct = encrypt_chunk_aes(ckey, plaintext) return { "type": MNP.FILE_CHUNK, "v": MNP_VERSION, "chunk_index": chunk_index, "plaintext_size": len(plaintext), "nonce": nonce, "ct": ct, } class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. Usage: transport = WebRTCTransport(sk_node, hub_pk_pem, gek, shared_root, index) answer_sdp = await transport.handle_offer(offer_sdp, peer_id) # Return answer_sdp to the browser via hub signaling """ def __init__( self, sk_node: Ed25519PrivateKey, hub_pk_pem: bytes, gek: bytes, shared_root: Path, index: GroupIndex, groups: dict[str, dict] | None = None, denylist: Any | None = None, stun_servers: list[str] | None = None, ): self._ctx: dict[str, Any] = { "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, "shared_root": shared_root, "index": index, "_peers": {}, } if groups: self._ctx["groups"] = groups if denylist: self._ctx["denylist"] = denylist self._stun = stun_servers or ["stun:stun.l.google.com:19302"] self._sessions: dict[str, WebRTCPeerSession] = {} async def handle_offer( self, offer_sdp: str, peer_id: str, ) -> tuple[str, list[dict]]: """ Process a WebRTC SDP offer from a browser client. Returns (answer_sdp, ice_candidates) to relay back via hub signaling. ICE candidates are embedded in the SDP (aiortc gathers before returning). """ from aiortc import RTCIceServer, RTCConfiguration config = RTCConfiguration( iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else [] ) pc = RTCPeerConnection(configuration=config) session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id) self._sessions[peer_id] = session @pc.on("datachannel") def on_datachannel(channel: RTCDataChannel): log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id) session._setup_channel(channel) @pc.on("connectionstatechange") async def on_state_change(): state = pc.connectionState log.info("WebRTC connection state: %s (peer=%s)", state, peer_id) if state in ("failed", "closed"): self._sessions.pop(peer_id, None) offer = RTCSessionDescription(sdp=offer_sdp, type="offer") await pc.setRemoteDescription(offer) answer = await pc.createAnswer() await pc.setLocalDescription(answer) log.info("WebRTC answer ready for peer=%s", peer_id) return pc.localDescription.sdp, [] async def close_peer(self, peer_id: str) -> None: session = self._sessions.pop(peer_id, None) if session: await session.close() async def close_all(self) -> None: for session in list(self._sessions.values()): await session.close() self._sessions.clear() @property def active_peers(self) -> int: return len(self._sessions)