diff --git a/libp2p/security/noise/patterns.py b/libp2p/security/noise/patterns.py index c37046795..fbe334972 100644 --- a/libp2p/security/noise/patterns.py +++ b/libp2p/security/noise/patterns.py @@ -90,14 +90,20 @@ async def handshake_inbound(self, conn: IRawConnection) -> ISecureConn: @abstractmethod async def handshake_outbound( - self, conn: IRawConnection, remote_peer: ID + self, conn: IRawConnection, remote_peer: ID | None ) -> ISecureConn: """ Perform outbound handshake as initiator. + ``remote_peer=None`` is for initiators that do not know the + responder's identity up front (e.g. a WebRTC-Direct listener, which + is the Noise initiator per spec): the remote is still authenticated by + its signed handshake payload; only the peer-ID equality check is + skipped. + Args: conn: Raw connection to perform handshake on - remote_peer: Expected remote peer ID for verification + remote_peer: Expected remote peer ID for verification, or ``None`` Returns: ISecureConn: Established secure connection @@ -279,7 +285,7 @@ async def handshake_inbound(self, conn: IRawConnection) -> ISecureConn: ) async def handshake_outbound( - self, conn: IRawConnection, remote_peer: ID + self, conn: IRawConnection, remote_peer: ID | None ) -> ISecureConn: logger.debug(f"Noise XX handshake_outbound started to peer {remote_peer}") noise_state = self.create_noise_state(prologue=self.prologue) @@ -339,7 +345,7 @@ async def handshake_outbound( f"{remote_peer}" ) remote_peer_id_from_pubkey = ID.from_pubkey(peer_handshake_payload.id_pubkey) - if remote_peer_id_from_pubkey != remote_peer: + if remote_peer is not None and remote_peer_id_from_pubkey != remote_peer: raise PeerIDMismatchesPubkey( "peer id does not correspond to the received pubkey: " f"remote_peer={remote_peer}, " diff --git a/libp2p/transport/webrtc/_aiortc_helpers.py b/libp2p/transport/webrtc/_aiortc_helpers.py index 600bde62e..8d6648a24 100644 --- a/libp2p/transport/webrtc/_aiortc_helpers.py +++ b/libp2p/transport/webrtc/_aiortc_helpers.py @@ -27,6 +27,7 @@ from .exceptions import WebRTCStreamError if TYPE_CHECKING: + from ._udp_mux import UdpMux from .connection import WebRTCConnection logger = logging.getLogger(__name__) @@ -37,8 +38,9 @@ # Timeout for HTTP SDP exchange (seconds). _SDP_HTTP_TIMEOUT = 15.0 -# Bounds for the HTTP /sdp dev harness — defend against memory-amplification -# DoS while the harness exists (until the STUN-based listener lands, #1352). +# Bounds for the experimental HTTP /sdp harness (config.enable_sdp_http_harness, +# off by default; the spec path is the STUN listener) — defend against +# memory-amplification DoS whenever it is enabled. _MAX_SDP_BODY_SIZE = 32 * 1024 # 32 KiB; SDP offers are typically 1–4 KiB _MAX_HEADER_LINES = 64 _MAX_HEADER_BYTES = 8 * 1024 # 8 KiB total across all header lines @@ -83,7 +85,7 @@ async def create_peer_connection( pc = RTCPeerConnection(configuration=config) # Replace aiortc's auto-generated cert. Must use the mangled name — # aiortc reads only `self.__certificates`, which mangles to this. - pc._RTCPeerConnection__certificates = [rtc_cert] # type: ignore[attr-defined] + set_private_attr(pc, "_RTCPeerConnection__certificates", [rtc_cert]) return pc @@ -162,10 +164,17 @@ def get_remote_fingerprint(pc: RTCPeerConnection) -> bytes: :returns: 32-byte SHA-256 digest. :raises ValueError: If the remote certificate is not available. """ - dtls = getattr(pc, "_dtlsTransport", None) + # aiortc keeps one DTLS transport per SCTP association; for a + # data-channel-only PC that is ``pc.sctp.transport``. The peer cert + # lives on the pyOpenSSL connection once the handshake completed. + sctp = getattr(pc, "sctp", None) + dtls = getattr(sctp, "transport", None) if sctp is not None else None if dtls is None: raise ValueError("DTLS transport not available on peer connection") - remote_cert = getattr(dtls, "_remote_certificate", None) + ssl_conn = getattr(dtls, "_ssl", None) + if ssl_conn is None: + raise ValueError("DTLS handshake has not started") + remote_cert = ssl_conn.get_peer_certificate(as_cryptography=True) if remote_cert is None: raise ValueError("Remote DTLS certificate not available") # remote_cert is a cryptography x509.Certificate @@ -175,6 +184,122 @@ def get_remote_fingerprint(pc: RTCPeerConnection) -> bytes: return hashlib.sha256(der).digest() +def set_private_attr(obj: Any, name: str, value: Any) -> None: + """ + Overwrite a private (possibly name-mangled) library attribute. + + Writing a wrong — or upgraded-away — name would silently create a new + attribute the library never reads; asserting existence first turns that + silent no-op into a failure at the line that caused it, and fails loudly + on the aiortc/aioice upgrade that renames or drops the slot. It separates + "wrote to the wrong name" from "the value never reached the handshake" + (which only the downstream invariant tests would catch). + """ + assert hasattr(obj, name), f"{type(obj).__name__} has no attribute {name!r}" + setattr(obj, name, value) + + +# ------------------------------------------------------------------ +# Peer-connection shutdown +# ------------------------------------------------------------------ + + +def _abort_ice_transports(pc: RTCPeerConnection) -> None: + """Force-close the aioice UDP transports of a data-channel-only *pc*.""" + sctp = getattr(pc, "sctp", None) + dtls = getattr(sctp, "transport", None) if sctp is not None else None + ice = getattr(dtls, "transport", None) if dtls is not None else None + conn = getattr(ice, "_connection", None) if ice is not None else None + for proto in list(getattr(conn, "_protocols", [])): + abort = getattr(getattr(proto, "transport", None), "abort", None) + if abort is not None: + abort() + + +async def close_peer_connection(pc: RTCPeerConnection, timeout: float = 5.0) -> None: + """ + Close *pc*, working around a Windows proactor close hang. + + When a datagram send is still in flight as ``transport.close()`` runs + (typical: DTLS just queued its close_notify), CPython's + ``_ProactorDatagramTransport._loop_writing`` early-returns on + ``_conn_lost`` and the deferred ``connection_lost`` is never delivered, + so aioice's ``StunProtocol.close()`` awaits its closed-future forever + (observed deterministically on Windows CPython 3.12/3.13). On timeout, + ``abort()`` the lingering transports — ``_force_close`` delivers + ``connection_lost`` unconditionally — and let ``close()`` finish. + """ + task = asyncio.ensure_future(pc.close()) + try: + await asyncio.wait_for(asyncio.shield(task), timeout) + return + except asyncio.TimeoutError: + logger.debug("pc.close() stalled; aborting ICE transports") + _abort_ice_transports(pc) + try: + await asyncio.wait_for(task, timeout) + except asyncio.TimeoutError: + task.cancel() + logger.warning("pc.close() still stalled after transport abort") + + +# ------------------------------------------------------------------ +# UdpMux bridge +# ------------------------------------------------------------------ + + +def attach_muxed_connection(pc: RTCPeerConnection, mux: UdpMux, conn: Any) -> None: + """ + Run *pc*'s ICE over the :class:`UdpMux`-backed ``aioice.Connection`` *conn*. + + aiortc offers no injection point for an external ICE connection: the + gatherer builds its own ``aioice.Connection`` when the SCTP/DTLS/ICE + stack is created (synchronously, on the first ``createDataChannel``). + This swaps that connection for *conn* — created via + :meth:`UdpMux.add_ice_connection` — so the peer connection sends and + receives on the shared UDP port instead of binding its own. + + Call **after** ``createDataChannel`` and **before** any + ``set*Description``. Touches aiortc/aioice privates (validated against + aiortc 1.15 / aioice 0.10): + + - ``iceGatherer._connection`` / ``iceTransport._connection`` — what + ``getLocalParameters()`` / ``getLocalCandidates()`` / ``start()`` use; + - ``iceTransport._recv`` / ``_send`` — bound at construction to the + *original* connection's methods and used by ``RTCDtlsTransport``; + forgetting these leaves DTLS talking to the orphaned connection. + + The orphaned original connection never bound a socket (gathering is + triggered later, and *conn* has it marked done), so it is simply dropped. + ``RTCIceTransport.stop()`` closes *conn*, which the mux tolerates. On + ICE ``completed`` the nominated remote address is registered on the mux + (belt and braces — the mux already learns it from the peer's STUN checks); + on ``closed``/``failed`` the connection's ufrag and addresses are dropped. + """ + sctp = pc.sctp + if sctp is None: + raise ValueError("call pc.createDataChannel() before attach_muxed_connection") + ice = ( + sctp.transport.transport + ) # RTCSctpTransport -> RTCDtlsTransport -> RTCIceTransport + set_private_attr(ice.iceGatherer, "_connection", conn) + set_private_attr(ice, "_connection", conn) + set_private_attr(ice, "_recv", conn.recv) + set_private_attr(ice, "_send", conn.send) + + ufrag = conn.local_username + + @ice.on("statechange") # type: ignore[misc,untyped-decorator] + def _on_ice_state() -> None: + state = ice.state + if state == "completed": + pair = conn._nominated.get(1) + if pair is not None: + mux.register_addr(pair.remote_addr, pair.protocol) + elif state in ("closed", "failed"): + mux.unregister(ufrag) + + # ------------------------------------------------------------------ # Callback wiring # ------------------------------------------------------------------ @@ -222,7 +347,7 @@ async def _send_on_channel(channel_id: int, data: bytes) -> None: ch.send(data) async def _close_pc() -> None: - await pc.close() + await close_peer_connection(pc) conn._create_channel_cb = _create_channel conn._send_on_channel_cb = _send_on_channel @@ -354,6 +479,11 @@ def _on_noise_msg(message: str | bytes) -> None: recv_queue.put_nowait(data) async def send(data: bytes) -> None: + # The negotiated channel exists before SCTP is up; aiortc raises + # InvalidStateError on send() until it is "open". The Noise + # initiator (the listener) sends first, right after DTLS connects, + # so wait here rather than in every caller. + await _wait_channel_open(channel) channel.send(data) async def recv() -> bytes: diff --git a/libp2p/transport/webrtc/_udp_mux.py b/libp2p/transport/webrtc/_udp_mux.py index bcb64511a..32133fdd8 100644 --- a/libp2p/transport/webrtc/_udp_mux.py +++ b/libp2p/transport/webrtc/_udp_mux.py @@ -57,13 +57,20 @@ def __init__( self, real_transport: asyncio.DatagramTransport, local_addr: tuple[str, int], + mux: UdpMux | None = None, ) -> None: self._real = real_transport self._local_addr = local_addr + self._mux = mux # Set by UdpMux after the StunProtocol is constructed (avoids circular ref). self._protocol: _HasConnectionLost | None = None def sendto(self, data: bytes, addr: tuple[str, int]) -> None: + # Anything we send to (our own connectivity checks included) will + # answer from that address without a USERNAME, so learn the mapping + # on the way out. + if self._mux is not None and self._protocol is not None: + self._mux._learn_addr((addr[0], addr[1]), self._protocol) self._real.sendto(data, addr) def close(self) -> None: @@ -101,15 +108,21 @@ class UdpMux(asyncio.DatagramProtocol): await conn.add_remote_candidate(None) # end-of-candidates await conn.connect() # do NOT call gather_candidates() - # After ICE selects a candidate pair: - mux.register_addr(("203.0.113.5", 54321), conn._protocols[0]) + # The peer's address is learned from its STUN checks; register_addr() + # can add more (e.g. the nominated pair's remote address). - # Teardown: + # Teardown (also drops the addresses learned for "abc"): mux.unregister("abc") - mux.unregister_addr(("203.0.113.5", 54321)) await mux.close() + + To run a full aiortc ``RTCPeerConnection`` over a muxed connection use + :func:`libp2p.transport.webrtc._aiortc_helpers.attach_muxed_connection`. """ + # Bound on addresses learned per connection (a real peer has a handful + # of candidates; anything beyond that is noise or an attack). + _MAX_ADDRS_PER_PROTOCOL = 16 + def __init__(self) -> None: self._transport: asyncio.DatagramTransport | None = None self._local_addr: tuple[str, int] | None = None @@ -117,8 +130,11 @@ def __init__(self) -> None: self._by_ufrag: dict[str, _HasConnectionLost] = {} # (host, port) -> StunProtocol (post-ICE DTLS/SCTP dispatch) self._by_addr: dict[tuple[str, int], _HasConnectionLost] = {} - # Called for a STUN packet whose ufrag is not registered (see - # set_unknown_stun_handler); lets a listener observe first-contact dials. + # id(protocol) -> number of _by_addr entries pointing at it + self._addr_count: dict[int, int] = {} + # Called with the full USERNAME for a BINDING REQUEST whose local ufrag + # is not registered (see set_unknown_stun_handler); lets a listener + # observe first-contact dials. self._unknown_stun_handler: ( Callable[[str, bytes, tuple[str, int]], None] | None ) = None @@ -156,33 +172,96 @@ 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 + except ValueError: + # Non-STUN (DTLS handshake, SCTP frames after ICE): route by addr. + # aioice's StunProtocol re-parses, hits the same ValueError and + # forwards the bytes to the connection's data path. + self._route_by_addr(data, norm, stun_shaped=False) + return + except struct.error: + # STUN-shaped but malformed: aioice parses attributes with + # struct.unpack, which raises struct.error (NOT a ValueError + # subclass) on a short fixed-width attribute. StunProtocol only + # catches ValueError, so handing this to protocol.datagram_received + # would let the same struct.error escape the loop callback. Deliver + # straight to the ICE connection's data path instead (DTLS/SCTP + # discard garbage) so a crafted packet can neither raise here nor + # log-flood. + self._route_by_addr(data, norm, stun_shaped=True) + return + + # Only BINDING REQUESTs carry USERNAME (``local:remote``); responses + # and indications must be routed by address. + username: str = msg.attributes.get("USERNAME", "") + ufrag = username.split(":")[0] + protocol = self._by_ufrag.get(ufrag) if ufrag else None + if protocol is not None: + # Learn the peer address from its STUN checks so post-ICE DTLS/SCTP + # from that address routes even if it races ahead of our own ICE + # "completed" transition (the controlling dialer sends its DTLS + # ClientHello as soon as *it* nominates). Same approach as pion's + # UDPMux. A spoofed STUN with a known ufrag can only cause the + # attacker's bytes to be fed to that connection's DTLS, which + # discards them. + self._learn_addr(norm, protocol) + protocol.datagram_received(data, norm) + return + is_first_contact = ( + ufrag != "" + and msg.message_class == _stun.Class.REQUEST + and msg.message_method == _stun.Method.BINDING + ) + if is_first_contact and self._unknown_stun_handler is not None: # First contact: an inbound BINDING REQUEST for an unregistered # ufrag. A WebRTC-Direct listener registers a handler to create the # connection (add_ice_connection) and replay this datagram. - if self._unknown_stun_handler is not None: - self._unknown_stun_handler(ufrag, data, norm) - return - logger.debug("UdpMux: no handler for ufrag %r from %s", ufrag, norm) - except (ValueError, struct.error): - # Not valid STUN. aioice raises ValueError for non-STUN framing, but - # a STUN-shaped packet with a malformed/short fixed-width attribute - # makes its struct.unpack raise struct.error (NOT a ValueError - # subclass). Both mean "not usable STUN" and must fall through to - # addr-based routing rather than escape datagram_received — a remote - # peer must not be able to crash or log-flood the mux with crafted - # packets. - # 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 + self._unknown_stun_handler(username, data, norm) + return + protocol = self._by_addr.get(norm) + if protocol is not None: + protocol.datagram_received(data, norm) + return + logger.debug("UdpMux: no handler for STUN ufrag %r from %s", ufrag, norm) + + def _learn_addr(self, addr: tuple[str, int], protocol: _HasConnectionLost) -> None: + """ + Map *addr* -> *protocol* for non-STUN routing. + + Latest wins (like pion's UDPMux): a peer that redials from the same + (ip, port) with a new ufrag must reach its *new* connection, not the + stale one still being torn down. Learned addresses are capped per + protocol because inbound STUN is unauthenticated at this layer — a + flood of requests carrying a live ufrag from many source ports must + not grow the table without bound. + """ + current = self._by_addr.get(addr) + if current is protocol: + return + n = self._addr_count.get(id(protocol), 0) + if n >= self._MAX_ADDRS_PER_PROTOCOL: + return + if current is not None: + self._addr_count[id(current)] = self._addr_count.get(id(current), 1) - 1 + self._by_addr[addr] = protocol + self._addr_count[id(protocol)] = n + 1 + + def _route_by_addr( + self, data: bytes, norm: tuple[str, int], *, stun_shaped: bool + ) -> None: + protocol = self._by_addr.get(norm) + if protocol is None: logger.debug("UdpMux: no handler for non-STUN datagram from %s", norm) + return + if not stun_shaped: + protocol.datagram_received(data, norm) + return + # Bypass StunProtocol's parser (see datagram_received); StunProtocol + # exposes the owning aioice.Connection as ``receiver``. + receiver = getattr(protocol, "receiver", None) + if receiver is not None and hasattr(receiver, "data_received"): + receiver.data_received(data, 1) + else: + logger.debug("UdpMux: dropped malformed STUN-shaped datagram from %s", norm) def error_received(self, exc: Exception) -> None: logger.warning("UdpMux socket error: %s", exc) @@ -202,13 +281,26 @@ def register_addr( self, addr: tuple[str, int], protocol: _HasConnectionLost ) -> None: """Route non-STUN packets from *addr* to *protocol* (call after ICE).""" + current = self._by_addr.get(addr) + if current is not None and current is not protocol: + self._addr_count[id(current)] = self._addr_count.get(id(current), 1) - 1 + if current is not protocol: + self._addr_count[id(protocol)] = self._addr_count.get(id(protocol), 0) + 1 self._by_addr[addr] = protocol def unregister(self, ufrag: str) -> None: - self._by_ufrag.pop(ufrag, None) + """Drop *ufrag* and every address learned/registered for its protocol.""" + protocol = self._by_ufrag.pop(ufrag, None) + if protocol is None: + return + for addr in [a for a, p in self._by_addr.items() if p is protocol]: + del self._by_addr[addr] + self._addr_count.pop(id(protocol), None) def unregister_addr(self, addr: tuple[str, int]) -> None: - self._by_addr.pop(addr, None) + protocol = self._by_addr.pop(addr, None) + if protocol is not None: + self._addr_count[id(protocol)] = self._addr_count.get(id(protocol), 1) - 1 def set_unknown_stun_handler( self, handler: Callable[[str, bytes, tuple[str, int]], None] | None @@ -216,9 +308,11 @@ def set_unknown_stun_handler( """ Register a callback for STUN packets whose ufrag is not yet registered. - The callback receives ``(ufrag, data, addr)``. For WebRTC-Direct this is - how a listener observes the *first* inbound BINDING REQUEST from a new - dialer — it can create and register the connection via + The callback receives ``(username, data, addr)`` where ``username`` is + the full STUN ``USERNAME`` (``local_ufrag:remote_ufrag``) of an inbound + BINDING REQUEST whose local ufrag is unregistered. For WebRTC-Direct + this is how a listener observes the *first* packet from a new dialer — + it can create and register the connection via :meth:`add_ice_connection` and replay ``data``. Called from the asyncio event-loop thread; keep it non-blocking (e.g. hand off to a queue). Pass ``None`` to clear. @@ -250,10 +344,10 @@ def add_ice_connection( 1. For each of the dialer's candidates, ``await conn.add_remote_candidate(c)``, then ``await conn.add_remote_candidate(None)`` to signal end-of-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. + 3. (Optional) call ``register_addr(remote_addr, conn._protocols[0])`` + once ICE selects a candidate pair; the mux already learns the peer's + address from its STUN checks. + 4. Call ``unregister(local_username)`` on teardown. """ if not _HAS_AIOICE: raise RuntimeError("aioice is required (install py-libp2p[webrtc])") @@ -266,7 +360,7 @@ def add_ice_connection( local_password=local_password, ) - muxed_transport = _MuxedTransport(self._transport, self._local_addr) + muxed_transport = _MuxedTransport(self._transport, self._local_addr, self) protocol = _ice.StunProtocol(conn) # Wire the fake transport so aioice can send responses. protocol.transport = muxed_transport # type: ignore[assignment] @@ -286,11 +380,26 @@ def add_ice_connection( protocol.local_candidate = local_candidate # Inject into aioice's internal protocol list, bypassing gather_candidates - # (which would bind extra UDP sockets). Also mark gathering complete and - # expose the candidate so Connection.connect() proceeds instead of raising - # "Local candidates gathering was not performed". + # (which would bind extra UDP sockets). Mark gathering both *started* + # and *complete*: ``_end`` lets Connection.connect() proceed instead of + # raising "Local candidates gathering was not performed"; ``_start`` + # makes a later gather_candidates() call — which aiortc's + # RTCIceGatherer.gather() issues unconditionally from + # setLocalDescription — a no-op instead of binding fresh sockets and + # appending their candidates to ours. + # Guard the private-slot writes: a renamed/dropped aioice attribute + # would otherwise be a silent no-op (a new attribute aioice never + # reads) and only fail far downstream. + for attr in ( + "_protocols", + "_local_candidates", + "_local_candidates_start", + "_local_candidates_end", + ): + assert hasattr(conn, attr), f"aioice Connection has no {attr!r}" conn._protocols.append(protocol) conn._local_candidates = [local_candidate] + conn._local_candidates_start = True conn._local_candidates_end = True self.register(local_username, protocol) # type: ignore[arg-type] return conn @@ -312,4 +421,5 @@ async def close(self) -> None: # dispatched to a half-torn-down protocol. self._by_ufrag.clear() self._by_addr.clear() + self._addr_count.clear() self._unknown_stun_handler = None diff --git a/libp2p/transport/webrtc/config.py b/libp2p/transport/webrtc/config.py index 63b491726..0dc44bddd 100644 --- a/libp2p/transport/webrtc/config.py +++ b/libp2p/transport/webrtc/config.py @@ -64,6 +64,18 @@ class WebRTCTransportConfig: default_factory=lambda: ["stun:stun.l.google.com:19302"] ) + # ------------------------------------------------------------------ + # WebRTC-Direct signaling + # ------------------------------------------------------------------ + # The spec path needs no signaling: the listener infers the dialer's + # offer from its first STUN packet (one shared UDP port, v1/v2 ufrag + # dispatch). The HTTP ``POST /sdp`` harness predates that and is kept + # only as an experimental py<->py debugging aid: when enabled the + # listener also serves it on TCP (same port number) and our dialer + # uses it instead of the STUN path. Not interoperable with other + # implementations. + enable_sdp_http_harness: bool = False + def get_or_generate_certificate(self) -> WebRTCCertificate: """ Return the configured certificate or generate a new one. diff --git a/libp2p/transport/webrtc/listener.py b/libp2p/transport/webrtc/listener.py index cf2fa892f..f3ea0f214 100644 --- a/libp2p/transport/webrtc/listener.py +++ b/libp2p/transport/webrtc/listener.py @@ -1,16 +1,31 @@ """ WebRTC Direct listener. -Runs a lightweight HTTP signaling server on TCP (same port number as the -WebRTC UDP endpoint) that accepts SDP offers and returns answers. After -the SDP exchange each incoming connection completes ICE/DTLS, a Noise XX -handshake over data-channel 0, and then hands the fully-authenticated -:class:`WebRTCConnection` to the registered handler. +Spec path (default): one shared UDP socket (:class:`UdpMux`). A dialer's +first STUN BINDING REQUEST carries ``USERNAME = server_ufrag:client_ufrag``; +the version prefix on ``server_ufrag`` (``libp2p+webrtc+v1/`` / +``libp2p+webrtc+v2/``) selects the flow. The listener infers the dialer's +SDP offer from that packet (credentials + source address), answers over the +muxed ICE agent, completes DTLS (as DTLS server, without verifying the +dialer's fingerprint — it cannot know it yet), then runs the Noise XX +handshake over data-channel 0 as the Noise *initiator* and hands the +authenticated :class:`WebRTCConnection` to the handler on the trio side. + +Experimental harness (``config.enable_sdp_http_harness``): additionally +serves ``POST /sdp`` on TCP (same port number) for py↔py debugging. Not +interoperable with other implementations. Published multiaddr format:: /ip4//udp//webrtc-direct/certhash//p2p/ +Threading model: aiortc runs on the :class:`AsyncioBridge` thread; the Noise +handshake and the handler run on trio inside a listener-owned nursery +(started as a system task, like the TCP listener). The asyncio → trio hop is +``trio.from_thread.run_sync(nursery.start_soon, ...)`` — non-blocking, so +the asyncio loop keeps servicing the data-channel callbacks the handshake +needs. + Spec: https://github.com/libp2p/specs/blob/master/webrtc/webrtc-direct.md """ @@ -22,6 +37,7 @@ from typing import TYPE_CHECKING, Any from multiaddr import Multiaddr +import trio from libp2p.abc import IListener from libp2p.crypto.keys import PrivateKey @@ -36,9 +52,11 @@ build_webrtc_direct_multiaddr, parse_webrtc_direct_multiaddr, ) +from .sdp import build_inferred_offer, parse_direct_username if TYPE_CHECKING: from ._asyncio_bridge import AsyncioBridge + from ._udp_mux import UdpMux logger = logging.getLogger(__name__) @@ -70,14 +88,32 @@ def __init__( self._closed = False self._signaling_server: asyncio.Server | None = None self._bridge: AsyncioBridge | None = None + self._mux: UdpMux | None = None + self._rtc_cert: Any = None + # Host used for the muxed connections' single host candidate. + self._candidate_host = "127.0.0.1" + + # trio side: long-lived nursery for inbound handshakes/handlers. + self._trio_token: trio.lowlevel.TrioToken | None = None + self._nursery: trio.Nursery | None = None + self._nursery_ready = trio.Event() + self._nursery_done = trio.Event() + # asyncio tasks driving inbound PCs (spec path); cancelled on close. + self._accept_tasks: set[asyncio.Future[Any]] = set() + # Bound the number of not-yet-authenticated inbound connections. + # ponytail: soft cap — updated from both the asyncio and trio threads + # (GIL-atomic int ops); use a lock if it ever needs to be exact. + self._in_flight = 0 async def listen(self, maddr: Multiaddr) -> None: """ Start listening for incoming WebRTC Direct connections. - Starts an HTTP signaling server on TCP that accepts SDP offers. - The published multiaddr advertises the same port on UDP (for - WebRTC data channels) and includes the DTLS certificate hash. + Binds one shared UDP socket and dispatches inbound STUN by ufrag (spec + path). If ``config.enable_sdp_http_harness`` is set, also starts the + experimental HTTP ``POST /sdp`` signaling server on TCP with the same + port number. The published multiaddr advertises the UDP port and the + DTLS certificate hash. :param maddr: A ``/webrtc-direct`` multiaddr. :raises WebRTCConnectionError: If binding fails. @@ -91,28 +127,39 @@ async def listen(self, maddr: Multiaddr) -> None: raise WebRTCConnectionError( "WebRTC certificate was not generated via aiortc" ) + self._rtc_cert = rtc_cert - from ._aiortc_helpers import run_signaling_server + from ._udp_mux import UdpMux - # Start HTTP signaling server on asyncio thread. - # Binds TCP on the same port as the WebRTC UDP endpoint. - self._signaling_server = await bridge.run_coro( - run_signaling_server( - host=host if host != "0.0.0.0" else "0.0.0.0", - port=port, - on_offer=self._make_offer_handler(bridge, rtc_cert), + # Shared UDP socket + STUN dispatch (runs on the asyncio thread). + # Bind before spawning anything so a bind failure leaves no orphans. + try: + mux, bound_port = await bridge.run_coro(UdpMux.create(host, port)) + except OSError as e: + raise WebRTCConnectionError( + f"Failed to bind WebRTC Direct listener on {host}:{port}: {e}" + ) from e + self._mux = mux + await self._start_trio_nursery() + advertised_host = host if host != "0.0.0.0" else "127.0.0.1" + self._candidate_host = advertised_host + mux.set_unknown_stun_handler(self._on_unknown_stun) + + if self._config.enable_sdp_http_harness: + from ._aiortc_helpers import run_signaling_server + + # Experimental py<->py harness: TCP, same port number as the UDP + # socket so one multiaddr describes both. + self._signaling_server = await bridge.run_coro( + run_signaling_server( + host=host, + port=bound_port, + on_offer=self._make_offer_handler(bridge, rtc_cert), + ) ) - ) - - # Determine the actual bound port (if port was 0). - bound_port = port - if self._signaling_server.sockets: - sock = self._signaling_server.sockets[0] - bound_port = sock.getsockname()[1] # Build advertised multiaddr with certhash and peer ID. certhash_mb = self._certificate.fingerprint_to_multibase() - advertised_host = host if host != "0.0.0.0" else "127.0.0.1" advertised = build_webrtc_direct_multiaddr( host=advertised_host, port=bound_port, @@ -122,6 +169,135 @@ async def listen(self, maddr: Multiaddr) -> None: self._listening_addrs.append(advertised) logger.info("WebRTC Direct listener on %s", advertised) + # ------------------------------------------------------------------ + # Spec path: STUN first contact -> inferred offer -> muxed PC + # ------------------------------------------------------------------ + + def _on_unknown_stun( + self, username: str, data: bytes, addr: tuple[str, int] + ) -> None: + """ + First inbound BINDING REQUEST from a new dialer (asyncio thread, sync). + + Validates the ``USERNAME``, registers a mux-backed ICE connection for + the server ufrag (so every following packet routes without coming + back here), replays the packet into it, and schedules the async + acceptance. Anything malformed is dropped silently — this is an + unauthenticated network input. + """ + mux, bridge = self._mux, self._bridge + if self._closed or mux is None or bridge is None: + return + # ponytail: in-flight cap only; add a per-source-IP token bucket if + # STUN floods from one address become a problem (spec: SHOULD rate-limit). + if self._in_flight >= self._config.max_in_flight_connections: + logger.debug("WebRTC Direct: in-flight cap reached, dropping %s", addr) + return + try: + version, server_ufrag, _client_ufrag = parse_direct_username(username) + if version != 1: + logger.debug( + "WebRTC Direct: v%d not supported yet, dropping %s", version, addr + ) + return + # v1: the dialer set ufrag == pwd on both offer and answer, so the + # server credential *is* the password. + conn = mux.add_ice_connection( + server_ufrag, server_ufrag, host=self._candidate_host + ) + except (WebRTCConnectionError, ValueError): + logger.debug("WebRTC Direct: rejected first contact from %s", addr) + return + + self._in_flight += 1 + # Re-dispatch: the ufrag is registered now, so this reaches the aioice + # protocol (which answers the check and queues it) and teaches the + # mux the peer's address. + mux.datagram_received(data, addr) + task = asyncio.ensure_future(self._accept_v1(conn, server_ufrag, addr)) + self._accept_tasks.add(task) + task.add_done_callback(self._accept_tasks.discard) + + async def _accept_v1( + self, conn: Any, server_ufrag: str, addr: tuple[str, int] + ) -> None: + """Build the muxed PC for a v1 first contact and drive it to connected.""" + from aiortc import RTCSessionDescription + + from ._aiortc_helpers import ( + attach_muxed_connection, + close_peer_connection, + create_noise_channel, + create_peer_connection, + make_noise_channel_callbacks, + set_private_attr, + ) + + mux, bridge = self._mux, self._bridge + assert mux is not None and bridge is not None + pc: Any = None + try: + # The listener never needs STUN servers: it is reachable on the + # advertised address by construction. + pc = await create_peer_connection(self._rtc_cert, ice_servers=[]) + noise_ch = await create_noise_channel(pc) + noise_send, noise_recv, _ = make_noise_channel_callbacks(noise_ch) + attach_muxed_connection(pc, mux, conn) + # Spec step 6.2/7: B cannot know A's DTLS fingerprint, so it must + # not verify it during DTLS; the Noise handshake authenticates. + set_private_attr( + pc.sctp.transport, "_validate_peer_identity", lambda _params: None + ) + + offer_sdp = build_inferred_offer( + client_ufrag=server_ufrag, + client_pwd=server_ufrag, + remote_host=addr[0], + remote_port=addr[1], + max_message_size=self._config.max_message_size, + ) + await pc.setRemoteDescription( + RTCSessionDescription(sdp=offer_sdp, type="offer") + ) + answer = await pc.createAnswer() + await pc.setLocalDescription(answer) + except BaseException: + self._in_flight -= 1 + logger.debug("WebRTC Direct: inbound setup failed", exc_info=True) + mux.unregister(server_ufrag) + if pc is not None: + try: + await close_peer_connection(pc) + except Exception: + pass + return + + await self._complete_inbound(pc, bridge, noise_send, noise_recv) + + async def _start_trio_nursery(self) -> None: + """ + Open a listener-owned nursery in a trio system task. + + ``IListener.listen`` takes no nursery, so (like the TCP listener) we + spawn a system task that holds one open until :meth:`close`. + """ + if self._nursery is not None: + return + self._trio_token = trio.lowlevel.current_trio_token() + + async def _run() -> None: + try: + async with trio.open_nursery() as nursery: + self._nursery = nursery + self._nursery_ready.set() + await trio.sleep_forever() + finally: + self._nursery = None + self._nursery_done.set() + + trio.lowlevel.spawn_system_task(_run) + await self._nursery_ready.wait() + def _make_offer_handler( self, bridge: AsyncioBridge, @@ -138,17 +314,27 @@ async def _handle_offer(offer_sdp: str) -> str: make_noise_channel_callbacks, ) - # Create PC, set remote (offer), create answer. - pc = await create_peer_connection(rtc_cert) - noise_ch = await create_noise_channel(pc) - noise_send, noise_recv, _ = make_noise_channel_callbacks(noise_ch) + if self._in_flight >= self._config.max_in_flight_connections: + raise WebRTCConnectionError( + "Too many in-flight inbound WebRTC connections" + ) + self._in_flight += 1 - offer = RTCSessionDescription(sdp=offer_sdp, type="offer") - await pc.setRemoteDescription(offer) + try: + # Create PC, set remote (offer), create answer. + pc = await create_peer_connection(rtc_cert) + noise_ch = await create_noise_channel(pc) + noise_send, noise_recv, _ = make_noise_channel_callbacks(noise_ch) - answer = await pc.createAnswer() - await pc.setLocalDescription(answer) - answer_sdp = pc.localDescription.sdp + offer = RTCSessionDescription(sdp=offer_sdp, type="offer") + await pc.setRemoteDescription(offer) + + answer = await pc.createAnswer() + await pc.setLocalDescription(answer) + answer_sdp = pc.localDescription.sdp + except BaseException: + self._in_flight -= 1 + raise # Spawn background task to complete the connection after ICE. asyncio.ensure_future( @@ -169,15 +355,62 @@ async def _complete_inbound( """ Finish an inbound connection after the SDP answer has been sent. - Runs on the asyncio thread. Waits for ICE, runs the Noise - handshake (via trio), and hands the connection to the handler. + Runs on the asyncio thread: waits for ICE/DTLS, reads the dialer's + DTLS fingerprint, then hands off to :meth:`_finish_inbound` on trio + without blocking the loop. """ try: - from ._aiortc_helpers import wait_for_connected, wire_pc_to_connection + from ._aiortc_helpers import ( + close_peer_connection, + get_remote_fingerprint, + wait_for_connected, + ) + + await wait_for_connected(pc, timeout=self._config.handshake_timeout) + dialer_fp = get_remote_fingerprint(pc) + + nursery, token = self._nursery, self._trio_token + if self._closed or nursery is None or token is None: + raise WebRTCConnectionError("listener closed") + trio.from_thread.run_sync( + nursery.start_soon, + self._finish_inbound, + pc, + bridge, + dialer_fp, + noise_send, + noise_recv, + trio_token=token, + ) + except BaseException: + self._in_flight -= 1 + logger.debug("Failed to complete inbound WebRTC connection", exc_info=True) + try: + await close_peer_connection(pc) + except Exception: + pass + + async def _finish_inbound( + self, + pc: Any, + bridge: AsyncioBridge, + dialer_fp: bytes, + noise_send: Any, + noise_recv: Any, + ) -> None: + """ + Trio side of an inbound connection: Noise handshake, then handler. + + Must run on trio: :class:`WebRTCConnection` captures the trio token + in its constructor, and the Noise pattern code is trio-async. + """ + from libp2p.crypto.x25519 import create_new_key_pair as create_x25519_keypair - await wait_for_connected(pc) + from ._aiortc_helpers import close_peer_connection, wire_pc_to_connection + from .noise_handshake import DataChannelReadWriter, perform_noise_handshake - # Build WebRTCConnection (on trio side via bridge). + conn: WebRTCConnection | None = None + try: conn = WebRTCConnection( peer_id=ID(b"\x00" * 32), # updated after Noise bridge=bridge, @@ -186,18 +419,6 @@ async def _complete_inbound( ) wire_pc_to_connection(pc, conn) - # Noise handshake must run on the trio side. - from libp2p.crypto.x25519 import ( - create_new_key_pair as create_x25519_keypair, - ) - - from .noise_handshake import ( - DataChannelReadWriter, - perform_noise_handshake, - ) - - noise_kp = create_x25519_keypair() - async def _trio_noise_send(data: bytes) -> None: await bridge.run_coro(noise_send(data)) @@ -207,53 +428,45 @@ async def _trio_noise_recv() -> bytes: noise_rw = DataChannelReadWriter( send_cb=_trio_noise_send, recv_cb=_trio_noise_recv, - is_initiator=False, + is_initiator=True, ) - - # perform_noise_handshake is a trio function; schedule it - # on the trio thread. - def _run_noise_and_handler() -> None: - # This runs on the trio thread via trio.from_thread. - import trio as _trio - - async def _inner() -> None: - authenticated_peer = await perform_noise_handshake( - conn=noise_rw, - local_peer=self._local_peer_id, - libp2p_privkey=self._private_key, - noise_static_key=noise_kp.private_key, - local_fingerprint=self._certificate.fingerprint, - remote_fingerprint=b"\x00" * 32, # TODO: extract from PC - is_initiator=False, - ) - conn.peer_id = authenticated_peer - await conn.start() - logger.info( - "Inbound WebRTC connection from %s", - authenticated_peer, - ) - await self._handler(conn) - - _trio.from_thread.run_sync( - lambda: None # placeholder — full wiring in follow-up + with trio.fail_after(self._config.handshake_timeout): + # Server = Noise initiator (spec); we do not know the dialer's + # peer ID up front, so remote_peer=None. + authenticated_peer = await perform_noise_handshake( + conn=noise_rw, + local_peer=self._local_peer_id, + libp2p_privkey=self._private_key, + noise_static_key=create_x25519_keypair().private_key, + dialer_fingerprint=dialer_fp, + server_fingerprint=self._certificate.fingerprint, + is_initiator=True, + remote_peer=None, ) + conn.peer_id = authenticated_peer + await conn.start() + self._in_flight -= 1 + logger.info("Inbound WebRTC connection from %s", authenticated_peer) + except BaseException: + self._in_flight -= 1 + logger.debug("Inbound WebRTC handshake failed", exc_info=True) + try: + if conn is not None: + await conn.close() + else: + await bridge.run_coro(close_peer_connection(pc)) + except Exception: + pass + return - # For now, log that the inbound connection flow reached this point. - # Full trio-side Noise handshake wiring requires a TrioToken and - # careful cross-thread coordination that will be completed when - # the loopback integration test validates the full path. - logger.info( - "Inbound WebRTC connection: ICE connected, " - "Noise handshake pending (full wiring in integration test)" - ) - + # The nursery lives in a trio system task: an exception escaping here + # would take down the whole trio run, not just this connection. + try: + await self._handler(conn) except Exception: - logger.debug( - "Failed to complete inbound WebRTC connection", - exc_info=True, - ) + logger.exception("WebRTC Direct handler failed for %s", conn.peer_id) try: - await pc.close() + await conn.close() except Exception: pass @@ -274,6 +487,28 @@ async def close(self) -> None: logger.debug("Error closing signaling server", exc_info=True) self._signaling_server = None + if self._mux is not None and self._bridge is not None: + mux, self._mux = self._mux, None + mux.set_unknown_stun_handler(None) + + async def _shutdown_mux() -> None: + # Cancel in-flight inbound setups so their PCs close now + # rather than after handshake_timeout, then drop the socket. + for task in list(self._accept_tasks): + task.cancel() + if self._accept_tasks: + await asyncio.gather(*self._accept_tasks, return_exceptions=True) + await mux.close() + + try: + await self._bridge.run_coro(_shutdown_mux()) + except Exception: + logger.debug("Error closing UdpMux", exc_info=True) + + if self._nursery is not None: + self._nursery.cancel_scope.cancel() + await self._nursery_done.wait() + self._listening_addrs.clear() logger.debug("WebRTC Direct listener closed") diff --git a/libp2p/transport/webrtc/noise_handshake.py b/libp2p/transport/webrtc/noise_handshake.py index b3946895b..f120305e6 100644 --- a/libp2p/transport/webrtc/noise_handshake.py +++ b/libp2p/transport/webrtc/noise_handshake.py @@ -29,9 +29,12 @@ from libp2p.crypto.keys import PrivateKey from libp2p.peer.id import ID from libp2p.security.noise.patterns import PatternXX +from libp2p.utils.varint import decode_varint_with_size -from .constants import NOISE_PROLOGUE_PREFIX +from .constants import MAX_PAYLOAD_SIZE, NOISE_PROLOGUE_PREFIX from .exceptions import WebRTCHandshakeError +from .pb.webrtc_pb2 import Message +from .stream import _frame logger = logging.getLogger(__name__) @@ -40,19 +43,25 @@ def build_noise_prologue( - local_fingerprint: bytes, - remote_fingerprint: bytes, + dialer_fingerprint: bytes, + server_fingerprint: bytes, ) -> bytes: """ Build the Noise prologue that binds the handshake to the DTLS session. - :param local_fingerprint: Raw SHA-256 of the local DTLS certificate. - :param remote_fingerprint: Raw SHA-256 of the remote DTLS certificate. + Per the WebRTC-Direct spec the prologue is + ``"libp2p-webrtc-noise:" || multihash(FP_A) || multihash(FP_B)`` where *A* + is the dialer (Noise **responder**) and *B* is the server (Noise + **initiator**). The order is fixed by role, not by "local"/"remote", so both + sides must pass the same two values in the same order. + + :param dialer_fingerprint: Raw SHA-256 of the dialer's DTLS certificate. + :param server_fingerprint: Raw SHA-256 of the server's DTLS certificate. :returns: The prologue bytes for ``NoiseState.set_prologue()``. """ - local_mh = _MH_SHA256_HEADER + local_fingerprint - remote_mh = _MH_SHA256_HEADER + remote_fingerprint - return NOISE_PROLOGUE_PREFIX + local_mh + remote_mh + dialer_mh = _MH_SHA256_HEADER + dialer_fingerprint + server_mh = _MH_SHA256_HEADER + server_fingerprint + return NOISE_PROLOGUE_PREFIX + dialer_mh + server_mh async def perform_noise_handshake( @@ -60,26 +69,32 @@ async def perform_noise_handshake( local_peer: ID, libp2p_privkey: PrivateKey, noise_static_key: PrivateKey, - local_fingerprint: bytes, - remote_fingerprint: bytes, + dialer_fingerprint: bytes, + server_fingerprint: bytes, is_initiator: bool, remote_peer: ID | None = None, ) -> ID: """ Run the Noise XX handshake over a data-channel-0 connection. + Roles follow the WebRTC-Direct spec: the **server (listener) is the Noise + initiator** and the **dialer is the responder**. This is independent of + which side opened the WebRTC connection. + :param conn: A :class:`IRawConnection` wrapping data channel 0. :param local_peer: The local peer's ID. :param libp2p_privkey: The local peer's libp2p identity private key. :param noise_static_key: An ephemeral X25519 key for the Noise session. - :param local_fingerprint: Raw SHA-256 of the local DTLS certificate. - :param remote_fingerprint: Raw SHA-256 of the remote DTLS certificate. - :param is_initiator: True if this peer initiated the connection. - :param remote_peer: Expected remote peer ID (for outbound connections). + :param dialer_fingerprint: Raw SHA-256 of the dialer's DTLS certificate. + :param server_fingerprint: Raw SHA-256 of the server's DTLS certificate. + :param is_initiator: True if this peer is the Noise initiator (the server). + :param remote_peer: Expected remote peer ID, or ``None`` if unknown (the + server does not know the dialer's ID; the dialer should verify the + returned ID against the ``/p2p/`` component itself). :returns: The authenticated remote peer ID. :raises WebRTCHandshakeError: If the handshake fails. """ - prologue = build_noise_prologue(local_fingerprint, remote_fingerprint) + prologue = build_noise_prologue(dialer_fingerprint, server_fingerprint) logger.debug( "Noise handshake prologue: %d bytes (initiator=%s)", len(prologue), @@ -95,10 +110,6 @@ async def perform_noise_handshake( try: if is_initiator: - if remote_peer is None: - raise WebRTCHandshakeError( - "remote_peer is required for outbound Noise handshake" - ) secure_conn = await pattern.handshake_outbound(conn, remote_peer) else: secure_conn = await pattern.handshake_inbound(conn) @@ -113,6 +124,29 @@ async def perform_noise_handshake( raise WebRTCHandshakeError(f"Noise handshake failed: {e}") from e +def _unframe(raw: bytes) -> tuple[bytes, bool]: + """ + Decode one uvarint-prefixed ``webrtc.pb.Message`` from a channel message. + + :returns: ``(payload, fin)`` — the ``message`` bytes (possibly empty for a + pure flag frame) and whether the peer signalled FIN/RESET. + :raises WebRTCHandshakeError: on malformed framing. + """ + try: + length, consumed = decode_varint_with_size(raw) + except ValueError as e: + raise WebRTCHandshakeError(f"malformed handshake frame: {e}") from e + if consumed == 0 or raw[consumed - 1] & 0x80 or consumed + length != len(raw): + raise WebRTCHandshakeError("malformed handshake frame length") + msg = Message() + try: + msg.ParseFromString(raw[consumed:]) + except Exception as e: # DecodeError + raise WebRTCHandshakeError(f"malformed handshake frame: {e}") from e + fin = msg.HasField("flag") and msg.flag in (Message.FIN, Message.RESET) + return msg.message, fin + + class DataChannelReadWriter(IRawConnection): """ Wraps a WebRTC data channel (stream) as an ``IRawConnection`` so the @@ -121,6 +155,14 @@ class DataChannelReadWriter(IRawConnection): The data channel is represented by ``send_cb`` and ``recv_cb`` callables rather than a direct aiortc reference. + + Wire format (spec, "Multiplexing" + go/js behaviour): the handshake runs + over a WebRTC *stream* on channel 0, i.e. every data-channel message is a + uvarint-length-prefixed ``webrtc.pb.Message`` whose ``message`` field + carries a chunk of the Noise byte stream (2-byte length prefix + payload). + ``write`` frames, ``read`` unframes and buffers, so the byte-oriented + Noise packet reader (``read_exactly(conn, 2)`` then the payload) works + unchanged and the bytes on the wire match other implementations. """ def __init__( @@ -131,15 +173,50 @@ def __init__( ) -> None: self._send_cb = send_cb self._recv_cb = recv_cb + self._buffer = bytearray() + self._closed = False self.is_initiator = is_initiator + async def _fill(self) -> bool: + """Pull one channel message into the buffer. False once the peer is done.""" + if self._closed: + return False + raw = await self._recv_cb() + if not raw: + self._closed = True + return False + payload, fin = _unframe(raw) + self._buffer.extend(payload) + if fin: + self._closed = True + return True + async def read(self, n: int | None = None) -> bytes: - """Read the next message from the data channel.""" - return await self._recv_cb() + """ + Read from the data channel. + + With ``n=None`` return whatever is buffered, or the payload of the next + channel message. With ``n`` return exactly ``n`` bytes, pulling further + channel messages as needed (short only if the peer closed). + """ + if n is None: + if not self._buffer: + await self._fill() + data = bytes(self._buffer) + self._buffer.clear() + return data + while len(self._buffer) < n: + if not await self._fill(): + break # closed; return what we have (read_exactly raises) + data = bytes(self._buffer[:n]) + del self._buffer[:n] + return data async def write(self, data: bytes) -> None: - """Write a message to the data channel.""" - await self._send_cb(data) + """Write Noise bytes to the data channel, framed as stream messages.""" + for offset in range(0, len(data), MAX_PAYLOAD_SIZE): + chunk = data[offset : offset + MAX_PAYLOAD_SIZE] + await self._send_cb(_frame(Message(message=chunk))) async def close(self) -> None: """No-op — the channel lifecycle is managed by the connection.""" diff --git a/libp2p/transport/webrtc/sdp.py b/libp2p/transport/webrtc/sdp.py index 1979ace47..e02edaaf3 100644 --- a/libp2p/transport/webrtc/sdp.py +++ b/libp2p/transport/webrtc/sdp.py @@ -1,19 +1,28 @@ """ -SDP construction for WebRTC Direct. +SDP construction and ICE-username helpers for WebRTC Direct. -For WebRTC Direct, there is no signaling exchange — the client constructs an -SDP offer locally from the server's multiaddr (IP, port, certificate hash). -The server answers with its own locally-constructed SDP. +WebRTC Direct has no signaling exchange: -All ICE credential injection is isolated in :meth:`SDPBuilder._apply_ice_credentials` -so that when Chrome removes ICE credential munging (libp2p/specs#672) only -that single method needs to change. +- the **dialer** builds the server's answer locally from the multiaddr (IP, + port, certhash) — :func:`build_synthetic_answer`; +- the **listener** infers the dialer's offer from the first inbound STUN + BINDING REQUEST (its ``USERNAME`` and source address) — + :func:`parse_direct_username` + :func:`build_inferred_offer`. + +The ICE username fragment doubles as the protocol version switch: +``libp2p+webrtc+v1/…`` (SDP munging) or ``libp2p+webrtc+v2/…`` (no munging, +libp2p/specs#715). Note that aiortc **ignores** ICE credentials in SDP text +passed to ``setLocalDescription`` — the real local credentials live on the +aioice ``Connection`` — so "munging" on our own dialer means setting those +fields, not editing SDP (see ``transport.py``). Spec: https://github.com/libp2p/specs/blob/master/webrtc/webrtc-direct.md """ from __future__ import annotations +import base64 +import re import secrets from .certificate import WebRTCCertificate, fingerprint_from_multibase @@ -38,8 +47,139 @@ a=sctp-port:5000 a=max-message-size:{max_message_size} a=candidate:1 1 UDP {priority} {host} {port} typ host +a=end-of-candidates """ +# --------------------------------------------------------------------------- +# ICE username / version helpers (spec: "Connection Establishment") +# --------------------------------------------------------------------------- + +#: ufrag prefix for the SDP-munging flow (browsers are removing munging). +WEBRTC_DIRECT_V1_PREFIX = "libp2p+webrtc+v1/" +#: ufrag prefix for the no-munging flow (libp2p/specs#715). +WEBRTC_DIRECT_V2_PREFIX = "libp2p+webrtc+v2/" +_VERSION_PREFIXES = {WEBRTC_DIRECT_V1_PREFIX: 1, WEBRTC_DIRECT_V2_PREFIX: 2} + +# RFC 8839 §5.4: ice-char = ALPHA / DIGIT / "+" / "/"; ufrag 4..256, pwd 22..256. +_ICE_CHARS = re.compile(r"[A-Za-z0-9+/]+") +ICE_UFRAG_MIN, ICE_UFRAG_MAX = 4, 256 +ICE_PWD_MIN, ICE_PWD_MAX = 22, 256 + +# Placeholder for the inferred offer: the listener cannot know the dialer's +# DTLS fingerprint yet (spec step 6.2), so DTLS fingerprint verification is +# disabled for inbound and Noise authenticates instead. aiortc still requires +# *a* fingerprint line to be present. +_PLACEHOLDER_FINGERPRINT = ":".join(["00"] * 32) + + +def is_ice_ufrag(value: str) -> bool: + return ICE_UFRAG_MIN <= len(value) <= ICE_UFRAG_MAX and bool( + _ICE_CHARS.fullmatch(value) + ) + + +def is_ice_pwd(value: str) -> bool: + return ICE_PWD_MIN <= len(value) <= ICE_PWD_MAX and bool( + _ICE_CHARS.fullmatch(value) + ) + + +def parse_direct_username(username: str) -> tuple[int, str, str]: + """ + Split and validate the STUN ``USERNAME`` of a WebRTC-Direct first contact. + + The dialer sets ``USERNAME = server_ufrag:client_ufrag`` (RFC 8445 §7.2.2) + where ``server_ufrag`` carries the version prefix. Both halves must be + valid ufrags. An unknown or missing prefix is rejected — the spec says + never to guess a version. + + :returns: ``(version, server_ufrag, client_ufrag)``. + :raises WebRTCConnectionError: on any malformed / unsupported input. + """ + server_ufrag, sep, client_ufrag = username.partition(":") + if not sep: + raise WebRTCConnectionError("STUN USERNAME is not 'server:client'") + if not (is_ice_ufrag(server_ufrag) and is_ice_ufrag(client_ufrag)): + raise WebRTCConnectionError("STUN USERNAME halves are not valid ICE ufrags") + for prefix, version in _VERSION_PREFIXES.items(): + if server_ufrag.startswith(prefix): + return version, server_ufrag, client_ufrag + raise WebRTCConnectionError( + "unknown WebRTC-Direct version prefix in ufrag (want libp2p+webrtc+v1/ " + "or libp2p+webrtc+v2/)" + ) + + +def make_v1_credential(random_len: int = 32) -> str: + """``libp2p+webrtc+v1/`` — used as both ufrag and pwd in v1.""" + return WEBRTC_DIRECT_V1_PREFIX + _generate_ice_credential(random_len) + + +def build_inferred_offer( + *, + client_ufrag: str, + client_pwd: str, + remote_host: str, + remote_port: int, + max_message_size: int = 16384, +) -> str: + """ + The dialer's offer as the listener reconstructs it (spec v1 step 6). + + ``a=setup:active`` (spec allows ``actpass`` or ``active``): with + ``actpass`` aiortc would answer as DTLS *client*, but the listener must + be the DTLS server. The fingerprint is a placeholder — see + :data:`_PLACEHOLDER_FINGERPRINT`. + """ + ip_version = "IP6" if ":" in remote_host else "IP4" + return _SDP_TEMPLATE.format( + session_id=secrets.randbelow(2**62), + ip_version=ip_version, + host=remote_host, + port=remote_port, + ice_ufrag=client_ufrag, + ice_pwd=client_pwd, + fingerprint=_PLACEHOLDER_FINGERPRINT, + setup_role="active", + max_message_size=max_message_size, + priority=2130706431, + ) + + +def build_synthetic_answer( + *, + host: str, + port: int, + certhash_multibase: str, + ufrag: str, + pwd: str, + max_message_size: int = 16384, +) -> str: + """ + The server's answer as the dialer synthesises it (spec v1 step 4). + + ICE-Lite, ``a=setup:passive`` (server = DTLS server), single host + candidate at the multiaddr's IP:port, fingerprint from the certhash (so + aiortc pins the server's DTLS certificate for us). + """ + fingerprint_bytes = fingerprint_from_multibase(certhash_multibase) + fingerprint_hex = ":".join(f"{b:02X}" for b in fingerprint_bytes) + ip_version = "IP6" if ":" in host else "IP4" + sdp = _SDP_TEMPLATE.format( + session_id=secrets.randbelow(2**62), + ip_version=ip_version, + host=host, + port=port, + ice_ufrag=ufrag, + ice_pwd=pwd, + fingerprint=fingerprint_hex, + setup_role="passive", + max_message_size=max_message_size, + priority=2130706431, + ) + # a=ice-lite is a session-level attribute: insert before the m= line. + return sdp.replace("m=application", "a=ice-lite\nm=application", 1) + class SDPBuilder: """ @@ -206,25 +346,23 @@ def _apply_ice_credentials( remote_pwd: str | None = None, ) -> str: """ - Apply ICE credentials to the SDP. + Apply ICE credentials to the SDP (legacy ``SDPBuilder`` path). - **This is the single seam for libp2p/specs#672.** When Chrome drops - ICE credential munging support, only this function needs to change. - Currently a no-op passthrough — local credentials are already in the - SDP template. The remote credentials are accepted but unused until - the spec changes require injecting them via a separate mechanism. + Kept for the experimental HTTP harness. The spec-path credential seams + are :func:`build_synthetic_answer` / :func:`build_inferred_offer` (SDP + text) and the aioice ``Connection`` fields set by the dialer (aiortc + ignores ICE credentials in SDP passed to ``setLocalDescription``). :param sdp: The SDP string with local credentials in template slots. :param ufrag: Local ICE username fragment. :param pwd: Local ICE password. :param fingerprint_hex: Colon-separated cert fingerprint. - :param remote_ufrag: Remote ICE ufrag (for answer SDP / ICE agent). - :param remote_pwd: Remote ICE pwd (for answer SDP / ICE agent). - :returns: The (possibly modified) SDP string. + :param remote_ufrag: Remote ICE ufrag (unused). + :param remote_pwd: Remote ICE pwd (unused). + :returns: The (unmodified) SDP string. """ - # Currently a passthrough. The SDP template already contains - # a=ice-ufrag, a=ice-pwd, and a=fingerprint lines. Remote - # credentials will be used when aiortc's ICE agent is wired up. + # Passthrough: the template already carries a=ice-ufrag / a=ice-pwd / + # a=fingerprint. return sdp @@ -260,5 +398,12 @@ def fingerprint_from_sdp(sdp: str) -> bytes: def _generate_ice_credential(length: int) -> str: - """Generate a random ICE credential string (alphanumeric).""" - return secrets.token_urlsafe(length)[:length] + """ + Random ICE credential of exactly *length* ice-chars. + + Standard base64 alphabet (``A-Za-z0-9+/``) — RFC 8839 ice-chars. Not + ``token_urlsafe``: ``-`` and ``_`` are not ice-chars and aioice rejects + them. + """ + raw = base64.b64encode(secrets.token_bytes(length)).decode("ascii") + return raw.replace("=", "")[:length] diff --git a/libp2p/transport/webrtc/transport.py b/libp2p/transport/webrtc/transport.py index 3ce4b3294..617d2e15c 100644 --- a/libp2p/transport/webrtc/transport.py +++ b/libp2p/transport/webrtc/transport.py @@ -15,7 +15,7 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from multiaddr import Multiaddr import trio @@ -35,7 +35,7 @@ is_webrtc_direct_multiaddr, parse_webrtc_direct_multiaddr, ) -from .sdp import SDPBuilder +from .sdp import SDPBuilder, build_synthetic_answer, make_v1_credential if TYPE_CHECKING: pass @@ -125,11 +125,13 @@ async def dial(self, maddr: Multiaddr) -> WebRTCConnection: # All aiortc calls go through the bridge (asyncio thread). from ._aiortc_helpers import ( + close_peer_connection, create_noise_channel, create_peer_connection, get_remote_fingerprint, make_noise_channel_callbacks, post_sdp, + set_private_attr, wait_for_connected, wire_pc_to_connection, ) @@ -143,9 +145,12 @@ async def dial(self, maddr: Multiaddr) -> WebRTCConnection: "Ensure aiortc is installed and config uses from_aiortc()." ) + pc: Any = None try: # 1. Create RTCPeerConnection + Noise channel - pc = await bridge.run_coro(create_peer_connection(rtc_cert)) + pc = await bridge.run_coro( + create_peer_connection(rtc_cert, ice_servers=self._config.ice_servers) + ) noise_ch = await bridge.run_coro(create_noise_channel(pc)) # make_noise_channel_callbacks is sync; wrap inline. @@ -154,14 +159,42 @@ async def _setup_noise() -> tuple: # type: ignore[type-arg] noise_send, noise_recv, _ = await bridge.run_coro(_setup_noise()) - # 2. Create offer, set local description - offer = await bridge.run_coro(pc.createOffer()) - await bridge.run_coro(pc.setLocalDescription(offer)) - - # 3. Exchange SDP via HTTP POST to the listener - answer_sdp = await bridge.run_coro( - post_sdp(host, port, pc.localDescription.sdp) - ) + if self._config.enable_sdp_http_harness: + # Experimental py<->py harness: real SDP exchange over HTTP. + offer = await bridge.run_coro(pc.createOffer()) + await bridge.run_coro(pc.setLocalDescription(offer)) + answer_sdp = await bridge.run_coro( + post_sdp(host, port, pc.localDescription.sdp) + ) + else: + # Spec v1 (SDP munging): the same "libp2p+webrtc+v1/" + # string is our ufrag *and* pwd on both the local offer and the + # synthetic remote answer, so the server can reconstruct + # everything from our first STUN USERNAME. aiortc regenerates + # local ICE credentials from its aioice Connection when + # setLocalDescription runs (SDP text edits are ignored), so + # "munging" means setting those fields. + cred = make_v1_credential() + + async def _set_local_ice_credentials() -> None: + assert pc.sctp is not None # createDataChannel ran above + ice_conn = pc.sctp.transport.transport._connection + set_private_attr(ice_conn, "_local_username", cred) + set_private_attr(ice_conn, "_local_password", cred) + + await bridge.run_coro(_set_local_ice_credentials()) + offer = await bridge.run_coro(pc.createOffer()) + await bridge.run_coro(pc.setLocalDescription(offer)) + # The listener is ICE-Lite/controlled and the DTLS server; its + # fingerprint comes from the certhash so aiortc pins it. + answer_sdp = build_synthetic_answer( + host=host, + port=port, + certhash_multibase=certhash, + ufrag=cred, + pwd=cred, + max_message_size=self._config.max_message_size, + ) # 4. Set remote description from aiortc import RTCSessionDescription @@ -176,7 +209,6 @@ async def _setup_noise() -> tuple: # type: ignore[type-arg] expected_fp = fingerprint_from_multibase(certhash) remote_fp = get_remote_fingerprint(pc) # sync, safe off-thread if remote_fp != expected_fp: - await bridge.run_coro(pc.close()) raise WebRTCConnectionError( "Remote DTLS fingerprint does not match certhash in the multiaddr" ) @@ -204,21 +236,33 @@ async def _trio_noise_send(data: bytes) -> None: async def _trio_noise_recv() -> bytes: return await bridge.run_coro(noise_recv()) + # Per the WebRTC-Direct spec the *server* is the Noise initiator + # and the dialer is the responder (this is independent of the + # transport-level ``is_initiator`` used for stream-ID parity). noise_rw = DataChannelReadWriter( send_cb=_trio_noise_send, recv_cb=_trio_noise_recv, - is_initiator=True, - ) - authenticated_peer = await perform_noise_handshake( - conn=noise_rw, - local_peer=self._local_peer_id, - libp2p_privkey=self._private_key, - noise_static_key=noise_kp.private_key, - local_fingerprint=self._certificate.fingerprint, - remote_fingerprint=expected_fp, - is_initiator=True, - remote_peer=remote_peer_id, + is_initiator=False, ) + # As responder our first step is a read; bound it — nothing else + # (swarm, transport) times out a hung dial. + with trio.fail_after(self._config.handshake_timeout): + authenticated_peer = await perform_noise_handshake( + conn=noise_rw, + local_peer=self._local_peer_id, + libp2p_privkey=self._private_key, + noise_static_key=noise_kp.private_key, + dialer_fingerprint=self._certificate.fingerprint, + server_fingerprint=expected_fp, + is_initiator=False, + ) + # The responder cannot pin the peer during the handshake, so + # verify the authenticated identity against ``/p2p/`` here. + if remote_peer_id is not None and authenticated_peer != remote_peer_id: + raise WebRTCConnectionError( + f"Remote peer ID {authenticated_peer} does not match " + f"/p2p/{remote_peer_id} in the multiaddr" + ) # 9. Finalize connection conn.peer_id = authenticated_peer @@ -229,9 +273,18 @@ async def _trio_noise_recv() -> bytes: ) return conn - except WebRTCConnectionError: - raise - except Exception as e: + except BaseException as e: + # Every failure path (ICE timeout, fingerprint/peer-ID mismatch, + # Noise failure, cancellation) must release the PC — it owns UDP + # sockets and asyncio tasks on the bridge loop. + if pc is not None: + with trio.CancelScope(shield=True): + try: + await bridge.run_coro(close_peer_connection(pc)) + except Exception: + pass + if isinstance(e, WebRTCConnectionError | trio.Cancelled): + raise raise WebRTCConnectionError(f"WebRTC Direct dial failed: {e}") from e def create_listener(self, handler_function: THandler) -> WebRTCDirectListener: diff --git a/newsfragments/1437.bugfix.rst b/newsfragments/1437.bugfix.rst new file mode 100644 index 000000000..36721b81a --- /dev/null +++ b/newsfragments/1437.bugfix.rst @@ -0,0 +1 @@ +Fixed the WebRTC-Direct inbound path, which never completed: the listener now runs the Noise XX handshake (as initiator, per spec) on the trio side and hands the authenticated ``WebRTCConnection`` to the handler; the dialer is the Noise responder and verifies the ``/p2p/`` peer ID after the handshake. Also fixed ``get_remote_fingerprint`` (read non-existent aiortc attributes, so ``dial()`` always failed), ``DataChannelReadWriter.read(n)`` now honours ``n`` (the Noise packet reader needs the 2-byte length prefix alone), the Noise prologue is role-ordered (dialer fingerprint, then server), and the noise channel ``send`` waits for the channel to open, and the handshake bytes are framed as ``webrtc.pb.Message`` stream frames on channel 0 like go/js (was raw). The dialer bounds the Noise phase with ``handshake_timeout`` and closes the peer connection on every failure path; a raising connection handler no longer takes down the listeners trio run. ``PatternXX.handshake_outbound`` accepts ``remote_peer=None`` for initiators that learn the peer ID from the handshake. diff --git a/newsfragments/1437.feature.rst b/newsfragments/1437.feature.rst new file mode 100644 index 000000000..dfe4459c0 --- /dev/null +++ b/newsfragments/1437.feature.rst @@ -0,0 +1 @@ +Added a spec-aligned WebRTC-Direct listener: one shared UDP port (``UdpMux``), inbound dials dispatched by the STUN ``USERNAME`` ufrag with ``libp2p+webrtc+v1/`` version-prefix validation (unknown/missing prefixes are rejected, never assumed v1), the dialer's offer inferred from its first STUN packet, DTLS fingerprint verification disabled for inbound per spec (Noise authenticates), and an in-flight cap on unauthenticated inbounds. Our dialer now speaks WebRTC-Direct v1 (``libp2p+webrtc+v1/`` ufrag == pwd on both the local offer and a synthesised ICE-Lite answer built from the multiaddr), so py↔py loopback exercises the real STUN path. The HTTP ``POST /sdp`` signaling harness is now opt-in via ``WebRTCTransportConfig(enable_sdp_http_harness=True)`` (experimental, py↔py only). New helpers in ``libp2p.transport.webrtc.sdp``: ``parse_direct_username``, ``build_inferred_offer``, ``build_synthetic_answer``, ``make_v1_credential``; ICE credentials are generated with ice-chars only. diff --git a/newsfragments/1437.internal.rst b/newsfragments/1437.internal.rst new file mode 100644 index 000000000..6a6a26b93 --- /dev/null +++ b/newsfragments/1437.internal.rst @@ -0,0 +1 @@ +``UdpMux`` can now carry a full aiortc ``RTCPeerConnection``: new ``attach_muxed_connection(pc, mux, conn)`` helper swaps the peer connection's ICE agent for a mux-backed one (including the DTLS ``_recv``/``_send`` bindings) and registers/unregisters the peer address on ICE state changes; ``add_ice_connection`` marks gathering *started* so aiortc's ``gather()`` no longer binds extra sockets; STUN responses (no ``USERNAME``) route by address; peer addresses are learned from STUN checks and outbound sends (latest wins, capped per connection); ``unregister(ufrag)`` drops the learned addresses; malformed STUN-shaped datagrams are delivered to the connection's data path instead of re-raising ``struct.error`` out of the loop. ``webrtc`` extra now requires ``aiortc>=1.15``. diff --git a/newsfragments/1437.misc.rst b/newsfragments/1437.misc.rst new file mode 100644 index 000000000..953f37710 --- /dev/null +++ b/newsfragments/1437.misc.rst @@ -0,0 +1 @@ +Install the ``webrtc`` extra (``aiortc``) in the tox test environments so the WebRTC transport tests run in CI instead of being skipped. diff --git a/pyproject.toml b/pyproject.toml index 8cf20218c..cbc621f19 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,7 +63,7 @@ classifiers = [ Homepage = "https://github.com/libp2p/py-libp2p" [project.optional-dependencies] -webrtc = ["aiortc>=1.5.0,<2.0"] +webrtc = ["aiortc>=1.15,<2.0"] [project.scripts] chat-demo = "examples.chat.chat:main" diff --git a/tests/core/security/noise/test_patterns.py b/tests/core/security/noise/test_patterns.py index 6898917a8..92342fc4f 100644 --- a/tests/core/security/noise/test_patterns.py +++ b/tests/core/security/noise/test_patterns.py @@ -295,6 +295,47 @@ async def do_responder(): assert init_error is not None assert isinstance(init_error, PeerIDMismatchesPubkey) + @pytest.mark.trio + async def test_handshake_outbound_without_expected_peer( + self, pattern_setup, nursery + ): + """ + ``remote_peer=None`` skips only the peer-ID equality check. + + The initiator still authenticates the responder via its signed + handshake payload and learns its peer ID from the connection. + """ + initiator_pattern, libp2p_keypair, noise_keypair = pattern_setup + + responder_keypair = create_new_key_pair() + responder_peer = ID.from_pubkey(responder_keypair.public_key) + responder_pattern = PatternXX( + local_peer=responder_peer, + libp2p_privkey=responder_keypair.private_key, + noise_static_key=noise_static_key_factory(), + ) + + from tests.utils.factories import raw_conn_factory + + async with raw_conn_factory(nursery) as (init_conn, resp_conn): + results: dict[str, ID] = {} + + async def do_initiator(): + secure = await initiator_pattern.handshake_outbound(init_conn, None) + results["init_sees"] = secure.get_remote_peer() + + async def do_responder(): + secure = await responder_pattern.handshake_inbound(resp_conn) + results["resp_sees"] = secure.get_remote_peer() + + with trio.fail_after(10): + async with trio.open_nursery() as handshake_nursery: + handshake_nursery.start_soon(do_initiator) + handshake_nursery.start_soon(do_responder) + + assert results["init_sees"] == responder_peer + assert results["resp_sees"] == ID.from_pubkey(libp2p_keypair.public_key) + @pytest.mark.trio async def test_handshake_not_finished(self, pattern_setup, nursery): """Test handshake failure when connection closes before completion.""" diff --git a/tests/core/transport/webrtc/test_aiortc_helpers.py b/tests/core/transport/webrtc/test_aiortc_helpers.py index 4993f29cc..32063c1a9 100644 --- a/tests/core/transport/webrtc/test_aiortc_helpers.py +++ b/tests/core/transport/webrtc/test_aiortc_helpers.py @@ -226,3 +226,71 @@ async def on_offer(sdp: str) -> str: finally: server.close() await server.wait_closed() + + +# --------------------------------------------------------------------------- +# get_remote_fingerprint +# --------------------------------------------------------------------------- + + +class TestGetRemoteFingerprint: + def test_matches_peer_certificate_after_dtls(self) -> None: + asyncio.run(self._run()) + + async def _run(self) -> None: + from aiortc import RTCSessionDescription + + from libp2p.transport.webrtc._aiortc_helpers import ( + create_noise_channel, + create_peer_connection, + get_remote_fingerprint, + wait_for_connected, + ) + from libp2p.transport.webrtc.certificate import WebRTCCertificate + + cert_a = WebRTCCertificate.from_aiortc() + cert_b = WebRTCCertificate.from_aiortc() + pc_a = await create_peer_connection(cert_a._rtc_certificate, ice_servers=[]) + pc_b = await create_peer_connection(cert_b._rtc_certificate, ice_servers=[]) + try: + with pytest.raises(ValueError): + get_remote_fingerprint(pc_a) # no SCTP/DTLS yet + + await create_noise_channel(pc_a) + await create_noise_channel(pc_b) + offer = await pc_a.createOffer() + await pc_a.setLocalDescription(offer) + await pc_b.setRemoteDescription( + RTCSessionDescription(sdp=pc_a.localDescription.sdp, type="offer") + ) + answer = await pc_b.createAnswer() + await pc_b.setLocalDescription(answer) + await pc_a.setRemoteDescription( + RTCSessionDescription(sdp=pc_b.localDescription.sdp, type="answer") + ) + await asyncio.gather( + wait_for_connected(pc_a, timeout=15.0), + wait_for_connected(pc_b, timeout=15.0), + ) + + assert get_remote_fingerprint(pc_a) == cert_b.fingerprint + assert get_remote_fingerprint(pc_b) == cert_a.fingerprint + finally: + await pc_a.close() + await pc_b.close() + + +class TestSetPrivateAttr: + def test_write_to_missing_name_fails_at_the_write(self) -> None: + from libp2p.transport.webrtc._aiortc_helpers import set_private_attr + + class Obj: + _slot = 1 + + obj = Obj() + set_private_attr(obj, "_slot", 2) + assert obj._slot == 2 + # A typo'd / upgraded-away name must not silently create a new + # attribute the library never reads. + with pytest.raises(AssertionError): + set_private_attr(obj, "_slott", 3) diff --git a/tests/core/transport/webrtc/test_noise_handshake.py b/tests/core/transport/webrtc/test_noise_handshake.py index f86c05da9..75305aed7 100644 --- a/tests/core/transport/webrtc/test_noise_handshake.py +++ b/tests/core/transport/webrtc/test_noise_handshake.py @@ -7,46 +7,59 @@ from unittest.mock import AsyncMock import pytest +import trio -from libp2p.transport.webrtc.constants import NOISE_PROLOGUE_PREFIX +from libp2p.crypto.ed25519 import create_new_key_pair +from libp2p.crypto.x25519 import create_new_key_pair as create_x25519_key_pair +from libp2p.peer.id import ID +from libp2p.transport.webrtc.certificate import WebRTCCertificate +from libp2p.transport.webrtc.constants import ( + MAX_MESSAGE_SIZE, + MAX_PAYLOAD_SIZE, + NOISE_PROLOGUE_PREFIX, +) +from libp2p.transport.webrtc.exceptions import WebRTCHandshakeError from libp2p.transport.webrtc.noise_handshake import ( DataChannelReadWriter, build_noise_prologue, + perform_noise_handshake, ) +from libp2p.transport.webrtc.pb.webrtc_pb2 import Message +from libp2p.transport.webrtc.stream import _frame class TestBuildNoisePrologue: def test_prologue_starts_with_prefix(self): - local_fp = b"\x01" * 32 - remote_fp = b"\x02" * 32 - prologue = build_noise_prologue(local_fp, remote_fp) + dialer_fp = b"\x01" * 32 + server_fp = b"\x02" * 32 + prologue = build_noise_prologue(dialer_fp, server_fp) assert prologue.startswith(NOISE_PROLOGUE_PREFIX) def test_prologue_contains_multihash_encoded_fingerprints(self): - local_fp = b"\xaa" * 32 - remote_fp = b"\xbb" * 32 - prologue = build_noise_prologue(local_fp, remote_fp) - # After prefix: local_mh (34 bytes) + remote_mh (34 bytes) + dialer_fp = b"\xaa" * 32 + server_fp = b"\xbb" * 32 + prologue = build_noise_prologue(dialer_fp, server_fp) + # After prefix: dialer_mh (34 bytes) + server_mh (34 bytes) after_prefix = prologue[len(NOISE_PROLOGUE_PREFIX) :] assert len(after_prefix) == 68 # 34 + 34 # Verify multihash headers assert after_prefix[0] == 0x12 # SHA-256 code assert after_prefix[1] == 32 # digest length - assert after_prefix[2:34] == local_fp + assert after_prefix[2:34] == dialer_fp assert after_prefix[34] == 0x12 assert after_prefix[35] == 32 - assert after_prefix[36:68] == remote_fp + assert after_prefix[36:68] == server_fp def test_prologue_total_length(self): - local_fp = b"\x00" * 32 - remote_fp = b"\xff" * 32 - prologue = build_noise_prologue(local_fp, remote_fp) - # prefix (20) + local_mh (34) + remote_mh (34) = 88 + dialer_fp = b"\x00" * 32 + server_fp = b"\xff" * 32 + prologue = build_noise_prologue(dialer_fp, server_fp) + # prefix (20) + dialer_mh (34) + server_mh (34) = 88 expected = len(NOISE_PROLOGUE_PREFIX) + 34 + 34 assert len(prologue) == expected def test_prologue_is_asymmetric(self): - """Swapping local/remote produces different prologues.""" + """Swapping dialer/server produces different prologues.""" fp_a = b"\x01" * 32 fp_b = b"\x02" * 32 p1 = build_noise_prologue(fp_a, fp_b) @@ -54,8 +67,6 @@ def test_prologue_is_asymmetric(self): assert p1 != p2 def test_prologue_with_real_fingerprints(self): - from libp2p.transport.webrtc.certificate import WebRTCCertificate - cert_a = WebRTCCertificate.generate() cert_b = WebRTCCertificate.generate() prologue = build_noise_prologue(cert_a.fingerprint, cert_b.fingerprint) @@ -64,28 +75,56 @@ def test_prologue_with_real_fingerprints(self): class TestDataChannelReadWriter: @pytest.mark.trio - async def test_write_calls_send_cb(self): + async def test_write_frames_as_stream_message(self): + """Wire format = uvarint-prefixed webrtc.pb.Message (spec/go/js).""" send_cb = AsyncMock() - recv_cb = AsyncMock(return_value=b"response") rw = DataChannelReadWriter( - send_cb=send_cb, - recv_cb=recv_cb, - is_initiator=True, + send_cb=send_cb, recv_cb=AsyncMock(), is_initiator=True ) await rw.write(b"hello") - send_cb.assert_called_once_with(b"hello") + send_cb.assert_called_once_with(_frame(Message(message=b"hello"))) @pytest.mark.trio - async def test_read_calls_recv_cb(self): + async def test_write_chunks_large_payloads(self): send_cb = AsyncMock() - recv_cb = AsyncMock(return_value=b"data-from-peer") rw = DataChannelReadWriter( - send_cb=send_cb, - recv_cb=recv_cb, + send_cb=send_cb, recv_cb=AsyncMock(), is_initiator=True + ) + await rw.write(b"x" * (MAX_PAYLOAD_SIZE + 1)) + assert send_cb.await_count == 2 + for call in send_cb.await_args_list: + assert len(call.args[0]) <= MAX_MESSAGE_SIZE + + @pytest.mark.trio + async def test_read_unframes_stream_message(self): + recv_cb = AsyncMock(return_value=_frame(Message(message=b"data-from-peer"))) + rw = DataChannelReadWriter( + send_cb=AsyncMock(), recv_cb=recv_cb, is_initiator=False + ) + assert await rw.read() == b"data-from-peer" + + @pytest.mark.trio + async def test_read_stops_at_fin(self): + frames = [ + _frame(Message(message=b"ab")), + _frame(Message(flag=Message.FIN)), + ] + rw = DataChannelReadWriter( + send_cb=AsyncMock(), + recv_cb=AsyncMock(side_effect=frames), is_initiator=False, ) - data = await rw.read() - assert data == b"data-from-peer" + assert await rw.read(4) == b"ab" # short read: peer finished + + @pytest.mark.trio + async def test_read_rejects_malformed_frame(self): + rw = DataChannelReadWriter( + send_cb=AsyncMock(), + recv_cb=AsyncMock(return_value=b"\xff\xff\xff"), + is_initiator=False, + ) + with pytest.raises(WebRTCHandshakeError): + await rw.read(1) @pytest.mark.trio async def test_close_is_noop(self): @@ -111,3 +150,145 @@ def test_transport_addresses_empty(self): is_initiator=True, ) assert rw.get_transport_addresses() == [] + + @pytest.mark.trio + async def test_read_n_splits_and_joins_channel_messages(self): + """ + ``read(n)`` must honour ``n``: the Noise packet reader asks for the + 2-byte length prefix first, then the payload, while a data channel + delivers whole messages. + """ + messages = [_frame(Message(message=m)) for m in (b"\x00\x03abc", b"de", b"fgh")] + recv_cb = AsyncMock(side_effect=messages) + rw = DataChannelReadWriter( + send_cb=AsyncMock(), recv_cb=recv_cb, is_initiator=False + ) + assert await rw.read(2) == b"\x00\x03" # prefix only, rest buffered + assert await rw.read(3) == b"abc" # from buffer, no recv + assert recv_cb.await_count == 1 + assert await rw.read(4) == b"defg" # spans two channel messages + assert await rw.read() == b"h" # n=None drains the remainder + + +def _memory_channel_pair() -> tuple[DataChannelReadWriter, DataChannelReadWriter]: + """Two DataChannelReadWriters wired back-to-back over trio memory channels.""" + a_to_b_send, a_to_b_recv = trio.open_memory_channel[bytes](16) + b_to_a_send, b_to_a_recv = trio.open_memory_channel[bytes](16) + + async def _send_a(data: bytes) -> None: + await a_to_b_send.send(data) + + async def _recv_a() -> bytes: + return await b_to_a_recv.receive() + + async def _send_b(data: bytes) -> None: + await b_to_a_send.send(data) + + async def _recv_b() -> bytes: + return await a_to_b_recv.receive() + + dialer = DataChannelReadWriter(_send_a, _recv_a, is_initiator=False) + server = DataChannelReadWriter(_send_b, _recv_b, is_initiator=True) + return dialer, server + + +class TestPerformNoiseHandshake: + @pytest.mark.trio + async def test_spec_roles_server_initiates_dialer_responds(self): + """ + Server = Noise initiator without knowing the dialer's ID; dialer = + responder. Both use the role-ordered prologue and learn the peer's ID. + """ + dialer_kp, server_kp = create_new_key_pair(), create_new_key_pair() + dialer_id, server_id = ( + ID.from_pubkey(dialer_kp.public_key), + ID.from_pubkey(server_kp.public_key), + ) + dialer_cert, server_cert = ( + WebRTCCertificate.generate(), + (WebRTCCertificate.generate()), + ) + dialer_rw, server_rw = _memory_channel_pair() + seen: dict[str, ID] = {} + + async def _server() -> None: + seen["server"] = await perform_noise_handshake( + conn=server_rw, + local_peer=server_id, + libp2p_privkey=server_kp.private_key, + noise_static_key=create_x25519_key_pair().private_key, + dialer_fingerprint=dialer_cert.fingerprint, + server_fingerprint=server_cert.fingerprint, + is_initiator=True, + remote_peer=None, + ) + + async def _dialer() -> None: + seen["dialer"] = await perform_noise_handshake( + conn=dialer_rw, + local_peer=dialer_id, + libp2p_privkey=dialer_kp.private_key, + noise_static_key=create_x25519_key_pair().private_key, + dialer_fingerprint=dialer_cert.fingerprint, + server_fingerprint=server_cert.fingerprint, + is_initiator=False, + ) + + with trio.fail_after(10): + async with trio.open_nursery() as nursery: + nursery.start_soon(_server) + nursery.start_soon(_dialer) + + assert seen["server"] == dialer_id + assert seen["dialer"] == server_id + + @pytest.mark.trio + async def test_prologue_mismatch_fails(self): + """A side that gets the fingerprint order wrong cannot complete.""" + dialer_kp, server_kp = create_new_key_pair(), create_new_key_pair() + dialer_cert, server_cert = ( + WebRTCCertificate.generate(), + (WebRTCCertificate.generate()), + ) + dialer_rw, server_rw = _memory_channel_pair() + errors: list[BaseException] = [] + + async def _run(rw, kp, is_initiator, dialer_fp, server_fp) -> None: + try: + await perform_noise_handshake( + conn=rw, + local_peer=ID.from_pubkey(kp.public_key), + libp2p_privkey=kp.private_key, + noise_static_key=create_x25519_key_pair().private_key, + dialer_fingerprint=dialer_fp, + server_fingerprint=server_fp, + is_initiator=is_initiator, + ) + except WebRTCHandshakeError as e: + errors.append(e) + + with trio.move_on_after(10): + async with trio.open_nursery() as nursery: + nursery.start_soon( + _run, + server_rw, + server_kp, + True, + dialer_cert.fingerprint, + server_cert.fingerprint, + ) + # dialer swaps the order -> different prologue -> decrypt fails + nursery.start_soon( + _run, + dialer_rw, + dialer_kp, + False, + server_cert.fingerprint, + dialer_cert.fingerprint, + ) + # first failure aborts; cancel the peer stuck waiting + while not errors: + await trio.sleep(0.01) + nursery.cancel_scope.cancel() + + assert errors diff --git a/tests/core/transport/webrtc/test_sdp.py b/tests/core/transport/webrtc/test_sdp.py index 9622c4a92..719ecc97b 100644 --- a/tests/core/transport/webrtc/test_sdp.py +++ b/tests/core/transport/webrtc/test_sdp.py @@ -102,3 +102,111 @@ def test_build_from_multiaddr_components(self): # Fingerprint in the SDP should match the cert extracted = fingerprint_from_sdp(sdp) assert extracted == cert.fingerprint + + +class TestDirectUsername: + def test_v1_and_v2_parse(self): + from libp2p.transport.webrtc.sdp import parse_direct_username + + assert parse_direct_username("libp2p+webrtc+v1/abcdEFGH:cli1") == ( + 1, + "libp2p+webrtc+v1/abcdEFGH", + "cli1", + ) + pwd = "p" * 22 + assert parse_direct_username(f"libp2p+webrtc+v2/{pwd}:cli+/9") == ( + 2, + f"libp2p+webrtc+v2/{pwd}", + "cli+/9", + ) + + @pytest.mark.parametrize( + "username", + [ + "libp2p+webrtc+v1/abcd", # no ':' + "abcdEFGH:cli1", # no version prefix -> MUST reject, never assume v1 + "libp2p+webrtc+v9/abcd:cli1", # unknown version + "libp2p+webrtc+v1/ab-cd:cli1", # '-' is not an ice-char + "libp2p+webrtc+v1/abcd:cl", # client ufrag < 4 + "libp2p+webrtc+v1/" + "a" * 250 + ":cli1", # server ufrag > 256 + "libp2p+webrtc+v1/abcd:", # empty client ufrag + "libp2p+webrtc+v1/abcd\n:cli1", # trailing newline is not an ice-char + "libp2p+webrtc+v1/abcd:cli1\n", + "", + ], + ) + def test_rejects_malformed(self, username): + from libp2p.transport.webrtc.exceptions import WebRTCConnectionError + from libp2p.transport.webrtc.sdp import parse_direct_username + + with pytest.raises(WebRTCConnectionError): + parse_direct_username(username) + + def test_v1_credential_is_a_valid_ufrag_and_pwd(self): + from libp2p.transport.webrtc.sdp import ( + WEBRTC_DIRECT_V1_PREFIX, + is_ice_pwd, + is_ice_ufrag, + make_v1_credential, + parse_direct_username, + ) + + cred = make_v1_credential() + assert cred.startswith(WEBRTC_DIRECT_V1_PREFIX) + assert is_ice_ufrag(cred) and is_ice_pwd(cred) + assert parse_direct_username(f"{cred}:abcd")[1] == cred + + def test_generated_credentials_are_ice_chars_only(self): + import re + + from libp2p.transport.webrtc.sdp import _generate_ice_credential + + for n in (4, 22, 32): + for _ in range(20): + c = _generate_ice_credential(n) + assert len(c) == n + assert re.fullmatch(r"[A-Za-z0-9+/]+", c), c + + +class TestSpecSdp: + def test_inferred_offer_shape(self): + from libp2p.transport.webrtc.sdp import build_inferred_offer + + sdp = build_inferred_offer( + client_ufrag="cli1", + client_pwd="libp2p+webrtc+v1/abcdefghijklmnop", + remote_host="203.0.113.5", + remote_port=54321, + ) + assert "a=ice-ufrag:cli1\n" in sdp + assert "a=ice-pwd:libp2p+webrtc+v1/abcdefghijklmnop\n" in sdp + assert "a=setup:active" in sdp # forces aiortc into DTLS server role + assert "c=IN IP4 203.0.113.5" in sdp + assert "a=candidate:1 1 UDP 2130706431 203.0.113.5 54321 typ host" in sdp + assert "a=end-of-candidates" in sdp + assert "a=max-message-size:16384" in sdp + assert "a=fingerprint:sha-256 " in sdp + assert "a=ice-lite" not in sdp + + def test_synthetic_answer_shape(self): + from libp2p.transport.webrtc.certificate import WebRTCCertificate + from libp2p.transport.webrtc.sdp import ( + build_synthetic_answer, + fingerprint_from_sdp, + ) + + cert = WebRTCCertificate.generate() + cred = "libp2p+webrtc+v1/abcdefghijklmnop" + sdp = build_synthetic_answer( + host="127.0.0.1", + port=4001, + certhash_multibase=cert.fingerprint_to_multibase(), + ufrag=cred, + pwd=cred, + ) + assert sdp.index("a=ice-lite") < sdp.index("m=application") + assert "a=setup:passive" in sdp + assert f"a=ice-ufrag:{cred}\n" in sdp and f"a=ice-pwd:{cred}\n" in sdp + assert "a=candidate:1 1 UDP 2130706431 127.0.0.1 4001 typ host" in sdp + assert "a=end-of-candidates" in sdp + assert fingerprint_from_sdp(sdp) == cert.fingerprint diff --git a/tests/core/transport/webrtc/test_udp_mux.py b/tests/core/transport/webrtc/test_udp_mux.py index 3a4624af0..7bf554647 100644 --- a/tests/core/transport/webrtc/test_udp_mux.py +++ b/tests/core/transport/webrtc/test_udp_mux.py @@ -54,6 +54,16 @@ def _dtls_like_bytes() -> bytes: return b"\x16\xfe\xff\x00\x01\x00\x00\x00\x00\x00\x00\x00\x05hello" +class _RecordingReceiver: + """Stand-in for aioice.ice.Connection's data path.""" + + def __init__(self) -> None: + self.data: list[tuple[bytes, int]] = [] + + def data_received(self, data: bytes, component: int) -> None: + self.data.append((data, component)) + + class _RecordingProtocol: """Stand-in for aioice.ice.StunProtocol — records what it receives.""" @@ -109,14 +119,14 @@ async def _unknown_ufrag_handler(self) -> None: mux, _ = await UdpMux.create("127.0.0.1", 0) seen: list[tuple[str, bytes, tuple[str, int]]] = [] mux.set_unknown_stun_handler( - lambda ufrag, data, addr: seen.append((ufrag, data, addr)) + lambda username, data, addr: seen.append((username, data, addr)) ) try: data = _stun_binding_request("newdialer:remote") mux.datagram_received(data, ("10.0.0.9", 5555)) # First contact for an unregistered ufrag reaches the handler with - # the ufrag, raw datagram (for replay), and source address. - assert seen == [("newdialer", data, ("10.0.0.9", 5555))] + # the full USERNAME, raw datagram (for replay), and source address. + assert seen == [("newdialer:remote", data, ("10.0.0.9", 5555))] finally: await mux.close() @@ -194,6 +204,87 @@ async def _unregister_addr(self) -> None: # --------------------------------------------------------------------------- +class TestAddrLearning: + def test_stun_for_known_ufrag_learns_remote_addr(self) -> None: + asyncio.run(self._learns()) + + async def _learns(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + remote = ("10.0.0.9", 40000) + mux.register("abc123", proto) + try: + req = _stun_binding_request("abc123:peer") # random txn id per call + mux.datagram_received(req, remote) + # A DTLS record from the same address now routes without an + # explicit register_addr(). + mux.datagram_received(_dtls_like_bytes(), remote) + assert [d for d, _ in proto.received] == [req, _dtls_like_bytes()] + finally: + await mux.close() + + def test_learned_addrs_are_capped_per_protocol(self) -> None: + asyncio.run(self._capped()) + + async def _capped(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() + mux.register("abc123", proto) + try: + cap = UdpMux._MAX_ADDRS_PER_PROTOCOL + for port in range(40000, 40000 + cap + 20): + mux.datagram_received( + _stun_binding_request("abc123:peer"), ("10.0.0.9", port) + ) + learned = [a for a, p in mux._by_addr.items() if p is proto] + assert len(learned) == cap + # STUN with the known ufrag still reaches the protocol regardless. + assert len(proto.received) == cap + 20 + finally: + await mux.close() + + def test_newer_connection_takes_over_reused_addr(self) -> None: + asyncio.run(self._takeover()) + + async def _takeover(self) -> None: + # A dialer that redials from the same (ip, port) with a new ufrag + # while the old connection is still registered must reach the new one. + mux, _ = await UdpMux.create("127.0.0.1", 0) + old, new = _RecordingProtocol(), _RecordingProtocol() + remote = ("10.0.0.9", 40000) + mux.register("oldufrag", old) + mux.register("newufrag", new) + try: + mux.datagram_received(_stun_binding_request("oldufrag:x"), remote) + mux.datagram_received(_stun_binding_request("newufrag:x"), remote) + mux.datagram_received(_dtls_like_bytes(), remote) + assert [d for d, _ in new.received][-1] == _dtls_like_bytes() + assert all(d != _dtls_like_bytes() for d, _ in old.received) + mux.unregister("newufrag") + assert remote not in mux._by_addr + finally: + await mux.close() + + def test_unregister_drops_learned_addrs(self) -> None: + asyncio.run(self._unregister_drops()) + + async def _unregister_drops(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto, other = _RecordingProtocol(), _RecordingProtocol() + remote, other_remote = ("10.0.0.9", 40000), ("10.0.0.10", 40001) + mux.register("abc123", proto) + mux.register_addr(other_remote, other) + try: + mux.datagram_received(_stun_binding_request("abc123:peer"), remote) + mux.unregister("abc123") + assert mux._by_ufrag == {} + assert mux._by_addr == {other_remote: other} # unrelated entry kept + mux.datagram_received(_dtls_like_bytes(), remote) + assert len(proto.received) == 1 # nothing after unregister + finally: + await mux.close() + + class TestMalformedStun: def test_struct_error_routes_by_addr_not_raise(self) -> None: asyncio.run(self._struct_error()) @@ -205,8 +296,12 @@ async def _struct_error(self) -> None: # usable STUN" and fall through to addr routing instead of letting the # exception escape — otherwise a remote peer can crash / log-flood the # mux with crafted packets. + # aioice's own StunProtocol.datagram_received re-parses and catches + # only ValueError, so the bytes must NOT go through it either — they + # are handed straight to the connection's data path (``receiver``). mux, _ = await UdpMux.create("127.0.0.1", 0) proto = _RecordingProtocol() + proto.receiver = _RecordingReceiver() # type: ignore[attr-defined] remote = ("10.0.0.2", 54321) mux.register_addr(remote, proto) try: @@ -216,7 +311,25 @@ async def _struct_error(self) -> None: ): # Must not raise. mux.datagram_received(data, remote) - assert proto.received == [(data, remote)] + assert proto.received == [] # StunProtocol parser bypassed + assert proto.receiver.data == [(data, 1)] # type: ignore[attr-defined] + finally: + await mux.close() + + def test_struct_error_without_receiver_is_dropped(self) -> None: + asyncio.run(self._struct_error_no_receiver()) + + async def _struct_error_no_receiver(self) -> None: + mux, _ = await UdpMux.create("127.0.0.1", 0) + proto = _RecordingProtocol() # no .receiver attribute + remote = ("10.0.0.2", 54321) + mux.register_addr(remote, proto) + try: + with patch.object( + _stun, "parse_message", side_effect=struct.error("bad attr") + ): + mux.datagram_received(b"\x00\x01" + b"\x00" * 26, remote) + assert proto.received == [] finally: await mux.close() @@ -369,3 +482,116 @@ async def _stun_for_conn(self) -> None: assert len(recorder.received) == 1 finally: await mux.close() + + +# --------------------------------------------------------------------------- +# attach_muxed_connection: full aiortc PC over the mux +# --------------------------------------------------------------------------- + + +class TestAttachMuxedConnection: + def test_pc_over_mux_connects_and_routes(self) -> None: + asyncio.run(asyncio.wait_for(self._pc_over_mux(), timeout=30)) + + async def _pc_over_mux(self) -> None: + # Bind the mux on all interfaces and advertise a real host address: + # the client PC gathers LAN host candidates only (aioice skips + # loopback), and on Windows a 127.0.0.1-bound socket cannot send to a + # LAN-bound peer socket (WinError 1231) - Linux's weak-host model hid + # this. Mirrors a real listener on 0.0.0.0. + from aioice.ice import get_host_addresses + from aiortc import RTCSessionDescription + + from libp2p.transport.webrtc._aiortc_helpers import ( + _wait_channel_open, + attach_muxed_connection, + close_peer_connection, + create_noise_channel, + create_peer_connection, + wait_for_connected, + ) + from libp2p.transport.webrtc.certificate import WebRTCCertificate + + hosts = get_host_addresses(use_ipv4=True, use_ipv6=False) or ["127.0.0.1"] + cand_host = hosts[0] + mux, port = await UdpMux.create("0.0.0.0", 0) + ufrag, pwd = "srvufrag1", "srvpassword1234567890ab" + cert_s, cert_c = ( + WebRTCCertificate.from_aiortc(), + WebRTCCertificate.from_aiortc(), + ) + pc_s = pc_c = None + try: + conn = mux.add_ice_connection(ufrag, pwd, host=cand_host) + pc_s = await create_peer_connection(cert_s._rtc_certificate, ice_servers=[]) + ch_s = await create_noise_channel(pc_s) + attach_muxed_connection(pc_s, mux, conn) + + pc_c = await create_peer_connection(cert_c._rtc_certificate, ice_servers=[]) + ch_c = await create_noise_channel(pc_c) + + offer = await pc_c.createOffer() + await pc_c.setLocalDescription(offer) + # Force the muxed side to be the DTLS server (WebRTC-Direct + # listeners must be): with actpass aiortc answers as DTLS client. + offer_sdp = pc_c.localDescription.sdp.replace( + "a=setup:actpass", "a=setup:active" + ) + await pc_s.setRemoteDescription( + RTCSessionDescription(sdp=offer_sdp, type="offer") + ) + answer = await pc_s.createAnswer() + await pc_s.setLocalDescription(answer) + answer_sdp = pc_s.localDescription.sdp + await pc_c.setRemoteDescription( + RTCSessionDescription(sdp=answer_sdp, type="answer") + ) + + # The answer advertises exactly the shared port and our creds — + # aiortc's gather() did not bind/append any extra sockets. + cands = [ + ln for ln in answer_sdp.splitlines() if ln.startswith("a=candidate") + ] + assert len(cands) == 1 and f" {port} " in cands[0], cands + assert f"a=ice-ufrag:{ufrag}" in answer_sdp + assert f"a=ice-pwd:{pwd}" in answer_sdp + assert "a=setup:passive" in answer_sdp + assert len(conn._protocols) == 1 + + await asyncio.gather( + wait_for_connected(pc_s, timeout=15.0), + wait_for_connected(pc_c, timeout=15.0), + ) + + # Data flows both ways over the shared socket. + await asyncio.gather(_wait_channel_open(ch_s), _wait_channel_open(ch_c)) + got_s: asyncio.Queue[bytes] = asyncio.Queue() + got_c: asyncio.Queue[bytes] = asyncio.Queue() + ch_s.on("message", lambda m: got_s.put_nowait(m)) + ch_c.on("message", lambda m: got_c.put_nowait(m)) + ch_c.send(b"client->server") + assert await asyncio.wait_for(got_s.get(), 5) == b"client->server" + ch_s.send(b"server->client") + assert await asyncio.wait_for(got_c.get(), 5) == b"server->client" + + # Peer address learned/registered for post-ICE routing. + assert ufrag in mux._by_ufrag + assert conn._nominated[1].remote_addr in mux._by_addr + + await close_peer_connection(pc_c) + await close_peer_connection(pc_s) + pc_s = pc_c = None + # closing the ICE transport unregisters everything for this ufrag + assert mux._by_ufrag == {} + assert mux._by_addr == {} + finally: + # Bound cleanup: a hung close() after a primary failure would + # otherwise outlive asyncio.wait_for and be killed by pytest-timeout + # (reported as an xdist "worker crashed", hiding the real error). + for pc in (pc_s, pc_c): + if pc is not None: + try: + await close_peer_connection(pc) + except Exception: + pass + await asyncio.wait_for(mux.close(), 10) diff --git a/tests/core/transport/webrtc/test_webrtc_direct_loopback.py b/tests/core/transport/webrtc/test_webrtc_direct_loopback.py index c2356aa46..0d45fa28c 100644 --- a/tests/core/transport/webrtc/test_webrtc_direct_loopback.py +++ b/tests/core/transport/webrtc/test_webrtc_direct_loopback.py @@ -2,16 +2,17 @@ Integration tests for WebRTC Direct listener setup (aiortc required). Verifies that a listener advertises a multiaddr with /certhash/ and /p2p/, -binds a real UDP port when port 0 is requested, and uses an aiortc-native -certificate. Does not exercise dial(), Noise, or stream echo — see -test_multiplexing_loopback.py for data-channel layer loopback and add a -transport-level loopback test in a follow-up. +binds a real port when port 0 is requested, uses an aiortc-native +certificate, and — end to end — that dial() reaches the listener's handler +through ICE/DTLS + Noise (server initiates, dialer responds) and a stream +echoes. See test_multiplexing_loopback.py for data-channel layer loopback. """ # pyrefly: ignore from __future__ import annotations import pytest +import trio try: import aiortc # noqa: F401 @@ -26,6 +27,33 @@ pytestmark = pytest.mark.skipif(not HAS_AIORTC, reason="aiortc not installed") +def _listen_addr() -> str: + """ + Listen on a concrete non-loopback IPv4 interface when one exists. + + aiortc dialers gather LAN host candidates only (aioice skips loopback) and + on Windows a LAN-bound UDP socket cannot send to 127.0.0.1 at all + (WinError 1214), so a listener advertising 127.0.0.1 is unreachable there. + Linux's weak-host model hides this. Falls back to loopback when the host + has no other IPv4 address. + """ + from libp2p.utils.address_validation import get_available_interfaces + + for maddr in get_available_interfaces(0, "udp"): + ip = maddr.value_for_protocol("ip4") if "/ip4/" in str(maddr) else None + if ip and ip != "127.0.0.1": + return f"/ip4/{ip}/udp/0/webrtc-direct" + return "/ip4/127.0.0.1/udp/0/webrtc-direct" + + +LISTEN_ADDR = _listen_addr() + + +def _host_port(maddr) -> tuple[str, int]: # type: ignore[no-untyped-def] + parts = str(maddr).split("/") + return parts[2], int(parts[4]) + + @pytest.mark.trio async def test_listener_advertises_certhash_in_multiaddr(): """Listener publishes a multiaddr that contains /certhash/ and /p2p/.""" @@ -65,7 +93,7 @@ async def noop_handler(conn: object) -> None: listener = transport.create_listener(noop_handler) from multiaddr import Multiaddr - await listener.listen(Multiaddr("/ip4/127.0.0.1/udp/0/webrtc-direct")) + await listener.listen(Multiaddr(LISTEN_ADDR)) addrs = listener.get_addrs() addr_str = str(addrs[0]) @@ -87,3 +115,286 @@ async def test_certificate_is_aiortc_native(): assert hasattr(transport.certificate, "_rtc_certificate") assert transport.certificate._rtc_certificate is not None await transport.close() + + +def _transport(**cfg): # type: ignore[no-untyped-def] + from libp2p.transport.webrtc.config import WebRTCTransportConfig + + # ice_servers=[] keeps loopback local (no external STUN round-trip). + kp = create_new_key_pair() + return WebRTCDirectTransport( + private_key=kp.private_key, + config=WebRTCTransportConfig(ice_servers=[], **cfg), + ), kp + + +@pytest.mark.trio +@pytest.mark.parametrize("harness", [False, True], ids=["stun", "sdp-http-harness"]) +async def test_dial_listen_open_stream_echo(harness): + """ + Full transport loopback: dial → ICE/DTLS → Noise (server initiates, + dialer responds) → handler gets an authenticated connection → stream echo. + Runs over the spec STUN path (default) and the experimental HTTP harness. + """ + from multiaddr import Multiaddr + + from libp2p.peer.id import ID + + server, server_kp = _transport(enable_sdp_http_harness=harness) + dialer, dialer_kp = _transport(enable_sdp_http_harness=harness) + server_id = ID.from_pubkey(server_kp.public_key) + dialer_id = ID.from_pubkey(dialer_kp.public_key) + + seen_by_server: list[ID] = [] + handler_done = trio.Event() + + async def echo_handler(conn) -> None: # type: ignore[no-untyped-def] + seen_by_server.append(conn.peer_id) + stream = await conn.accept_stream() + data = await stream.read() + await stream.write(data) + await handler_done.wait() + + listener = server.create_listener(echo_handler) + await listener.listen(Multiaddr(LISTEN_ADDR)) + (maddr,) = listener.get_addrs() + assert listener._mux is not None # STUN path is always on + assert (listener._signaling_server is not None) is harness + + try: + with trio.fail_after(30): + conn = await dialer.dial(maddr) + assert conn.peer_id == server_id + stream = await conn.open_stream() + await stream.write(b"ping-over-webrtc-direct") + assert await stream.read() == b"ping-over-webrtc-direct" + assert seen_by_server == [dialer_id] + handler_done.set() + await conn.close() + finally: + await listener.close() + await dialer.close() + await server.close() + + +@pytest.mark.trio +async def test_concurrent_dials_share_one_udp_port(): + """Two dialers hit the same advertised port; the mux demuxes by ufrag.""" + from multiaddr import Multiaddr + + from libp2p.peer.id import ID + + server, _ = _transport() + dialers = [_transport() for _ in range(2)] + dialer_ids = {ID.from_pubkey(kp.public_key) for _, kp in dialers} + seen: set[ID] = set() + release = trio.Event() + + async def handler(conn) -> None: # type: ignore[no-untyped-def] + seen.add(conn.peer_id) + stream = await conn.accept_stream() + await stream.write(await stream.read()) + await release.wait() + + listener = server.create_listener(handler) + await listener.listen(Multiaddr(LISTEN_ADDR)) + (maddr,) = listener.get_addrs() + conns = [] + + async def _dial(t) -> None: # type: ignore[no-untyped-def] + conn = await t.dial(maddr) + stream = await conn.open_stream() + payload = f"hello from {id(t)}".encode() + await stream.write(payload) + assert await stream.read() == payload + conns.append(conn) + + try: + with trio.fail_after(30): + async with trio.open_nursery() as nursery: + for t, _ in dialers: + nursery.start_soon(_dial, t) + assert seen == dialer_ids + # Both dials went through the one mux socket: two ufrags registered. + assert len(listener._mux._by_ufrag) == 2 + release.set() + for c in conns: + await c.close() + finally: + await listener.close() + for t, _ in dialers: + await t.close() + await server.close() + + +@pytest.mark.trio +async def test_unknown_version_prefix_is_rejected(): + """A first contact without a libp2p+webrtc+vN/ prefix must not create a PC.""" + import asyncio + + from multiaddr import Multiaddr + + from libp2p.transport.webrtc._udp_mux import UdpMux + from tests.core.transport.webrtc.test_udp_mux import _stun_binding_request + + server, _ = _transport() + listener = server.create_listener(lambda conn: trio.sleep_forever()) + await listener.listen(Multiaddr(LISTEN_ADDR)) + (maddr,) = listener.get_addrs() + host, port = _host_port(maddr) + bridge = await server._ensure_bridge() + + async def _poke() -> None: + # Fresh UDP endpoint on the asyncio thread; send crafted requests. + loop = asyncio.get_running_loop() + transport, _ = await loop.create_datagram_endpoint( + asyncio.DatagramProtocol, local_addr=("0.0.0.0", 0) + ) + try: + for username in ("noprefix1:cli1", "libp2p+webrtc+v9/abcd:cli1", "x:y"): + transport.sendto(_stun_binding_request(username), (host, port)) + await asyncio.sleep(0.2) + finally: + transport.close() + + try: + await bridge.run_coro(_poke()) + assert isinstance(listener._mux, UdpMux) + assert listener._mux._by_ufrag == {} + assert listener._in_flight == 0 + finally: + await listener.close() + await server.close() + + +@pytest.mark.trio +async def test_in_flight_cap_drops_excess_first_contacts(): + """Beyond max_in_flight_connections, new first contacts are dropped.""" + import asyncio + + from multiaddr import Multiaddr + + from tests.core.transport.webrtc.test_udp_mux import _stun_binding_request + + server, _ = _transport(max_in_flight_connections=1) + listener = server.create_listener(lambda conn: trio.sleep_forever()) + await listener.listen(Multiaddr(LISTEN_ADDR)) + (maddr,) = listener.get_addrs() + host, port = _host_port(maddr) + bridge = await server._ensure_bridge() + + async def _poke() -> None: + loop = asyncio.get_running_loop() + transport, _ = await loop.create_datagram_endpoint( + asyncio.DatagramProtocol, local_addr=("0.0.0.0", 0) + ) + try: + for i in range(3): + cred = f"libp2p+webrtc+v1/{'a' * 20}{i}" + transport.sendto(_stun_binding_request(f"{cred}:{cred}"), (host, port)) + await asyncio.sleep(0.3) + finally: + transport.close() + + try: + await bridge.run_coro(_poke()) + assert listener._in_flight == 1 + assert len(listener._mux._by_ufrag) == 1 + assert len(listener._accept_tasks) == 1 + # close() cancels the in-flight setup instead of letting it wait out + # handshake_timeout. + with trio.fail_after(5): + await listener.close() + assert listener._accept_tasks == set() + finally: + await listener.close() + await server.close() + + +@pytest.mark.trio +async def test_listen_bind_failure_raises_connection_error(): + """A port already in use surfaces as WebRTCConnectionError, no orphans.""" + from multiaddr import Multiaddr + + from libp2p.transport.webrtc.exceptions import WebRTCConnectionError + + server, _ = _transport() + first = server.create_listener(lambda conn: trio.sleep_forever()) + await first.listen(Multiaddr("/ip4/127.0.0.1/udp/0/webrtc-direct")) + (maddr,) = first.get_addrs() + port = int(str(maddr).split("/udp/")[1].split("/")[0]) + second = server.create_listener(lambda conn: trio.sleep_forever()) + try: + with pytest.raises(WebRTCConnectionError): + await second.listen(Multiaddr(f"/ip4/127.0.0.1/udp/{port}/webrtc-direct")) + assert second._nursery is None and second._mux is None + finally: + await first.close() + await server.close() + + +@pytest.mark.trio +async def test_dial_rejects_wrong_p2p_id(): + """A /p2p/ component that doesn't match the authenticated server fails.""" + from multiaddr import Multiaddr + + from libp2p.peer.id import ID + from libp2p.transport.webrtc.exceptions import WebRTCConnectionError + + server, _ = _transport() + dialer, _ = _transport() + + async def handler(conn) -> None: # type: ignore[no-untyped-def] + await trio.sleep_forever() + + listener = server.create_listener(handler) + await listener.listen(Multiaddr(LISTEN_ADDR)) + (maddr,) = listener.get_addrs() + impostor = ID.from_pubkey(create_new_key_pair().public_key) + wrong = Multiaddr(str(maddr).rsplit("/p2p/", 1)[0] + f"/p2p/{impostor}") + + try: + with trio.fail_after(30): + with pytest.raises(WebRTCConnectionError): + await dialer.dial(wrong) + finally: + await listener.close() + await dialer.close() + await server.close() + + +@pytest.mark.trio +async def test_handler_exception_does_not_crash_listener(): + """ + A raising handler must only affect its own connection: the listener + nursery lives in a trio system task, so an escaping exception would + otherwise abort the whole trio run. + """ + from multiaddr import Multiaddr + + server, _ = _transport() + dialer, _ = _transport() + calls = 0 + + async def bad_handler(conn) -> None: # type: ignore[no-untyped-def] + nonlocal calls + calls += 1 + raise RuntimeError("boom") + + listener = server.create_listener(bad_handler) + await listener.listen(Multiaddr(LISTEN_ADDR)) + (maddr,) = listener.get_addrs() + try: + with trio.fail_after(30): + conn = await dialer.dial(maddr) # handshake completes before handler + while calls == 0: + await trio.sleep(0.01) + await conn.close() + # Listener still alive and usable. + conn2 = await dialer.dial(maddr) + await conn2.close() + assert calls == 2 + finally: + await listener.close() + await dialer.close() + await server.close() diff --git a/tox.ini b/tox.ini index 4d2a1b6a3..2189e8522 100644 --- a/tox.ini +++ b/tox.ini @@ -18,7 +18,7 @@ per-file-ignores=__init__.py:F401 usedevelop=True commands_pre= uv pip install --upgrade pip - uv pip install --group dev -e . + uv pip install --group dev -e ".[webrtc]" commands= core: pytest -n auto --timeout=1200 --durations=40 --durations-min=1.0 {posargs:tests/core} demos: pytest -n auto --timeout=1200 --durations=40 --durations-min=1.0 {posargs:tests/examples}