1 use arc_swap::ArcSwapOption;
2 use std::collections::HashMap;
3 use std::future::Future;
4 use std::pin::Pin;
5 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
6 use std::sync::Arc;
7
8 use bytes::Bytes;
9 use dtls::config::ClientAuthType;
10 use dtls::conn::DTLSConn;
11 use dtls::extension::extension_use_srtp::SrtpProtectionProfile;
12 use interceptor::stream_info::StreamInfo;
13 use interceptor::{Interceptor, RTCPReader, RTPReader};
14 use sha2::{Digest, Sha256};
15 use srtp::protection_profile::ProtectionProfile;
16 use srtp::session::Session;
17 use srtp::stream::Stream;
18 use tokio::sync::{mpsc, Mutex};
19 use util::Conn;
20
21 use dtls_role::*;
22
23 use crate::api::setting_engine::SettingEngine;
24 use crate::dtls_transport::dtls_parameters::DTLSParameters;
25 use crate::dtls_transport::dtls_transport_state::RTCDtlsTransportState;
26 use crate::error::{flatten_errs, Error, Result};
27 use crate::ice_transport::ice_role::RTCIceRole;
28 use crate::ice_transport::ice_transport_state::RTCIceTransportState;
29 use crate::ice_transport::RTCIceTransport;
30 use crate::mux::endpoint::Endpoint;
31 use crate::mux::mux_func::{match_dtls, match_srtcp, match_srtp, MatchFunc};
32 use crate::peer_connection::certificate::RTCCertificate;
33 use crate::rtp_transceiver::SSRC;
34 use crate::stats::stats_collector::StatsCollector;
35
36 #[cfg(test)]
37 mod dtls_transport_test;
38
39 pub mod dtls_fingerprint;
40 pub mod dtls_parameters;
41 pub mod dtls_role;
42 pub mod dtls_transport_state;
43
default_srtp_protection_profiles() -> Vec<SrtpProtectionProfile>44 pub(crate) fn default_srtp_protection_profiles() -> Vec<SrtpProtectionProfile> {
45 vec![
46 SrtpProtectionProfile::Srtp_Aead_Aes_128_Gcm,
47 SrtpProtectionProfile::Srtp_Aes128_Cm_Hmac_Sha1_80,
48 ]
49 }
50
51 pub type OnDTLSTransportStateChangeHdlrFn = Box<
52 dyn (FnMut(RTCDtlsTransportState) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>)
53 + Send
54 + Sync,
55 >;
56
57 /// DTLSTransport allows an application access to information about the DTLS
58 /// transport over which RTP and RTCP packets are sent and received by
59 /// RTPSender and RTPReceiver, as well other data such as SCTP packets sent
60 /// and received by data channels.
61 #[derive(Default)]
62 pub struct RTCDtlsTransport {
63 pub(crate) ice_transport: Arc<RTCIceTransport>,
64 pub(crate) certificates: Vec<RTCCertificate>,
65 pub(crate) setting_engine: Arc<SettingEngine>,
66
67 pub(crate) remote_parameters: Mutex<DTLSParameters>,
68 pub(crate) remote_certificate: Mutex<Bytes>,
69 pub(crate) state: AtomicU8, //DTLSTransportState,
70 pub(crate) srtp_protection_profile: Mutex<ProtectionProfile>,
71 pub(crate) on_state_change_handler: ArcSwapOption<Mutex<OnDTLSTransportStateChangeHdlrFn>>,
72 pub(crate) conn: Mutex<Option<Arc<DTLSConn>>>,
73
74 pub(crate) srtp_session: Mutex<Option<Arc<Session>>>,
75 pub(crate) srtcp_session: Mutex<Option<Arc<Session>>>,
76 pub(crate) srtp_endpoint: Mutex<Option<Arc<Endpoint>>>,
77 pub(crate) srtcp_endpoint: Mutex<Option<Arc<Endpoint>>>,
78
79 pub(crate) simulcast_streams: Mutex<HashMap<SSRC, Arc<Stream>>>,
80
81 pub(crate) srtp_ready_signal: Arc<AtomicBool>,
82 pub(crate) srtp_ready_tx: Mutex<Option<mpsc::Sender<()>>>,
83 pub(crate) srtp_ready_rx: Mutex<Option<mpsc::Receiver<()>>>,
84
85 pub(crate) dtls_matcher: Option<MatchFunc>,
86 }
87
88 impl RTCDtlsTransport {
new( ice_transport: Arc<RTCIceTransport>, certificates: Vec<RTCCertificate>, setting_engine: Arc<SettingEngine>, ) -> Self89 pub(crate) fn new(
90 ice_transport: Arc<RTCIceTransport>,
91 certificates: Vec<RTCCertificate>,
92 setting_engine: Arc<SettingEngine>,
93 ) -> Self {
94 let (srtp_ready_tx, srtp_ready_rx) = mpsc::channel(1);
95 RTCDtlsTransport {
96 ice_transport,
97 certificates,
98 setting_engine,
99 srtp_ready_signal: Arc::new(AtomicBool::new(false)),
100 srtp_ready_tx: Mutex::new(Some(srtp_ready_tx)),
101 srtp_ready_rx: Mutex::new(Some(srtp_ready_rx)),
102 state: AtomicU8::new(RTCDtlsTransportState::New as u8),
103 dtls_matcher: Some(Box::new(match_dtls)),
104 ..Default::default()
105 }
106 }
107
conn(&self) -> Option<Arc<DTLSConn>>108 pub(crate) async fn conn(&self) -> Option<Arc<DTLSConn>> {
109 let conn = self.conn.lock().await;
110 conn.clone()
111 }
112
113 /// returns the currently-configured ICETransport or None
114 /// if one has not been configured
ice_transport(&self) -> &RTCIceTransport115 pub fn ice_transport(&self) -> &RTCIceTransport {
116 &self.ice_transport
117 }
118
119 /// state_change requires the caller holds the lock
state_change(&self, state: RTCDtlsTransportState)120 async fn state_change(&self, state: RTCDtlsTransportState) {
121 self.state.store(state as u8, Ordering::SeqCst);
122 if let Some(handler) = &*self.on_state_change_handler.load() {
123 let mut f = handler.lock().await;
124 f(state).await;
125 }
126 }
127
128 /// on_state_change sets a handler that is fired when the DTLS
129 /// connection state changes.
on_state_change(&self, f: OnDTLSTransportStateChangeHdlrFn)130 pub fn on_state_change(&self, f: OnDTLSTransportStateChangeHdlrFn) {
131 self.on_state_change_handler
132 .store(Some(Arc::new(Mutex::new(f))));
133 }
134
135 /// state returns the current dtls_transport transport state.
state(&self) -> RTCDtlsTransportState136 pub fn state(&self) -> RTCDtlsTransportState {
137 self.state.load(Ordering::SeqCst).into()
138 }
139
140 /// write_rtcp sends a user provided RTCP packet to the connected peer. If no peer is connected the
141 /// packet is discarded.
write_rtcp( &self, pkts: &[Box<dyn rtcp::packet::Packet + Send + Sync>], ) -> Result<usize>142 pub async fn write_rtcp(
143 &self,
144 pkts: &[Box<dyn rtcp::packet::Packet + Send + Sync>],
145 ) -> Result<usize> {
146 let srtcp_session = self.srtcp_session.lock().await;
147 if let Some(srtcp_session) = &*srtcp_session {
148 let raw = rtcp::packet::marshal(pkts)?;
149 Ok(srtcp_session.write(&raw, false).await?)
150 } else {
151 Ok(0)
152 }
153 }
154
155 /// get_local_parameters returns the DTLS parameters of the local DTLSTransport upon construction.
get_local_parameters(&self) -> Result<DTLSParameters>156 pub fn get_local_parameters(&self) -> Result<DTLSParameters> {
157 let mut fingerprints = vec![];
158
159 for c in &self.certificates {
160 fingerprints.extend(c.get_fingerprints());
161 }
162
163 Ok(DTLSParameters {
164 role: DTLSRole::Auto, // always returns the default role
165 fingerprints,
166 })
167 }
168
169 /// get_remote_certificate returns the certificate chain in use by the remote side
170 /// returns an empty list prior to selection of the remote certificate
get_remote_certificate(&self) -> Bytes171 pub async fn get_remote_certificate(&self) -> Bytes {
172 let remote_certificate = self.remote_certificate.lock().await;
173 remote_certificate.clone()
174 }
175
start_srtp(&self) -> Result<()>176 pub(crate) async fn start_srtp(&self) -> Result<()> {
177 let profile = {
178 let srtp_protection_profile = self.srtp_protection_profile.lock().await;
179 *srtp_protection_profile
180 };
181
182 let mut srtp_config = srtp::config::Config {
183 profile,
184 ..Default::default()
185 };
186
187 if self.setting_engine.replay_protection.srtp != 0 {
188 srtp_config.remote_rtp_options = Some(srtp::option::srtp_replay_protection(
189 self.setting_engine.replay_protection.srtp,
190 ));
191 } else if self.setting_engine.disable_srtp_replay_protection {
192 srtp_config.remote_rtp_options = Some(srtp::option::srtp_no_replay_protection());
193 }
194
195 if let Some(conn) = self.conn().await {
196 let conn_state = conn.connection_state().await;
197 srtp_config
198 .extract_session_keys_from_dtls(conn_state, self.role().await == DTLSRole::Client)
199 .await?;
200 } else {
201 return Err(Error::ErrDtlsTransportNotStarted);
202 }
203
204 {
205 let mut srtp_session = self.srtp_session.lock().await;
206 *srtp_session = {
207 let se = self.srtp_endpoint.lock().await;
208 if let Some(srtp_endpoint) = &*se {
209 Some(Arc::new(
210 Session::new(
211 Arc::clone(srtp_endpoint) as Arc<dyn Conn + Send + Sync>,
212 srtp_config,
213 true,
214 )
215 .await?,
216 ))
217 } else {
218 None
219 }
220 };
221 }
222
223 let mut srtcp_config = srtp::config::Config {
224 profile,
225 ..Default::default()
226 };
227 if self.setting_engine.replay_protection.srtcp != 0 {
228 srtcp_config.remote_rtcp_options = Some(srtp::option::srtcp_replay_protection(
229 self.setting_engine.replay_protection.srtcp,
230 ));
231 } else if self.setting_engine.disable_srtcp_replay_protection {
232 srtcp_config.remote_rtcp_options = Some(srtp::option::srtcp_no_replay_protection());
233 }
234
235 if let Some(conn) = self.conn().await {
236 let conn_state = conn.connection_state().await;
237 srtcp_config
238 .extract_session_keys_from_dtls(conn_state, self.role().await == DTLSRole::Client)
239 .await?;
240 } else {
241 return Err(Error::ErrDtlsTransportNotStarted);
242 }
243
244 {
245 let mut srtcp_session = self.srtcp_session.lock().await;
246 *srtcp_session = {
247 let se = self.srtcp_endpoint.lock().await;
248 if let Some(srtcp_endpoint) = &*se {
249 Some(Arc::new(
250 Session::new(
251 Arc::clone(srtcp_endpoint) as Arc<dyn Conn + Send + Sync>,
252 srtcp_config,
253 false,
254 )
255 .await?,
256 ))
257 } else {
258 None
259 }
260 };
261 }
262
263 {
264 let mut srtp_ready_tx = self.srtp_ready_tx.lock().await;
265 srtp_ready_tx.take();
266 if srtp_ready_tx.is_none() {
267 self.srtp_ready_signal.store(true, Ordering::SeqCst);
268 }
269 }
270
271 Ok(())
272 }
273
get_srtp_session(&self) -> Option<Arc<Session>>274 pub(crate) async fn get_srtp_session(&self) -> Option<Arc<Session>> {
275 let srtp_session = self.srtp_session.lock().await;
276 srtp_session.clone()
277 }
278
get_srtcp_session(&self) -> Option<Arc<Session>>279 pub(crate) async fn get_srtcp_session(&self) -> Option<Arc<Session>> {
280 let srtcp_session = self.srtcp_session.lock().await;
281 srtcp_session.clone()
282 }
283
role(&self) -> DTLSRole284 pub(crate) async fn role(&self) -> DTLSRole {
285 // If remote has an explicit role use the inverse
286 {
287 let remote_parameters = self.remote_parameters.lock().await;
288 match remote_parameters.role {
289 DTLSRole::Client => return DTLSRole::Server,
290 DTLSRole::Server => return DTLSRole::Client,
291 _ => {}
292 };
293 }
294
295 // If SettingEngine has an explicit role
296 match self.setting_engine.answering_dtls_role {
297 DTLSRole::Server => return DTLSRole::Server,
298 DTLSRole::Client => return DTLSRole::Client,
299 _ => {}
300 };
301
302 // Remote was auto and no explicit role was configured via SettingEngine
303 if self.ice_transport.role().await == RTCIceRole::Controlling {
304 return DTLSRole::Server;
305 }
306
307 DEFAULT_DTLS_ROLE_ANSWER
308 }
309
collect_stats(&self, collector: &StatsCollector)310 pub(crate) async fn collect_stats(&self, collector: &StatsCollector) {
311 for cert in &self.certificates {
312 cert.collect_stats(collector).await;
313 }
314 }
315
prepare_transport( &self, remote_parameters: DTLSParameters, ) -> Result<(DTLSRole, dtls::config::Config)>316 async fn prepare_transport(
317 &self,
318 remote_parameters: DTLSParameters,
319 ) -> Result<(DTLSRole, dtls::config::Config)> {
320 self.ensure_ice_conn()?;
321
322 if self.state() != RTCDtlsTransportState::New {
323 return Err(Error::ErrInvalidDTLSStart);
324 }
325
326 {
327 let mut srtp_endpoint = self.srtp_endpoint.lock().await;
328 *srtp_endpoint = self.ice_transport.new_endpoint(Box::new(match_srtp)).await;
329 }
330 {
331 let mut srtcp_endpoint = self.srtcp_endpoint.lock().await;
332 *srtcp_endpoint = self.ice_transport.new_endpoint(Box::new(match_srtcp)).await;
333 }
334 {
335 let mut rp = self.remote_parameters.lock().await;
336 *rp = remote_parameters;
337 }
338
339 let certificate = if let Some(cert) = self.certificates.first() {
340 cert.dtls_certificate.clone()
341 } else {
342 return Err(Error::ErrNonCertificate);
343 };
344 self.state_change(RTCDtlsTransportState::Connecting).await;
345
346 Ok((
347 self.role().await,
348 dtls::config::Config {
349 certificates: vec![certificate],
350 srtp_protection_profiles: if !self
351 .setting_engine
352 .srtp_protection_profiles
353 .is_empty()
354 {
355 self.setting_engine.srtp_protection_profiles.clone()
356 } else {
357 default_srtp_protection_profiles()
358 },
359 client_auth: ClientAuthType::RequireAnyClientCert,
360 insecure_skip_verify: true,
361 insecure_verification: self.setting_engine.allow_insecure_verification_algorithm,
362 ..Default::default()
363 },
364 ))
365 }
366
367 /// start DTLS transport negotiation with the parameters of the remote DTLS transport
start(&self, remote_parameters: DTLSParameters) -> Result<()>368 pub async fn start(&self, remote_parameters: DTLSParameters) -> Result<()> {
369 let dtls_conn_result = if let Some(dtls_endpoint) =
370 self.ice_transport.new_endpoint(Box::new(match_dtls)).await
371 {
372 let (role, mut dtls_config) = self.prepare_transport(remote_parameters).await?;
373 if self.setting_engine.replay_protection.dtls != 0 {
374 dtls_config.replay_protection_window = self.setting_engine.replay_protection.dtls;
375 }
376
377 // Connect as DTLS Client/Server, function is blocking and we
378 // must not hold the DTLSTransport lock
379 if role == DTLSRole::Client {
380 dtls::conn::DTLSConn::new(
381 dtls_endpoint as Arc<dyn Conn + Send + Sync>,
382 dtls_config,
383 true,
384 None,
385 )
386 .await
387 } else {
388 dtls::conn::DTLSConn::new(
389 dtls_endpoint as Arc<dyn Conn + Send + Sync>,
390 dtls_config,
391 false,
392 None,
393 )
394 .await
395 }
396 } else {
397 Err(dtls::Error::Other(
398 "ice_transport.new_endpoint failed".to_owned(),
399 ))
400 };
401
402 let dtls_conn = match dtls_conn_result {
403 Ok(dtls_conn) => dtls_conn,
404 Err(err) => {
405 self.state_change(RTCDtlsTransportState::Failed).await;
406 return Err(err.into());
407 }
408 };
409
410 let srtp_profile = dtls_conn.selected_srtpprotection_profile();
411 {
412 let mut srtp_protection_profile = self.srtp_protection_profile.lock().await;
413 *srtp_protection_profile = match srtp_profile {
414 dtls::extension::extension_use_srtp::SrtpProtectionProfile::Srtp_Aead_Aes_128_Gcm => {
415 srtp::protection_profile::ProtectionProfile::AeadAes128Gcm
416 }
417 dtls::extension::extension_use_srtp::SrtpProtectionProfile::Srtp_Aes128_Cm_Hmac_Sha1_80 => {
418 srtp::protection_profile::ProtectionProfile::Aes128CmHmacSha1_80
419 }
420 _ => {
421 if let Err(err) = dtls_conn.close().await {
422 log::error!("{}", err);
423 }
424
425 self.state_change(RTCDtlsTransportState::Failed).await;
426 return Err(Error::ErrNoSRTPProtectionProfile);
427 }
428 };
429 }
430
431 // Check the fingerprint if a certificate was exchanged
432 let remote_certs = &dtls_conn.connection_state().await.peer_certificates;
433 if remote_certs.is_empty() {
434 if let Err(err) = dtls_conn.close().await {
435 log::error!("{}", err);
436 }
437
438 self.state_change(RTCDtlsTransportState::Failed).await;
439 return Err(Error::ErrNoRemoteCertificate);
440 }
441
442 {
443 let mut remote_certificate = self.remote_certificate.lock().await;
444 *remote_certificate = Bytes::from(remote_certs[0].clone());
445 }
446
447 if !self
448 .setting_engine
449 .disable_certificate_fingerprint_verification
450 {
451 if let Err(err) = self.validate_fingerprint(&remote_certs[0]).await {
452 if let Err(close_err) = dtls_conn.close().await {
453 log::error!("{}", close_err);
454 }
455
456 self.state_change(RTCDtlsTransportState::Failed).await;
457 return Err(err);
458 }
459 }
460
461 {
462 let mut conn = self.conn.lock().await;
463 *conn = Some(Arc::new(dtls_conn));
464 }
465 self.state_change(RTCDtlsTransportState::Connected).await;
466
467 self.start_srtp().await
468 }
469
470 /// stops and closes the DTLSTransport object.
stop(&self) -> Result<()>471 pub async fn stop(&self) -> Result<()> {
472 // Try closing everything and collect the errors
473 let mut close_errs: Vec<Error> = vec![];
474 {
475 let srtp_session = {
476 let mut srtp_session = self.srtp_session.lock().await;
477 srtp_session.take()
478 };
479 if let Some(srtp_session) = srtp_session {
480 match srtp_session.close().await {
481 Ok(_) => {}
482 Err(err) => {
483 close_errs.push(err.into());
484 }
485 };
486 }
487 }
488
489 {
490 let srtcp_session = {
491 let mut srtcp_session = self.srtcp_session.lock().await;
492 srtcp_session.take()
493 };
494 if let Some(srtcp_session) = srtcp_session {
495 match srtcp_session.close().await {
496 Ok(_) => {}
497 Err(err) => {
498 close_errs.push(err.into());
499 }
500 };
501 }
502 }
503
504 {
505 let simulcast_streams: Vec<Arc<Stream>> = {
506 let mut simulcast_streams = self.simulcast_streams.lock().await;
507 simulcast_streams.drain().map(|(_, v)| v).collect()
508 };
509 for ss in simulcast_streams {
510 match ss.close().await {
511 Ok(_) => {}
512 Err(err) => {
513 close_errs.push(Error::new(format!(
514 "simulcast_streams ssrc={}: {}",
515 ss.get_ssrc(),
516 err
517 )));
518 }
519 };
520 }
521 }
522
523 if let Some(conn) = self.conn().await {
524 // dtls_transport connection may be closed on sctp close.
525 match conn.close().await {
526 Ok(_) => {}
527 Err(err) => {
528 if err.to_string() != dtls::Error::ErrConnClosed.to_string() {
529 close_errs.push(err.into());
530 }
531 }
532 }
533 }
534
535 self.state_change(RTCDtlsTransportState::Closed).await;
536
537 flatten_errs(close_errs)
538 }
539
validate_fingerprint(&self, remote_cert: &[u8]) -> Result<()>540 pub(crate) async fn validate_fingerprint(&self, remote_cert: &[u8]) -> Result<()> {
541 let remote_parameters = self.remote_parameters.lock().await;
542 for fp in &remote_parameters.fingerprints {
543 if fp.algorithm != "sha-256" {
544 return Err(Error::ErrUnsupportedFingerprintAlgorithm);
545 }
546
547 let mut h = Sha256::new();
548 h.update(remote_cert);
549 let hashed = h.finalize();
550 let values: Vec<String> = hashed.iter().map(|x| format! {"{x:02x}"}).collect();
551 let remote_value = values.join(":").to_lowercase();
552
553 if remote_value == fp.value.to_lowercase() {
554 return Ok(());
555 }
556 }
557
558 Err(Error::ErrNoMatchingCertificateFingerprint)
559 }
560
ensure_ice_conn(&self) -> Result<()>561 pub(crate) fn ensure_ice_conn(&self) -> Result<()> {
562 if self.ice_transport.state() == RTCIceTransportState::New {
563 Err(Error::ErrICEConnectionNotStarted)
564 } else {
565 Ok(())
566 }
567 }
568
store_simulcast_stream(&self, ssrc: SSRC, stream: Arc<Stream>)569 pub(crate) async fn store_simulcast_stream(&self, ssrc: SSRC, stream: Arc<Stream>) {
570 let mut simulcast_streams = self.simulcast_streams.lock().await;
571 simulcast_streams.insert(ssrc, stream);
572 }
573
remove_simulcast_stream(&self, ssrc: SSRC)574 pub(crate) async fn remove_simulcast_stream(&self, ssrc: SSRC) {
575 let mut simulcast_streams = self.simulcast_streams.lock().await;
576 simulcast_streams.remove(&ssrc);
577 }
578
streams_for_ssrc( &self, ssrc: SSRC, stream_info: &StreamInfo, interceptor: &Arc<dyn Interceptor + Send + Sync>, ) -> Result<( Arc<srtp::stream::Stream>, Arc<dyn RTPReader + Send + Sync>, Arc<srtp::stream::Stream>, Arc<dyn RTCPReader + Send + Sync>, )>579 pub(crate) async fn streams_for_ssrc(
580 &self,
581 ssrc: SSRC,
582 stream_info: &StreamInfo,
583 interceptor: &Arc<dyn Interceptor + Send + Sync>,
584 ) -> Result<(
585 Arc<srtp::stream::Stream>,
586 Arc<dyn RTPReader + Send + Sync>,
587 Arc<srtp::stream::Stream>,
588 Arc<dyn RTCPReader + Send + Sync>,
589 )> {
590 let srtp_session = self
591 .get_srtp_session()
592 .await
593 .ok_or(Error::ErrDtlsTransportNotStarted)?;
594 //log::debug!("streams_for_ssrc: srtp_session.listen ssrc={}", ssrc);
595 let rtp_read_stream = srtp_session.open(ssrc).await;
596 let rtp_stream_reader = Arc::clone(&rtp_read_stream) as Arc<dyn RTPReader + Send + Sync>;
597 let rtp_interceptor = interceptor
598 .bind_remote_stream(stream_info, rtp_stream_reader)
599 .await;
600
601 let srtcp_session = self
602 .get_srtcp_session()
603 .await
604 .ok_or(Error::ErrDtlsTransportNotStarted)?;
605 //log::debug!("streams_for_ssrc: srtcp_session.listen ssrc={}", ssrc);
606 let rtcp_read_stream = srtcp_session.open(ssrc).await;
607 let rtcp_stream_reader = Arc::clone(&rtcp_read_stream) as Arc<dyn RTCPReader + Send + Sync>;
608 let rtcp_interceptor = interceptor.bind_rtcp_reader(rtcp_stream_reader).await;
609
610 Ok((
611 rtp_read_stream,
612 rtp_interceptor,
613 rtcp_read_stream,
614 rtcp_interceptor,
615 ))
616 }
617 }
618