sipx_media/ice/
gather.rs

1//! Gathering local candidates, and the description they make (RFC 8445 §5.1.1, RFC 8839 §5.1,
2//! §13.2; [spec] §5, §15).
3//!
4//! Two kinds of candidate, and both come from sockets that are already bound. A **host**
5//! candidate is what [`MediaPort`](crate::MediaPort) got from the OS. A **server-reflexive** one
6//! is what a STUN server says it sees, obtained over
7//! [`sipx_transport::stun`](https://docs.rs/sipx-transport) — the Binding client that already
8//! exists, because [spec] §15 says a second one would be a second thing to get wrong.
9//!
10//! The candidates are *priced* by the agent and not here. §5.1.1.3's foundation and §5.1.2.1's
11//! local preference are properties of the whole gathered set — "MUST be unique for each" candidate
12//! of a type — and are not facts any single candidate knows about itself, so what this module
13//! produces is a stream of [`Gathered`] and what it reads back is
14//! [`Agent::local_candidates`](super::Agent::local_candidates).
15//!
16//! [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
17
18use std::net::SocketAddr;
19use std::sync::Arc;
20use std::sync::atomic::Ordering;
21use std::time::Duration;
22
23use tokio::net::UdpSocket;
24
25use sipx_sdp::ice::{
26    Candidate, CandidateType, ComponentId, Credentials, Foundation, RelatedAddress, Transport,
27};
28
29use super::agent::{Agent, Config, Input, Output};
30use super::candidate::{Gathered, LocalBase, LocalCandidate};
31use super::negotiate::Negotiation;
32use crate::counters::DiscardMeters;
33
34/// How long to wait for one Binding Response before trying again (RFC 5389 §7.2.1's initial RTO).
35const STUN_RTO: Duration = Duration::from_millis(500);
36
37/// Everything gathering needs that the sockets do not already supply.
38#[derive(Debug, Clone)]
39pub struct Gathering {
40    /// Our short-term credentials for this ICE session (RFC 8839 §5.4), which go in the offer or
41    /// the answer and key every check in both directions ([spec] §11.2).
42    ///
43    /// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
44    pub credentials: Credentials,
45    /// Whether sipx sent the initial offer — RFC 8445 §6.1.1's role determination between two
46    /// full agents.
47    pub offerer: bool,
48    /// §7.1.3's 64-bit tiebreaker, drawn once per ICE session.
49    pub tiebreaker: u64,
50    /// The STUN server to ask for a server-reflexive candidate (§5.1.1.2). `None` gathers host
51    /// candidates alone, which is the right answer on a network with no NAT and the only answer
52    /// when no server is configured.
53    pub stun_server: Option<SocketAddr>,
54    /// How long to keep asking it before giving up.
55    ///
56    /// Gathering that never finishes is an offer that never goes out, so this is a deadline and
57    /// not a retry count: whatever has been gathered when it expires is what is offered, and a
58    /// STUN server that is down costs one call setup this long and nothing else.
59    pub stun_timeout: Duration,
60    /// The agent's own configuration — §14's timers and §6.1.2.5's pair limit.
61    pub agent: Config,
62}
63
64impl Gathering {
65    /// Gather host candidates only, with a fresh tiebreaker.
66    ///
67    /// The tiebreaker is drawn here rather than taken because §7.1.3 wants a random one per ICE
68    /// session and a caller that has no opinion should not have to have one; a caller that does —
69    /// a test walking §7.3.1.1's `T = V` row — sets the field afterwards.
70    #[must_use]
71    pub fn new(credentials: Credentials, offerer: bool) -> Self {
72        Self {
73            credentials,
74            offerer,
75            tiebreaker: rand::random(),
76            stun_server: None,
77            stun_timeout: Duration::from_secs(2),
78            agent: Config::default(),
79        }
80    }
81}
82
83/// One bound socket, and which component and base it is.
84#[derive(Debug, Clone, Copy)]
85pub(crate) struct Base<'a> {
86    /// The index the agent will name this socket by ([spec] §2).
87    ///
88    /// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
89    pub index: LocalBase,
90    /// Which component of the stream it carries.
91    pub component: ComponentId,
92    /// The socket itself, still exclusively ours: gathering runs before any receive loop does.
93    pub socket: &'a UdpSocket,
94}
95
96/// What sipx will put in its own description, and the agent that will drive it.
97///
98/// Held by the caller across offer/answer: it is built when the port is bound and the offer is
99/// written, and it is consumed when the session starts. The agent inside it has already been told
100/// what was gathered, so the only thing still missing is the peer's half.
101#[derive(Debug)]
102pub struct LocalDescription {
103    agent: Agent,
104    /// Outputs the agent produced before there was a driver to perform them.
105    ///
106    /// There are none today — forming checklists arms Ta and sends nothing — but they are carried
107    /// rather than dropped, because "the agent emitted something and nobody did it" is the one
108    /// failure this type could introduce silently.
109    pending: Vec<Output>,
110    credentials: Credentials,
111    candidates: Vec<Candidate>,
112    defaults: Vec<(ComponentId, SocketAddr)>,
113}
114
115impl LocalDescription {
116    /// Our credentials, for `a=ice-ufrag` and `a=ice-pwd`.
117    #[must_use]
118    pub const fn credentials(&self) -> &Credentials {
119        &self.credentials
120    }
121
122    /// The gathered candidates, priced, in descending priority.
123    #[must_use]
124    pub fn candidates(&self) -> &[Candidate] {
125        &self.candidates
126    }
127
128    /// Where the peer should send this component if it turns out not to do ICE — the `c=`/`m=`
129    /// default destination (RFC 8839 §4.2.1; [spec] §13.2).
130    ///
131    /// "The candidate sipx would use if the peer turned out not to do ICE", which for a full agent
132    /// is the highest-priority one for the component.
133    ///
134    /// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
135    #[must_use]
136    pub fn default_destination(&self, component: ComponentId) -> Option<SocketAddr> {
137        self.defaults
138            .iter()
139            .find(|(id, _)| *id == component)
140            .map(|(_, address)| *address)
141    }
142
143    /// The media-level attributes for the offer or the answer (RFC 8839 §4.2.1, §4.2.2).
144    ///
145    /// `a=ice-options:ice2` goes out on both: [spec] §8 makes aggressive nomination unavailable
146    /// rather than optional, and `ice2` is how a peer is told that no pair will be re-nominated
147    /// mid-session.
148    ///
149    /// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
150    #[must_use]
151    pub fn attributes(&self) -> Vec<sipx_sdp::Attribute> {
152        let mut attributes = vec![
153            sipx_sdp::Attribute::valued("ice-ufrag", self.credentials.ufrag()),
154            sipx_sdp::Attribute::valued("ice-pwd", self.credentials.pwd()),
155            sipx_sdp::Attribute::valued("ice-options", sipx_sdp::ice::ICE2),
156        ];
157        attributes.extend(
158            self.candidates
159                .iter()
160                .map(|candidate| sipx_sdp::Attribute::valued("candidate", candidate.to_value())),
161        );
162        attributes
163    }
164
165    /// Feed the agent the peer's half of the exchange.
166    ///
167    /// Returns whether ICE will actually be driven for this stream. `false` for
168    /// [`Negotiation::Absent`] and [`Negotiation::Mismatch`] alike — RFC 8839 §5.3 says ICE MUST
169    /// NOT be used for a mismatched stream, so the two differ in what the *answer* says and not in
170    /// what the media port does.
171    pub fn accept(&mut self, negotiation: &Negotiation) -> bool {
172        let Negotiation::Ice {
173            credentials,
174            candidates,
175            lite,
176        } = negotiation
177        else {
178            return false;
179        };
180        self.pending
181            .extend(self.agent.handle(Input::RemoteDescription {
182                credentials: credentials.clone(),
183                candidates: candidates.clone(),
184                lite: *lite,
185            }));
186        true
187    }
188
189    /// Whether [`Self::accept`] has been given a peer description ICE can run against.
190    #[must_use]
191    pub(crate) fn running(&self) -> bool {
192        !self.agent.remote_candidates().is_empty()
193    }
194
195    /// Take the agent and whatever it has already asked for, for the driver to run.
196    pub(crate) fn into_driver_parts(self) -> (Agent, Vec<Output>) {
197        (self.agent, self.pending)
198    }
199}
200
201/// Gather over these sockets and build the description they make (§5.1.1).
202///
203/// Every socket contributes a host candidate; the first one contributes a server-reflexive
204/// candidate too when a STUN server is configured and answers. `GatheringDone` is fed at the end,
205/// which is what lets the agent form checklists as soon as the peer's half arrives.
206pub(crate) async fn gather(
207    bases: &[Base<'_>],
208    config: &Gathering,
209    discards: Arc<DiscardMeters>,
210) -> LocalDescription {
211    let mut agent = Agent::new(
212        config.agent,
213        config.offerer,
214        config.credentials.clone(),
215        config.tiebreaker,
216    );
217    let mut pending = Vec::new();
218
219    for base in bases {
220        let Ok(address) = base.socket.local_addr() else {
221            continue;
222        };
223        if address.ip().is_unspecified() {
224            // A wildcard bind has no host candidate: `0.0.0.0` is not somewhere a peer can send,
225            // and offering it would advertise a path that cannot work while hiding that no usable
226            // one was found. Enumerating the interfaces behind a wildcard bind is what a
227            // gathering agent would do instead, and it is not something this crate can do without
228            // a platform dependency it does not have.
229            tracing::debug!(%address, "no host candidate for a wildcard bind");
230            continue;
231        }
232        pending.extend(agent.handle(Input::LocalCandidate(Gathered {
233            base: base.index,
234            base_address: address,
235            address,
236            kind: CandidateType::Host,
237            component: base.component,
238            server: None,
239        })));
240
241        let Some(server) = config.stun_server else {
242            continue;
243        };
244        let Some(mapped) = reflexive(base.socket, server, config.stun_timeout, &discards).await
245        else {
246            continue;
247        };
248        if mapped == address {
249            // §5.1.3: a server-reflexive candidate whose address is one of our host candidates is
250            // redundant and is discarded. On a network with no NAT that is every one of them.
251            discards
252                .ice_redundant_candidates
253                .fetch_add(1, Ordering::Relaxed);
254            tracing::debug!(%address, "no nat: the reflexive candidate is the host one");
255            continue;
256        }
257        pending.extend(agent.handle(Input::LocalCandidate(Gathered {
258            base: base.index,
259            base_address: address,
260            address: mapped,
261            kind: CandidateType::ServerReflexive,
262            component: base.component,
263            server: Some(server.ip()),
264        })));
265    }
266
267    pending.extend(agent.handle(Input::GatheringDone));
268
269    let candidates = lines(agent.local_candidates());
270    let defaults = defaults(&candidates);
271    LocalDescription {
272        agent,
273        pending,
274        credentials: config.credentials.clone(),
275        candidates,
276        defaults,
277    }
278}
279
280/// Ask a STUN server what address it sees, over [`sipx_transport::stun`] (§5.1.1.2).
281///
282/// The socket is the one the candidate is for, because that is what makes the answer a candidate:
283/// the mapping a NAT holds is per source address and port, so a reflexive address learned on any
284/// other socket describes a path media will never take.
285///
286/// Retransmission is RFC 5389 §7.2.1's, truncated at the deadline rather than at Rc: gathering is
287/// on the call-setup path, and an offer that waits out the full ladder for a server that is down
288/// is an offer nobody sends.
289async fn reflexive(
290    socket: &UdpSocket,
291    server: SocketAddr,
292    within: Duration,
293    discards: &DiscardMeters,
294) -> Option<SocketAddr> {
295    let id = sipx_transport::stun::new_transaction_id();
296    let request = sipx_transport::stun::binding_request(&id);
297    let deadline = tokio::time::Instant::now().checked_add(within)?;
298    let mut rto = STUN_RTO;
299    let mut datagram = vec![0u8; 1500];
300
301    while tokio::time::Instant::now() < deadline {
302        if socket.send_to(&request, server).await.is_err() {
303            return None;
304        }
305        let wait = deadline
306            .saturating_duration_since(tokio::time::Instant::now())
307            .min(rto);
308        let until = tokio::time::Instant::now().checked_add(wait)?;
309        loop {
310            let read = tokio::time::timeout_at(until, socket.recv_from(&mut datagram)).await;
311            let Ok(Ok((len, from))) = read else {
312                break;
313            };
314            if from != server {
315                // Something else on the media port. It is not this transaction's business, and
316                // gathering runs before there is anywhere to hand it, so it is dropped.
317                discards
318                    .ice_gathering_foreign_datagrams
319                    .fetch_add(1, Ordering::Relaxed);
320                tracing::debug!(%from, %server, "dropping a datagram from outside the STUN gathering transaction");
321                continue;
322            }
323            let Some(reply) = sipx_transport::stun::parse_reply(datagram.get(..len)?) else {
324                continue;
325            };
326            if reply.id() != id {
327                continue;
328            }
329            return match reply {
330                sipx_transport::stun::Reply::Bound { mapped, .. } => mapped,
331                // §4.4.2 of RFC 5626 calls an error response a failed flow; here it is simply no
332                // reflexive candidate, and the host ones still stand.
333                sipx_transport::stun::Reply::Failed { .. } => None,
334            };
335        }
336        rto = rto.saturating_mul(2);
337    }
338    None
339}
340
341/// Turn the agent's priced candidates into `a=candidate` lines (RFC 8839 §5.1).
342///
343/// Shared with the driver, which signals the same list again for every later exchange on the call
344/// ([spec] §13.5) — one ordering rule and one set of `raddr`/`rport` decisions, so a re-offer
345/// cannot describe the same sockets differently from the offer that opened the session.
346///
347/// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
348pub(crate) fn lines(candidates: &[LocalCandidate]) -> Vec<Candidate> {
349    let mut lines: Vec<Candidate> = candidates
350        .iter()
351        .filter_map(|candidate| {
352            Some(Candidate {
353                foundation: Foundation::new(&candidate.foundation.0.to_string())?,
354                component: candidate.gathered.component,
355                transport: Transport::Udp,
356                priority: candidate.priority,
357                address: candidate.gathered.address.ip(),
358                port: candidate.gathered.address.port(),
359                kind: candidate.gathered.kind,
360                // §5.1: `raddr`/`rport` MUST be present for a reflexive candidate and MUST be
361                // absent for a host one. The base is what they name.
362                related: related(candidate),
363                extensions: Vec::new(),
364            })
365        })
366        .collect();
367    // Descending priority, so the first line for a component is also the default destination and
368    // a reader of the offer sees them in the order ICE will reason about them.
369    lines.sort_by(|left, right| {
370        right
371            .priority
372            .get()
373            .cmp(&left.priority.get())
374            .then_with(|| left.component.get().cmp(&right.component.get()))
375    });
376    lines
377}
378
379/// The `raddr`/`rport` a candidate of this type carries (RFC 8839 §5.1).
380fn related(candidate: &LocalCandidate) -> Option<RelatedAddress> {
381    match candidate.gathered.kind {
382        CandidateType::Host => None,
383        CandidateType::ServerReflexive | CandidateType::PeerReflexive | CandidateType::Relayed => {
384            Some(RelatedAddress {
385                address: candidate.gathered.base_address.ip(),
386                port: candidate.gathered.base_address.port(),
387            })
388        }
389    }
390}
391
392/// The highest-priority candidate for each component ([spec] §13.2).
393///
394/// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
395fn defaults(candidates: &[Candidate]) -> Vec<(ComponentId, SocketAddr)> {
396    let mut defaults: Vec<(ComponentId, SocketAddr)> = Vec::new();
397    for candidate in candidates {
398        if defaults.iter().any(|(id, _)| *id == candidate.component) {
399            continue;
400        }
401        defaults.push((
402            candidate.component,
403            SocketAddr::new(candidate.address, candidate.port),
404        ));
405    }
406    defaults
407}
408
409#[cfg(test)]
410#[allow(
411    clippy::unwrap_used,
412    clippy::expect_used,
413    clippy::panic,
414    clippy::indexing_slicing
415)]
416mod tests {
417    use super::*;
418
419    fn credentials() -> Credentials {
420        Credentials::new("8hhY", "asd88fgpdd777uzjYhagZg").expect("valid")
421    }
422
423    async fn bound() -> UdpSocket {
424        UdpSocket::bind("127.0.0.1:0".parse::<SocketAddr>().unwrap())
425            .await
426            .expect("a loopback port")
427    }
428
429    /// The two components come off the two sockets [`MediaPort`](crate::MediaPort) binds, and the
430    /// priorities are §4's table: host, single address, RTP and RTCP.
431    #[tokio::test]
432    async fn host_candidates_come_off_the_bound_sockets() {
433        let (media, control) = (bound().await, bound().await);
434        let description = gather(
435            &[
436                Base {
437                    index: LocalBase(0),
438                    component: ComponentId::RTP,
439                    socket: &media,
440                },
441                Base {
442                    index: LocalBase(1),
443                    component: ComponentId::RTCP,
444                    socket: &control,
445                },
446            ],
447            &Gathering::new(credentials(), true),
448            Arc::new(DiscardMeters::default()),
449        )
450        .await;
451
452        assert_eq!(description.candidates().len(), 2);
453        let first = &description.candidates()[0];
454        assert_eq!(first.component, ComponentId::RTP);
455        assert_eq!(first.kind, CandidateType::Host);
456        assert_eq!(first.priority.get(), 2_130_706_431);
457        assert_eq!(first.related, None, "a host candidate carries no raddr");
458        assert_eq!(description.candidates()[1].priority.get(), 2_130_706_430);
459
460        assert_eq!(
461            description.default_destination(ComponentId::RTP),
462            Some(media.local_addr().unwrap())
463        );
464        assert_eq!(
465            description.default_destination(ComponentId::RTCP),
466            Some(control.local_addr().unwrap())
467        );
468    }
469
470    /// [spec] §6.1: sipx offers component 2 only when the control port was actually obtained.
471    /// A driver that offered it anyway would have the peer checking an address nothing is bound
472    /// to, and RFC 8445 §6.1.2.2's reduction to the minimum component count is exactly what a
473    /// peer does with the offer that does not.
474    ///
475    /// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
476    #[tokio::test]
477    async fn no_control_port_means_no_second_component() {
478        let rtp = bound().await;
479        let description = gather(
480            &[Base {
481                index: LocalBase(0),
482                component: ComponentId::RTP,
483                socket: &rtp,
484            }],
485            &Gathering::new(credentials(), true),
486            Arc::new(DiscardMeters::default()),
487        )
488        .await;
489
490        assert_eq!(description.candidates().len(), 1);
491        assert_eq!(description.candidates()[0].component, ComponentId::RTP);
492        assert_eq!(description.default_destination(ComponentId::RTCP), None);
493    }
494
495    /// A wildcard bind is not a candidate: `0.0.0.0` is nowhere to send to, and advertising it
496    /// would hide that nothing usable was gathered behind a line that looks like one.
497    #[tokio::test]
498    async fn a_wildcard_bind_yields_no_host_candidate() {
499        let any = UdpSocket::bind("0.0.0.0:0".parse::<SocketAddr>().unwrap())
500            .await
501            .expect("bound");
502        let description = gather(
503            &[Base {
504                index: LocalBase(0),
505                component: ComponentId::RTP,
506                socket: &any,
507            }],
508            &Gathering::new(credentials(), true),
509            Arc::new(DiscardMeters::default()),
510        )
511        .await;
512        assert!(description.candidates().is_empty());
513    }
514
515    /// §5.1.1.2, over the Binding client [spec] §15 says to reuse: the address the server reports
516    /// becomes a candidate whose base is the socket it was learned on, with `raddr`/`rport`
517    /// naming that base as §5.1 requires.
518    ///
519    /// [spec]: https://github.com/codewandler/sipx/blob/main/docs/specs/ice.md
520    #[tokio::test]
521    async fn a_server_reflexive_candidate_comes_from_the_stun_server() {
522        let server = bound().await;
523        let server_address = server.local_addr().unwrap();
524        let reported: SocketAddr = "198.51.100.7:31337".parse().unwrap();
525        tokio::spawn(async move {
526            let mut datagram = vec![0u8; 1500];
527            let Ok((_len, from)) = server.recv_from(&mut datagram).await else {
528                return;
529            };
530            let id: [u8; 12] = datagram[8..20].try_into().expect("a header");
531            let _ = server.send_to(&binding_response(&id, reported), from).await;
532        });
533
534        let rtp = bound().await;
535        let mut gathering = Gathering::new(credentials(), true);
536        gathering.stun_server = Some(server_address);
537        let description = gather(
538            &[Base {
539                index: LocalBase(0),
540                component: ComponentId::RTP,
541                socket: &rtp,
542            }],
543            &gathering,
544            Arc::new(DiscardMeters::default()),
545        )
546        .await;
547
548        let reflexive = description
549            .candidates()
550            .iter()
551            .find(|candidate| candidate.kind == CandidateType::ServerReflexive)
552            .expect("the server's answer became a candidate");
553        assert_eq!(reflexive.address, reported.ip());
554        assert_eq!(reflexive.port, reported.port());
555        assert_eq!(reflexive.priority.get(), 1_694_498_815);
556        let related = reflexive.related.as_ref().expect("srflx carries raddr");
557        assert_eq!(related.address, rtp.local_addr().unwrap().ip());
558        assert_eq!(related.port, rtp.local_addr().unwrap().port());
559
560        // §13.2's default destination is the highest-priority candidate, which is the host one.
561        assert_eq!(
562            description.default_destination(ComponentId::RTP),
563            Some(rtp.local_addr().unwrap())
564        );
565    }
566
567    /// A STUN server that never answers costs the deadline and nothing else: the host candidates
568    /// are still offered, because an offer that waits for a server that is down is an offer
569    /// nobody sends.
570    #[tokio::test]
571    async fn a_silent_stun_server_still_yields_the_host_candidates() {
572        // Bound and never read, so the request arrives and no answer ever comes back.
573        let black_hole = bound().await;
574        let rtp = bound().await;
575        let mut gathering = Gathering::new(credentials(), true);
576        gathering.stun_server = Some(black_hole.local_addr().unwrap());
577        gathering.stun_timeout = Duration::from_millis(120);
578
579        let description = gather(
580            &[Base {
581                index: LocalBase(0),
582                component: ComponentId::RTP,
583                socket: &rtp,
584            }],
585            &gathering,
586            Arc::new(DiscardMeters::default()),
587        )
588        .await;
589        assert_eq!(description.candidates().len(), 1);
590        assert_eq!(description.candidates()[0].kind, CandidateType::Host);
591    }
592
593    /// RFC 5389 §15.2's `XOR-MAPPED-ADDRESS` in a Binding Response, for the fake server above.
594    fn binding_response(id: &[u8; 12], mapped: SocketAddr) -> Vec<u8> {
595        let SocketAddr::V4(v4) = mapped else {
596            panic!("the fixture is IPv4");
597        };
598        let cookie = sipx_transport::stun::MAGIC_COOKIE;
599        let mut value = vec![0u8, 0x01];
600        value.extend_from_slice(&(v4.port() ^ u16::try_from(cookie >> 16).unwrap()).to_be_bytes());
601        let octets = u32::from(*v4.ip()) ^ cookie;
602        value.extend_from_slice(&octets.to_be_bytes());
603
604        let mut message = vec![0x01, 0x01];
605        message.extend_from_slice(&u16::try_from(value.len() + 4).unwrap().to_be_bytes());
606        message.extend_from_slice(&cookie.to_be_bytes());
607        message.extend_from_slice(id);
608        message.extend_from_slice(&0x0020u16.to_be_bytes());
609        message.extend_from_slice(&u16::try_from(value.len()).unwrap().to_be_bytes());
610        message.extend_from_slice(&value);
611        message
612    }
613}
sipx_media/ice/gather.rs

sipx_media/ice/
gather.rs
