1#[cfg(test)]
2mod agent_test;
3
4pub mod agent_config;
5mod agent_proto;
6pub mod agent_selector;
7pub mod agent_stats;
8
9use agent_config::*;
10use bytes::BytesMut;
11use log::{debug, error, info, trace, warn};
12use mdns::{Mdns, QueryId};
13use sansio::Protocol;
14use std::collections::{HashMap, VecDeque};
15use std::net::{IpAddr, Ipv4Addr, SocketAddr};
16use std::sync::Arc;
17use std::time::{Duration, Instant};
18use stun::attributes::*;
19use stun::fingerprint::*;
20use stun::integrity::*;
21use stun::message::*;
22use stun::textattrs::*;
23use stun::xoraddr::*;
24
25use crate::candidate::candidate_peer_reflexive::CandidatePeerReflexiveConfig;
26use crate::candidate::{candidate_pair::*, *};
27use crate::mdns::{MulticastDnsMode, create_multicast_dns, generate_multicast_dns_name};
28use crate::network_type::NetworkType;
29use crate::rand::*;
30use crate::state::*;
31use crate::tcp_type::TcpType;
32use crate::url::*;
33use shared::error::*;
34use shared::{TaggedBytesMut, TransportContext, TransportProtocol};
35
36const ZERO_DURATION: Duration = Duration::from_secs(0);
37
38#[derive(Debug, Clone)]
39pub(crate) struct BindingRequest {
40 pub(crate) timestamp: Instant,
41 pub(crate) transaction_id: TransactionId,
42 pub(crate) destination: SocketAddr,
43 pub(crate) is_use_candidate: bool,
44}
45
46impl Default for BindingRequest {
47 fn default() -> Self {
48 Self {
49 timestamp: Instant::now(),
50 transaction_id: TransactionId::default(),
51 destination: SocketAddr::new(Ipv4Addr::new(0, 0, 0, 0).into(), 0),
52 is_use_candidate: false,
53 }
54 }
55}
56
57#[derive(Default, Clone)]
58pub struct Credentials {
59 pub ufrag: String,
60 pub pwd: String,
61}
62
63#[derive(Default, Clone)]
64pub(crate) struct UfragPwd {
65 pub(crate) local_credentials: Credentials,
66 pub(crate) remote_credentials: Option<Credentials>,
67}
68
69fn assert_inbound_username(m: &Message, expected_username: &str) -> Result<()> {
70 let mut username = Username::new(ATTR_USERNAME, String::new());
71 username.get_from(m)?;
72
73 if username.to_string() != expected_username {
74 return Err(Error::Other(format!(
75 "{:?} expected({}) actual({})",
76 Error::ErrMismatchUsername,
77 expected_username,
78 username,
79 )));
80 }
81
82 Ok(())
83}
84
85fn assert_inbound_message_integrity(m: &mut Message, key: &[u8]) -> Result<()> {
86 let message_integrity_attr = MessageIntegrity(key.to_vec());
87 message_integrity_attr.check(m)
88}
89
90pub enum Event {
91 ConnectionStateChange(ConnectionState),
92 SelectedCandidatePairChange(Box<Candidate>, Box<Candidate>),
93 RoleChange(bool),
96}
97
98pub struct Agent {
100 pub(crate) tie_breaker: u64,
101 pub(crate) is_controlling: bool,
102 pub(crate) lite: bool,
103
104 pub(crate) start_time: Instant,
105
106 pub(crate) connection_state: ConnectionState,
107 pub(crate) last_connection_state: ConnectionState,
108
109 pub(crate) ufrag_pwd: UfragPwd,
111
112 pub(crate) local_candidates: Vec<Candidate>,
113 pub(crate) remote_candidates: Vec<Candidate>,
114 pub(crate) candidate_pairs: Vec<CandidatePair>,
115 pub(crate) nominated_pair: Option<usize>,
116 pub(crate) selected_pair: Option<usize>,
117
118 pub(crate) pending_binding_requests: Vec<BindingRequest>,
120
121 pub(crate) insecure_skip_verify: bool,
123 pub(crate) max_binding_requests: u16,
124 pub(crate) host_acceptance_min_wait: Duration,
125 pub(crate) srflx_acceptance_min_wait: Duration,
126 pub(crate) prflx_acceptance_min_wait: Duration,
127 pub(crate) relay_acceptance_min_wait: Duration,
128 pub(crate) disconnected_timeout: Duration,
131 pub(crate) failed_timeout: Duration,
134 pub(crate) keepalive_interval: Duration,
137 pub(crate) last_consent_sent: Instant,
139 pub(crate) check_interval: Duration,
141 pub(crate) checking_duration: Instant,
142 pub(crate) last_checking_time: Instant,
143
144 pub(crate) mdns: Option<Mdns>,
145 pub(crate) mdns_queries: HashMap<QueryId, Candidate>,
146
147 pub(crate) mdns_mode: MulticastDnsMode,
148 pub(crate) mdns_local_name: String,
149 pub(crate) mdns_local_ip: Option<IpAddr>,
150
151 pub(crate) candidate_types: Vec<CandidateType>,
152 pub(crate) network_types: Vec<NetworkType>,
153 pub(crate) urls: Vec<Url>,
154
155 pub(crate) write_outs: VecDeque<TaggedBytesMut>,
156 pub(crate) event_outs: VecDeque<Event>,
157}
158
159impl Default for Agent {
160 fn default() -> Self {
161 Self {
162 tie_breaker: 0,
163 is_controlling: false,
164 lite: false,
165 start_time: Instant::now(),
166 connection_state: Default::default(),
167 last_connection_state: Default::default(),
168 ufrag_pwd: Default::default(),
169 local_candidates: vec![],
170 remote_candidates: vec![],
171 candidate_pairs: vec![],
172 nominated_pair: None,
173 selected_pair: None,
174 pending_binding_requests: vec![],
175 insecure_skip_verify: false,
176 max_binding_requests: 0,
177 host_acceptance_min_wait: Default::default(),
178 srflx_acceptance_min_wait: Default::default(),
179 prflx_acceptance_min_wait: Default::default(),
180 relay_acceptance_min_wait: Default::default(),
181 disconnected_timeout: Default::default(),
182 failed_timeout: Default::default(),
183 keepalive_interval: Default::default(),
184 last_consent_sent: Instant::now(),
185 check_interval: Default::default(),
186 checking_duration: Instant::now(),
187 last_checking_time: Instant::now(),
188 mdns_mode: MulticastDnsMode::Disabled,
189 mdns_local_name: "".to_owned(),
190 mdns_local_ip: None,
191 mdns_queries: HashMap::new(),
192 mdns: None,
193 candidate_types: vec![],
194 network_types: vec![],
195 urls: vec![],
196 write_outs: Default::default(),
197 event_outs: Default::default(),
198 }
199 }
200}
201
202impl Agent {
203 pub fn new(config: Arc<AgentConfig>) -> Result<Self> {
205 let mut mdns_local_name = config.multicast_dns_local_name.clone();
206 if mdns_local_name.is_empty() {
207 mdns_local_name = generate_multicast_dns_name();
208 }
209
210 if !mdns_local_name.ends_with(".local") || mdns_local_name.split('.').count() != 2 {
211 return Err(Error::ErrInvalidMulticastDnshostName);
212 }
213
214 let mdns_mode = config.multicast_dns_mode;
215 let mdns = create_multicast_dns(
216 mdns_mode,
217 &mdns_local_name,
218 &config.multicast_dns_local_ip,
219 &config.multicast_dns_query_timeout,
220 )
221 .unwrap_or_else(|err| {
222 warn!("Failed to initialize mDNS {mdns_local_name}: {err}");
225 None
226 });
227
228 let candidate_types = if config.candidate_types.is_empty() {
229 default_candidate_types()
230 } else {
231 config.candidate_types.clone()
232 };
233
234 if config.lite && (candidate_types.len() != 1 || candidate_types[0] != CandidateType::Host)
235 {
236 return Err(Error::ErrLiteUsingNonHostCandidates);
237 }
238
239 if !config.urls.is_empty()
240 && !contains_candidate_type(CandidateType::ServerReflexive, &candidate_types)
241 && !contains_candidate_type(CandidateType::Relay, &candidate_types)
242 {
243 return Err(Error::ErrUselessUrlsProvided);
244 }
245
246 let mut agent = Self {
247 tie_breaker: rand::random::<u64>(),
248 is_controlling: config.is_controlling,
249 lite: config.lite,
250
251 start_time: Instant::now(),
252
253 nominated_pair: None,
254 selected_pair: None,
255 candidate_pairs: vec![],
256
257 connection_state: ConnectionState::New,
258
259 insecure_skip_verify: config.insecure_skip_verify,
260
261 max_binding_requests: if let Some(max_binding_requests) = config.max_binding_requests {
265 max_binding_requests
266 } else {
267 DEFAULT_MAX_BINDING_REQUESTS
268 },
269 host_acceptance_min_wait: if let Some(host_acceptance_min_wait) =
270 config.host_acceptance_min_wait
271 {
272 host_acceptance_min_wait
273 } else {
274 DEFAULT_HOST_ACCEPTANCE_MIN_WAIT
275 },
276 srflx_acceptance_min_wait: if let Some(srflx_acceptance_min_wait) =
277 config.srflx_acceptance_min_wait
278 {
279 srflx_acceptance_min_wait
280 } else {
281 DEFAULT_SRFLX_ACCEPTANCE_MIN_WAIT
282 },
283 prflx_acceptance_min_wait: if let Some(prflx_acceptance_min_wait) =
284 config.prflx_acceptance_min_wait
285 {
286 prflx_acceptance_min_wait
287 } else {
288 DEFAULT_PRFLX_ACCEPTANCE_MIN_WAIT
289 },
290 relay_acceptance_min_wait: if let Some(relay_acceptance_min_wait) =
291 config.relay_acceptance_min_wait
292 {
293 relay_acceptance_min_wait
294 } else {
295 DEFAULT_RELAY_ACCEPTANCE_MIN_WAIT
296 },
297
298 disconnected_timeout: if let Some(disconnected_timeout) = config.disconnected_timeout {
301 disconnected_timeout
302 } else {
303 DEFAULT_DISCONNECTED_TIMEOUT
304 },
305
306 failed_timeout: if let Some(failed_timeout) = config.failed_timeout {
309 failed_timeout
310 } else {
311 DEFAULT_FAILED_TIMEOUT
312 },
313
314 keepalive_interval: if let Some(keepalive_interval) = config.keepalive_interval {
317 keepalive_interval
318 } else {
319 DEFAULT_KEEPALIVE_INTERVAL
320 },
321
322 check_interval: if config.check_interval == Duration::from_secs(0) {
324 DEFAULT_CHECK_INTERVAL
325 } else {
326 config.check_interval
327 },
328 last_consent_sent: Instant::now(),
329 checking_duration: Instant::now(),
330 last_checking_time: Instant::now(),
331 last_connection_state: ConnectionState::Unspecified,
332
333 mdns,
334 mdns_queries: HashMap::new(),
335
336 mdns_mode,
337 mdns_local_name,
338 mdns_local_ip: config.multicast_dns_local_ip,
339
340 ufrag_pwd: UfragPwd::default(),
341
342 local_candidates: vec![],
343 remote_candidates: vec![],
344
345 pending_binding_requests: vec![],
347
348 candidate_types,
349 network_types: config.network_types.clone(),
350 urls: config.urls.clone(),
351
352 write_outs: VecDeque::new(),
353 event_outs: VecDeque::new(),
354 };
355
356 if let Err(err) = agent.restart(config.local_ufrag.clone(), config.local_pwd.clone(), false)
358 {
359 let _ = agent.close();
360 return Err(err);
361 }
362
363 Ok(agent)
364 }
365
366 pub fn add_local_candidate(&mut self, mut c: Candidate) -> Result<bool> {
368 if !self.network_types.is_empty() {
370 let candidate_network_type = c.network_type();
371 if !self.network_types.contains(&candidate_network_type) {
372 debug!(
373 "Ignoring local candidate with network type {:?} (not in configured network types: {:?})",
374 candidate_network_type, self.network_types
375 );
376 return Ok(false);
377 }
378 }
379
380 let candidate_type = c.candidate_type();
382 if !self.candidate_types.is_empty() && !self.candidate_types.contains(&candidate_type) {
383 debug!(
384 "Ignoring local candidate with type {:?} (not in configured candidate types: {:?})",
385 candidate_type, self.candidate_types
386 );
387 return Ok(false);
388 }
389
390 if c.candidate_type() == CandidateType::Host
391 && self.mdns_mode == MulticastDnsMode::QueryAndGather
392 && c.network_type == NetworkType::Udp4
393 && self
394 .mdns_local_ip
395 .is_some_and(|local_ip| local_ip == c.addr().ip())
396 {
397 trace!(
400 "mDNS hides local ip {} with local name {}",
401 c.address, self.mdns_local_name
402 );
403 c.address = self.mdns_local_name.clone();
404 }
405
406 for cand in &self.local_candidates {
407 if cand.equal(&c) {
408 return Ok(false);
409 }
410 }
411
412 self.local_candidates.push(c);
413 let local_index = self.local_candidates.len() - 1;
414
415 for remote_index in 0..self.remote_candidates.len() {
416 if self.local_candidates[local_index]
417 .can_pair_with(&self.remote_candidates[remote_index])
418 {
419 self.add_pair(local_index, remote_index);
420 }
421 }
422
423 self.request_connectivity_check();
424
425 Ok(true)
426 }
427
428 pub fn add_remote_candidate(&mut self, c: Candidate) -> Result<bool> {
430 if !self.network_types.is_empty() {
432 let candidate_network_type = c.network_type();
433 if !self.network_types.contains(&candidate_network_type) {
434 debug!(
435 "Ignoring remote candidate with network type {:?} (not in configured network types: {:?})",
436 candidate_network_type, self.network_types
437 );
438 return Ok(false);
439 }
440 }
441
442 if c.tcp_type() == TcpType::Active {
446 debug!(
447 "Ignoring remote candidate with tcptype active: {}",
448 c.address()
449 );
450 return Ok(false);
451 }
452
453 if c.candidate_type() == CandidateType::Host && c.address().ends_with(".local") {
455 if self.mdns_mode == MulticastDnsMode::Disabled {
456 warn!(
457 "remote mDNS candidate added, but mDNS is disabled: ({})",
458 c.address()
459 );
460 return Ok(false);
461 }
462
463 if c.candidate_type() != CandidateType::Host {
464 return Err(Error::ErrAddressParseFailed);
465 }
466
467 if let Some(mdns_conn) = &mut self.mdns {
468 let query_id = mdns_conn.query(c.address());
469 self.mdns_queries.insert(query_id, c);
470 }
471
472 return Ok(false);
473 }
474
475 self.trigger_request_connectivity_check(vec![c]);
476 Ok(true)
477 }
478
479 fn trigger_request_connectivity_check(&mut self, remote_candidates: Vec<Candidate>) {
480 for c in remote_candidates {
481 if !self.remote_candidates.iter().any(|cand| cand.equal(&c)) {
482 self.remote_candidates.push(c);
483 let remote_index = self.remote_candidates.len() - 1;
484
485 for local_index in 0..self.local_candidates.len() {
486 if self.local_candidates[local_index]
487 .can_pair_with(&self.remote_candidates[remote_index])
488 {
489 self.add_pair(local_index, remote_index);
490 }
491 }
492
493 self.request_connectivity_check();
494 }
495 }
496 }
497
498 pub fn set_remote_credentials(
500 &mut self,
501 remote_ufrag: String,
502 remote_pwd: String,
503 ) -> Result<()> {
504 if remote_ufrag.is_empty() {
505 return Err(Error::ErrRemoteUfragEmpty);
506 } else if remote_pwd.is_empty() {
507 return Err(Error::ErrRemotePwdEmpty);
508 }
509
510 self.ufrag_pwd.remote_credentials = Some(Credentials {
511 ufrag: remote_ufrag,
512 pwd: remote_pwd,
513 });
514
515 Ok(())
516 }
517
518 pub fn get_remote_credentials(&self) -> Option<&Credentials> {
520 self.ufrag_pwd.remote_credentials.as_ref()
521 }
522
523 pub fn get_local_credentials(&self) -> &Credentials {
525 &self.ufrag_pwd.local_credentials
526 }
527
528 pub fn role(&self) -> bool {
529 self.is_controlling
530 }
531
532 pub fn set_role(&mut self, is_controlling: bool) {
533 self.is_controlling = is_controlling;
534 }
535
536 pub fn state(&self) -> ConnectionState {
537 self.connection_state
538 }
539
540 pub fn is_valid_non_stun_traffic(&mut self, transport: TransportContext) -> bool {
541 self.find_local_candidate(transport.local_addr, transport.transport_protocol)
542 .is_some()
543 && self.validate_non_stun_traffic(transport.peer_addr)
544 }
545
546 fn get_timeout_interval(&self) -> Duration {
547 let (check_interval, keepalive_interval, disconnected_timeout, failed_timeout) = (
548 self.check_interval,
549 self.keepalive_interval,
550 self.disconnected_timeout,
551 self.failed_timeout,
552 );
553 let mut interval = DEFAULT_CHECK_INTERVAL;
554
555 let mut update_interval = |x: Duration| {
556 if x != ZERO_DURATION && (interval == ZERO_DURATION || interval > x) {
557 interval = x;
558 }
559 };
560
561 match self.last_connection_state {
562 ConnectionState::New | ConnectionState::Checking => {
563 update_interval(check_interval);
565 }
566 ConnectionState::Connected | ConnectionState::Disconnected => {
567 update_interval(keepalive_interval);
568 }
569 _ => {}
570 };
571 update_interval(disconnected_timeout);
573 update_interval(failed_timeout);
574 interval
575 }
576
577 pub fn get_selected_candidate_pair(&self) -> Option<(&Candidate, &Candidate)> {
579 if let Some(pair_index) = self.get_selected_pair() {
580 let candidate_pair = &self.candidate_pairs[pair_index];
581 Some((
582 &self.local_candidates[candidate_pair.local_index],
583 &self.remote_candidates[candidate_pair.remote_index],
584 ))
585 } else {
586 None
587 }
588 }
589
590 pub fn get_best_available_candidate_pair(&self) -> Option<(&Candidate, &Candidate)> {
591 if let Some(pair_index) = self.get_best_available_pair() {
592 let candidate_pair = &self.candidate_pairs[pair_index];
593 Some((
594 &self.local_candidates[candidate_pair.local_index],
595 &self.remote_candidates[candidate_pair.remote_index],
596 ))
597 } else {
598 None
599 }
600 }
601
602 pub fn start_connectivity_checks(
604 &mut self,
605 is_controlling: bool,
606 remote_ufrag: String,
607 remote_pwd: String,
608 ) -> Result<()> {
609 debug!(
610 "Started agent: isControlling? {}, remoteUfrag: {}, remotePwd: {}",
611 is_controlling, remote_ufrag, remote_pwd
612 );
613 self.set_remote_credentials(remote_ufrag, remote_pwd)?;
614 self.is_controlling = is_controlling;
615 self.start();
616
617 self.update_connection_state(ConnectionState::Checking);
618 self.request_connectivity_check();
619
620 Ok(())
621 }
622
623 pub fn restart(
626 &mut self,
627 mut ufrag: String,
628 mut pwd: String,
629 keep_local_candidates: bool,
630 ) -> Result<()> {
631 if ufrag.is_empty() {
632 ufrag = generate_ufrag();
633 }
634 if pwd.is_empty() {
635 pwd = generate_pwd();
636 }
637
638 if ufrag.len() * 8 < 24 {
639 return Err(Error::ErrLocalUfragInsufficientBits);
640 }
641 if pwd.len() * 8 < 128 {
642 return Err(Error::ErrLocalPwdInsufficientBits);
643 }
644
645 self.ufrag_pwd.local_credentials.ufrag = ufrag;
647 self.ufrag_pwd.local_credentials.pwd = pwd;
648 self.ufrag_pwd.remote_credentials = None;
649
650 self.pending_binding_requests = vec![];
651
652 self.candidate_pairs = vec![];
653
654 self.set_selected_pair(None);
655 self.delete_all_candidates(keep_local_candidates);
656 self.start();
657
658 if self.connection_state != ConnectionState::New {
661 self.update_connection_state(ConnectionState::Checking);
662 }
663
664 Ok(())
665 }
666
667 pub fn get_local_candidates(&self) -> &[Candidate] {
669 &self.local_candidates
670 }
671
672 fn contact(&mut self, now: Instant) {
673 if self.connection_state == ConnectionState::Failed {
674 self.last_connection_state = self.connection_state;
677 return;
678 }
679 if self.connection_state == ConnectionState::Checking {
680 if self.last_connection_state != self.connection_state {
682 self.checking_duration = now;
683 }
684
685 if now
687 .checked_duration_since(self.checking_duration)
688 .unwrap_or_else(|| Duration::from_secs(0))
689 > self.disconnected_timeout + self.failed_timeout
690 {
691 self.update_connection_state(ConnectionState::Failed);
692 self.last_connection_state = self.connection_state;
693 return;
694 }
695 }
696
697 self.contact_candidates();
698
699 self.last_connection_state = self.connection_state;
700 self.last_checking_time = now;
701 }
702
703 pub(crate) fn update_connection_state(&mut self, new_state: ConnectionState) {
704 if self.connection_state != new_state {
705 if new_state == ConnectionState::Failed {
707 self.set_selected_pair(None);
708 self.delete_all_candidates(false);
709 }
710
711 info!(
712 "[{}]: Setting new connection state: {}",
713 self.get_name(),
714 new_state
715 );
716 self.connection_state = new_state;
717 self.event_outs
718 .push_back(Event::ConnectionStateChange(new_state));
719 }
720 }
721
722 pub(crate) fn set_selected_pair(&mut self, selected_pair: Option<usize>) {
723 if let Some(pair_index) = selected_pair {
724 trace!(
725 "[{}]: Set selected candidate pair: {:?}",
726 self.get_name(),
727 self.candidate_pairs[pair_index]
728 );
729
730 self.candidate_pairs[pair_index].nominated = true;
731 self.selected_pair = Some(pair_index);
732
733 self.update_connection_state(ConnectionState::Connected);
734
735 let candidate_pair = &self.candidate_pairs[pair_index];
737 self.event_outs
738 .push_back(Event::SelectedCandidatePairChange(
739 Box::new(self.local_candidates[candidate_pair.local_index].clone()),
740 Box::new(self.remote_candidates[candidate_pair.remote_index].clone()),
741 ));
742 } else {
743 self.selected_pair = None;
744 }
745 }
746
747 pub(crate) fn ping_all_candidates(&mut self) {
748 let mut pairs: Vec<(usize, usize)> = vec![];
749
750 let name = self.get_name().to_string();
751 if self.candidate_pairs.is_empty() {
752 warn!(
753 "[{}]: pingAllCandidates called with no candidate pairs. Connection is not possible yet.",
754 name,
755 );
756 }
757 for p in &mut self.candidate_pairs {
758 if p.state == CandidatePairState::Waiting {
759 p.state = CandidatePairState::InProgress;
760 } else if p.state != CandidatePairState::InProgress {
761 continue;
762 }
763
764 if p.binding_request_count > self.max_binding_requests {
765 trace!(
766 "[{}]: max requests reached for pair {} (local_addr {} <-> remote_addr {}), marking it as failed",
767 name,
768 *p,
769 self.local_candidates[p.local_index].addr(),
770 self.remote_candidates[p.remote_index].addr()
771 );
772 p.state = CandidatePairState::Failed;
773 } else {
774 p.binding_request_count += 1;
775 let local = p.local_index;
776 let remote = p.remote_index;
777 pairs.push((local, remote));
778 }
779 }
780
781 if !pairs.is_empty() {
782 trace!(
783 "[{}]: pinging all {} candidates",
784 self.get_name(),
785 pairs.len()
786 );
787 }
788
789 for (local, remote) in pairs {
790 self.ping_candidate(local, remote);
791 }
792 }
793
794 pub(crate) fn add_pair(&mut self, local_index: usize, remote_index: usize) {
795 let p = CandidatePair::new(
796 local_index,
797 remote_index,
798 self.local_candidates[local_index].priority(),
799 self.remote_candidates[remote_index].priority(),
800 self.is_controlling,
801 );
802 self.candidate_pairs.push(p);
803 }
804
805 pub(crate) fn find_pair(&self, local_index: usize, remote_index: usize) -> Option<usize> {
806 for (index, p) in self.candidate_pairs.iter().enumerate() {
807 if p.local_index == local_index && p.remote_index == remote_index {
808 return Some(index);
809 }
810 }
811 None
812 }
813
814 pub(crate) fn validate_selected_pair(&mut self) -> bool {
817 let (valid, disconnected_time) = {
818 self.selected_pair.as_ref().map_or_else(
819 || (false, Duration::from_secs(0)),
820 |&pair_index| {
821 let remote_index = self.candidate_pairs[pair_index].remote_index;
822
823 let disconnected_time = Instant::now()
824 .duration_since(self.remote_candidates[remote_index].last_received());
825 (true, disconnected_time)
826 },
827 )
828 };
829
830 if valid {
831 let mut total_time_to_failure = self.failed_timeout;
833 if total_time_to_failure != Duration::from_secs(0) {
834 total_time_to_failure += self.disconnected_timeout;
835 }
836
837 if total_time_to_failure != Duration::from_secs(0)
838 && disconnected_time > total_time_to_failure
839 {
840 self.update_connection_state(ConnectionState::Failed);
841 } else if self.disconnected_timeout != Duration::from_secs(0)
842 && disconnected_time > self.disconnected_timeout
843 {
844 self.update_connection_state(ConnectionState::Disconnected);
845 } else {
846 self.update_connection_state(ConnectionState::Connected);
847 }
848 }
849
850 valid
851 }
852
853 pub(crate) fn check_keepalive(&mut self) {
856 let (local_index, remote_index, pair_index) = {
857 self.selected_pair
858 .as_ref()
859 .map_or((None, None, None), |&pair_index| {
860 let p = &self.candidate_pairs[pair_index];
861 (Some(p.local_index), Some(p.remote_index), Some(pair_index))
862 })
863 };
864
865 if let (Some(local_index), Some(remote_index), Some(pair_index)) =
866 (local_index, remote_index, pair_index)
867 && self.keepalive_interval != Duration::from_secs(0)
868 && self.last_consent_sent.elapsed() >= self.keepalive_interval
869 {
870 self.last_consent_sent = Instant::now();
871 self.candidate_pairs[pair_index].on_consent_request_sent();
872 self.ping_candidate(local_index, remote_index);
873 }
874 }
875
876 fn request_connectivity_check(&mut self) {
877 if self.ufrag_pwd.remote_credentials.is_some() {
878 self.contact(Instant::now());
879 }
880 }
881
882 pub(crate) fn delete_all_candidates(&mut self, keep_local_candidates: bool) {
887 if !keep_local_candidates {
888 self.local_candidates.clear();
889 }
890 self.remote_candidates.clear();
891 }
892
893 pub(crate) fn find_remote_candidate(&self, addr: SocketAddr) -> Option<usize> {
894 for (index, c) in self.remote_candidates.iter().enumerate() {
895 if c.addr() == addr {
896 return Some(index);
897 }
898 }
899 None
900 }
901
902 pub(crate) fn find_local_candidate(
903 &self,
904 addr: SocketAddr,
905 transport_protocol: TransportProtocol,
906 ) -> Option<usize> {
907 for (index, c) in self.local_candidates.iter().enumerate() {
908 if c.network_type().to_protocol() != transport_protocol {
909 continue;
910 }
911
912 if c.tcp_type() == TcpType::Active && transport_protocol == TransportProtocol::TCP {
916 if c.addr().ip() == addr.ip() {
917 return Some(index);
918 }
919 } else if c.addr() == addr {
920 return Some(index);
921 } else if let Some(related_address) = c.related_address()
922 && related_address.address == addr.ip().to_string()
923 && related_address.port == addr.port()
924 {
925 return Some(index);
926 }
927 }
928 None
929 }
930
931 pub(crate) fn send_binding_request(
932 &mut self,
933 m: &Message,
934 local_index: usize,
935 remote_index: usize,
936 ) {
937 trace!(
938 "[{}]: ping STUN from {} to {}",
939 self.get_name(),
940 self.local_candidates[local_index],
941 self.remote_candidates[remote_index],
942 );
943
944 self.invalidate_pending_binding_requests(Instant::now());
945
946 self.pending_binding_requests.push(BindingRequest {
947 timestamp: Instant::now(),
948 transaction_id: m.transaction_id,
949 destination: self.remote_candidates[remote_index].addr(),
950 is_use_candidate: m.contains(ATTR_USE_CANDIDATE),
951 });
952
953 if let Some(pair_index) = self.find_pair(local_index, remote_index) {
955 self.candidate_pairs[pair_index].on_request_sent();
956 }
957
958 self.send_stun(m, local_index, remote_index);
959 }
960
961 pub(crate) fn send_binding_success(
962 &mut self,
963 m: &Message,
964 local_index: usize,
965 remote_index: usize,
966 ) {
967 let addr = self.remote_candidates[remote_index].addr();
968 let (ip, port) = (addr.ip(), addr.port());
969 let local_pwd = self.ufrag_pwd.local_credentials.pwd.clone();
970
971 let (out, result) = {
972 let mut out = Message::new();
973 let result = out.build(&[
974 Box::new(m.clone()),
975 Box::new(BINDING_SUCCESS),
976 Box::new(XorMappedAddress { ip, port }),
977 Box::new(MessageIntegrity::new_short_term_integrity(local_pwd)),
978 Box::new(FINGERPRINT),
979 ]);
980 (out, result)
981 };
982
983 if let Err(err) = result {
984 warn!(
985 "[{}]: Failed to handle inbound ICE from: {} to: {} error: {}",
986 self.get_name(),
987 self.local_candidates[local_index],
988 self.remote_candidates[remote_index],
989 err
990 );
991 } else {
992 if let Some(pair_index) = self.find_pair(local_index, remote_index) {
994 self.candidate_pairs[pair_index].on_response_sent();
995 }
996 self.send_stun(&out, local_index, remote_index);
997 }
998 }
999
1000 pub(crate) fn send_role_conflict_error(
1003 &mut self,
1004 m: &Message,
1005 local_index: usize,
1006 remote_index: usize,
1007 ) {
1008 use stun::error_code::*;
1009
1010 let local_pwd = self.ufrag_pwd.local_credentials.pwd.clone();
1011
1012 let (out, result) = {
1013 let mut out = Message::new();
1014 let result = out.build(&[
1015 Box::new(m.clone()),
1016 Box::new(stun::message::BINDING_ERROR),
1017 Box::new(CODE_ROLE_CONFLICT),
1018 Box::new(MessageIntegrity::new_short_term_integrity(local_pwd)),
1019 Box::new(FINGERPRINT),
1020 ]);
1021 (out, result)
1022 };
1023
1024 if let Err(err) = result {
1025 warn!(
1026 "[{}]: Failed to send role conflict error from: {} to: {} error: {}",
1027 self.get_name(),
1028 self.local_candidates[local_index],
1029 self.remote_candidates[remote_index],
1030 err
1031 );
1032 } else {
1033 debug!(
1034 "[{}]: Sent 487 Role Conflict error from {} to {}",
1035 self.get_name(),
1036 self.local_candidates[local_index],
1037 self.remote_candidates[remote_index]
1038 );
1039 self.send_stun(&out, local_index, remote_index);
1040 }
1041 }
1042
1043 pub(crate) fn switch_role(&mut self) {
1046 self.is_controlling = !self.is_controlling;
1047
1048 for pair in &mut self.candidate_pairs {
1051 pair.ice_role_controlling = self.is_controlling;
1052 }
1053
1054 self.nominated_pair = None;
1056
1057 info!(
1058 "[{}]: Role switched, recomputed {} candidate pair priorities",
1059 self.get_name(),
1060 self.candidate_pairs.len()
1061 );
1062
1063 self.event_outs
1064 .push_back(Event::RoleChange(self.is_controlling));
1065 }
1066
1067 pub(crate) fn invalidate_pending_binding_requests(&mut self, filter_time: Instant) {
1072 let pending_binding_requests = &mut self.pending_binding_requests;
1073 let initial_size = pending_binding_requests.len();
1074
1075 let mut temp = vec![];
1076 for binding_request in pending_binding_requests.drain(..) {
1077 if filter_time
1078 .checked_duration_since(binding_request.timestamp)
1079 .map(|duration| duration < MAX_BINDING_REQUEST_TIMEOUT)
1080 .unwrap_or(true)
1081 {
1082 temp.push(binding_request);
1083 }
1084 }
1085
1086 *pending_binding_requests = temp;
1087 let bind_requests_remaining = pending_binding_requests.len();
1088 let bind_requests_removed = initial_size - bind_requests_remaining;
1089 if bind_requests_removed > 0 {
1090 trace!(
1091 "[{}]: Discarded {} binding requests because they expired, still {} remaining",
1092 self.get_name(),
1093 bind_requests_removed,
1094 bind_requests_remaining,
1095 );
1096 }
1097 }
1098
1099 pub(crate) fn handle_inbound_binding_success(
1102 &mut self,
1103 id: TransactionId,
1104 ) -> Option<BindingRequest> {
1105 self.invalidate_pending_binding_requests(Instant::now());
1106
1107 let pending_binding_requests = &mut self.pending_binding_requests;
1108 for i in 0..pending_binding_requests.len() {
1109 if pending_binding_requests[i].transaction_id == id {
1110 let valid_binding_request = pending_binding_requests.remove(i);
1111 return Some(valid_binding_request);
1112 }
1113 }
1114 None
1115 }
1116
1117 pub(crate) fn handle_inbound(
1119 &mut self,
1120 now: Instant,
1121 m: &mut Message,
1122 local_index: usize,
1123 remote_addr: SocketAddr,
1124 ) -> Result<()> {
1125 if m.typ.method != METHOD_BINDING
1126 || !(m.typ.class == CLASS_SUCCESS_RESPONSE
1127 || m.typ.class == CLASS_REQUEST
1128 || m.typ.class == CLASS_INDICATION)
1129 {
1130 trace!(
1131 "[{}]: unhandled STUN from {} to {} class({}) method({})",
1132 self.get_name(),
1133 remote_addr,
1134 self.local_candidates[local_index],
1135 m.typ.class,
1136 m.typ.method
1137 );
1138 return Err(Error::ErrUnhandledStunpacket);
1139 }
1140
1141 if self.is_controlling {
1143 if m.contains(ATTR_ICE_CONTROLLING) {
1144 let mut remote_controlling = crate::attributes::control::AttrControlling::default();
1146 if let Err(err) = remote_controlling.get_from(m) {
1147 warn!(
1148 "[{}]: Failed to get remote ICE-CONTROLLING attribute: {}",
1149 self.get_name(),
1150 err
1151 );
1152 return Err(err);
1153 }
1154
1155 debug!(
1156 "[{}]: Role conflict detected (both controlling), local tiebreaker: {}, remote tiebreaker: {}",
1157 self.get_name(),
1158 self.tie_breaker,
1159 remote_controlling.0
1160 );
1161
1162 if m.typ.class == CLASS_REQUEST {
1164 if let Some(remote_index) = self.find_remote_candidate(remote_addr) {
1166 self.send_role_conflict_error(m, local_index, remote_index);
1167 }
1168
1169 if self.tie_breaker < remote_controlling.0 {
1171 info!(
1172 "[{}]: Switching from controlling to controlled due to role conflict (smaller tiebreaker)",
1173 self.get_name()
1174 );
1175 self.switch_role();
1176 }
1177 }
1178 } else if m.contains(ATTR_USE_CANDIDATE) {
1180 debug!(
1181 "[{}]: useCandidate && a.isControlling == true",
1182 self.get_name(),
1183 );
1184 return Err(Error::ErrUnexpectedStunrequestMessage);
1185 }
1186 } else if m.contains(ATTR_ICE_CONTROLLED) {
1187 let mut remote_controlled = crate::attributes::control::AttrControlled::default();
1189 if let Err(err) = remote_controlled.get_from(m) {
1190 warn!(
1191 "[{}]: Failed to get remote ICE-CONTROLLED attribute: {}",
1192 self.get_name(),
1193 err
1194 );
1195 return Err(err);
1196 }
1197
1198 debug!(
1199 "[{}]: Role conflict detected (both controlled), local tiebreaker: {}, remote tiebreaker: {}",
1200 self.get_name(),
1201 self.tie_breaker,
1202 remote_controlled.0
1203 );
1204
1205 if m.typ.class == CLASS_REQUEST {
1207 if let Some(remote_index) = self.find_remote_candidate(remote_addr) {
1209 self.send_role_conflict_error(m, local_index, remote_index);
1210 }
1211
1212 if self.tie_breaker > remote_controlled.0 {
1214 info!(
1215 "[{}]: Switching from controlled to controlling due to role conflict (larger tiebreaker)",
1216 self.get_name()
1217 );
1218 self.switch_role();
1219 }
1220 }
1221 }
1223
1224 let Some(remote_credentials) = &self.ufrag_pwd.remote_credentials else {
1225 debug!(
1226 "[{}]: ufrag_pwd.remote_credentials.is_none",
1227 self.get_name(),
1228 );
1229 return Err(Error::ErrPasswordEmpty);
1230 };
1231
1232 let mut remote_candidate_index = self.find_remote_candidate(remote_addr);
1233 if m.typ.class == CLASS_SUCCESS_RESPONSE {
1234 if let Err(err) = assert_inbound_message_integrity(m, remote_credentials.pwd.as_bytes())
1235 {
1236 warn!(
1237 "[{}]: discard message from ({}), {}",
1238 self.get_name(),
1239 remote_addr,
1240 err
1241 );
1242 return Err(err);
1243 }
1244
1245 if let Some(remote_index) = &remote_candidate_index {
1246 self.handle_success_response(now, m, local_index, *remote_index, remote_addr);
1247 } else {
1248 warn!(
1249 "[{}]: discard success message from ({}), no such remote",
1250 self.get_name(),
1251 remote_addr
1252 );
1253 return Err(Error::ErrUnhandledStunpacket);
1254 }
1255 } else if m.typ.class == CLASS_REQUEST {
1256 {
1257 let username = self.ufrag_pwd.local_credentials.ufrag.clone()
1258 + ":"
1259 + remote_credentials.ufrag.as_str();
1260 if let Err(err) = assert_inbound_username(m, &username) {
1261 warn!(
1262 "[{}]: discard message from ({}), {}",
1263 self.get_name(),
1264 remote_addr,
1265 err
1266 );
1267 return Err(err);
1268 } else if let Err(err) = assert_inbound_message_integrity(
1269 m,
1270 self.ufrag_pwd.local_credentials.pwd.as_bytes(),
1271 ) {
1272 warn!(
1273 "[{}]: discard message from ({}), {}",
1274 self.get_name(),
1275 remote_addr,
1276 err
1277 );
1278 return Err(err);
1279 }
1280 }
1281
1282 if remote_candidate_index.is_none() {
1283 let network_type = self.local_candidates[local_index].network_type();
1285 let (ip, port) = (remote_addr.ip(), remote_addr.port());
1286
1287 let prflx_candidate_config = CandidatePeerReflexiveConfig {
1288 base_config: CandidateConfig {
1289 network: network_type.to_string(),
1290 address: ip.to_string(),
1291 port,
1292 component: self.local_candidates[local_index].component(),
1293 ..CandidateConfig::default()
1294 },
1295 rel_addr: "".to_owned(),
1296 rel_port: 0,
1297 };
1298
1299 match prflx_candidate_config.new_candidate_peer_reflexive() {
1300 Ok(prflx_candidate) => {
1301 if let Ok(added) = self.add_remote_candidate(prflx_candidate)
1302 && added
1303 {
1304 remote_candidate_index = Some(self.remote_candidates.len() - 1);
1305 }
1306 }
1307 Err(err) => {
1308 error!(
1309 "[{}]: Failed to create new remote prflx candidate ({})",
1310 self.get_name(),
1311 err
1312 );
1313 return Err(err);
1314 }
1315 };
1316
1317 debug!(
1318 "[{}]: adding a new peer-reflexive candidate: {} ",
1319 self.get_name(),
1320 remote_addr
1321 );
1322 }
1323
1324 trace!(
1325 "[{}]: inbound STUN (Request) from {} to {}",
1326 self.get_name(),
1327 remote_addr,
1328 self.local_candidates[local_index]
1329 );
1330
1331 if let Some(remote_index) = &remote_candidate_index {
1332 self.handle_binding_request(m, local_index, *remote_index);
1333 }
1334 }
1335
1336 if let Some(remote_index) = remote_candidate_index {
1337 self.remote_candidates[remote_index].seen(false);
1338 }
1339
1340 Ok(())
1341 }
1342
1343 pub(crate) fn validate_non_stun_traffic(&mut self, remote_addr: SocketAddr) -> bool {
1346 self.find_remote_candidate(remote_addr)
1347 .is_some_and(|remote_index| {
1348 self.remote_candidates[remote_index].seen(false);
1349 true
1350 })
1351 }
1352
1353 pub(crate) fn send_stun(&mut self, msg: &Message, local_index: usize, remote_index: usize) {
1354 let peer_addr = self.remote_candidates[remote_index].addr();
1355 let local_addr = self.local_candidates[local_index].addr();
1356 let transport_protocol = if self.local_candidates[local_index].network_type().is_tcp() {
1357 TransportProtocol::TCP
1358 } else {
1359 TransportProtocol::UDP
1360 };
1361
1362 self.write_outs.push_back(TaggedBytesMut {
1363 now: Instant::now(),
1364 transport: TransportContext {
1365 local_addr,
1366 peer_addr,
1367 ecn: None,
1368 transport_protocol,
1369 },
1370 message: BytesMut::from(&msg.raw[..]),
1371 });
1372
1373 self.local_candidates[local_index].seen(true);
1374 }
1375
1376 fn handle_inbound_candidate_msg(
1377 &mut self,
1378 local_index: usize,
1379 msg: TaggedBytesMut,
1380 ) -> Result<()> {
1381 if is_stun_message(&msg.message) {
1382 let mut m = Message {
1383 raw: msg.message.to_vec(),
1384 ..Message::default()
1385 };
1386
1387 if let Err(err) = m.decode() {
1388 warn!(
1389 "[{}]: Failed to handle decode ICE from {} to {}: {}",
1390 self.get_name(),
1391 msg.transport.local_addr,
1392 msg.transport.peer_addr,
1393 err
1394 );
1395 Err(err)
1396 } else {
1397 self.handle_inbound(msg.now, &mut m, local_index, msg.transport.peer_addr)
1398 }
1399 } else {
1400 if !self.validate_non_stun_traffic(msg.transport.peer_addr) {
1401 warn!(
1402 "[{}]: Discarded message, not a valid remote candidate from {}",
1403 self.get_name(),
1404 msg.transport.peer_addr,
1405 );
1406 } else {
1407 warn!(
1408 "[{}]: non-STUN traffic message from a valid remote candidate from {}",
1409 self.get_name(),
1410 msg.transport.peer_addr
1411 );
1412 }
1413 Err(Error::ErrNonStunmessage)
1414 }
1415 }
1416
1417 pub(crate) fn get_name(&self) -> &str {
1418 if self.is_controlling {
1419 "controlling"
1420 } else {
1421 "controlled"
1422 }
1423 }
1424
1425 pub(crate) fn get_selected_pair(&self) -> Option<usize> {
1426 self.selected_pair
1427 }
1428
1429 pub(crate) fn get_best_available_pair(&self) -> Option<usize> {
1430 let mut best_pair_index: Option<usize> = None;
1431
1432 for (index, p) in self.candidate_pairs.iter().enumerate() {
1433 if p.state == CandidatePairState::Failed {
1434 continue;
1435 }
1436
1437 if let Some(pair_index) = &mut best_pair_index {
1438 let b = &self.candidate_pairs[*pair_index];
1439 if b.priority() < p.priority() {
1440 *pair_index = index;
1441 }
1442 } else {
1443 best_pair_index = Some(index);
1444 }
1445 }
1446
1447 best_pair_index
1448 }
1449
1450 pub(crate) fn get_best_valid_candidate_pair(&self) -> Option<usize> {
1451 let mut best_pair_index: Option<usize> = None;
1452
1453 for (index, p) in self.candidate_pairs.iter().enumerate() {
1454 if p.state != CandidatePairState::Succeeded {
1455 continue;
1456 }
1457
1458 if let Some(pair_index) = &mut best_pair_index {
1459 let b = &self.candidate_pairs[*pair_index];
1460 if b.priority() < p.priority() {
1461 *pair_index = index;
1462 }
1463 } else {
1464 best_pair_index = Some(index);
1465 }
1466 }
1467
1468 best_pair_index
1469 }
1470}