From ebc683446e8b071d7265bb5714dba23778487b00 Mon Sep 17 00:00:00 2001 From: Yash Kumar Saini Date: Fri, 10 Jul 2026 13:30:53 +0530 Subject: [PATCH 1/2] feat(webrtc): add UdpMux for shared-port WebRTC-Direct inbound dispatch (#1352) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements `_udp_mux.UdpMux` — a shared UDP DatagramProtocol that owns a single OS socket and demultiplexes concurrent inbound WebRTC-Direct dials: - Pre-ICE STUN BINDING REQUESTs routed by `USERNAME` ufrag prefix - Post-ICE DTLS/SCTP frames routed by remote (host, port) - `_MuxedTransport` adapter gives each aioice.Connection a fake transport backed by the shared socket, bypassing `_gather_candidates` entirely - `add_ice_connection()` creates an `aioice.Connection` with preset `local_username`/`local_password` and injects a mux-backed StunProtocol into its `_protocols` list without binding a new UDP socket This is the foundational primitive for a spec-aligned WebRTC-Direct listener (libp2p/specs#715) that advertises one fixed UDP port in its multiaddr. Follow-up: wire UdpMux into listener.py for the full v2 listener flow. --- libp2p/transport/webrtc/_udp_mux.py | 250 +++++++++++++++++ newsfragments/1352.feature.rst | 1 + tests/core/transport/webrtc/test_udp_mux.py | 287 ++++++++++++++++++++ 3 files changed, 538 insertions(+) create mode 100644 libp2p/transport/webrtc/_udp_mux.py create mode 100644 newsfragments/1352.feature.rst create mode 100644 tests/core/transport/webrtc/test_udp_mux.py diff --git a/libp2p/transport/webrtc/_udp_mux.py b/libp2p/transport/webrtc/_udp_mux.py new file mode 100644 index 000000000..72f1ce77b --- /dev/null +++ b/libp2p/transport/webrtc/_udp_mux.py @@ -0,0 +1,250 @@ +""" +UdpMux: shared UDP socket dispatcher for WebRTC-Direct inbound connections. + +Routes incoming datagrams to per-dial aioice.Connection instances: + - STUN packets: by local ufrag prefix in the USERNAME attribute (pre-ICE) + - non-STUN data: by remote (host, port) pair (post-ICE DTLS / SCTP) + +This is the foundational primitive for a spec-aligned WebRTC-Direct listener +that advertises a single fixed UDP port and demuxes concurrent inbound dials +without spinning up a new port per peer. + +Refs: libp2p/specs#715 (WebRTC-Direct v2), libp2p/py-libp2p#1352 (spike). +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Optional, Protocol + +try: + import aioice.ice as _ice + import aioice.stun as _stun + from aioice.candidate import Candidate + + _HAS_AIOICE = True +except ImportError: + _HAS_AIOICE = False + + +class _HasConnectionLost(Protocol): + def connection_lost(self, exc: Optional[Exception]) -> None: ... + def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: ... + +logger = logging.getLogger(__name__) + + +class _MuxedTransport: + """ + Fake DatagramTransport backed by the mux's real shared socket. + + Passed to aioice's StunProtocol so that aioice can send datagrams without + ever owning its own UDP socket. close() signals the protocol rather than + closing the shared socket. + """ + + def __init__( + self, + real_transport: asyncio.DatagramTransport, + local_addr: tuple[str, int], + ) -> None: + self._real = real_transport + self._local_addr = local_addr + # Set by UdpMux after the StunProtocol is constructed (avoids circular ref). + self._protocol: Optional[_HasConnectionLost] = None + + def sendto(self, data: bytes, addr: tuple[str, int]) -> None: + self._real.sendto(data, addr) + + def close(self) -> None: + # Let the protocol know the "transport" is gone so aioice's + # `await protocol.close()` can complete without blocking forever. + if self._protocol is not None: + self._protocol.connection_lost(None) + + def get_extra_info(self, key: str, default: object = None) -> object: + if key == "sockname": + return self._local_addr + return default + + +class UdpMux(asyncio.DatagramProtocol): + """ + Shared UDP socket for WebRTC-Direct inbound connections. + + Create one instance per listener port via :meth:`create`, then call + :meth:`add_ice_connection` for each inbound dial. After ICE completes, + call :meth:`register_addr` with the elected remote address so that + subsequent DTLS/SCTP datagrams from that peer are dispatched correctly. + + Example:: + + mux, port = await UdpMux.create("0.0.0.0", 0) + + # When a new STUN BINDING REQUEST arrives for ufrag "abc": + conn = mux.add_ice_connection("abc", "password123", host="0.0.0.0") + # ... feed candidates, call conn.connect(), await ICE completion ... + + # After ICE selects a candidate pair: + mux.register_addr(("203.0.113.5", 54321), conn._protocols[0]) + + # Teardown: + mux.unregister("abc") + mux.unregister_addr(("203.0.113.5", 54321)) + await mux.close() + """ + + def __init__(self) -> None: + self._transport: Optional[asyncio.DatagramTransport] = None + self._local_addr: Optional[tuple[str, int]] = None + # ufrag -> StunProtocol (pre-ICE STUN dispatch) + self._by_ufrag: dict[str, _HasConnectionLost] = {} + # (host, port) -> StunProtocol (post-ICE DTLS/SCTP dispatch) + self._by_addr: dict[tuple[str, int], _HasConnectionLost] = {} + + # ------------------------------------------------------------------ + # Factory + # ------------------------------------------------------------------ + + @classmethod + async def create(cls, host: str, port: int) -> tuple["UdpMux", int]: + """ + Bind a shared UDP socket on *host*:*port* (use ``port=0`` for OS choice). + Returns ``(mux, bound_port)``. + """ + loop = asyncio.get_event_loop() + mux = cls() + transport, _ = await loop.create_datagram_endpoint( + lambda: mux, local_addr=(host, port) + ) + # connection_made is called synchronously inside create_datagram_endpoint, + # so _transport and _local_addr are set by the time we return. + assert mux._local_addr is not None + return mux, mux._local_addr[1] + + # ------------------------------------------------------------------ + # asyncio.DatagramProtocol + # ------------------------------------------------------------------ + + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self._transport = transport # type: ignore[assignment] + addr = transport.get_extra_info("sockname") + self._local_addr = (addr[0], addr[1]) + + def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: + norm: tuple[str, int] = (addr[0], addr[1]) + try: + msg = _stun.parse_message(data) + username: str = msg.attributes.get("USERNAME", "") + ufrag = username.split(":")[0] + protocol = self._by_ufrag.get(ufrag) + if protocol is not None: + protocol.datagram_received(data, norm) + return + logger.debug("UdpMux: no handler for ufrag %r from %s", ufrag, norm) + except ValueError: + # Non-STUN (DTLS handshake, SCTP frames after ICE): route by addr. + protocol = self._by_addr.get(norm) + if protocol is not None: + protocol.datagram_received(data, norm) + return + logger.debug("UdpMux: no handler for non-STUN datagram from %s", norm) + + def error_received(self, exc: Exception) -> None: + logger.warning("UdpMux socket error: %s", exc) + + def connection_lost(self, exc: Optional[Exception]) -> None: + logger.debug("UdpMux connection lost: %s", exc) + + # ------------------------------------------------------------------ + # Registration + # ------------------------------------------------------------------ + + def register(self, ufrag: str, protocol: _HasConnectionLost) -> None: + """Route STUN packets with local ufrag *ufrag* to *protocol*.""" + self._by_ufrag[ufrag] = protocol + + def register_addr(self, addr: tuple[str, int], protocol: _HasConnectionLost) -> None: + """Route non-STUN packets from *addr* to *protocol* (call after ICE).""" + self._by_addr[addr] = protocol + + def unregister(self, ufrag: str) -> None: + self._by_ufrag.pop(ufrag, None) + + def unregister_addr(self, addr: tuple[str, int]) -> None: + self._by_addr.pop(addr, None) + + # ------------------------------------------------------------------ + # Connection factory + # ------------------------------------------------------------------ + + def add_ice_connection( + self, + local_username: str, + local_password: str, + *, + host: str, + ) -> "_ice.Connection": + """ + Create an ``aioice.Connection`` backed by this mux (no own UDP socket). + + The connection is pre-registered for *local_username* so that inbound + STUN connectivity checks are dispatched correctly before the caller + has a chance to set remote candidates. + + The caller must: + 1. Call ``conn.set_remote_candidates([...])`` with the dialer's candidates. + 2. Await ``conn.connect()`` to complete ICE negotiation. + 3. Call ``register_addr(remote_addr, conn._protocols[0])`` once ICE + selects a candidate pair so that post-ICE DTLS/SCTP is routed. + 4. Call ``unregister(local_username)`` and + ``unregister_addr(remote_addr)`` on teardown. + """ + if not _HAS_AIOICE: + raise RuntimeError("aioice is required (install py-libp2p[webrtc])") + assert self._transport is not None, "UdpMux not yet bound (call create() first)" + assert self._local_addr is not None + + conn = _ice.Connection( + ice_controlling=False, + local_username=local_username, + local_password=local_password, + ) + + muxed_transport = _MuxedTransport(self._transport, self._local_addr) + protocol = _ice.StunProtocol(conn) + # Wire the fake transport so aioice can send responses. + protocol.transport = muxed_transport # type: ignore[assignment] + # Back-reference so _MuxedTransport.close() can resolve protocol.close(). + # aioice types connection_lost(exc: Exception) but asyncio protocol is Optional. + muxed_transport._protocol = protocol # type: ignore[assignment] + + protocol.local_candidate = Candidate( + foundation=_ice.candidate_foundation("host", "udp", host), + component=1, + transport="udp", + priority=_ice.candidate_priority(1, "host"), + host=host, + port=self._local_addr[1], + type="host", + ) + + # Inject into aioice's internal protocol list, bypassing _gather_candidates. + conn._protocols.append(protocol) + self.register(local_username, protocol) # type: ignore[arg-type] + return conn + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + @property + def local_addr(self) -> Optional[tuple[str, int]]: + return self._local_addr + + async def close(self) -> None: + """Close the shared UDP socket.""" + if self._transport is not None: + self._transport.close() + self._transport = None diff --git a/newsfragments/1352.feature.rst b/newsfragments/1352.feature.rst new file mode 100644 index 000000000..bf88653a9 --- /dev/null +++ b/newsfragments/1352.feature.rst @@ -0,0 +1 @@ +Added ``libp2p.transport.webrtc._udp_mux.UdpMux`` — a shared UDP socket dispatcher for WebRTC-Direct inbound connections. Routes pre-ICE STUN datagrams by ``USERNAME`` ufrag prefix and post-ICE DTLS/SCTP frames by remote address, enabling a single fixed port to demultiplex concurrent inbound dials without spinning up a separate socket per peer (prerequisite for a spec-aligned WebRTC-Direct v2 listener, libp2p/specs#715). diff --git a/tests/core/transport/webrtc/test_udp_mux.py b/tests/core/transport/webrtc/test_udp_mux.py new file mode 100644 index 000000000..5fce44f3c --- /dev/null +++ b/tests/core/transport/webrtc/test_udp_mux.py @@ -0,0 +1,287 @@ +""" +Tests for libp2p.transport.webrtc._udp_mux.UdpMux. + +All tests are sync wrappers so they run outside trio_mode and don't +interfere with the project-wide trio backend. + +Coverage: + - STUN datagram is dispatched to the registered protocol by ufrag + - STUN with unknown ufrag is silently dropped (no crash) + - Non-STUN datagram is dispatched by remote addr (post-ICE path) + - Non-STUN from unknown addr is silently dropped + - _MuxedTransport.close() resolves protocol's __closed future + - add_ice_connection() registers the connection and sets up the candidate +""" + +from __future__ import annotations + +import asyncio + +import pytest + +try: + import aioice.stun as _stun + from aioice.ice import Connection as _Connection + + from libp2p.transport.webrtc._udp_mux import UdpMux, _MuxedTransport + + HAS_AIOICE = True +except ImportError: + HAS_AIOICE = False + +pytestmark = pytest.mark.skipif(not HAS_AIOICE, reason="aioice not installed") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _stun_binding_request(username: str) -> bytes: + """Craft a minimal STUN BINDING REQUEST with the given USERNAME.""" + msg = _stun.Message( + message_method=_stun.Method.BINDING, + message_class=_stun.Class.REQUEST, + ) + msg.attributes["USERNAME"] = username + return bytes(msg) + + +def _dtls_like_bytes() -> bytes: + """Return bytes that look like DTLS (non-STUN) — TLS record header.""" + # DTLS 1.2 record: content_type=22 (handshake), version=0xfeff, ... + return b"\x16\xfe\xff\x00\x01\x00\x00\x00\x00\x00\x00\x00\x05hello" + + +class _RecordingProtocol: + """Stand-in for aioice.ice.StunProtocol — records what it receives.""" + + def __init__(self) -> None: + self.received: list[tuple[bytes, tuple[str, int]]] = [] + + def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: + self.received.append((data, addr)) + + def connection_lost(self, exc: object) -> None: + pass + + +# --------------------------------------------------------------------------- +# STUN dispatch +# --------------------------------------------------------------------------- + + +class TestStunDispatch: + def test_known_ufrag_dispatches_to_protocol(self) -> None: + asyncio.run(self._known_ufrag()) + + async def _known_ufrag(self) -> None: + mux, port = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + mux.register("abc123", proto) + try: + data = _stun_binding_request("abc123:remote456") + mux.datagram_received(data, ("10.0.0.1", 12345)) + assert len(proto.received) == 1 + assert proto.received[0][1] == ("10.0.0.1", 12345) + finally: + await mux.close() + + def test_unknown_ufrag_is_dropped_silently(self) -> None: + asyncio.run(self._unknown_ufrag()) + + async def _unknown_ufrag(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + mux.register("abc123", proto) + try: + data = _stun_binding_request("different:remote") + mux.datagram_received(data, ("10.0.0.1", 12345)) + assert proto.received == [] + finally: + await mux.close() + + def test_unregister_stops_dispatch(self) -> None: + asyncio.run(self._unregister()) + + async def _unregister(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + mux.register("abc123", proto) + mux.unregister("abc123") + try: + data = _stun_binding_request("abc123:remote456") + mux.datagram_received(data, ("10.0.0.1", 12345)) + assert proto.received == [] + finally: + await mux.close() + + +# --------------------------------------------------------------------------- +# Addr dispatch (post-ICE non-STUN) +# --------------------------------------------------------------------------- + + +class TestAddrDispatch: + def test_known_addr_dispatches_to_protocol(self) -> None: + asyncio.run(self._known_addr()) + + async def _known_addr(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + remote = ("10.0.0.2", 54321) + mux.register_addr(remote, proto) + try: + data = _dtls_like_bytes() + mux.datagram_received(data, remote) + assert len(proto.received) == 1 + assert proto.received[0][0] == data + finally: + await mux.close() + + def test_unknown_addr_is_dropped_silently(self) -> None: + asyncio.run(self._unknown_addr()) + + async def _unknown_addr(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + mux.register_addr(("10.0.0.2", 54321), proto) + try: + data = _dtls_like_bytes() + mux.datagram_received(data, ("10.0.0.3", 9999)) + assert proto.received == [] + finally: + await mux.close() + + def test_unregister_addr_stops_dispatch(self) -> None: + asyncio.run(self._unregister_addr()) + + async def _unregister_addr(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + remote = ("10.0.0.2", 54321) + mux.register_addr(remote, proto) + mux.unregister_addr(remote) + try: + data = _dtls_like_bytes() + mux.datagram_received(data, remote) + assert proto.received == [] + finally: + await mux.close() + + +# --------------------------------------------------------------------------- +# _MuxedTransport +# --------------------------------------------------------------------------- + + +class TestMuxedTransport: + def test_sendto_delegates_to_real_transport(self) -> None: + asyncio.run(self._sendto()) + + async def _sendto(self) -> None: + mux, mux_port = await UdpMux.create("127.0.0.1", 0) + try: + # Open a real UDP socket to receive what the mux sends. + loop = asyncio.get_event_loop() + received: list[bytes] = [] + + class _Echo(asyncio.DatagramProtocol): + def datagram_received(self, data, addr): + received.append(data) + + server_transport, _ = await loop.create_datagram_endpoint( + _Echo, local_addr=("127.0.0.1", 0) + ) + server_addr = server_transport.get_extra_info("sockname")[:2] + try: + mt = _MuxedTransport(mux._transport, mux.local_addr) + mt.sendto(b"hello", server_addr) + await asyncio.sleep(0.05) # give event loop time to deliver + assert received == [b"hello"] + finally: + server_transport.close() + finally: + await mux.close() + + def test_get_extra_info_sockname(self) -> None: + asyncio.run(self._sockname()) + + async def _sockname(self) -> None: + mux, port = await UdpMux.create("127.0.0.1", 0) + try: + mt = _MuxedTransport(mux._transport, mux.local_addr) + assert mt.get_extra_info("sockname") == ("127.0.0.1", port) + assert mt.get_extra_info("unknown") is None + finally: + await mux.close() + + def test_close_calls_connection_lost_on_protocol(self) -> None: + asyncio.run(self._close()) + + async def _close(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + try: + proto = _RecordingProtocol() + mt = _MuxedTransport(mux._transport, mux.local_addr) + mt._protocol = proto + connection_lost_called = [] + proto.connection_lost = lambda exc: connection_lost_called.append(exc) + mt.close() + assert connection_lost_called == [None] + finally: + await mux.close() + + +# --------------------------------------------------------------------------- +# add_ice_connection +# --------------------------------------------------------------------------- + + +class TestAddIceConnection: + def test_connection_is_registered_and_has_candidate(self) -> None: + asyncio.run(self._add_conn()) + + async def _add_conn(self) -> None: + mux, port = await UdpMux.create("127.0.0.1", 0) + # RFC 5245: ufrag >= 4 chars, password >= 22 chars + ufrag = "myufrag1" + password = "mypassword1234567890ab" + try: + conn = mux.add_ice_connection( + ufrag, password, host="127.0.0.1" + ) + # Registered for STUN dispatch + assert ufrag in mux._by_ufrag + # Has one protocol with a local candidate pointing at the mux port + assert len(conn._protocols) == 1 + cand = conn._protocols[0].local_candidate + assert cand is not None + assert cand.port == port + assert cand.host == "127.0.0.1" + assert cand.transport == "udp" + # local_username / local_password set correctly + assert conn.local_username == ufrag + assert conn.local_password == password + finally: + await mux.close() + + def test_stun_for_connection_dispatches_to_its_protocol(self) -> None: + asyncio.run(self._stun_for_conn()) + + async def _stun_for_conn(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + # RFC 5245: ufrag >= 4 chars, password >= 22 chars + ufrag = "uf1x" + password = "pw1password1234567890ab" + try: + # Replace the protocol with a recorder to observe dispatch + conn = mux.add_ice_connection(ufrag, password, host="127.0.0.1") + recorder = _RecordingProtocol() + mux._by_ufrag[ufrag] = recorder + + data = _stun_binding_request(f"{ufrag}:remote_uf") + mux.datagram_received(data, ("192.168.1.1", 9000)) + assert len(recorder.received) == 1 + finally: + await mux.close() From e83c55c86dc8d11c26aa15f9b8660dc318351ef9 Mon Sep 17 00:00:00 2001 From: Yash Kumar Saini Date: Fri, 10 Jul 2026 14:14:23 +0530 Subject: [PATCH 2/2] fix(webrtc/udpmux): resolve CI lint failures - Remove Optional import, use X | None syntax (ruff pyupgrade) - Fix import order in try block: from before import, alphabetical - Add _udp_mux.py to pyrefly project_excludes (aioice not installed in lint tox env, same treatment as other webrtc-optional files) - Drop unused _Connection import and unused `conn` variable in tests --- libp2p/transport/webrtc/_udp_mux.py | 25 ++++++++++++--------- pyproject.toml | 1 + tests/core/transport/webrtc/test_udp_mux.py | 11 +++++---- 3 files changed, 20 insertions(+), 17 deletions(-) diff --git a/libp2p/transport/webrtc/_udp_mux.py b/libp2p/transport/webrtc/_udp_mux.py index 72f1ce77b..8a4475703 100644 --- a/libp2p/transport/webrtc/_udp_mux.py +++ b/libp2p/transport/webrtc/_udp_mux.py @@ -16,12 +16,12 @@ import asyncio import logging -from typing import Optional, Protocol +from typing import Protocol try: + from aioice.candidate import Candidate import aioice.ice as _ice import aioice.stun as _stun - from aioice.candidate import Candidate _HAS_AIOICE = True except ImportError: @@ -29,9 +29,10 @@ class _HasConnectionLost(Protocol): - def connection_lost(self, exc: Optional[Exception]) -> None: ... + def connection_lost(self, exc: Exception | None) -> None: ... def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: ... + logger = logging.getLogger(__name__) @@ -52,7 +53,7 @@ def __init__( self._real = real_transport self._local_addr = local_addr # Set by UdpMux after the StunProtocol is constructed (avoids circular ref). - self._protocol: Optional[_HasConnectionLost] = None + self._protocol: _HasConnectionLost | None = None def sendto(self, data: bytes, addr: tuple[str, int]) -> None: self._real.sendto(data, addr) @@ -96,8 +97,8 @@ class UdpMux(asyncio.DatagramProtocol): """ def __init__(self) -> None: - self._transport: Optional[asyncio.DatagramTransport] = None - self._local_addr: Optional[tuple[str, int]] = None + self._transport: asyncio.DatagramTransport | None = None + self._local_addr: tuple[str, int] | None = None # ufrag -> StunProtocol (pre-ICE STUN dispatch) self._by_ufrag: dict[str, _HasConnectionLost] = {} # (host, port) -> StunProtocol (post-ICE DTLS/SCTP dispatch) @@ -108,7 +109,7 @@ def __init__(self) -> None: # ------------------------------------------------------------------ @classmethod - async def create(cls, host: str, port: int) -> tuple["UdpMux", int]: + async def create(cls, host: str, port: int) -> tuple[UdpMux, int]: """ Bind a shared UDP socket on *host*:*port* (use ``port=0`` for OS choice). Returns ``(mux, bound_port)``. @@ -154,7 +155,7 @@ def datagram_received(self, data: bytes, addr: tuple[str, int]) -> None: def error_received(self, exc: Exception) -> None: logger.warning("UdpMux socket error: %s", exc) - def connection_lost(self, exc: Optional[Exception]) -> None: + def connection_lost(self, exc: Exception | None) -> None: logger.debug("UdpMux connection lost: %s", exc) # ------------------------------------------------------------------ @@ -165,7 +166,9 @@ def register(self, ufrag: str, protocol: _HasConnectionLost) -> None: """Route STUN packets with local ufrag *ufrag* to *protocol*.""" self._by_ufrag[ufrag] = protocol - def register_addr(self, addr: tuple[str, int], protocol: _HasConnectionLost) -> None: + def register_addr( + self, addr: tuple[str, int], protocol: _HasConnectionLost + ) -> None: """Route non-STUN packets from *addr* to *protocol* (call after ICE).""" self._by_addr[addr] = protocol @@ -185,7 +188,7 @@ def add_ice_connection( local_password: str, *, host: str, - ) -> "_ice.Connection": + ) -> _ice.Connection: """ Create an ``aioice.Connection`` backed by this mux (no own UDP socket). @@ -240,7 +243,7 @@ def add_ice_connection( # ------------------------------------------------------------------ @property - def local_addr(self) -> Optional[tuple[str, int]]: + def local_addr(self) -> tuple[str, int] | None: return self._local_addr async def close(self) -> None: diff --git a/pyproject.toml b/pyproject.toml index 585e91cfe..f84753631 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -329,6 +329,7 @@ project_excludes = [ "./tests/core/transport/webrtc", "./libp2p/transport/webrtc/_asyncio_bridge.py", "./libp2p/transport/webrtc/_aiortc_helpers.py", + "./libp2p/transport/webrtc/_udp_mux.py", "./libp2p/transport/webrtc/certificate.py", "./libp2p/transport/webrtc/transport.py", "./libp2p/transport/webrtc/listener.py", diff --git a/tests/core/transport/webrtc/test_udp_mux.py b/tests/core/transport/webrtc/test_udp_mux.py index 5fce44f3c..0b171be56 100644 --- a/tests/core/transport/webrtc/test_udp_mux.py +++ b/tests/core/transport/webrtc/test_udp_mux.py @@ -21,7 +21,6 @@ try: import aioice.stun as _stun - from aioice.ice import Connection as _Connection from libp2p.transport.webrtc._udp_mux import UdpMux, _MuxedTransport @@ -248,9 +247,7 @@ async def _add_conn(self) -> None: ufrag = "myufrag1" password = "mypassword1234567890ab" try: - conn = mux.add_ice_connection( - ufrag, password, host="127.0.0.1" - ) + conn = mux.add_ice_connection(ufrag, password, host="127.0.0.1") # Registered for STUN dispatch assert ufrag in mux._by_ufrag # Has one protocol with a local candidate pointing at the mux port @@ -275,8 +272,10 @@ async def _stun_for_conn(self) -> None: ufrag = "uf1x" password = "pw1password1234567890ab" try: - # Replace the protocol with a recorder to observe dispatch - conn = mux.add_ice_connection(ufrag, password, host="127.0.0.1") + # Register the connection so the ufrag is in the dispatch table, + # then replace it with a recorder to observe dispatch without + # triggering real aioice connectivity checks. + mux.add_ice_connection(ufrag, password, host="127.0.0.1") recorder = _RecordingProtocol() mux._by_ufrag[ufrag] = recorder