1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
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))
|