1 use crate::agent::agent_internal::*; 2 use crate::candidate::*; 3 use crate::control::*; 4 use crate::priority::*; 5 use crate::use_candidate::*; 6 7 use stun::{agent::*, attributes::*, fingerprint::*, integrity::*, message::*, textattrs::*}; 8 9 use async_trait::async_trait; 10 use std::net::SocketAddr; 11 use std::sync::atomic::Ordering; 12 use std::sync::Arc; 13 use tokio::time::{Duration, Instant}; 14 15 #[async_trait] 16 trait ControllingSelector { 17 async fn start(&self); 18 async fn contact_candidates(&self); 19 async fn ping_candidate( 20 &self, 21 local: &Arc<dyn Candidate + Send + Sync>, 22 remote: &Arc<dyn Candidate + Send + Sync>, 23 ); 24 async fn handle_success_response( 25 &self, 26 m: &Message, 27 local: &Arc<dyn Candidate + Send + Sync>, 28 remote: &Arc<dyn Candidate + Send + Sync>, 29 remote_addr: SocketAddr, 30 ); 31 async fn handle_binding_request( 32 &self, 33 m: &Message, 34 local: &Arc<dyn Candidate + Send + Sync>, 35 remote: &Arc<dyn Candidate + Send + Sync>, 36 ); 37 } 38 39 #[async_trait] 40 trait ControlledSelector { 41 async fn start(&self); 42 async fn contact_candidates(&self); 43 async fn ping_candidate( 44 &self, 45 local: &Arc<dyn Candidate + Send + Sync>, 46 remote: &Arc<dyn Candidate + Send + Sync>, 47 ); 48 async fn handle_success_response( 49 &self, 50 m: &Message, 51 local: &Arc<dyn Candidate + Send + Sync>, 52 remote: &Arc<dyn Candidate + Send + Sync>, 53 remote_addr: SocketAddr, 54 ); 55 async fn handle_binding_request( 56 &self, 57 m: &Message, 58 local: &Arc<dyn Candidate + Send + Sync>, 59 remote: &Arc<dyn Candidate + Send + Sync>, 60 ); 61 } 62 63 impl AgentInternal { 64 async fn is_nominatable(&self, c: &Arc<dyn Candidate + Send + Sync>) -> bool { 65 let start_time = self.start_time.lock().await; 66 match c.candidate_type() { 67 CandidateType::Host => { 68 Instant::now() 69 .checked_duration_since(*start_time) 70 .unwrap_or_else(|| Duration::from_secs(0)) 71 .as_nanos() 72 > self.host_acceptance_min_wait.as_nanos() 73 } 74 CandidateType::ServerReflexive => { 75 Instant::now() 76 .checked_duration_since(*start_time) 77 .unwrap_or_else(|| Duration::from_secs(0)) 78 .as_nanos() 79 > self.srflx_acceptance_min_wait.as_nanos() 80 } 81 CandidateType::PeerReflexive => { 82 Instant::now() 83 .checked_duration_since(*start_time) 84 .unwrap_or_else(|| Duration::from_secs(0)) 85 .as_nanos() 86 > self.prflx_acceptance_min_wait.as_nanos() 87 } 88 CandidateType::Relay => { 89 Instant::now() 90 .checked_duration_since(*start_time) 91 .unwrap_or_else(|| Duration::from_secs(0)) 92 .as_nanos() 93 > self.relay_acceptance_min_wait.as_nanos() 94 } 95 CandidateType::Unspecified => { 96 log::error!( 97 "is_nominatable invalid candidate type {}", 98 c.candidate_type() 99 ); 100 false 101 } 102 } 103 } 104 105 async fn nominate_pair(&self) { 106 let result = { 107 let nominated_pair = self.nominated_pair.lock().await; 108 if let Some(pair) = &*nominated_pair { 109 // The controlling agent MUST include the USE-CANDIDATE attribute in 110 // order to nominate a candidate pair (Section 8.1.1). The controlled 111 // agent MUST NOT include the USE-CANDIDATE attribute in a Binding 112 // request. 113 114 let (msg, result) = { 115 let ufrag_pwd = self.ufrag_pwd.lock().await; 116 let username = 117 ufrag_pwd.remote_ufrag.clone() + ":" + ufrag_pwd.local_ufrag.as_str(); 118 let mut msg = Message::new(); 119 let result = msg.build(&[ 120 Box::new(BINDING_REQUEST), 121 Box::new(TransactionId::new()), 122 Box::new(Username::new(ATTR_USERNAME, username)), 123 Box::new(UseCandidateAttr::default()), 124 Box::new(AttrControlling(self.tie_breaker.load(Ordering::SeqCst))), 125 Box::new(PriorityAttr(pair.local.priority())), 126 Box::new(MessageIntegrity::new_short_term_integrity( 127 ufrag_pwd.remote_pwd.clone(), 128 )), 129 Box::new(FINGERPRINT), 130 ]); 131 (msg, result) 132 }; 133 134 if let Err(err) = result { 135 log::error!("{}", err); 136 None 137 } else { 138 log::trace!( 139 "ping STUN (nominate candidate pair from {} to {}", 140 pair.local, 141 pair.remote 142 ); 143 let local = pair.local.clone(); 144 let remote = pair.remote.clone(); 145 Some((msg, local, remote)) 146 } 147 } else { 148 None 149 } 150 }; 151 152 if let Some((msg, local, remote)) = result { 153 self.send_binding_request(&msg, &local, &remote).await; 154 } 155 } 156 157 pub(crate) async fn start(&self) { 158 if self.is_controlling.load(Ordering::SeqCst) { 159 ControllingSelector::start(self).await; 160 } else { 161 ControlledSelector::start(self).await; 162 } 163 } 164 165 pub(crate) async fn contact_candidates(&self) { 166 if self.is_controlling.load(Ordering::SeqCst) { 167 ControllingSelector::contact_candidates(self).await; 168 } else { 169 ControlledSelector::contact_candidates(self).await; 170 } 171 } 172 173 pub(crate) async fn ping_candidate( 174 &self, 175 local: &Arc<dyn Candidate + Send + Sync>, 176 remote: &Arc<dyn Candidate + Send + Sync>, 177 ) { 178 if self.is_controlling.load(Ordering::SeqCst) { 179 ControllingSelector::ping_candidate(self, local, remote).await; 180 } else { 181 ControlledSelector::ping_candidate(self, local, remote).await; 182 } 183 } 184 185 pub(crate) async fn handle_success_response( 186 &self, 187 m: &Message, 188 local: &Arc<dyn Candidate + Send + Sync>, 189 remote: &Arc<dyn Candidate + Send + Sync>, 190 remote_addr: SocketAddr, 191 ) { 192 if self.is_controlling.load(Ordering::SeqCst) { 193 ControllingSelector::handle_success_response(self, m, local, remote, remote_addr).await; 194 } else { 195 ControlledSelector::handle_success_response(self, m, local, remote, remote_addr).await; 196 } 197 } 198 199 pub(crate) async fn handle_binding_request( 200 &self, 201 m: &Message, 202 local: &Arc<dyn Candidate + Send + Sync>, 203 remote: &Arc<dyn Candidate + Send + Sync>, 204 ) { 205 if self.is_controlling.load(Ordering::SeqCst) { 206 ControllingSelector::handle_binding_request(self, m, local, remote).await; 207 } else { 208 ControlledSelector::handle_binding_request(self, m, local, remote).await; 209 } 210 } 211 } 212 213 #[async_trait] 214 impl ControllingSelector for AgentInternal { 215 async fn start(&self) { 216 { 217 let mut nominated_pair = self.nominated_pair.lock().await; 218 *nominated_pair = None; 219 } 220 { 221 let mut start_time = self.start_time.lock().await; 222 *start_time = Instant::now(); 223 } 224 } 225 226 async fn contact_candidates(&self) { 227 // A lite selector should not contact candidates 228 if self.lite.load(Ordering::SeqCst) { 229 // This only happens if both peers are lite. See RFC 8445 S6.1.1 and S6.2 230 log::trace!("now falling back to full agent"); 231 } 232 233 let nominated_pair_is_some = { 234 let nominated_pair = self.nominated_pair.lock().await; 235 nominated_pair.is_some() 236 }; 237 238 if self.agent_conn.get_selected_pair().await.is_some() { 239 if self.validate_selected_pair().await { 240 log::trace!("[{}]: checking keepalive", self.get_name()); 241 self.check_keepalive().await; 242 } 243 } else if nominated_pair_is_some { 244 self.nominate_pair().await; 245 } else { 246 let has_nominated_pair = 247 if let Some(p) = self.agent_conn.get_best_valid_candidate_pair().await { 248 self.is_nominatable(&p.local).await && self.is_nominatable(&p.remote).await 249 } else { 250 false 251 }; 252 253 if has_nominated_pair { 254 if let Some(p) = self.agent_conn.get_best_valid_candidate_pair().await { 255 log::trace!( 256 "Nominatable pair found, nominating ({}, {})", 257 p.local.to_string(), 258 p.remote.to_string() 259 ); 260 p.nominated.store(true, Ordering::SeqCst); 261 { 262 let mut nominated_pair = self.nominated_pair.lock().await; 263 *nominated_pair = Some(p); 264 } 265 } 266 267 self.nominate_pair().await; 268 } else { 269 self.ping_all_candidates().await; 270 } 271 } 272 } 273 274 async fn ping_candidate( 275 &self, 276 local: &Arc<dyn Candidate + Send + Sync>, 277 remote: &Arc<dyn Candidate + Send + Sync>, 278 ) { 279 let (msg, result) = { 280 let ufrag_pwd = self.ufrag_pwd.lock().await; 281 let username = ufrag_pwd.remote_ufrag.clone() + ":" + ufrag_pwd.local_ufrag.as_str(); 282 let mut msg = Message::new(); 283 let result = msg.build(&[ 284 Box::new(BINDING_REQUEST), 285 Box::new(TransactionId::new()), 286 Box::new(Username::new(ATTR_USERNAME, username)), 287 Box::new(AttrControlling(self.tie_breaker.load(Ordering::SeqCst))), 288 Box::new(PriorityAttr(local.priority())), 289 Box::new(MessageIntegrity::new_short_term_integrity( 290 ufrag_pwd.remote_pwd.clone(), 291 )), 292 Box::new(FINGERPRINT), 293 ]); 294 (msg, result) 295 }; 296 297 if let Err(err) = result { 298 log::error!("{}", err); 299 } else { 300 self.send_binding_request(&msg, local, remote).await; 301 } 302 } 303 304 async fn handle_success_response( 305 &self, 306 m: &Message, 307 local: &Arc<dyn Candidate + Send + Sync>, 308 remote: &Arc<dyn Candidate + Send + Sync>, 309 remote_addr: SocketAddr, 310 ) { 311 if let Some(pending_request) = self.handle_inbound_binding_success(m.transaction_id).await { 312 let transaction_addr = pending_request.destination; 313 314 // Assert that NAT is not symmetric 315 // https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1 316 if transaction_addr != remote_addr { 317 log::debug!("discard message: transaction source and destination does not match expected({}), actual({})", transaction_addr, remote); 318 return; 319 } 320 321 log::trace!( 322 "inbound STUN (SuccessResponse) from {} to {}", 323 remote, 324 local 325 ); 326 let selected_pair_is_none = self.agent_conn.get_selected_pair().await.is_none(); 327 328 if let Some(p) = self.find_pair(local, remote).await { 329 p.state 330 .store(CandidatePairState::Succeeded as u8, Ordering::SeqCst); 331 log::trace!( 332 "Found valid candidate pair: {}, p.state: {}, isUseCandidate: {}, {}", 333 p, 334 p.state.load(Ordering::SeqCst), 335 pending_request.is_use_candidate, 336 selected_pair_is_none 337 ); 338 if pending_request.is_use_candidate && selected_pair_is_none { 339 self.set_selected_pair(Some(Arc::clone(&p))).await; 340 } 341 } else { 342 // This shouldn't happen 343 log::error!("Success response from invalid candidate pair"); 344 } 345 } else { 346 log::warn!( 347 "discard message from ({}), unknown TransactionID 0x{:?}", 348 remote, 349 m.transaction_id 350 ); 351 } 352 } 353 354 async fn handle_binding_request( 355 &self, 356 m: &Message, 357 local: &Arc<dyn Candidate + Send + Sync>, 358 remote: &Arc<dyn Candidate + Send + Sync>, 359 ) { 360 self.send_binding_success(m, local, remote).await; 361 log::trace!("controllingSelector: sendBindingSuccess"); 362 363 if let Some(p) = self.find_pair(local, remote).await { 364 let nominated_pair_is_none = { 365 let nominated_pair = self.nominated_pair.lock().await; 366 nominated_pair.is_none() 367 }; 368 369 log::trace!( 370 "controllingSelector: after findPair {}, p.state: {}, {}", 371 p, 372 p.state.load(Ordering::SeqCst), 373 nominated_pair_is_none, 374 //self.agent_conn.get_selected_pair().await.is_none() //, {} 375 ); 376 if p.state.load(Ordering::SeqCst) == CandidatePairState::Succeeded as u8 377 && nominated_pair_is_none 378 && self.agent_conn.get_selected_pair().await.is_none() 379 { 380 if let Some(best_pair) = self.agent_conn.get_best_available_candidate_pair().await { 381 log::trace!( 382 "controllingSelector: getBestAvailableCandidatePair {}", 383 best_pair 384 ); 385 if best_pair == p 386 && self.is_nominatable(&p.local).await 387 && self.is_nominatable(&p.remote).await 388 { 389 log::trace!("The candidate ({}, {}) is the best candidate available, marking it as nominated", 390 p.local, p.remote); 391 { 392 let mut nominated_pair = self.nominated_pair.lock().await; 393 *nominated_pair = Some(p); 394 } 395 self.nominate_pair().await; 396 } 397 } else { 398 log::trace!("No best pair available"); 399 } 400 } 401 } else { 402 log::trace!("controllingSelector: addPair"); 403 self.add_pair(local.clone(), remote.clone()).await; 404 } 405 } 406 } 407 408 #[async_trait] 409 impl ControlledSelector for AgentInternal { 410 async fn start(&self) {} 411 412 async fn contact_candidates(&self) { 413 // A lite selector should not contact candidates 414 if self.lite.load(Ordering::SeqCst) { 415 self.validate_selected_pair().await; 416 } else if self.agent_conn.get_selected_pair().await.is_some() { 417 if self.validate_selected_pair().await { 418 log::trace!("[{}]: checking keepalive", self.get_name()); 419 self.check_keepalive().await; 420 } 421 } else { 422 self.ping_all_candidates().await; 423 } 424 } 425 426 async fn ping_candidate( 427 &self, 428 local: &Arc<dyn Candidate + Send + Sync>, 429 remote: &Arc<dyn Candidate + Send + Sync>, 430 ) { 431 let (msg, result) = { 432 let ufrag_pwd = self.ufrag_pwd.lock().await; 433 let username = ufrag_pwd.remote_ufrag.clone() + ":" + ufrag_pwd.local_ufrag.as_str(); 434 let mut msg = Message::new(); 435 let result = msg.build(&[ 436 Box::new(BINDING_REQUEST), 437 Box::new(TransactionId::new()), 438 Box::new(Username::new(ATTR_USERNAME, username)), 439 Box::new(AttrControlled(self.tie_breaker.load(Ordering::SeqCst))), 440 Box::new(PriorityAttr(local.priority())), 441 Box::new(MessageIntegrity::new_short_term_integrity( 442 ufrag_pwd.remote_pwd.clone(), 443 )), 444 Box::new(FINGERPRINT), 445 ]); 446 (msg, result) 447 }; 448 449 if let Err(err) = result { 450 log::error!("{}", err); 451 } else { 452 self.send_binding_request(&msg, local, remote).await; 453 } 454 } 455 456 async fn handle_success_response( 457 &self, 458 m: &Message, 459 local: &Arc<dyn Candidate + Send + Sync>, 460 remote: &Arc<dyn Candidate + Send + Sync>, 461 remote_addr: SocketAddr, 462 ) { 463 // https://tools.ietf.org/html/rfc8445#section-7.3.1.5 464 // If the controlled agent does not accept the request from the 465 // controlling agent, the controlled agent MUST reject the nomination 466 // request with an appropriate error code response (e.g., 400) 467 // [RFC5389]. 468 469 if let Some(pending_request) = self.handle_inbound_binding_success(m.transaction_id).await { 470 let transaction_addr = pending_request.destination; 471 472 // Assert that NAT is not symmetric 473 // https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1 474 if transaction_addr != remote_addr { 475 log::debug!("discard message: transaction source and destination does not match expected({}), actual({})", transaction_addr, remote); 476 return; 477 } 478 479 log::trace!( 480 "inbound STUN (SuccessResponse) from {} to {}", 481 remote, 482 local 483 ); 484 485 if let Some(p) = self.find_pair(local, remote).await { 486 p.state 487 .store(CandidatePairState::Succeeded as u8, Ordering::SeqCst); 488 log::trace!("Found valid candidate pair: {}", p); 489 } else { 490 // This shouldn't happen 491 log::error!("Success response from invalid candidate pair"); 492 } 493 } else { 494 log::warn!( 495 "discard message from ({}), unknown TransactionID 0x{:?}", 496 remote, 497 m.transaction_id 498 ); 499 } 500 } 501 502 async fn handle_binding_request( 503 &self, 504 m: &Message, 505 local: &Arc<dyn Candidate + Send + Sync>, 506 remote: &Arc<dyn Candidate + Send + Sync>, 507 ) { 508 if self.find_pair(local, remote).await.is_none() { 509 self.add_pair(local.clone(), remote.clone()).await; 510 } 511 512 if let Some(p) = self.find_pair(local, remote).await { 513 let use_candidate = m.contains(ATTR_USE_CANDIDATE); 514 if use_candidate { 515 // https://tools.ietf.org/html/rfc8445#section-7.3.1.5 516 517 if p.state.load(Ordering::SeqCst) == CandidatePairState::Succeeded as u8 { 518 // If the state of this pair is Succeeded, it means that the check 519 // previously sent by this pair produced a successful response and 520 // generated a valid pair (Section 7.2.5.3.2). The agent sets the 521 // nominated flag value of the valid pair to true. 522 if self.agent_conn.get_selected_pair().await.is_none() { 523 self.set_selected_pair(Some(Arc::clone(&p))).await; 524 } 525 self.send_binding_success(m, local, remote).await; 526 } else { 527 // If the received Binding request triggered a new check to be 528 // enqueued in the triggered-check queue (Section 7.3.1.4), once the 529 // check is sent and if it generates a successful response, and 530 // generates a valid pair, the agent sets the nominated flag of the 531 // pair to true. If the request fails (Section 7.2.5.2), the agent 532 // MUST remove the candidate pair from the valid list, set the 533 // candidate pair state to Failed, and set the checklist state to 534 // Failed. 535 self.ping_candidate(local, remote).await; 536 } 537 } else { 538 self.send_binding_success(m, local, remote).await; 539 self.ping_candidate(local, remote).await; 540 } 541 } 542 } 543 } 544