1 #[cfg(test)] 2 mod agent_gather_test; 3 #[cfg(test)] 4 mod agent_test; 5 #[cfg(test)] 6 mod agent_transport_test; 7 #[cfg(test)] 8 pub(crate) mod agent_vnet_test; 9 10 pub mod agent_config; 11 pub mod agent_gather; 12 pub(crate) mod agent_internal; 13 pub mod agent_selector; 14 pub mod agent_stats; 15 pub mod agent_transport; 16 17 use crate::candidate::*; 18 use crate::error::*; 19 use crate::external_ip_mapper::*; 20 use crate::mdns::*; 21 use crate::network_type::*; 22 use crate::state::*; 23 use crate::udp_mux::UDPMux; 24 use crate::udp_network::UDPNetwork; 25 use crate::url::*; 26 use agent_config::*; 27 use agent_internal::*; 28 use agent_stats::*; 29 30 use mdns::conn::*; 31 use std::collections::HashMap; 32 use std::net::{Ipv4Addr, SocketAddr}; 33 use stun::{agent::*, attributes::*, fingerprint::*, integrity::*, message::*, xoraddr::*}; 34 use util::{vnet::net::*, Buffer}; 35 36 use crate::agent::agent_gather::GatherCandidatesInternalParams; 37 use crate::rand::*; 38 use crate::tcp_type::TcpType; 39 use std::future::Future; 40 use std::pin::Pin; 41 use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering}; 42 use std::sync::Arc; 43 use std::time::SystemTime; 44 use tokio::sync::{broadcast, mpsc, Mutex}; 45 use tokio::time::{Duration, Instant}; 46 47 #[derive(Debug, Clone)] 48 pub(crate) struct BindingRequest { 49 pub(crate) timestamp: Instant, 50 pub(crate) transaction_id: TransactionId, 51 pub(crate) destination: SocketAddr, 52 pub(crate) is_use_candidate: bool, 53 } 54 55 impl Default for BindingRequest { 56 fn default() -> Self { 57 Self { 58 timestamp: Instant::now(), 59 transaction_id: TransactionId::default(), 60 destination: SocketAddr::new(Ipv4Addr::new(0, 0, 0, 0).into(), 0), 61 is_use_candidate: false, 62 } 63 } 64 } 65 66 pub type OnConnectionStateChangeHdlrFn = Box< 67 dyn (FnMut(ConnectionState) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) 68 + Send 69 + Sync, 70 >; 71 pub type OnSelectedCandidatePairChangeHdlrFn = Box< 72 dyn (FnMut( 73 &Arc<dyn Candidate + Send + Sync>, 74 &Arc<dyn Candidate + Send + Sync>, 75 ) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) 76 + Send 77 + Sync, 78 >; 79 pub type OnCandidateHdlrFn = Box< 80 dyn (FnMut( 81 Option<Arc<dyn Candidate + Send + Sync>>, 82 ) -> Pin<Box<dyn Future<Output = ()> + Send + 'static>>) 83 + Send 84 + Sync, 85 >; 86 pub type GatherCandidateCancelFn = Box<dyn Fn() + Send + Sync>; 87 88 struct ChanReceivers { 89 chan_state_rx: mpsc::Receiver<ConnectionState>, 90 chan_candidate_rx: mpsc::Receiver<Option<Arc<dyn Candidate + Send + Sync>>>, 91 chan_candidate_pair_rx: mpsc::Receiver<()>, 92 } 93 94 /// Represents the ICE agent. 95 pub struct Agent { 96 pub(crate) internal: Arc<AgentInternal>, 97 98 pub(crate) udp_network: UDPNetwork, 99 pub(crate) interface_filter: Arc<Option<InterfaceFilterFn>>, 100 pub(crate) ip_filter: Arc<Option<IpFilterFn>>, 101 pub(crate) mdns_mode: MulticastDnsMode, 102 pub(crate) mdns_name: String, 103 pub(crate) mdns_conn: Option<Arc<DnsConn>>, 104 pub(crate) net: Arc<Net>, 105 106 // 1:1 D-NAT IP address mapping 107 pub(crate) ext_ip_mapper: Arc<Option<ExternalIpMapper>>, 108 pub(crate) gathering_state: Arc<AtomicU8>, //GatheringState, 109 pub(crate) candidate_types: Vec<CandidateType>, 110 pub(crate) urls: Vec<Url>, 111 pub(crate) network_types: Vec<NetworkType>, 112 113 pub(crate) gather_candidate_cancel: Option<GatherCandidateCancelFn>, 114 } 115 116 impl Agent { 117 /// Creates a new Agent. 118 pub async fn new(config: AgentConfig) -> Result<Self> { 119 let mut mdns_name = config.multicast_dns_host_name.clone(); 120 if mdns_name.is_empty() { 121 mdns_name = generate_multicast_dns_name(); 122 } 123 124 if !mdns_name.ends_with(".local") || mdns_name.split('.').count() != 2 { 125 return Err(Error::ErrInvalidMulticastDnshostName); 126 } 127 128 let mut mdns_mode = config.multicast_dns_mode; 129 if mdns_mode == MulticastDnsMode::Unspecified { 130 mdns_mode = MulticastDnsMode::QueryOnly; 131 } 132 133 let mdns_conn = 134 match create_multicast_dns(mdns_mode, &mdns_name, &config.multicast_dns_dest_addr) { 135 Ok(c) => c, 136 Err(err) => { 137 // Opportunistic mDNS: If we can't open the connection, that's ok: we 138 // can continue without it. 139 log::warn!("Failed to initialize mDNS {}: {}", mdns_name, err); 140 None 141 } 142 }; 143 144 let (mut ai, chan_receivers) = AgentInternal::new(&config); 145 let (chan_state_rx, chan_candidate_rx, chan_candidate_pair_rx) = ( 146 chan_receivers.chan_state_rx, 147 chan_receivers.chan_candidate_rx, 148 chan_receivers.chan_candidate_pair_rx, 149 ); 150 151 config.init_with_defaults(&mut ai); 152 153 let candidate_types = if config.candidate_types.is_empty() { 154 default_candidate_types() 155 } else { 156 config.candidate_types.clone() 157 }; 158 159 if ai.lite.load(Ordering::SeqCst) 160 && (candidate_types.len() != 1 || candidate_types[0] != CandidateType::Host) 161 { 162 Self::close_multicast_conn(&mdns_conn).await; 163 return Err(Error::ErrLiteUsingNonHostCandidates); 164 } 165 166 if !config.urls.is_empty() 167 && !contains_candidate_type(CandidateType::ServerReflexive, &candidate_types) 168 && !contains_candidate_type(CandidateType::Relay, &candidate_types) 169 { 170 Self::close_multicast_conn(&mdns_conn).await; 171 return Err(Error::ErrUselessUrlsProvided); 172 } 173 174 let ext_ip_mapper = match config.init_ext_ip_mapping(mdns_mode, &candidate_types) { 175 Ok(ext_ip_mapper) => ext_ip_mapper, 176 Err(err) => { 177 Self::close_multicast_conn(&mdns_conn).await; 178 return Err(err); 179 } 180 }; 181 182 let net = if let Some(net) = config.net { 183 if net.is_virtual() { 184 log::warn!("vnet is enabled"); 185 if mdns_mode != MulticastDnsMode::Disabled { 186 log::warn!("vnet does not support mDNS yet"); 187 } 188 } 189 190 net 191 } else { 192 Arc::new(Net::new(None)) 193 }; 194 195 let agent = Self { 196 udp_network: config.udp_network, 197 internal: Arc::new(ai), 198 interface_filter: Arc::clone(&config.interface_filter), 199 ip_filter: Arc::clone(&config.ip_filter), 200 mdns_mode, 201 mdns_name, 202 mdns_conn, 203 net, 204 ext_ip_mapper: Arc::new(ext_ip_mapper), 205 gathering_state: Arc::new(AtomicU8::new(0)), //GatheringState::New, 206 candidate_types, 207 urls: config.urls.clone(), 208 network_types: config.network_types.clone(), 209 210 gather_candidate_cancel: None, //TODO: add cancel 211 }; 212 213 agent 214 .internal 215 .start_on_connection_state_change_routine( 216 chan_state_rx, 217 chan_candidate_rx, 218 chan_candidate_pair_rx, 219 ) 220 .await; 221 222 // Restart is also used to initialize the agent for the first time 223 if let Err(err) = agent.restart(config.local_ufrag, config.local_pwd).await { 224 Self::close_multicast_conn(&agent.mdns_conn).await; 225 let _ = agent.close().await; 226 return Err(err); 227 } 228 229 Ok(agent) 230 } 231 232 pub async fn get_bytes_received(&self) -> usize { 233 self.internal.agent_conn.bytes_received() 234 } 235 236 pub async fn get_bytes_sent(&self) -> usize { 237 self.internal.agent_conn.bytes_sent() 238 } 239 240 /// Sets a handler that is fired when the connection state changes. 241 pub async fn on_connection_state_change(&self, f: OnConnectionStateChangeHdlrFn) { 242 let mut on_connection_state_change_hdlr = 243 self.internal.on_connection_state_change_hdlr.lock().await; 244 *on_connection_state_change_hdlr = Some(f); 245 } 246 247 /// Sets a handler that is fired when the final candidate pair is selected. 248 pub async fn on_selected_candidate_pair_change(&self, f: OnSelectedCandidatePairChangeHdlrFn) { 249 let mut on_selected_candidate_pair_change_hdlr = self 250 .internal 251 .on_selected_candidate_pair_change_hdlr 252 .lock() 253 .await; 254 *on_selected_candidate_pair_change_hdlr = Some(f); 255 } 256 257 /// Sets a handler that is fired when new candidates gathered. When the gathering process 258 /// complete the last candidate is nil. 259 pub async fn on_candidate(&self, f: OnCandidateHdlrFn) { 260 let mut on_candidate_hdlr = self.internal.on_candidate_hdlr.lock().await; 261 *on_candidate_hdlr = Some(f); 262 } 263 264 /// Adds a new remote candidate. 265 pub async fn add_remote_candidate(&self, c: &Arc<dyn Candidate + Send + Sync>) -> Result<()> { 266 // cannot check for network yet because it might not be applied 267 // when mDNS hostame is used. 268 if c.tcp_type() == TcpType::Active { 269 // TCP Candidates with tcptype active will probe server passive ones, so 270 // no need to do anything with them. 271 log::info!("Ignoring remote candidate with tcpType active: {}", c); 272 return Ok(()); 273 } 274 275 // If we have a mDNS Candidate lets fully resolve it before adding it locally 276 if c.candidate_type() == CandidateType::Host && c.address().ends_with(".local") { 277 if self.mdns_mode == MulticastDnsMode::Disabled { 278 log::warn!( 279 "remote mDNS candidate added, but mDNS is disabled: ({})", 280 c.address() 281 ); 282 return Ok(()); 283 } 284 285 if c.candidate_type() != CandidateType::Host { 286 return Err(Error::ErrAddressParseFailed); 287 } 288 289 let ai = Arc::clone(&self.internal); 290 let host_candidate = Arc::clone(c); 291 let mdns_conn = self.mdns_conn.clone(); 292 tokio::spawn(async move { 293 if let Some(mdns_conn) = mdns_conn { 294 if let Ok(candidate) = 295 Self::resolve_and_add_multicast_candidate(mdns_conn, host_candidate).await 296 { 297 ai.add_remote_candidate(&candidate).await; 298 } 299 } 300 }); 301 } else { 302 let ai = Arc::clone(&self.internal); 303 let candidate = Arc::clone(c); 304 tokio::spawn(async move { 305 ai.add_remote_candidate(&candidate).await; 306 }); 307 } 308 309 Ok(()) 310 } 311 312 /// Returns the local candidates. 313 pub async fn get_local_candidates(&self) -> Result<Vec<Arc<dyn Candidate + Send + Sync>>> { 314 let mut res = vec![]; 315 316 { 317 let local_candidates = self.internal.local_candidates.lock().await; 318 for candidates in local_candidates.values() { 319 for candidate in candidates { 320 res.push(Arc::clone(candidate)); 321 } 322 } 323 } 324 325 Ok(res) 326 } 327 328 /// Returns the local user credentials. 329 pub async fn get_local_user_credentials(&self) -> (String, String) { 330 let ufrag_pwd = self.internal.ufrag_pwd.lock().await; 331 (ufrag_pwd.local_ufrag.clone(), ufrag_pwd.local_pwd.clone()) 332 } 333 334 /// Returns the remote user credentials. 335 pub async fn get_remote_user_credentials(&self) -> (String, String) { 336 let ufrag_pwd = self.internal.ufrag_pwd.lock().await; 337 (ufrag_pwd.remote_ufrag.clone(), ufrag_pwd.remote_pwd.clone()) 338 } 339 340 /// Cleans up the Agent. 341 pub async fn close(&self) -> Result<()> { 342 if let Some(gather_candidate_cancel) = &self.gather_candidate_cancel { 343 gather_candidate_cancel(); 344 } 345 346 if let UDPNetwork::Muxed(ref udp_mux) = self.udp_network { 347 let (ufrag, _) = self.get_local_user_credentials().await; 348 udp_mux.remove_conn_by_ufrag(&ufrag).await; 349 } 350 351 //FIXME: deadlock here 352 self.internal.close().await 353 } 354 355 /// Returns the selected pair or nil if there is none 356 pub async fn get_selected_candidate_pair(&self) -> Option<Arc<CandidatePair>> { 357 self.internal.agent_conn.get_selected_pair().await 358 } 359 360 /// Sets the credentials of the remote agent. 361 pub async fn set_remote_credentials( 362 &self, 363 remote_ufrag: String, 364 remote_pwd: String, 365 ) -> Result<()> { 366 self.internal 367 .set_remote_credentials(remote_ufrag, remote_pwd) 368 .await 369 } 370 371 /// Restarts the ICE Agent with the provided ufrag/pwd 372 /// If no ufrag/pwd is provided the Agent will generate one itself. 373 /// 374 /// Restart must only be called when `GatheringState` is `GatheringStateComplete` 375 /// a user must then call `GatherCandidates` explicitly to start generating new ones. 376 pub async fn restart(&self, mut ufrag: String, mut pwd: String) -> Result<()> { 377 if ufrag.is_empty() { 378 ufrag = generate_ufrag(); 379 } 380 if pwd.is_empty() { 381 pwd = generate_pwd(); 382 } 383 384 if ufrag.len() * 8 < 24 { 385 return Err(Error::ErrLocalUfragInsufficientBits); 386 } 387 if pwd.len() * 8 < 128 { 388 return Err(Error::ErrLocalPwdInsufficientBits); 389 } 390 391 if GatheringState::from(self.gathering_state.load(Ordering::SeqCst)) 392 == GatheringState::Gathering 393 { 394 return Err(Error::ErrRestartWhenGathering); 395 } 396 self.gathering_state 397 .store(GatheringState::New as u8, Ordering::SeqCst); 398 399 { 400 let done_tx = self.internal.done_tx.lock().await; 401 if done_tx.is_none() { 402 return Err(Error::ErrClosed); 403 } 404 } 405 406 // Clear all agent needed to take back to fresh state 407 { 408 let mut ufrag_pwd = self.internal.ufrag_pwd.lock().await; 409 ufrag_pwd.local_ufrag = ufrag; 410 ufrag_pwd.local_pwd = pwd; 411 ufrag_pwd.remote_ufrag = String::new(); 412 ufrag_pwd.remote_pwd = String::new(); 413 } 414 { 415 let mut pending_binding_requests = self.internal.pending_binding_requests.lock().await; 416 *pending_binding_requests = vec![]; 417 } 418 419 { 420 let mut checklist = self.internal.agent_conn.checklist.lock().await; 421 *checklist = vec![]; 422 } 423 424 self.internal.set_selected_pair(None).await; 425 self.internal.delete_all_candidates().await; 426 self.internal.start().await; 427 428 // Restart is used by NewAgent. Accept/Connect should be used to move to checking 429 // for new Agents 430 if self.internal.connection_state.load(Ordering::SeqCst) != ConnectionState::New as u8 { 431 self.internal 432 .update_connection_state(ConnectionState::Checking) 433 .await; 434 } 435 436 Ok(()) 437 } 438 439 /// Initiates the trickle based gathering process. 440 pub async fn gather_candidates(&self) -> Result<()> { 441 if self.gathering_state.load(Ordering::SeqCst) != GatheringState::New as u8 { 442 return Err(Error::ErrMultipleGatherAttempted); 443 } 444 445 { 446 let on_candidate_hdlr = self.internal.on_candidate_hdlr.lock().await; 447 if on_candidate_hdlr.is_none() { 448 return Err(Error::ErrNoOnCandidateHandler); 449 } 450 } 451 452 if let Some(gather_candidate_cancel) = &self.gather_candidate_cancel { 453 gather_candidate_cancel(); // Cancel previous gathering routine 454 } 455 456 //TODO: a.gatherCandidateCancel = cancel 457 458 let params = GatherCandidatesInternalParams { 459 udp_network: self.udp_network.clone(), 460 candidate_types: self.candidate_types.clone(), 461 urls: self.urls.clone(), 462 network_types: self.network_types.clone(), 463 mdns_mode: self.mdns_mode, 464 mdns_name: self.mdns_name.clone(), 465 net: Arc::clone(&self.net), 466 interface_filter: self.interface_filter.clone(), 467 ip_filter: self.ip_filter.clone(), 468 ext_ip_mapper: Arc::clone(&self.ext_ip_mapper), 469 agent_internal: Arc::clone(&self.internal), 470 gathering_state: Arc::clone(&self.gathering_state), 471 chan_candidate_tx: Arc::clone(&self.internal.chan_candidate_tx), 472 }; 473 tokio::spawn(async move { 474 Self::gather_candidates_internal(params).await; 475 }); 476 477 Ok(()) 478 } 479 480 /// Returns a list of candidate pair stats. 481 pub async fn get_candidate_pairs_stats(&self) -> Vec<CandidatePairStats> { 482 self.internal.get_candidate_pairs_stats().await 483 } 484 485 /// Returns a list of local candidates stats. 486 pub async fn get_local_candidates_stats(&self) -> Vec<CandidateStats> { 487 self.internal.get_local_candidates_stats().await 488 } 489 490 /// Returns a list of remote candidates stats. 491 pub async fn get_remote_candidates_stats(&self) -> Vec<CandidateStats> { 492 self.internal.get_remote_candidates_stats().await 493 } 494 495 async fn resolve_and_add_multicast_candidate( 496 mdns_conn: Arc<DnsConn>, 497 c: Arc<dyn Candidate + Send + Sync>, 498 ) -> Result<Arc<dyn Candidate + Send + Sync>> { 499 //TODO: hook up _close_query_signal_tx to Agent or Candidate's Close signal? 500 let (_close_query_signal_tx, close_query_signal_rx) = mpsc::channel(1); 501 let src = match mdns_conn.query(&c.address(), close_query_signal_rx).await { 502 Ok((_, src)) => src, 503 Err(err) => { 504 log::warn!("Failed to discover mDNS candidate {}: {}", c.address(), err); 505 return Err(err.into()); 506 } 507 }; 508 509 c.set_ip(&src.ip()).await?; 510 511 Ok(c) 512 } 513 514 async fn close_multicast_conn(mdns_conn: &Option<Arc<DnsConn>>) { 515 if let Some(conn) = mdns_conn { 516 if let Err(err) = conn.close().await { 517 log::warn!("failed to close mDNS Conn: {}", err); 518 } 519 } 520 } 521 } 522