diff options
6 files changed, 309 insertions, 3 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/config.py b/packages/meshbay-node/src/meshbay_node/config.py index b94d03a..0355615 100644 --- a/packages/meshbay-node/src/meshbay_node/config.py +++ b/packages/meshbay-node/src/meshbay_node/config.py @@ -65,7 +65,9 @@ transcode_incompatible_video = true # ice_interfaces = ["wlp0s20f3", "eth0"] # STUN servers for WebRTC ICE candidate gathering (NAT traversal). By default -# four public servers are used; set this to override. +# four public servers are used; set this to override. Every server in the list +# is queried in parallel each gather and the first answer wins, so one slow or +# blocked server no longer stalls the WebRTC answer. # stun_servers = ["stun:stun.l.google.com:19302", "stun:stun.cloudflare.com:3478"] # Browser and native clients reach this node over WebRTC DataChannel via hub diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 5c6d86c..2130440 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -476,6 +476,11 @@ class NodeDaemon: install_ice_filter( self._config.node.ice_interfaces or None, ) + # aiortc keeps only the FIRST entry of RTCConfiguration.iceServers, so + # the multi-STUN fallback only exists on the node if aioice itself + # fans out — see transport/stun_multi. + from meshbay_node.transport.stun_multi import install as install_stun_multi + install_stun_multi(self._config.node.stun_servers or None) first = next(iter(groups_ctx.values()), None) if WEBRTC_AVAILABLE: diff --git a/packages/meshbay-node/src/meshbay_node/ops.py b/packages/meshbay-node/src/meshbay_node/ops.py index cd5dc32..34a569f 100644 --- a/packages/meshbay-node/src/meshbay_node/ops.py +++ b/packages/meshbay-node/src/meshbay_node/ops.py @@ -877,6 +877,8 @@ async def set_node_settings(state: dict, settings: dict) -> dict: webrtc = state.get("webrtc") if webrtc and hasattr(webrtc, '_stun'): webrtc._stun = updated["stun_servers"] + from meshbay_node.transport.stun_multi import set_servers as _set_stun + _set_stun(updated["stun_servers"]) if "ice_interfaces" in updated: from meshbay_node.transport.ice_filter import install as install_ice_filter install_ice_filter(updated["ice_interfaces"] or None) diff --git a/packages/meshbay-node/src/meshbay_node/transport/stun_multi.py b/packages/meshbay-node/src/meshbay_node/transport/stun_multi.py new file mode 100644 index 0000000..5e3ec88 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/transport/stun_multi.py @@ -0,0 +1,148 @@ +""" +Make the node actually use every configured STUN server, not just the first. + +`aiortc` accepts a list in `RTCConfiguration.iceServers` but keeps **only the +first** STUN URI (`connection_kwargs`: *"only a single STUN server is +supported"*), and `aioice.ice.Connection` has a single `stun_server` field. So +the four defaults in `config.DEFAULT_STUN_SERVERS` — and anything the operator +adds on the Node page or with `meshbay-node stun add` — collapsed to +`stun:stun.l.google.com:19302` on the node side. When that one server was slow +or unreachable from the node's network, `get_component_candidates` burned its +full 5 s timeout with no server-reflexive candidate to show for it, adding +seconds to every browser connection. The "fallback" was configuration only. + +This module patches `aioice.ice.server_reflexive_candidate` — the same +monkey-patch technique `ice_filter.py` uses on `get_host_addresses` — so a +single ICE gather races the STUN binding request against **all** configured +servers on the one bound socket and takes the first answer. One reachable +server anywhere in the list now yields a reflexive candidate in one RTT instead +of a 5 s stall. + + install(["stun:a:3478", "stun:b:19302"]) → patch + set the list + set_servers([...]) → update the list live (CLI / panel) + set_servers([]) / install(None) → no fan-out; behave exactly like + upstream with whatever single + server aiortc passed +""" + +import asyncio +import logging +import socket + +import aioice.ice as _ice + +log = logging.getLogger(__name__) + +# Slightly under aioice's own `get_component_candidates(timeout=5)` so that when +# every server is unreachable we raise (→ no srflx candidate, same as today) +# just before the outer wait cancels us, rather than at the same instant. +_QUERY_TIMEOUT = 4.0 + +_servers: list[tuple[str, int]] = [] +_installed = False +_original = _ice.server_reflexive_candidate + + +def _parse(url: str) -> tuple[str, int]: + """`stun:host:port` / `stun:host` → (host, port). Default port 3478 (RFC 7064).""" + rest = url[5:] if url.startswith("stun:") else url + rest = rest.split("?", 1)[0].strip() # drop any ?transport=... tail + if not rest: + raise ValueError(f"empty STUN host in {url!r}") + host, sep, port = rest.rpartition(":") + if sep and host and port.isdigit(): + return host, int(port) + return rest, 3478 + + +async def _query_one(protocol, addr: tuple[str, int]): + # A fresh Message per call: StunProtocol.request asserts the transaction id + # is not already registered, and each Message gets a random id at build time. + request = _ice.stun.Message( + message_method=_ice.stun.Method.BINDING, + message_class=_ice.stun.Class.REQUEST, + ) + response, _addr = await protocol.request(request, addr) + return response + + +async def _first_ok(tasks: list[asyncio.Task]): + """First task that returns without raising wins; if all raise, re-raise the last.""" + pending = set(tasks) + last_exc: BaseException | None = None + while pending: + done, pending = await asyncio.wait( + pending, return_when=asyncio.FIRST_COMPLETED) + for task in done: + exc = task.exception() + if exc is None: + return task.result() + last_exc = exc + raise last_exc if last_exc is not None else RuntimeError("no STUN query ran") + + +async def _fanout_server_reflexive_candidate(protocol, stun_server): + """Drop-in for aioice.ice.server_reflexive_candidate that tries every server.""" + targets = list(_servers) if _servers else [tuple(stun_server)] + + loop = asyncio.get_event_loop() + + async def _resolve(host: str, port: int): + return (await loop.run_in_executor(None, socket.gethostbyname, host), port) + + resolved: list[tuple[str, int]] = [] + for res in await asyncio.gather( + *(_resolve(h, p) for h, p in targets), return_exceptions=True + ): + if isinstance(res, BaseException): + log.debug("STUN resolve failed: %s", res) + else: + resolved.append(res) + if not resolved: + raise OSError("no configured STUN server could be resolved") + + query_tasks = [asyncio.ensure_future(_query_one(protocol, a)) for a in resolved] + try: + response = await asyncio.wait_for(_first_ok(query_tasks), _QUERY_TIMEOUT) + finally: + for task in query_tasks: + task.cancel() + await asyncio.gather(*query_tasks, return_exceptions=True) + + local = protocol.local_candidate + return _ice.Candidate( + foundation=_ice.candidate_foundation("srflx", "udp", local.host), + component=local.component, + transport=local.transport, + priority=_ice.candidate_priority(local.component, "srflx"), + host=response.attributes["XOR-MAPPED-ADDRESS"][0], + port=response.attributes["XOR-MAPPED-ADDRESS"][1], + type="srflx", + related_address=local.host, + related_port=local.port, + ), None + + +def set_servers(stun_urls: list[str] | None) -> None: + """Replace the fan-out server list. Safe to call before or after install().""" + global _servers + parsed: list[tuple[str, int]] = [] + for url in stun_urls or []: + try: + parsed.append(_parse(str(url))) + except ValueError: + log.warning("Ignoring malformed STUN URL: %r", url) + _servers = parsed + if _installed: + log.info("STUN fan-out list: %s", + ", ".join(f"{h}:{p}" for h, p in parsed) or "(none)") + + +def install(stun_urls: list[str] | None) -> None: + """Monkey-patch aioice so an ICE gather queries every configured STUN server.""" + global _installed + set_servers(stun_urls) + if not _installed: + _ice.server_reflexive_candidate = _fanout_server_reflexive_candidate + _installed = True + log.info("STUN multi-server fan-out installed (%d servers)", len(_servers)) 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 717a27a..8402633 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -4631,6 +4631,10 @@ class WebRTCTransport: """ from aiortc import RTCIceServer, RTCConfiguration + # 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 [] ) @@ -4666,10 +4670,17 @@ class WebRTCTransport: offer = RTCSessionDescription(sdp=offer_sdp, type="offer") await pc.setRemoteDescription(offer) answer = await pc.createAnswer() + gather_start = time.monotonic() await pc.setLocalDescription(answer) - log.info("WebRTC answer ready for peer=%s", peer_id) - return pc.localDescription.sdp, [] + # 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") + log.info("WebRTC answer ready for peer=%s (ICE gather %.2fs, %d srflx)", + peer_id, time.monotonic() - gather_start, srflx) + return answer_sdp, [] async def close_peer(self, peer_id: str) -> None: session = self._sessions.pop(peer_id, None) diff --git a/packages/meshbay-node/tests/test_stun_multi.py b/packages/meshbay-node/tests/test_stun_multi.py new file mode 100644 index 0000000..119071d --- /dev/null +++ b/packages/meshbay-node/tests/test_stun_multi.py @@ -0,0 +1,138 @@ +""" +transport/stun_multi — the node must query every configured STUN server, not +just the first (which is all aiortc/aioice keep on their own). +""" + +import asyncio +import time + +import aioice.ice as _ice +import pytest +from meshbay_node.transport import stun_multi + + +@pytest.fixture(autouse=True) +def _restore(): + """Each test gets a clean, un-patched module.""" + saved = (_ice.server_reflexive_candidate, stun_multi._installed, + list(stun_multi._servers)) + stun_multi._installed = False + stun_multi._servers = [] + yield + (_ice.server_reflexive_candidate, stun_multi._installed, + stun_multi._servers) = saved[0], saved[1], saved[2] + + +@pytest.fixture +def _identity_dns(monkeypatch): + monkeypatch.setattr(stun_multi.socket, "gethostbyname", lambda host: host) + + +class _Local: + host, port, component, transport = "192.168.1.50", 54000, 1, "udp" + + +class _Response: + def __init__(self, ip, port): + self.attributes = {"XOR-MAPPED-ADDRESS": (ip, port)} + + +class _Protocol: + """A fake StunProtocol whose .request() behaviour is scripted per address.""" + + def __init__(self, behaviour): + self.local_candidate = _Local() + self._behaviour = behaviour # {(host, port): callable | value} + self.cancelled = 0 + + async def request(self, message, addr): + action = self._behaviour.get(addr) + try: + if action is None: + raise OSError(f"unreachable {addr}") + if callable(action): + return await action(addr) + return action, addr + except asyncio.CancelledError: + self.cancelled += 1 + raise + + +@pytest.mark.parametrize("url,expected", [ + ("stun:stun.l.google.com:19302", ("stun.l.google.com", 19302)), + ("stun:stun.cloudflare.com:3478", ("stun.cloudflare.com", 3478)), + ("stun:example.org", ("example.org", 3478)), + ("stun:example.org?transport=udp", ("example.org", 3478)), +]) +def test_parse(url, expected): + assert stun_multi._parse(url) == expected + + +def test_set_servers_parses_and_drops_empty(): + stun_multi.set_servers(["stun:a:1", "stun:b", "stun:", "stun:c:3478?transport=udp"]) + assert stun_multi._servers == [("a", 1), ("b", 3478), ("c", 3478)] + + +def test_install_patches_and_is_idempotent(): + stun_multi.install(["stun:a:3478"]) + assert _ice.server_reflexive_candidate is stun_multi._fanout_server_reflexive_candidate + first = _ice.server_reflexive_candidate + stun_multi.install(["stun:b:3478", "stun:c:3478"]) + assert _ice.server_reflexive_candidate is first # not re-wrapped + assert stun_multi._servers == [("b", 3478), ("c", 3478)] # list still updated + + +async def test_fanout_first_answer_wins(_identity_dns): + async def slow(addr): + await asyncio.sleep(30) + + proto = _Protocol({ + ("slow", 3478): slow, + ("fast", 19302): _Response("203.0.113.7", 5555), + # ("broken", 3478) -> not in map -> OSError + }) + stun_multi.set_servers(["stun:slow:3478", "stun:fast:19302", "stun:broken:3478"]) + + t0 = time.monotonic() + cand, extra = await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1)) + elapsed = time.monotonic() - t0 + + assert extra is None + assert cand.type == "srflx" + assert (cand.host, cand.port) == ("203.0.113.7", 5555) + assert cand.related_address == _Local.host + assert elapsed < 1.0 # did not wait on `slow` + await asyncio.sleep(0) # let cancellations propagate + assert proto.cancelled >= 1 # the slow query was torn down + + +async def test_all_servers_fail_raises_without_long_hang(_identity_dns, monkeypatch): + monkeypatch.setattr(stun_multi, "_QUERY_TIMEOUT", 0.3) + proto = _Protocol({}) # every addr -> OSError + stun_multi.set_servers(["stun:a:3478", "stun:b:3478"]) + + t0 = time.monotonic() + with pytest.raises(OSError): + await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1)) + assert time.monotonic() - t0 < 2.0 + + +async def test_empty_list_falls_back_to_passed_server(_identity_dns): + proto = _Protocol({("fallback.example", 3478): _Response("198.51.100.4", 9)}) + stun_multi.set_servers([]) # no fan-out configured + + cand, _ = await stun_multi._fanout_server_reflexive_candidate( + proto, ("fallback.example", 3478)) + assert (cand.host, cand.port) == ("198.51.100.4", 9) + + +async def test_unresolvable_names_raise(monkeypatch): + def boom(host): + raise OSError("name resolution failed") + + monkeypatch.setattr(stun_multi.socket, "gethostbyname", boom) + proto = _Protocol({}) + stun_multi.set_servers(["stun:a:3478", "stun:b:3478"]) + + with pytest.raises(OSError): + await stun_multi._fanout_server_reflexive_candidate(proto, ("x", 1)) |