1 use super::agent_vnet_test::*; 2 use super::*; 3 use crate::candidate::candidate_base::*; 4 use crate::candidate::candidate_host::*; 5 use crate::candidate::candidate_peer_reflexive::*; 6 use crate::candidate::candidate_relay::*; 7 use crate::candidate::candidate_server_reflexive::*; 8 use crate::control::AttrControlling; 9 use crate::priority::PriorityAttr; 10 use crate::use_candidate::UseCandidateAttr; 11 12 use crate::agent::agent_transport_test::pipe; 13 use async_trait::async_trait; 14 use std::net::Ipv4Addr; 15 use std::ops::Sub; 16 use std::str::FromStr; 17 use stun::message::*; 18 use stun::textattrs::Username; 19 use util::{vnet::*, Conn}; 20 use waitgroup::{WaitGroup, Worker}; 21 22 #[tokio::test] 23 async fn test_pair_search() -> Result<()> { 24 let config = AgentConfig::default(); 25 let a = Agent::new(config).await?; 26 27 { 28 { 29 let checklist = a.internal.agent_conn.checklist.lock().await; 30 assert!( 31 checklist.is_empty(), 32 "TestPairSearch is only a valid test if a.validPairs is empty on construction" 33 ); 34 } 35 36 let cp = a 37 .internal 38 .agent_conn 39 .get_best_available_candidate_pair() 40 .await; 41 assert!(cp.is_none(), "No Candidate pairs should exist"); 42 } 43 44 a.close().await?; 45 46 Ok(()) 47 } 48 49 #[tokio::test] 50 async fn test_pair_priority() -> Result<()> { 51 let a = Agent::new(AgentConfig::default()).await?; 52 53 let host_config = CandidateHostConfig { 54 base_config: CandidateBaseConfig { 55 network: "udp".to_owned(), 56 address: "192.168.1.1".to_owned(), 57 port: 19216, 58 component: 1, 59 ..Default::default() 60 }, 61 ..Default::default() 62 }; 63 let host_local: Arc<dyn Candidate + Send + Sync> = Arc::new(host_config.new_candidate_host()?); 64 65 let relay_config = CandidateRelayConfig { 66 base_config: CandidateBaseConfig { 67 network: "udp".to_owned(), 68 address: "1.2.3.4".to_owned(), 69 port: 12340, 70 component: 1, 71 ..Default::default() 72 }, 73 rel_addr: "4.3.2.1".to_owned(), 74 rel_port: 43210, 75 ..Default::default() 76 }; 77 78 let relay_remote = relay_config.new_candidate_relay()?; 79 80 let srflx_config = CandidateServerReflexiveConfig { 81 base_config: CandidateBaseConfig { 82 network: "udp".to_owned(), 83 address: "10.10.10.2".to_owned(), 84 port: 19218, 85 component: 1, 86 ..Default::default() 87 }, 88 rel_addr: "4.3.2.1".to_owned(), 89 rel_port: 43212, 90 }; 91 92 let srflx_remote = srflx_config.new_candidate_server_reflexive()?; 93 94 let prflx_config = CandidatePeerReflexiveConfig { 95 base_config: CandidateBaseConfig { 96 network: "udp".to_owned(), 97 address: "10.10.10.2".to_owned(), 98 port: 19217, 99 component: 1, 100 ..Default::default() 101 }, 102 rel_addr: "4.3.2.1".to_owned(), 103 rel_port: 43211, 104 }; 105 106 let prflx_remote = prflx_config.new_candidate_peer_reflexive()?; 107 108 let host_config = CandidateHostConfig { 109 base_config: CandidateBaseConfig { 110 network: "udp".to_owned(), 111 address: "1.2.3.5".to_owned(), 112 port: 12350, 113 component: 1, 114 ..Default::default() 115 }, 116 ..Default::default() 117 }; 118 let host_remote = host_config.new_candidate_host()?; 119 120 let remotes: Vec<Arc<dyn Candidate + Send + Sync>> = vec![ 121 Arc::new(relay_remote), 122 Arc::new(srflx_remote), 123 Arc::new(prflx_remote), 124 Arc::new(host_remote), 125 ]; 126 127 { 128 for remote in remotes { 129 if a.internal.find_pair(&host_local, &remote).await.is_none() { 130 a.internal 131 .add_pair(host_local.clone(), remote.clone()) 132 .await; 133 } 134 135 if let Some(p) = a.internal.find_pair(&host_local, &remote).await { 136 p.state 137 .store(CandidatePairState::Succeeded as u8, Ordering::SeqCst); 138 } 139 140 if let Some(best_pair) = a 141 .internal 142 .agent_conn 143 .get_best_available_candidate_pair() 144 .await 145 { 146 assert_eq!( 147 best_pair.to_string(), 148 CandidatePair { 149 remote: remote.clone(), 150 local: host_local.clone(), 151 ..Default::default() 152 } 153 .to_string(), 154 "Unexpected bestPair {} (expected remote: {})", 155 best_pair, 156 remote, 157 ); 158 } else { 159 panic!("expected Some, but got None"); 160 } 161 } 162 } 163 164 a.close().await?; 165 Ok(()) 166 } 167 168 #[tokio::test] 169 async fn test_agent_get_stats() -> Result<()> { 170 let (conn_a, conn_b, agent_a, agent_b) = pipe(None, None).await?; 171 assert_eq!(agent_a.get_bytes_received(), 0); 172 assert_eq!(agent_a.get_bytes_sent(), 0); 173 assert_eq!(agent_b.get_bytes_received(), 0); 174 assert_eq!(agent_b.get_bytes_sent(), 0); 175 176 let _na = conn_a.send(&[0u8; 10]).await?; 177 let mut buf = vec![0u8; 10]; 178 let _nb = conn_b.recv(&mut buf).await?; 179 180 assert_eq!(agent_a.get_bytes_received(), 0); 181 assert_eq!(agent_a.get_bytes_sent(), 10); 182 183 assert_eq!(agent_b.get_bytes_received(), 10); 184 assert_eq!(agent_b.get_bytes_sent(), 0); 185 186 Ok(()) 187 } 188 189 #[tokio::test] 190 async fn test_on_selected_candidate_pair_change() -> Result<()> { 191 let a = Agent::new(AgentConfig::default()).await?; 192 let (callback_called_tx, mut callback_called_rx) = mpsc::channel::<()>(1); 193 let callback_called_tx = Arc::new(Mutex::new(Some(callback_called_tx))); 194 let cb: OnSelectedCandidatePairChangeHdlrFn = Box::new(move |_, _| { 195 let callback_called_tx_clone = Arc::clone(&callback_called_tx); 196 Box::pin(async move { 197 let mut tx = callback_called_tx_clone.lock().await; 198 tx.take(); 199 }) 200 }); 201 a.on_selected_candidate_pair_change(cb); 202 203 let host_config = CandidateHostConfig { 204 base_config: CandidateBaseConfig { 205 network: "udp".to_owned(), 206 address: "192.168.1.1".to_owned(), 207 port: 19216, 208 component: 1, 209 ..Default::default() 210 }, 211 ..Default::default() 212 }; 213 let host_local = host_config.new_candidate_host()?; 214 215 let relay_config = CandidateRelayConfig { 216 base_config: CandidateBaseConfig { 217 network: "udp".to_owned(), 218 address: "1.2.3.4".to_owned(), 219 port: 12340, 220 component: 1, 221 ..Default::default() 222 }, 223 rel_addr: "4.3.2.1".to_owned(), 224 rel_port: 43210, 225 ..Default::default() 226 }; 227 let relay_remote = relay_config.new_candidate_relay()?; 228 229 // select the pair 230 let p = Arc::new(CandidatePair::new( 231 Arc::new(host_local), 232 Arc::new(relay_remote), 233 false, 234 )); 235 a.internal.set_selected_pair(Some(p)).await; 236 237 // ensure that the callback fired on setting the pair 238 let _ = callback_called_rx.recv().await; 239 240 a.close().await?; 241 Ok(()) 242 } 243 244 #[tokio::test] 245 async fn test_handle_peer_reflexive_udp_pflx_candidate() -> Result<()> { 246 let a = Agent::new(AgentConfig::default()).await?; 247 248 let host_config = CandidateHostConfig { 249 base_config: CandidateBaseConfig { 250 network: "udp".to_owned(), 251 address: "192.168.0.2".to_owned(), 252 port: 777, 253 component: 1, 254 conn: Some(Arc::new(MockConn {})), 255 ..Default::default() 256 }, 257 ..Default::default() 258 }; 259 260 let local: Arc<dyn Candidate + Send + Sync> = Arc::new(host_config.new_candidate_host()?); 261 let remote = SocketAddr::from_str("172.17.0.3:999")?; 262 263 let (username, local_pwd, tie_breaker) = { 264 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 265 ( 266 ufrag_pwd.local_ufrag.to_owned() + ":" + ufrag_pwd.remote_ufrag.as_str(), 267 ufrag_pwd.local_pwd.clone(), 268 a.internal.tie_breaker.load(Ordering::SeqCst), 269 ) 270 }; 271 272 let mut msg = Message::new(); 273 msg.build(&[ 274 Box::new(BINDING_REQUEST), 275 Box::new(TransactionId::new()), 276 Box::new(Username::new(ATTR_USERNAME, username)), 277 Box::new(UseCandidateAttr::new()), 278 Box::new(AttrControlling(tie_breaker)), 279 Box::new(PriorityAttr(local.priority())), 280 Box::new(MessageIntegrity::new_short_term_integrity(local_pwd)), 281 Box::new(FINGERPRINT), 282 ])?; 283 284 { 285 a.internal.handle_inbound(&mut msg, &local, remote).await; 286 287 let remote_candidates = a.internal.remote_candidates.lock().await; 288 // length of remote candidate list must be one now 289 assert_eq!( 290 remote_candidates.len(), 291 1, 292 "failed to add a network type to the remote candidate list" 293 ); 294 295 // length of remote candidate list for a network type must be 1 296 if let Some(cands) = remote_candidates.get(&local.network_type()) { 297 assert_eq!( 298 cands.len(), 299 1, 300 "failed to add prflx candidate to remote candidate list" 301 ); 302 303 let c = &cands[0]; 304 305 assert_eq!( 306 c.candidate_type(), 307 CandidateType::PeerReflexive, 308 "candidate type must be prflx" 309 ); 310 311 assert_eq!(c.address(), "172.17.0.3", "IP address mismatch"); 312 313 assert_eq!(c.port(), 999, "Port number mismatch"); 314 } else { 315 panic!( 316 "expected non-empty remote candidate for network type {}", 317 local.network_type() 318 ); 319 } 320 } 321 322 a.close().await?; 323 Ok(()) 324 } 325 326 #[tokio::test] 327 async fn test_handle_peer_reflexive_unknown_remote() -> Result<()> { 328 let a = Agent::new(AgentConfig::default()).await?; 329 330 let mut tid = TransactionId::default(); 331 tid.0[..3].copy_from_slice("ABC".as_bytes()); 332 333 let remote_pwd = { 334 { 335 let mut pending_binding_requests = a.internal.pending_binding_requests.lock().await; 336 *pending_binding_requests = vec![BindingRequest { 337 timestamp: Instant::now(), 338 transaction_id: tid, 339 destination: SocketAddr::from_str("0.0.0.0:0")?, 340 is_use_candidate: false, 341 }]; 342 } 343 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 344 ufrag_pwd.remote_pwd.clone() 345 }; 346 347 let host_config = CandidateHostConfig { 348 base_config: CandidateBaseConfig { 349 network: "udp".to_owned(), 350 address: "192.168.0.2".to_owned(), 351 port: 777, 352 component: 1, 353 conn: Some(Arc::new(MockConn {})), 354 ..Default::default() 355 }, 356 ..Default::default() 357 }; 358 359 let local: Arc<dyn Candidate + Send + Sync> = Arc::new(host_config.new_candidate_host()?); 360 let remote = SocketAddr::from_str("172.17.0.3:999")?; 361 362 let mut msg = Message::new(); 363 msg.build(&[ 364 Box::new(BINDING_SUCCESS), 365 Box::new(tid), 366 Box::new(MessageIntegrity::new_short_term_integrity(remote_pwd)), 367 Box::new(FINGERPRINT), 368 ])?; 369 370 { 371 a.internal.handle_inbound(&mut msg, &local, remote).await; 372 373 let remote_candidates = a.internal.remote_candidates.lock().await; 374 assert_eq!( 375 remote_candidates.len(), 376 0, 377 "unknown remote was able to create a candidate" 378 ); 379 } 380 381 a.close().await?; 382 Ok(()) 383 } 384 385 //use std::io::Write; 386 387 // Assert that Agent on startup sends message, and doesn't wait for connectivityTicker to fire 388 #[tokio::test] 389 async fn test_connectivity_on_startup() -> Result<()> { 390 /*env_logger::Builder::new() 391 .format(|buf, record| { 392 writeln!( 393 buf, 394 "{}:{} [{}] {} - {}", 395 record.file().unwrap_or("unknown"), 396 record.line().unwrap_or(0), 397 record.level(), 398 chrono::Local::now().format("%H:%M:%S.%6f"), 399 record.args() 400 ) 401 }) 402 .filter(None, log::LevelFilter::Trace) 403 .init();*/ 404 405 // Create a network with two interfaces 406 let wan = Arc::new(Mutex::new(router::Router::new(router::RouterConfig { 407 cidr: "0.0.0.0/0".to_owned(), 408 ..Default::default() 409 })?)); 410 411 let net0 = Arc::new(net::Net::new(Some(net::NetConfig { 412 static_ips: vec!["192.168.0.1".to_owned()], 413 ..Default::default() 414 }))); 415 let net1 = Arc::new(net::Net::new(Some(net::NetConfig { 416 static_ips: vec!["192.168.0.2".to_owned()], 417 ..Default::default() 418 }))); 419 420 connect_net2router(&net0, &wan).await?; 421 connect_net2router(&net1, &wan).await?; 422 start_router(&wan).await?; 423 424 let (a_notifier, mut a_connected) = on_connected(); 425 let (b_notifier, mut b_connected) = on_connected(); 426 427 let keepalive_interval = Some(Duration::from_secs(3600)); //time.Hour 428 let check_interval = Duration::from_secs(3600); //time.Hour 429 let cfg0 = AgentConfig { 430 network_types: supported_network_types(), 431 multicast_dns_mode: MulticastDnsMode::Disabled, 432 net: Some(net0), 433 434 keepalive_interval, 435 check_interval, 436 ..Default::default() 437 }; 438 439 let a_agent = Arc::new(Agent::new(cfg0).await?); 440 a_agent.on_connection_state_change(a_notifier); 441 442 let cfg1 = AgentConfig { 443 network_types: supported_network_types(), 444 multicast_dns_mode: MulticastDnsMode::Disabled, 445 net: Some(net1), 446 447 keepalive_interval, 448 check_interval, 449 ..Default::default() 450 }; 451 452 let b_agent = Arc::new(Agent::new(cfg1).await?); 453 b_agent.on_connection_state_change(b_notifier); 454 455 // Manual signaling 456 let (a_ufrag, a_pwd) = a_agent.get_local_user_credentials().await; 457 let (b_ufrag, b_pwd) = b_agent.get_local_user_credentials().await; 458 459 gather_and_exchange_candidates(&a_agent, &b_agent).await?; 460 461 let (accepted_tx, mut accepted_rx) = mpsc::channel::<()>(1); 462 let (accepting_tx, mut accepting_rx) = mpsc::channel::<()>(1); 463 let (_a_cancel_tx, a_cancel_rx) = mpsc::channel(1); 464 let (_b_cancel_tx, b_cancel_rx) = mpsc::channel(1); 465 466 let accepting_tx = Arc::new(Mutex::new(Some(accepting_tx))); 467 a_agent.on_connection_state_change(Box::new(move |s: ConnectionState| { 468 let accepted_tx_clone = Arc::clone(&accepting_tx); 469 Box::pin(async move { 470 if s == ConnectionState::Checking { 471 let mut tx = accepted_tx_clone.lock().await; 472 tx.take(); 473 } 474 }) 475 })); 476 477 tokio::spawn(async move { 478 let result = a_agent.accept(a_cancel_rx, b_ufrag, b_pwd).await; 479 assert!(result.is_ok(), "agent accept expected OK"); 480 drop(accepted_tx); 481 }); 482 483 let _ = accepting_rx.recv().await; 484 485 let _ = b_agent.dial(b_cancel_rx, a_ufrag, a_pwd).await?; 486 487 // Ensure accepted 488 let _ = accepted_rx.recv().await; 489 490 // Ensure pair selected 491 // Note: this assumes ConnectionStateConnected is thrown after selecting the final pair 492 let _ = a_connected.recv().await; 493 let _ = b_connected.recv().await; 494 495 { 496 let mut w = wan.lock().await; 497 w.stop().await?; 498 } 499 500 Ok(()) 501 } 502 503 #[tokio::test] 504 async fn test_connectivity_lite() -> Result<()> { 505 /*env_logger::Builder::new() 506 .format(|buf, record| { 507 writeln!( 508 buf, 509 "{}:{} [{}] {} - {}", 510 record.file().unwrap_or("unknown"), 511 record.line().unwrap_or(0), 512 record.level(), 513 chrono::Local::now().format("%H:%M:%S.%6f"), 514 record.args() 515 ) 516 }) 517 .filter(None, log::LevelFilter::Trace) 518 .init();*/ 519 520 let stun_server_url = Url { 521 scheme: SchemeType::Stun, 522 host: "1.2.3.4".to_owned(), 523 port: 3478, 524 proto: ProtoType::Udp, 525 ..Default::default() 526 }; 527 528 let nat_type = nat::NatType { 529 mapping_behavior: nat::EndpointDependencyType::EndpointIndependent, 530 filtering_behavior: nat::EndpointDependencyType::EndpointIndependent, 531 ..Default::default() 532 }; 533 534 let v = build_vnet(nat_type, nat_type).await?; 535 536 let (a_notifier, mut a_connected) = on_connected(); 537 let (b_notifier, mut b_connected) = on_connected(); 538 539 let cfg0 = AgentConfig { 540 urls: vec![stun_server_url], 541 network_types: supported_network_types(), 542 multicast_dns_mode: MulticastDnsMode::Disabled, 543 net: Some(Arc::clone(&v.net0)), 544 ..Default::default() 545 }; 546 547 let a_agent = Arc::new(Agent::new(cfg0).await?); 548 a_agent.on_connection_state_change(a_notifier); 549 550 let cfg1 = AgentConfig { 551 urls: vec![], 552 lite: true, 553 candidate_types: vec![CandidateType::Host], 554 network_types: supported_network_types(), 555 multicast_dns_mode: MulticastDnsMode::Disabled, 556 net: Some(Arc::clone(&v.net1)), 557 ..Default::default() 558 }; 559 560 let b_agent = Arc::new(Agent::new(cfg1).await?); 561 b_agent.on_connection_state_change(b_notifier); 562 563 let _ = connect_with_vnet(&a_agent, &b_agent).await?; 564 565 // Ensure pair selected 566 // Note: this assumes ConnectionStateConnected is thrown after selecting the final pair 567 let _ = a_connected.recv().await; 568 let _ = b_connected.recv().await; 569 570 v.close().await?; 571 572 Ok(()) 573 } 574 575 struct MockPacketConn; 576 577 #[async_trait] 578 impl Conn for MockPacketConn { 579 async fn connect(&self, _addr: SocketAddr) -> std::result::Result<(), util::Error> { 580 Ok(()) 581 } 582 583 async fn recv(&self, _buf: &mut [u8]) -> std::result::Result<usize, util::Error> { 584 Ok(0) 585 } 586 587 async fn recv_from( 588 &self, 589 _buf: &mut [u8], 590 ) -> std::result::Result<(usize, SocketAddr), util::Error> { 591 Ok((0, SocketAddr::new(Ipv4Addr::new(0, 0, 0, 0).into(), 0))) 592 } 593 594 async fn send(&self, _buf: &[u8]) -> std::result::Result<usize, util::Error> { 595 Ok(0) 596 } 597 598 async fn send_to( 599 &self, 600 _buf: &[u8], 601 _target: SocketAddr, 602 ) -> std::result::Result<usize, util::Error> { 603 Ok(0) 604 } 605 606 fn local_addr(&self) -> std::result::Result<SocketAddr, util::Error> { 607 Ok(SocketAddr::new(Ipv4Addr::new(0, 0, 0, 0).into(), 0)) 608 } 609 610 fn remote_addr(&self) -> Option<SocketAddr> { 611 None 612 } 613 614 async fn close(&self) -> std::result::Result<(), util::Error> { 615 Ok(()) 616 } 617 } 618 619 fn build_msg(c: MessageClass, username: String, key: String) -> Result<Message> { 620 let mut msg = Message::new(); 621 msg.build(&[ 622 Box::new(MessageType::new(METHOD_BINDING, c)), 623 Box::new(TransactionId::new()), 624 Box::new(Username::new(ATTR_USERNAME, username)), 625 Box::new(MessageIntegrity::new_short_term_integrity(key)), 626 Box::new(FINGERPRINT), 627 ])?; 628 Ok(msg) 629 } 630 631 #[tokio::test] 632 async fn test_inbound_validity() -> Result<()> { 633 /*env_logger::Builder::new() 634 .format(|buf, record| { 635 writeln!( 636 buf, 637 "{}:{} [{}] {} - {}", 638 record.file().unwrap_or("unknown"), 639 record.line().unwrap_or(0), 640 record.level(), 641 chrono::Local::now().format("%H:%M:%S.%6f"), 642 record.args() 643 ) 644 }) 645 .filter(None, log::LevelFilter::Trace) 646 .init();*/ 647 648 let remote = SocketAddr::from_str("172.17.0.3:999")?; 649 let local: Arc<dyn Candidate + Send + Sync> = Arc::new( 650 CandidateHostConfig { 651 base_config: CandidateBaseConfig { 652 network: "udp".to_owned(), 653 address: "192.168.0.2".to_owned(), 654 port: 777, 655 component: 1, 656 conn: Some(Arc::new(MockPacketConn {})), 657 ..Default::default() 658 }, 659 ..Default::default() 660 } 661 .new_candidate_host()?, 662 ); 663 664 //"Invalid Binding requests should be discarded" 665 { 666 let a = Agent::new(AgentConfig::default()).await?; 667 668 { 669 let local_pwd = { 670 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 671 ufrag_pwd.local_pwd.clone() 672 }; 673 a.internal 674 .handle_inbound( 675 &mut build_msg(CLASS_REQUEST, "invalid".to_owned(), local_pwd)?, 676 &local, 677 remote, 678 ) 679 .await; 680 { 681 let remote_candidates = a.internal.remote_candidates.lock().await; 682 assert_ne!( 683 remote_candidates.len(), 684 1, 685 "Binding with invalid Username was able to create prflx candidate" 686 ); 687 } 688 689 let username = { 690 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 691 format!("{}:{}", ufrag_pwd.local_ufrag, ufrag_pwd.remote_ufrag) 692 }; 693 a.internal 694 .handle_inbound( 695 &mut build_msg(CLASS_REQUEST, username, "Invalid".to_owned())?, 696 &local, 697 remote, 698 ) 699 .await; 700 { 701 let remote_candidates = a.internal.remote_candidates.lock().await; 702 assert_ne!( 703 remote_candidates.len(), 704 1, 705 "Binding with invalid MessageIntegrity was able to create prflx candidate" 706 ); 707 } 708 } 709 710 a.close().await?; 711 } 712 713 //"Invalid Binding success responses should be discarded" 714 { 715 let a = Agent::new(AgentConfig::default()).await?; 716 717 { 718 let username = { 719 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 720 format!("{}:{}", ufrag_pwd.local_ufrag, ufrag_pwd.remote_ufrag) 721 }; 722 a.internal 723 .handle_inbound( 724 &mut build_msg(CLASS_SUCCESS_RESPONSE, username, "Invalid".to_owned())?, 725 &local, 726 remote, 727 ) 728 .await; 729 { 730 let remote_candidates = a.internal.remote_candidates.lock().await; 731 assert_ne!( 732 remote_candidates.len(), 733 1, 734 "Binding with invalid Username was able to create prflx candidate" 735 ); 736 } 737 } 738 739 a.close().await?; 740 } 741 742 //"Discard non-binding messages" 743 { 744 let a = Agent::new(AgentConfig::default()).await?; 745 746 { 747 let username = { 748 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 749 format!("{}:{}", ufrag_pwd.local_ufrag, ufrag_pwd.remote_ufrag) 750 }; 751 a.internal 752 .handle_inbound( 753 &mut build_msg(CLASS_ERROR_RESPONSE, username, "Invalid".to_owned())?, 754 &local, 755 remote, 756 ) 757 .await; 758 let remote_candidates = a.internal.remote_candidates.lock().await; 759 assert_ne!( 760 remote_candidates.len(), 761 1, 762 "non-binding message was able to create prflxRemote" 763 ); 764 } 765 766 a.close().await?; 767 } 768 769 //"Valid bind request" 770 { 771 let a = Agent::new(AgentConfig::default()).await?; 772 773 { 774 let (username, local_pwd) = { 775 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 776 ( 777 format!("{}:{}", ufrag_pwd.local_ufrag, ufrag_pwd.remote_ufrag), 778 ufrag_pwd.local_pwd.clone(), 779 ) 780 }; 781 a.internal 782 .handle_inbound( 783 &mut build_msg(CLASS_REQUEST, username, local_pwd)?, 784 &local, 785 remote, 786 ) 787 .await; 788 let remote_candidates = a.internal.remote_candidates.lock().await; 789 assert_eq!( 790 remote_candidates.len(), 791 1, 792 "Binding with valid values was unable to create prflx candidate" 793 ); 794 } 795 796 a.close().await?; 797 } 798 799 //"Valid bind without fingerprint" 800 { 801 let a = Agent::new(AgentConfig::default()).await?; 802 803 { 804 let (username, local_pwd) = { 805 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 806 ( 807 format!("{}:{}", ufrag_pwd.local_ufrag, ufrag_pwd.remote_ufrag), 808 ufrag_pwd.local_pwd.clone(), 809 ) 810 }; 811 812 let mut msg = Message::new(); 813 msg.build(&[ 814 Box::new(BINDING_REQUEST), 815 Box::new(TransactionId::new()), 816 Box::new(Username::new(ATTR_USERNAME, username)), 817 Box::new(MessageIntegrity::new_short_term_integrity(local_pwd)), 818 ])?; 819 820 a.internal.handle_inbound(&mut msg, &local, remote).await; 821 let remote_candidates = a.internal.remote_candidates.lock().await; 822 assert_eq!( 823 remote_candidates.len(), 824 1, 825 "Binding with valid values (but no fingerprint) was unable to create prflx candidate" 826 ); 827 } 828 829 a.close().await?; 830 } 831 832 //"Success with invalid TransactionID" 833 { 834 let a = Agent::new(AgentConfig::default()).await?; 835 836 { 837 let remote = SocketAddr::from_str("172.17.0.3:999")?; 838 839 let mut t_id = TransactionId::default(); 840 t_id.0[..3].copy_from_slice(b"ABC"); 841 842 let remote_pwd = { 843 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 844 ufrag_pwd.remote_pwd.clone() 845 }; 846 847 let mut msg = Message::new(); 848 msg.build(&[ 849 Box::new(BINDING_SUCCESS), 850 Box::new(t_id), 851 Box::new(MessageIntegrity::new_short_term_integrity(remote_pwd)), 852 Box::new(FINGERPRINT), 853 ])?; 854 855 a.internal.handle_inbound(&mut msg, &local, remote).await; 856 857 { 858 let remote_candidates = a.internal.remote_candidates.lock().await; 859 assert_eq!( 860 remote_candidates.len(), 861 0, 862 "unknown remote was able to create a candidate" 863 ); 864 } 865 } 866 867 a.close().await?; 868 } 869 870 Ok(()) 871 } 872 873 #[tokio::test] 874 async fn test_invalid_agent_starts() -> Result<()> { 875 let a = Agent::new(AgentConfig::default()).await?; 876 877 let (_cancel_tx1, cancel_rx1) = mpsc::channel(1); 878 let result = a.dial(cancel_rx1, "".to_owned(), "bar".to_owned()).await; 879 assert!(result.is_err()); 880 if let Err(err) = result { 881 assert_eq!(Error::ErrRemoteUfragEmpty, err); 882 } 883 884 let (_cancel_tx2, cancel_rx2) = mpsc::channel(1); 885 let result = a.dial(cancel_rx2, "foo".to_owned(), "".to_owned()).await; 886 assert!(result.is_err()); 887 if let Err(err) = result { 888 assert_eq!(Error::ErrRemotePwdEmpty, err); 889 } 890 891 let (cancel_tx3, cancel_rx3) = mpsc::channel(1); 892 tokio::spawn(async move { 893 tokio::time::sleep(Duration::from_millis(100)).await; 894 drop(cancel_tx3); 895 }); 896 897 let result = a.dial(cancel_rx3, "foo".to_owned(), "bar".to_owned()).await; 898 assert!(result.is_err()); 899 if let Err(err) = result { 900 assert_eq!(Error::ErrCanceledByCaller, err); 901 } 902 903 let (_cancel_tx4, cancel_rx4) = mpsc::channel(1); 904 let result = a.dial(cancel_rx4, "foo".to_owned(), "bar".to_owned()).await; 905 assert!(result.is_err()); 906 if let Err(err) = result { 907 assert_eq!(Error::ErrMultipleStart, err); 908 } 909 910 a.close().await?; 911 912 Ok(()) 913 } 914 915 //use std::io::Write; 916 917 // Assert that Agent emits Connecting/Connected/Disconnected/Failed/Closed messages 918 #[tokio::test] 919 async fn test_connection_state_callback() -> Result<()> { 920 /*env_logger::Builder::new() 921 .format(|buf, record| { 922 writeln!( 923 buf, 924 "{}:{} [{}] {} - {}", 925 record.file().unwrap_or("unknown"), 926 record.line().unwrap_or(0), 927 record.level(), 928 chrono::Local::now().format("%H:%M:%S.%6f"), 929 record.args() 930 ) 931 }) 932 .filter(None, log::LevelFilter::Trace) 933 .init();*/ 934 935 let disconnected_duration = Duration::from_secs(1); 936 let failed_duration = Duration::from_secs(1); 937 let keepalive_interval = Duration::from_secs(0); 938 939 let cfg0 = AgentConfig { 940 urls: vec![], 941 network_types: supported_network_types(), 942 disconnected_timeout: Some(disconnected_duration), 943 failed_timeout: Some(failed_duration), 944 keepalive_interval: Some(keepalive_interval), 945 ..Default::default() 946 }; 947 948 let cfg1 = AgentConfig { 949 urls: vec![], 950 network_types: supported_network_types(), 951 disconnected_timeout: Some(disconnected_duration), 952 failed_timeout: Some(failed_duration), 953 keepalive_interval: Some(keepalive_interval), 954 ..Default::default() 955 }; 956 957 let a_agent = Arc::new(Agent::new(cfg0).await?); 958 let b_agent = Arc::new(Agent::new(cfg1).await?); 959 960 let (is_checking_tx, mut is_checking_rx) = mpsc::channel::<()>(1); 961 let (is_connected_tx, mut is_connected_rx) = mpsc::channel::<()>(1); 962 let (is_disconnected_tx, mut is_disconnected_rx) = mpsc::channel::<()>(1); 963 let (is_failed_tx, mut is_failed_rx) = mpsc::channel::<()>(1); 964 let (is_closed_tx, mut is_closed_rx) = mpsc::channel::<()>(1); 965 966 let is_checking_tx = Arc::new(Mutex::new(Some(is_checking_tx))); 967 let is_connected_tx = Arc::new(Mutex::new(Some(is_connected_tx))); 968 let is_disconnected_tx = Arc::new(Mutex::new(Some(is_disconnected_tx))); 969 let is_failed_tx = Arc::new(Mutex::new(Some(is_failed_tx))); 970 let is_closed_tx = Arc::new(Mutex::new(Some(is_closed_tx))); 971 972 a_agent.on_connection_state_change(Box::new(move |c: ConnectionState| { 973 let is_checking_tx_clone = Arc::clone(&is_checking_tx); 974 let is_connected_tx_clone = Arc::clone(&is_connected_tx); 975 let is_disconnected_tx_clone = Arc::clone(&is_disconnected_tx); 976 let is_failed_tx_clone = Arc::clone(&is_failed_tx); 977 let is_closed_tx_clone = Arc::clone(&is_closed_tx); 978 Box::pin(async move { 979 match c { 980 ConnectionState::Checking => { 981 log::debug!("drop is_checking_tx"); 982 let mut tx = is_checking_tx_clone.lock().await; 983 tx.take(); 984 } 985 ConnectionState::Connected => { 986 log::debug!("drop is_connected_tx"); 987 let mut tx = is_connected_tx_clone.lock().await; 988 tx.take(); 989 } 990 ConnectionState::Disconnected => { 991 log::debug!("drop is_disconnected_tx"); 992 let mut tx = is_disconnected_tx_clone.lock().await; 993 tx.take(); 994 } 995 ConnectionState::Failed => { 996 log::debug!("drop is_failed_tx"); 997 let mut tx = is_failed_tx_clone.lock().await; 998 tx.take(); 999 } 1000 ConnectionState::Closed => { 1001 log::debug!("drop is_closed_tx"); 1002 let mut tx = is_closed_tx_clone.lock().await; 1003 tx.take(); 1004 } 1005 _ => {} 1006 }; 1007 }) 1008 })); 1009 1010 connect_with_vnet(&a_agent, &b_agent).await?; 1011 1012 log::debug!("wait is_checking_tx"); 1013 let _ = is_checking_rx.recv().await; 1014 log::debug!("wait is_connected_rx"); 1015 let _ = is_connected_rx.recv().await; 1016 log::debug!("wait is_disconnected_rx"); 1017 let _ = is_disconnected_rx.recv().await; 1018 log::debug!("wait is_failed_rx"); 1019 let _ = is_failed_rx.recv().await; 1020 1021 a_agent.close().await?; 1022 b_agent.close().await?; 1023 1024 log::debug!("wait is_closed_rx"); 1025 let _ = is_closed_rx.recv().await; 1026 1027 Ok(()) 1028 } 1029 1030 #[tokio::test] 1031 async fn test_invalid_gather() -> Result<()> { 1032 //"Gather with no OnCandidate should error" 1033 let a = Agent::new(AgentConfig::default()).await?; 1034 1035 if let Err(err) = a.gather_candidates() { 1036 assert_eq!( 1037 Error::ErrNoOnCandidateHandler, 1038 err, 1039 "trickle GatherCandidates succeeded without OnCandidate" 1040 ); 1041 } 1042 1043 a.close().await?; 1044 1045 Ok(()) 1046 } 1047 1048 #[tokio::test] 1049 async fn test_candidate_pair_stats() -> Result<()> { 1050 let a = Agent::new(AgentConfig::default()).await?; 1051 1052 let host_local: Arc<dyn Candidate + Send + Sync> = Arc::new( 1053 CandidateHostConfig { 1054 base_config: CandidateBaseConfig { 1055 network: "udp".to_owned(), 1056 address: "192.168.1.1".to_owned(), 1057 port: 19216, 1058 component: 1, 1059 ..Default::default() 1060 }, 1061 ..Default::default() 1062 } 1063 .new_candidate_host()?, 1064 ); 1065 1066 let relay_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1067 CandidateRelayConfig { 1068 base_config: CandidateBaseConfig { 1069 network: "udp".to_owned(), 1070 address: "1.2.3.4".to_owned(), 1071 port: 2340, 1072 component: 1, 1073 ..Default::default() 1074 }, 1075 rel_addr: "4.3.2.1".to_owned(), 1076 rel_port: 43210, 1077 ..Default::default() 1078 } 1079 .new_candidate_relay()?, 1080 ); 1081 1082 let srflx_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1083 CandidateServerReflexiveConfig { 1084 base_config: CandidateBaseConfig { 1085 network: "udp".to_owned(), 1086 address: "10.10.10.2".to_owned(), 1087 port: 19218, 1088 component: 1, 1089 ..Default::default() 1090 }, 1091 rel_addr: "4.3.2.1".to_owned(), 1092 rel_port: 43212, 1093 } 1094 .new_candidate_server_reflexive()?, 1095 ); 1096 1097 let prflx_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1098 CandidatePeerReflexiveConfig { 1099 base_config: CandidateBaseConfig { 1100 network: "udp".to_owned(), 1101 address: "10.10.10.2".to_owned(), 1102 port: 19217, 1103 component: 1, 1104 ..Default::default() 1105 }, 1106 rel_addr: "4.3.2.1".to_owned(), 1107 rel_port: 43211, 1108 } 1109 .new_candidate_peer_reflexive()?, 1110 ); 1111 1112 let host_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1113 CandidateHostConfig { 1114 base_config: CandidateBaseConfig { 1115 network: "udp".to_owned(), 1116 address: "1.2.3.5".to_owned(), 1117 port: 12350, 1118 component: 1, 1119 ..Default::default() 1120 }, 1121 ..Default::default() 1122 } 1123 .new_candidate_host()?, 1124 ); 1125 1126 for remote in &[ 1127 Arc::clone(&relay_remote), 1128 Arc::clone(&srflx_remote), 1129 Arc::clone(&prflx_remote), 1130 Arc::clone(&host_remote), 1131 ] { 1132 let p = a.internal.find_pair(&host_local, remote).await; 1133 1134 if p.is_none() { 1135 a.internal 1136 .add_pair(Arc::clone(&host_local), Arc::clone(remote)) 1137 .await; 1138 } 1139 } 1140 1141 { 1142 if let Some(p) = a.internal.find_pair(&host_local, &prflx_remote).await { 1143 p.state 1144 .store(CandidatePairState::Failed as u8, Ordering::SeqCst); 1145 } 1146 } 1147 1148 let stats = a.get_candidate_pairs_stats().await; 1149 assert_eq!(stats.len(), 4, "expected 4 candidate pairs stats"); 1150 1151 let (mut relay_pair_stat, mut srflx_pair_stat, mut prflx_pair_stat, mut host_pair_stat) = ( 1152 CandidatePairStats::default(), 1153 CandidatePairStats::default(), 1154 CandidatePairStats::default(), 1155 CandidatePairStats::default(), 1156 ); 1157 1158 for cps in stats { 1159 assert_eq!( 1160 cps.local_candidate_id, 1161 host_local.id(), 1162 "invalid local candidate id" 1163 ); 1164 1165 if cps.remote_candidate_id == relay_remote.id() { 1166 relay_pair_stat = cps; 1167 } else if cps.remote_candidate_id == srflx_remote.id() { 1168 srflx_pair_stat = cps; 1169 } else if cps.remote_candidate_id == prflx_remote.id() { 1170 prflx_pair_stat = cps; 1171 } else if cps.remote_candidate_id == host_remote.id() { 1172 host_pair_stat = cps; 1173 } else { 1174 panic!("invalid remote candidate ID"); 1175 } 1176 } 1177 1178 assert_eq!( 1179 relay_pair_stat.remote_candidate_id, 1180 relay_remote.id(), 1181 "missing host-relay pair stat" 1182 ); 1183 assert_eq!( 1184 srflx_pair_stat.remote_candidate_id, 1185 srflx_remote.id(), 1186 "missing host-srflx pair stat" 1187 ); 1188 assert_eq!( 1189 prflx_pair_stat.remote_candidate_id, 1190 prflx_remote.id(), 1191 "missing host-prflx pair stat" 1192 ); 1193 assert_eq!( 1194 host_pair_stat.remote_candidate_id, 1195 host_remote.id(), 1196 "missing host-host pair stat" 1197 ); 1198 assert_eq!( 1199 prflx_pair_stat.state, 1200 CandidatePairState::Failed, 1201 "expected host-prfflx pair to have state failed, it has state {} instead", 1202 prflx_pair_stat.state 1203 ); 1204 1205 a.close().await?; 1206 1207 Ok(()) 1208 } 1209 1210 #[tokio::test] 1211 async fn test_local_candidate_stats() -> Result<()> { 1212 let a = Agent::new(AgentConfig::default()).await?; 1213 1214 let host_local: Arc<dyn Candidate + Send + Sync> = Arc::new( 1215 CandidateHostConfig { 1216 base_config: CandidateBaseConfig { 1217 network: "udp".to_owned(), 1218 address: "192.168.1.1".to_owned(), 1219 port: 19216, 1220 component: 1, 1221 ..Default::default() 1222 }, 1223 ..Default::default() 1224 } 1225 .new_candidate_host()?, 1226 ); 1227 1228 let srflx_local: Arc<dyn Candidate + Send + Sync> = Arc::new( 1229 CandidateServerReflexiveConfig { 1230 base_config: CandidateBaseConfig { 1231 network: "udp".to_owned(), 1232 address: "192.168.1.1".to_owned(), 1233 port: 19217, 1234 component: 1, 1235 ..Default::default() 1236 }, 1237 rel_addr: "4.3.2.1".to_owned(), 1238 rel_port: 43212, 1239 } 1240 .new_candidate_server_reflexive()?, 1241 ); 1242 1243 { 1244 let mut local_candidates = a.internal.local_candidates.lock().await; 1245 local_candidates.insert( 1246 NetworkType::Udp4, 1247 vec![Arc::clone(&host_local), Arc::clone(&srflx_local)], 1248 ); 1249 } 1250 1251 let local_stats = a.get_local_candidates_stats().await; 1252 assert_eq!( 1253 local_stats.len(), 1254 2, 1255 "expected 2 local candidates stats, got {} instead", 1256 local_stats.len() 1257 ); 1258 1259 let (mut host_local_stat, mut srflx_local_stat) = 1260 (CandidateStats::default(), CandidateStats::default()); 1261 for stats in local_stats { 1262 let candidate = if stats.id == host_local.id() { 1263 host_local_stat = stats.clone(); 1264 Arc::clone(&host_local) 1265 } else if stats.id == srflx_local.id() { 1266 srflx_local_stat = stats.clone(); 1267 Arc::clone(&srflx_local) 1268 } else { 1269 panic!("invalid local candidate ID"); 1270 }; 1271 1272 assert_eq!( 1273 stats.candidate_type, 1274 candidate.candidate_type(), 1275 "invalid stats CandidateType" 1276 ); 1277 assert_eq!( 1278 stats.priority, 1279 candidate.priority(), 1280 "invalid stats CandidateType" 1281 ); 1282 assert_eq!(stats.ip, candidate.address(), "invalid stats IP"); 1283 } 1284 1285 assert_eq!( 1286 host_local_stat.id, 1287 host_local.id(), 1288 "missing host local stat" 1289 ); 1290 assert_eq!( 1291 srflx_local_stat.id, 1292 srflx_local.id(), 1293 "missing srflx local stat" 1294 ); 1295 1296 a.close().await?; 1297 1298 Ok(()) 1299 } 1300 1301 #[tokio::test] 1302 async fn test_remote_candidate_stats() -> Result<()> { 1303 let a = Agent::new(AgentConfig::default()).await?; 1304 1305 let relay_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1306 CandidateRelayConfig { 1307 base_config: CandidateBaseConfig { 1308 network: "udp".to_owned(), 1309 address: "1.2.3.4".to_owned(), 1310 port: 12340, 1311 component: 1, 1312 ..Default::default() 1313 }, 1314 rel_addr: "4.3.2.1".to_owned(), 1315 rel_port: 43210, 1316 ..Default::default() 1317 } 1318 .new_candidate_relay()?, 1319 ); 1320 1321 let srflx_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1322 CandidateServerReflexiveConfig { 1323 base_config: CandidateBaseConfig { 1324 network: "udp".to_owned(), 1325 address: "10.10.10.2".to_owned(), 1326 port: 19218, 1327 component: 1, 1328 ..Default::default() 1329 }, 1330 rel_addr: "4.3.2.1".to_owned(), 1331 rel_port: 43212, 1332 } 1333 .new_candidate_server_reflexive()?, 1334 ); 1335 1336 let prflx_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1337 CandidatePeerReflexiveConfig { 1338 base_config: CandidateBaseConfig { 1339 network: "udp".to_owned(), 1340 address: "10.10.10.2".to_owned(), 1341 port: 19217, 1342 component: 1, 1343 ..Default::default() 1344 }, 1345 rel_addr: "4.3.2.1".to_owned(), 1346 rel_port: 43211, 1347 } 1348 .new_candidate_peer_reflexive()?, 1349 ); 1350 1351 let host_remote: Arc<dyn Candidate + Send + Sync> = Arc::new( 1352 CandidateHostConfig { 1353 base_config: CandidateBaseConfig { 1354 network: "udp".to_owned(), 1355 address: "1.2.3.5".to_owned(), 1356 port: 12350, 1357 component: 1, 1358 ..Default::default() 1359 }, 1360 ..Default::default() 1361 } 1362 .new_candidate_host()?, 1363 ); 1364 1365 { 1366 let mut remote_candidates = a.internal.remote_candidates.lock().await; 1367 remote_candidates.insert( 1368 NetworkType::Udp4, 1369 vec![ 1370 Arc::clone(&relay_remote), 1371 Arc::clone(&srflx_remote), 1372 Arc::clone(&prflx_remote), 1373 Arc::clone(&host_remote), 1374 ], 1375 ); 1376 } 1377 1378 let remote_stats = a.get_remote_candidates_stats().await; 1379 assert_eq!( 1380 remote_stats.len(), 1381 4, 1382 "expected 4 remote candidates stats, got {} instead", 1383 remote_stats.len() 1384 ); 1385 1386 let (mut relay_remote_stat, mut srflx_remote_stat, mut prflx_remote_stat, mut host_remote_stat) = ( 1387 CandidateStats::default(), 1388 CandidateStats::default(), 1389 CandidateStats::default(), 1390 CandidateStats::default(), 1391 ); 1392 for stats in remote_stats { 1393 let candidate = if stats.id == relay_remote.id() { 1394 relay_remote_stat = stats.clone(); 1395 Arc::clone(&relay_remote) 1396 } else if stats.id == srflx_remote.id() { 1397 srflx_remote_stat = stats.clone(); 1398 Arc::clone(&srflx_remote) 1399 } else if stats.id == prflx_remote.id() { 1400 prflx_remote_stat = stats.clone(); 1401 Arc::clone(&prflx_remote) 1402 } else if stats.id == host_remote.id() { 1403 host_remote_stat = stats.clone(); 1404 Arc::clone(&host_remote) 1405 } else { 1406 panic!("invalid remote candidate ID"); 1407 }; 1408 1409 assert_eq!( 1410 stats.candidate_type, 1411 candidate.candidate_type(), 1412 "invalid stats CandidateType" 1413 ); 1414 assert_eq!( 1415 stats.priority, 1416 candidate.priority(), 1417 "invalid stats CandidateType" 1418 ); 1419 assert_eq!(stats.ip, candidate.address(), "invalid stats IP"); 1420 } 1421 1422 assert_eq!( 1423 relay_remote_stat.id, 1424 relay_remote.id(), 1425 "missing relay remote stat" 1426 ); 1427 assert_eq!( 1428 srflx_remote_stat.id, 1429 srflx_remote.id(), 1430 "missing srflx remote stat" 1431 ); 1432 assert_eq!( 1433 prflx_remote_stat.id, 1434 prflx_remote.id(), 1435 "missing prflx remote stat" 1436 ); 1437 assert_eq!( 1438 host_remote_stat.id, 1439 host_remote.id(), 1440 "missing host remote stat" 1441 ); 1442 1443 a.close().await?; 1444 1445 Ok(()) 1446 } 1447 1448 #[tokio::test] 1449 async fn test_init_ext_ip_mapping() -> Result<()> { 1450 // a.extIPMapper should be nil by default 1451 let a = Agent::new(AgentConfig::default()).await?; 1452 assert!( 1453 a.ext_ip_mapper.is_none(), 1454 "a.extIPMapper should be none by default" 1455 ); 1456 a.close().await?; 1457 1458 // a.extIPMapper should be nil when NAT1To1IPs is a non-nil empty array 1459 let a = Agent::new(AgentConfig { 1460 nat_1to1_ips: vec![], 1461 nat_1to1_ip_candidate_type: CandidateType::Host, 1462 ..Default::default() 1463 }) 1464 .await?; 1465 assert!( 1466 a.ext_ip_mapper.is_none(), 1467 "a.extIPMapper should be none by default" 1468 ); 1469 a.close().await?; 1470 1471 // NewAgent should return an error when 1:1 NAT for host candidate is enabled 1472 // but the candidate type does not appear in the CandidateTypes. 1473 if let Err(err) = Agent::new(AgentConfig { 1474 nat_1to1_ips: vec!["1.2.3.4".to_owned()], 1475 nat_1to1_ip_candidate_type: CandidateType::Host, 1476 candidate_types: vec![CandidateType::Relay], 1477 ..Default::default() 1478 }) 1479 .await 1480 { 1481 assert_eq!( 1482 Error::ErrIneffectiveNat1to1IpMappingHost, 1483 err, 1484 "Unexpected error: {}", 1485 err 1486 ); 1487 } else { 1488 panic!("expected error, but got ok"); 1489 } 1490 1491 // NewAgent should return an error when 1:1 NAT for srflx candidate is enabled 1492 // but the candidate type does not appear in the CandidateTypes. 1493 if let Err(err) = Agent::new(AgentConfig { 1494 nat_1to1_ips: vec!["1.2.3.4".to_owned()], 1495 nat_1to1_ip_candidate_type: CandidateType::ServerReflexive, 1496 candidate_types: vec![CandidateType::Relay], 1497 ..Default::default() 1498 }) 1499 .await 1500 { 1501 assert_eq!( 1502 Error::ErrIneffectiveNat1to1IpMappingSrflx, 1503 err, 1504 "Unexpected error: {}", 1505 err 1506 ); 1507 } else { 1508 panic!("expected error, but got ok"); 1509 } 1510 1511 // NewAgent should return an error when 1:1 NAT for host candidate is enabled 1512 // along with mDNS with MulticastDNSModeQueryAndGather 1513 if let Err(err) = Agent::new(AgentConfig { 1514 nat_1to1_ips: vec!["1.2.3.4".to_owned()], 1515 nat_1to1_ip_candidate_type: CandidateType::Host, 1516 multicast_dns_mode: MulticastDnsMode::QueryAndGather, 1517 ..Default::default() 1518 }) 1519 .await 1520 { 1521 assert_eq!( 1522 Error::ErrMulticastDnsWithNat1to1IpMapping, 1523 err, 1524 "Unexpected error: {}", 1525 err 1526 ); 1527 } else { 1528 panic!("expected error, but got ok"); 1529 } 1530 1531 // NewAgent should return if newExternalIPMapper() returns an error. 1532 if let Err(err) = Agent::new(AgentConfig { 1533 nat_1to1_ips: vec!["bad.2.3.4".to_owned()], // bad IP 1534 nat_1to1_ip_candidate_type: CandidateType::Host, 1535 ..Default::default() 1536 }) 1537 .await 1538 { 1539 assert_eq!( 1540 Error::ErrInvalidNat1to1IpMapping, 1541 err, 1542 "Unexpected error: {}", 1543 err 1544 ); 1545 } else { 1546 panic!("expected error, but got ok"); 1547 } 1548 1549 Ok(()) 1550 } 1551 1552 #[tokio::test] 1553 async fn test_binding_request_timeout() -> Result<()> { 1554 const EXPECTED_REMOVAL_COUNT: usize = 2; 1555 1556 let a = Agent::new(AgentConfig::default()).await?; 1557 1558 let now = Instant::now(); 1559 { 1560 { 1561 let mut pending_binding_requests = a.internal.pending_binding_requests.lock().await; 1562 pending_binding_requests.push(BindingRequest { 1563 timestamp: now, // valid 1564 ..Default::default() 1565 }); 1566 pending_binding_requests.push(BindingRequest { 1567 timestamp: now.sub(Duration::from_millis(3900)), // valid 1568 ..Default::default() 1569 }); 1570 pending_binding_requests.push(BindingRequest { 1571 timestamp: now.sub(Duration::from_millis(4100)), // invalid 1572 ..Default::default() 1573 }); 1574 pending_binding_requests.push(BindingRequest { 1575 timestamp: now.sub(Duration::from_secs(75)), // invalid 1576 ..Default::default() 1577 }); 1578 } 1579 1580 a.internal.invalidate_pending_binding_requests(now).await; 1581 { 1582 let pending_binding_requests = a.internal.pending_binding_requests.lock().await; 1583 assert_eq!(pending_binding_requests.len(), EXPECTED_REMOVAL_COUNT, "Binding invalidation due to timeout did not remove the correct number of binding requests") 1584 } 1585 } 1586 1587 a.close().await?; 1588 1589 Ok(()) 1590 } 1591 1592 // test_agent_credentials checks if local username fragments and passwords (if set) meet RFC standard 1593 // and ensure it's backwards compatible with previous versions of the pion/ice 1594 #[tokio::test] 1595 async fn test_agent_credentials() -> Result<()> { 1596 // Agent should not require any of the usernames and password to be set 1597 // If set, they should follow the default 16/128 bits random number generator strategy 1598 1599 let a = Agent::new(AgentConfig::default()).await?; 1600 { 1601 let ufrag_pwd = a.internal.ufrag_pwd.lock().await; 1602 assert!(ufrag_pwd.local_ufrag.as_bytes().len() * 8 >= 24); 1603 assert!(ufrag_pwd.local_pwd.as_bytes().len() * 8 >= 128); 1604 } 1605 a.close().await?; 1606 1607 // Should honor RFC standards 1608 // Local values MUST be unguessable, with at least 128 bits of 1609 // random number generator output used to generate the password, and 1610 // at least 24 bits of output to generate the username fragment. 1611 1612 if let Err(err) = Agent::new(AgentConfig { 1613 local_ufrag: "xx".to_owned(), 1614 ..Default::default() 1615 }) 1616 .await 1617 { 1618 assert_eq!(Error::ErrLocalUfragInsufficientBits, err); 1619 } else { 1620 panic!("expected error, but got ok"); 1621 } 1622 1623 if let Err(err) = Agent::new(AgentConfig { 1624 local_pwd: "xxxxxx".to_owned(), 1625 ..Default::default() 1626 }) 1627 .await 1628 { 1629 assert_eq!(Error::ErrLocalPwdInsufficientBits, err); 1630 } else { 1631 panic!("expected error, but got ok"); 1632 } 1633 1634 Ok(()) 1635 } 1636 1637 // Assert that Agent on Failure deletes all existing candidates 1638 // User can then do an ICE Restart to bring agent back 1639 #[tokio::test] 1640 async fn test_connection_state_failed_delete_all_candidates() -> Result<()> { 1641 let one_second = Duration::from_secs(1); 1642 let keepalive_interval = Duration::from_secs(0); 1643 1644 let cfg0 = AgentConfig { 1645 network_types: supported_network_types(), 1646 disconnected_timeout: Some(one_second), 1647 failed_timeout: Some(one_second), 1648 keepalive_interval: Some(keepalive_interval), 1649 ..Default::default() 1650 }; 1651 let cfg1 = AgentConfig { 1652 network_types: supported_network_types(), 1653 disconnected_timeout: Some(one_second), 1654 failed_timeout: Some(one_second), 1655 keepalive_interval: Some(keepalive_interval), 1656 ..Default::default() 1657 }; 1658 1659 let a_agent = Arc::new(Agent::new(cfg0).await?); 1660 let b_agent = Arc::new(Agent::new(cfg1).await?); 1661 1662 let (is_failed_tx, mut is_failed_rx) = mpsc::channel::<()>(1); 1663 let is_failed_tx = Arc::new(Mutex::new(Some(is_failed_tx))); 1664 a_agent.on_connection_state_change(Box::new(move |c: ConnectionState| { 1665 let is_failed_tx_clone = Arc::clone(&is_failed_tx); 1666 Box::pin(async move { 1667 if c == ConnectionState::Failed { 1668 let mut tx = is_failed_tx_clone.lock().await; 1669 tx.take(); 1670 } 1671 }) 1672 })); 1673 1674 connect_with_vnet(&a_agent, &b_agent).await?; 1675 let _ = is_failed_rx.recv().await; 1676 1677 { 1678 { 1679 let remote_candidates = a_agent.internal.remote_candidates.lock().await; 1680 assert_eq!(remote_candidates.len(), 0); 1681 } 1682 { 1683 let local_candidates = a_agent.internal.local_candidates.lock().await; 1684 assert_eq!(local_candidates.len(), 0); 1685 } 1686 } 1687 1688 a_agent.close().await?; 1689 b_agent.close().await?; 1690 1691 Ok(()) 1692 } 1693 1694 // Assert that the ICE Agent can go directly from Connecting -> Failed on both sides 1695 #[tokio::test] 1696 async fn test_connection_state_connecting_to_failed() -> Result<()> { 1697 let one_second = Duration::from_secs(1); 1698 let keepalive_interval = Duration::from_secs(0); 1699 1700 let cfg0 = AgentConfig { 1701 disconnected_timeout: Some(one_second), 1702 failed_timeout: Some(one_second), 1703 keepalive_interval: Some(keepalive_interval), 1704 ..Default::default() 1705 }; 1706 let cfg1 = AgentConfig { 1707 disconnected_timeout: Some(one_second), 1708 failed_timeout: Some(one_second), 1709 keepalive_interval: Some(keepalive_interval), 1710 ..Default::default() 1711 }; 1712 1713 let a_agent = Arc::new(Agent::new(cfg0).await?); 1714 let b_agent = Arc::new(Agent::new(cfg1).await?); 1715 1716 let is_failed = WaitGroup::new(); 1717 let is_checking = WaitGroup::new(); 1718 1719 let connection_state_check = move |wf: Worker, wc: Worker| { 1720 let wf = Arc::new(Mutex::new(Some(wf))); 1721 let wc = Arc::new(Mutex::new(Some(wc))); 1722 let hdlr_fn: OnConnectionStateChangeHdlrFn = Box::new(move |c: ConnectionState| { 1723 let wf_clone = Arc::clone(&wf); 1724 let wc_clone = Arc::clone(&wc); 1725 Box::pin(async move { 1726 if c == ConnectionState::Failed { 1727 let mut f = wf_clone.lock().await; 1728 f.take(); 1729 } else if c == ConnectionState::Checking { 1730 let mut c = wc_clone.lock().await; 1731 c.take(); 1732 } else if c == ConnectionState::Connected || c == ConnectionState::Completed { 1733 panic!("Unexpected ConnectionState: {}", c); 1734 } 1735 }) 1736 }); 1737 hdlr_fn 1738 }; 1739 1740 let (wf1, wc1) = (is_failed.worker(), is_checking.worker()); 1741 a_agent.on_connection_state_change(connection_state_check(wf1, wc1)); 1742 1743 let (wf2, wc2) = (is_failed.worker(), is_checking.worker()); 1744 b_agent.on_connection_state_change(connection_state_check(wf2, wc2)); 1745 1746 let agent_a = Arc::clone(&a_agent); 1747 tokio::spawn(async move { 1748 let (_cancel_tx, cancel_rx) = mpsc::channel(1); 1749 let result = agent_a 1750 .accept(cancel_rx, "InvalidFrag".to_owned(), "InvalidPwd".to_owned()) 1751 .await; 1752 assert!(result.is_err()); 1753 }); 1754 1755 let agent_b = Arc::clone(&b_agent); 1756 tokio::spawn(async move { 1757 let (_cancel_tx, cancel_rx) = mpsc::channel(1); 1758 let result = agent_b 1759 .dial(cancel_rx, "InvalidFrag".to_owned(), "InvalidPwd".to_owned()) 1760 .await; 1761 assert!(result.is_err()); 1762 }); 1763 1764 is_checking.wait().await; 1765 is_failed.wait().await; 1766 1767 a_agent.close().await?; 1768 b_agent.close().await?; 1769 1770 Ok(()) 1771 } 1772 1773 #[tokio::test] 1774 async fn test_agent_restart_during_gather() -> Result<()> { 1775 //"Restart During Gather" 1776 1777 let agent = Agent::new(AgentConfig::default()).await?; 1778 1779 agent 1780 .gathering_state 1781 .store(GatheringState::Gathering as u8, Ordering::SeqCst); 1782 1783 if let Err(err) = agent.restart("".to_owned(), "".to_owned()).await { 1784 assert_eq!(Error::ErrRestartWhenGathering, err); 1785 } else { 1786 panic!("expected error, but got ok"); 1787 } 1788 1789 agent.close().await?; 1790 1791 Ok(()) 1792 } 1793 1794 #[tokio::test] 1795 async fn test_agent_restart_when_closed() -> Result<()> { 1796 //"Restart When Closed" 1797 1798 let agent = Agent::new(AgentConfig::default()).await?; 1799 agent.close().await?; 1800 1801 if let Err(err) = agent.restart("".to_owned(), "".to_owned()).await { 1802 assert_eq!(Error::ErrClosed, err); 1803 } else { 1804 panic!("expected error, but got ok"); 1805 } 1806 1807 Ok(()) 1808 } 1809 1810 #[tokio::test] 1811 async fn test_agent_restart_one_side() -> Result<()> { 1812 let one_second = Duration::from_secs(1); 1813 1814 //"Restart One Side" 1815 let (_, _, agent_a, agent_b) = pipe( 1816 Some(AgentConfig { 1817 disconnected_timeout: Some(one_second), 1818 failed_timeout: Some(one_second), 1819 ..Default::default() 1820 }), 1821 Some(AgentConfig { 1822 disconnected_timeout: Some(one_second), 1823 failed_timeout: Some(one_second), 1824 ..Default::default() 1825 }), 1826 ) 1827 .await?; 1828 1829 let (cancel_tx, mut cancel_rx) = mpsc::channel::<()>(1); 1830 let cancel_tx = Arc::new(Mutex::new(Some(cancel_tx))); 1831 agent_b.on_connection_state_change(Box::new(move |c: ConnectionState| { 1832 let cancel_tx_clone = Arc::clone(&cancel_tx); 1833 Box::pin(async move { 1834 if c == ConnectionState::Failed || c == ConnectionState::Disconnected { 1835 let mut tx = cancel_tx_clone.lock().await; 1836 tx.take(); 1837 } 1838 }) 1839 })); 1840 1841 agent_a.restart("".to_owned(), "".to_owned()).await?; 1842 1843 let _ = cancel_rx.recv().await; 1844 1845 agent_a.close().await?; 1846 agent_b.close().await?; 1847 1848 Ok(()) 1849 } 1850 1851 #[tokio::test] 1852 async fn test_agent_restart_both_side() -> Result<()> { 1853 let one_second = Duration::from_secs(1); 1854 //"Restart Both Sides" 1855 1856 // Get all addresses of candidates concatenated 1857 let generate_candidate_address_strings = 1858 |res: Result<Vec<Arc<dyn Candidate + Send + Sync>>>| -> String { 1859 assert!(res.is_ok()); 1860 1861 let mut out = String::new(); 1862 if let Ok(candidates) = res { 1863 for c in candidates { 1864 out += c.address().as_str(); 1865 out += ":"; 1866 out += c.port().to_string().as_str(); 1867 } 1868 } 1869 out 1870 }; 1871 1872 // Store the original candidates, confirm that after we reconnect we have new pairs 1873 let (_, _, agent_a, agent_b) = pipe( 1874 Some(AgentConfig { 1875 disconnected_timeout: Some(one_second), 1876 failed_timeout: Some(one_second), 1877 ..Default::default() 1878 }), 1879 Some(AgentConfig { 1880 disconnected_timeout: Some(one_second), 1881 failed_timeout: Some(one_second), 1882 ..Default::default() 1883 }), 1884 ) 1885 .await?; 1886 1887 let conn_afirst_candidates = 1888 generate_candidate_address_strings(agent_a.get_local_candidates().await); 1889 let conn_bfirst_candidates = 1890 generate_candidate_address_strings(agent_b.get_local_candidates().await); 1891 1892 let (a_notifier, mut a_connected) = on_connected(); 1893 agent_a.on_connection_state_change(a_notifier); 1894 1895 let (b_notifier, mut b_connected) = on_connected(); 1896 agent_b.on_connection_state_change(b_notifier); 1897 1898 // Restart and Re-Signal 1899 agent_a.restart("".to_owned(), "".to_owned()).await?; 1900 agent_b.restart("".to_owned(), "".to_owned()).await?; 1901 1902 // Exchange Candidates and Credentials 1903 let (ufrag, pwd) = agent_b.get_local_user_credentials().await; 1904 agent_a.set_remote_credentials(ufrag, pwd).await?; 1905 1906 let (ufrag, pwd) = agent_a.get_local_user_credentials().await; 1907 agent_b.set_remote_credentials(ufrag, pwd).await?; 1908 1909 gather_and_exchange_candidates(&agent_a, &agent_b).await?; 1910 1911 // Wait until both have gone back to connected 1912 let _ = a_connected.recv().await; 1913 let _ = b_connected.recv().await; 1914 1915 // Assert that we have new candiates each time 1916 assert_ne!( 1917 conn_afirst_candidates, 1918 generate_candidate_address_strings(agent_a.get_local_candidates().await) 1919 ); 1920 assert_ne!( 1921 conn_bfirst_candidates, 1922 generate_candidate_address_strings(agent_b.get_local_candidates().await) 1923 ); 1924 1925 agent_a.close().await?; 1926 agent_b.close().await?; 1927 1928 Ok(()) 1929 } 1930 1931 #[tokio::test] 1932 async fn test_get_remote_credentials() -> Result<()> { 1933 let a = Agent::new(AgentConfig::default()).await?; 1934 1935 let (remote_ufrag, remote_pwd) = { 1936 let mut ufrag_pwd = a.internal.ufrag_pwd.lock().await; 1937 ufrag_pwd.remote_ufrag = "remoteUfrag".to_owned(); 1938 ufrag_pwd.remote_pwd = "remotePwd".to_owned(); 1939 ( 1940 ufrag_pwd.remote_ufrag.to_owned(), 1941 ufrag_pwd.remote_pwd.to_owned(), 1942 ) 1943 }; 1944 1945 let (actual_ufrag, actual_pwd) = a.get_remote_user_credentials().await; 1946 1947 assert_eq!(actual_ufrag, remote_ufrag); 1948 assert_eq!(actual_pwd, remote_pwd); 1949 1950 a.close().await?; 1951 1952 Ok(()) 1953 } 1954 1955 #[tokio::test] 1956 async fn test_close_in_connection_state_callback() -> Result<()> { 1957 let disconnected_duration = Duration::from_secs(1); 1958 let failed_duration = Duration::from_secs(1); 1959 let keepalive_interval = Duration::from_secs(0); 1960 1961 let cfg0 = AgentConfig { 1962 urls: vec![], 1963 network_types: supported_network_types(), 1964 disconnected_timeout: Some(disconnected_duration), 1965 failed_timeout: Some(failed_duration), 1966 keepalive_interval: Some(keepalive_interval), 1967 check_interval: Duration::from_millis(500), 1968 ..Default::default() 1969 }; 1970 1971 let cfg1 = AgentConfig { 1972 urls: vec![], 1973 network_types: supported_network_types(), 1974 disconnected_timeout: Some(disconnected_duration), 1975 failed_timeout: Some(failed_duration), 1976 keepalive_interval: Some(keepalive_interval), 1977 check_interval: Duration::from_millis(500), 1978 ..Default::default() 1979 }; 1980 1981 let a_agent = Arc::new(Agent::new(cfg0).await?); 1982 let b_agent = Arc::new(Agent::new(cfg1).await?); 1983 1984 let (is_closed_tx, mut is_closed_rx) = mpsc::channel::<()>(1); 1985 let (is_connected_tx, mut is_connected_rx) = mpsc::channel::<()>(1); 1986 let is_closed_tx = Arc::new(Mutex::new(Some(is_closed_tx))); 1987 let is_connected_tx = Arc::new(Mutex::new(Some(is_connected_tx))); 1988 a_agent.on_connection_state_change(Box::new(move |c: ConnectionState| { 1989 let is_closed_tx_clone = Arc::clone(&is_closed_tx); 1990 let is_connected_tx_clone = Arc::clone(&is_connected_tx); 1991 Box::pin(async move { 1992 if c == ConnectionState::Connected { 1993 let mut tx = is_connected_tx_clone.lock().await; 1994 tx.take(); 1995 } else if c == ConnectionState::Closed { 1996 let mut tx = is_closed_tx_clone.lock().await; 1997 tx.take(); 1998 } 1999 }) 2000 })); 2001 2002 connect_with_vnet(&a_agent, &b_agent).await?; 2003 2004 let _ = is_connected_rx.recv().await; 2005 a_agent.close().await?; 2006 2007 let _ = is_closed_rx.recv().await; 2008 b_agent.close().await?; 2009 2010 Ok(()) 2011 } 2012 2013 #[tokio::test] 2014 async fn test_run_task_in_connection_state_callback() -> Result<()> { 2015 let one_second = Duration::from_secs(1); 2016 let keepalive_interval = Duration::from_secs(0); 2017 2018 let cfg0 = AgentConfig { 2019 urls: vec![], 2020 network_types: supported_network_types(), 2021 disconnected_timeout: Some(one_second), 2022 failed_timeout: Some(one_second), 2023 keepalive_interval: Some(keepalive_interval), 2024 check_interval: Duration::from_millis(50), 2025 ..Default::default() 2026 }; 2027 2028 let cfg1 = AgentConfig { 2029 urls: vec![], 2030 network_types: supported_network_types(), 2031 disconnected_timeout: Some(one_second), 2032 failed_timeout: Some(one_second), 2033 keepalive_interval: Some(keepalive_interval), 2034 check_interval: Duration::from_millis(50), 2035 ..Default::default() 2036 }; 2037 2038 let a_agent = Arc::new(Agent::new(cfg0).await?); 2039 let b_agent = Arc::new(Agent::new(cfg1).await?); 2040 2041 let (is_complete_tx, mut is_complete_rx) = mpsc::channel::<()>(1); 2042 let is_complete_tx = Arc::new(Mutex::new(Some(is_complete_tx))); 2043 a_agent.on_connection_state_change(Box::new(move |c: ConnectionState| { 2044 let is_complete_tx_clone = Arc::clone(&is_complete_tx); 2045 Box::pin(async move { 2046 if c == ConnectionState::Connected { 2047 let mut tx = is_complete_tx_clone.lock().await; 2048 tx.take(); 2049 } 2050 }) 2051 })); 2052 2053 connect_with_vnet(&a_agent, &b_agent).await?; 2054 2055 let _ = is_complete_rx.recv().await; 2056 let _ = a_agent.get_local_user_credentials().await; 2057 a_agent.restart("".to_owned(), "".to_owned()).await?; 2058 2059 a_agent.close().await?; 2060 b_agent.close().await?; 2061 2062 Ok(()) 2063 } 2064 2065 #[tokio::test] 2066 async fn test_run_task_in_selected_candidate_pair_change_callback() -> Result<()> { 2067 let one_second = Duration::from_secs(1); 2068 let keepalive_interval = Duration::from_secs(0); 2069 2070 let cfg0 = AgentConfig { 2071 urls: vec![], 2072 network_types: supported_network_types(), 2073 disconnected_timeout: Some(one_second), 2074 failed_timeout: Some(one_second), 2075 keepalive_interval: Some(keepalive_interval), 2076 check_interval: Duration::from_millis(50), 2077 ..Default::default() 2078 }; 2079 2080 let cfg1 = AgentConfig { 2081 urls: vec![], 2082 network_types: supported_network_types(), 2083 disconnected_timeout: Some(one_second), 2084 failed_timeout: Some(one_second), 2085 keepalive_interval: Some(keepalive_interval), 2086 check_interval: Duration::from_millis(50), 2087 ..Default::default() 2088 }; 2089 2090 let a_agent = Arc::new(Agent::new(cfg0).await?); 2091 let b_agent = Arc::new(Agent::new(cfg1).await?); 2092 2093 let (is_tested_tx, mut is_tested_rx) = mpsc::channel::<()>(1); 2094 let is_tested_tx = Arc::new(Mutex::new(Some(is_tested_tx))); 2095 a_agent.on_selected_candidate_pair_change(Box::new( 2096 move |_: &Arc<dyn Candidate + Send + Sync>, _: &Arc<dyn Candidate + Send + Sync>| { 2097 let is_tested_tx_clone = Arc::clone(&is_tested_tx); 2098 Box::pin(async move { 2099 let mut tx = is_tested_tx_clone.lock().await; 2100 tx.take(); 2101 }) 2102 }, 2103 )); 2104 2105 let (is_complete_tx, mut is_complete_rx) = mpsc::channel::<()>(1); 2106 let is_complete_tx = Arc::new(Mutex::new(Some(is_complete_tx))); 2107 a_agent.on_connection_state_change(Box::new(move |c: ConnectionState| { 2108 let is_complete_tx_clone = Arc::clone(&is_complete_tx); 2109 Box::pin(async move { 2110 if c == ConnectionState::Connected { 2111 let mut tx = is_complete_tx_clone.lock().await; 2112 tx.take(); 2113 } 2114 }) 2115 })); 2116 2117 connect_with_vnet(&a_agent, &b_agent).await?; 2118 2119 let _ = is_complete_rx.recv().await; 2120 let _ = is_tested_rx.recv().await; 2121 2122 let _ = a_agent.get_local_user_credentials().await; 2123 2124 a_agent.close().await?; 2125 b_agent.close().await?; 2126 2127 Ok(()) 2128 } 2129 2130 // Assert that a Lite agent goes to disconnected and failed 2131 #[tokio::test] 2132 async fn test_lite_lifecycle() -> Result<()> { 2133 let (a_notifier, mut a_connected_rx) = on_connected(); 2134 2135 let a_agent = Arc::new( 2136 Agent::new(AgentConfig { 2137 network_types: supported_network_types(), 2138 multicast_dns_mode: MulticastDnsMode::Disabled, 2139 ..Default::default() 2140 }) 2141 .await?, 2142 ); 2143 2144 a_agent.on_connection_state_change(a_notifier); 2145 2146 let disconnected_duration = Duration::from_secs(1); 2147 let failed_duration = Duration::from_secs(1); 2148 let keepalive_interval = Duration::from_secs(0); 2149 2150 let b_agent = Arc::new( 2151 Agent::new(AgentConfig { 2152 lite: true, 2153 candidate_types: vec![CandidateType::Host], 2154 network_types: supported_network_types(), 2155 multicast_dns_mode: MulticastDnsMode::Disabled, 2156 disconnected_timeout: Some(disconnected_duration), 2157 failed_timeout: Some(failed_duration), 2158 keepalive_interval: Some(keepalive_interval), 2159 check_interval: Duration::from_millis(500), 2160 ..Default::default() 2161 }) 2162 .await?, 2163 ); 2164 2165 let (b_connected_tx, mut b_connected_rx) = mpsc::channel::<()>(1); 2166 let (b_disconnected_tx, mut b_disconnected_rx) = mpsc::channel::<()>(1); 2167 let (b_failed_tx, mut b_failed_rx) = mpsc::channel::<()>(1); 2168 let b_connected_tx = Arc::new(Mutex::new(Some(b_connected_tx))); 2169 let b_disconnected_tx = Arc::new(Mutex::new(Some(b_disconnected_tx))); 2170 let b_failed_tx = Arc::new(Mutex::new(Some(b_failed_tx))); 2171 2172 b_agent.on_connection_state_change(Box::new(move |c: ConnectionState| { 2173 let b_connected_tx_clone = Arc::clone(&b_connected_tx); 2174 let b_disconnected_tx_clone = Arc::clone(&b_disconnected_tx); 2175 let b_failed_tx_clone = Arc::clone(&b_failed_tx); 2176 2177 Box::pin(async move { 2178 if c == ConnectionState::Connected { 2179 let mut tx = b_connected_tx_clone.lock().await; 2180 tx.take(); 2181 } else if c == ConnectionState::Disconnected { 2182 let mut tx = b_disconnected_tx_clone.lock().await; 2183 tx.take(); 2184 } else if c == ConnectionState::Failed { 2185 let mut tx = b_failed_tx_clone.lock().await; 2186 tx.take(); 2187 } 2188 }) 2189 })); 2190 2191 connect_with_vnet(&b_agent, &a_agent).await?; 2192 2193 let _ = a_connected_rx.recv().await; 2194 let _ = b_connected_rx.recv().await; 2195 a_agent.close().await?; 2196 2197 let _ = b_disconnected_rx.recv().await; 2198 let _ = b_failed_rx.recv().await; 2199 2200 b_agent.close().await?; 2201 2202 Ok(()) 2203 } 2204