Skip to content
rachel/esp32-radio
File

Blob: firmware/vendor/str0m/tests/common.rs

rust696 lines
1#![allow(unused)]
2use std::cell::Cell;
3use std::io::Cursor;
4use std::net::{Ipv4Addr, SocketAddr};
5use std::ops::{Deref, DerefMut};
6use std::sync::{Arc, Once};
7use std::time::{Duration, Instant};
8 
9use netem::{Input as NetemInput, Netem, NetemConfig, Output as NetemOutput};
10 
11use pcap_file::pcap::PcapReader;
12use str0m::Candidate;
13use str0m::change::SdpApi;
14use str0m::crypto::CryptoProvider;
15use str0m::format::Codec;
16use str0m::format::PayloadParams;
17use str0m::net::Protocol;
18use str0m::net::Receive;
19use str0m::rtp::ExtensionMap;
20use str0m::rtp::RtpHeader;
21use str0m::{Event, Input, Output, Rtc, RtcConfig, RtcError};
22use tracing::Span;
23use tracing::info_span;
24 
25/// Peer for test peers - Left or Right.
26/// Used to determine which crypto provider to use based on environment variables.
27#[derive(Debug, Clone, Copy, PartialEq, Eq)]
28pub enum Peer {
29    Left,
30    Right,
31}
32 
33impl Peer {
34    /// Create a tracing span for this peer.
35    pub fn span(&self) -> Span {
36        match self {
37            Peer::Left => info_span!("L"),
38            Peer::Right => info_span!("R"),
39        }
40    }
41 
42    /// Get the crypto provider for this peer based on environment variables.
43    /// Returns None if no environment variable is set.
44    pub fn crypto_provider(&self) -> Option<Arc<CryptoProvider>> {
45        let env_var = match self {
46            Peer::Left => "L_CRYPTO",
47            Peer::Right => "R_CRYPTO",
48        };
49 
50        if let Ok(crypto_name) = std::env::var(env_var) {
51            Some(Arc::new(get_crypto_provider_by_name(&crypto_name)))
52        } else {
53            None
54        }
55    }
56}
57 
58/// Owned version of Receive for queueing.
59#[derive(Clone)]
60pub struct PendingPacket {
61    pub proto: Protocol,
62    pub source: SocketAddr,
63    pub destination: SocketAddr,
64    pub contents: Vec<u8>,
65}
66 
67impl AsRef<[u8]> for PendingPacket {
68    fn as_ref(&self) -> &[u8] {
69        &self.contents
70    }
71}
72 
73pub struct TestRtc {
74    pub span: Span,
75    pub rtc: Rtc,
76    pub start: Instant,
77    pub last: Instant,
78    pub events: Vec<(Instant, Event)>,
79    pub pending: Netem<PendingPacket>,
80    pub forced_time_advance: Duration,
81}
82 
83impl TestRtc {
84    pub fn new(peer: Peer) -> Self {
85        let now = Instant::now();
86        let rtc = if let Some(crypto) = peer.crypto_provider() {
87            Rtc::builder().set_crypto_provider(crypto).build(now)
88        } else {
89            Rtc::new(now)
90        };
91 
92        Self::new_with_rtc(peer.span(), rtc)
93    }
94 
95    /// Like [`TestRtc::new`], but lets the caller customise the `RtcConfig`
96    /// builder — e.g. to enable a non-default codec such as H.266.
97    pub fn new_with_config(peer: Peer, f: impl FnOnce(RtcConfig) -> RtcConfig) -> Self {
98        let now = Instant::now();
99        let builder = Rtc::builder();
100        let builder = if let Some(crypto) = peer.crypto_provider() {
101            builder.set_crypto_provider(crypto)
102        } else {
103            builder
104        };
105        let rtc = f(builder).build(now);
106        Self::new_with_rtc(peer.span(), rtc)
107    }
108 
109    pub fn new_with_rtc(span: Span, rtc: Rtc) -> Self {
110        let now = Instant::now();
111        TestRtc {
112            span,
113            rtc,
114            start: now,
115            last: now,
116            events: vec![],
117            pending: Netem::new(NetemConfig::new()),
118            forced_time_advance: Duration::from_millis(10),
119        }
120    }
121 
122    /// Set the forced time advance duration when RTC returns v==rtc.last.
123    /// This prevents the test from getting stuck when RTC has no pending timeouts.
124    /// Should be set to the packet interval for the target bitrate (e.g., 0.2ms for 50 Mbps).
125    pub fn set_forced_time_advance(&mut self, duration: Duration) {
126        self.forced_time_advance = duration;
127    }
128 
129    /// Configure network emulation for incoming traffic to this RTC.
130    /// Call this on the RECEIVER to affect traffic coming TO this peer.
131    /// This preserves any packets already queued in the netem.
132    pub fn set_netem(&mut self, config: NetemConfig) {
133        self.pending.set_config(config);
134    }
135 
136    pub fn add_host_candidate(&mut self, socket: SocketAddr) -> Candidate {
137        self.rtc
138            .add_local_candidate(Candidate::host(socket, "udp").unwrap())
139            .unwrap()
140            .clone()
141    }
142 
143    pub fn duration(&self) -> Duration {
144        self.last - self.start
145    }
146 
147    pub fn params_opus(&self) -> PayloadParams {
148        self.rtc
149            .codec_config()
150            .find(|p| p.spec().codec == Codec::Opus)
151            .cloned()
152            .unwrap()
153    }
154 
155    pub fn params_vp8(&self) -> PayloadParams {
156        self.rtc
157            .codec_config()
158            .find(|p| p.spec().codec == Codec::Vp8)
159            .cloned()
160            .unwrap()
161    }
162 
163    pub fn params_vp9(&self) -> PayloadParams {
164        self.rtc
165            .codec_config()
166            .find(|p| p.spec().codec == Codec::Vp9)
167            .cloned()
168            .unwrap()
169    }
170 
171    pub fn params_h264(&self) -> PayloadParams {
172        self.rtc
173            .codec_config()
174            .find(|p| p.spec().codec == Codec::H264)
175            .cloned()
176            .unwrap()
177    }
178 
179    pub fn params_av1(&self) -> PayloadParams {
180        self.rtc
181            .codec_config()
182            .find(|p| p.spec().codec == Codec::Av1)
183            .cloned()
184            .unwrap()
185    }
186 
187    pub fn params_h265(&self) -> PayloadParams {
188        self.rtc
189            .codec_config()
190            .find(|p| p.spec().codec == Codec::H265)
191            .cloned()
192            .unwrap()
193    }
194 
195    pub fn params_h266(&self) -> PayloadParams {
196        self.rtc
197            .codec_config()
198            .find(|p| p.spec().codec == Codec::H266)
199            .cloned()
200            .unwrap()
201    }
202}
203 
204/// Progress time forward by processing the next event.
205///
206/// We have 4 event sources:
207/// - l.last: l's rtc timeout
208/// - r.last: r's rtc timeout
209/// - l.pending: packet ready to deliver to l
210/// - r.pending: packet ready to deliver to r
211///
212/// Pick the earliest, process it, then try to progress again for any
213/// more even that is within 5ms of the first time.
214pub fn progress(l: &mut TestRtc, r: &mut TestRtc) -> Result<(), RtcError> {
215    let mut first_time = None;
216 
217    loop {
218        // Find earliest event
219        let l_netem = l.pending.poll_timeout();
220        let r_netem = r.pending.poll_timeout();
221 
222        // Determine which event is next: (time, is_l, is_netem)
223        let mut next = (l.last, true, false); // default: l's rtc
224 
225        if r.last < next.0 {
226            next = (r.last, false, false);
227        }
228        if l_netem < next.0 {
229            next = (l_netem, true, true);
230        }
231        if r_netem < next.0 {
232            next = (r_netem, false, true);
233        }
234 
235        let (time, is_l, is_netem) = next;
236 
237        if let Some(first_time) = first_time {
238            // The idea is that we try to advance all the components that might be
239            // within some distance of each other.
240            let elapsed = time.saturating_duration_since(first_time);
241            if elapsed >= Duration::from_millis(5) {
242                break;
243            }
244        } else {
245            first_time = Some(time);
246        }
247 
248        progress_one(l, r, time, is_l, is_netem)?;
249    }
250 
251    Ok(())
252}
253 
254thread_local! {
255    /// When set, [`rtc_poll_to_timeout`] asserts every [`Output::Transmit`]
256    /// for which `contents.len() <= mtu`.
257    static STRICT_MTU: Cell<Option<usize>> = const { Cell::new(None) };
258}
259 
260struct StrictMtuGuard(Option<usize>);
261impl Drop for StrictMtuGuard {
262    fn drop(&mut self) {
263        STRICT_MTU.with(|c| c.set(self.0));
264    }
265}
266 
267/// Like [`progress`] but asserts every outgoing datagram is `<= mtu` bytes.
268pub fn progress_strict_mtu(l: &mut TestRtc, r: &mut TestRtc, mtu: usize) -> Result<(), RtcError> {
269    let prev = STRICT_MTU.with(|c| c.replace(Some(mtu)));
270    let _guard = StrictMtuGuard(prev);
271    progress(l, r)
272}
273 
274fn progress_one(
275    l: &mut TestRtc,
276    r: &mut TestRtc,
277    time: Instant,
278    is_l: bool,
279    is_netem: bool,
280) -> Result<(), RtcError> {
281    if is_netem {
282        // Deliver packet from netem to rtc (no timeout processing)
283        if is_l {
284            netem_to_rtc(l, time, &mut r.pending)?;
285        } else {
286            netem_to_rtc(r, time, &mut l.pending)?;
287        }
288    } else {
289        // Process rtc timeout and poll outputs
290        if is_l {
291            rtc_timeout(l, time, &mut r.pending)?;
292        } else {
293            rtc_timeout(r, time, &mut l.pending)?;
294        }
295    }
296    Ok(())
297}
298 
299/// Deliver one packet from rtc.pending to rtc. No timeout processing.
300fn netem_to_rtc(
301    rtc: &mut TestRtc,
302    time: Instant,
303    other_netem: &mut Netem<PendingPacket>,
304) -> Result<(), RtcError> {
305    rtc.pending.handle_input(NetemInput::Timeout(time));
306 
307    let Some(NetemOutput::Packet(packet)) = rtc.pending.poll_output() else {
308        return Ok(());
309    };
310 
311    let input = Input::Receive(
312        time,
313        Receive {
314            proto: packet.proto,
315            source: packet.source,
316            destination: packet.destination,
317            contents: (&packet.contents[..]).try_into()?,
318        },
319    );
320    rtc.span.in_scope(|| rtc.rtc.handle_input(input))?;
321 
322    rtc_poll_to_timeout(rtc, time, other_netem)?;
323 
324    Ok(())
325}
326 
327/// Process rtc timeout and poll until next timeout, queueing transmits in other_netem.
328fn rtc_timeout(
329    rtc: &mut TestRtc,
330    time: Instant,
331    other_netem: &mut Netem<PendingPacket>,
332) -> Result<(), RtcError> {
333    rtc.span
334        .in_scope(|| rtc.rtc.handle_input(Input::Timeout(time)))?;
335 
336    rtc_poll_to_timeout(rtc, time, other_netem)?;
337 
338    Ok(())
339}
340 
341fn rtc_poll_to_timeout(
342    rtc: &mut TestRtc,
343    time: Instant,
344    other_netem: &mut Netem<PendingPacket>,
345) -> Result<(), RtcError> {
346    loop {
347        let next = rtc.span.in_scope(|| rtc.rtc.poll_output())?;
348        // println!("next: {:?}", next);
349        match next {
350            Output::Timeout(v) => {
351                let tick = rtc.last + rtc.forced_time_advance;
352                rtc.last = if v == rtc.last { tick } else { tick.min(v) };
353                break;
354            }
355            Output::Transmit(v) => {
356                if let Some(mtu) = STRICT_MTU.with(|c| c.get()) {
357                    assert!(
358                        v.contents.len() <= mtu,
359                        "outgoing datagram {} bytes exceeds strict mtu {} (proto={:?}, {}->{})",
360                        v.contents.len(),
361                        mtu,
362                        v.proto,
363                        v.source,
364                        v.destination,
365                    );
366                }
367                let packet = PendingPacket {
368                    proto: v.proto,
369                    source: v.source,
370                    destination: v.destination,
371                    contents: v.contents.to_vec(),
372                };
373                other_netem.handle_input(NetemInput::Packet(time, packet));
374            }
375            Output::Event(v) => {
376                rtc.events.push((rtc.last, v));
377            }
378        }
379    }
380    Ok(())
381}
382 
383/// Perform a change to the session via an offer and answer.
384///
385/// The closure is passed the [`SdpApi`] for the offer side to make any changes, these are then
386/// applied locally and the offer is negotiated with the answerer.
387pub fn negotiate<F, R>(offerer: &mut TestRtc, answerer: &mut TestRtc, mut do_change: F) -> R
388where
389    F: FnMut(&mut SdpApi) -> R,
390{
391    let (offer, pending, result) = offerer.span.in_scope(|| {
392        let mut change = offerer.rtc.sdp_api();
393 
394        let result = do_change(&mut change);
395 
396        let (offer, pending) = change.apply().unwrap();
397 
398        (offer, pending, result)
399    });
400 
401    let answer = answerer
402        .span
403        .in_scope(|| answerer.rtc.sdp_api().accept_offer(offer).unwrap());
404 
405    offerer.span.in_scope(|| {
406        offerer
407            .rtc
408            .sdp_api()
409            .accept_answer(pending, answer)
410            .unwrap();
411    });
412 
413    result
414}
415 
416impl Deref for TestRtc {
417    type Target = Rtc;
418 
419    fn deref(&self) -> &Self::Target {
420        &self.rtc
421    }
422}
423 
424impl DerefMut for TestRtc {
425    fn deref_mut(&mut self) -> &mut Self::Target {
426        &mut self.rtc
427    }
428}
429 
430pub fn init_log() {
431    use tracing_subscriber::{EnvFilter, fmt, prelude::*};
432 
433    let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("off"));
434 
435    static START: Once = Once::new();
436 
437    START.call_once(|| {
438        tracing_subscriber::registry()
439            .with(fmt::layer())
440            .with(env_filter)
441            .init();
442    });
443}
444 
445pub fn init_crypto_default() {
446    str0m::crypto::from_feature_flags().install_process_default();
447}
448 
449/// Create a crypto provider from a string name.
450/// Supported names: "aws-lc-rs", "rust-crypto", "openssl", "wincrypto", "apple-crypto"
451fn get_crypto_provider_by_name(name: &str) -> CryptoProvider {
452    match name {
453        #[cfg(feature = "aws-lc-rs")]
454        "aws-lc-rs" | "aws" => str0m_aws_lc_rs::default_provider(),
455 
456        #[cfg(feature = "rust-crypto")]
457        "rust-crypto" => str0m_rust_crypto::default_provider(),
458 
459        // If both openssl and openssl-dimpl are declared, only openssl-dimpl is available.
460        #[cfg(all(feature = "openssl", not(feature = "openssl-dimpl")))]
461        "openssl" => str0m_openssl::default_provider(),
462 
463        #[cfg(feature = "openssl-dimpl")]
464        "openssl-dimpl" => str0m_openssl::default_provider(),
465 
466        #[cfg(all(feature = "wincrypto-dimpl", target_os = "windows"))]
467        "wincrypto-dimpl" => str0m_wincrypto::default_provider(),
468 
469        // If both wincrypto and wincrypto-dimpl are declared, only wincrypto-dimpl is available.
470        #[cfg(all(
471            feature = "wincrypto",
472            not(feature = "wincrypto-dimpl"),
473            target_os = "windows"
474        ))]
475        "wincrypto" => str0m_wincrypto::default_provider(),
476 
477        #[cfg(all(feature = "apple-crypto", target_vendor = "apple"))]
478        "apple-crypto" => str0m_apple_crypto::default_provider(),
479 
480        _ => {
481            let mut available: Vec<&'static str> = Vec::new();
482            #[cfg(feature = "aws-lc-rs")]
483            available.push("aws-lc-rs");
484            #[cfg(feature = "rust-crypto")]
485            available.push("rust-crypto");
486            // If both openssl and openssl-dimpl are declared, only openssl-dimpl is available.
487            #[cfg(all(feature = "openssl", not(feature = "openssl-dimpl")))]
488            available.push("openssl");
489            #[cfg(feature = "openssl-dimpl")]
490            available.push("openssl-dimpl");
491            // If both wincrypto and wincrypto-dimpl are declared, only wincrypto-dimpl is available.
492            #[cfg(all(
493                feature = "wincrypto",
494                not(feature = "wincrypto-dimpl"),
495                target_os = "windows"
496            ))]
497            available.push("wincrypto");
498            #[cfg(all(feature = "wincrypto-dimpl", target_os = "windows"))]
499            available.push("wincrypto-dimpl");
500            #[cfg(all(feature = "apple-crypto", target_vendor = "apple"))]
501            available.push("apple-crypto");
502 
503            panic!(
504                "Unknown or unavailable crypto provider '{}'. Available providers: [{}]",
505                name,
506                available.join(", ")
507            )
508        }
509    }
510}
511 
512pub fn connect_l_r() -> (TestRtc, TestRtc) {
513    let mut rtc1_builder = Rtc::builder().set_rtp_mode(true).enable_raw_packets(true);
514 
515    if let Some(crypto) = Peer::Left.crypto_provider() {
516        rtc1_builder = rtc1_builder.set_crypto_provider(crypto);
517    }
518 
519    let mut rtc2_builder = Rtc::builder().set_rtp_mode(true).enable_raw_packets(true);
520 
521    if let Some(crypto) = Peer::Right.crypto_provider() {
522        rtc2_builder = rtc2_builder.set_crypto_provider(crypto);
523    }
524 
525    let now = Instant::now();
526    connect_l_r_with_rtc(rtc1_builder.build(now), rtc2_builder.build(now))
527}
528 
529pub fn connect_l_r_with_rtc(rtc1: Rtc, rtc2: Rtc) -> (TestRtc, TestRtc) {
530    let mut l = TestRtc::new_with_rtc(info_span!("L"), rtc1);
531    let mut r = TestRtc::new_with_rtc(info_span!("R"), rtc2);
532 
533    let host1 = Candidate::host((Ipv4Addr::new(1, 1, 1, 1), 1000).into(), "udp").unwrap();
534    let host2 = Candidate::host((Ipv4Addr::new(2, 2, 2, 2), 2000).into(), "udp").unwrap();
535    l.add_local_candidate(host1.clone());
536    l.add_remote_candidate(host2.clone());
537    r.add_local_candidate(host2);
538    r.add_remote_candidate(host1);
539 
540    let finger_l = l.direct_api().local_dtls_fingerprint().clone();
541    let finger_r = r.direct_api().local_dtls_fingerprint().clone();
542 
543    l.direct_api().set_remote_fingerprint(finger_r);
544    r.direct_api().set_remote_fingerprint(finger_l);
545 
546    let creds_l = l.direct_api().local_ice_credentials();
547    let creds_r = r.direct_api().local_ice_credentials();
548 
549    l.direct_api().set_remote_ice_credentials(creds_r);
550    r.direct_api().set_remote_ice_credentials(creds_l);
551 
552    l.direct_api().set_ice_controlling(true);
553    r.direct_api().set_ice_controlling(false);
554 
555    l.direct_api().start_dtls(true).unwrap();
556    r.direct_api().start_dtls(false).unwrap();
557 
558    l.direct_api().start_sctp(true);
559    r.direct_api().start_sctp(false);
560 
561    loop {
562        if l.is_connected() || r.is_connected() {
563            break;
564        }
565        progress(&mut l, &mut r).expect("clean progress");
566    }
567 
568    (l, r)
569}
570 
571pub type PcapData = Vec<(Duration, RtpHeader, Vec<u8>)>;
572 
573pub fn vp8_data() -> PcapData {
574    load_pcap_data(include_bytes!("data/vp8.pcap"))
575}
576 
577pub fn vp9_contiguous_data() -> PcapData {
578    load_pcap_data(include_bytes!("data/contiguous_vp9.pcap"))
579}
580 
581pub fn vp9_data() -> PcapData {
582    load_pcap_data(include_bytes!("data/vp9.pcap"))
583}
584 
585pub fn h264_data() -> PcapData {
586    load_pcap_data(include_bytes!("data/h264.pcap"))
587}
588 
589pub fn av1_data() -> PcapData {
590    load_pcap_data(include_bytes!("data/av1.pcap"))
591}
592 
593pub fn h265_data() -> PcapData {
594    load_pcap_data(include_bytes!("data/h265.pcap"))
595}
596 
597pub fn h266_data() -> PcapData {
598    load_pcap_data(include_bytes!("data/h266.pcap"))
599}
600 
601// ---------------------------------------------------------------------------
602// SNAP (SCTP Negotiation Acceleration Protocol) test helpers
603// ---------------------------------------------------------------------------
604 
605/// Extract the `a=sctp-init:<value>` from an SDP string, if present.
606pub fn extract_sctp_init(sdp: &str) -> Option<String> {
607    sdp.lines()
608        .find(|l| l.starts_with("a=sctp-init:"))
609        .map(|l| l.trim_start_matches("a=sctp-init:").to_string())
610}
611 
612/// Replace the `a=sctp-init:` attribute value in an SDP string.
613///
614/// Preserves original line endings (`\r\n`).
615pub fn replace_sctp_init(sdp: &str, replacement: &str) -> String {
616    let mut result = String::with_capacity(sdp.len());
617    // split("\r\n") produces an extra empty element after a trailing CRLF;
618    // skip it to avoid appending a spurious double-CRLF at the end.
619    for line in sdp.split("\r\n") {
620        if line.is_empty() && result.ends_with("\r\n") {
621            continue;
622        }
623        if line.starts_with("a=sctp-init:") {
624            result.push_str(&format!("a=sctp-init:{replacement}\r\n"));
625        } else {
626            result.push_str(line);
627            result.push_str("\r\n");
628        }
629    }
630    result
631}
632 
633/// Remove the `a=sctp-init:` attribute from an SDP string.
634///
635/// Preserves original line endings (`\r\n`).
636pub fn remove_sctp_init(sdp: &str) -> String {
637    let mut result = String::with_capacity(sdp.len());
638    // split("\r\n") produces an extra empty element after a trailing CRLF;
639    // skip it to avoid appending a spurious double-CRLF at the end.
640    for line in sdp.split("\r\n") {
641        if line.is_empty() && result.ends_with("\r\n") {
642            continue;
643        }
644        if !line.starts_with("a=sctp-init:") {
645            result.push_str(line);
646            result.push_str("\r\n");
647        }
648    }
649    result
650}
651 
652/// Generate SNAP init data for out-of-band SCTP negotiation.
653///
654/// Returns `Some((local_init_bytes, SctpInitData))` when `use_snap` is true, `None` otherwise.
655pub fn snap_init_data(use_snap: bool) -> Option<(Vec<u8>, str0m::channel::SctpInitData)> {
656    if use_snap {
657        let mut data = str0m::channel::SctpInitData::new();
658        let local_init = data
659            .local_init_chunk()
660            .expect("generate_snap_token should not fail");
661        Some((local_init, data))
662    } else {
663        None
664    }
665}
666 
667pub fn load_pcap_data(data: &[u8]) -> PcapData {
668    let reader = Cursor::new(data);
669    let mut r = PcapReader::new(reader).expect("pcap reader");
670 
671    let exts = ExtensionMap::standard();
672 
673    let mut ret = vec![];
674 
675    let mut first = None;
676 
677    while let Some(pkt) = r.next_packet() {
678        let pkt = pkt.unwrap();
679 
680        if first.is_none() {
681            first = Some(pkt.timestamp);
682        }
683        let relative_time = pkt.timestamp - first.unwrap();
684 
685        // This magic number 42 is the ethernet/IP/UDP framing of the packet.
686        let rtp_data = &pkt.data[42..];
687 
688        let header = RtpHeader::_parse(rtp_data, &exts).unwrap();
689        let payload = &rtp_data[header.header_len..];
690 
691        ret.push((relative_time, header, payload.to_vec()));
692    }
693 
694    ret
695}