sipx_media/
browser.rs

1//! The security gate for one browser-audio ICE component.
2//!
3//! This module is the I/O-free boundary from `docs/specs/webrtc-audio.md` §7. The live socket
4//! owner consults it before handing bytes to ICE, DTLS, SRTP or SRTCP; keeping it free of sockets
5//! makes every hostile-input and ordering branch deterministic in a test.
6
7use std::net::SocketAddr;
8
9use std::future::Future;
10#[cfg(feature = "dtls")]
11use std::io::{Read, Write};
12use std::sync::atomic::{AtomicUsize, Ordering};
13use std::sync::{Arc, Mutex as StdMutex};
14#[cfg(feature = "dtls")]
15use std::time::Duration;
16
17use sipx_sdp::ice::CandidateType;
18#[cfg(feature = "dtls")]
19use tokio::net::UdpSocket;
20#[cfg(feature = "dtls")]
21use tokio::sync::{Mutex, mpsc};
22
23#[cfg(all(test, feature = "dtls"))]
24use std::sync::atomic::AtomicBool;
25#[cfg(all(test, feature = "dtls"))]
26use tokio::sync::Notify;
27
28#[cfg(all(test, feature = "dtls"))]
29static ACTIVE_SUPERVISORS: AtomicUsize = AtomicUsize::new(0);
30#[cfg(all(test, feature = "dtls"))]
31static DTLS_HANDSHAKING: AtomicBool = AtomicBool::new(false);
32#[cfg(all(test, feature = "dtls"))]
33static SUPERVISOR_CHANGED: Notify = Notify::const_new();
34#[cfg(all(test, feature = "dtls"))]
35static HANDSHAKE_STARTED: Notify = Notify::const_new();
36
37/// Largest inbound datagram the browser-audio component admits to a protocol parser.
38pub const MAX_DATAGRAM: usize = 2048;
39
40/// Maximum browser-profile tasks alive at one instant (spec §7.2).
41pub const MAX_PROFILE_TASKS: usize = 6;
42
43/// Per-component accounting tied to the lifetime of every profile-owned task.
44#[derive(Debug, Default)]
45pub(crate) struct ProfileTasks {
46    active: AtomicUsize,
47    peak: AtomicUsize,
48}
49
50impl ProfileTasks {
51    fn enter(self: &Arc<Self>) -> ProfileTaskPermit {
52        let active = self.active.fetch_add(1, Ordering::SeqCst) + 1;
53        self.peak.fetch_max(active, Ordering::SeqCst);
54        ProfileTaskPermit {
55            tasks: Arc::clone(self),
56        }
57    }
58
59    #[cfg(all(test, feature = "dtls"))]
60    pub(crate) fn counts(&self) -> (usize, usize) {
61        (
62            self.active.load(Ordering::SeqCst),
63            self.peak.load(Ordering::SeqCst),
64        )
65    }
66}
67
68struct ProfileTaskPermit {
69    tasks: Arc<ProfileTasks>,
70}
71
72impl Drop for ProfileTaskPermit {
73    fn drop(&mut self) {
74        self.tasks.active.fetch_sub(1, Ordering::SeqCst);
75    }
76}
77
78pub(crate) async fn profile_task<F: Future>(tasks: Arc<ProfileTasks>, future: F) -> F::Output {
79    let _permit = tasks.enter();
80    future.await
81}
82
83/// One protocol carried by the nominated component.
84#[derive(Debug, Clone, Copy, PartialEq, Eq)]
85pub enum IngressClass {
86    /// An ICE/STUN datagram.
87    Stun,
88    /// A DTLS record.
89    Dtls,
90    /// A protected RTP packet.
91    Srtp,
92    /// A protected RTCP packet.
93    Srtcp,
94}
95
96/// Why a datagram changed no browser-component protocol state.
97#[derive(Debug, Clone, Copy, PartialEq, Eq)]
98#[non_exhaustive]
99pub enum IngressDrop {
100    /// The UDP payload was empty.
101    Empty,
102    /// An RTP/RTCP first byte arrived without the second byte needed by RFC 5761.
103    TruncatedClassPrefix,
104    /// The payload exceeded [`MAX_DATAGRAM`].
105    Oversized,
106    /// RFC 5764 assigns no protocol to the first-byte range.
107    UnknownProtocol,
108    /// Protected traffic or DTLS arrived before ICE selected a pair.
109    BeforeNomination,
110    /// Traffic came from an address other than the selected remote candidate.
111    WrongPeer,
112    /// Protected media arrived before all directional contexts were installed.
113    KeysUnavailable,
114    /// A new DTLS record arrived after the handshake boundary closed.
115    UnexpectedDtls,
116    /// The component no longer admits traffic.
117    Closed,
118}
119
120/// Classify one bounded datagram by RFC 5764 §5.1.2 and RFC 5761 §4.
121///
122/// No protocol parser is attempted here. Length is checked before either classifier byte is read.
123pub fn classify_datagram(datagram: &[u8]) -> Result<IngressClass, IngressDrop> {
124    if datagram.len() > MAX_DATAGRAM {
125        return Err(IngressDrop::Oversized);
126    }
127    let first = *datagram.first().ok_or(IngressDrop::Empty)?;
128    match first {
129        0..=1 => Ok(IngressClass::Stun),
130        20..=63 => Ok(IngressClass::Dtls),
131        128..=191 => {
132            let second = *datagram.get(1).ok_or(IngressDrop::TruncatedClassPrefix)?;
133            if (192..=223).contains(&second) {
134                Ok(IngressClass::Srtcp)
135            } else {
136                Ok(IngressClass::Srtp)
137            }
138        }
139        _ => Err(IngressDrop::UnknownProtocol),
140    }
141}
142
143/// The exact ICE pair allowed to carry DTLS and protected media.
144#[derive(Debug, Clone, Copy, PartialEq, Eq)]
145pub struct SelectedComponent {
146    /// Bound local candidate address.
147    pub local: SocketAddr,
148    /// Nominated remote candidate address.
149    pub remote: SocketAddr,
150    /// ICE generation whose credentials authenticated the pair.
151    pub ice_generation: u64,
152    /// How the local candidate was obtained.
153    pub local_kind: CandidateType,
154    /// How the remote candidate was obtained.
155    pub remote_kind: CandidateType,
156}
157
158impl SelectedComponent {
159    /// A host-to-host selected pair.
160    #[must_use]
161    pub const fn new(local: SocketAddr, remote: SocketAddr, ice_generation: u64) -> Self {
162        Self {
163            local,
164            remote,
165            ice_generation,
166            local_kind: CandidateType::Host,
167            remote_kind: CandidateType::Host,
168        }
169    }
170
171    /// Record the candidate types the ICE agent selected.
172    #[must_use]
173    pub const fn with_candidate_types(
174        mut self,
175        local_kind: CandidateType,
176        remote_kind: CandidateType,
177    ) -> Self {
178        self.local_kind = local_kind;
179        self.remote_kind = remote_kind;
180        self
181    }
182}
183
184/// Security phase of one component generation.
185#[derive(Debug, Clone, Copy, PartialEq, Eq)]
186pub enum ComponentState {
187    /// ICE is checking and no media peer exists yet.
188    IceChecking,
189    /// ICE selected the immutable pair; DTLS may start.
190    Nominated,
191    /// DTLS records from the selected peer are admitted.
192    DtlsHandshaking,
193    /// All directional SRTP/SRTCP key material is installed, but delivery is not enabled.
194    KeysInstalled,
195    /// Protected RTP and RTCP are admitted.
196    Running,
197    /// Admissions and key material are closed.
198    Closed,
199}
200
201/// Exact, monotonic drops decided by [`ComponentIngress`].
202#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
203pub struct IngressCounts {
204    /// Empty payloads.
205    pub ingress_empty: u64,
206    /// One-byte RTP/RTCP prefixes.
207    pub ingress_truncated_prefix: u64,
208    /// Oversized payloads.
209    pub ingress_oversized: u64,
210    /// Unassigned first-byte ranges.
211    pub ingress_unknown_protocol: u64,
212    /// DTLS before nomination.
213    pub dtls_before_nomination: u64,
214    /// SRTP before nomination.
215    pub srtp_before_nomination: u64,
216    /// SRTCP before nomination.
217    pub srtcp_before_nomination: u64,
218    /// DTLS from a non-nominated peer.
219    pub dtls_wrong_peer: u64,
220    /// SRTP from a non-nominated peer.
221    pub srtp_wrong_peer: u64,
222    /// SRTCP from a non-nominated peer.
223    pub srtcp_wrong_peer: u64,
224    /// SRTP received before key installation and media admission.
225    pub srtp_keys_unavailable: u64,
226    /// SRTCP received before key installation and media admission.
227    pub srtcp_keys_unavailable: u64,
228    /// DTLS records received after its admission phase.
229    pub dtls_unexpected_records: u64,
230    /// Classified traffic received after closure.
231    pub ingress_closed: u64,
232    /// STUN datagrams too short or malformed for ICE.
233    pub stun_malformed: u64,
234    /// DTLS records too short or malformed for the handshake.
235    pub dtls_malformed: u64,
236    /// SRTP packets too short or malformed for RTP.
237    pub srtp_malformed: u64,
238    /// SRTCP packets too short or malformed for RTCP.
239    pub srtcp_malformed: u64,
240    /// STUN handoffs refused by their bounded queue.
241    pub stun_queue_refusals: u64,
242    /// DTLS handoffs refused by their bounded queue.
243    pub dtls_queue_refusals: u64,
244    /// SRTP handoffs refused by their bounded queue.
245    pub srtp_queue_refusals: u64,
246    /// SRTCP handoffs refused by their bounded queue.
247    pub srtcp_queue_refusals: u64,
248    /// SRTP authentication failures excluding replay.
249    pub srtp_authentication_failures: u64,
250    /// SRTCP authentication failures excluding replay.
251    pub srtcp_authentication_failures: u64,
252    /// STUN messages whose short-term credential did not verify.
253    pub stun_authentication_failures: u64,
254    /// Replayed or too-old SRTP packets.
255    pub srtp_replays: u64,
256    /// Replayed or too-old SRTCP packets.
257    pub srtcp_replays: u64,
258    /// Authenticated RTCP compound packets applied by the media worker.
259    pub srtcp_processed: u64,
260}
261
262impl IngressCounts {
263    /// Sum every count, saturating rather than wrapping to a plausible low value.
264    #[must_use]
265    pub fn total(self) -> u64 {
266        [
267            self.ingress_empty,
268            self.ingress_truncated_prefix,
269            self.ingress_oversized,
270            self.ingress_unknown_protocol,
271            self.dtls_before_nomination,
272            self.srtp_before_nomination,
273            self.srtcp_before_nomination,
274            self.dtls_wrong_peer,
275            self.srtp_wrong_peer,
276            self.srtcp_wrong_peer,
277            self.srtp_keys_unavailable,
278            self.srtcp_keys_unavailable,
279            self.dtls_unexpected_records,
280            self.ingress_closed,
281            self.stun_malformed,
282            self.dtls_malformed,
283            self.srtp_malformed,
284            self.srtcp_malformed,
285            self.stun_queue_refusals,
286            self.dtls_queue_refusals,
287            self.srtp_queue_refusals,
288            self.srtcp_queue_refusals,
289            self.srtp_authentication_failures,
290            self.srtcp_authentication_failures,
291            self.stun_authentication_failures,
292            self.srtp_replays,
293            self.srtcp_replays,
294        ]
295        .into_iter()
296        .fold(0, u64::saturating_add)
297    }
298}
299
300/// Result of applying both classification and the component security state.
301#[derive(Debug, Clone, Copy, PartialEq, Eq)]
302pub enum IngressDisposition {
303    /// The named protocol may consume this datagram now.
304    Accepted(IngressClass),
305    /// The datagram was accounted for and must be released.
306    Dropped(IngressDrop),
307}
308
309/// Read-only facts for diagnostics and the independent browser proof.
310#[derive(Debug, Clone, Copy, PartialEq, Eq)]
311pub struct BrowserComponentSnapshot {
312    /// Current security phase.
313    pub state: ComponentState,
314    /// Selected pair, once ICE has nominated one.
315    pub selected: Option<SelectedComponent>,
316    /// Exact gate drops so far.
317    pub counts: IngressCounts,
318}
319
320/// A rejected component transition.
321#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
322#[non_exhaustive]
323pub enum ComponentError {
324    /// ICE selected a pair from a generation this gate does not own.
325    #[error("ICE nomination generation {got} does not match current generation {expected}")]
326    Generation {
327        /// Current generation.
328        expected: u64,
329        /// Generation attached to the nomination.
330        got: u64,
331    },
332    /// An operation was attempted outside its one permitted phase.
333    #[error("component operation {operation} is invalid in state {state:?}")]
334    State {
335        /// Operation being attempted.
336        operation: &'static str,
337        /// Current state.
338        state: ComponentState,
339    },
340    /// DTLS was asked to start for a source other than the selected peer.
341    #[error("DTLS peer {got} is not nominated peer {expected}")]
342    WrongPeer {
343        /// Selected peer.
344        expected: SocketAddr,
345        /// Requested peer.
346        got: SocketAddr,
347    },
348}
349
350/// I/O-free ingress and key-installation gate for one browser component.
351#[derive(Debug)]
352pub struct ComponentIngress {
353    generation: u64,
354    state: ComponentState,
355    selected: Option<SelectedComponent>,
356    keys: Option<crate::SrtpKeys>,
357    counts: IngressCounts,
358}
359
360impl ComponentIngress {
361    /// Begin checking one ICE generation.
362    #[must_use]
363    pub const fn new(generation: u64) -> Self {
364        Self {
365            generation,
366            state: ComponentState::IceChecking,
367            selected: None,
368            keys: None,
369            counts: IngressCounts {
370                ingress_empty: 0,
371                ingress_truncated_prefix: 0,
372                ingress_oversized: 0,
373                ingress_unknown_protocol: 0,
374                dtls_before_nomination: 0,
375                srtp_before_nomination: 0,
376                srtcp_before_nomination: 0,
377                dtls_wrong_peer: 0,
378                srtp_wrong_peer: 0,
379                srtcp_wrong_peer: 0,
380                srtp_keys_unavailable: 0,
381                srtcp_keys_unavailable: 0,
382                dtls_unexpected_records: 0,
383                ingress_closed: 0,
384                stun_malformed: 0,
385                dtls_malformed: 0,
386                srtp_malformed: 0,
387                srtcp_malformed: 0,
388                stun_queue_refusals: 0,
389                dtls_queue_refusals: 0,
390                srtp_queue_refusals: 0,
391                srtcp_queue_refusals: 0,
392                srtp_authentication_failures: 0,
393                srtcp_authentication_failures: 0,
394                stun_authentication_failures: 0,
395                srtp_replays: 0,
396                srtcp_replays: 0,
397                srtcp_processed: 0,
398            },
399        }
400    }
401
402    /// Freeze the pair selected by ICE for this generation.
403    pub fn nominate(&mut self, selected: SelectedComponent) -> Result<(), ComponentError> {
404        if selected.ice_generation != self.generation {
405            return Err(ComponentError::Generation {
406                expected: self.generation,
407                got: selected.ice_generation,
408            });
409        }
410        if self.state != ComponentState::IceChecking {
411            return Err(ComponentError::State {
412                operation: "nominate",
413                state: self.state,
414            });
415        }
416        self.selected = Some(selected);
417        self.state = ComponentState::Nominated;
418        Ok(())
419    }
420
421    /// Admit DTLS records from the nominated peer.
422    pub fn begin_dtls(&mut self, peer: SocketAddr) -> Result<(), ComponentError> {
423        if self.state != ComponentState::Nominated {
424            return Err(ComponentError::State {
425                operation: "begin_dtls",
426                state: self.state,
427            });
428        }
429        let expected =
430            self.selected
431                .map(|selected| selected.remote)
432                .ok_or(ComponentError::State {
433                    operation: "begin_dtls",
434                    state: self.state,
435                })?;
436        if peer != expected {
437            return Err(ComponentError::WrongPeer {
438                expected,
439                got: peer,
440            });
441        }
442        self.state = ComponentState::DtlsHandshaking;
443        Ok(())
444    }
445
446    /// Atomically install both directions from a fingerprint-verified DTLS result.
447    pub fn install_verified_keys(
448        &mut self,
449        keys: crate::dtls::VerifiedKeys,
450    ) -> Result<(), ComponentError> {
451        if self.state != ComponentState::DtlsHandshaking {
452            return Err(ComponentError::State {
453                operation: "install_verified_keys",
454                state: self.state,
455            });
456        }
457        self.keys = Some(keys.into_srtp_keys());
458        self.state = ComponentState::KeysInstalled;
459        Ok(())
460    }
461
462    /// Enable protected delivery and move the directional master material into the media session.
463    pub fn start_media(&mut self) -> Result<crate::SrtpKeys, ComponentError> {
464        if self.state != ComponentState::KeysInstalled {
465            return Err(ComponentError::State {
466                operation: "start_media",
467                state: self.state,
468            });
469        }
470        let keys = self.keys.take().ok_or(ComponentError::State {
471            operation: "start_media",
472            state: self.state,
473        })?;
474        self.state = ComponentState::Running;
475        Ok(keys)
476    }
477
478    /// Classify and apply the nominated-peer/key-state boundary, counting one refusal on failure.
479    pub fn admit(&mut self, source: SocketAddr, datagram: &[u8]) -> IngressDisposition {
480        let class = match classify_datagram(datagram) {
481            Ok(class) => class,
482            Err(reason) => {
483                self.note_drop(None, reason);
484                return IngressDisposition::Dropped(reason);
485            }
486        };
487        let reason = if self.state == ComponentState::Closed {
488            Some(IngressDrop::Closed)
489        } else if class == IngressClass::Stun {
490            None
491        } else if self.selected.is_none() {
492            Some(IngressDrop::BeforeNomination)
493        } else if self
494            .selected
495            .is_some_and(|selected| selected.remote != source)
496        {
497            Some(IngressDrop::WrongPeer)
498        } else {
499            match class {
500                IngressClass::Stun => None,
501                IngressClass::Dtls => (!matches!(
502                    self.state,
503                    ComponentState::Nominated | ComponentState::DtlsHandshaking
504                ))
505                .then_some(IngressDrop::UnexpectedDtls),
506                IngressClass::Srtp | IngressClass::Srtcp => {
507                    (self.state != ComponentState::Running).then_some(IngressDrop::KeysUnavailable)
508                }
509            }
510        };
511        if let Some(reason) = reason {
512            self.note_drop(Some(class), reason);
513            IngressDisposition::Dropped(reason)
514        } else {
515            IngressDisposition::Accepted(class)
516        }
517    }
518
519    /// Refuse future admissions and erase any key material not moved into a session.
520    pub fn close(&mut self) {
521        self.keys = None;
522        self.state = ComponentState::Closed;
523    }
524
525    /// Current low-cardinality diagnostic facts. No key material is exposed.
526    #[must_use]
527    pub const fn snapshot(&self) -> BrowserComponentSnapshot {
528        BrowserComponentSnapshot {
529            state: self.state,
530            selected: self.selected,
531            counts: self.counts,
532        }
533    }
534
535    pub(crate) fn note_malformed(&mut self, class: IngressClass) {
536        let counter = match class {
537            IngressClass::Stun => &mut self.counts.stun_malformed,
538            IngressClass::Dtls => &mut self.counts.dtls_malformed,
539            IngressClass::Srtp => &mut self.counts.srtp_malformed,
540            IngressClass::Srtcp => &mut self.counts.srtcp_malformed,
541        };
542        *counter = counter.saturating_add(1);
543    }
544
545    #[cfg(feature = "dtls")]
546    pub(crate) fn note_queue_full(&mut self, class: IngressClass) {
547        let counter = match class {
548            IngressClass::Stun => &mut self.counts.stun_queue_refusals,
549            IngressClass::Dtls => &mut self.counts.dtls_queue_refusals,
550            IngressClass::Srtp => &mut self.counts.srtp_queue_refusals,
551            IngressClass::Srtcp => &mut self.counts.srtcp_queue_refusals,
552        };
553        *counter = counter.saturating_add(1);
554    }
555
556    pub(crate) fn note_authentication_failure(&mut self, class: IngressClass) {
557        let counter = match class {
558            IngressClass::Stun => &mut self.counts.stun_authentication_failures,
559            IngressClass::Srtp => &mut self.counts.srtp_authentication_failures,
560            IngressClass::Srtcp => &mut self.counts.srtcp_authentication_failures,
561            IngressClass::Dtls => return,
562        };
563        *counter = counter.saturating_add(1);
564    }
565
566    pub(crate) fn note_replay(&mut self, class: IngressClass) {
567        let counter = match class {
568            IngressClass::Srtp => &mut self.counts.srtp_replays,
569            IngressClass::Srtcp => &mut self.counts.srtcp_replays,
570            IngressClass::Stun | IngressClass::Dtls => return,
571        };
572        *counter = counter.saturating_add(1);
573    }
574
575    pub(crate) fn note_srtcp_processed(&mut self) {
576        self.counts.srtcp_processed = self.counts.srtcp_processed.saturating_add(1);
577    }
578
579    fn note_drop(&mut self, class: Option<IngressClass>, reason: IngressDrop) {
580        let counter = match (class, reason) {
581            (_, IngressDrop::Empty) => &mut self.counts.ingress_empty,
582            (_, IngressDrop::TruncatedClassPrefix) => &mut self.counts.ingress_truncated_prefix,
583            (_, IngressDrop::Oversized) => &mut self.counts.ingress_oversized,
584            (_, IngressDrop::UnknownProtocol) => &mut self.counts.ingress_unknown_protocol,
585            (Some(IngressClass::Dtls), IngressDrop::BeforeNomination) => {
586                &mut self.counts.dtls_before_nomination
587            }
588            (Some(IngressClass::Srtp), IngressDrop::BeforeNomination) => {
589                &mut self.counts.srtp_before_nomination
590            }
591            (Some(IngressClass::Srtcp), IngressDrop::BeforeNomination) => {
592                &mut self.counts.srtcp_before_nomination
593            }
594            (Some(IngressClass::Dtls), IngressDrop::WrongPeer) => &mut self.counts.dtls_wrong_peer,
595            (Some(IngressClass::Srtp), IngressDrop::WrongPeer) => &mut self.counts.srtp_wrong_peer,
596            (Some(IngressClass::Srtcp), IngressDrop::WrongPeer) => {
597                &mut self.counts.srtcp_wrong_peer
598            }
599            (Some(IngressClass::Srtp), IngressDrop::KeysUnavailable) => {
600                &mut self.counts.srtp_keys_unavailable
601            }
602            (Some(IngressClass::Srtcp), IngressDrop::KeysUnavailable) => {
603                &mut self.counts.srtcp_keys_unavailable
604            }
605            (Some(IngressClass::Dtls), IngressDrop::UnexpectedDtls) => {
606                &mut self.counts.dtls_unexpected_records
607            }
608            (_, IngressDrop::Closed) => &mut self.counts.ingress_closed,
609            // STUN is accepted by this gate; malformed/authentication and queue outcomes belong
610            // to its parser/adapter. All other combinations are structurally unreachable.
611            _ => return,
612        };
613        *counter = counter.saturating_add(1);
614    }
615}
616
617/// Why the browser component could not advance from ICE through verified DTLS keying.
618#[cfg(feature = "dtls")]
619#[derive(Debug, thiserror::Error)]
620#[non_exhaustive]
621pub enum BrowserStartError {
622    /// Codec, pacing or mux configuration was invalid before the owner started.
623    #[error("media setup: {0}")]
624    Setup(#[from] crate::SetupError),
625    /// The named browser path was started without negotiated RTP/RTCP multiplexing.
626    #[error("browser audio requires RTP/RTCP multiplexing")]
627    RtcpMuxRequired,
628    /// ICE concluded component 1 without selecting a pair.
629    #[error("ICE failed before nominating the browser-audio component")]
630    IceFailed,
631    /// ICE stopped before reporting selection or failure.
632    #[error("ICE stopped before nominating the browser-audio component")]
633    IceStopped,
634    /// The bounded DTLS handshake did not complete.
635    #[error("DTLS handshake exceeded its configured deadline")]
636    DtlsTimeout,
637    /// The component security gate rejected an internal transition.
638    #[error("browser component transition: {0}")]
639    Component(#[from] ComponentError),
640    /// The blocking DTLS worker did not complete normally.
641    #[error("DTLS worker: {0}")]
642    Worker(String),
643    /// DTLS, fingerprint verification or key derivation failed.
644    #[error("DTLS: {0}")]
645    Dtls(#[from] crate::dtls::Error),
646    /// The DTLS adapter could not be constructed.
647    #[error("DTLS adapter: {0}")]
648    Adapter(String),
649}
650
651#[cfg(feature = "dtls")]
652#[derive(Debug)]
653pub(crate) struct Datagram {
654    pub(crate) source: SocketAddr,
655    pub(crate) bytes: Vec<u8>,
656}
657
658/// Receivers attached to the ordinary RTP/SRTCP workers after key installation.
659#[cfg(feature = "dtls")]
660#[derive(Debug)]
661pub(crate) struct MediaIngress {
662    pub(crate) srtp: mpsc::Receiver<Datagram>,
663    pub(crate) srtcp: mpsc::Receiver<Datagram>,
664}
665
666/// The live, still-bound component and its one receive owner.
667#[cfg(feature = "dtls")]
668#[derive(Debug)]
669pub(crate) struct Runtime {
670    pub(crate) socket: Arc<UdpSocket>,
671    pub(crate) media: MediaIngress,
672    pub(crate) ice: crate::ice::driver::Handle,
673    pub(crate) ingress: Arc<StdMutex<ComponentIngress>>,
674    pub(crate) owner: tokio::task::JoinHandle<()>,
675    pub(crate) ice_owner: tokio::task::JoinHandle<()>,
676    pub(crate) stop: Arc<crate::session::Stop>,
677    pub(crate) profile_tasks: Arc<ProfileTasks>,
678}
679
680#[cfg(feature = "dtls")]
681const DTLS_QUEUE: usize = 64;
682#[cfg(feature = "dtls")]
683const SRTP_QUEUE: usize = 64;
684#[cfg(feature = "dtls")]
685const SRTCP_QUEUE: usize = 32;
686#[cfg(feature = "dtls")]
687const OUTBOUND_QUEUE: usize = 64;
688#[cfg(feature = "dtls")]
689const MAX_OUTBOUND_DATAGRAM: usize = 1200;
690
691/// Start the sole receiver, await exact ICE nomination, run DTLS through it and install keys.
692#[cfg(feature = "dtls")]
693#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
694pub(crate) async fn prepare(
695    socket: Arc<UdpSocket>,
696    local: crate::ice::LocalDescription,
697    ice_generation: u64,
698    identity: crate::dtls::openssl::Identity,
699    role: crate::dtls::Role,
700    fingerprint: sipx_sdp::fingerprint::Fingerprint,
701    timeout: Duration,
702    stop: Arc<crate::session::Stop>,
703    discards: Arc<crate::counters::DiscardMeters>,
704) -> Result<(Runtime, crate::SrtpKeys), BrowserStartError> {
705    let (finished, result) = tokio::sync::oneshot::channel();
706    let profile_tasks = Arc::new(ProfileTasks::default());
707    let mut cancellation = StopOnDrop {
708        stop: Arc::clone(&stop),
709        armed: true,
710    };
711    tokio::spawn(async move {
712        let task_permit = profile_tasks.enter();
713        #[cfg(all(test, feature = "dtls"))]
714        let _activity = SupervisorActivity::new();
715        let outcome = prepare_inner(
716            socket,
717            local,
718            ice_generation,
719            identity,
720            role,
721            fingerprint,
722            timeout,
723            stop,
724            discards,
725            Arc::clone(&profile_tasks),
726        )
727        .await;
728        // The preparation supervisor is not a running-session task. End its counted lifetime
729        // before waking the caller that will attach the four media workers.
730        drop(task_permit);
731        if let Err(returned) = finished.send(outcome)
732            && let Ok((runtime, _keys)) = returned
733        {
734            cleanup_runtime(runtime).await;
735        }
736    });
737    let outcome = result.await.map_err(|_| {
738        BrowserStartError::Worker(
739            "browser-component supervisor stopped without a result".to_owned(),
740        )
741    })?;
742    cancellation.armed = false;
743    outcome
744}
745
746#[cfg(feature = "dtls")]
747struct StopOnDrop {
748    stop: Arc<crate::session::Stop>,
749    armed: bool,
750}
751
752#[cfg(feature = "dtls")]
753impl Drop for StopOnDrop {
754    fn drop(&mut self) {
755        if self.armed {
756            self.stop.stop();
757        }
758    }
759}
760
761#[cfg(feature = "dtls")]
762#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
763async fn prepare_inner(
764    socket: Arc<UdpSocket>,
765    local: crate::ice::LocalDescription,
766    ice_generation: u64,
767    identity: crate::dtls::openssl::Identity,
768    role: crate::dtls::Role,
769    fingerprint: sipx_sdp::fingerprint::Fingerprint,
770    timeout: Duration,
771    stop: Arc<crate::session::Stop>,
772    discards: Arc<crate::counters::DiscardMeters>,
773    profile_tasks: Arc<ProfileTasks>,
774) -> Result<(Runtime, crate::SrtpKeys), BrowserStartError> {
775    let placeholder = SocketAddr::new(
776        socket
777            .local_addr()
778            .map_err(|error| {
779                BrowserStartError::Adapter(format!("component has no local address: {error}"))
780            })?
781            .ip(),
782        0,
783    );
784    let destinations = crate::ice::driver::Destinations {
785        rtp: Arc::new(Mutex::new(placeholder)),
786        rtcp: Arc::new(Mutex::new(None)),
787    };
788    let (agent, pending) = local.into_driver_parts();
789    let peering = agent.peering().cloned();
790    let crate::ice::driver::OwnedDriver {
791        handle: ice,
792        task: ice_owner,
793    } = crate::ice::driver::spawn_owned(
794        agent,
795        pending,
796        vec![Arc::clone(&socket)],
797        destinations,
798        Arc::clone(&stop),
799        discards,
800        Arc::clone(&profile_tasks),
801    );
802
803    let ingress = Arc::new(StdMutex::new(ComponentIngress::new(ice_generation)));
804    let (dtls_in, dtls_rx) = std::sync::mpsc::sync_channel(DTLS_QUEUE);
805    let (dtls_out, dtls_writes) = mpsc::channel(OUTBOUND_QUEUE);
806    let (audio_packets, srtp_rx) = mpsc::channel(SRTP_QUEUE);
807    let (control_packets, srtcp_rx) = mpsc::channel(SRTCP_QUEUE);
808    let owner = tokio::spawn(profile_task(
809        Arc::clone(&profile_tasks),
810        owner_loop(
811            Arc::clone(&socket),
812            ice.clone(),
813            Arc::clone(&ingress),
814            peering,
815            dtls_in,
816            dtls_writes,
817            audio_packets,
818            control_packets,
819            Arc::clone(&stop),
820        ),
821    ));
822    let mut tasks = PreparingTasks {
823        stop: Arc::clone(&stop),
824        owner: Some(owner),
825        ice_owner: Some(ice_owner),
826        dtls: None,
827        armed: true,
828    };
829
830    let selected = match ice.wait_selected(ice_generation).await {
831        Ok(selected) => selected,
832        Err(crate::ice::driver::SelectionError::Failed) => {
833            tasks.cleanup().await;
834            return Err(BrowserStartError::IceFailed);
835        }
836        Err(crate::ice::driver::SelectionError::Stopped) => {
837            tasks.cleanup().await;
838            return Err(BrowserStartError::IceStopped);
839        }
840    };
841    let transition = {
842        let mut gate = lock_ingress(&ingress);
843        gate.nominate(selected)
844            .and_then(|()| gate.begin_dtls(selected.remote))
845    };
846    if let Err(error) = transition {
847        tasks.cleanup().await;
848        return Err(error.into());
849    }
850    #[cfg(all(test, feature = "dtls"))]
851    {
852        DTLS_HANDSHAKING.store(true, Ordering::SeqCst);
853        HANDSHAKE_STARTED.notify_waiters();
854    }
855
856    let adapter = DtlsAdapter {
857        inbound: dtls_rx,
858        outbound: dtls_out,
859        timeout: timeout.saturating_add(Duration::from_millis(250)),
860    };
861    let dtls_tasks = Arc::clone(&profile_tasks);
862    tasks.dtls = Some(tokio::task::spawn_blocking(move || {
863        let _permit = dtls_tasks.enter();
864        let mut handshake = crate::dtls::openssl::Session::with_io(adapter, &identity)
865            .map_err(|error| crate::dtls::Error::Dtls(error.to_string()))?;
866        crate::dtls::establish_verified(&mut handshake, role, Some(&fingerprint))
867    }));
868    let handshake = if let Some(worker) = tasks.dtls.as_mut() {
869        tokio::time::timeout(timeout, worker).await
870    } else {
871        tasks.cleanup().await;
872        return Err(BrowserStartError::Worker(
873            "DTLS worker was not retained".to_owned(),
874        ));
875    };
876    let verified = match handshake {
877        Err(_elapsed) => {
878            tasks.cleanup().await;
879            return Err(BrowserStartError::DtlsTimeout);
880        }
881        Ok(Err(error)) => {
882            tasks.dtls.take();
883            tasks.cleanup().await;
884            return Err(BrowserStartError::Worker(error.to_string()));
885        }
886        Ok(Ok(Err(error))) => {
887            tasks.dtls.take();
888            tasks.cleanup().await;
889            return Err(error.into());
890        }
891        Ok(Ok(Ok(verified))) => {
892            tasks.dtls.take();
893            verified
894        }
895    };
896
897    let installation = {
898        let mut gate = lock_ingress(&ingress);
899        gate.install_verified_keys(verified)
900            .and_then(|()| gate.start_media())
901    };
902    let keys = match installation {
903        Ok(keys) => keys,
904        Err(error) => {
905            tasks.cleanup().await;
906            return Err(error.into());
907        }
908    };
909    let owner = tasks.owner.take().ok_or(BrowserStartError::IceStopped)?;
910    let ice_owner = tasks
911        .ice_owner
912        .take()
913        .ok_or(BrowserStartError::IceStopped)?;
914    tasks.armed = false;
915    Ok((
916        Runtime {
917            socket,
918            media: MediaIngress {
919                srtp: srtp_rx,
920                srtcp: srtcp_rx,
921            },
922            ice,
923            ingress,
924            owner,
925            ice_owner,
926            stop,
927            profile_tasks,
928        },
929        keys,
930    ))
931}
932
933#[cfg(feature = "dtls")]
934async fn cleanup_runtime(runtime: Runtime) {
935    runtime.stop.stop();
936    lock_ingress(&runtime.ingress).close();
937    runtime.owner.abort();
938    runtime.ice_owner.abort();
939    // discard: cancellation is the requested terminal result; awaiting only proves both owned
940    // tasks were reaped, so their expected cancelled JoinErrors carry no additional outcome.
941    let _ = runtime.owner.await;
942    let _ = runtime.ice_owner.await;
943}
944
945#[cfg(feature = "dtls")]
946struct PreparingTasks {
947    stop: Arc<crate::session::Stop>,
948    owner: Option<tokio::task::JoinHandle<()>>,
949    ice_owner: Option<tokio::task::JoinHandle<()>>,
950    dtls: Option<tokio::task::JoinHandle<Result<crate::dtls::VerifiedKeys, crate::dtls::Error>>>,
951    armed: bool,
952}
953
954#[cfg(feature = "dtls")]
955impl PreparingTasks {
956    async fn cleanup(&mut self) {
957        self.stop.stop();
958        if let Some(task) = &self.dtls {
959            task.abort();
960        }
961        if let Some(task) = self.dtls.take() {
962            // discard: the caller already retained the typed handshake outcome; this await only
963            // reaps a worker that was either completed or explicitly aborted.
964            let _ = task.await;
965        }
966        if let Some(task) = self.owner.take() {
967            // discard: the preparation outcome is already terminal and stop is set; awaiting the
968            // owner proves cleanup, while a cancellation JoinError changes no public result.
969            let _ = task.await;
970        }
971        if let Some(task) = self.ice_owner.take() {
972            // discard: the preparation outcome is already terminal and stop is set; awaiting the
973            // ICE owner proves cleanup, while a cancellation JoinError changes no public result.
974            let _ = task.await;
975        }
976        self.armed = false;
977    }
978}
979
980#[cfg(feature = "dtls")]
981impl Drop for PreparingTasks {
982    fn drop(&mut self) {
983        if !self.armed {
984            return;
985        }
986        self.stop.stop();
987        if let Some(task) = &self.dtls {
988            task.abort();
989        }
990        if let Some(task) = &self.owner {
991            task.abort();
992        }
993        if let Some(task) = &self.ice_owner {
994            task.abort();
995        }
996    }
997}
998
999#[cfg(all(test, feature = "dtls"))]
1000struct SupervisorActivity;
1001
1002#[cfg(all(test, feature = "dtls"))]
1003impl SupervisorActivity {
1004    fn new() -> Self {
1005        ACTIVE_SUPERVISORS.fetch_add(1, Ordering::SeqCst);
1006        SUPERVISOR_CHANGED.notify_waiters();
1007        Self
1008    }
1009}
1010
1011#[cfg(all(test, feature = "dtls"))]
1012impl Drop for SupervisorActivity {
1013    fn drop(&mut self) {
1014        ACTIVE_SUPERVISORS.fetch_sub(1, Ordering::SeqCst);
1015        DTLS_HANDSHAKING.store(false, Ordering::SeqCst);
1016        SUPERVISOR_CHANGED.notify_waiters();
1017    }
1018}
1019
1020#[cfg(feature = "dtls")]
1021#[allow(clippy::too_many_arguments)]
1022async fn owner_loop(
1023    socket: Arc<UdpSocket>,
1024    ice: crate::ice::driver::Handle,
1025    ingress: Arc<StdMutex<ComponentIngress>>,
1026    peering: Option<crate::ice::stun::Peering>,
1027    dtls: std::sync::mpsc::SyncSender<Vec<u8>>,
1028    mut dtls_writes: mpsc::Receiver<Vec<u8>>,
1029    audio_packets: mpsc::Sender<Datagram>,
1030    control_packets: mpsc::Sender<Datagram>,
1031    stop: Arc<crate::session::Stop>,
1032) {
1033    let mut datagram = vec![0u8; MAX_DATAGRAM + 1];
1034    let mut dtls_writes_open = true;
1035    loop {
1036        let received = tokio::select! {
1037            () = stop.wait() => return,
1038            write = dtls_writes.recv(), if dtls_writes_open => {
1039                let Some(bytes) = write else {
1040                    dtls_writes_open = false;
1041                    continue;
1042                };
1043                let peer = lock_ingress(&ingress).snapshot().selected.map(|pair| pair.remote);
1044                if let Some(peer) = peer
1045                    && socket.send_to(&bytes, peer).await.is_err()
1046                {
1047                    return;
1048                }
1049                continue;
1050            }
1051            received = socket.recv_from(&mut datagram) => received,
1052        };
1053        let Ok((length, source)) = received else {
1054            return;
1055        };
1056        let bytes = datagram.get(..length).unwrap_or_default();
1057        let disposition = lock_ingress(&ingress).admit(source, bytes);
1058        let IngressDisposition::Accepted(class) = disposition else {
1059            continue;
1060        };
1061        if bytes.len() < minimum_length(class) {
1062            lock_ingress(&ingress).note_malformed(class);
1063            continue;
1064        }
1065        if class == IngressClass::Stun {
1066            let accepted = {
1067                let mut gate = lock_ingress(&ingress);
1068                account_stun(&mut gate, bytes, peering.as_ref())
1069            };
1070            if !accepted {
1071                continue;
1072            }
1073        }
1074        let admitted = match class {
1075            IngressClass::Stun => ice.datagram(source, crate::ice::LocalBase(0), bytes.to_vec()),
1076            IngressClass::Dtls => dtls.try_send(bytes.to_vec()).is_ok(),
1077            IngressClass::Srtp => audio_packets
1078                .try_send(Datagram {
1079                    source,
1080                    bytes: bytes.to_vec(),
1081                })
1082                .is_ok(),
1083            IngressClass::Srtcp => control_packets
1084                .try_send(Datagram {
1085                    source,
1086                    bytes: bytes.to_vec(),
1087                })
1088                .is_ok(),
1089        };
1090        if !admitted {
1091            lock_ingress(&ingress).note_queue_full(class);
1092        }
1093    }
1094}
1095
1096#[cfg(feature = "dtls")]
1097enum StunDrop {
1098    Malformed,
1099    AuthenticationFailed,
1100}
1101
1102#[cfg(feature = "dtls")]
1103fn validate_stun(
1104    bytes: &[u8],
1105    peering: Option<&crate::ice::stun::Peering>,
1106) -> Result<(), StunDrop> {
1107    use crate::ice::stun::{Class, Message};
1108
1109    let message = Message::decode(bytes).map_err(|_| StunDrop::Malformed)?;
1110    let peering = peering.ok_or(StunDrop::AuthenticationFailed)?;
1111    let authenticated = match message.class() {
1112        Class::Request => {
1113            message.username() == Some(peering.inbound_username().as_str())
1114                && message.verify_integrity(peering.inbound_key())
1115        }
1116        Class::Success | Class::Error => message.verify_integrity(peering.outbound_key()),
1117        Class::Indication => !message.has_integrity(),
1118    };
1119    if authenticated {
1120        Ok(())
1121    } else {
1122        Err(StunDrop::AuthenticationFailed)
1123    }
1124}
1125
1126#[cfg(feature = "dtls")]
1127fn account_stun(
1128    ingress: &mut ComponentIngress,
1129    bytes: &[u8],
1130    peering: Option<&crate::ice::stun::Peering>,
1131) -> bool {
1132    match validate_stun(bytes, peering) {
1133        Ok(()) => true,
1134        Err(StunDrop::Malformed) => {
1135            ingress.note_malformed(IngressClass::Stun);
1136            false
1137        }
1138        Err(StunDrop::AuthenticationFailed) => {
1139            ingress.note_authentication_failure(IngressClass::Stun);
1140            false
1141        }
1142    }
1143}
1144
1145#[cfg(all(test, feature = "dtls"))]
1146#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
1147mod tests {
1148    use super::*;
1149    use crate::ice::stun::{Peering, RoleAttribute, connectivity_check, new_transaction_id};
1150    use crate::ice::{Gathering, Negotiation, Timers};
1151    use crate::{Codec, Config, MediaPort};
1152    use sipx_sdp::RtcpMode;
1153    use sipx_sdp::ice::{Credentials, Priority};
1154
1155    fn credentials(ufrag: &str, password: &str) -> Credentials {
1156        Credentials::new(ufrag, password).expect("valid ICE credentials")
1157    }
1158
1159    fn gathering(ufrag: &str, offerer: bool) -> Gathering {
1160        let mut gathering =
1161            Gathering::new(credentials(ufrag, "browserPassword0123456789AB"), offerer);
1162        gathering.agent.timers = Timers {
1163            ta: Duration::from_millis(20),
1164            tn: Duration::from_millis(250),
1165            tr: Duration::from_millis(200),
1166            ..Timers::default()
1167        };
1168        gathering
1169    }
1170
1171    fn peer(local: &crate::ice::LocalDescription) -> Negotiation {
1172        Negotiation::Ice {
1173            credentials: local.credentials().clone(),
1174            candidates: local.candidates().to_vec(),
1175            lite: false,
1176        }
1177    }
1178
1179    fn config(remote: SocketAddr) -> Config {
1180        let mut config = Config::new(remote, Codec::Pcmu);
1181        config.rtcp_mode = RtcpMode::Mux;
1182        config.rtcp_interval = None;
1183        config
1184    }
1185
1186    async fn wait_for_supervisors(expected: usize) {
1187        let wait = async {
1188            loop {
1189                let notified = SUPERVISOR_CHANGED.notified();
1190                tokio::pin!(notified);
1191                notified.as_mut().enable();
1192                if ACTIVE_SUPERVISORS.load(Ordering::SeqCst) == expected {
1193                    return;
1194                }
1195                notified.await;
1196            }
1197        };
1198        tokio::time::timeout(Duration::from_secs(2), wait)
1199            .await // bound on failure: supervisor cleanup has no timing semantics.
1200            .expect("supervisor count reaches the expected value");
1201    }
1202
1203    async fn wait_for_handshake() {
1204        let wait = async {
1205            loop {
1206                let notified = HANDSHAKE_STARTED.notified();
1207                tokio::pin!(notified);
1208                notified.as_mut().enable();
1209                if DTLS_HANDSHAKING.load(Ordering::SeqCst) {
1210                    return;
1211                }
1212                notified.await;
1213            }
1214        };
1215        tokio::time::timeout(Duration::from_secs(2), wait)
1216            .await // bound on failure: waits for the exact gate transition.
1217            .expect("DTLS handshaking begins");
1218    }
1219
1220    #[test]
1221    fn structurally_sized_bad_stun_is_counted_exactly_once() {
1222        let local = credentials("local1", "localPassword0123456789AB");
1223        let remote = credentials("remote", "remotePassword0123456789A");
1224        let peering = Peering::new(local.clone(), remote.clone());
1225        let mut ingress = ComponentIngress::new(0);
1226
1227        let bad_cookie = [0u8; 20];
1228        assert!(!account_stun(&mut ingress, &bad_cookie, Some(&peering)));
1229        assert_eq!(ingress.snapshot().counts.stun_malformed, 1);
1230        assert_eq!(ingress.snapshot().counts.total(), 1);
1231
1232        let forged_remote = Peering::new(remote, credentials("local1", "wrongPassword0123456789A"));
1233        let forged = connectivity_check(
1234            new_transaction_id(),
1235            &forged_remote,
1236            Priority::new(100).expect("priority"),
1237            RoleAttribute::Controlled { tiebreaker: 7 },
1238        )
1239        .expect("encoded check");
1240        assert!(!account_stun(&mut ingress, &forged, Some(&peering)));
1241        let counts = ingress.snapshot().counts;
1242        assert_eq!(counts.stun_authentication_failures, 1);
1243        assert_eq!(counts.stun_malformed, 1);
1244        assert_eq!(counts.total(), 2);
1245    }
1246
1247    #[test]
1248    fn every_bounded_ingress_queue_refusal_is_counted_once() {
1249        let mut ingress = ComponentIngress::new(0);
1250        for class in [
1251            IngressClass::Stun,
1252            IngressClass::Dtls,
1253            IngressClass::Srtp,
1254            IngressClass::Srtcp,
1255        ] {
1256            ingress.note_queue_full(class);
1257        }
1258        let counts = ingress.snapshot().counts;
1259        assert_eq!(counts.stun_queue_refusals, 1);
1260        assert_eq!(counts.dtls_queue_refusals, 1);
1261        assert_eq!(counts.srtp_queue_refusals, 1);
1262        assert_eq!(counts.srtcp_queue_refusals, 1);
1263        assert_eq!(counts.total(), 4);
1264    }
1265
1266    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1267    async fn cancellation_reaps_ice_and_dtls_preparation_without_a_detached_receiver() {
1268        use crate::dtls::Role;
1269        use crate::dtls::openssl::Identity;
1270
1271        // Cancel while ICE is checking and no peer is running.
1272        let alice_port = MediaPort::bind("127.0.0.1:0".parse().expect("address"))
1273            .await
1274            .expect("Alice port");
1275        let bob_port = MediaPort::bind("127.0.0.1:0".parse().expect("address"))
1276            .await
1277            .expect("Bob fixture port");
1278        let alice_addr = alice_port.local_addr();
1279        let alice_gathering = gathering("alice4", true);
1280        let bob_gathering = gathering("bob004", false);
1281        let (mut alice_ice, bob_ice) = tokio::join!(
1282            alice_port.gather_with_rtcp_mode(&alice_gathering, RtcpMode::Mux),
1283            bob_port.gather_with_rtcp_mode(&bob_gathering, RtcpMode::Mux),
1284        );
1285        assert!(alice_ice.accept(&peer(&bob_ice)));
1286        let identity = Identity::generate().expect("identity");
1287        let fingerprint = Identity::generate()
1288            .expect("peer identity")
1289            .fingerprint()
1290            .expect("peer fingerprint");
1291        let checking = tokio::spawn(alice_port.start_browser_audio(
1292            config(bob_port.local_addr()),
1293            alice_ice,
1294            0,
1295            identity,
1296            Role::Client,
1297            fingerprint,
1298            Duration::from_secs(5),
1299        ));
1300        wait_for_supervisors(1).await;
1301        assert!(!DTLS_HANDSHAKING.load(Ordering::SeqCst));
1302        checking.abort();
1303        // discard: this test requested cancellation and asserts cleanup through the supervisor
1304        // census and immediate port rebind below, rather than through the expected JoinError.
1305        let _ = checking.await;
1306        wait_for_supervisors(0).await;
1307        drop(bob_port);
1308        drop(
1309            UdpSocket::bind(alice_addr)
1310                .await
1311                .expect("ICE cancellation released the port"),
1312        );
1313
1314        // Cancel after exact nomination has opened the DTLS adapter.
1315        let alice_port = MediaPort::bind("127.0.0.1:0".parse().expect("address"))
1316            .await
1317            .expect("Alice port");
1318        let bob_port = MediaPort::bind("127.0.0.1:0".parse().expect("address"))
1319            .await
1320            .expect("Bob port");
1321        let (alice_addr, bob_addr) = (alice_port.local_addr(), bob_port.local_addr());
1322        let alice_gathering = gathering("alice5", true);
1323        let bob_gathering = gathering("bob005", false);
1324        let (mut alice_ice, mut bob_ice) = tokio::join!(
1325            alice_port.gather_with_rtcp_mode(&alice_gathering, RtcpMode::Mux),
1326            bob_port.gather_with_rtcp_mode(&bob_gathering, RtcpMode::Mux),
1327        );
1328        assert!(alice_ice.accept(&peer(&bob_ice)));
1329        assert!(bob_ice.accept(&peer(&alice_ice)));
1330        let bob = bob_port
1331            .start_with_ice(config(alice_addr), bob_ice)
1332            .expect("ordinary ICE peer");
1333        let identity = Identity::generate().expect("identity");
1334        let fingerprint = Identity::generate()
1335            .expect("peer identity")
1336            .fingerprint()
1337            .expect("peer fingerprint");
1338        let handshaking = tokio::spawn(alice_port.start_browser_audio(
1339            config(bob_addr),
1340            alice_ice,
1341            0,
1342            identity,
1343            Role::Client,
1344            fingerprint,
1345            Duration::from_secs(5),
1346        ));
1347        wait_for_handshake().await;
1348        handshaking.abort();
1349        // discard: this test requested cancellation and asserts cleanup through the supervisor
1350        // census and immediate port rebind below, rather than through the expected JoinError.
1351        let _ = handshaking.await;
1352        wait_for_supervisors(0).await;
1353        drop(bob);
1354        drop(
1355            UdpSocket::bind(alice_addr)
1356                .await
1357                .expect("DTLS cancellation released the port"),
1358        );
1359    }
1360
1361    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
1362    async fn actual_spawn_lifetimes_stay_inside_the_phase_task_bound() {
1363        use crate::dtls::Role;
1364        use crate::dtls::openssl::Identity;
1365
1366        let alice_port = MediaPort::bind("127.0.0.1:0".parse().expect("address"))
1367            .await
1368            .expect("Alice port");
1369        let bob_port = MediaPort::bind("127.0.0.1:0".parse().expect("address"))
1370            .await
1371            .expect("Bob port");
1372        let (alice_addr, bob_addr) = (alice_port.local_addr(), bob_port.local_addr());
1373        let alice_gathering = gathering("alice6", true);
1374        let bob_gathering = gathering("bob006", false);
1375        let (mut alice_ice, mut bob_ice) = tokio::join!(
1376            alice_port.gather_with_rtcp_mode(&alice_gathering, RtcpMode::Mux),
1377            bob_port.gather_with_rtcp_mode(&bob_gathering, RtcpMode::Mux),
1378        );
1379        assert!(alice_ice.accept(&peer(&bob_ice)));
1380        assert!(bob_ice.accept(&peer(&alice_ice)));
1381        let alice_identity = Identity::generate().expect("Alice identity");
1382        let bob_identity = Identity::generate().expect("Bob identity");
1383        let alice_fingerprint = alice_identity.fingerprint().expect("Alice fingerprint");
1384        let bob_fingerprint = bob_identity.fingerprint().expect("Bob fingerprint");
1385        let mut alice_config = config(bob_addr);
1386        let mut bob_config = config(alice_addr);
1387        alice_config.rtcp_interval = Some(Duration::from_millis(50));
1388        bob_config.rtcp_interval = Some(Duration::from_millis(50));
1389
1390        let (alice, bob) = tokio::time::timeout(Duration::from_secs(8), async {
1391            tokio::join!(
1392                alice_port.start_browser_audio(
1393                    alice_config,
1394                    alice_ice,
1395                    0,
1396                    alice_identity,
1397                    Role::Client,
1398                    bob_fingerprint,
1399                    Duration::from_secs(5),
1400                ),
1401                bob_port.start_browser_audio(
1402                    bob_config,
1403                    bob_ice,
1404                    0,
1405                    bob_identity,
1406                    Role::Server,
1407                    alice_fingerprint,
1408                    Duration::from_secs(5),
1409                ),
1410            )
1411        })
1412        .await // bound on failure: the task census requires a completed handshake.
1413        .expect("both components reach Running");
1414        let alice = alice.expect("Alice starts");
1415        let bob = bob.expect("Bob starts");
1416
1417        tokio::time::timeout(Duration::from_secs(1), async {
1418            loop {
1419                let alice_peak = alice.browser_task_counts().map_or(0, |(_, _, peak)| peak);
1420                let bob_peak = bob.browser_task_counts().map_or(0, |(_, _, peak)| peak);
1421                if alice_peak == MAX_PROFILE_TASKS && bob_peak == MAX_PROFILE_TASKS {
1422                    break;
1423                }
1424                tokio::task::yield_now().await;
1425            }
1426        })
1427        .await // bound on failure: every running worker must enter its counted future promptly.
1428        .expect("running workers enter the task census");
1429        for session in [&alice, &bob] {
1430            let (preparing_peak, active, peak) =
1431                session.browser_task_counts().expect("browser task census");
1432            assert_eq!(preparing_peak, 4);
1433            assert_eq!(peak, MAX_PROFILE_TASKS);
1434            assert!(active <= MAX_PROFILE_TASKS);
1435        }
1436        let alice_tasks = alice.browser_task_probe().expect("Alice task probe");
1437        let bob_tasks = bob.browser_task_probe().expect("Bob task probe");
1438        drop(alice);
1439        drop(bob);
1440        tokio::time::timeout(Duration::from_secs(1), async {
1441            while alice_tasks.counts().0 != 0 || bob_tasks.counts().0 != 0 {
1442                tokio::task::yield_now().await;
1443            }
1444        })
1445        .await // bound on failure: Drop aborts and reaps every retained profile worker.
1446        .expect("running-session cancellation leaves no profile task");
1447        drop(
1448            UdpSocket::bind(alice_addr)
1449                .await
1450                .expect("Alice running cancellation released its port"),
1451        );
1452        drop(
1453            UdpSocket::bind(bob_addr)
1454                .await
1455                .expect("Bob running cancellation released its port"),
1456        );
1457    }
1458}
1459
1460#[cfg(feature = "dtls")]
1461const fn minimum_length(class: IngressClass) -> usize {
1462    match class {
1463        IngressClass::Stun => 20,
1464        IngressClass::Dtls => 13,
1465        IngressClass::Srtp => 12,
1466        IngressClass::Srtcp => 8,
1467    }
1468}
1469
1470pub(crate) fn lock_ingress(
1471    ingress: &Arc<StdMutex<ComponentIngress>>,
1472) -> std::sync::MutexGuard<'_, ComponentIngress> {
1473    ingress
1474        .lock()
1475        .unwrap_or_else(std::sync::PoisonError::into_inner)
1476}
1477
1478#[cfg(feature = "dtls")]
1479struct DtlsAdapter {
1480    inbound: std::sync::mpsc::Receiver<Vec<u8>>,
1481    outbound: mpsc::Sender<Vec<u8>>,
1482    timeout: Duration,
1483}
1484
1485#[cfg(feature = "dtls")]
1486impl Read for DtlsAdapter {
1487    fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
1488        let datagram = self.inbound.recv_timeout(self.timeout).map_err(|error| {
1489            let kind = match error {
1490                std::sync::mpsc::RecvTimeoutError::Timeout => std::io::ErrorKind::TimedOut,
1491                std::sync::mpsc::RecvTimeoutError::Disconnected => {
1492                    std::io::ErrorKind::UnexpectedEof
1493                }
1494            };
1495            std::io::Error::new(kind, error)
1496        })?;
1497        let destination = buffer.get_mut(..datagram.len()).ok_or_else(|| {
1498            std::io::Error::new(
1499                std::io::ErrorKind::InvalidData,
1500                "DTLS record exceeds the adapter read buffer",
1501            )
1502        })?;
1503        destination.copy_from_slice(&datagram);
1504        Ok(datagram.len())
1505    }
1506}
1507
1508#[cfg(feature = "dtls")]
1509impl Write for DtlsAdapter {
1510    fn write(&mut self, buffer: &[u8]) -> std::io::Result<usize> {
1511        if buffer.len() > MAX_OUTBOUND_DATAGRAM {
1512            return Err(std::io::Error::new(
1513                std::io::ErrorKind::InvalidData,
1514                "DTLS record exceeds the browser-component outbound bound",
1515            ));
1516        }
1517        self.outbound
1518            .blocking_send(buffer.to_vec())
1519            .map_err(|error| std::io::Error::new(std::io::ErrorKind::BrokenPipe, error))?;
1520        Ok(buffer.len())
1521    }
1522
1523    fn flush(&mut self) -> std::io::Result<()> {
1524        Ok(())
1525    }
1526}
sipx_media/browser.rs

sipx_media/
browser.rs
