diff --git a/.github/workflows/ts.yml b/.github/workflows/ts.yml index ac34aa39d36..b289358039e 100644 --- a/.github/workflows/ts.yml +++ b/.github/workflows/ts.yml @@ -25,6 +25,7 @@ on: # Workspace dependencies of moq-net, moq-tokio and moq-mux on the TS path. - "rs/kio/**" - "rs/moq-flate/**" + - "rs/moq-ice/**" - "rs/moq-msf/**" - "rs/moq-pattern/**" - "rs/moq-sock/**" diff --git a/Cargo.lock b/Cargo.lock index ecc6f8a8d14..e4070b5efcc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4689,6 +4689,17 @@ dependencies = [ "url", ] +[[package]] +name = "moq-ice" +version = "0.1.0" +dependencies = [ + "str0m", + "stun-types", + "thiserror 2.0.21", + "tokio", + "tracing", +] + [[package]] name = "moq-json" version = "0.5.8" @@ -4949,6 +4960,7 @@ dependencies = [ "axum", "bytes", "hang", + "moq-ice", "moq-mux", "moq-net", "moq-tokio", @@ -5083,11 +5095,13 @@ dependencies = [ "hmac 0.13.0", "http", "humantime", + "if-addrs", "jni 0.22.4", "kio 0.6.2", "libc", "mdns-sd", "moq-clock", + "moq-ice", "moq-net", "moq-quic", "moq-sock", @@ -5107,6 +5121,8 @@ dependencies = [ "serde_with", "sha2 0.11.0", "socket2 0.6.5", + "str0m", + "stun-types", "subtle", "tempfile", "thiserror 2.0.21", diff --git a/Cargo.toml b/Cargo.toml index e6c06ec0ef1..c8bac8f17cc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,7 @@ members = [ "rs/moq-flate", "rs/moq-gst", "rs/moq-hls", + "rs/moq-ice", "rs/moq-json", "rs/moq-loc", "rs/moq-msf", @@ -61,6 +62,7 @@ default-members = [ # "rs/moq-gst", # requires GStreamer "rs/moq-flate", "rs/moq-hls", + "rs/moq-ice", "rs/moq-json", "rs/moq-loc", "rs/moq-msf", @@ -133,6 +135,8 @@ hmac = "0.13" humantime = "2.3" humantime-serde = "1.1.1" hyper-util = { version = "0.1.21", features = ["tokio"] } +# Interface addresses for WebRTC host candidates (moq-tokio's `webrtc` feature). +if-addrs = "0.15" io-uring = "0.7.15" jsonwebtoken = "11" kio = { version = "0.6.2", path = "rs/kio" } @@ -153,6 +157,7 @@ moq-clock = { version = "0.11.0", path = "rs/moq-clock" } moq-clock-macros = { version = "0.1.0", path = "rs/moq-clock/macros" } moq-flate = { version = "0.2.0", path = "rs/moq-flate" } moq-hls = { version = "0.5.11", path = "rs/moq-hls", default-features = false } +moq-ice = { version = "0.1.0", path = "rs/moq-ice" } moq-json = { version = "0.5.8", path = "rs/moq-json" } moq-loc = { version = "0.2.19", path = "rs/moq-loc" } moq-msf = { version = "0.5.1", path = "rs/moq-msf" } @@ -227,6 +232,10 @@ serde_with = "3" sha2 = "0.11" slab = "0.4.12" socket2 = "0.6" +# Sans-IO WebRTC. The default `examples` feature pulls in a web server, and aws-lc-rs +# is the provider rustls already brings. +str0m = { version = "0.24.1", default-features = false, features = ["aws-lc-rs"] } +stun-types = "2.0.3" subtle = "2.6" tempfile = "3" thiserror = "2" diff --git a/doc/concept/transport.md b/doc/concept/transport.md index cc0b6d6475a..dd41787939f 100644 --- a/doc/concept/transport.md +++ b/doc/concept/transport.md @@ -71,6 +71,12 @@ records, with no datagrams. The application supplies signaling and ICE and keeps ownership of the peer connection; closing the transport closes only its channel. SCTP ordering means one lost packet can hold back every stream. +Native peers use `moq-tokio`'s `webrtc` feature, which runs its own ICE agent: +host candidates on every interface, server-reflexive candidates from the STUN +servers it is given, and mDNS resolution of a browser's `.local` candidates. +The application still carries the offer, answer, and candidates; the offering +peer connects as the MoQ client and the answering peer accepts. + ## iroh (peer-to-peer, experimental) [iroh](https://www.iroh.computer/) swaps "dial a hostname" for "dial a public diff --git a/doc/lib/rs/index.md b/doc/lib/rs/index.md index 171427dc9b3..8690d4c9edb 100644 --- a/doc/lib/rs/index.md +++ b/doc/lib/rs/index.md @@ -16,6 +16,7 @@ The reference implementation. Every crate is on | [moq-pattern](https://docs.rs/moq-pattern) | Path patterns such as `room/**`. Re-exported by moq-net and moq-auth. | | [moq-tokio](https://docs.rs/moq-tokio) | Connects and listens over QUIC, WebSocket, or iroh, from config or CLI flags. | | [moq-sock](https://docs.rs/moq-sock) | Socket binding and CPU pinning for thread-per-core listeners. | +| [moq-ice](https://docs.rs/moq-ice) | The ICE agent behind native WebRTC: drives a str0m connection over UDP, with a STUN client. | | [moq-uring](https://docs.rs/moq-uring) | Experimental Linux io\_uring server for moq-lite. | | [hang](/lib/rs/hang) | The media layer: catalog, containers, ordered frame delivery. | | [moq-mux](/lib/rs/moq-mux) | Import and export fMP4/CMAF, MPEG-TS, Matroska, FLV, and Annex-B. | diff --git a/quest/m2/p2p/README.md b/quest/m2/p2p/README.md index 2e1f0446b59..00e0e2bcc49 100644 --- a/quest/m2/p2p/README.md +++ b/quest/m2/p2p/README.md @@ -148,7 +148,6 @@ Backend choice does not change the policy that watching tabs never re-serve. - [Peer grants](/quest/m2/p2p/peer-grant.md) - the relay issues a node-bound, asymmetrically signed grant a browser can verify; HS256 keys issue none - [Signaling and policy](/quest/m2/p2p/signal.md) - opted-in peers find each other under the prefix, the application picks who to dial, and the roster-size gate decides whether STUN is used -- [Native data channel transport](/quest/m2/p2p/webrtc.md) - `moq-tokio` holds a moq-net session with a browser over str0m with a full ICE agent - [UDP demux](/quest/m2/p2p/udp-demux.md) - one classifier splits QUIC from STUN and WebRTC on the relay's socket and on a native pair's - [Native pairs over ICE and QUIC](/quest/m2/p2p/native-quic.md) - two native peers connect through ICE, then `moq-quic` over the nominated path with fingerprints pinned from the offer - [moq-cli joins](/quest/m2/p2p/cli.md) - `--p2p` publishes a roster entry, dials native peers over ICE and QUIC, and serves browsers as a transit hop diff --git a/quest/m2/p2p/cli.md b/quest/m2/p2p/cli.md index c68851fbf15..2ad8969a151 100644 --- a/quest/m2/p2p/cli.md +++ b/quest/m2/p2p/cli.md @@ -41,6 +41,5 @@ Docs: a "P2P" section in `doc/bin/cli.md` beside the LAN cluster one. ## Required -- [Native data channel transport](/quest/m2/p2p/webrtc.md) - [Native pairs over ICE and QUIC](/quest/m2/p2p/native-quic.md) - the native-to-native path - [Signaling and policy](/quest/m2/p2p/signal.md) - the roster and offer schema this mirrors diff --git a/quest/m2/p2p/native-quic.md b/quest/m2/p2p/native-quic.md index 2b2466e26eb..4d05a99735e 100644 --- a/quest/m2/p2p/native-quic.md +++ b/quest/m2/p2p/native-quic.md @@ -7,7 +7,7 @@ selected by their applications connect directly across a NAT: ICE finds a path through host and server-reflexive candidates, then a `moq-quic` connection runs over the nominated pair, with each side's certificate pinned by the fingerprint the other sent in its offer or answer. A pair with a -browser still uses a [data channel](/quest/m2/p2p/webrtc.md). +browser still uses a data channel (`moq_tokio::webrtc`). ## Plan @@ -18,9 +18,10 @@ Decided with the maintainer on 2026-10-13: the relay as rendezvous and fallback. - ICE is str0m's standalone sans-IO agent, the `is` crate (0.11, already in the lockfile through str0m; str0m re-exports it as `str0m::ice`). Gather - candidates through the module [webrtc](/quest/m2/p2p/webrtc.md) lifts out - of `moq-rtc`, with host candidates, `.local` resolution, and the STUN - Binding client; don't copy it. + candidates with `moq_tokio::webrtc`'s host gathering, `.local` resolution, + and the STUN Binding client in `moq-ice`; don't copy them. The first two + are private to the data channel transport today, so move them into + `moq-ice` rather than export them from `moq-tokio`. - QUIC is `moq-quic`, fed datagrams from the nominated pair. - Authentication: each side generates a self-signed certificate and sends its SHA-256 fingerprint in its offer or answer, as DTLS fingerprints ride @@ -60,5 +61,4 @@ covers LAN and NAT pairs between two `moq-cli --p2p` nodes. ## Required - [UDP demux](/quest/m2/p2p/udp-demux.md) - the classifier that splits STUN from QUIC -- [Native data channel transport](/quest/m2/p2p/webrtc.md) - the shared ICE gathering module - [Signaling and policy](/quest/m2/p2p/signal.md) - the offer schema and roster this extends diff --git a/quest/m2/p2p/webrtc.md b/quest/m2/p2p/webrtc.md deleted file mode 100644 index 92506d78bc1..00000000000 --- a/quest/m2/p2p/webrtc.md +++ /dev/null @@ -1,43 +0,0 @@ -# [L] Native data channel transport - -## Goal - -`moq-tokio` gains a `webrtc` feature: a data channel transport implementing -moq-net's poll `Session` over str0m, qmux over one ordered channel, with a -full ICE agent that gathers host candidates, resolves browsers' `.local` -candidates with mdns-sd, and gathers server-reflexive candidates from the -STUN servers it is given. `Client` and `Server` can hold a moq-net session -with a browser peer across a NAT. - -## Plan - -str0m is already a dependency through `moq-rtc` and mdns-sd through the -`mdns` feature; no new stack. The feature is off by default in `moq-tokio` -and on in `moq-cli`. - -ICE: a full agent with host candidates on every non-loopback interface as raw -addresses, a dual-stack socket per session, and a STUN Binding client for -each configured server so the agent offers server-reflexive candidates. -`moq-rtc` gathers host candidates only today; lift its run loop, socket -reader, and candidate pick into a module both crates share and add the STUN -client there, rather than copying it. Remote `.local` candidates are resolved -with an mdns-sd query before `add_remote_candidate`. - -Signaling is the caller's: the transport exposes the local description and a -stream of local candidates, and accepts the remote description and -candidates, as plain async methods. No callbacks. - -Transport: implement `qmux::transport::{Transport, Writer, Reader}` over one -reliable ordered str0m channel, the way `ws::Upgraded` does, and feed -`qmux::Session` through `transport::Session` like `websocket.rs`. One record -per message, `max_record_size` 16 KiB by default, clamped to the negotiated -message size. The channel is always ordered, with no knob. - -Tests: an in-process str0m pair over loopback runs moq-net's session tests; -a second pair puts a fake STUN server between them and asserts the reflexive -candidate is offered and selected. Browser interop is the harness's job. - -## Related - -- [moq-cli joins](/quest/m2/p2p/cli.md) - the first consumer -- [Native pairs over ICE and QUIC](/quest/m2/p2p/native-quic.md) - reuses the ICE gathering module this lifts out of `moq-rtc` diff --git a/rs/moq-auth/src/request.rs b/rs/moq-auth/src/request.rs index f16c97594fa..77f06509ab3 100644 --- a/rs/moq-auth/src/request.rs +++ b/rs/moq-auth/src/request.rs @@ -146,7 +146,7 @@ pub enum Transport { Rtmp, /// An SRT gateway session. Srt, - /// A WebRTC gateway session, including WHIP and WHEP. + /// A WebRTC session: a WHIP or WHEP gateway, or MoQ over a data channel. WebRtc, /// A WebTransport connection over HTTP/3. WebTransport, diff --git a/rs/moq-cli/Cargo.toml b/rs/moq-cli/Cargo.toml index 60920fa86a4..e0e8cec4199 100644 --- a/rs/moq-cli/Cargo.toml +++ b/rs/moq-cli/Cargo.toml @@ -26,7 +26,7 @@ path = "src/main.rs" doc = false [features] -default = ["cluster-lan", "quic", "websocket", "nvidia", "s3", "gcs", "azure"] +default = ["cluster-lan", "quic", "websocket", "webrtc", "nvidia", "s3", "gcs", "azure"] # Experimental transport with a heavy dependency tree, enabled only on request. iroh = ["moq-relay/iroh", "moq-tokio/iroh"] # LAN discovery and meshing (`--cluster-lan`). @@ -38,6 +38,8 @@ quic = ["moq-tokio/quic"] # congestion-control testing. qlog = ["moq-tokio/qlog"] websocket = ["moq-tokio/websocket"] +# Native WebRTC data channels, for holding sessions with browser peers. +webrtc = ["moq-tokio/webrtc"] # The cloud object stores `export archive` and `import archive` reach (`s3://`, # `gs://`, `az://`). Local `file://` URLs need none of them. s3 = ["object_store/aws"] diff --git a/rs/moq-ice/CHANGELOG.md b/rs/moq-ice/CHANGELOG.md new file mode 100644 index 00000000000..2a718d0b3af --- /dev/null +++ b/rs/moq-ice/CHANGELOG.md @@ -0,0 +1,12 @@ +# Changelog + +All notable changes to this project will be documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). + +## [Unreleased] + +### Added + +- First release: the ICE agent shared by `moq-tokio`'s WebRTC data channel transport and `moq-rtc`'s WHIP/WHEP gateway, with a STUN Binding client and per-peer wire counters. diff --git a/rs/moq-ice/Cargo.toml b/rs/moq-ice/Cargo.toml new file mode 100644 index 00000000000..e34bbdc354c --- /dev/null +++ b/rs/moq-ice/Cargo.toml @@ -0,0 +1,23 @@ +[package] +name = "moq-ice" +description = "Media over QUIC - drives a str0m WebRTC connection's ICE over a tokio UDP socket" +authors = ["Luke Curley"] +repository = "https://github.com/moq-dev/moq" +license = "MIT OR Apache-2.0" + +version = "0.1.0" +edition = "2024" +rust-version.workspace = true + +keywords = ["webrtc", "ice", "stun", "str0m", "moq"] +categories = ["network-programming", "multimedia"] + +[dependencies] +str0m = { workspace = true } +stun-types = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["macros", "net", "rt", "sync", "time"] } +tracing = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/rs/moq-ice/README.md b/rs/moq-ice/README.md new file mode 100644 index 00000000000..23d9f8b6a60 --- /dev/null +++ b/rs/moq-ice/README.md @@ -0,0 +1,12 @@ +# moq-ice + +Drives a [str0m](https://github.com/algesten/str0m) WebRTC connection over a +tokio UDP socket: the run loop that feeds it datagrams and timeouts, a STUN +Binding client for server-reflexive candidates, and per-peer wire counters. + +str0m is sans-IO and requires every mutation to be followed by draining its +output. `Agent::rtc_mut` hands out a guard that drains when dropped, so callers +can't forget. + +This is infrastructure, not an entry point: `moq-tokio`'s `webrtc` feature +builds MoQ data channels on it, and `moq-rtc` builds its WHIP/WHEP gateway. diff --git a/rs/moq-ice/src/agent.rs b/rs/moq-ice/src/agent.rs new file mode 100644 index 00000000000..20f310f2e53 --- /dev/null +++ b/rs/moq-ice/src/agent.rs @@ -0,0 +1,555 @@ +use std::collections::VecDeque; +use std::net::{IpAddr, SocketAddr}; +use std::ops::{Deref, DerefMut}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use str0m::net::{Protocol, Receive}; +use str0m::{Candidate, IceConnectionState, Input, Output, Rtc}; +use stun_types::attribute::XorMappedAddress; +use stun_types::message::{BINDING, Message, MessageClass, MessageWriteVec, TransactionId}; +use stun_types::prelude::*; +use tokio::net::UdpSocket; + +use crate::{Inbound, Socket, Wire}; + +/// Backstop deadline for reaching a connected ICE state, covering the one case +/// str0m's agent deliberately never times out: a peer that provides no remote +/// candidates and sends nothing. str0m ends a connection whose candidate pairs were +/// tried and exhausted (after ~21s at its defaults), but with no remote candidates +/// it treats the session as still possible forever, since trickle ICE could add +/// one. Set above str0m's own exhaustion so a slow but real peer is not clipped. +/// +/// It counts from when the peer could hold every local candidate: once gathering +/// ends, so a caller that sends its description only after the last candidate is +/// not clipped by an unreachable STUN server's retransmissions, and for an +/// [`Agent::awaiting_answer`], once the answer applies. +const ESTABLISH_TIMEOUT: Duration = Duration::from_secs(30); + +/// RFC 8489 section 6.2.1: the first retransmission timeout of a STUN request. +const STUN_RTO: Duration = Duration::from_millis(500); + +/// RFC 8489 section 6.2.1: how many times a STUN request is sent (`Rc`). +const STUN_SENDS: u32 = 7; + +/// RFC 8489 section 6.2.1: how long to wait after the last send, in RTOs (`Rm`). +const STUN_LAST_WAIT: u32 = 16; + +/// Why an [`Agent`] stopped. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum Error { + /// The peer disconnected, str0m closed the connection, or the inbound queue ended. + #[error("peer connection closed")] + Closed, + + /// ICE never connected before the establishment deadline. + #[error("ICE did not connect before the establishment deadline")] + Timeout, + + /// str0m failed (SDP, DTLS, SCTP, or ICE state). + #[error("rtc error: {0}")] + Rtc(String), +} + +impl Error { + fn rtc(err: impl std::fmt::Display) -> Self { + Self::Rtc(err.to_string()) + } +} + +/// What [`Agent::next`] surfaced. +#[derive(Debug)] +#[non_exhaustive] +// Moved once per poll; boxing would allocate for every data channel message. +#[allow(clippy::large_enum_variant)] +pub enum Event { + /// An event from str0m. ICE disconnection is reported as [`Error::Closed`] instead. + Rtc(str0m::Event), + /// A server-reflexive candidate, already added locally, to signal to the peer. + Candidate(Candidate), + /// Every STUN server answered or gave up: no more local candidates follow. + Gathered, +} + +/// A STUN Binding request in flight to one server. +struct Query { + server: SocketAddr, + /// The host candidate the request leaves from, which a reply's mapping is the reflexive form of. + base: SocketAddr, + id: TransactionId, + request: Vec, + sends: u32, + rto: Duration, + /// When to retransmit, or give up after the last send. + deadline: Instant, +} + +/// Drives one [`Rtc`] over a UDP socket until the connection ends. +pub struct Agent { + rtc: Rtc, + /// Send side, possibly shared by several agents (a demuxed server socket). + socket: Arc, + /// Whether `socket` is AF_INET6, which needs IPv4 destinations mapped. + socket_v6: bool, + /// The host candidates advertised to the peer. Each inbound datagram is tagged + /// with the one whose family matches its source: str0m drops a STUN request + /// whose destination is not a local candidate, and a wildcard bind is not one. + /// Never empty. + locals: Vec, + inbound: Inbound, + stun: Vec, + /// When every STUN query answered or gave up, once [`Event::Gathered`] went out. + gathered: Option, + events: VecDeque, + /// Why the connection ended, found while draining and reported by the next [`Agent::next`]. + failed: Option, + /// When str0m next wants a timeout, as of the last drain. + timeout: Instant, + /// Datagrams the socket refused for lack of buffer space, sent first next time. + outbox: VecDeque<(Vec, SocketAddr)>, + /// When the [`ESTABLISH_TIMEOUT`] started, or `None` while an offer awaits its answer. + started: Option, + connected: bool, +} + +impl Agent { + /// Drive `rtc` over `socket`, which must have a host candidate. + pub fn new(rtc: Rtc, socket: Socket) -> std::io::Result { + let Socket { + udp: socket, + locals, + inbound, + } = socket; + if locals.is_empty() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "no host candidate to receive on", + )); + } + let socket_v6 = socket.local_addr()?.is_ipv6(); + let now = Instant::now(); + let mut agent = Self { + rtc, + socket, + socket_v6, + locals, + inbound, + stun: Vec::new(), + gathered: None, + events: VecDeque::new(), + failed: None, + timeout: now, + outbox: VecDeque::new(), + started: Some(now), + connected: false, + }; + // Signaling already changed the connection. + agent.drain(); + Ok(agent) + } + + /// Ask each STUN server for this socket's public mapping, offering what comes + /// back as a server-reflexive candidate. Servers of a family with no host + /// candidate are skipped. + pub fn with_stun(mut self, servers: &[SocketAddr]) -> Self { + let now = Instant::now(); + for &server in servers { + let server = canonical(server); + let Some(base) = self.locals.iter().find(|l| l.is_ipv4() == server.is_ipv4()).copied() else { + tracing::debug!(%server, "no host candidate of this family; skipping STUN server"); + continue; + }; + let request = Message::builder_request(BINDING, MessageWriteVec::new()); + let id = request.transaction_id(); + self.stun.push(Query { + server, + base, + id, + request: request.finish(), + sends: 0, + rto: STUN_RTO, + deadline: now, + }); + } + self + } + + /// The connection being driven, read only. + pub fn rtc(&self) -> &Rtc { + &self.rtc + } + + /// Hold off the deadline for ICE to connect until the first [`Agent::rtc_mut`], + /// for an offer whose answer has yet to arrive: signaling time is the caller's. + pub fn awaiting_answer(mut self) -> Self { + self.started = None; + self + } + + /// Change the connection through a guard that drains what the change produced + /// when dropped: transmits are sent, and events wait for [`Agent::next`]. + pub fn rtc_mut(&mut self) -> RtcMut<'_> { + self.started.get_or_insert_with(Instant::now); + self.change() + } + + /// [`Agent::rtc_mut`] for the agent's own changes, which never start the deadline. + fn change(&mut self) -> RtcMut<'_> { + RtcMut(self) + } + + /// The counters of the bytes exchanged with the peer. + pub fn wire(&self) -> &Wire { + &self.inbound.wire + } + + /// Drive the connection until it surfaces an [`Event`]. + /// + /// Cancel safe: the connection is always drained between calls, and a datagram + /// the socket could not take yet is kept and sent first on the next call. + pub async fn next(&mut self) -> Result { + loop { + if let Some(event) = self.events.pop_front() { + return Ok(event); + } + if let Some(err) = self.failed.take() { + return Err(err); + } + if self.gathered.is_none() && self.stun.is_empty() { + self.gathered = Some(Instant::now()); + return Ok(Event::Gathered); + } + + while let Some((data, dst)) = self.outbox.front() { + match self.socket.send_to(data, *dst).await { + Ok(len) => self.inbound.wire.sent(len), + Err(err) => tracing::warn!(%err, dst = %dst, "send failed"), + } + self.outbox.pop_front(); + } + + // A dead Rtc (DTLS failure, explicit disconnect) reports a never-firing + // timeout instead of an error, which would park this loop forever. + if !self.rtc.is_alive() { + return Err(Error::Closed); + } + let establish = self.establish(); + if establish.is_some_and(|at| at <= Instant::now()) { + return Err(Error::Timeout); + } + + let deadline = self.stun.iter().map(|q| q.deadline).chain(establish); + let deadline = deadline.fold(self.timeout, Instant::min); + + let now = Instant::now(); + if deadline > now { + tokio::select! { + packet = self.inbound.packets.recv() => { + let Some((data, src)) = packet else { + // Every sender dropped: nothing more will arrive. + return Err(Error::Closed); + }; + self.receive(&data, src); + continue; + } + _ = tokio::time::sleep(deadline - now) => {} + } + } + + let now = Instant::now(); + let result = self.change().handle_input(Input::Timeout(now)); + if let Err(err) = result { + self.fail(Error::rtc(err)); + } + self.retransmit(now); + } + } + + /// When [`ESTABLISH_TIMEOUT`] expires, or `None` while it is held or ICE connected. + fn establish(&self) -> Option { + match (self.started, self.gathered) { + (Some(started), Some(gathered)) if !self.connected => Some(started.max(gathered) + ESTABLISH_TIMEOUT), + _ => None, + } + } + + /// Poll str0m until it asks for a timeout: transmits are sent, events queued. + fn drain(&mut self) { + while self.failed.is_none() { + let output = match self.rtc.poll_output() { + Ok(output) => output, + Err(err) => return self.fail(Error::rtc(err)), + }; + match output { + Output::Timeout(timeout) => { + self.timeout = timeout; + return; + } + Output::Transmit(transmit) => self.send(&transmit.contents, transmit.destination), + Output::Event(event) => { + if let str0m::Event::IceConnectionStateChange(state) = &event { + tracing::debug!(?state, "ice state"); + if *state == IceConnectionState::Disconnected { + return self.fail(Error::Closed); + } + self.connected |= state.is_connected(); + } + self.events.push_back(Event::Rtc(event)); + } + } + } + } + + /// Keep the first reason the connection ended. + fn fail(&mut self, err: Error) { + self.failed.get_or_insert(err); + } + + /// Send without waiting, keeping the datagram for later when the socket is full. + fn send(&mut self, data: &[u8], dst: SocketAddr) { + let dst = to_family(dst, self.socket_v6); + match self.socket.try_send_to(data, dst) { + Ok(len) => self.inbound.wire.sent(len), + // Changes can keep coming while the socket stays full, so the backlog is + // bounded like the inbound queue; ICE, SCTP, and RTP recover the loss. + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock && self.outbox.len() < crate::INBOX => { + self.outbox.push_back((data.to_vec(), dst)) + } + Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => { + tracing::debug!(%dst, len = data.len(), "dropped a datagram: the socket is full") + } + Err(err) => tracing::warn!(%err, %dst, "send failed"), + } + } + + fn receive(&mut self, data: &[u8], src: SocketAddr) { + if self.stun_reply(data, src) { + return; + } + + let local = pick_local(&self.locals, src); + // Anything can arrive on a socket a server-reflexive candidate exposes, so a + // datagram that is not STUN, DTLS, or RTP is noise, not a failure. + let Ok(receive) = Receive::new(Protocol::Udp, src, local, data) else { + tracing::debug!(%src, len = data.len(), "dropped unrecognized datagram"); + return; + }; + let result = self.change().handle_input(Input::Receive(Instant::now(), receive)); + if let Err(err) = result { + self.fail(Error::rtc(err)); + } + } + + /// Consume a reply to one of our STUN queries, returning false for anything else. + fn stun_reply(&mut self, data: &[u8], src: SocketAddr) -> bool { + if self.stun.is_empty() { + return false; + } + let Ok(reply) = Message::from_bytes(data) else { + return false; + }; + let id = reply.transaction_id(); + let Some(index) = self.stun.iter().position(|q| q.id == id && q.server == src) else { + return false; + }; + let query = self.stun.swap_remove(index); + + if !reply.has_method(BINDING) || !reply.has_class(MessageClass::Success) { + tracing::debug!(server = %query.server, "STUN server refused the binding request"); + return true; + } + let Ok(mapped) = reply.attribute::() else { + tracing::debug!(server = %query.server, "STUN reply carries no mapped address"); + return true; + }; + + let mapped = canonical(mapped.addr(id)); + match Candidate::server_reflexive(mapped, query.base, "udp") { + // None: redundant with a candidate the agent already has, as when there is no NAT. + Ok(candidate) => { + let added = self.change().add_local_candidate(candidate).cloned(); + if let Some(candidate) = added { + self.events.push_back(Event::Candidate(candidate)); + } + } + Err(err) => tracing::debug!(%err, %mapped, "unusable server-reflexive address"), + } + true + } + + /// Send each STUN request that is due, dropping those out of attempts. + fn retransmit(&mut self, now: Instant) { + let mut due = Vec::new(); + self.stun.retain_mut(|query| { + if query.deadline > now { + return true; + } + if query.sends == STUN_SENDS { + tracing::debug!(server = %query.server, "STUN server did not answer"); + return false; + } + query.sends += 1; + query.deadline = now + + match query.sends { + STUN_SENDS => STUN_RTO * STUN_LAST_WAIT, + _ => query.rto, + }; + query.rto *= 2; + due.push((query.request.clone(), query.server)); + true + }); + for (request, server) in due { + self.send(&request, server); + } + } +} + +/// Mutable access to an [`Agent`]'s connection, which drains it when dropped. +pub struct RtcMut<'a>(&'a mut Agent); + +impl Deref for RtcMut<'_> { + type Target = Rtc; + + fn deref(&self) -> &Rtc { + &self.0.rtc + } +} + +impl DerefMut for RtcMut<'_> { + fn deref_mut(&mut self) -> &mut Rtc { + &mut self.0.rtc + } +} + +impl Drop for RtcMut<'_> { + fn drop(&mut self) { + self.0.drain(); + } +} + +/// Unmap an IPv4-mapped address to the `V4` it really is; others pass through. +pub fn canonical(addr: SocketAddr) -> SocketAddr { + SocketAddr::new(addr.ip().to_canonical(), addr.port()) +} + +/// Adapt a destination to what the socket accepts: an AF_INET6 socket only sends +/// to IPv6, so an IPv4 destination goes back to its mapped form. +fn to_family(addr: SocketAddr, socket_v6: bool) -> SocketAddr { + match addr.ip() { + IpAddr::V4(ip) if socket_v6 => SocketAddr::new(IpAddr::V6(ip.to_ipv6_mapped()), addr.port()), + _ => addr, + } +} + +/// The advertised candidate to tag an inbound datagram with: the first of the +/// source's family, falling back to the first, so a dual-stack peer's packets keep +/// a same-family host candidate and str0m's pairing stays consistent. +fn pick_local(locals: &[SocketAddr], src: SocketAddr) -> SocketAddr { + locals + .iter() + .find(|l| l.is_ipv4() == src.is_ipv4()) + .copied() + .unwrap_or(locals[0]) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pick_local_matches_address_family() { + let v4: SocketAddr = "1.2.3.4:5000".parse().unwrap(); + let v6: SocketAddr = "[2001:db8::1]:5000".parse().unwrap(); + let locals = vec![v4, v6]; + assert_eq!(pick_local(&locals, "9.9.9.9:1".parse().unwrap()), v4); + assert_eq!(pick_local(&locals, "[2001:db8::2]:1".parse().unwrap()), v6); + // No same-family candidate falls back to the first. + assert_eq!(pick_local(&[v4], "[2001:db8::2]:1".parse().unwrap()), v4); + } + + #[tokio::test] + async fn wire_counts_only_accepted_payload() { + let socket = Arc::new(UdpSocket::bind("127.0.0.1:0").await.unwrap()); + let peer = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + let local = socket.local_addr().unwrap(); + let inbound = crate::reader(socket.clone()); + let wire = inbound.wire.clone(); + let socket = Socket { + udp: socket, + locals: vec![local], + inbound, + }; + let mut agent = Agent::new(Rtc::builder().build(Instant::now()), socket).unwrap(); + + // Until tokio has seen the socket writable, a send waits in the outbox. + agent.socket.writable().await.unwrap(); + agent.send(&[42; 73], peer.local_addr().unwrap()); + // UDP refuses an oversized datagram, which is never counted as sent. + agent.send(&vec![0; 65_536], peer.local_addr().unwrap()); + assert_eq!(wire.stats(), crate::Stats { tx: 73, rx: 0 }); + + peer.send_to(&[7; 11], local).await.unwrap(); + let (data, _) = agent.inbound.packets.recv().await.unwrap(); + assert_eq!(data.len(), 11); + drop(agent); + assert_eq!(wire.stats(), crate::Stats { tx: 73, rx: 11 }); + } + + /// An agent on a fresh socket, which tokio reports unwritable until the runtime + /// polls it, so every send waits in the outbox until the test yields. + fn unpolled() -> Agent { + let socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap(); + socket.set_nonblocking(true).unwrap(); + let socket = Arc::new(UdpSocket::from_std(socket).unwrap()); + let local = socket.local_addr().unwrap(); + let socket = Socket { + inbound: crate::reader(socket.clone()), + udp: socket, + locals: vec![local], + }; + Agent::new(Rtc::builder().build(Instant::now()), socket).unwrap() + } + + #[tokio::test] + async fn outbox_is_bounded() { + let mut agent = unpolled(); + let dst = agent.locals[0]; + for _ in 0..crate::INBOX + 10 { + agent.send(&[1; 100], dst); + } + assert_eq!(agent.outbox.len(), crate::INBOX); + } + + #[tokio::test] + async fn awaiting_answer_holds_the_deadline() { + let mut agent = unpolled().awaiting_answer(); + assert_eq!(agent.started, None); + // The agent's own changes, such as timeouts and STUN, leave it held. + drop(agent.change()); + assert_eq!(agent.started, None); + drop(agent.rtc_mut()); + assert!(agent.started.is_some()); + } + + #[tokio::test] + async fn gathering_holds_the_deadline() { + let mut agent = unpolled().with_stun(&["127.0.0.1:3478".parse().unwrap()]); + // An unreachable STUN server retransmits for longer than the deadline, so the + // deadline waits for gathering to end. + assert_eq!(agent.establish(), None); + agent.stun.clear(); + assert!(matches!(agent.next().await, Ok(Event::Gathered))); + let gathered = agent.gathered.unwrap(); + assert_eq!(agent.establish(), Some(gathered + ESTABLISH_TIMEOUT)); + } + + #[test] + fn mapped_addresses_round_trip() { + let v4: SocketAddr = "1.2.3.4:5000".parse().unwrap(); + let mapped = to_family(v4, true); + assert_eq!(mapped, "[::ffff:1.2.3.4]:5000".parse().unwrap()); + assert_eq!(canonical(mapped), v4); + assert_eq!(to_family(v4, false), v4); + } +} diff --git a/rs/moq-ice/src/lib.rs b/rs/moq-ice/src/lib.rs new file mode 100644 index 00000000000..af782457c8e --- /dev/null +++ b/rs/moq-ice/src/lib.rs @@ -0,0 +1,27 @@ +//! The ICE side of a str0m peer connection: one UDP socket, its candidates, and +//! the loop that drives an [`str0m::Rtc`] over it. +//! +//! str0m is sans-IO, so [`Agent::next`] alternates between feeding it datagrams +//! and timeouts and draining what it produces (transmits, events, the next +//! timeout), and runs a STUN Binding client on the same socket so the agent can +//! offer server-reflexive candidates. Shared by `moq-tokio`'s data channel +//! transport and `moq-rtc`'s WHIP/WHEP gateway, which only differ in what they do +//! with the events. +//! +//! str0m requires every mutation of an `Rtc` to be followed by draining its output +//! to a timeout before the next one. The agent owns that cycle: [`Agent::rtc`] is +//! read only, and the [`RtcMut`] guard from [`Agent::rtc_mut`] drains when dropped, +//! queueing the events for [`Agent::next`]. +//! +//! A dual-stack socket (`[::]`) reports an IPv4 peer as an IPv4-mapped IPv6 +//! address and refuses a send to a real `V4` destination. Neither form leaks past +//! this crate: inbound sources are unmapped by [`canonical`], since ICE pairs +//! candidates by address family, and outbound destinations are mapped back. + +#![warn(missing_docs)] + +mod agent; +mod wire; + +pub use agent::{Agent, Error, Event, RtcMut, canonical}; +pub use wire::{INBOX, Inbound, Packet, Socket, Stats, Wire, reader}; diff --git a/rs/moq-ice/src/wire.rs b/rs/moq-ice/src/wire.rs new file mode 100644 index 00000000000..53c022371fd --- /dev/null +++ b/rs/moq-ice/src/wire.rs @@ -0,0 +1,106 @@ +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use tokio::net::UdpSocket; +use tokio::sync::mpsc; + +use crate::canonical; + +/// One inbound UDP datagram and the address it came from. +pub type Packet = (Vec, SocketAddr); + +/// Bound on a session's inbound datagram queue, sized like a socket buffer: past +/// it, datagrams are dropped rather than buffered, since ICE and SCTP tolerate loss +/// and a stalled session must not grow memory without limit. +pub const INBOX: usize = 256; + +/// The UDP socket an [`crate::Agent`] runs on: where it sends, what it receives, +/// and the host candidates it is reachable at. +pub struct Socket { + /// Sends datagrams, possibly shared by several agents (a demuxed server socket). + pub udp: Arc, + /// The host candidates advertised to the peer, not the socket's bind address. + /// Must not be empty. + pub locals: Vec, + /// The datagrams routed to this agent. + pub inbound: Inbound, +} + +/// The datagrams routed to one [`crate::Agent`], and the counters they are tallied in. +pub struct Inbound { + /// Ends once every sender is dropped, which ends the agent. + pub packets: mpsc::Receiver, + /// Counts `rx` where datagrams are routed, so a full queue's drops count too; + /// the agent counts `tx`. + pub wire: Wire, +} + +/// Cumulative UDP payload bytes exchanged with one peer, excluding IP and UDP +/// headers. Clones share counters, which outlive the agent. +#[derive(Clone, Debug, Default)] +pub struct Wire(Arc); + +#[derive(Debug, Default)] +struct Counters { + tx: AtomicU64, + rx: AtomicU64, +} + +/// A snapshot of a [`Wire`]'s counters. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Stats { + /// Bytes the socket accepted, including ICE, DTLS, and retransmissions. + pub tx: u64, + /// Bytes routed to the peer, including those its full queue dropped. + pub rx: u64, +} + +impl Wire { + /// Read both counters without consuming them. + pub fn stats(&self) -> Stats { + Stats { + tx: self.0.tx.load(Ordering::Relaxed), + rx: self.0.rx.load(Ordering::Relaxed), + } + } + + /// Count a datagram routed to this peer. + pub fn received(&self, bytes: usize) { + self.0.rx.fetch_add(bytes as u64, Ordering::Relaxed); + } + + pub(crate) fn sent(&self, bytes: usize) { + self.0.tx.fetch_add(bytes as u64, Ordering::Relaxed); + } +} + +/// Pump every datagram from a socket the caller owns 1:1 into a queue for +/// [`Socket::inbound`], until the queue's receiver is dropped. +pub fn reader(socket: Arc) -> Inbound { + let (tx, rx) = mpsc::channel(INBOX); + let wire = Wire::default(); + let counted = wire.clone(); + tokio::spawn(async move { + let mut buf = vec![0u8; 65_535]; + loop { + tokio::select! { + res = socket.recv_from(&mut buf) => match res { + Ok((len, src)) => { + counted.received(len); + // Bounded like a socket buffer: drop on full. + if let Err(mpsc::error::TrySendError::Closed(_)) = tx.try_send((buf[..len].to_vec(), canonical(src))) { + break; + } + } + Err(err) => { + tracing::warn!(%err, "webrtc socket recv failed"); + break; + } + }, + _ = tx.closed() => break, + } + } + }); + Inbound { packets: rx, wire } +} diff --git a/rs/moq-relay/Cargo.toml b/rs/moq-relay/Cargo.toml index d1fd9a5c26c..80c050a081d 100644 --- a/rs/moq-relay/Cargo.toml +++ b/rs/moq-relay/Cargo.toml @@ -76,7 +76,7 @@ rustls-acme = { version = "0.15.4", default-features = false, features = ["aws-l serde = { workspace = true } serde_json = { workspace = true } serde_with = { workspace = true, features = ["json", "base64"] } -stun-types = "2.0.3" +stun-types = { workspace = true } sysinfo = { version = "0.39", default-features = false, features = ["system"] } thiserror = { workspace = true } tokio = { workspace = true, features = ["full"] } diff --git a/rs/moq-relay/src/auth.rs b/rs/moq-relay/src/auth.rs index 1e9878a7994..71bfb668702 100644 --- a/rs/moq-relay/src/auth.rs +++ b/rs/moq-relay/src/auth.rs @@ -998,17 +998,7 @@ impl Admission { /// The `moq_auth::Request` for an accepted transport request: every fact the /// transport knows, nothing parsed on the server's behalf. pub fn request_for(auth: &Auth, request: &moq_tokio::server::Request) -> Request { - let transport = match request.transport() { - moq_tokio::Transport::Quic => moq_auth::Transport::Quic, - moq_tokio::Transport::WebTransport => moq_auth::Transport::WebTransport, - moq_tokio::Transport::Iroh => moq_auth::Transport::Iroh, - moq_tokio::Transport::WebSocket => moq_auth::Transport::WebSocket, - moq_tokio::Transport::Tcp => moq_auth::Transport::Tcp, - moq_tokio::Transport::Unix => moq_auth::Transport::Unix, - // A transport this build does not know is still a session on the wire; the - // server sees the same facts either way. - other => unreachable!("unknown transport {other}"), - }; + let transport = transport(request.transport()); // A URL-less transport reports its root as empty; the contract says what was dialed. let path = match request.path() { "" => "/".to_string(), @@ -1032,6 +1022,22 @@ pub fn request_for(auth: &Auth, request: &moq_tokio::server::Request) -> Request out } +/// The auth transport for a session's transport. +fn transport(transport: moq_tokio::Transport) -> moq_auth::Transport { + match transport { + moq_tokio::Transport::Quic => moq_auth::Transport::Quic, + moq_tokio::Transport::WebTransport => moq_auth::Transport::WebTransport, + moq_tokio::Transport::Iroh => moq_auth::Transport::Iroh, + moq_tokio::Transport::WebSocket => moq_auth::Transport::WebSocket, + moq_tokio::Transport::Tcp => moq_auth::Transport::Tcp, + moq_tokio::Transport::Unix => moq_auth::Transport::Unix, + moq_tokio::Transport::WebRtc => moq_auth::Transport::WebRtc, + // A transport this build does not know is still a session on the wire; the + // server sees the same facts either way. + other => unreachable!("unknown transport {other}"), + } +} + /// The auth role for a session role, `None` for one `moq-auth` cannot name yet. pub(crate) fn role(role: moq_net::Role) -> Option { match role { @@ -1120,6 +1126,22 @@ mod tests { assert!(lease.ended().await.to_string().contains("cannot enforce bitrate caps")); } + #[test] + fn every_transport_has_an_auth_name() { + use moq_tokio::Transport as T; + for transport in [ + T::Quic, + T::WebTransport, + T::Iroh, + T::WebSocket, + T::Tcp, + T::Unix, + T::WebRtc, + ] { + assert_eq!(super::transport(transport).as_str(), transport.as_str()); + } + } + #[test] fn exactly_one_source() { assert!(config(None, &[]).validate(false).is_err()); diff --git a/rs/moq-rtc/Cargo.toml b/rs/moq-rtc/Cargo.toml index 50162b504a2..cd79d6ef3cf 100644 --- a/rs/moq-rtc/Cargo.toml +++ b/rs/moq-rtc/Cargo.toml @@ -26,12 +26,14 @@ server = ["dep:axum"] axum = { workspace = true, optional = true } bytes = { workspace = true } hang = { workspace = true } +# Drives each str0m connection over its UDP socket. +moq-ice = { workspace = true } moq-mux = { workspace = true } moq-net = { workspace = true } # `RedactedUrl` keeps a dialed URL's query token and userinfo out of the logs. moq-tokio = { workspace = true } reqwest = { workspace = true, features = ["rustls"] } -str0m = "0.24" +str0m = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true, features = ["full"] } tracing = { workspace = true } diff --git a/rs/moq-rtc/src/client/whep.rs b/rs/moq-rtc/src/client/whep.rs index 27e11bd4fd1..1b148235f92 100644 --- a/rs/moq-rtc/src/client/whep.rs +++ b/rs/moq-rtc/src/client/whep.rs @@ -27,9 +27,9 @@ pub(crate) async fn dial( .with_bandwidth(client.config().bandwidth.clone()); let sink = Box::new(IngestSink::new(broadcast, config)?); - let (socket, candidates) = session::bind_udp(&client.config().ice_candidates).await?; + let socket = session::bind_udp(&client.config().ice_candidates).await?; let mut rtc = Rtc::new(Instant::now()); - for addr in &candidates { + for addr in &socket.locals { let cand = Candidate::host(*addr, "udp").map_err(Error::rtc)?; rtc.add_local_candidate(cand); } @@ -63,11 +63,7 @@ pub(crate) async fn dial( // The URL routinely carries the auth token in its query, or userinfo. tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), "whep client connected"); - // 1:1 socket (no demux on the client): pump its datagrams into the session. - // The session tags each datagram with the advertised candidate matching its - // family (str0m matches the destination against a host candidate, not the bind). - let inbound = session::spawn_socket_reader(socket.clone()); - let mut session = session::Session::ingest(rtc, socket, candidates, inbound, sink); + let mut session = session::Session::ingest(rtc, socket, sink)?; for mid in mids { session.open_media(mid)?; } diff --git a/rs/moq-rtc/src/client/whip.rs b/rs/moq-rtc/src/client/whip.rs index 055992ed14c..3727953664b 100644 --- a/rs/moq-rtc/src/client/whip.rs +++ b/rs/moq-rtc/src/client/whip.rs @@ -30,11 +30,11 @@ pub(crate) async fn dial( return Err(Error::NoRenditions); } - let (socket, candidates) = session::bind_udp(&client.config().ice_candidates).await?; + let socket = session::bind_udp(&client.config().ice_candidates).await?; // Restrict to codecs the catalog can actually source so the remote // answer doesn't pick a codec we have no rendition for. let mut rtc = session::rtc_with_codecs(&codecs); - for addr in &candidates { + for addr in &socket.locals { let cand = Candidate::host(*addr, "udp").map_err(Error::rtc)?; rtc.add_local_candidate(cand); } @@ -75,11 +75,7 @@ pub(crate) async fn dial( // The URL routinely carries the auth token in its query, or userinfo. tracing::info!(url = %moq_tokio::RedactedUrl::new(&url), "whip client connected"); - // 1:1 socket (no demux on the client): pump its datagrams into the session. - // The session tags each datagram with the advertised candidate matching its - // family (str0m matches the destination against a host candidate, not the bind). - let inbound = session::spawn_socket_reader(socket.clone()); - let mut session = session::Session::egress(rtc, socket, candidates, inbound, source); + let mut session = session::Session::egress(rtc, socket, source)?; for mid in mids { session.open_media(mid)?; } diff --git a/rs/moq-rtc/src/error.rs b/rs/moq-rtc/src/error.rs index e36cf3a7131..a8c4f54fbb2 100644 --- a/rs/moq-rtc/src/error.rs +++ b/rs/moq-rtc/src/error.rs @@ -62,10 +62,6 @@ pub enum Error { #[error("rtc error: {0}")] Rtc(String), - /// Error feeding a received UDP datagram into the WebRTC engine. - #[error("rtc input error: {0}")] - RtcInput(String), - /// An internal video bridge could not be initialized. #[error("video bridge initialization failed")] BridgeFailed, @@ -83,10 +79,6 @@ impl Error { pub(crate) fn rtc(err: impl std::fmt::Display) -> Self { Self::Rtc(err.to_string()) } - - pub(crate) fn rtc_input(err: impl std::fmt::Display) -> Self { - Self::RtcInput(err.to_string()) - } } /// Convenience alias for results from the WebRTC <-> MoQ gateway. diff --git a/rs/moq-rtc/src/lib.rs b/rs/moq-rtc/src/lib.rs index 7075db3947b..e7f05fc51db 100644 --- a/rs/moq-rtc/src/lib.rs +++ b/rs/moq-rtc/src/lib.rs @@ -55,7 +55,6 @@ mod codec; mod egress; mod error; mod ingest; -mod net; mod sdp; mod session; diff --git a/rs/moq-rtc/src/net.rs b/rs/moq-rtc/src/net.rs deleted file mode 100644 index 47f2924e412..00000000000 --- a/rs/moq-rtc/src/net.rs +++ /dev/null @@ -1,73 +0,0 @@ -//! Address-family adaptation at the UDP socket boundary. -//! -//! A dual-stack listener (`[::]`, which Linux serves both families from unless -//! `bindv6only` is set) reports an IPv4 peer as an IPv4-mapped IPv6 address -//! (`::ffff:a.b.c.d`), and refuses a send to a real `V4` destination. Neither -//! form should leak past this boundary: -//! -//! - **Inbound**: [`canonical`] unmaps, so an IPv4 peer looks like the `V4` it -//! is. ICE pairing keys off the address family (see -//! [`pick_local`](crate::session::pick_local)), so without this an IPv4 peer -//! on a dual-stack socket would be paired against the IPv6 host candidate. -//! - **Outbound**: [`to_family`] re-maps, because the destination str0m hands -//! back is the canonical address we fed it. - -use std::net::{IpAddr, SocketAddr}; - -/// Unmap an IPv4-mapped peer address to the `V4` it really is; other addresses -/// pass through unchanged. -pub(crate) fn canonical(addr: SocketAddr) -> SocketAddr { - SocketAddr::new(addr.ip().to_canonical(), addr.port()) -} - -/// Adapt a destination to what `socket_is_v6`'s socket will accept: an AF_INET6 -/// socket can only send to an IPv6 address, so an IPv4 destination has to go -/// back to its mapped form. -pub(crate) fn to_family(dst: SocketAddr, socket_is_v6: bool) -> SocketAddr { - match (socket_is_v6, dst.ip()) { - (true, IpAddr::V4(ip)) => SocketAddr::new(IpAddr::V6(ip.to_ipv6_mapped()), dst.port()), - _ => dst, - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn canonical_unmaps_v4_mapped_peers() { - // What a dual-stack socket reports for an IPv4 peer. - let mapped: SocketAddr = "[::ffff:1.2.3.4]:5000".parse().unwrap(); - assert_eq!(canonical(mapped), "1.2.3.4:5000".parse().unwrap()); - } - - #[test] - fn canonical_passes_through_native_addresses() { - let v4: SocketAddr = "1.2.3.4:5000".parse().unwrap(); - let v6: SocketAddr = "[2001:db8::1]:5000".parse().unwrap(); - assert_eq!(canonical(v4), v4); - assert_eq!(canonical(v6), v6); - } - - #[test] - fn to_family_remaps_v4_for_a_dual_stack_socket() { - let v4: SocketAddr = "1.2.3.4:5000".parse().unwrap(); - assert_eq!(to_family(v4, true), "[::ffff:1.2.3.4]:5000".parse().unwrap()); - } - - #[test] - fn to_family_leaves_matching_families_alone() { - let v4: SocketAddr = "1.2.3.4:5000".parse().unwrap(); - let v6: SocketAddr = "[2001:db8::1]:5000".parse().unwrap(); - assert_eq!(to_family(v4, false), v4); - assert_eq!(to_family(v6, true), v6); - } - - /// The round trip a dual-stack server does per IPv4 peer: canonicalize the - /// source on the way in, then re-map str0m's destination on the way out. - #[test] - fn canonical_and_to_family_round_trip() { - let mapped: SocketAddr = "[::ffff:1.2.3.4]:5000".parse().unwrap(); - assert_eq!(to_family(canonical(mapped), true), mapped); - } -} diff --git a/rs/moq-rtc/src/server/mux.rs b/rs/moq-rtc/src/server/mux.rs index b7d68527702..df5a852ec86 100644 --- a/rs/moq-rtc/src/server/mux.rs +++ b/rs/moq-rtc/src/server/mux.rs @@ -23,12 +23,13 @@ use std::collections::HashMap; use std::net::SocketAddr; use std::sync::{Arc, Mutex}; +use moq_ice::{Inbound, Packet}; use str0m::ice::{IceCreds, StunMessage}; use tokio::net::UdpSocket; use tokio::sync::mpsc; use crate::Result; -use crate::session::{Inbound, Packet, SESSION_INBOX, advertised_candidates}; +use crate::session::advertised_candidates; /// Per-session routing table, shared between the demux task and the accept path. #[derive(Default)] @@ -44,7 +45,7 @@ struct Registry { struct Entry { ufrag: String, sender: mpsc::Sender, - wire: crate::wire::Source, + wire: moq_ice::Wire, } /// The shared media socket plus its demux routing table. @@ -124,8 +125,8 @@ impl Mux { /// [`Registration`] guard the session task must hold for its lifetime. pub(crate) fn register(&self) -> (IceCreds, Inbound, Registration) { let creds = IceCreds::new(); - let (tx, rx) = mpsc::channel(SESSION_INBOX); - let wire = crate::wire::Source::new(); + let (tx, rx) = mpsc::channel(moq_ice::INBOX); + let wire = moq_ice::Wire::default(); let entry = Arc::new(Entry { ufrag: creds.ufrag.clone(), sender: tx, @@ -170,8 +171,8 @@ async fn demux(socket: Arc, registry: Arc>) { } }; // A dual-stack bind reports IPv4 peers as `::ffff:a.b.c.d`; unmap before - // this address becomes a registry key or reaches ICE (see crate::net). - let src = crate::net::canonical(src); + // this address becomes a registry key or reaches ICE (see `moq_ice::canonical`). + let src = moq_ice::canonical(src); let data = &buf[..len]; // Cached peers need one routing lock; only a new address parses STUN. @@ -254,7 +255,7 @@ mod tests { let address_b = "127.0.0.1:10002".parse().unwrap(); let mut registry = mux.registry.lock().unwrap(); assert!(registry.route(100, address_a, Some("unknown")).is_none()); - for _ in 0..SESSION_INBOX { + for _ in 0..moq_ice::INBOX { let entry = registry.route(11, address_a, Some(&a.ufrag)).unwrap(); entry.sender.try_send((vec![0; 11], address_a)).unwrap(); } @@ -267,12 +268,12 @@ mod tests { drop(registry); assert_eq!( source_a.stats(), - crate::wire::Stats { + moq_ice::Stats { tx: 0, - rx: 11 * SESSION_INBOX as u64 + 23 + rx: 11 * moq_ice::INBOX as u64 + 23 } ); - assert_eq!(source_b.stats(), crate::wire::Stats { tx: 0, rx: 37 }); + assert_eq!(source_b.stats(), moq_ice::Stats { tx: 0, rx: 37 }); // Unregister even while an inbox is still held: cached addresses cannot // keep attributing traffic after the peer ends. @@ -299,7 +300,7 @@ mod tests { let mut addresses = Vec::new(); for peer in 0..peers { let (sender, receiver) = mpsc::channel(1); - let wire = crate::wire::Source::new(); + let wire = moq_ice::Wire::default(); let address = SocketAddr::from(([127, 0, 0, 1], peer as u16 + 1)); let entry = Arc::new(Entry { ufrag: peer.to_string(), diff --git a/rs/moq-rtc/src/server/whep.rs b/rs/moq-rtc/src/server/whep.rs index c309b1d9c72..c8915c19a3c 100644 --- a/rs/moq-rtc/src/server/whep.rs +++ b/rs/moq-rtc/src/server/whep.rs @@ -159,7 +159,12 @@ pub async fn accept( let answer = rtc.sdp_api().accept_offer(offer).map_err(Error::rtc)?; let resource_id = sdp::new_resource_id(); - let session = session::Session::egress(rtc, mux.socket(), mux.candidates().to_vec(), inbound, source); + let socket = moq_ice::Socket { + udp: mux.socket(), + locals: mux.candidates().to_vec(), + inbound, + }; + let session = session::Session::egress(rtc, socket, source)?; // Register before returning so a DELETE that races startup still finds the // session; Response::run unregisters itself when it ends. diff --git a/rs/moq-rtc/src/server/whip.rs b/rs/moq-rtc/src/server/whip.rs index 1663ffc748e..425b0952f20 100644 --- a/rs/moq-rtc/src/server/whip.rs +++ b/rs/moq-rtc/src/server/whip.rs @@ -153,7 +153,12 @@ pub async fn accept( handle.announce(moq_net::origin::Route::default().with_epoch(epoch.clone()))?; let resource_id = sdp::new_resource_id(); - let session = session::Session::ingest(rtc, mux.socket(), mux.candidates().to_vec(), inbound, sink); + let socket = moq_ice::Socket { + udp: mux.socket(), + locals: mux.candidates().to_vec(), + inbound, + }; + let session = session::Session::ingest(rtc, socket, sink)?; // Register before returning so a DELETE that races the first packet still // finds the session; Response::run unregisters itself when it ends. diff --git a/rs/moq-rtc/src/session.rs b/rs/moq-rtc/src/session.rs index 3569e01ca57..2321a8e2d35 100644 --- a/rs/moq-rtc/src/session.rs +++ b/rs/moq-rtc/src/session.rs @@ -1,9 +1,7 @@ //! str0m session driver shared by every HTTP role / media direction. //! -//! str0m is sans-IO, so we drive the [`str0m::Rtc`] instance from a tokio -//! task that owns a UDP socket. [`Session::run`] alternates between -//! [`Rtc::poll_output`] (drain pending transmits / events) and -//! [`Rtc::handle_input`] (feed UDP packets or timeouts). +//! [`moq_ice::Agent`] drives the sans-IO [`str0m::Rtc`] over UDP; +//! [`Session::run`] feeds it egress writes and handles the media events it surfaces. //! //! The session itself doesn't care whether the [`Rtc`] was populated by //! accepting an SDP offer (server side) or by minting one and posting it @@ -13,53 +11,21 @@ use std::collections::{HashMap, HashSet}; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::sync::Arc; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; -use str0m::{Candidate, Event, IceConnectionState, Input, Output, Rtc, net::Receive}; +use str0m::{Candidate, Event, Rtc}; use tokio::net::UdpSocket; use tokio::sync::mpsc; use crate::egress::{EgressClock, EgressSource, Format, WriteRequest}; use crate::{Error, Result, codec}; -/// One inbound UDP datagram plus its source address, the unit fed to a session. -/// The [`server`](crate::server) paths get these from the shared-socket demux -/// (`crate::server::mux`); the client paths get them from a 1:1 reader -/// ([`spawn_socket_reader`]). -pub(crate) type Packet = (Vec, SocketAddr); - -pub(crate) struct Inbound { - pub packets: mpsc::Receiver, - pub wire: crate::wire::Source, -} - -/// Bound on a session's inbound datagram queue, sized like a socket buffer: -/// past this, datagrams are dropped rather than buffered (WebRTC tolerates loss -/// and a stalled session must not grow memory without limit). -pub(crate) const SESSION_INBOX: usize = 256; - /// str0m's outbound video buffer depth (packets), which also backs NACK resends. /// Raised above the str0m default (1000) so a late-joining peer can recover a /// large keyframe and the rest of the current group via NACK; see /// [`rtc_config_with_codecs`]. const EGRESS_SEND_BUFFER_VIDEO: usize = 3000; -/// Backstop deadline for a session to reach a connected ICE state, covering the one -/// case str0m's ICE agent deliberately never times out: a peer that answers the SDP -/// but provides NO remote candidates and sends nothing (an abandoned WHIP/WHEP -/// offer, or a probe that only exercises signalling). str0m DOES end a connection -/// whose candidate pairs were tried and exhausted -- the agent goes to -/// `IceConnectionState::Disconnected` (handled in `handle_event`) after -/// ~`StunTiming::timeout()` (~21s at the defaults). But when `remote_candidates` -/// stays empty the agent treats the session as "still possible" forever (trickle -/// ICE: more candidates could arrive), so it sits in `Checking` indefinitely, -/// pinning this task, its broadcast announcement, and its mux registration. Nothing -/// upstream ends it, so we do. Set ABOVE str0m's ~21s pair-exhaustion so a -/// connection that actually started checks is ended by str0m's native path (and a -/// slow-but-real TURN/lossy peer isn't clipped); this only fires for the -/// never-any-candidate case. -const ICE_ESTABLISH_TIMEOUT: Duration = Duration::from_secs(30); - /// Receives `MediaData` events from str0m and dispatches to the right codec /// [`Bridge`](codec::Bridge). Used as the per-session sink in [`Session::run`] /// for any flow where RTP arrives from the peer (`server publish` / WHIP @@ -114,23 +80,7 @@ impl MediaFlow { /// `inbound` rather than being read off the socket directly, so several /// sessions can share one socket behind the `crate::server::mux`. pub struct Session { - rtc: Rtc, - /// Send side. Shared across sessions on the server (the mux socket); owned - /// 1:1 on the client. Receiving happens via `inbound`, not this socket. - socket: Arc, - /// The local ICE candidates we advertised. Each inbound datagram is tagged - /// (for str0m) with the candidate whose address family matches the packet's - /// source, so a dual-stack peer reaching us over IPv6 isn't told the packet - /// arrived on an IPv4 host candidate. MUST be the advertised candidates, not - /// the socket's bind address: str0m drops a STUN binding request whose - /// destination doesn't match a host candidate ("unknown interface"), and the - /// shared mux socket binds a wildcard (`0.0.0.0`) while advertising concrete - /// IPs. Never empty (falls back to the bound address). - locals: Vec, - /// Inbound datagrams routed to this session (demux on the server, a 1:1 - /// reader on the client). `None` from `recv` means every sender dropped, so - /// the session is done. - inbound: Inbound, + agent: moq_ice::Agent, role: MediaRole, opened_media: HashSet, /// Egress write requests. `Some` only for [`MediaRole::Egress`] @@ -146,53 +96,35 @@ pub struct Session { } impl Session { - /// Convenience for the ingest case (WHIP server, WHEP client). `locals` are the - /// advertised ICE candidates (see the field docs), not the socket bind. - pub fn ingest( - rtc: Rtc, - socket: Arc, - locals: Vec, - inbound: Inbound, - sink: Box, - ) -> Self { - Self { - rtc, - socket, - locals, - inbound, + /// Convenience for the ingest case (WHIP server, WHEP client). The socket's + /// `inbound` comes from the server mux, or [`moq_ice::reader`] for a 1:1 client socket. + pub fn ingest(rtc: Rtc, socket: moq_ice::Socket, sink: Box) -> Result { + Ok(Self { + agent: moq_ice::Agent::new(rtc, socket)?, role: MediaRole::Ingest(sink), opened_media: HashSet::new(), writes_rx: None, ingest_clock: IngestClock::default(), egress_clock: EgressClock::default(), - } + }) } - /// Convenience for the egress case (WHEP server, WHIP client). `locals` are the - /// advertised ICE candidates (see the field docs), not the socket bind. - pub fn egress( - rtc: Rtc, - socket: Arc, - locals: Vec, - inbound: Inbound, - mut source: EgressSource, - ) -> Self { + /// Convenience for the egress case (WHEP server, WHIP client). The socket's + /// `inbound` comes from the server mux, or [`moq_ice::reader`] for a 1:1 client socket. + pub fn egress(rtc: Rtc, socket: moq_ice::Socket, mut source: EgressSource) -> Result { let writes_rx = source.take_writes(); - Self { - rtc, - socket, - locals, - inbound, + Ok(Self { + agent: moq_ice::Agent::new(rtc, socket)?, role: MediaRole::Egress(Box::new(source)), opened_media: HashSet::new(), writes_rx: Some(writes_rx), ingest_clock: IngestClock::default(), egress_clock: EgressClock::default(), - } + }) } pub(crate) fn wire(&self) -> crate::wire::Source { - self.inbound.wire.clone() + crate::wire::Source::new(self.agent.wire().clone()) } /// Open a locally offered media line after its answer has been applied. @@ -215,96 +147,29 @@ impl Session { } async fn run_loop(&mut self) -> Result<()> { - let started = Instant::now(); - let mut connected = false; - // str0m hands back the canonical destination we fed it, so a dual-stack - // socket needs IPv4 re-mapped before each send (see crate::net). - let socket_v6 = self.socket.local_addr().map_err(Error::Io)?.is_ipv6(); loop { - // A dead Rtc (DTLS/SDP failure, explicit disconnect) makes poll_output - // return a never-firing timeout instead of erroring, which would hang - // this task forever holding the broadcast announcement + mux - // registration. Bail so those release. - if !self.rtc.is_alive() { - return Err(Error::SessionClosed); - } - - // Abort a session that never finishes connecting (see - // ICE_ESTABLISH_TIMEOUT); once connected, str0m's own timeouts take over. - if !connected && started.elapsed() >= ICE_ESTABLISH_TIMEOUT { - return Err(Error::IceTimeout); - } - - let timeout = match self.rtc.poll_output().map_err(Error::rtc)? { - Output::Timeout(t) => t, - Output::Transmit(t) => { - let dst = crate::net::to_family(t.destination, socket_v6); - if let Err(err) = transmit(&self.socket, &self.inbound.wire, &t.contents, dst).await { - tracing::warn!(%err, %dst, "send failed"); - } - continue; - } - Output::Event(event) => { - if let Event::IceConnectionStateChange(state) = &event { - connected |= state.is_connected(); - } - self.handle_event(event)?; - continue; - } - }; - - let now = Instant::now(); - let mut duration = timeout.saturating_duration_since(now); - // While still connecting, never sleep past the establishment deadline, so - // the check above fires on time even if str0m scheduled a far-off timeout. - if !connected { - duration = duration.min(ICE_ESTABLISH_TIMEOUT.saturating_sub(started.elapsed())); - } - if duration.is_zero() { - self.rtc.handle_input(Input::Timeout(now)).map_err(Error::rtc)?; - continue; - } - - // Wait for the earliest of: an inbound UDP packet, an egress - // write request (if egress), or the str0m-requested timeout. tokio::select! { + // Egress writes get drained promptly; `Agent::next` is cancel safe. biased; - // Egress writes get drained promptly. Without `biased` an - // idle socket select could starve them. Some(req) = async { match self.writes_rx.as_mut() { Some(rx) => rx.recv().await, None => std::future::pending::>().await, } } => { - let now = Instant::now(); - let wallclock = self.egress_clock.wallclock(req.time, now); - crate::egress::dispatch(&mut self.rtc, req, wallclock); - } - - packet = self.inbound.packets.recv() => { - match packet { - Some((data, src)) => { - let now = Instant::now(); - // Tag the packet with the advertised candidate matching its - // address family, not the socket bind (see the `locals` docs). - let local = pick_local(&self.locals, src); - let recv = Receive::new(str0m::net::Protocol::Udp, src, local, &data) - .map_err(Error::rtc_input)?; - self.rtc.handle_input(Input::Receive(now, recv)).map_err(Error::rtc)?; - } - // Every sender dropped: the demux unregistered us (or the - // 1:1 reader stopped). Nothing more will arrive, so end. - None => return Err(Error::SessionClosed), - } + let wallclock = self.egress_clock.wallclock(req.time, Instant::now()); + crate::egress::dispatch(&mut self.agent.rtc_mut(), req, wallclock); } - _ = tokio::time::sleep(duration) => { - self.rtc - .handle_input(Input::Timeout(Instant::now())) - .map_err(Error::rtc)?; - } + event = self.agent.next() => match event { + Ok(moq_ice::Event::Rtc(event)) => self.handle_event(event)?, + // No STUN servers are configured, so there is nothing to trickle. + Ok(_) => {} + Err(moq_ice::Error::Closed) => return Err(Error::SessionClosed), + Err(moq_ice::Error::Timeout) => return Err(Error::IceTimeout), + Err(err) => return Err(Error::rtc(err)), + }, } } } @@ -313,9 +178,6 @@ impl Session { match event { Event::IceConnectionStateChange(state) => { tracing::debug!(?state, "ice state"); - if state == IceConnectionState::Disconnected { - return Err(Error::SessionClosed); - } } Event::MediaAdded(added) => self.handle_media(added.mid)?, Event::MediaData(data) => { @@ -356,7 +218,7 @@ impl Session { // str0m's CodecConfig is the negotiated set; `remote_pts` lists this // `mid`'s share of it in the peer's order. - let Some(media) = self.rtc.media(mid) else { + let Some(media) = self.agent.rtc().media(mid) else { tracing::warn!(?mid, "negotiated media is missing; ignoring"); return Ok(()); }; @@ -371,7 +233,15 @@ impl Session { let negotiated: Vec<_> = media .remote_pts() .iter() - .filter_map(|pt| self.rtc.codec_config().params().iter().find(|p| p.pt() == *pt).copied()) + .filter_map(|pt| { + self.agent + .rtc() + .codec_config() + .params() + .iter() + .find(|p| p.pt() == *pt) + .copied() + }) .collect(); let Some(first) = negotiated.first() else { tracing::warn!(?mid, "no codec params for media; ignoring"); @@ -407,12 +277,6 @@ impl Session { } } -async fn transmit(socket: &UdpSocket, wire: &crate::wire::Source, data: &[u8], dst: SocketAddr) -> std::io::Result<()> { - let len = socket.send_to(data, dst).await?; - wire.sent(len); - Ok(()) -} - /// Per-session clock that rebases each ingested track's raw RTP timestamps onto /// one timeline so audio and video stay in sync. /// @@ -559,18 +423,6 @@ pub(crate) fn log_session_end(role: &str, result: &Result<()>) { } } -/// Pick the advertised local candidate to tag an inbound packet with: the first -/// one whose address family matches `src`, falling back to the first candidate -/// (the list is never empty). Keeps a dual-stack peer's packets tagged with a -/// same-family host candidate so str0m's ICE pairing stays consistent. -fn pick_local(locals: &[SocketAddr], src: SocketAddr) -> SocketAddr { - locals - .iter() - .find(|l| l.is_ipv4() == src.is_ipv4()) - .copied() - .unwrap_or(locals[0]) -} - /// Convert a str0m [`MediaTime`](str0m::media::MediaTime) to microseconds. fn media_time_to_micros(time: &str0m::media::MediaTime) -> u64 { // MediaTime stores `numer / denom` seconds; cast through i128 so the @@ -657,19 +509,21 @@ pub fn rtc_with_codecs(codecs: &[str0m::format::Codec]) -> Rtc { rtc_config_with_codecs(codecs).build(std::time::Instant::now()) } -/// Bind an ephemeral UDP socket for a single client session and return it -/// (shared with its [reader task](spawn_socket_reader)) plus the ICE candidates -/// to advertise. +/// Bind an ephemeral UDP socket for a single client session, read by its own +/// [`moq_ice::reader`], with the ICE candidates to advertise. /// /// The client paths are 1:1 (one socket per dialed session, no demux); the /// server paths share one socket via `crate::server::mux` instead. `advertise` /// IPs are used verbatim (reusing the bound port); empty falls back to the /// bound address, substituting loopback when that address is unspecified. -pub async fn bind_udp(advertise: &[SocketAddr]) -> Result<(Arc, Vec)> { - let socket = UdpSocket::bind(("0.0.0.0", 0)).await?; - let local = socket.local_addr()?; - let candidates = advertised_candidates(advertise, local)?; - Ok((Arc::new(socket), candidates)) +pub async fn bind_udp(advertise: &[SocketAddr]) -> Result { + let udp = Arc::new(UdpSocket::bind(("0.0.0.0", 0)).await?); + let locals = advertised_candidates(advertise, udp.local_addr()?)?; + Ok(moq_ice::Socket { + inbound: moq_ice::reader(udp.clone()), + udp, + locals, + }) } /// Pair configured ICE candidates with the bound UDP port and validate them. @@ -699,41 +553,6 @@ pub(crate) fn advertised_candidates(advertise: &[SocketAddr], local: SocketAddr) Ok(candidates) } -/// Spawn a 1:1 reader pumping every datagram from `socket` into a channel, for -/// the client paths (one socket per session, so no demux is needed). Mirrors the -/// inbound side of `crate::server::mux` for a single session. -pub fn spawn_socket_reader(socket: Arc) -> Inbound { - let (tx, rx) = mpsc::channel(SESSION_INBOX); - let wire = crate::wire::Source::new(); - let source = wire.clone(); - tokio::spawn(async move { - let mut buf = vec![0u8; 65_535]; - loop { - let packet = tokio::select! { - biased; - _ = tx.closed() => break, - packet = socket.recv_from(&mut buf) => packet, - }; - match packet { - // Bounded like a socket buffer: drop on full, stop once the - // session's receiver is gone. - Ok((len, src)) => { - source.received(len); - let src = crate::net::canonical(src); - if let Err(mpsc::error::TrySendError::Closed(_)) = tx.try_send((buf[..len].to_vec(), src)) { - break; - } - } - Err(err) => { - tracing::warn!(%err, "webrtc client socket recv failed"); - break; - } - } - } - }); - Inbound { packets: rx, wire } -} - #[cfg(test)] mod tests { use std::time::{Duration, UNIX_EPOCH}; @@ -744,33 +563,6 @@ mod tests { use super::*; - #[tokio::test] - async fn wire_tx_counts_only_successful_udp_payload() { - let socket = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let peer = UdpSocket::bind("127.0.0.1:0").await.unwrap(); - let wire = crate::wire::Source::new(); - let packet = [42; 73]; - transmit(&socket, &wire, &packet, peer.local_addr().unwrap()) - .await - .unwrap(); - let mut buffer = [0; 100]; - let (len, _) = tokio::time::timeout(Duration::from_secs(5), peer.recv_from(&mut buffer)) - .await - .unwrap() - .unwrap(); - assert_eq!(&buffer[..len], &packet); - assert_eq!(wire.stats(), crate::wire::Stats { tx: 73, rx: 0 }); - // An oversized datagram is refused by UDP, never counted as sent. - assert!( - transmit(&socket, &wire, &vec![0; 65_536], peer.local_addr().unwrap()) - .await - .is_err() - ); - assert_eq!(wire.stats().tx, 73); - drop(socket); - assert_eq!(wire.stats().tx, 73); - } - #[test] fn media_flow_respects_negotiated_direction() { use str0m::media::Direction::{Inactive, RecvOnly, SendOnly, SendRecv}; @@ -824,19 +616,6 @@ mod tests { assert!(advertised_candidates(&advertised, local).is_err()); } - #[test] - fn pick_local_matches_address_family() { - let v4: SocketAddr = "1.2.3.4:5000".parse().unwrap(); - let v6: SocketAddr = "[2001:db8::1]:5000".parse().unwrap(); - let locals = vec![v4, v6]; - let src_v4: SocketAddr = "9.9.9.9:1".parse().unwrap(); - let src_v6: SocketAddr = "[2001:db8::2]:1".parse().unwrap(); - assert_eq!(pick_local(&locals, src_v4), v4); - assert_eq!(pick_local(&locals, src_v6), v6); - // No same-family candidate falls back to the first. - assert_eq!(pick_local(&[v4], src_v6), v4); - } - #[test] fn ingest_clock_rebases_first_frame_to_zero() { let mut clock = IngestClock::default(); diff --git a/rs/moq-rtc/src/wire.rs b/rs/moq-rtc/src/wire.rs index d8246dc6bcc..2580970d852 100644 --- a/rs/moq-rtc/src/wire.rs +++ b/rs/moq-rtc/src/wire.rs @@ -1,47 +1,19 @@ //! Cumulative UDP payload bytes for one WebRTC peer. -use std::sync::{ - Arc, - atomic::{AtomicU64, Ordering}, -}; - /// A peer's cumulative datagram bytes, excluding IP and UDP headers. -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] -pub struct Stats { - /// Bytes successfully handed to the UDP socket, including control and retransmissions. - pub tx: u64, - /// Bytes routed to this peer, including control and packets dropped by its full inbox. - pub rx: u64, -} +pub use moq_ice::Stats; /// A read-only source whose counters survive the WebRTC session ending. #[derive(Clone, Debug)] -pub struct Source(Arc); - -#[derive(Debug, Default)] -struct Counters { - tx: AtomicU64, - rx: AtomicU64, -} +pub struct Source(moq_ice::Wire); impl Source { - pub(crate) fn new() -> Self { - Self(Arc::default()) + pub(crate) fn new(wire: moq_ice::Wire) -> Self { + Self(wire) } /// Read cumulative sent and received UDP payload bytes; neither read consumes them. pub fn stats(&self) -> Stats { - Stats { - tx: self.0.tx.load(Ordering::Relaxed), - rx: self.0.rx.load(Ordering::Relaxed), - } - } - - pub(crate) fn sent(&self, bytes: usize) { - self.0.tx.fetch_add(bytes as u64, Ordering::Relaxed); - } - - pub(crate) fn received(&self, bytes: usize) { - self.0.rx.fetch_add(bytes as u64, Ordering::Relaxed); + self.0.stats() } } diff --git a/rs/moq-tokio/Cargo.toml b/rs/moq-tokio/Cargo.toml index 2d879bf4614..9e4ddcf3e15 100644 --- a/rs/moq-tokio/Cargo.toml +++ b/rs/moq-tokio/Cargo.toml @@ -60,6 +60,8 @@ tcp = ["_transport", "dep:qmux", "dep:dns-lookup"] # Unix-domain-socket qmux transport (`unix://`), unix-only. Adds peer-credential # inspection on accept. Pulls in `tcp` (shares qmux's stream transport). uds = ["tcp", "qmux/uds"] +# Native WebRTC data channels over str0m, with a full ICE agent (the `webrtc` module). +webrtc = ["_transport", "dep:qmux", "dep:moq-ice", "dep:str0m", "dep:if-addrs", "dep:mdns-sd"] ring = ["_certs", "rustls/ring", "rcgen/ring", "moq-quic?/rustls-ring"] [dependencies] @@ -72,10 +74,13 @@ hex = { workspace = true } hmac = { workspace = true, optional = true } http = { version = "1", optional = true } humantime = "2.3" +if-addrs = { workspace = true, optional = true } kio = { workspace = true, optional = true } mdns-sd = { workspace = true, optional = true } moq-clock = { workspace = true, optional = true } +# The ICE agent behind the `webrtc` module. Private, like str0m: neither appears in this crate's API. +moq-ice = { workspace = true, optional = true } moq-net = { workspace = true } # The QUIC state machine the `quic` module drives; the crypto-provider features pick its provider. moq-quic = { workspace = true, features = ["bloom"], optional = true } @@ -101,6 +106,7 @@ serde = { version = "1", features = ["derive"] } serde_with = { version = "3", features = ["hex"] } sha2 = { workspace = true, optional = true } socket2 = { workspace = true, optional = true } +str0m = { workspace = true, optional = true } subtle = { workspace = true, optional = true } thiserror = "2" tikv-jemalloc-ctl = { version = "0.6", optional = true } @@ -138,6 +144,8 @@ chrono = "0.4" # be driven with synthetic daemon events. flume = "0.12" rcgen = { workspace = true, features = ["pem", "aws_lc_rs"] } +# A fake STUN server for the WebRTC server-reflexive candidate test. +stun-types = { workspace = true } tempfile = "3" # test-util adds tokio::time::pause/auto-advance for the time-dependent tests. tokio = { workspace = true, features = ["test-util"] } diff --git a/rs/moq-tokio/src/client.rs b/rs/moq-tokio/src/client.rs index 1b6eb42674e..31786a6259f 100644 --- a/rs/moq-tokio/src/client.rs +++ b/rs/moq-tokio/src/client.rs @@ -46,13 +46,7 @@ pub struct Client { /// every transport (passed into the QUIC backend's `connect` and used directly for /// raw TCP/UDS qmux and WebSocket). Resolved once in [`Client::new`] so the ALPN list /// can't diverge between transports. - #[cfg(any( - feature = "quic", - feature = "iroh", - feature = "websocket", - feature = "tcp", - feature = "uds" - ))] + #[cfg(feature = "_transport")] versions: moq_net::Versions, /// The URL from [`connect.url`](crate::connect::Config::url), dialed by [`Client::publish`] / [`Client::consume`]. connect: Option, @@ -144,13 +138,6 @@ impl Client { moq: moq_net::Client::new() .with_versions(versions.clone()) .with_extensions(config.extensions), - #[cfg(any( - feature = "quic", - feature = "iroh", - feature = "websocket", - feature = "tcp", - feature = "uds" - ))] versions, connect: config.url, timeout, @@ -311,6 +298,39 @@ impl Client { Some(self.with_subscriber(origin).connect(url)) } + /// Run a MoQ client session over the data channel an [`crate::webrtc::Offerer`] opened. + /// + /// Waits for ICE and the channel to come up, bounded by the connect timeout. A + /// peer link cannot be redialed, so there is no reconnect: the session ends with + /// the peer connection. + #[cfg(feature = "webrtc")] + pub async fn connect_webrtc(&self, peer: &crate::webrtc::Offerer) -> crate::Result { + let protocol = peer.protocol(); + if !self.versions.alpns().contains(&protocol) { + return Err(crate::webrtc::Error::Protocol(protocol.to_string()).into()); + } + + let attempt = async { + let opened = peer.channel().await?; + let config = opened.config; + let session = qmux::Session::connect(opened.channel, config) + .await + .map_err(crate::webrtc::Error::from)?; + // A data channel has no request URI, so a path would ride the SETUP; a peer link has none. + let transport = crate::transport::Session::new(session, moq_net::transport::Binding::InBand); + Ok::<_, Error>(connect_session(&self.moq, transport).await?) + }; + + let session = match self.timeout.is_zero() { + true => attempt.await?, + false => tokio::time::timeout(self.timeout, attempt) + .await + .map_err(|_| Error::ConnectTimeout(self.timeout))??, + }; + tracing::info!(version = %session.version(), transport = %crate::Transport::WebRtc, "connected"); + Ok(session) + } + /// Dial the given URL and complete the MoQ handshake. /// /// Errors if no transport feature is compiled in. @@ -763,13 +783,7 @@ fn race_error(quic: Error, websocket: Error) -> Error { } } -#[cfg(any( - feature = "quic", - feature = "iroh", - feature = "websocket", - feature = "tcp", - all(feature = "uds", unix) -))] +#[cfg(feature = "_transport")] async fn connect_session( client: &moq_net::Client, transport: S, diff --git a/rs/moq-tokio/src/error.rs b/rs/moq-tokio/src/error.rs index 497bb9a7ab8..37cf8a290df 100644 --- a/rs/moq-tokio/src/error.rs +++ b/rs/moq-tokio/src/error.rs @@ -199,11 +199,11 @@ pub enum Error { #[error("{0}")] Reconnect(String), - /// The connection was stopped locally, by closing it or dropping the last handle. + /// The connection or listener was stopped locally, by closing it or dropping the last handle. /// /// Not a failure: it is what a caller asked for. Distinct from [`Self::Reconnect`] /// so a status watcher can tell an expected teardown from a connection that gave up. - #[error("connection stopped")] + #[error("stopped locally")] Stopped, /// A task serving this connection panicked. The panic itself is logged. @@ -238,6 +238,11 @@ pub enum Error { #[cfg(all(feature = "uds", unix))] #[error(transparent)] Unix(Arc), + + /// The WebRTC data channel transport failed. + #[cfg(feature = "webrtc")] + #[error(transparent)] + WebRtc(Arc), } impl Error { @@ -404,6 +409,13 @@ impl From for Error { } } +#[cfg(feature = "webrtc")] +impl From for Error { + fn from(err: crate::webrtc::Error) -> Self { + Self::WebRtc(Arc::new(err)) + } +} + /// Convenience alias for results produced by this crate. pub type Result = std::result::Result; diff --git a/rs/moq-tokio/src/lib.rs b/rs/moq-tokio/src/lib.rs index a1566a9e838..82c6a029d1e 100644 --- a/rs/moq-tokio/src/lib.rs +++ b/rs/moq-tokio/src/lib.rs @@ -7,6 +7,7 @@ //! - TCP via the `tcp://` (plaintext) and `tls://` schemes (qmux; requires `tcp` feature) //! - Unix domain socket via the `unix://` scheme (qmux, peer-credential aware; requires `uds` feature, unix-only) //! - Iroh P2P (requires `iroh` feature) +//! - A WebRTC data channel with a browser peer, signaled by the caller (qmux; requires `webrtc` feature) //! //! See [`Client`] for connecting to relays and [`Server`] for accepting //! connections. The `mdns` feature finds peers to connect to on the local network. @@ -62,15 +63,17 @@ pub mod worker; mod util; #[cfg(feature = "watch")] pub mod watch; +#[cfg(feature = "webrtc")] +pub mod webrtc; #[cfg(feature = "websocket")] pub mod websocket; -/// Session settings for every qmux transport (WebSocket, `tcp://`, `tls://`, `unix://`). +/// Session settings for every qmux transport (WebSocket, `tcp://`, `tls://`, `unix://`, WebRTC). /// /// moq-lite holds a stream open per subscription, so a qmux session grants the same /// stream count as QUIC rather than qmux's default of 100, past which a peer's opens /// wait on credit forever. -#[cfg(any(feature = "websocket", feature = "tcp"))] +#[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] fn qmux_config() -> qmux::Config { let mut config = qmux::Config::default(); config.max_streams_bidi = quic::DEFAULT_MAX_STREAMS; diff --git a/rs/moq-tokio/src/server.rs b/rs/moq-tokio/src/server.rs index f7f8107fd98..5dad079b1c7 100644 --- a/rs/moq-tokio/src/server.rs +++ b/rs/moq-tokio/src/server.rs @@ -18,9 +18,19 @@ use url::Url; // the stream listeners hand back an already-built `Request`. #[cfg(any(feature = "quic", feature = "iroh", feature = "websocket"))] use futures::FutureExt; -use futures::future::BoxFuture; -use futures::stream::FuturesUnordered; -use futures::stream::StreamExt; +// The pending handshakes of the transports `Server` accepts itself; WebRTC +// signaling is the caller's, so a WebRTC-only build has none. +#[cfg(any( + feature = "quic", + feature = "iroh", + feature = "websocket", + feature = "tcp", + all(feature = "uds", unix) +))] +use futures::{ + future::BoxFuture, + stream::{FuturesUnordered, StreamExt}, +}; impl crate::listen::Config { /// Build the [`Server`] this config describes, binding its listeners. @@ -227,6 +237,13 @@ pub struct Server { versions: moq_net::Versions, /// From [`crate::listen::Config::resolved_timeout`]. timeout: Option, + #[cfg(any( + feature = "quic", + feature = "iroh", + feature = "websocket", + feature = "tcp", + all(feature = "uds", unix) + ))] accept: FuturesUnordered>>, #[cfg(any(feature = "tcp", all(feature = "uds", unix)))] streams: StreamListeners, @@ -411,6 +428,13 @@ impl Server { Ok(Server { activity, + #[cfg(any( + feature = "quic", + feature = "iroh", + feature = "websocket", + feature = "tcp", + all(feature = "uds", unix) + ))] accept: Default::default(), moq, versions, @@ -507,11 +531,11 @@ impl Server { feature = "tcp", all(feature = "uds", unix) )))] - /// Returns the next partially established session. - /// - /// Panics: no transport feature is compiled in, so nothing can be accepted. + /// Returns the next partially established session: none, since no compiled + /// transport is one this server listens on (WebRTC sessions arrive through + /// [`Listener::accept_webrtc`]). async fn accept_next(&mut self) -> Option { - unreachable!("no transport compiled; enable a QUIC backend, websocket, tcp, or uds feature"); + None } /// The accept-loop health of every listener this server owns that performs a real @@ -866,6 +890,56 @@ impl Listener { self.server.accept_next().await } + /// Read the MoQ SETUP from the data channel the remote peer opened on an + /// [`crate::webrtc::Answerer`], leaving admission to the returned [`Request`]. + /// + /// Waits for ICE and the channel to come up, bounded by the handshake timeout. + /// The channel must carry the answerer's protocol, a MoQ ALPN this server + /// speaks. Returns [`Error::Stopped`] once the listener is closing. + #[cfg(feature = "webrtc")] + pub async fn accept_webrtc(&self, peer: &crate::webrtc::Answerer) -> crate::Result { + use crate::webrtc::Error as WebRtcError; + + let server = &self.server; + let active = server.activity.enter().ok_or(Error::Stopped)?; + if !server.versions.alpns().contains(&peer.protocol()) { + return Err(WebRtcError::Protocol(peer.protocol().to_string()).into()); + } + let deadline = server + .timeout + .and_then(|timeout| tokio::time::Instant::now().checked_add(timeout)); + + let handshake = async { + let opened = peer.channel().await?; + if opened.protocol != peer.protocol() { + return Err(WebRtcError::Protocol(opened.protocol)); + } + let config = opened.config; + let protocol = opened.protocol; + let session = qmux::Session::accept(opened.channel, config).await?; + Ok((session, protocol)) + }; + let (session, protocol) = within(deadline, handshake).await?; + + let (request, deadline) = + accept_request(&server.moq, session, moq_net::transport::Binding::InBand, deadline).await?; + Ok(Request { + transport: Transport::WebRtc, + url: None, + authority: None, + identity: None, + link: Link { + alpn: Some(protocol), + ..Default::default() + }, + kind: RequestKind::Qmux(Box::new(request)), + #[cfg(feature = "quic")] + quic: None, + deadline, + active, + }) + } + /// Close every listener and connection, once each QUIC peer has been sent the /// close. /// @@ -966,7 +1040,8 @@ impl Setup { feature = "iroh", feature = "websocket", feature = "tcp", - all(feature = "uds", unix) + all(feature = "uds", unix), + feature = "webrtc" ))] /// Read the peer's MoQ SETUP by `deadline`, closing the session with a timeout /// code if it hasn't arrived. The deadline comes back too, since answering the @@ -1027,7 +1102,8 @@ impl Deadline { feature = "iroh", feature = "websocket", feature = "tcp", - all(feature = "uds", unix) + all(feature = "uds", unix), + feature = "webrtc" ))] fn new(at: tokio::time::Instant, session: S) -> Self where @@ -1070,7 +1146,8 @@ impl Deadline { feature = "iroh", feature = "websocket", feature = "tcp", - all(feature = "uds", unix) + all(feature = "uds", unix), + feature = "webrtc" ))] async fn within( deadline: Option, @@ -1520,7 +1597,12 @@ pub(crate) enum RequestKind { Quic(Box>>), #[cfg(feature = "iroh")] Iroh(Box>>), - #[cfg(any(feature = "tcp", all(feature = "uds", unix), feature = "websocket"))] + #[cfg(any( + feature = "tcp", + all(feature = "uds", unix), + feature = "websocket", + feature = "webrtc" + ))] Qmux(Box>>), } @@ -1604,7 +1686,12 @@ macro_rules! request_ref { RequestKind::Quic($r) => $body, #[cfg(feature = "iroh")] RequestKind::Iroh($r) => $body, - #[cfg(any(feature = "tcp", all(feature = "uds", unix), feature = "websocket"))] + #[cfg(any( + feature = "tcp", + all(feature = "uds", unix), + feature = "websocket", + feature = "webrtc" + ))] RequestKind::Qmux($r) => $body, } }; @@ -1618,7 +1705,12 @@ macro_rules! request_into { RequestKind::Quic($r) => $body, #[cfg(feature = "iroh")] RequestKind::Iroh($r) => $body, - #[cfg(any(feature = "tcp", all(feature = "uds", unix), feature = "websocket"))] + #[cfg(any( + feature = "tcp", + all(feature = "uds", unix), + feature = "websocket", + feature = "webrtc" + ))] RequestKind::Qmux($r) => $body, } }; @@ -1632,7 +1724,12 @@ macro_rules! request_map { RequestKind::Quic($r) => RequestKind::Quic(Box::new($body)), #[cfg(feature = "iroh")] RequestKind::Iroh($r) => RequestKind::Iroh(Box::new($body)), - #[cfg(any(feature = "tcp", all(feature = "uds", unix), feature = "websocket"))] + #[cfg(any( + feature = "tcp", + all(feature = "uds", unix), + feature = "websocket", + feature = "webrtc" + ))] RequestKind::Qmux($r) => RequestKind::Qmux(Box::new($body)), } }; diff --git a/rs/moq-tokio/src/transport.rs b/rs/moq-tokio/src/transport.rs index e908eb41d18..2244aa9b8e5 100644 --- a/rs/moq-tokio/src/transport.rs +++ b/rs/moq-tokio/src/transport.rs @@ -37,6 +37,8 @@ pub enum Transport { Unix, /// WebTransport over HTTP/3 on QUIC. WebTransport, + /// A WebRTC data channel using qmux framing. + WebRtc, } impl Transport { @@ -49,6 +51,7 @@ impl Transport { Self::Tcp => "tcp", Self::Unix => "unix", Self::WebTransport => "webtransport", + Self::WebRtc => "webrtc", } } } @@ -906,10 +909,10 @@ impl GroupedSend for web_transport_iroh::SendStream { } } -#[cfg(any(feature = "websocket", feature = "tcp"))] +#[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] impl Grouped for qmux::Session {} -#[cfg(any(feature = "websocket", feature = "tcp"))] +#[cfg(any(feature = "websocket", feature = "tcp", feature = "webrtc"))] impl GroupedSend for qmux::SendStream { type Group = moq_net::transport::poll::Ungrouped; diff --git a/rs/moq-tokio/src/webrtc.rs b/rs/moq-tokio/src/webrtc.rs new file mode 100644 index 00000000000..5664d861797 --- /dev/null +++ b/rs/moq-tokio/src/webrtc.rs @@ -0,0 +1,962 @@ +//! MoQ over one reliable, ordered WebRTC data channel, for sessions with browser peers. +//! +//! An [`Offerer`] or [`Answerer`] is one side of a str0m peer connection with a +//! full ICE agent: a dual-stack UDP socket, a host candidate per non-loopback +//! interface address, a server-reflexive candidate from each configured STUN +//! server, and remote `.local` candidates resolved over mDNS. Signaling is the +//! caller's: send the [`Offer`] or [`Answer`] and the local [`Candidate`]s to the +//! remote peer, and feed its replies back. +//! +//! The [`Offerer`] opens the data channel, labelled `moq` and carrying the agreed +//! MoQ ALPN as its protocol, and runs the MoQ client over it with +//! [`crate::Client::connect_webrtc`]. The [`Answerer`] adopts that channel and +//! runs the MoQ server with [`crate::Listener::accept_webrtc`]. This matches +//! `@moq/p2p`, where `Transport.connect` creates the channel and +//! `Transport.accept` adopts it. +//! +//! The channel carries qmux-01, one record per SCTP message, capped at 16 KiB and +//! the peer's SCTP message limit. Ordered delivery means one lost chunk stalls +//! every stream until SCTP retransmits it. + +use std::net::{IpAddr, SocketAddr}; +use std::str::FromStr; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use bytes::Bytes; +use str0m::Rtc; +use str0m::change::{SdpAnswer, SdpOffer, SdpPendingOffer}; +use str0m::channel::{ChannelConfig, ChannelId, Reliability}; +use tokio::net::UdpSocket; +use tokio::sync::{mpsc, oneshot}; + +mod channel; + +/// The data channel label, as `@moq/p2p` creates it. +const LABEL: &str = "moq"; + +/// The largest qmux record, as `@moq/p2p` sends: SCTP flow control is per +/// message, so a smaller record keeps one stream from holding the association. +const RECORD_SIZE: u64 = 16 * 1024; + +/// qmux-01 does not allow a peer to advertise a smaller record limit. +const MIN_RECORD_SIZE: u64 = 16_382; + +/// SCTP's message limit when a description omits `a=max-message-size` (RFC 8841 section 6.1). +const DEFAULT_MAX_MESSAGE_SIZE: u64 = 64 * 1024; + +/// Bytes the channel may hold unsent before qmux's writer waits, as `@moq/p2p` allows. +const SEND_WINDOW: usize = RECORD_SIZE as usize * 4; + +/// Unread bytes the channel tolerates past qmux's advertised `max_data`, for the +/// frame headers and control frames that flow control does not count. A peer that +/// honors flow control with reasonably full records stays under it; one that +/// ignores it has the channel closed instead of growing memory without limit. +const READ_SLACK: u64 = 16 * RECORD_SIZE; + +/// How long to wait for a `.local` name to resolve: mdns-sd's first query plus its +/// retransmissions at one and two seconds (RFC 6762 section 5.2). +const MDNS_TIMEOUT: Duration = Duration::from_secs(5); + +/// The default STUN port (RFC 8489 section 18.10). +const STUN_PORT: u16 = 3478; + +/// Errors from a WebRTC peer connection or its data channel. +#[derive(Debug, thiserror::Error)] +#[non_exhaustive] +pub enum Error { + /// Binding the UDP socket or resolving a STUN server failed. + #[error(transparent)] + Io(#[from] std::io::Error), + + /// The peer disconnected, or the connection was closed. + #[error("peer connection closed")] + Closed, + + /// ICE never connected before the establishment deadline. + #[error("ICE did not connect before the establishment deadline")] + Timeout, + + /// The WebRTC stack failed (DTLS, SCTP, or ICE state). + #[error("rtc error: {0}")] + Rtc(String), + + /// A session description failed to parse or apply. + #[error("invalid session description: {0}")] + Sdp(String), + + /// An ICE candidate failed to parse. + #[error("invalid ICE candidate: {0}")] + Candidate(String), + + /// An ICE server URL is malformed or not a `stun:` URL. + #[error("unsupported ICE server: {0}")] + Server(String), + + /// No interface has an address to offer as a host candidate. + #[error("no network interface address to offer as a host candidate")] + NoHost, + + /// A `.local` candidate could not be resolved. + #[error("mDNS resolution failed: {0}")] + Mdns(String), + + /// The data channel was already taken by a session. + #[error("the data channel is already in use")] + Taken, + + /// The protocol is not a MoQ ALPN this endpoint speaks, or the channel's is not the one signaling agreed. + #[error("unsupported MoQ protocol: {0:?}")] + Protocol(String), + + /// The peer opened an unordered or partially reliable channel, which qmux cannot run over. + #[error("the data channel must be reliable and ordered")] + Unreliable, + + /// The peer's SCTP message limit is below qmux's minimum record size. + #[error("SCTP message limit {0} is smaller than qmux's minimum record size")] + MessageSize(u64), + + /// The qmux handshake over the channel failed. + #[error("qmux handshake failed: {0}")] + Qmux(String), +} + +crate::error::from_message!(qmux::Error => Qmux); + +impl Error { + /// Adopt the agent's error. Not a `From` impl, which would put moq-ice in this crate's API. + fn ice(err: moq_ice::Error) -> Self { + match err { + moq_ice::Error::Closed => Self::Closed, + moq_ice::Error::Timeout => Self::Timeout, + moq_ice::Error::Rtc(err) => Self::Rtc(err), + err => Self::Rtc(err.to_string()), + } + } +} + +/// How an [`Offerer`] or [`Answerer`] gathers its local candidates. +#[derive(Clone, Debug, Default)] +#[non_exhaustive] +pub struct Config { + /// STUN servers asked for this host's public address. Empty means LAN only: + /// host candidates and nothing else. + pub stun: Vec, +} + +/// A STUN server, written the way an `RTCIceServer` URL names one: `stun:host[:port]`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Stun { + host: String, + port: u16, +} + +impl FromStr for Stun { + type Err = Error; + + fn from_str(url: &str) -> Result { + let invalid = || Error::Server(url.to_string()); + let (scheme, rest) = url.split_once(':').ok_or_else(invalid)?; + if !scheme.eq_ignore_ascii_case("stun") { + // TURN is out of scope: the relay already carries what a TURN server would. + return Err(invalid()); + } + + let (host, port) = match rest.strip_prefix('[') { + Some(rest) => { + let (host, port) = rest.split_once(']').ok_or_else(invalid)?; + let port = match port { + "" => None, + port => Some(port.strip_prefix(':').ok_or_else(invalid)?), + }; + ( + host.parse::().map_err(|_| invalid())?.to_string(), + port, + ) + } + None => { + let (host, port) = match rest.split_once(':') { + Some((host, port)) => (host, Some(port)), + None => (rest, None), + }; + // A DNS name or IPv4 address, so a fragment or stray character fails here + // instead of as a lookup that only warns. + if host.is_empty() + || !host + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '.' | '_')) + { + return Err(invalid()); + } + (host.to_string(), port) + } + }; + let port = match port { + Some(port) => port.parse().map_err(|_| invalid())?, + None => STUN_PORT, + }; + Ok(Self { host, port }) + } +} + +impl std::fmt::Display for Stun { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self.host.contains(':') { + true => write!(f, "stun:[{}]:{}", self.host, self.port), + false => write!(f, "stun:{}:{}", self.host, self.port), + } + } +} + +/// An SDP offer, as `RTCSessionDescription.sdp` carries it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Offer(String); + +/// An SDP answer, as `RTCSessionDescription.sdp` carries it. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Answer(String); + +/// An ICE candidate, as `RTCIceCandidate.candidate` carries it: `candidate:...`. +/// +/// The connection has one media section, so a browser takes it with `sdpMLineIndex: 0`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Candidate(String); + +macro_rules! text { + ($ty:ident) => { + impl $ty { + /// Wrap the text a browser produced. + pub fn new(text: impl Into) -> Self { + Self(text.into()) + } + + /// The text to hand a browser. + pub fn as_str(&self) -> &str { + &self.0 + } + } + + impl std::fmt::Display for $ty { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } + } + }; +} + +text!(Offer); +text!(Answer); +text!(Candidate); + +/// The offering side of a WebRTC peer connection, which opens the MoQ data channel +/// and runs the MoQ client over it with [`crate::Client::connect_webrtc`]. +/// +/// A background task drives the connection from creation, so ICE checks and STUN +/// replies are answered while signaling is still in progress. It runs until both +/// this handle and any session over its data channel are gone. Dropping the handle +/// stops signaling but leaves a running session alone. +pub struct Offerer { + peer: Peer, + /// The MoQ ALPN the channel carries. + protocol: String, +} + +impl Offerer { + /// Start a peer connection with a data channel for MoQ `protocol`, the ALPN the + /// peers agreed through signaling, returning the offer to send. + pub async fn new(config: &Config, protocol: &str) -> Result<(Self, Offer), Error> { + Self::with_hosts(config, protocol, hosts()?).await + } + + async fn with_hosts(config: &Config, protocol: &str, hosts: Vec) -> Result<(Self, Offer), Error> { + let protocol = negotiated(protocol)?; + let socket = bind(&hosts)?; + let mut rtc = rtc(&socket.locals)?; + + let mut api = rtc.sdp_api(); + let id = api.add_channel_with_config(ChannelConfig { + label: LABEL.to_string(), + ordered: true, + reliability: Reliability::Reliable, + negotiated: None, + protocol: protocol.to_string(), + }); + let (offer, pending) = api.apply().expect("adding a channel is an SDP change"); + + let role = Role::Offerer { + channel: id, + pending: Some(pending), + }; + let peer = Peer::spawn(config, rtc, socket, role, None).await?; + let offerer = Self { peer, protocol }; + Ok((offerer, Offer(offer.to_sdp_string()))) + } + + /// Apply the remote peer's answer to this offer. + pub async fn accept(&self, answer: &Answer) -> Result<(), Error> { + let (sdp, mdns) = split_mdns(answer.as_str()); + let max_message = max_message_size(answer.as_str())?; + let parsed = SdpAnswer::from_sdp_string(&sdp).map_err(|err| Error::Sdp(err.to_string()))?; + + let (tx, rx) = oneshot::channel(); + self.peer.send(Command::Answer(parsed, max_message, tx))?; + rx.await.map_err(|_| Error::Closed)??; + self.peer.resolve(mdns); + Ok(()) + } + + /// The next local candidate to send the remote peer, or `None` once gathering is done. + /// + /// Host candidates ride in the offer; these are the server-reflexive ones that + /// STUN finds afterwards. + pub async fn candidate(&mut self) -> Option { + self.peer.candidates.recv().await + } + + /// Add a candidate the remote peer sent, resolving a `.local` address over mDNS first. + /// + /// An empty candidate, which a browser sends to mark the end of its candidates, is ignored. + pub async fn add_candidate(&self, candidate: &Candidate) -> Result<(), Error> { + self.peer.add_candidate(candidate).await + } + + /// The MoQ ALPN this peer's data channel carries. + pub(crate) fn protocol(&self) -> &str { + &self.protocol + } + + /// Wait for the data channel to open, and take it. + pub(crate) async fn channel(&self) -> Result { + self.peer.channel().await + } +} + +/// The answering side of a WebRTC peer connection, which adopts the remote peer's +/// MoQ data channel and serves it with [`crate::Listener::accept_webrtc`]. +/// +/// Like [`Offerer`], a background task drives the connection until both this +/// handle and any session over its data channel are gone. +pub struct Answerer { + peer: Peer, + /// The MoQ ALPN the remote peer's channel must carry. + protocol: String, +} + +impl Answerer { + /// Start a peer connection for the remote peer's offer and MoQ `protocol`, the + /// ALPN the peers agreed through signaling, returning the answer to send. + pub async fn new(config: &Config, offer: &Offer, protocol: &str) -> Result<(Self, Answer), Error> { + Self::with_hosts(config, offer, protocol, hosts()?).await + } + + async fn with_hosts( + config: &Config, + offer: &Offer, + protocol: &str, + hosts: Vec, + ) -> Result<(Self, Answer), Error> { + let protocol = negotiated(protocol)?; + let (sdp, mdns) = split_mdns(offer.as_str()); + let max_message = max_message_size(offer.as_str())?; + let parsed = SdpOffer::from_sdp_string(&sdp).map_err(|err| Error::Sdp(err.to_string()))?; + + let socket = bind(&hosts)?; + let mut rtc = rtc(&socket.locals)?; + let answer = rtc + .sdp_api() + .accept_offer(parsed) + .map_err(|err| Error::Sdp(err.to_string()))?; + + let peer = Peer::spawn(config, rtc, socket, Role::Answerer, Some(max_message)).await?; + peer.resolve(mdns); + Ok((Self { peer, protocol }, Answer(answer.to_sdp_string()))) + } + + /// The next local candidate to send the remote peer, or `None` once gathering is done. + /// + /// Host candidates ride in the answer; these are the server-reflexive ones that + /// STUN finds afterwards. + pub async fn candidate(&mut self) -> Option { + self.peer.candidates.recv().await + } + + /// Add a candidate the remote peer sent, resolving a `.local` address over mDNS first. + /// + /// An empty candidate, which a browser sends to mark the end of its candidates, is ignored. + pub async fn add_candidate(&self, candidate: &Candidate) -> Result<(), Error> { + self.peer.add_candidate(candidate).await + } + + /// The MoQ ALPN the remote peer's data channel must carry. + pub(crate) fn protocol(&self) -> &str { + &self.protocol + } + + /// Wait for the remote peer's data channel to open, and take it. + pub(crate) async fn channel(&self) -> Result { + self.peer.channel().await + } +} + +/// Check a MoQ ALPN agreed through signaling, as `@moq/p2p` does. +fn negotiated(protocol: &str) -> Result { + match protocol.is_empty() || protocol.contains(char::is_whitespace) { + true => Err(Error::Protocol(protocol.to_string())), + false => Ok(protocol.to_string()), + } +} + +/// The signaling handle both roles share: commands to the driver task and the +/// candidates it gathers. +struct Peer { + commands: mpsc::UnboundedSender, + candidates: mpsc::UnboundedReceiver, +} + +impl Peer { + async fn spawn( + config: &Config, + rtc: Rtc, + socket: moq_ice::Socket, + role: Role, + max_message: Option, + ) -> Result { + let stun = resolve_stun(&config.stun).await; + let mut agent = moq_ice::Agent::new(rtc, socket)?.with_stun(&stun); + if let Role::Offerer { .. } = role { + // The answer applies through `rtc_mut`, which starts the deadline. + agent = agent.awaiting_answer(); + } + + let (commands, rx) = mpsc::unbounded_channel(); + let (candidates_tx, candidates) = mpsc::unbounded_channel(); + let driver = Driver { + agent, + commands: rx, + candidates: Some(candidates_tx), + role, + max_message, + channel: None, + waiter: None, + taken: false, + }; + use tracing::Instrument; + tokio::spawn(driver.run().instrument(tracing::Span::current())); + + Ok(Self { commands, candidates }) + } + + async fn add_candidate(&self, candidate: &Candidate) -> Result<(), Error> { + let line = candidate.as_str().trim(); + let line = line.strip_prefix("a=").unwrap_or(line); + if line.is_empty() { + return Ok(()); + } + let candidate = parse_candidate(line).await?; + self.send(Command::Candidate(candidate)) + } + + async fn channel(&self) -> Result { + let (tx, rx) = oneshot::channel(); + self.send(Command::Channel(tx))?; + let mut opened = rx.await.map_err(|_| Error::Closed)??; + // The session keeps the driver alive once this handle is gone. + opened.channel.writer._driver = Some(self.commands.clone()); + Ok(opened) + } + + fn send(&self, command: Command) -> Result<(), Error> { + self.commands.send(command).map_err(|_| Error::Closed) + } + + /// Add the `.local` candidates a description carried once mDNS resolves them. + fn resolve(&self, lines: Vec) { + for line in lines { + let commands = self.commands.clone(); + tokio::spawn(async move { + match parse_candidate(&line).await { + Ok(candidate) => { + let _ = commands.send(Command::Candidate(candidate)); + } + Err(err) => tracing::debug!(%err, "dropped a remote candidate"), + } + }); + } + } +} + +/// A data channel that opened, ready to carry qmux. +pub(crate) struct Opened { + pub channel: channel::Channel, + /// The MoQ ALPN the channel carries. + pub protocol: String, + /// The qmux session settings for this channel. + pub config: qmux::Config, +} + +/// The qmux session settings for a channel carrying `protocol`, advertising +/// `max_record_size`: 16 KiB, capped by the peer's SCTP message limit. +fn session_config(protocol: &str, max_record_size: u64) -> qmux::Config { + let mut config = crate::qmux_config(); + config.version = qmux::Version::QMux01; + config.protocol = qmux::Protocol::Negotiated(protocol.to_string()); + config.max_record_size = max_record_size; + // The binding has no datagrams. + config.max_datagram_frame_size = 0; + config +} + +pub(crate) enum Command { + Answer(SdpAnswer, u64, oneshot::Sender>), + Candidate(str0m::Candidate), + Channel(oneshot::Sender>), +} + +enum Role { + Offerer { + channel: ChannelId, + /// Until the answer is applied. + pending: Option, + }, + Answerer, +} + +/// The open data channel, from the driver's side. +struct Open { + id: ChannelId, + protocol: String, + reads: channel::Inbox, + /// `None` once the session's writer closed. + writes: Option>, + config: qmux::Config, + /// The session's halves, until a session takes them. + halves: Option, +} + +/// Owns the connection and runs it on its own task. +struct Driver { + agent: moq_ice::Agent, + commands: mpsc::UnboundedReceiver, + /// Dropped once gathering finishes, ending [`Offerer::candidate`] and [`Answerer::candidate`]. + candidates: Option>, + role: Role, + /// The remote peer's SCTP message limit, once its description is applied. + max_message: Option, + channel: Option>, + /// The session waiting for the channel to open. + waiter: Option>>, + /// Whether a session already asked for the channel, which only one may carry. + taken: bool, +} + +impl Driver { + async fn run(mut self) { + let err = loop { + let writable = self.writable(); + let writes = match &mut self.channel { + Some(Ok(Open { + writes: Some(writes), .. + })) => Some(writes), + _ => None, + }; + + tokio::select! { + event = self.agent.next() => match event { + Ok(event) => if let Err(err) = self.event(event) { + break err; + }, + Err(err) => break Error::ice(err), + }, + command = self.commands.recv() => match command { + Some(command) => self.command(command), + // Neither the signaling handle nor a session is left to use the connection. + None => return, + }, + data = next_write(writes), if writable => { + if let Err(err) = self.write(data) { + break err; + } + } + } + }; + + match &err { + Error::Closed => tracing::debug!("peer connection closed"), + err => tracing::debug!(%err, "peer connection failed"), + } + if let Some(waiter) = self.waiter.take() { + let _ = waiter.send(Err(err)); + } + } + + /// Whether the channel can take another record without passing [`SEND_WINDOW`]. + fn writable(&mut self) -> bool { + let Some(Ok(Open { + id, writes: Some(_), .. + })) = &self.channel + else { + return false; + }; + let id = *id; + self.agent + .rtc_mut() + .channel(id) + .is_some_and(|mut channel| channel.buffered_amount() < SEND_WINDOW) + } + + /// Hand a record from the session to SCTP, or close the channel when the session's writer closed. + fn write(&mut self, data: Option) -> Result<(), Error> { + let Some(Ok(open)) = &mut self.channel else { + return Ok(()); + }; + let id = open.id; + let Some(data) = data else { + open.writes = None; + self.agent.rtc_mut().direct_api().close_data_channel(id); + return Ok(()); + }; + let written = { + let mut rtc = self.agent.rtc_mut(); + let mut channel = rtc.channel(id).ok_or(Error::Closed)?; + channel.write(true, &data).map_err(|err| Error::Rtc(err.to_string()))? + }; + // The send window keeps the buffer well below str0m's, so a refusal is a bug, not backpressure. + match written { + true => Ok(()), + false => Err(Error::Rtc("SCTP refused a record within the send window".into())), + } + } + + fn command(&mut self, command: Command) { + match command { + Command::Answer(answer, max_message, reply) => { + // Only an `Offerer` sends an answer, so the answering role never has one pending. + let pending = match &mut self.role { + Role::Offerer { pending, .. } => pending.take(), + Role::Answerer => None, + }; + let result = match pending { + Some(pending) => { + let accepted = self.agent.rtc_mut().sdp_api().accept_answer(pending, answer); + accepted + .map(|()| self.max_message = Some(max_message)) + .map_err(|err| Error::Sdp(err.to_string())) + } + None => Err(Error::Sdp("the answer was already applied".into())), + }; + let _ = reply.send(result); + } + Command::Candidate(candidate) => self.agent.rtc_mut().add_remote_candidate(candidate), + Command::Channel(reply) => { + if self.taken { + let _ = reply.send(Err(Error::Taken)); + return; + } + self.taken = true; + self.waiter = Some(reply); + self.deliver(); + } + } + } + + fn event(&mut self, event: moq_ice::Event) -> Result<(), Error> { + let event = match event { + moq_ice::Event::Candidate(candidate) => { + if let Some(candidates) = &self.candidates { + let _ = candidates.send(Candidate(candidate.to_sdp_string())); + } + return Ok(()); + } + moq_ice::Event::Gathered => { + self.candidates = None; + return Ok(()); + } + moq_ice::Event::Rtc(event) => event, + _ => return Ok(()), + }; + + match event { + str0m::Event::ChannelOpen(id, _) => self.open(id), + str0m::Event::ChannelData(data) => { + let Some(Ok(open)) = &self.channel else { + return Ok(()); + }; + if data.id != open.id { + return Ok(()); + } + let record = if !data.binary { + Err(qmux::Error::ProtocolViolation) + } else if data.data.len() as u64 > open.config.max_record_size { + Err(qmux::Error::FrameTooLarge) + } else { + Ok(Bytes::from(data.data)) + }; + if !open.reads.push(record) { + tracing::debug!("closing the data channel: the peer broke qmux framing or flow control"); + let id = open.id; + self.agent.rtc_mut().direct_api().close_data_channel(id); + self.channel = Some(Err(Error::Closed)); + } + } + str0m::Event::ChannelClose(id) => { + if let Some(Ok(open)) = &self.channel + && open.id == id + { + // The session's reader sees the end of its records. + self.channel = Some(Err(Error::Closed)); + } + } + _ => {} + } + Ok(()) + } + + /// Take a channel that opened as the session's, if it is the one this role expects. + fn open(&mut self, id: ChannelId) { + let expected = match &self.role { + Role::Offerer { channel, .. } => *channel == id, + // The remote peer may open other channels first; only the `moq` one is the session's. + Role::Answerer => self + .agent + .rtc_mut() + .channel(id) + .and_then(|channel| channel.config().map(|config| config.label == LABEL)) + .unwrap_or(false), + }; + if !expected || self.channel.is_some() { + tracing::debug!(?id, "closing an unexpected data channel"); + self.agent.rtc_mut().direct_api().close_data_channel(id); + return; + } + + let opened = self.adopt(id); + if opened.is_err() { + self.agent.rtc_mut().direct_api().close_data_channel(id); + } + self.channel = Some(opened); + self.deliver(); + } + + fn adopt(&mut self, id: ChannelId) -> Result { + let max_message = self.max_message.unwrap_or(DEFAULT_MAX_MESSAGE_SIZE); + let max_record_size = RECORD_SIZE.min(max_message); + if max_record_size < MIN_RECORD_SIZE { + return Err(Error::MessageSize(max_message)); + } + + let protocol = { + let mut rtc = self.agent.rtc_mut(); + let mut channel = rtc.channel(id).ok_or(Error::Closed)?; + let config = channel.config().ok_or(Error::Unreliable)?; + if !config.ordered || config.reliability != Reliability::Reliable { + return Err(Error::Unreliable); + } + let protocol = config.protocol.clone(); + channel.set_buffered_amount_low_threshold(SEND_WINDOW / 2); + protocol + }; + + let config = session_config(&protocol, max_record_size); + let (reads, reader) = channel::queue(config.max_data + READ_SLACK); + let (tx, writes) = mpsc::channel(1); + Ok(Open { + id, + protocol, + reads, + writes: Some(writes), + config, + halves: Some(channel::Channel { + writer: channel::Writer { + tx: Some(tx), + max: usize::try_from(max_message).unwrap_or(usize::MAX), + _driver: None, + }, + reader, + }), + }) + } + + /// Answer a waiting session once the channel opened or failed to. + fn deliver(&mut self) { + let Some(channel) = &mut self.channel else { + return; + }; + let Some(waiter) = self.waiter.take() else { + return; + }; + let result = match channel { + Ok(open) => Ok(Opened { + channel: open.halves.take().expect("only the first session takes the halves"), + protocol: open.protocol.clone(), + config: open.config.clone(), + }), + Err(err) => Err(match err { + Error::Unreliable => Error::Unreliable, + Error::MessageSize(size) => Error::MessageSize(*size), + _ => Error::Closed, + }), + }; + let _ = waiter.send(result); + } +} + +/// The session's next record, or `None` when its writer closed. Pends without a channel. +async fn next_write(writes: Option<&mut mpsc::Receiver>) -> Option { + match writes { + Some(writes) => writes.recv().await, + None => std::future::pending().await, + } +} + +/// A str0m connection with this socket's host candidates. +fn rtc(locals: &[SocketAddr]) -> Result { + let mut rtc = Rtc::builder().build(Instant::now()); + for &addr in locals { + let candidate = str0m::Candidate::host(addr, "udp").map_err(|err| Error::Candidate(err.to_string()))?; + rtc.add_local_candidate(candidate); + } + Ok(rtc) +} + +/// Every address worth offering as a host candidate: not loopback, and not +/// link-local, which IPv6 cannot use without a scope and str0m refuses for IPv4. +fn hosts() -> Result, Error> { + let mut hosts = Vec::new(); + for interface in if_addrs::get_if_addrs()? { + let ip = interface.ip(); + if interface.is_loopback() || interface.is_link_local() || ip.is_multicast() || hosts.contains(&ip) { + continue; + } + hosts.push(ip); + } + Ok(hosts) +} + +/// Bind a dual-stack socket on an ephemeral port, pairing it with each host +/// address, and start reading it. +fn bind(hosts: &[IpAddr]) -> Result { + let socket = crate::bind::udp(crate::bind::Udp::new("[::]:0".parse().unwrap())) + .or_else(|_| crate::bind::udp(crate::bind::Udp::new("0.0.0.0:0".parse().unwrap())))?; + socket.set_nonblocking(true)?; + // A `[::]` socket that stayed v6-only can't send to IPv4 candidates. + let ipv4 = socket.local_addr()?.is_ipv4() || crate::bind::udp_is_dual_stack(&socket); + let socket = UdpSocket::from_std(socket)?; + let local = socket.local_addr()?; + + let locals: Vec = hosts + .iter() + .filter(|ip| if ip.is_ipv4() { ipv4 } else { local.is_ipv6() }) + .map(|&ip| SocketAddr::new(ip, local.port())) + .collect(); + if locals.is_empty() { + return Err(Error::NoHost); + } + let udp = Arc::new(socket); + Ok(moq_ice::Socket { + inbound: moq_ice::reader(udp.clone()), + udp, + locals, + }) +} + +/// Look up each STUN server. One that does not resolve is skipped, since a missing +/// reflexive candidate only costs the paths that needed it. +async fn resolve_stun(servers: &[Stun]) -> Vec { + let mut addrs = Vec::new(); + for server in servers { + match tokio::net::lookup_host((server.host.as_str(), server.port)).await { + Ok(found) => addrs.extend(found), + Err(err) => tracing::warn!(%err, %server, "failed to resolve STUN server"), + } + } + addrs +} + +/// The SCTP message limit a description advertises, where `0` means unlimited. +fn max_message_size(sdp: &str) -> Result { + let Some(value) = sdp + .lines() + .find_map(|line| line.trim().strip_prefix("a=max-message-size:")) + else { + return Ok(DEFAULT_MAX_MESSAGE_SIZE); + }; + match value.trim().parse::() { + Ok(0) => Ok(u64::MAX), + Ok(size) => Ok(size), + Err(_) => Err(Error::Sdp(format!("invalid max-message-size: {value}"))), + } +} + +/// Split a description's `.local` candidates out, since str0m only parses IP addresses. +fn split_mdns(sdp: &str) -> (String, Vec) { + let mut kept = String::with_capacity(sdp.len()); + let mut mdns = Vec::new(); + for line in sdp.split_inclusive('\n') { + if let Some(candidate) = line.trim_end().strip_prefix("a=") + && candidate.starts_with("candidate:") + && candidate_address(candidate).is_some_and(|addr| addr.ends_with(".local")) + { + mdns.push(candidate.to_string()); + continue; + } + kept.push_str(line); + } + (kept, mdns) +} + +/// The connection address of a `candidate:` line (RFC 8839 section 5.1). +fn candidate_address(line: &str) -> Option<&str> { + line.split_whitespace().nth(4) +} + +/// Parse a `candidate:` line, resolving a `.local` address over mDNS first. +async fn parse_candidate(line: &str) -> Result { + let line = match candidate_address(line) { + Some(host) if host.ends_with(".local") => { + let ip = mdns(host).await?; + let mut fields: Vec = line.split_whitespace().map(str::to_string).collect(); + fields[4] = ip.to_string(); + fields.join(" ") + } + _ => line.to_string(), + }; + str0m::Candidate::from_sdp_string(&line).map_err(|err| Error::Candidate(err.to_string())) +} + +/// The process-wide mDNS daemon, started on the first `.local` candidate. +static MDNS: Mutex> = Mutex::new(None); + +/// Resolve a browser's `.local` host name, preferring an IPv4 address. +async fn mdns(host: &str) -> Result { + let daemon = { + let mut daemon = MDNS.lock().unwrap(); + match daemon.as_ref() { + Some(daemon) => daemon.clone(), + None => daemon + .insert(mdns_sd::ServiceDaemon::new().map_err(|err| Error::Mdns(err.to_string()))?) + .clone(), + } + }; + + let name = format!("{host}."); + let timeout = MDNS_TIMEOUT.as_millis() as u64; + let events = daemon + .resolve_hostname(&name, Some(timeout)) + .map_err(|err| Error::Mdns(err.to_string()))?; + let found = loop { + match events.recv_async().await { + Ok(mdns_sd::HostnameResolutionEvent::AddressesFound(_, addrs)) if !addrs.is_empty() => { + let addrs: Vec = addrs.iter().map(|addr| addr.to_ip_addr()).collect(); + break addrs.iter().find(|ip| ip.is_ipv4()).or(addrs.first()).copied(); + } + Ok(mdns_sd::HostnameResolutionEvent::SearchTimeout(_)) | Err(_) => break None, + Ok(_) => {} + } + }; + let _ = daemon.stop_resolve_hostname(&name); + found.ok_or_else(|| Error::Mdns(format!("{host} did not resolve"))) +} + +#[cfg(test)] +mod tests; diff --git a/rs/moq-tokio/src/webrtc/channel.rs b/rs/moq-tokio/src/webrtc/channel.rs new file mode 100644 index 00000000000..e5c1c7ca1a6 --- /dev/null +++ b/rs/moq-tokio/src/webrtc/channel.rs @@ -0,0 +1,154 @@ +//! qmux message I/O over one reliable, ordered data channel. +//! +//! Each SCTP message carries one qmux-01 record without its size field, as a +//! WebSocket message does. The peer's driver task owns the channel; these halves +//! only hand messages to and from it. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; + +use bytes::Bytes; +use tokio::sync::mpsc; + +/// The data channel, as the qmux transport a [`qmux::Session`] runs over. +pub(crate) struct Channel { + pub writer: Writer, + pub reader: Reader, +} + +impl qmux::transport::Transport for Channel { + type Writer = Writer; + type Reader = Reader; + + fn split(self) -> (Writer, Reader) { + (self.writer, self.reader) + } +} + +/// Hands outbound records to the driver, which waits for SCTP to drain before +/// taking more, so a full send window holds back qmux's writer. +pub(crate) struct Writer { + /// `None` once closed: dropping the sender is what tells the driver to close the channel. + pub tx: Option>, + /// The largest message the peer accepts. + pub max: usize, + /// Keeps the driver running while the session outlives the signaling handle. + pub _driver: Option>, +} + +impl qmux::transport::Writer for Writer { + async fn send(&mut self, data: Bytes) -> Result<(), qmux::Error> { + if data.len() > self.max { + return Err(qmux::Error::FrameTooLarge); + } + let tx = self.tx.as_ref().ok_or(qmux::Error::Closed)?; + tx.send(data).await.map_err(|_| qmux::Error::Closed) + } + + async fn close(&mut self) -> Result<(), qmux::Error> { + self.tx = None; + Ok(()) + } +} + +/// A queue of inbound records bounded by bytes the session has yet to read. +/// +/// The driver never stops reading the socket to wait for the session: that would +/// also stall ICE consent checks and the SACKs that free our own send buffer. +/// qmux's flow control bounds what an honest peer can have in flight, so the +/// queue only overflows when the peer ignores it, and then the channel closes. +pub(crate) fn queue(budget: u64) -> (Inbox, Reader) { + let (tx, rx) = mpsc::unbounded_channel(); + let unread = Arc::new(AtomicU64::new(0)); + let inbox = Inbox { + tx, + unread: unread.clone(), + budget, + }; + (inbox, Reader { rx, unread }) +} + +/// The driver's end of the inbound queue. +pub(crate) struct Inbox { + tx: mpsc::UnboundedSender>, + /// Bytes queued but not yet read, shared with the [`Reader`]. + unread: Arc, + budget: u64, +} + +impl Inbox { + /// Queue a record for the session, returning false once its reads have ended: + /// on an error, or with a flow control error when the record would leave more + /// than the budget unread. + pub fn push(&self, record: Result) -> bool { + // A closed reader drops the record. + let data = match record { + Ok(data) => data, + Err(err) => { + let _ = self.tx.send(Err(err)); + return false; + } + }; + // qmux records are never empty, so the byte budget also bounds their count. + if data.is_empty() { + let _ = self.tx.send(Err(qmux::Error::ProtocolViolation)); + return false; + } + let len = data.len() as u64; + if self.unread.fetch_add(len, Ordering::Relaxed) + len > self.budget { + let _ = self.tx.send(Err(qmux::Error::FlowControlError)); + return false; + } + let _ = self.tx.send(Ok(data)); + true + } +} + +/// Receives the records the driver read off the channel. +pub(crate) struct Reader { + rx: mpsc::UnboundedReceiver>, + unread: Arc, +} + +impl qmux::transport::Reader for Reader { + async fn recv(&mut self) -> Result { + let record = self.rx.recv().await.unwrap_or(Err(qmux::Error::Closed))?; + self.unread.fetch_sub(record.len() as u64, Ordering::Relaxed); + Ok(record) + } +} + +#[cfg(test)] +mod tests { + use qmux::transport::Reader as _; + + use super::*; + + #[tokio::test] + async fn queue_closes_past_its_budget() { + let (inbox, mut reader) = queue(10); + assert!(inbox.push(Ok(Bytes::from_static(&[0; 6])))); + assert!(inbox.push(Ok(Bytes::from_static(&[1; 4])))); + // Reading frees budget for the next record. + assert_eq!(reader.recv().await.unwrap().len(), 6); + assert!(inbox.push(Ok(Bytes::from_static(&[2; 6])))); + assert!(!inbox.push(Ok(Bytes::from_static(&[3; 1])))); + + // Records queued before the overflow still arrive, then the error. + assert_eq!(reader.recv().await.unwrap()[0], 1); + assert_eq!(reader.recv().await.unwrap()[0], 2); + assert!(matches!(reader.recv().await, Err(qmux::Error::FlowControlError))); + } + + #[tokio::test] + async fn queue_ends_on_empty_or_error_records() { + // Neither costs any bytes, so the budget alone would never stop a flood of them. + let (inbox, mut reader) = queue(10); + assert!(!inbox.push(Ok(Bytes::new()))); + assert!(matches!(reader.recv().await, Err(qmux::Error::ProtocolViolation))); + + let (inbox, mut reader) = queue(10); + assert!(!inbox.push(Err(qmux::Error::FrameTooLarge))); + assert!(matches!(reader.recv().await, Err(qmux::Error::FrameTooLarge))); + } +} diff --git a/rs/moq-tokio/src/webrtc/tests.rs b/rs/moq-tokio/src/webrtc/tests.rs new file mode 100644 index 00000000000..b3afab026bb --- /dev/null +++ b/rs/moq-tokio/src/webrtc/tests.rs @@ -0,0 +1,364 @@ +use std::net::{IpAddr, Ipv4Addr}; +use std::time::Duration; + +use super::*; + +/// Real sockets and real ICE, so these run on the wall clock. +const TIMEOUT: Duration = Duration::from_secs(10); + +const LOOPBACK: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST); + +/// A protocol both the default client and server speak. +fn protocol() -> &'static str { + crate::connect::Config::default().versions().alpns()[0] +} + +/// A one-frame broadcast named `test` and the handles that keep it alive. +fn broadcast(payload: &[u8]) -> (moq_net::origin::Producer, impl Sized) { + let origin = crate::origin::spawn(); + let broadcast = origin.create_broadcast("test").unwrap(); + broadcast.announce(Default::default()).unwrap(); + let track = broadcast + .create_track( + "video", + moq_net::track::Info::default().with_timescale(moq_net::Timescale::MILLI), + ) + .unwrap(); + let mut group = track.append_group().unwrap(); + group.write_frame(moq_net::Timestamp::ZERO, payload.to_vec()).unwrap(); + group.finish().unwrap(); + (origin, (broadcast, track)) +} + +/// Serve a one-frame broadcast from the answering peer, subscribe from the +/// offering one, and read the frame back across the data channel. +async fn transfer(offerer: &Offerer, answerer: &Answerer, payload: &[u8]) { + let (origin, _keep) = broadcast(payload); + let listener = crate::listen::Config::default() + .init_streams() + .unwrap() + .listen() + .await + .unwrap(); + let server = async { + let request = listener.accept_webrtc(answerer).await.unwrap(); + assert_eq!(request.transport(), crate::Transport::WebRtc); + request.with_publisher(&origin).ok().await.unwrap() + }; + + let subscribed = crate::origin::spawn(); + let consumer = subscribed.consume(); + let mut announced = consumer.announced(); + let client = crate::client::Config::default() + .init() + .unwrap() + .with_subscriber(subscribed); + let ((_server, _driver), _client) = tokio::time::timeout(TIMEOUT, async { + tokio::join!(server, async { client.connect_webrtc(offerer).await.unwrap() }) + }) + .await + .expect("handshake timed out"); + + let read = async { + let Some(moq_net::announce::Event::Start(route)) = announced.next().await else { + panic!("expected the broadcast to be announced"); + }; + assert_eq!(route.prefix.as_str(), "test"); + let broadcast = consumer.request_broadcast("test", None).await.unwrap(); + let mut track = broadcast.track("video").unwrap().subscribe(None).await.unwrap(); + let mut group = track.recv_group().await.unwrap().unwrap(); + group.read_frame().await.unwrap().unwrap() + }; + let frame = tokio::time::timeout(TIMEOUT, read).await.expect("transfer timed out"); + assert!(frame.payload[..] == *payload, "the frame arrived corrupted"); +} + +#[tokio::test] +async fn session_over_loopback() { + let config = Config::default(); + let (offerer, offer) = Offerer::with_hosts(&config, protocol(), vec![LOOPBACK]).await.unwrap(); + let (answerer, answer) = Answerer::with_hosts(&config, &offer, protocol(), vec![LOOPBACK]) + .await + .unwrap(); + offerer.accept(&answer).await.unwrap(); + + transfer(&offerer, &answerer, b"hello").await; +} + +/// A frame many records long fills the SCTP send window, so the writer has to +/// wait for the peer's acknowledgements and resume. +#[tokio::test] +async fn large_frame_over_loopback() { + let config = Config::default(); + let (offerer, offer) = Offerer::with_hosts(&config, protocol(), vec![LOOPBACK]).await.unwrap(); + let (answerer, answer) = Answerer::with_hosts(&config, &offer, protocol(), vec![LOOPBACK]) + .await + .unwrap(); + offerer.accept(&answer).await.unwrap(); + + let payload: Vec = (0..1024 * 1024).map(|i| i as u8).collect(); + transfer(&offerer, &answerer, &payload).await; +} + +/// Open the data channel between two loopback peers, without a MoQ session over it. +async fn channels() -> (Offerer, Answerer, Opened, Opened) { + let config = Config::default(); + let (offerer, offer) = Offerer::with_hosts(&config, protocol(), vec![LOOPBACK]).await.unwrap(); + let (answerer, answer) = Answerer::with_hosts(&config, &offer, protocol(), vec![LOOPBACK]) + .await + .unwrap(); + offerer.accept(&answer).await.unwrap(); + let (sent, received) = tokio::time::timeout(TIMEOUT, async { tokio::join!(offerer.channel(), answerer.channel()) }) + .await + .expect("the channel did not open"); + (offerer, answerer, sent.unwrap(), received.unwrap()) +} + +/// A full record, numbered so its order can be checked. +fn record(index: usize) -> Bytes { + Bytes::from(vec![index as u8; RECORD_SIZE as usize]) +} + +/// A session that is not reading must not stop the peer connection: the agent +/// keeps acknowledging, so the remote writer gets well past the send window. +#[tokio::test] +async fn unread_records_do_not_stall_the_writer() { + use qmux::transport::{Reader as _, Writer as _}; + + let (_offerer, _answerer, sent, received) = channels().await; + let (mut writer, mut reader) = (sent.channel.writer, received.channel.reader); + let count = 8 * SEND_WINDOW / RECORD_SIZE as usize; + tokio::time::timeout(TIMEOUT, async { + for index in 0..count { + writer.send(record(index)).await.unwrap(); + } + }) + .await + .expect("the writer stalled while the reader was idle"); + + for index in 0..count { + assert_eq!(reader.recv().await.unwrap(), record(index)); + } +} + +/// A peer that sends past qmux's receive window while the session is not reading +/// has the channel closed, after the records within the budget. +#[tokio::test] +async fn ignoring_flow_control_closes_the_channel() { + use qmux::transport::{Reader as _, Writer as _}; + + let (_offerer, _answerer, sent, received) = channels().await; + let budget = received.config.max_data + READ_SLACK; + let within = (budget / RECORD_SIZE) as usize; + let (mut writer, mut reader) = (sent.channel.writer, received.channel.reader); + tokio::time::timeout(TIMEOUT, async { + for index in 0..within + 8 { + if writer.send(record(index)).await.is_err() { + break; + } + } + }) + .await + .expect("the writer stalled"); + + for index in 0..within { + assert_eq!(reader.recv().await.unwrap(), record(index)); + } + assert!(matches!(reader.recv().await, Err(qmux::Error::FlowControlError))); +} + +/// Answer STUN Binding requests with the source address, as a public server would. +async fn stun_server() -> SocketAddr { + use stun_types::attribute::XorMappedAddress; + use stun_types::message::{Message, MessageWriteVec}; + use stun_types::prelude::*; + + let socket = UdpSocket::bind((LOOPBACK, 0)).await.unwrap(); + let addr = socket.local_addr().unwrap(); + tokio::spawn(async move { + let mut buf = [0; 1500]; + while let Ok((len, src)) = socket.recv_from(&mut buf).await { + let request = Message::from_bytes(&buf[..len]).unwrap(); + let mut reply = Message::builder_success(&request, MessageWriteVec::new()); + reply + .add_attribute(&XorMappedAddress::new(src, request.transaction_id())) + .unwrap(); + socket.send_to(&reply.finish(), src).await.unwrap(); + } + }); + addr +} + +/// Drop the candidates a description carries, so only trickled ones reach the peer. +fn strip_candidates(sdp: &str) -> String { + sdp.split_inclusive('\n') + .filter(|line| !line.starts_with("a=candidate:")) + .collect() +} + +/// The offerer sits behind a pretend NAT: it advertises 127.0.0.2, which the +/// answerer never learns, and reaches the world as 127.0.0.1. With no candidates +/// in either description and none trickled back, the answerer can only reach the +/// offerer through the server-reflexive candidate STUN found. +#[tokio::test] +async fn server_reflexive_candidate_is_offered_and_selected() { + let stun = stun_server().await; + let config = Config { + stun: vec![format!("stun:{stun}").parse().unwrap()], + }; + let private = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2)); + let (mut offerer, offer) = Offerer::with_hosts(&config, protocol(), vec![private]).await.unwrap(); + let offer = Offer::new(strip_candidates(offer.as_str())); + let (answerer, answer) = Answerer::with_hosts(&Config::default(), &offer, protocol(), vec![LOOPBACK]) + .await + .unwrap(); + offerer + .accept(&Answer::new(strip_candidates(answer.as_str()))) + .await + .unwrap(); + + let candidate = tokio::time::timeout(TIMEOUT, offerer.candidate()) + .await + .expect("gathering timed out") + .expect("no server-reflexive candidate"); + let parsed = str0m::Candidate::from_sdp_string(candidate.as_str()).unwrap(); + assert_eq!(parsed.kind(), str0m::CandidateKind::ServerReflexive); + assert_eq!(parsed.addr().ip(), LOOPBACK); + answerer.add_candidate(&candidate).await.unwrap(); + assert_eq!( + offerer.candidate().await, + None, + "gathering should end after the only server answers" + ); + + transfer(&offerer, &answerer, b"hello").await; +} + +#[tokio::test] +async fn unsupported_protocol_is_refused() { + let (offerer, _) = Offerer::with_hosts(&Config::default(), "not-moq", vec![LOOPBACK]) + .await + .unwrap(); + let client = crate::client::Config::default().init().unwrap(); + assert!(matches!( + client.connect_webrtc(&offerer).await, + Err(crate::Error::WebRtc(err)) if matches!(*err, Error::Protocol(_)) + )); +} + +/// A channel carrying another supported ALPN than the one signaling agreed is refused. +#[tokio::test] +async fn unsignaled_protocol_is_refused() { + let versions = crate::connect::Config::default().versions(); + let [agreed, other, ..] = versions.alpns()[..] else { + panic!("expected two supported ALPNs"); + }; + let config = Config::default(); + let (offerer, offer) = Offerer::with_hosts(&config, other, vec![LOOPBACK]).await.unwrap(); + let (answerer, answer) = Answerer::with_hosts(&config, &offer, agreed, vec![LOOPBACK]) + .await + .unwrap(); + offerer.accept(&answer).await.unwrap(); + + let listener = crate::listen::Config::default() + .init_streams() + .unwrap() + .listen() + .await + .unwrap(); + let accepted = tokio::time::timeout(TIMEOUT, listener.accept_webrtc(&answerer)) + .await + .expect("the channel did not open"); + assert!(matches!( + accepted, + Err(crate::Error::WebRtc(err)) if matches!(&*err, Error::Protocol(protocol) if protocol == other) + )); +} + +#[test] +fn stun_urls() { + let parse = |url: &str| url.parse::().map(|stun| stun.to_string()).ok(); + assert_eq!( + parse("stun:stun.example.com").as_deref(), + Some("stun:stun.example.com:3478") + ); + assert_eq!(parse("STUN:1.2.3.4:19302").as_deref(), Some("stun:1.2.3.4:19302")); + assert_eq!(parse("stun:[::1]").as_deref(), Some("stun:[::1]:3478")); + assert_eq!(parse("stun:[::1]:5349").as_deref(), Some("stun:[::1]:5349")); + for bad in [ + "turn:turn.example.com", + "stuns:stun.example.com", + "stun:", + "stun:host:port", + "stun:host:3478?transport=udp", + "stun:[::1]x", + "stun:[nope]", + "stun:server.example#fragment", + "stun:server example", + "stun:server\texample", + "stun.example.com", + ] { + assert_eq!(parse(bad), None, "{bad}"); + } +} + +#[test] +fn max_message_sizes() { + assert_eq!(max_message_size("v=0\r\n").unwrap(), DEFAULT_MAX_MESSAGE_SIZE); + assert_eq!(max_message_size("a=max-message-size:262144\r\n").unwrap(), 262_144); + assert_eq!(max_message_size("a=max-message-size:0\r\n").unwrap(), u64::MAX); + assert!(max_message_size("a=max-message-size:big\r\n").is_err()); +} + +#[test] +fn mdns_candidates_are_split_out() { + let sdp = "v=0\r\n\ + a=candidate:1 1 udp 2113937151 0a1b2c3d.local 54321 typ host\r\n\ + a=candidate:2 1 udp 2113937151 192.168.1.2 54321 typ host\r\n\ + a=end-of-candidates\r\n"; + let (kept, mdns) = split_mdns(sdp); + assert_eq!( + kept, + "v=0\r\na=candidate:2 1 udp 2113937151 192.168.1.2 54321 typ host\r\na=end-of-candidates\r\n" + ); + assert_eq!(mdns, ["candidate:1 1 udp 2113937151 0a1b2c3d.local 54321 typ host"]); +} + +#[tokio::test] +async fn ip_candidates_parse_without_mdns() { + let candidate = parse_candidate("candidate:2 1 udp 2113937151 192.168.1.2 54321 typ host") + .await + .unwrap(); + assert_eq!(candidate.addr(), "192.168.1.2:54321".parse().unwrap()); + assert!(parse_candidate("candidate:2 1 udp nope").await.is_err()); +} + +/// Building a peer mutates its `Rtc` several times before the agent's first drain: +/// a host candidate per address, then the channel or the answer. str0m wants a +/// drain after each, which is a no-op while none of them queues output; pin that, +/// so a str0m upgrade that changes it fails here. +#[test] +fn construction_queues_no_output() { + fn idle(rtc: &mut Rtc) { + assert!(matches!(rtc.poll_output(), Ok(str0m::Output::Timeout(_)))); + } + fn host(rtc: &mut Rtc, addr: &str) { + let candidate = str0m::Candidate::host(addr.parse().unwrap(), "udp").unwrap(); + rtc.add_local_candidate(candidate); + idle(rtc); + } + + let mut offerer = Rtc::builder().build(Instant::now()); + host(&mut offerer, "127.0.0.1:5000"); + host(&mut offerer, "[::1]:5000"); + let mut api = offerer.sdp_api(); + api.add_channel(LABEL.to_string()); + let (offer, _pending) = api.apply().unwrap(); + idle(&mut offerer); + + let mut answerer = Rtc::builder().build(Instant::now()); + host(&mut answerer, "127.0.0.1:5001"); + host(&mut answerer, "[::1]:5001"); + answerer.sdp_api().accept_offer(offer).unwrap(); + idle(&mut answerer); +} diff --git a/sh/rs/tokio-features.sh b/sh/rs/tokio-features.sh index 555ef55272a..9bbb5cc363e 100755 --- a/sh/rs/tokio-features.sh +++ b/sh/rs/tokio-features.sh @@ -22,7 +22,7 @@ if ((all)); then cargo clippy --locked -p moq-tokio --no-default-features --features quic,aws-lc-rs -- -D warnings cargo clippy --locked -p moq-tokio --no-default-features --features quic,ring -- -D warnings - for feature in aws-lc-rs ring tcp uds websocket; do + for feature in aws-lc-rs ring tcp uds websocket webrtc; do cargo clippy --locked -p moq-tokio --no-default-features --features "$feature" -- -D warnings done fi