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