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