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