""" 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 logging import time from typing import Any from aiortc import RTCDataChannel, RTCPeerConnection, RTCSessionDescription from cryptography.hazmat.primitives.asymmetric.ed25519 import ( Ed25519PrivateKey, ) from meshbay_node import transfers as transfers_mod from meshbay_node.indexer import GroupIndex # Re-imported under its original name: every call site and existing test in # this module still refers to it as `_probe_video`. The implementation lives # in media_probe.py so the indexer package (imported just above) can call it # too, for index-time enrichment, without a circular import. from meshbay_node.roots import ( RootSet, ) from meshbay_node.transport.webrtc.admin import AdminMixin from meshbay_node.transport.webrtc.admission import AdmissionMixin from meshbay_node.transport.webrtc.apps.music import MusicMixin from meshbay_node.transport.webrtc.apps.streaming import StreamingMixin from meshbay_node.transport.webrtc.apps.subtitles import SubtitlesMixin from meshbay_node.transport.webrtc.apps.video_meta import VideoMetaMixin from meshbay_node.transport.webrtc.blobs import BlobsMixin from meshbay_node.transport.webrtc.chat import ChatMixin from meshbay_node.transport.webrtc.core import _WEBRTC_TRACE, SessionCore from meshbay_node.transport.webrtc.dispatch import DispatchMixin from meshbay_node.transport.webrtc.files import FilesMixin from meshbay_node.transport.webrtc.group_ops import GroupOpsMixin from meshbay_node.transport.webrtc.handshake import HandshakeMixin from meshbay_node.transport.webrtc.node_ops import NodeOpsMixin from meshbay_node.transport.webrtc.transfer_handlers import TransferMixin from meshbay_node.transport.webrtc.upload_handlers import UploadMixin log = logging.getLogger(__name__) # How many peer connections this node holds at once, and how long one may stay # without completing the MNP handshake. The budget above bounds what *one* # unauthenticated peer costs; these bound how many there may be and how long # each lasts, which is the other half and was missing. The hub meters offers # per account (signaling.py) — a limit on each caller, not on this machine — # so the cost to an operator grew with the number of people in their groups. # Sized to be unreachable in ordinary use: a browser holds one connection per # open group, and a handshake unfinished after a minute is not going to finish. MAX_PEER_SESSIONS = 64 UNAUTHENTICATED_SESSION_TIMEOUT = 60 # seconds class WebRTCPeerSession( DispatchMixin, AdminMixin, AdmissionMixin, BlobsMixin, ChatMixin, FilesMixin, GroupOpsMixin, HandshakeMixin, NodeOpsMixin, TransferMixin, UploadMixin, StreamingMixin, VideoMetaMixin, MusicMixin, SubtitlesMixin, SessionCore, ): """One WebRTC peer connection, handling MNP over a DataChannel.""" class WebRTCTransport: """ Manages WebRTC peer connections for browser clients. Usage: transport = WebRTCTransport(sk_node, hub_pk_pem, gek, roots, 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, roots: RootSet, index: GroupIndex, groups: dict[str, dict] | None = None, denylist: Any | None = None, stun_servers: list[str] | None = None, max_concurrent_streams: int | None = None, max_concurrent_downloads: int | None = None, max_concurrent_uploads: int | None = None, max_upload_gb: float | None = None, transcode_incompatible_video: bool = True, ): self._ctx: dict[str, Any] = { "sk_node": sk_node, "hub_pk_pem": hub_pk_pem, "gek": gek, "roots": roots, "index": index, "_peers": {}, # None means "the operator said nothing" — the default applies. It # is read once, when the first stream builds the semaphore. "max_concurrent_streams": max_concurrent_streams, # Read once, when the first transfer builds the pools. None means # the operator said nothing and transfers.py's defaults apply. "max_concurrent_downloads": max_concurrent_downloads, "max_concurrent_uploads": max_concurrent_uploads, # The per-file upload ceiling, in GB. None means the operator said # nothing and MAX_UPLOAD_BYTES stands. "max_upload_gb": max_upload_gb, # Operator opt-out (node.toml) for the HEVC-etc. transcode # fallback in _stream_video_inner — real CPU cost, unlike copy. "transcode_incompatible_video": transcode_incompatible_video, } if groups: self._ctx["groups"] = groups if denylist: self._ctx["denylist"] = denylist from meshbay_node.config import DEFAULT_STUN_SERVERS self._stun = stun_servers or list(DEFAULT_STUN_SERVERS) self._sessions: dict[str, WebRTCPeerSession] = {} self._reapers: set[asyncio.Task] = set() def set_capacity(self, *, max_concurrent_streams: int | None = None, max_concurrent_downloads: int | None = None, max_concurrent_uploads: int | None = None, max_upload_gb: float | None = None) -> dict: """Resize a live pool without restarting the daemon. `ops.set_node_settings` used to do this by assigning `webrtc._stream_sem`, an attribute that has never existed — the pool is `ctx["_transcode_sem"]`, and `hasattr(webrtc, "_stream_sem")` is always False. So the hot-swap was a no-op and **`max_concurrent_streams` has never taken effect from the Node page without a restart**, contrary to docs/MESHBAY_DESIGN.md §6.8. This is the one implementation, on the object that owns the state, so the next two caps do not each grow their own copy of the mistake. What resizing means, stated because it is a decision and not a detail: **the new cap governs new streams; the ones already running are never interrupted.** A slot is held for the length of a film, so lowering the cap below what is in flight cannot take a viewer's film away — it stops the next one starting. The replacement pool is therefore created with the permits that remain (`new - in_flight`, floored at zero), not with a full set, or lowering the cap would briefly allow more viewers than either the old value or the new one. """ changed: dict = {} if max_concurrent_streams is not None: n = int(max_concurrent_streams) if n < 1: raise ValueError("max_concurrent_streams must be positive") before = self._ctx.get("max_concurrent_streams") self._ctx["max_concurrent_streams"] = n if self._ctx.get("_transcode_sem") is not None: in_flight = self._ctx.get("_streams_in_flight", 0) self._ctx["_transcode_sem"] = asyncio.Semaphore( max(0, n - in_flight)) log.info("stream: capacity %s -> %d (%d in flight, %d free now)", before, n, in_flight, max(0, n - in_flight)) else: # Nothing has streamed yet; the pool is built from this value on # first use, so there is nothing to resize. log.info("stream: capacity %s -> %d (no pool built yet)", before, n) changed["max_concurrent_streams"] = n pools = {} if max_concurrent_downloads is not None: pools[transfers_mod.DOWNLOAD] = int(max_concurrent_downloads) if max_concurrent_uploads is not None: pools[transfers_mod.UPLOAD] = int(max_concurrent_uploads) for key, value in pools.items(): if value < 1: raise ValueError(f"max_concurrent_{key}s must be positive") if pools: # Kept on the context whether or not a pool exists yet: the pools # are built on the first transfer, and would otherwise come up with # the defaults after an operator had already changed them. for key, value in pools.items(): self._ctx[f"max_concurrent_{key}s"] = value changed[f"max_concurrent_{key}s"] = value slots = self._ctx.get("_transfer_slots") if slots is not None: granted = slots.set_caps(node=pools) log.info("transfer: capacity now %s (%d started at once)", slots.summary(), len(granted)) # Raising a cap can start queued transfers immediately, and the # peers waiting on them have to be told: a grant nobody hears # about is the "stuck at waiting" report this design exists to # prevent. for lease in granted: self._notify_granted(lease) if max_upload_gb is not None: gb = float(max_upload_gb) if gb <= 0: raise ValueError("max_upload_gb must be greater than zero") self._ctx["max_upload_gb"] = gb changed["max_upload_gb"] = gb log.info("upload: per-file ceiling now %g GB", gb) return changed def _notify_granted(self, lease) -> None: """Tell the connection that owns `lease` it may start. On the transport rather than the session because a cap change has no session behind it — it arrives from the loopback API. """ groups = self._ctx.get("groups") registries = ([g.get("_peers", {}) for g in groups.values()] if groups else [self._ctx.get("_peers", {})]) for reg in registries: session = reg.get(lease.session_key) if session is not None: try: session._send( session._transfer_state_msg(lease, "granted")) except Exception: pass return 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 RTCConfiguration, RTCIceServer # aiortc keeps only the first STUN entry it sees here; the actual # multi-server fan-out is done by transport/stun_multi, which patches # aioice. The full list is still passed so a one-server deploy and the # tests that read `_stun` stay coherent. config = RTCConfiguration( iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else [] ) # Before anything is allocated. Every offer costs an RTCPeerConnection # with its own DTLS and SCTP stacks, and nothing here used to bound how # many a node would hold: the hub meters offers *per account* # (signaling.py), which is a limit on each caller and not on this # machine, so the cost # grew with the number of members in the group. An operator's node must # not be exhaustible by the people they invited. if len(self._sessions) >= MAX_PEER_SESSIONS: log.warning("Refusing WebRTC offer: %d peer sessions already open", len(self._sessions)) raise RuntimeError("Node is at its peer-connection limit") pc = RTCPeerConnection(configuration=config) session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id) self._sessions[peer_id] = session self._reap_if_unauthenticated(peer_id) @pc.on("datachannel") def on_datachannel(channel: RTCDataChannel): log.info("WebRTC DataChannel opened: %s (peer=%s)", channel.label, peer_id) session._setup_channel(channel) if _WEBRTC_TRACE: @pc.on("iceconnectionstatechange") def on_ice_state_change(): log.info("WebRTC ICE state: %s (peer=%s)", pc.iceConnectionState, peer_id) @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"): gone = self._sessions.pop(peer_id, None) if gone is not None: # Popping only forgets the session. Its stream went on # transcoding until the credit timeout — measured at 91s # after the connection closed — holding one of the node's # two slots the whole time. Closing the viewer, the tab or # the browser all arrive here, so this is the one place # that covers every way of walking away. # # And the group's peer set forgets it too, as close() does: # otherwise every later broadcast to the group is written # to a closed channel, and every reconnect leaves one more # dead session held until the node restarts. if gone._user_id: gone._unregister_peer() await gone.shutdown_tasks() offer = RTCSessionDescription(sdp=offer_sdp, type="offer") await pc.setRemoteDescription(offer) answer = await pc.createAnswer() gather_start = time.monotonic() await pc.setLocalDescription(answer) # ICE gathering runs inside setLocalDescription (non-trickle). A slow or # unreachable STUN server shows up here as seconds of wait and zero # srflx lines — the symptom the multi-server fan-out exists to prevent. answer_sdp = pc.localDescription.sdp srflx = answer_sdp.count(" typ srflx") # The host addresses this node put in the answer. When a peer reports # "DataChannel closed" the first question is whether the node offered # anything that peer could route to at all — on a NAT'd host or a VM the # only host candidate is an address no one else can reach, and the log # otherwise looks identical to a working connection. host_addrs: set[str] = set() for line in answer_sdp.splitlines(): if line.startswith("a=candidate:") and " typ host " in line: parts = line.split() if len(parts) > 5: host_addrs.add(parts[4]) log.info( "WebRTC answer ready for peer=%s (ICE gather %.2fs, host: %s, %d srflx)", peer_id, time.monotonic() - gather_start, ", ".join(sorted(host_addrs)) or "none", srflx) return answer_sdp, [] def _reap_if_unauthenticated(self, peer_id: str) -> None: """Close a session that never completes the handshake. A peer that connects and then says nothing is indistinguishable from a working one until it is asked to prove something, and it was never asked: `connectionstatechange` reaps a connection that *fails*, and one that succeeds and stays silent was held for the node's lifetime. That is the cheapest way to spend someone else's memory — no GEK, no token, no group, just an open connection. `_user_id` is set by the GEK proof (`_do_handshake_response`), so it is the one honest test of whether this peer ever became anybody. """ async def reap() -> None: try: await asyncio.sleep(UNAUTHENTICATED_SESSION_TIMEOUT) session = self._sessions.get(peer_id) if session is not None and not session._user_id: log.warning("Closing peer %s: no handshake within %ds", peer_id[:8], UNAUTHENTICATED_SESSION_TIMEOUT) await self.close_peer(peer_id) except asyncio.CancelledError: raise except Exception as e: log.warning("Reaping peer %s failed: %s", peer_id[:8], e) # Held in a set for the same reason every other task here is: asyncio # keeps only a weak reference, and a reaper collected mid-sleep reaps # nothing (see WebRTCPeerSession.__init__). task = asyncio.ensure_future(reap()) self._reapers.add(task) task.add_done_callback(self._reapers.discard) 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 task in list(self._reapers): task.cancel() self._reapers.clear() for session in list(self._sessions.values()): await session.close() self._sessions.clear() @property def active_peers(self) -> int: return len(self._sessions)