1
//! Real UDP/DTLS/SCTP tests without credentials, Cloudflare, or hardware.
2
use radio_core::signaling::ChannelIds;
3
use radio_webrtc::{Certificate, Peer, PeerState, SendOutcome, Stream};
4
use std::{
5
    net::{Ipv4Addr, UdpSocket},
6
    sync::Arc,
7
    thread,
8
    time::{Duration, Instant},
9
};
10
use str0m::{
11
    Candidate, Event, Input, Output, Rtc,
12
    change::SdpOffer,
13
    channel::{ChannelConfig, ChannelId, Reliability},
14
    crypto::dtls::DtlsCert,
15
    net::{Protocol, Receive},
16
};
17
 
18
const CERT: &[u8] = include_bytes!("fixtures/certificate.der");
19
const KEY: &[u8] = include_bytes!("fixtures/key.der");
20
 
21
struct Listener {
22
    rtc: Rtc,
23
    socket: UdpSocket,
24
    messages: Vec<Vec<u8>>,
25
    audio: usize,
26
    drop_data: bool,
27
    data_packets: usize,
28
}
29
 
30
impl Listener {
31
    fn new(offer: &str) -> (Self, String) {
32
        let socket = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).unwrap();
33
        socket.set_nonblocking(true).unwrap();
34
        let mut rtc = Rtc::builder()
35
            .set_crypto_provider(Arc::new(str0m_rust_crypto::default_provider()))
36
            .set_dtls_cert(DtlsCert {
37
                certificate: CERT.to_vec(),
38
                private_key: KEY.to_vec(),
39
            })
40
            .clear_codecs()
41
            .enable_opus(true)
42
            .build(Instant::now());
43
        rtc.add_local_candidate(Candidate::host(socket.local_addr().unwrap(), "udp").unwrap());
44
        let mut this = Self {
45
            rtc,
46
            socket,
47
            messages: Vec::new(),
48
            audio: 0,
49
            drop_data: false,
50
            data_packets: 0,
51
        };
52
        this.drain();
53
        let answer = this
54
            .rtc
55
            .sdp_api()
56
            .accept_offer(SdpOffer::from_sdp_string(offer).unwrap())
57
            .unwrap();
58
        this.drain();
59
        this.channel(0, true);
60
        (this, answer.to_sdp_string())
61
    }
62
 
63
    fn channel(&mut self, id: u16, reliable: bool) -> ChannelId {
64
        let id = self.rtc.direct_api().create_data_channel(ChannelConfig {
65
            label: format!("test-{id}"),
66
            negotiated: Some(id),
67
            ordered: reliable,
68
            reliability: if reliable {
69
                Reliability::Reliable
70
            } else {
71
                Reliability::MaxRetransmits { retransmits: 0 }
72
            },
73
            ..Default::default()
74
        });
75
        self.drain();
76
        id
77
    }
78
 
79
    fn drain(&mut self) {
80
        loop {
81
            match self.rtc.poll_output().unwrap() {
82
                Output::Timeout(_) => return,
83
                Output::Transmit(packet) => {
84
                    if packet.contents.first() == Some(&23) {
85
                        self.data_packets += 1;
86
                        if self.drop_data && self.data_packets % 5 == 0 {
87
                            continue;
88
                        }
89
                    }
90
                    self.socket
91
                        .send_to(&packet.contents, packet.destination)
92
                        .unwrap();
93
                }
94
                Output::Event(Event::ChannelData(data)) => self.messages.push(data.data),
95
                Output::Event(Event::MediaData(data)) => {
96
                    assert_eq!(data.data.as_ref(), radio_core::protocol::SILENCE);
97
                    self.audio += 1;
98
                }
99
                _ => {}
100
            }
101
        }
102
    }
103
 
104
    fn poll(&mut self) {
105
        let mut buffer = [0; 2048];
106
        while let Ok((n, source)) = self.socket.recv_from(&mut buffer) {
107
            let receive = Receive {
108
                proto: Protocol::Udp,
109
                source,
110
                destination: self.socket.local_addr().unwrap(),
111
                contents: buffer[..n].try_into().unwrap(),
112
            };
113
            self.rtc
114
                .handle_input(Input::Receive(Instant::now(), receive))
115
                .unwrap();
116
            self.drain();
117
        }
118
        self.rtc
119
            .handle_input(Input::Timeout(Instant::now()))
120
            .unwrap();
121
        self.drain();
122
    }
123
}
124
 
125
fn connected_pair() -> (Peer, Listener) {
126
    let certificate = Certificate::from_der(CERT.to_vec(), KEY.to_vec()).unwrap();
127
    let (mut peer, offer) = Peer::open(
128
        Ipv4Addr::LOCALHOST,
129
        certificate,
130
        Arc::new(str0m_rust_crypto::default_provider()),
131
    )
132
    .unwrap();
133
    let (mut listener, answer) = Listener::new(&offer);
134
    peer.answer(&answer).unwrap();
135
    let deadline = Instant::now() + Duration::from_secs(10);
136
    while peer.state() != PeerState::Connected {
137
        assert!(Instant::now() < deadline, "SCTP connection timed out");
138
        peer.poll().unwrap();
139
        listener.poll();
140
        thread::sleep(Duration::from_millis(1));
141
    }
142
    (peer, listener)
143
}
144
 
145
#[test]
146
fn negotiated_ids_audio_commands_loss_and_teardown() {
147
    for _ in 0..2 {
148
        let (mut peer, mut listener) = connected_pair();
149
        let ids = ChannelIds::new(6, 10).unwrap();
150
        peer.create_channels(ids).unwrap();
151
        assert!(peer.create_channels(ids).is_err());
152
        let robot = listener.channel(6, true);
153
        listener.channel(10, false);
154
        listener.drop_data = true;
155
        let mut commands = Vec::new();
156
        let mut sent = 0;
157
        let mut audio_sent = 0;
158
        let mut next_audio = Instant::now();
159
        let deadline = Instant::now() + Duration::from_secs(10);
160
        while commands.len() < 20 || listener.audio < 5 || listener.messages.len() < 20 {
161
            assert!(Instant::now() < deadline, "media/data delivery timed out");
162
            peer.poll().unwrap();
163
            listener.poll();
164
            if sent < 20
165
                && let Some(mut channel) = listener.rtc.channel(robot)
166
            {
167
                assert!(channel.write(false, &[sent]).unwrap());
168
                listener.drain();
169
                assert_eq!(
170
                    peer.data(Stream::Robot, &[sent]).unwrap(),
171
                    SendOutcome::Sent
172
                );
173
                assert_eq!(
174
                    peer.data(Stream::Spectrum, &[sent]).unwrap(),
175
                    SendOutcome::Sent
176
                );
177
                sent += 1;
178
            }
179
            if Instant::now() >= next_audio {
180
                peer.audio(audio_sent * 20, radio_core::protocol::SILENCE)
181
                    .unwrap();
182
                audio_sent += 1;
183
                next_audio += Duration::from_millis(20);
184
            }
185
            let mut out = [0; 512];
186
            if peer.command(&mut out).unwrap() > 0 {
187
                commands.push(out[0]);
188
            }
189
            thread::sleep(Duration::from_millis(1));
190
        }
191
        assert_eq!(commands, (0..20).collect::<Vec<_>>());
192
        assert_eq!(peer.dropped(), 0);
193
        peer.close().unwrap();
194
        let deadline = Instant::now() + Duration::from_secs(3);
195
        while peer.state() != PeerState::Lost {
196
            assert!(Instant::now() < deadline, "local close timed out");
197
            peer.poll().unwrap();
198
            listener.poll();
199
        }
200
        drop(peer);
201
    }
202
}
203
 
204
#[test]
205
fn data_backpressure_is_recoverable_but_unavailable_and_closed_channels_are_errors() {
206
    let (mut peer, mut listener) = connected_pair();
207
    assert!(peer.data(Stream::Robot, b"before channels").is_err());
208
 
209
    let ids = ChannelIds::new(6, 10).unwrap();
210
    peer.create_channels(ids).unwrap();
211
    let robot = listener.channel(6, true);
212
    listener.channel(10, false);
213
 
214
    let payload = [0x42; 2048];
215
    // The application performs this normal protocol drive after handling its
216
    // Start request and before entering the playback/send portion of the loop.
217
    peer.poll().unwrap();
218
    listener.poll();
219
    assert_eq!(
220
        peer.data(Stream::Robot, &payload).unwrap(),
221
        SendOutcome::Sent
222
    );
223
    let mut accepted = 1;
224
    while let SendOutcome::Sent = peer.data(Stream::Robot, &payload).unwrap() {
225
        accepted += 1;
226
        assert!(
227
            accepted < 128,
228
            "reliable channel never applied backpressure"
229
        );
230
    }
231
    assert!(accepted > 0);
232
    assert_eq!(peer.state(), PeerState::Connected);
233
 
234
    let deadline = Instant::now() + Duration::from_secs(10);
235
    loop {
236
        assert!(
237
            Instant::now() < deadline,
238
            "data channel stayed backpressured"
239
        );
240
        peer.poll().unwrap();
241
        listener.poll();
242
        match peer.data(Stream::Robot, b"after backpressure").unwrap() {
243
            SendOutcome::Sent => break,
244
            SendOutcome::Backpressured => thread::sleep(Duration::from_millis(1)),
245
        }
246
    }
247
    assert_eq!(peer.state(), PeerState::Connected);
248
 
249
    listener.rtc.direct_api().close_data_channel(robot);
250
    listener.drain();
251
    let deadline = Instant::now() + Duration::from_secs(10);
252
    while peer.state() != PeerState::Lost {
253
        assert!(Instant::now() < deadline, "remote channel close timed out");
254
        peer.poll().unwrap();
255
        listener.poll();
256
        thread::sleep(Duration::from_millis(1));
257
    }
258
    assert!(peer.data(Stream::Robot, b"after close").is_err());
259
}
260
 
261
#[test]
262
fn an_answer_with_the_wrong_certificate_fingerprint_is_rejected() {
263
    let certificate = Certificate::from_der(CERT.to_vec(), KEY.to_vec()).unwrap();
264
    let (mut peer, offer) = Peer::open(
265
        Ipv4Addr::LOCALHOST,
266
        certificate,
267
        Arc::new(str0m_rust_crypto::default_provider()),
268
    )
269
    .unwrap();
270
    let (mut listener, answer) = Listener::new(&offer);
271
    let tampered = answer
272
        .lines()
273
        .map(|line| {
274
            if line.starts_with("a=fingerprint:sha-256 ") {
275
                format!("a=fingerprint:sha-256 {}", ["00"; 32].join(":"))
276
            } else {
277
                line.to_owned()
278
            }
279
        })
280
        .collect::<Vec<_>>()
281
        .join("\r\n")
282
        + "\r\n";
283
    assert_ne!(answer, tampered);
284
    peer.answer(&tampered).unwrap();
285
    let deadline = Instant::now() + Duration::from_secs(3);
286
    let mut rejected = false;
287
    while Instant::now() < deadline {
288
        listener.poll();
289
        if peer.poll().is_err() || peer.state() == PeerState::Lost {
290
            rejected = true;
291
            break;
292
        }
293
        assert_ne!(peer.state(), PeerState::Connected);
294
        thread::sleep(Duration::from_millis(1));
295
    }
296
    assert!(rejected, "fingerprint mismatch must fail the handshake");
297
}