tests.rs (33639B)
1 use std::{ 2 collections::BTreeMap, 3 net::SocketAddr, 4 sync::{Arc, Mutex as StdMutex}, 5 }; 6 7 use crate::{ 8 app::{ 9 GossipEnvelope, NETWORK_ID, NodeCore, PROTOCOL_VERSION, PeerBook, PeerDirection, 10 ProtocolHello, 11 }, 12 domain::{Ledger, Wallet}, 13 }; 14 use tokio::io::AsyncWriteExt; 15 16 use super::test_support::{allocations, gossip_network, node, queue_plaintext_burn}; 17 18 #[tokio::test] 19 async fn full_outbound_queue_is_metric_not_peer_error() { 20 let wallet = Wallet::from_seed("full-outbound-queue"); 21 let node = Arc::new(tokio::sync::Mutex::new(node( 22 "full-outbound-queue", 23 wallet.clone(), 24 allocations(&[wallet], 1_000), 25 ))); 26 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::from_addresses(vec![ 27 "127.0.0.1:9444".to_string(), 28 ]))); 29 let network = super::GossipNetwork { 30 inner: Arc::new(super::GossipNetworkInner { 31 node, 32 peers: Arc::clone(&peers), 33 listen_addr: "127.0.0.1:9544".parse().unwrap(), 34 p2p_announce_addr: tokio::sync::Mutex::new(None), 35 node_id: super::new_node_id(), 36 accept_task: tokio::sync::Mutex::new(None), 37 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 38 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 39 metrics: super::P2pMetricsCounters::default(), 40 }), 41 }; 42 let (sender, _receiver) = tokio::sync::mpsc::channel(1); 43 sender 44 .try_send(vec![GossipEnvelope::PeerStatus { 45 height: 1, 46 tip_hash: "queued".to_string(), 47 time_ms: 1_000, 48 }]) 49 .unwrap(); 50 network 51 .inner 52 .sessions 53 .lock() 54 .await 55 .insert("127.0.0.1:9444".to_string(), sender); 56 57 network 58 .broadcast(vec![GossipEnvelope::PeerStatus { 59 height: 2, 60 tip_hash: "new".to_string(), 61 time_ms: 2_000, 62 }]) 63 .await 64 .unwrap(); 65 66 assert_eq!(network.metrics().outbound_queue_full, 1); 67 let peer = peers.lock().await.list().pop().unwrap(); 68 assert_eq!(peer.last_error, None); 69 assert_eq!(peer.last_error_ms, None); 70 } 71 72 #[tokio::test] 73 async fn single_block_fork_error_requests_chain_snapshot() { 74 let alice = Wallet::from_seed("single-block-fork-alice"); 75 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 76 let mut local_node = node( 77 "local-single-block-fork", 78 alice.clone(), 79 allocations.clone(), 80 ); 81 let mut remote_node = node("remote-single-block-fork", alice.clone(), allocations); 82 83 queue_plaintext_burn(&mut local_node, &alice, 1); 84 local_node.drain_outbox(); 85 local_node.mine_one_at(1).unwrap(); 86 local_node.drain_outbox(); 87 88 queue_plaintext_burn(&mut remote_node, &alice, 1); 89 remote_node.drain_outbox(); 90 remote_node.mine_one_at(2).unwrap(); 91 remote_node.drain_outbox(); 92 queue_plaintext_burn(&mut remote_node, &alice, 1); 93 remote_node.drain_outbox(); 94 let remote_block = remote_node.mine_one_at(3).unwrap(); 95 assert_eq!(remote_block.height, 2); 96 assert_ne!( 97 remote_block.prev_hash, 98 local_node.ledger().tip_hash().to_string() 99 ); 100 101 let network = gossip_network( 102 Arc::new(tokio::sync::Mutex::new(local_node)), 103 Arc::new(tokio::sync::Mutex::new(PeerBook::default())), 104 "127.0.0.1:9544".parse().unwrap(), 105 None, 106 ); 107 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); 108 let client = tokio::net::TcpStream::connect(listener.local_addr().unwrap()) 109 .await 110 .unwrap(); 111 let (server, remote_addr) = listener.accept().await.unwrap(); 112 let (_server_reader, mut server_writer) = server.into_split(); 113 let (client_reader, _client_writer) = client.into_split(); 114 let mut client_reader = super::LimitedLineReader::new(client_reader); 115 let mut known_peer = None; 116 117 super::process_envelope( 118 &network, 119 &mut server_writer, 120 remote_addr, 121 &mut known_peer, 122 GossipEnvelope::Block(remote_block), 123 ) 124 .await 125 .unwrap(); 126 127 let line = tokio::time::timeout(std::time::Duration::from_secs(1), client_reader.read_line()) 128 .await 129 .unwrap() 130 .unwrap() 131 .unwrap(); 132 assert!(matches!( 133 super::parse_envelope(&line).unwrap(), 134 GossipEnvelope::ChainSnapshotRequest 135 )); 136 } 137 138 #[tokio::test] 139 async fn hello_rejects_wrong_network_or_genesis_without_banning() { 140 let alice = Wallet::from_seed("hello-alice"); 141 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 142 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 143 let network = super::GossipNetwork { 144 inner: Arc::new(super::GossipNetworkInner { 145 node, 146 peers: Arc::new(tokio::sync::Mutex::new(PeerBook::default())), 147 listen_addr: "127.0.0.1:9544".parse().unwrap(), 148 p2p_announce_addr: tokio::sync::Mutex::new(None), 149 node_id: super::new_node_id(), 150 accept_task: tokio::sync::Mutex::new(None), 151 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 152 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 153 metrics: super::P2pMetricsCounters::default(), 154 }), 155 }; 156 157 let wrong_network = ProtocolHello { 158 protocol_version: PROTOCOL_VERSION, 159 network_id: "other-network".to_string(), 160 genesis_hash: network 161 .inner 162 .node 163 .lock() 164 .await 165 .ledger() 166 .genesis_hash() 167 .to_string(), 168 listen_addr: None, 169 node_id: None, 170 height: 0, 171 tip_hash: "tip".to_string(), 172 time_ms: 1_000, 173 }; 174 assert!( 175 super::process_hello( 176 &network, 177 "127.0.0.1:9545".parse().unwrap(), 178 &mut None, 179 wrong_network, 180 ) 181 .await 182 .unwrap_err() 183 .to_string() 184 .contains("wrong network") 185 ); 186 187 let wrong_genesis = ProtocolHello { 188 protocol_version: PROTOCOL_VERSION, 189 network_id: NETWORK_ID.to_string(), 190 genesis_hash: "not-local-genesis".to_string(), 191 listen_addr: Some("127.0.0.1:9545".to_string()), 192 node_id: None, 193 height: 0, 194 tip_hash: "tip".to_string(), 195 time_ms: 1_000, 196 }; 197 assert!( 198 super::process_hello( 199 &network, 200 "127.0.0.1:9545".parse().unwrap(), 201 &mut None, 202 wrong_genesis, 203 ) 204 .await 205 .unwrap_err() 206 .to_string() 207 .contains("wrong genesis") 208 ); 209 210 let wrong_protocol = ProtocolHello { 211 protocol_version: PROTOCOL_VERSION + 1, 212 network_id: NETWORK_ID.to_string(), 213 genesis_hash: network 214 .inner 215 .node 216 .lock() 217 .await 218 .ledger() 219 .genesis_hash() 220 .to_string(), 221 listen_addr: Some("127.0.0.1:9545".to_string()), 222 node_id: None, 223 height: 0, 224 tip_hash: "tip".to_string(), 225 time_ms: 1_000, 226 }; 227 assert!( 228 super::process_hello( 229 &network, 230 "127.0.0.1:9545".parse().unwrap(), 231 &mut None, 232 wrong_protocol, 233 ) 234 .await 235 .unwrap_err() 236 .to_string() 237 .contains("unsupported protocol version") 238 ); 239 240 assert!(network.inner.peers.lock().await.list().is_empty()); 241 } 242 243 #[tokio::test] 244 async fn hello_records_remote_clock_observation() { 245 let alice = Wallet::from_seed("hello-clock-alice"); 246 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 247 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 248 let network = super::GossipNetwork { 249 inner: Arc::new(super::GossipNetworkInner { 250 node, 251 peers: Arc::new(tokio::sync::Mutex::new(PeerBook::default())), 252 listen_addr: "127.0.0.1:9544".parse().unwrap(), 253 p2p_announce_addr: tokio::sync::Mutex::new(None), 254 node_id: super::new_node_id(), 255 accept_task: tokio::sync::Mutex::new(None), 256 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 257 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 258 metrics: super::P2pMetricsCounters::default(), 259 }), 260 }; 261 let remote_time_ms = crate::app::now_ms().saturating_add(60_000); 262 let hello = ProtocolHello { 263 protocol_version: PROTOCOL_VERSION, 264 network_id: NETWORK_ID.to_string(), 265 genesis_hash: network 266 .inner 267 .node 268 .lock() 269 .await 270 .ledger() 271 .genesis_hash() 272 .to_string(), 273 listen_addr: Some("127.0.0.1:9545".to_string()), 274 node_id: None, 275 height: 0, 276 tip_hash: "tip".to_string(), 277 time_ms: remote_time_ms, 278 }; 279 280 let mut known_peer = None; 281 super::process_hello( 282 &network, 283 "127.0.0.1:9545".parse().unwrap(), 284 &mut known_peer, 285 hello, 286 ) 287 .await 288 .unwrap(); 289 290 let peers = network.inner.peers.lock().await.list(); 291 let peer = peers 292 .iter() 293 .find(|peer| peer.address == "127.0.0.1:9545") 294 .unwrap(); 295 assert!(peer.last_clock_offset_ms.unwrap() > 30_000); 296 assert_eq!(peer.last_clock_offset_accepted, Some(true)); 297 } 298 299 #[tokio::test] 300 async fn hello_remembers_advertised_address_after_signed_session_and_dialback() { 301 let alice = Wallet::from_seed("hello-dialback-alice"); 302 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 303 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 304 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::default())); 305 let network = super::GossipNetwork { 306 inner: Arc::new(super::GossipNetworkInner { 307 node: Arc::clone(&node), 308 peers: Arc::clone(&peers), 309 listen_addr: "127.0.0.1:9544".parse().unwrap(), 310 p2p_announce_addr: tokio::sync::Mutex::new(None), 311 node_id: super::new_node_id(), 312 accept_task: tokio::sync::Mutex::new(None), 313 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 314 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 315 metrics: super::P2pMetricsCounters::default(), 316 }), 317 }; 318 let remote_node_id = super::new_node_id(); 319 let remote_addr = spawn_hello_server(ProtocolHello { 320 protocol_version: PROTOCOL_VERSION, 321 network_id: NETWORK_ID.to_string(), 322 genesis_hash: node.lock().await.ledger().genesis_hash().to_string(), 323 listen_addr: None, 324 node_id: Some(remote_node_id.clone()), 325 height: 0, 326 tip_hash: "tip".to_string(), 327 time_ms: 1_000, 328 }) 329 .await; 330 let original_addr = spawn_verification_responder(remote_node_id.clone()).await; 331 let stream = tokio::net::TcpStream::connect(original_addr).await.unwrap(); 332 let remote_socket = stream.peer_addr().unwrap(); 333 let (reader, mut writer) = stream.into_split(); 334 let mut reader = super::LimitedLineReader::new(reader); 335 let hello = ProtocolHello { 336 protocol_version: PROTOCOL_VERSION, 337 network_id: NETWORK_ID.to_string(), 338 genesis_hash: node.lock().await.ledger().genesis_hash().to_string(), 339 listen_addr: Some(remote_addr.to_string()), 340 node_id: Some(remote_node_id), 341 height: 0, 342 tip_hash: "tip".to_string(), 343 time_ms: 1_000, 344 }; 345 let mut known_peer = None; 346 347 super::process_hello_with_verification( 348 &network, 349 &mut writer, 350 &mut reader, 351 "test-original-peer", 352 remote_socket, 353 &mut known_peer, 354 hello, 355 ) 356 .await 357 .unwrap(); 358 359 assert_eq!(known_peer, Some(remote_addr.to_string())); 360 let listed = peers.lock().await.list(); 361 let peer = listed 362 .iter() 363 .find(|peer| peer.address == remote_addr.to_string()) 364 .unwrap(); 365 assert_eq!(peer.direction, PeerDirection::Discovered); 366 assert!( 367 peers 368 .lock() 369 .await 370 .addresses() 371 .contains(&remote_addr.to_string()) 372 ); 373 } 374 375 #[tokio::test] 376 async fn hello_ignores_advertised_address_when_connected_peer_cannot_sign_claimed_node_id() { 377 let alice = Wallet::from_seed("hello-dialback-spoof-alice"); 378 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 379 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 380 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::default())); 381 let network = super::GossipNetwork { 382 inner: Arc::new(super::GossipNetworkInner { 383 node: Arc::clone(&node), 384 peers: Arc::clone(&peers), 385 listen_addr: "127.0.0.1:9544".parse().unwrap(), 386 p2p_announce_addr: tokio::sync::Mutex::new(None), 387 node_id: super::new_node_id(), 388 accept_task: tokio::sync::Mutex::new(None), 389 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 390 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 391 metrics: super::P2pMetricsCounters::default(), 392 }), 393 }; 394 let victim_node_id = super::new_node_id(); 395 let attacker_node_id = super::new_node_id(); 396 let remote_addr = spawn_hello_server(ProtocolHello { 397 protocol_version: PROTOCOL_VERSION, 398 network_id: NETWORK_ID.to_string(), 399 genesis_hash: node.lock().await.ledger().genesis_hash().to_string(), 400 listen_addr: None, 401 node_id: Some(victim_node_id.clone()), 402 height: 0, 403 tip_hash: "tip".to_string(), 404 time_ms: 1_000, 405 }) 406 .await; 407 let original_addr = spawn_verification_responder(attacker_node_id).await; 408 let stream = tokio::net::TcpStream::connect(original_addr).await.unwrap(); 409 let remote_socket = stream.peer_addr().unwrap(); 410 let (reader, mut writer) = stream.into_split(); 411 let mut reader = super::LimitedLineReader::new(reader); 412 let hello = ProtocolHello { 413 protocol_version: PROTOCOL_VERSION, 414 network_id: NETWORK_ID.to_string(), 415 genesis_hash: node.lock().await.ledger().genesis_hash().to_string(), 416 listen_addr: Some(remote_addr.to_string()), 417 node_id: Some(victim_node_id), 418 height: 0, 419 tip_hash: "tip".to_string(), 420 time_ms: 1_000, 421 }; 422 let mut known_peer = None; 423 424 super::process_hello_with_verification( 425 &network, 426 &mut writer, 427 &mut reader, 428 "test-attacker-peer", 429 remote_socket, 430 &mut known_peer, 431 hello, 432 ) 433 .await 434 .unwrap(); 435 436 assert!(known_peer.is_none()); 437 assert!( 438 !peers 439 .lock() 440 .await 441 .addresses() 442 .contains(&remote_addr.to_string()) 443 ); 444 } 445 446 #[tokio::test] 447 async fn dialback_rejects_address_that_signs_with_different_node_id() { 448 let alice = Wallet::from_seed("hello-dialback-mismatch-alice"); 449 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 450 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 451 let network = super::GossipNetwork { 452 inner: Arc::new(super::GossipNetworkInner { 453 node: Arc::clone(&node), 454 peers: Arc::new(tokio::sync::Mutex::new(PeerBook::default())), 455 listen_addr: "127.0.0.1:9544".parse().unwrap(), 456 p2p_announce_addr: tokio::sync::Mutex::new(None), 457 node_id: super::new_node_id(), 458 accept_task: tokio::sync::Mutex::new(None), 459 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 460 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 461 metrics: super::P2pMetricsCounters::default(), 462 }), 463 }; 464 let honest_node_id = super::new_node_id(); 465 let claimed_node_id = super::new_node_id(); 466 let remote_addr = spawn_hello_server(ProtocolHello { 467 protocol_version: PROTOCOL_VERSION, 468 network_id: NETWORK_ID.to_string(), 469 genesis_hash: node.lock().await.ledger().genesis_hash().to_string(), 470 listen_addr: None, 471 node_id: Some(honest_node_id), 472 height: 0, 473 tip_hash: "tip".to_string(), 474 time_ms: 1_000, 475 }) 476 .await; 477 478 assert!( 479 !super::verify_advertised_peer_node_id( 480 &network, 481 &remote_addr.to_string(), 482 &claimed_node_id 483 ) 484 .await 485 ); 486 } 487 488 #[tokio::test] 489 async fn inbound_verification_only_session_closes_after_response() { 490 let alice = Wallet::from_seed("verification-only-close-alice"); 491 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 492 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 493 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::default())); 494 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); 495 let listen_addr = listener.local_addr().unwrap(); 496 drop(listener); 497 let network = super::GossipNetwork::start(node, peers, listen_addr, None, true) 498 .await 499 .unwrap(); 500 501 let stream = tokio::net::TcpStream::connect(listen_addr).await.unwrap(); 502 let (reader, mut writer) = stream.into_split(); 503 let mut reader = super::LimitedLineReader::new(reader); 504 let hello_line = reader.read_line().await.unwrap().unwrap(); 505 let node_id = match super::parse_envelope(&hello_line).unwrap() { 506 GossipEnvelope::Hello(hello) => hello.node_id.unwrap(), 507 other => panic!("expected hello, got {other:?}"), 508 }; 509 let nonce = super::new_verification_nonce(); 510 super::write_envelope( 511 &mut writer, 512 &GossipEnvelope::PeerVerificationChallenge { 513 address: listen_addr.to_string(), 514 nonce: nonce.clone(), 515 }, 516 ) 517 .await 518 .unwrap(); 519 520 let response_line = tokio::time::timeout(std::time::Duration::from_secs(1), reader.read_line()) 521 .await 522 .unwrap() 523 .unwrap() 524 .unwrap(); 525 match super::parse_envelope(&response_line).unwrap() { 526 GossipEnvelope::PeerVerificationResponse { 527 address, 528 nonce: response_nonce, 529 node_id: response_node_id, 530 signature, 531 } => assert!(super::peer_verification_response_is_valid( 532 &address, 533 &response_nonce, 534 &response_node_id, 535 &signature, 536 &listen_addr.to_string(), 537 &nonce, 538 &node_id, 539 )), 540 other => panic!("expected verification response, got {other:?}"), 541 } 542 543 let closed = tokio::time::timeout(std::time::Duration::from_secs(1), reader.read_line()) 544 .await 545 .unwrap() 546 .unwrap(); 547 assert!(closed.is_none()); 548 network.set_accept_inbound(false).await.unwrap(); 549 } 550 551 #[tokio::test] 552 async fn setup_placeholder_accepts_remote_genesis_and_adopts_snapshot() { 553 let local_wallet = Wallet::from_seed("setup-placeholder-local"); 554 let local_ledger = Ledger::new(BTreeMap::new(), 1); 555 let local_node = Arc::new(tokio::sync::Mutex::new(NodeCore::from_ledger( 556 local_wallet, 557 local_ledger.clone(), 558 0, 559 ))); 560 let network = super::GossipNetwork { 561 inner: Arc::new(super::GossipNetworkInner { 562 node: local_node, 563 peers: Arc::new(tokio::sync::Mutex::new(PeerBook::from_addresses(vec![ 564 "iuna.jhx.app:9444".to_string(), 565 ]))), 566 listen_addr: "127.0.0.1:9544".parse().unwrap(), 567 p2p_announce_addr: tokio::sync::Mutex::new(None), 568 node_id: super::new_node_id(), 569 accept_task: tokio::sync::Mutex::new(None), 570 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 571 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 572 metrics: super::P2pMetricsCounters::default(), 573 }), 574 }; 575 576 let remote_wallet = Wallet::from_seed("setup-placeholder-remote"); 577 let remote_snapshot = node( 578 "remote", 579 remote_wallet.clone(), 580 allocations(std::slice::from_ref(&remote_wallet), 1_000), 581 ) 582 .chain_snapshot(); 583 let remote_genesis = remote_snapshot.blocks[0].hash.clone(); 584 let hello = ProtocolHello { 585 protocol_version: PROTOCOL_VERSION, 586 network_id: NETWORK_ID.to_string(), 587 genesis_hash: remote_genesis.clone(), 588 listen_addr: Some("142.132.164.59:9444".to_string()), 589 node_id: None, 590 height: 5, 591 tip_hash: "remote-tip".to_string(), 592 time_ms: 1_000, 593 }; 594 let mut known_peer = Some("iuna.jhx.app:9444".to_string()); 595 let peer_status = super::process_hello( 596 &network, 597 "142.132.164.59:51234".parse().unwrap(), 598 &mut known_peer, 599 hello, 600 ) 601 .await 602 .unwrap(); 603 604 assert!(peer_status.request_snapshot); 605 assert!(!peer_status.push_snapshot); 606 assert_eq!(known_peer.as_deref(), Some("iuna.jhx.app:9444")); 607 let listed = network.inner.peers.lock().await.list(); 608 assert_eq!(listed.len(), 1); 609 let peer = listed 610 .into_iter() 611 .find(|peer| peer.address == "iuna.jhx.app:9444") 612 .unwrap(); 613 assert_eq!(peer.misbehavior_score, 0); 614 assert!(!peer.is_banned_at(crate::app::now_ms())); 615 616 let adopted = 617 super::validate_snapshot_extension(local_ledger, remote_snapshot, crate::app::now_ms()) 618 .await 619 .unwrap(); 620 assert_eq!(adopted.genesis_hash(), remote_genesis); 621 assert!( 622 network 623 .inner 624 .node 625 .lock() 626 .await 627 .import_verified_ledger(adopted) 628 .unwrap() 629 ); 630 assert_eq!( 631 network.inner.node.lock().await.ledger().genesis_hash(), 632 remote_genesis 633 ); 634 } 635 636 #[tokio::test] 637 async fn real_node_accepts_setup_placeholder_peer_and_pushes_snapshot() { 638 let wallet = Wallet::from_seed("setup-placeholder-peer-real-node"); 639 let node = Arc::new(tokio::sync::Mutex::new(node( 640 "real", 641 wallet.clone(), 642 allocations(std::slice::from_ref(&wallet), 1_000), 643 ))); 644 let network = super::GossipNetwork { 645 inner: Arc::new(super::GossipNetworkInner { 646 node: Arc::clone(&node), 647 peers: Arc::new(tokio::sync::Mutex::new(PeerBook::default())), 648 listen_addr: "127.0.0.1:9544".parse().unwrap(), 649 p2p_announce_addr: tokio::sync::Mutex::new(None), 650 node_id: super::new_node_id(), 651 accept_task: tokio::sync::Mutex::new(None), 652 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 653 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 654 metrics: super::P2pMetricsCounters::default(), 655 }), 656 }; 657 let setup_ledger = Ledger::new(BTreeMap::new(), 1); 658 let hello = ProtocolHello { 659 protocol_version: PROTOCOL_VERSION, 660 network_id: NETWORK_ID.to_string(), 661 genesis_hash: setup_ledger.genesis_hash().to_string(), 662 listen_addr: Some("127.0.0.1:9545".to_string()), 663 node_id: None, 664 height: 0, 665 tip_hash: setup_ledger.status().tip_hash, 666 time_ms: 1_000, 667 }; 668 669 let peer_status = super::process_hello( 670 &network, 671 "127.0.0.1:51234".parse().unwrap(), 672 &mut None, 673 hello, 674 ) 675 .await 676 .unwrap(); 677 678 assert!(!peer_status.request_snapshot); 679 assert!(peer_status.push_snapshot); 680 let payload = super::catchup_payload_for_peer(&node, &peer_status).await; 681 assert!(matches!( 682 payload.as_slice(), 683 [GossipEnvelope::ChainSnapshot(_)] 684 )); 685 } 686 687 #[tokio::test] 688 async fn hello_ignores_private_advertised_listen_address() { 689 let alice = Wallet::from_seed("hello-private-listen-alice"); 690 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 691 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 692 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::default())); 693 let network = super::GossipNetwork { 694 inner: Arc::new(super::GossipNetworkInner { 695 node: Arc::clone(&node), 696 peers: Arc::clone(&peers), 697 listen_addr: "0.0.0.0:9444".parse().unwrap(), 698 p2p_announce_addr: tokio::sync::Mutex::new(None), 699 node_id: super::new_node_id(), 700 accept_task: tokio::sync::Mutex::new(None), 701 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 702 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 703 metrics: super::P2pMetricsCounters::default(), 704 }), 705 }; 706 let status = node.lock().await.ledger().status(); 707 let hello = ProtocolHello { 708 protocol_version: PROTOCOL_VERSION, 709 network_id: NETWORK_ID.to_string(), 710 genesis_hash: node.lock().await.ledger().genesis_hash().to_string(), 711 listen_addr: Some("10.42.1.1:12138".to_string()), 712 node_id: None, 713 height: status.height, 714 tip_hash: status.tip_hash, 715 time_ms: 1_000, 716 }; 717 718 let mut known_peer = None; 719 super::process_hello( 720 &network, 721 "142.132.164.59:51234".parse().unwrap(), 722 &mut known_peer, 723 hello, 724 ) 725 .await 726 .unwrap(); 727 728 assert!(known_peer.is_none()); 729 assert!(peers.lock().await.addresses().is_empty()); 730 } 731 732 #[tokio::test] 733 async fn hello_ignores_loopback_alias_for_unspecified_self() { 734 let alice = Wallet::from_seed("hello-self-alias-alice"); 735 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 736 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 737 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::default())); 738 let network = super::GossipNetwork { 739 inner: Arc::new(super::GossipNetworkInner { 740 node, 741 peers: Arc::clone(&peers), 742 listen_addr: "0.0.0.0:9545".parse().unwrap(), 743 p2p_announce_addr: tokio::sync::Mutex::new(None), 744 node_id: super::new_node_id(), 745 accept_task: tokio::sync::Mutex::new(None), 746 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 747 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 748 metrics: super::P2pMetricsCounters::default(), 749 }), 750 }; 751 let hello = ProtocolHello { 752 protocol_version: PROTOCOL_VERSION, 753 network_id: NETWORK_ID.to_string(), 754 genesis_hash: network 755 .inner 756 .node 757 .lock() 758 .await 759 .ledger() 760 .genesis_hash() 761 .to_string(), 762 listen_addr: Some("127.0.0.1:9545".to_string()), 763 node_id: None, 764 height: 0, 765 tip_hash: "tip".to_string(), 766 time_ms: 1_000, 767 }; 768 769 super::process_hello( 770 &network, 771 "127.0.0.1:52144".parse().unwrap(), 772 &mut None, 773 hello, 774 ) 775 .await 776 .unwrap(); 777 778 assert_eq!(network.metrics().self_peer_rejections, 1); 779 assert!(peers.lock().await.addresses().is_empty()); 780 let listed = peers.lock().await.list(); 781 assert_eq!(listed.len(), 1); 782 assert_eq!(listed[0].direction, PeerDirection::Inbound); 783 } 784 785 #[tokio::test] 786 async fn hello_removes_outbound_peer_that_announces_self_address() { 787 let alice = Wallet::from_seed("hello-self-outbound-alice"); 788 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 789 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 790 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::from_addresses(vec![ 791 "10.42.1.1:16987".to_string(), 792 ]))); 793 let network = super::GossipNetwork { 794 inner: Arc::new(super::GossipNetworkInner { 795 node, 796 peers: Arc::clone(&peers), 797 listen_addr: "0.0.0.0:9444".parse().unwrap(), 798 p2p_announce_addr: tokio::sync::Mutex::new(None), 799 node_id: super::new_node_id(), 800 accept_task: tokio::sync::Mutex::new(None), 801 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 802 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 803 metrics: super::P2pMetricsCounters::default(), 804 }), 805 }; 806 let hello = ProtocolHello { 807 protocol_version: PROTOCOL_VERSION, 808 network_id: NETWORK_ID.to_string(), 809 genesis_hash: network 810 .inner 811 .node 812 .lock() 813 .await 814 .ledger() 815 .genesis_hash() 816 .to_string(), 817 listen_addr: Some("127.0.0.1:9444".to_string()), 818 node_id: None, 819 height: 0, 820 tip_hash: "tip".to_string(), 821 time_ms: 1_000, 822 }; 823 let mut known_peer = Some("10.42.1.1:16987".to_string()); 824 825 super::process_hello( 826 &network, 827 "10.42.1.1:16987".parse().unwrap(), 828 &mut known_peer, 829 hello, 830 ) 831 .await 832 .unwrap(); 833 834 assert_eq!(network.metrics().self_peer_rejections, 1); 835 assert!(known_peer.is_none()); 836 assert!(peers.lock().await.addresses().is_empty()); 837 } 838 839 #[tokio::test] 840 async fn hello_removes_outbound_peer_with_same_node_id() { 841 let alice = Wallet::from_seed("hello-self-node-id-alice"); 842 let allocations = allocations(std::slice::from_ref(&alice), 1_000); 843 let node = Arc::new(tokio::sync::Mutex::new(node("alice", alice, allocations))); 844 let peers = Arc::new(tokio::sync::Mutex::new(PeerBook::from_addresses(vec![ 845 "142.132.164.59:9444".to_string(), 846 ]))); 847 let network = super::GossipNetwork { 848 inner: Arc::new(super::GossipNetworkInner { 849 node, 850 peers: Arc::clone(&peers), 851 listen_addr: "0.0.0.0:9444".parse().unwrap(), 852 p2p_announce_addr: tokio::sync::Mutex::new(None), 853 node_id: super::new_node_id(), 854 accept_task: tokio::sync::Mutex::new(None), 855 sessions: tokio::sync::Mutex::new(BTreeMap::new()), 856 inbound_limiter: Arc::new(StdMutex::new(super::InboundConnectionLimiter::default())), 857 metrics: super::P2pMetricsCounters::default(), 858 }), 859 }; 860 let hello = ProtocolHello { 861 protocol_version: PROTOCOL_VERSION, 862 network_id: NETWORK_ID.to_string(), 863 genesis_hash: network 864 .inner 865 .node 866 .lock() 867 .await 868 .ledger() 869 .genesis_hash() 870 .to_string(), 871 listen_addr: Some("0.0.0.0:9444".to_string()), 872 node_id: Some(network.inner.node_id.clone()), 873 height: 0, 874 tip_hash: "tip".to_string(), 875 time_ms: 1_000, 876 }; 877 let mut known_peer = Some("142.132.164.59:9444".to_string()); 878 879 super::process_hello( 880 &network, 881 "142.132.164.59:52144".parse().unwrap(), 882 &mut known_peer, 883 hello, 884 ) 885 .await 886 .unwrap(); 887 888 assert_eq!(network.metrics().self_peer_rejections, 1); 889 assert!(known_peer.is_none()); 890 assert!(peers.lock().await.addresses().is_empty()); 891 } 892 893 async fn spawn_hello_server(hello: ProtocolHello) -> SocketAddr { 894 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); 895 let addr = listener.local_addr().unwrap(); 896 tokio::spawn(async move { 897 let Ok((stream, _)) = listener.accept().await else { 898 return; 899 }; 900 let node_id = hello.node_id.clone(); 901 let (reader, mut writer) = stream.into_split(); 902 let line = serde_json::to_string(&GossipEnvelope::Hello(hello)).unwrap(); 903 let _ = writer.write_all(line.as_bytes()).await; 904 let _ = writer.write_all(b"\n").await; 905 let Some(node_id) = node_id else { 906 return; 907 }; 908 let mut reader = super::LimitedLineReader::new(reader); 909 let Ok(Some(line)) = reader.read_line().await else { 910 return; 911 }; 912 let Ok(GossipEnvelope::PeerVerificationChallenge { address, nonce }) = 913 super::parse_envelope(&line) 914 else { 915 return; 916 }; 917 let Some(response) = 918 super::peer_verification_response_for_node_id(&node_id, &address, &nonce) 919 else { 920 return; 921 }; 922 let line = serde_json::to_string(&response).unwrap(); 923 let _ = writer.write_all(line.as_bytes()).await; 924 let _ = writer.write_all(b"\n").await; 925 }); 926 addr 927 } 928 929 async fn spawn_verification_responder(node_id: String) -> SocketAddr { 930 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); 931 let addr = listener.local_addr().unwrap(); 932 tokio::spawn(async move { 933 let Ok((stream, _)) = listener.accept().await else { 934 return; 935 }; 936 let (reader, mut writer) = stream.into_split(); 937 let mut reader = super::LimitedLineReader::new(reader); 938 let Ok(Some(line)) = reader.read_line().await else { 939 return; 940 }; 941 let Ok(GossipEnvelope::PeerVerificationChallenge { address, nonce }) = 942 super::parse_envelope(&line) 943 else { 944 return; 945 }; 946 let Some(response) = 947 super::peer_verification_response_for_node_id(&node_id, &address, &nonce) 948 else { 949 return; 950 }; 951 let line = serde_json::to_string(&response).unwrap(); 952 let _ = writer.write_all(line.as_bytes()).await; 953 let _ = writer.write_all(b"\n").await; 954 }); 955 addr 956 }