1use crate::{
22 config::{
23 FullNetworkConfiguration, IncomingRequest, NodeKeyConfig, NotificationHandshake, Params,
24 SetConfig, TransportConfig,
25 },
26 error::Error,
27 event::{DhtEvent, Event},
28 litep2p::{
29 discovery::{Discovery, DiscoveryEvent},
30 ipfs_dht::IpfsDht,
31 peerstore::Peerstore,
32 service::{Litep2pNetworkService, NetworkServiceCommand},
33 shim::{
34 notification::{
35 config::{NotificationProtocolConfig, ProtocolControlHandle},
36 peerset::PeersetCommand,
37 },
38 request_response::{RequestResponseConfig, RequestResponseProtocol},
39 },
40 },
41 peer_store::PeerStoreProvider,
42 service::{
43 metrics::{register_without_sources, MetricSources, Metrics, NotificationMetrics},
44 out_events,
45 traits::{BandwidthSink, NetworkBackend, NetworkService},
46 },
47 webrtc, NetworkStatus, NotificationService, ProtocolName,
48};
49
50use codec::Encode;
51use futures::StreamExt;
52use litep2p::{
53 config::ConfigBuilder,
54 crypto::ed25519::Keypair,
55 error::{DialError, NegotiationError},
56 executor::Executor,
57 protocol::{
58 libp2p::kademlia::{QueryId, Record},
59 request_response::ConfigBuilder as RequestResponseConfigBuilder,
60 },
61 transport::{
62 tcp::config::Config as TcpTransportConfig, webrtc::config::Config as WebRtcTransportConfig,
63 websocket::config::Config as WebSocketTransportConfig, ConnectionLimitsConfig, Endpoint,
64 },
65 types::{
66 multiaddr::{Multiaddr, Protocol},
67 ConnectionId,
68 },
69 Litep2p, Litep2pEvent, ProtocolName as Litep2pProtocolName,
70};
71use prometheus_endpoint::Registry;
72
73use sc_network_common::{role::Roles, ExHashT};
74use sc_network_types::{
75 kad::{Key as RecordKey, PeerRecord, Record as P2PRecord},
76 multiaddr::Protocol as NetworkProtocol,
77 PeerId,
78};
79use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver};
80use sp_runtime::traits::Block as BlockT;
81
82use std::{
83 cmp,
84 collections::{hash_map::Entry, HashMap, HashSet},
85 fs,
86 future::Future,
87 iter,
88 pin::Pin,
89 sync::{
90 atomic::{AtomicUsize, Ordering},
91 Arc,
92 },
93 time::{Duration, Instant},
94};
95
96mod discovery;
97mod ipfs_dht;
98mod peerstore;
99mod service;
100mod shim;
101
102struct Litep2pBandwidthSink {
104 sink: litep2p::BandwidthSink,
105}
106
107impl BandwidthSink for Litep2pBandwidthSink {
108 fn total_inbound(&self) -> u64 {
109 self.sink.inbound() as u64
110 }
111
112 fn total_outbound(&self) -> u64 {
113 self.sink.outbound() as u64
114 }
115}
116
117struct Litep2pExecutor {
119 executor: Box<dyn Fn(Pin<Box<dyn Future<Output = ()> + Send>>) + Send + Sync>,
121}
122
123impl Executor for Litep2pExecutor {
124 fn run(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
125 (self.executor)(future)
126 }
127
128 fn run_with_name(&self, _: &'static str, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
129 (self.executor)(future)
130 }
131}
132
133const LOG_TARGET: &str = "sub-libp2p";
135
136struct ConnectionContext {
138 endpoints: HashMap<ConnectionId, Endpoint>,
140
141 num_connections: usize,
143
144 supported_protocols: Option<HashSet<ProtocolName>>,
147}
148
149#[derive(Debug)]
151enum KadQuery {
152 FindNode(PeerId, Instant),
154 GetValue(RecordKey, Instant),
156 PutValue(RecordKey, Instant),
158 GetProviders(RecordKey, Instant),
160 AddProvider(RecordKey, Instant),
162}
163
164pub struct Litep2pNetworkBackend {
166 litep2p: Litep2p,
168
169 network_service: Arc<dyn NetworkService>,
171
172 cmd_rx: TracingUnboundedReceiver<NetworkServiceCommand>,
174
175 peerset_handles: HashMap<ProtocolName, ProtocolControlHandle>,
177
178 pending_queries: HashMap<QueryId, KadQuery>,
180
181 discovery: Discovery,
183
184 num_connected: Arc<AtomicUsize>,
186
187 peers: HashMap<litep2p::PeerId, ConnectionContext>,
189
190 peerstore_handle: Arc<dyn PeerStoreProvider>,
192
193 block_announce_protocol: ProtocolName,
195
196 event_streams: out_events::OutChannels,
198
199 metrics: Option<Metrics>,
201}
202
203impl Litep2pNetworkBackend {
204 fn parse_addresses(
207 addresses: impl Iterator<Item = Multiaddr>,
208 ) -> HashMap<PeerId, Vec<Multiaddr>> {
209 addresses
210 .into_iter()
211 .filter_map(|address| match address.iter().next() {
212 Some(
213 Protocol::Dns(_) |
214 Protocol::Dns4(_) |
215 Protocol::Dns6(_) |
216 Protocol::Ip6(_) |
217 Protocol::Ip4(_),
218 ) => match address.iter().find(|protocol| std::matches!(protocol, Protocol::P2p(_)))
219 {
220 Some(Protocol::P2p(peer_id)) => Some((peer_id.into(), Some(address))),
221 _ => None,
222 },
223 Some(Protocol::P2p(peer_id)) => Some((peer_id.into(), None)),
224 _ => None,
225 })
226 .fold(HashMap::new(), |mut acc, (peer, maybe_address)| {
227 let entry = acc.entry(peer).or_default();
228 maybe_address.map(|address| entry.push(address));
229
230 acc
231 })
232 }
233
234 fn add_addresses(&mut self, peers: impl Iterator<Item = Multiaddr>) -> HashSet<PeerId> {
236 Self::parse_addresses(peers.into_iter())
237 .into_iter()
238 .filter_map(|(peer, addresses)| {
239 if addresses.is_empty() {
241 return Some(peer);
242 }
243
244 if self.litep2p.add_known_address(peer.into(), addresses.clone().into_iter()) == 0 {
245 log::warn!(
246 target: LOG_TARGET,
247 "couldn't add any addresses for {peer:?} and it won't be added as reserved peer",
248 );
249 return None;
250 }
251
252 self.peerstore_handle.add_known_peer(peer);
253 Some(peer)
254 })
255 .collect()
256 }
257}
258
259impl Litep2pNetworkBackend {
260 fn get_keypair(node_key: &NodeKeyConfig) -> Result<(Keypair, litep2p::PeerId), Error> {
262 let secret: litep2p::crypto::ed25519::SecretKey =
263 node_key.clone().into_keypair()?.secret().into();
264
265 let local_identity = Keypair::from(secret);
266 let local_public = local_identity.public();
267 let local_peer_id = local_public.to_peer_id();
268
269 Ok((local_identity, local_peer_id))
270 }
271
272 fn configure_transport<B: BlockT + 'static, H: ExHashT>(
274 config: &FullNetworkConfiguration<B, H, Self>,
275 keypair: Keypair,
276 ) -> Result<ConfigBuilder, Error> {
277 let _ = match config.network_config.transport {
278 TransportConfig::MemoryOnly => panic!("memory transport not supported"),
279 TransportConfig::Normal { .. } => false,
280 };
281 let config_builder = ConfigBuilder::new();
282
283 let listen_addr_len = config.network_config.listen_addresses.len();
284 let mut tcp_addresses = Vec::with_capacity(listen_addr_len);
285 let mut websocket_addresses = Vec::with_capacity(listen_addr_len);
286 let mut webrtc_addresses = Vec::with_capacity(listen_addr_len);
287
288 for addr in &config.network_config.listen_addresses {
289 let mut iter = addr.iter();
290
291 let ip_version = iter.next();
292 let Some(NetworkProtocol::Ip4(_) | NetworkProtocol::Ip6(_)) = ip_version else {
293 log::error!(
294 target: LOG_TARGET,
295 "unknown protocol {ip_version:?}, ignoring {addr:?}",
296 );
297 continue;
298 };
299
300 let transport_layer = iter.next();
301 let protocol_type = iter.next();
302
303 match (&transport_layer, &protocol_type) {
304 (Some(NetworkProtocol::Tcp(_)), Some(NetworkProtocol::P2p(_)) | None) => {
306 tcp_addresses.push(addr.clone());
307 },
308 (
310 Some(NetworkProtocol::Tcp(_)),
311 Some(NetworkProtocol::Ws(_) | NetworkProtocol::Wss(_)),
312 ) => {
313 websocket_addresses.push(addr.clone());
314 },
315 (Some(NetworkProtocol::Udp(_)), Some(NetworkProtocol::WebRTCDirect)) => {
317 webrtc::validate_listen_address(addr)?;
319 webrtc_addresses.push(addr.clone());
320 },
321 _ => {
322 log::error!(
323 target: LOG_TARGET,
324 "unknown transport layer {transport_layer:?} and protocol type {protocol_type:?}, ignoring {addr:?}",
325 );
326 },
327 };
328 }
329
330 let mut config_builder = config_builder
331 .with_websocket(WebSocketTransportConfig {
332 listen_addresses: websocket_addresses.into_iter().map(Into::into).collect(),
333 yamux_config: litep2p::yamux::Config::default(),
334 nodelay: true,
335 ..Default::default()
336 })
337 .with_tcp(TcpTransportConfig {
338 listen_addresses: tcp_addresses.into_iter().map(Into::into).collect(),
339 yamux_config: litep2p::yamux::Config::default(),
340 nodelay: true,
341 ..Default::default()
342 });
343
344 if !webrtc_addresses.is_empty() {
345 let certificate =
346 webrtc::derive_certificate(keypair.secret()).map_err(Error::Litep2p)?;
347 log::info!(target: LOG_TARGET, "WebRTC certhash: {}", certificate.certhash_b64());
348 config_builder = config_builder.with_webrtc(WebRtcTransportConfig {
349 listen_addresses: webrtc_addresses.into_iter().map(Into::into).collect(),
350 certificate: Some(certificate),
351 ..Default::default()
352 });
353 }
354
355 Ok(config_builder.with_keypair(keypair))
356 }
357}
358
359#[async_trait::async_trait]
360impl<B: BlockT + 'static, H: ExHashT> NetworkBackend<B, H> for Litep2pNetworkBackend {
361 const SUPPORTS_IPFS: bool = true;
362
363 type NotificationProtocolConfig = NotificationProtocolConfig;
364 type RequestResponseProtocolConfig = RequestResponseConfig;
365 type NetworkService<Block, Hash> = Arc<Litep2pNetworkService>;
366 type PeerStore = Peerstore;
367
368 fn new(mut params: Params<B, H, Self>) -> Result<Self, Error>
369 where
370 Self: Sized,
371 {
372 if let Err(err) = rustls::crypto::ring::default_provider().install_default() {
374 log::warn!(
375 target: LOG_TARGET,
376 "failed to install ring CryptoProvider for rustls, another provider might be installed: {err:?}",
377 );
378 }
379
380 let (keypair, local_peer_id) =
381 Self::get_keypair(¶ms.network_config.network_config.node_key)?;
382 let (cmd_tx, cmd_rx) = tracing_unbounded("mpsc_network_worker", 100_000);
383
384 params.network_config.network_config.boot_nodes = params
385 .network_config
386 .network_config
387 .boot_nodes
388 .into_iter()
389 .filter(|boot_node| boot_node.peer_id != local_peer_id.into())
390 .collect();
391 params.network_config.network_config.default_peers_set.reserved_nodes = params
392 .network_config
393 .network_config
394 .default_peers_set
395 .reserved_nodes
396 .into_iter()
397 .filter(|reserved_node| {
398 if reserved_node.peer_id == local_peer_id.into() {
399 log::warn!(
400 target: LOG_TARGET,
401 "Local peer ID used in reserved node, ignoring: {reserved_node}",
402 );
403 false
404 } else {
405 true
406 }
407 })
408 .collect();
409
410 if let Some(path) = ¶ms.network_config.network_config.net_config_path {
411 fs::create_dir_all(path)?;
412 }
413
414 log::info!(target: LOG_TARGET, "Local node identity is: {local_peer_id}");
415 log::info!(target: LOG_TARGET, "Running litep2p network backend");
416
417 params.network_config.sanity_check_addresses()?;
418 params.network_config.sanity_check_bootnodes()?;
419
420 let mut config_builder =
421 Self::configure_transport(¶ms.network_config, keypair.clone())?;
422 let known_addresses = params.network_config.known_addresses();
423 let peer_store_handle = params.network_config.peer_store_handle();
424 let executor = Arc::new(Litep2pExecutor { executor: params.executor });
425
426 let FullNetworkConfiguration {
427 notification_protocols,
428 request_response_protocols,
429 network_config,
430 ..
431 } = params.network_config;
432
433 let block_announce_protocol = params.block_announce_config.protocol_name().clone();
439 let mut notif_protocols = HashMap::from_iter([(
440 params.block_announce_config.protocol_name().clone(),
441 params.block_announce_config.handle,
442 )]);
443
444 config_builder = notification_protocols
446 .into_iter()
447 .fold(config_builder, |config_builder, mut config| {
448 config.config.set_handshake(Roles::from(¶ms.role).encode());
449 notif_protocols.insert(config.protocol_name, config.handle);
450
451 config_builder.with_notification_protocol(config.config)
452 })
453 .with_notification_protocol(params.block_announce_config.config);
454
455 let metrics = match ¶ms.metrics_registry {
457 Some(registry) => Some(register_without_sources(registry)?),
458 None => None,
459 };
460
461 let (mut request_response_receivers, request_response_senders): (
467 HashMap<_, _>,
468 HashMap<_, _>,
469 ) = request_response_protocols
470 .iter()
471 .map(|config| {
472 let (tx, rx) = tracing_unbounded("outbound-requests", 10_000);
473 ((config.protocol_name.clone(), rx), (config.protocol_name.clone(), tx))
474 })
475 .unzip();
476
477 config_builder = request_response_protocols.into_iter().fold(
478 config_builder,
479 |config_builder, config| {
480 let (protocol_config, handle) = RequestResponseConfigBuilder::new(
481 Litep2pProtocolName::from(config.protocol_name.clone()),
482 )
483 .with_max_size(cmp::max(config.max_request_size, config.max_response_size) as usize)
484 .with_fallback_names(config.fallback_names.into_iter().map(From::from).collect())
485 .with_timeout(config.request_timeout)
486 .build();
487
488 let protocol = RequestResponseProtocol::new(
489 config.protocol_name.clone(),
490 handle,
491 Arc::clone(&peer_store_handle),
492 config.inbound_queue,
493 request_response_receivers
494 .remove(&config.protocol_name)
495 .expect("receiver exists as it was just added and there are no duplicate protocols; qed"),
496 request_response_senders.clone(),
497 metrics.clone(),
498 );
499
500 executor.run(Box::pin(async move {
501 protocol.run().await;
502 }));
503
504 config_builder.with_request_response_protocol(protocol_config)
505 },
506 );
507
508 let known_addresses: HashMap<litep2p::PeerId, Vec<Multiaddr>> =
510 known_addresses.into_iter().fold(HashMap::new(), |mut acc, (peer, address)| {
511 let address = match address.iter().last() {
512 Some(
513 NetworkProtocol::Ws(_) | NetworkProtocol::Wss(_) | NetworkProtocol::Tcp(_),
514 ) => address.with(NetworkProtocol::P2p(peer.into())),
515 Some(NetworkProtocol::WebRTCDirect | NetworkProtocol::Certhash(_)) => {
516 address.with(NetworkProtocol::P2p(peer.into()))
517 },
518 Some(NetworkProtocol::P2p(_)) => address,
519 _ => return acc,
520 };
521
522 acc.entry(peer.into()).or_default().push(address.into());
523 peer_store_handle.add_known_peer(peer);
524
525 acc
526 });
527
528 let listen_addresses = Arc::new(Default::default());
530 let (discovery, ping_config, identify_config, kademlia_config, maybe_mdns_config) =
531 Discovery::new(
532 local_peer_id,
533 &network_config,
534 params.genesis_hash,
535 params.fork_id.as_deref(),
536 ¶ms.protocol_id,
537 known_addresses.clone(),
538 Arc::clone(&listen_addresses),
539 Arc::clone(&peer_store_handle),
540 );
541
542 if let Some(ipfs) = params.ipfs_config {
543 config_builder = config_builder.with_libp2p_bitswap(ipfs.litep2p_bitswap_config);
544
545 if !ipfs.bootnodes.is_empty() {
546 let (ipfs_dht, kad_config) = IpfsDht::new(ipfs.bootnodes, ipfs.block_provider);
547 config_builder = config_builder.with_libp2p_kademlia(kad_config);
548 executor.run(Box::pin(ipfs_dht.run()));
549 } else {
550 log::warn!(
551 target: LOG_TARGET,
552 "Not starting IPFS DHT publisher because no IPFS bootnodes are configured. \
553 Only direct Bitswap requests will be handled.",
554 );
555 }
556 }
557
558 config_builder = config_builder
559 .with_known_addresses(known_addresses.clone().into_iter())
560 .with_libp2p_ping(ping_config)
561 .with_libp2p_identify(identify_config)
562 .with_libp2p_kademlia(kademlia_config)
563 .with_connection_limits(ConnectionLimitsConfig::default().max_incoming_connections(
564 Some(crate::MAX_CONNECTIONS_ESTABLISHED_INCOMING as usize),
565 ))
566 .with_keep_alive_timeout(network_config.idle_connection_timeout)
567 .with_system_resolver()
570 .with_executor(executor);
571
572 if let Some(config) = maybe_mdns_config {
573 config_builder = config_builder.with_mdns(config);
574 }
575
576 let litep2p =
577 Litep2p::new(config_builder.build()).map_err(|error| Error::Litep2p(error))?;
578
579 litep2p.listen_addresses().for_each(|address| {
580 log::debug!(target: LOG_TARGET, "listening on: {address}");
581
582 listen_addresses.write().insert(address.clone());
583 });
584
585 let public_addresses = litep2p.public_addresses();
586 for address in network_config.public_addresses.iter() {
587 if let Err(err) = public_addresses.add_address(address.clone().into()) {
588 log::warn!(
589 target: LOG_TARGET,
590 "failed to add public address {address:?}: {err:?}",
591 );
592 }
593 }
594
595 let network_service = Arc::new(Litep2pNetworkService::new(
596 local_peer_id,
597 keypair.clone(),
598 cmd_tx,
599 Arc::clone(&peer_store_handle),
600 notif_protocols.clone(),
601 block_announce_protocol.clone(),
602 request_response_senders,
603 Arc::clone(&listen_addresses),
604 public_addresses,
605 ));
606
607 let num_connected = Arc::new(Default::default());
609 let bandwidth: Arc<dyn BandwidthSink> =
610 Arc::new(Litep2pBandwidthSink { sink: litep2p.bandwidth_sink() });
611
612 if let Some(registry) = ¶ms.metrics_registry {
613 MetricSources::register(registry, bandwidth)?;
614 }
615
616 Ok(Self {
617 network_service,
618 cmd_rx,
619 metrics,
620 peerset_handles: notif_protocols,
621 num_connected,
622 discovery,
623 pending_queries: HashMap::new(),
624 peerstore_handle: peer_store_handle,
625 block_announce_protocol,
626 event_streams: out_events::OutChannels::new(params.metrics_registry.as_ref())?,
627 peers: HashMap::new(),
628 litep2p,
629 })
630 }
631
632 fn network_service(&self) -> Arc<dyn NetworkService> {
633 Arc::clone(&self.network_service)
634 }
635
636 fn peer_store(
637 bootnodes: Vec<sc_network_types::PeerId>,
638 metrics_registry: Option<Registry>,
639 ) -> Self::PeerStore {
640 Peerstore::new(bootnodes, metrics_registry)
641 }
642
643 fn register_notification_metrics(registry: Option<&Registry>) -> NotificationMetrics {
644 NotificationMetrics::new(registry)
645 }
646
647 fn notification_config(
649 protocol_name: ProtocolName,
650 fallback_names: Vec<ProtocolName>,
651 max_notification_size: u64,
652 handshake: Option<NotificationHandshake>,
653 set_config: SetConfig,
654 metrics: NotificationMetrics,
655 peerstore_handle: Arc<dyn PeerStoreProvider>,
656 ) -> (Self::NotificationProtocolConfig, Box<dyn NotificationService>) {
657 Self::NotificationProtocolConfig::new(
658 protocol_name,
659 fallback_names,
660 max_notification_size as usize,
661 handshake,
662 set_config,
663 metrics,
664 peerstore_handle,
665 )
666 }
667
668 fn request_response_config(
670 protocol_name: ProtocolName,
671 fallback_names: Vec<ProtocolName>,
672 max_request_size: u64,
673 max_response_size: u64,
674 request_timeout: Duration,
675 inbound_queue: Option<async_channel::Sender<IncomingRequest>>,
676 ) -> Self::RequestResponseProtocolConfig {
677 Self::RequestResponseProtocolConfig::new(
678 protocol_name,
679 fallback_names,
680 max_request_size,
681 max_response_size,
682 request_timeout,
683 inbound_queue,
684 )
685 }
686
687 async fn run(mut self) {
689 log::debug!(target: LOG_TARGET, "starting litep2p network backend");
690
691 loop {
692 let num_connected_peers = self
693 .peerset_handles
694 .get(&self.block_announce_protocol)
695 .map_or(0usize, |handle| handle.connected_peers.load(Ordering::Relaxed));
696 self.num_connected.store(num_connected_peers, Ordering::Relaxed);
697
698 tokio::select! {
699 command = self.cmd_rx.next() => match command {
700 None => return,
701 Some(command) => match command {
702 NetworkServiceCommand::FindClosestPeers { target } => {
703 let query_id = self.discovery.find_node(target.into()).await;
704 self.pending_queries.insert(query_id, KadQuery::FindNode(target, Instant::now()));
705 }
706 NetworkServiceCommand::GetValue{ key } => {
707 let query_id = self.discovery.get_value(key.clone()).await;
708 self.pending_queries.insert(query_id, KadQuery::GetValue(key, Instant::now()));
709 }
710 NetworkServiceCommand::PutValue { key, value } => {
711 let query_id = self.discovery.put_value(key.clone(), value).await;
712 self.pending_queries.insert(query_id, KadQuery::PutValue(key, Instant::now()));
713 }
714 NetworkServiceCommand::PutValueTo { record, peers, update_local_storage} => {
715 let kademlia_key = record.key.clone();
716 let query_id = self.discovery.put_value_to_peers(record.into(), peers, update_local_storage).await;
717 self.pending_queries.insert(query_id, KadQuery::PutValue(kademlia_key, Instant::now()));
718 }
719 NetworkServiceCommand::StoreRecord { key, value, publisher, expires } => {
720 self.discovery.store_record(key, value, publisher.map(Into::into), expires).await;
721 }
722 NetworkServiceCommand::StartProviding { key } => {
723 let query_id = self.discovery.start_providing(key.clone()).await;
724 self.pending_queries.insert(query_id, KadQuery::AddProvider(key, Instant::now()));
725 }
726 NetworkServiceCommand::StopProviding { key } => {
727 self.discovery.stop_providing(key).await;
728 }
729 NetworkServiceCommand::GetProviders { key } => {
730 let query_id = self.discovery.get_providers(key.clone()).await;
731 self.pending_queries.insert(query_id, KadQuery::GetProviders(key, Instant::now()));
732 }
733 NetworkServiceCommand::EventStream { tx } => {
734 let identified = self.peers.iter().filter_map(|(peer, context)| {
735 let supported_protocols = context.supported_protocols.clone()?;
736 Some(Event::PeerIdentified { peer: (*peer).into(), supported_protocols })
737 });
738 self.event_streams.push_with(tx, identified);
739 }
740 NetworkServiceCommand::Status { tx } => {
741 let _ = tx.send(NetworkStatus {
742 num_connected_peers: self
743 .peerset_handles
744 .get(&self.block_announce_protocol)
745 .map_or(0usize, |handle| handle.connected_peers.load(Ordering::Relaxed)),
746 total_bytes_inbound: self.litep2p.bandwidth_sink().inbound() as u64,
747 total_bytes_outbound: self.litep2p.bandwidth_sink().outbound() as u64,
748 });
749 }
750 NetworkServiceCommand::AddPeersToReservedSet {
751 protocol,
752 peers,
753 } => {
754 let peers = self.add_addresses(peers.into_iter().map(Into::into));
755
756 match self.peerset_handles.get(&protocol) {
757 Some(handle) => {
758 let _ = handle.tx.unbounded_send(PeersetCommand::AddReservedPeers { peers });
759 }
760 None => log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist"),
761 };
762 }
763 NetworkServiceCommand::AddKnownAddress { peer, address } => {
764 let mut address: Multiaddr = address.into();
765
766 if !address.iter().any(|protocol| std::matches!(protocol, Protocol::P2p(_))) {
767 address.push(Protocol::P2p(litep2p::PeerId::from(peer).into()));
768 }
769
770 if self.litep2p.add_known_address(peer.into(), iter::once(address.clone())) > 0 {
771 self.peerstore_handle.add_known_peer(peer);
775 } else {
776 log::debug!(
777 target: LOG_TARGET,
778 "couldn't add known address ({address}) for {peer:?}, unsupported transport"
779 );
780 }
781 },
782 NetworkServiceCommand::SetReservedPeers { protocol, peers } => {
783 let peers = self.add_addresses(peers.into_iter().map(Into::into));
784
785 match self.peerset_handles.get(&protocol) {
786 Some(handle) => {
787 let _ = handle.tx.unbounded_send(PeersetCommand::SetReservedPeers { peers });
788 }
789 None => log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist"),
790 }
791
792 },
793 NetworkServiceCommand::DisconnectPeer {
794 protocol,
795 peer,
796 } => {
797 let Some(handle) = self.peerset_handles.get(&protocol) else {
798 log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist");
799 continue
800 };
801
802 let _ = handle.tx.unbounded_send(PeersetCommand::DisconnectPeer { peer });
803 }
804 NetworkServiceCommand::SetReservedOnly {
805 protocol,
806 reserved_only,
807 } => {
808 let Some(handle) = self.peerset_handles.get(&protocol) else {
809 log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist");
810 continue
811 };
812
813 let _ = handle.tx.unbounded_send(PeersetCommand::SetReservedOnly { reserved_only });
814 }
815 NetworkServiceCommand::RemoveReservedPeers {
816 protocol,
817 peers,
818 } => {
819 let Some(handle) = self.peerset_handles.get(&protocol) else {
820 log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist");
821 continue
822 };
823
824 let _ = handle.tx.unbounded_send(PeersetCommand::RemoveReservedPeers { peers });
825 }
826 }
827 },
828 event = self.discovery.next() => match event {
829 None => return,
830 Some(DiscoveryEvent::Discovered { addresses }) => {
831 for (peer, addresses) in Litep2pNetworkBackend::parse_addresses(addresses.into_iter()) {
833 if self.litep2p.add_known_address(peer.into(), addresses.clone().into_iter()) > 0 {
834 self.peerstore_handle.add_known_peer(peer);
835 }
836 }
837 }
838 Some(DiscoveryEvent::RoutingTableUpdate { peers }) => {
839 let peers = peers.into_iter().map(Into::into).collect::<Vec<_>>();
840
841 for peer in &peers {
842 self.peerstore_handle.add_known_peer(*peer);
843 }
844
845 if !peers.is_empty() {
846 self.event_streams.send(Event::PeerRoutingTableUpdate(peers));
847 }
848 }
849 Some(DiscoveryEvent::FindNodeSuccess { query_id, target, peers }) => {
850 match self.pending_queries.remove(&query_id) {
851 Some(KadQuery::FindNode(_, started)) => {
852 log::trace!(
853 target: LOG_TARGET,
854 "`FIND_NODE` for {target:?} ({query_id:?}) succeeded",
855 );
856
857 self.event_streams.send(
858 Event::Dht(
859 DhtEvent::ClosestPeersFound(
860 target.into(),
861 peers
862 .into_iter()
863 .map(|(peer, addrs)| (
864 peer.into(),
865 addrs.into_iter().map(Into::into).collect(),
866 ))
867 .collect(),
868 )
869 )
870 );
871
872 if let Some(ref metrics) = self.metrics {
873 metrics
874 .kademlia_query_duration
875 .with_label_values(&["node-find"])
876 .observe(started.elapsed().as_secs_f64());
877 }
878 },
879 query => {
880 log::error!(
881 target: LOG_TARGET,
882 "Missing/invalid pending query for `FIND_NODE`: {query:?}"
883 );
884 debug_assert!(false);
885 }
886 }
887 },
888 Some(DiscoveryEvent::GetRecordPartialResult { query_id, record }) => {
889 if !self.pending_queries.contains_key(&query_id) {
890 log::error!(
891 target: LOG_TARGET,
892 "Missing/invalid pending query for `GET_VALUE` partial result: {query_id:?}"
893 );
894
895 continue
896 }
897
898 let peer_id: sc_network_types::PeerId = record.peer.into();
899 let record = PeerRecord {
900 record: P2PRecord {
901 key: record.record.key.to_vec().into(),
902 value: record.record.value,
903 publisher: record.record.publisher.map(|peer_id| {
904 let peer_id: sc_network_types::PeerId = peer_id.into();
905 peer_id.into()
906 }),
907 expires: record.record.expires,
908 },
909 peer: Some(peer_id.into()),
910 };
911
912 self.event_streams.send(
913 Event::Dht(
914 DhtEvent::ValueFound(
915 record.into()
916 )
917 )
918 );
919 }
920 Some(DiscoveryEvent::GetRecordSuccess { query_id }) => {
921 match self.pending_queries.remove(&query_id) {
922 Some(KadQuery::GetValue(key, started)) => {
923 log::trace!(
924 target: LOG_TARGET,
925 "`GET_VALUE` for {key:?} ({query_id:?}) succeeded",
926 );
927
928 if let Some(ref metrics) = self.metrics {
929 metrics
930 .kademlia_query_duration
931 .with_label_values(&["value-get"])
932 .observe(started.elapsed().as_secs_f64());
933 }
934 },
935 query => {
936 log::error!(
937 target: LOG_TARGET,
938 "Missing/invalid pending query for `GET_VALUE`: {query:?}"
939 );
940 debug_assert!(false);
941 },
942 }
943 }
944 Some(DiscoveryEvent::PutRecordSuccess { query_id }) => {
945 match self.pending_queries.remove(&query_id) {
946 Some(KadQuery::PutValue(key, started)) => {
947 log::trace!(
948 target: LOG_TARGET,
949 "`PUT_VALUE` for {key:?} ({query_id:?}) succeeded",
950 );
951
952 self.event_streams.send(Event::Dht(
953 DhtEvent::ValuePut(key)
954 ));
955
956 if let Some(ref metrics) = self.metrics {
957 metrics
958 .kademlia_query_duration
959 .with_label_values(&["value-put"])
960 .observe(started.elapsed().as_secs_f64());
961 }
962 },
963 query => {
964 log::error!(
965 target: LOG_TARGET,
966 "Missing/invalid pending query for `PUT_VALUE`: {query:?}"
967 );
968 debug_assert!(false);
969 }
970 }
971 }
972 Some(DiscoveryEvent::GetProvidersSuccess { query_id, providers }) => {
973 match self.pending_queries.remove(&query_id) {
974 Some(KadQuery::GetProviders(key, started)) => {
975 log::trace!(
976 target: LOG_TARGET,
977 "`GET_PROVIDERS` for {key:?} ({query_id:?}) succeeded",
978 );
979
980 providers.iter().for_each(|p| {
985 self.litep2p.add_known_address(p.peer, p.addresses.clone().into_iter());
986 });
987
988 self.event_streams.send(Event::Dht(
989 DhtEvent::ProvidersFound(
990 key.clone().into(),
991 providers.into_iter().map(|p| p.peer.into()).collect()
992 )
993 ));
994
995 self.event_streams.send(Event::Dht(
998 DhtEvent::NoMoreProviders(key.into())
999 ));
1000
1001 if let Some(ref metrics) = self.metrics {
1002 metrics
1003 .kademlia_query_duration
1004 .with_label_values(&["providers-get"])
1005 .observe(started.elapsed().as_secs_f64());
1006 }
1007 },
1008 query => {
1009 log::error!(
1010 target: LOG_TARGET,
1011 "Missing/invalid pending query for `GET_PROVIDERS`: {query:?}"
1012 );
1013 debug_assert!(false);
1014 }
1015 }
1016 }
1017 Some(DiscoveryEvent::AddProviderSuccess { query_id, provided_key }) => {
1018 match self.pending_queries.remove(&query_id) {
1019 Some(KadQuery::AddProvider(key, started)) => {
1020 debug_assert_eq!(key, provided_key.into());
1021
1022 log::trace!(
1023 target: LOG_TARGET,
1024 "`ADD_PROVIDER` for {key:?} ({query_id:?}) succeeded",
1025 );
1026
1027 self.event_streams.send(Event::Dht(
1028 DhtEvent::StartedProviding(key.into())
1029 ));
1030
1031 if let Some(ref metrics) = self.metrics {
1032 metrics
1033 .kademlia_query_duration
1034 .with_label_values(&["provider-add"])
1035 .observe(started.elapsed().as_secs_f64());
1036 }
1037 }
1038 Some(_) => {
1039 log::error!(
1040 target: LOG_TARGET,
1041 "Invalid pending query for `ADD_PROVIDER`: {query_id:?}"
1042 );
1043 debug_assert!(false);
1044 }
1045 None => {
1046 log::trace!(
1047 target: LOG_TARGET,
1048 "`ADD_PROVIDER` for key {provided_key:?} ({query_id:?}) succeeded (republishing)",
1049 );
1050 }
1051 }
1052 }
1053 Some(DiscoveryEvent::QueryFailed { query_id }) => {
1054 match self.pending_queries.remove(&query_id) {
1055 Some(KadQuery::FindNode(peer_id, started)) => {
1056 log::debug!(
1057 target: LOG_TARGET,
1058 "`FIND_NODE` ({query_id:?}) failed for target {peer_id:?}",
1059 );
1060
1061 self.event_streams.send(Event::Dht(
1062 DhtEvent::ClosestPeersNotFound(peer_id.into())
1063 ));
1064
1065 if let Some(ref metrics) = self.metrics {
1066 metrics
1067 .kademlia_query_duration
1068 .with_label_values(&["node-find-failed"])
1069 .observe(started.elapsed().as_secs_f64());
1070 }
1071 },
1072 Some(KadQuery::GetValue(key, started)) => {
1073 log::debug!(
1074 target: LOG_TARGET,
1075 "`GET_VALUE` ({query_id:?}) failed for key {key:?}",
1076 );
1077
1078 self.event_streams.send(Event::Dht(
1079 DhtEvent::ValueNotFound(key)
1080 ));
1081
1082 if let Some(ref metrics) = self.metrics {
1083 metrics
1084 .kademlia_query_duration
1085 .with_label_values(&["value-get-failed"])
1086 .observe(started.elapsed().as_secs_f64());
1087 }
1088 },
1089 Some(KadQuery::PutValue(key, started)) => {
1090 log::debug!(
1091 target: LOG_TARGET,
1092 "`PUT_VALUE` ({query_id:?}) failed for key {key:?}",
1093 );
1094
1095 self.event_streams.send(Event::Dht(
1096 DhtEvent::ValuePutFailed(key)
1097 ));
1098
1099 if let Some(ref metrics) = self.metrics {
1100 metrics
1101 .kademlia_query_duration
1102 .with_label_values(&["value-put-failed"])
1103 .observe(started.elapsed().as_secs_f64());
1104 }
1105 },
1106 Some(KadQuery::GetProviders(key, started)) => {
1107 log::debug!(
1108 target: LOG_TARGET,
1109 "`GET_PROVIDERS` ({query_id:?}) failed for key {key:?}"
1110 );
1111
1112 self.event_streams.send(Event::Dht(
1113 DhtEvent::ProvidersNotFound(key)
1114 ));
1115
1116 if let Some(ref metrics) = self.metrics {
1117 metrics
1118 .kademlia_query_duration
1119 .with_label_values(&["providers-get-failed"])
1120 .observe(started.elapsed().as_secs_f64());
1121 }
1122 },
1123 Some(KadQuery::AddProvider(key, started)) => {
1124 log::debug!(
1125 target: LOG_TARGET,
1126 "`ADD_PROVIDER` ({query_id:?}) failed with key {key:?}",
1127 );
1128
1129 self.event_streams.send(Event::Dht(
1130 DhtEvent::StartProvidingFailed(key)
1131 ));
1132
1133 if let Some(ref metrics) = self.metrics {
1134 metrics
1135 .kademlia_query_duration
1136 .with_label_values(&["provider-add-failed"])
1137 .observe(started.elapsed().as_secs_f64());
1138 }
1139 },
1140 None => {
1141 log::debug!(
1142 target: LOG_TARGET,
1143 "non-existent query (likely republishing a provider) failed ({query_id:?})",
1144 );
1145 }
1146 }
1147 }
1148 Some(DiscoveryEvent::Identified { peer, listen_addresses, supported_protocols, .. }) => {
1149 let protocols: HashSet<ProtocolName> =
1150 supported_protocols.iter().cloned().map(Into::into).collect();
1151 if let Some(context) = self.peers.get_mut(&peer) {
1152 context.supported_protocols = Some(protocols.clone());
1153 }
1154 self.event_streams.send(Event::PeerIdentified {
1155 peer: peer.into(),
1156 supported_protocols: protocols,
1157 });
1158 self.discovery.add_self_reported_address(peer, supported_protocols, listen_addresses).await;
1159 }
1160 Some(DiscoveryEvent::ExternalAddressDiscovered { address }) => {
1161 match self.litep2p.public_addresses().add_address(address.clone().into()) {
1162 Ok(inserted) => if inserted {
1163 log::info!(target: LOG_TARGET, "๐ Discovered new external address for our node: {address}");
1164 },
1165 Err(err) => {
1166 log::warn!(
1167 target: LOG_TARGET,
1168 "๐ Failed to add discovered external address {address:?}: {err:?}",
1169 );
1170 },
1171 }
1172 }
1173 Some(DiscoveryEvent::ExternalAddressExpired{ address }) => {
1174 let local_peer_id = self.litep2p.local_peer_id();
1175
1176 let address = if !std::matches!(address.iter().last(), Some(Protocol::P2p(_))) {
1178 address.with(Protocol::P2p((*local_peer_id).into()))
1179 } else {
1180 address
1181 };
1182
1183 if self.litep2p.public_addresses().remove_address(&address) {
1184 log::info!(target: LOG_TARGET, "๐ Expired external address for our node: {address}");
1185 } else {
1186 log::warn!(
1187 target: LOG_TARGET,
1188 "๐ Failed to remove expired external address {address:?}"
1189 );
1190 }
1191 }
1192 Some(DiscoveryEvent::Ping { peer, rtt }) => {
1193 log::trace!(
1194 target: LOG_TARGET,
1195 "ping time with {peer:?}: {rtt:?}",
1196 );
1197 }
1198 Some(DiscoveryEvent::IncomingRecord { record: Record { key, value, publisher, expires }} ) => {
1199 self.event_streams.send(Event::Dht(
1200 DhtEvent::PutRecordRequest(
1201 key.into(),
1202 value,
1203 publisher.map(Into::into),
1204 expires,
1205 )
1206 ));
1207 },
1208
1209 Some(DiscoveryEvent::RandomKademliaStarted) => {
1210 if let Some(metrics) = self.metrics.as_ref() {
1211 metrics.kademlia_random_queries_total.inc();
1212 }
1213 }
1214 },
1215 event = self.litep2p.next_event() => match event {
1216 Some(Litep2pEvent::ConnectionEstablished { peer, endpoint }) => {
1217 let direction = match endpoint {
1218 Endpoint::Dialer { .. } => "out",
1219 Endpoint::Listener { .. } => {
1220 if let Some(metrics) = &self.metrics {
1225 metrics.incoming_connections_total.inc();
1226 }
1227
1228 "in"
1229 },
1230 };
1231 if let Some(metrics) = &self.metrics {
1232 metrics.connections_opened_total.with_label_values(&[direction]).inc();
1233 }
1234
1235 match self.peers.entry(peer) {
1236 Entry::Vacant(entry) => {
1237 entry.insert(ConnectionContext {
1238 endpoints: HashMap::from_iter([(endpoint.connection_id(), endpoint)]),
1239 num_connections: 1usize,
1240 supported_protocols: None,
1241 });
1242 if let Some(metrics) = &self.metrics {
1243 metrics.distinct_peers_connections_opened_total.inc();
1244 }
1245 }
1246 Entry::Occupied(entry) => {
1247 let entry = entry.into_mut();
1248 entry.num_connections += 1;
1249 entry.endpoints.insert(endpoint.connection_id(), endpoint);
1250 }
1251 }
1252 }
1253 Some(Litep2pEvent::ConnectionClosed { peer, connection_id }) => {
1254 let Some(context) = self.peers.get_mut(&peer) else {
1255 log::debug!(target: LOG_TARGET, "unknown peer disconnected: {peer:?} ({connection_id:?})");
1256 continue
1257 };
1258
1259 let direction = match context.endpoints.remove(&connection_id) {
1260 None => {
1261 log::debug!(target: LOG_TARGET, "connection {connection_id:?} doesn't exist for {peer:?} ");
1262 continue
1263 }
1264 Some(endpoint) => {
1265 context.num_connections -= 1;
1266
1267 match endpoint {
1268 Endpoint::Dialer { .. } => "out",
1269 Endpoint::Listener { .. } => "in",
1270 }
1271 }
1272 };
1273
1274 if let Some(metrics) = &self.metrics {
1275 metrics.connections_closed_total.with_label_values(&[direction, "actively-closed"]).inc();
1276 }
1277
1278 if context.num_connections == 0 {
1279 self.peers.remove(&peer);
1280 if let Some(metrics) = &self.metrics {
1281 metrics.distinct_peers_connections_closed_total.inc();
1282 }
1283 }
1284 }
1285 Some(Litep2pEvent::DialFailure { address, error }) => {
1286 log::debug!(
1287 target: LOG_TARGET,
1288 "failed to dial peer at {address:?}: {error:?}",
1289 );
1290
1291 if let Some(metrics) = &self.metrics {
1292 let reason = match error {
1293 DialError::Timeout => "timeout",
1294 DialError::AddressError(_) => "invalid-address",
1295 DialError::DnsError(_) => "cannot-resolve-dns",
1296 DialError::NegotiationError(error) => match error {
1297 NegotiationError::Timeout => "timeout",
1298 NegotiationError::PeerIdMissing => "missing-peer-id",
1299 NegotiationError::StateMismatch => "state-mismatch",
1300 NegotiationError::PeerIdMismatch(_,_) => "peer-id-missmatch",
1301 NegotiationError::MultistreamSelectError(_) => "multistream-select-error",
1302 NegotiationError::SnowError(_) => "noise-error",
1303 NegotiationError::ParseError(_) => "parse-error",
1304 NegotiationError::IoError(_) => "io-error",
1305 NegotiationError::WebSocket(_) => "webscoket-error",
1306 NegotiationError::BadSignature => "bad-signature",
1307 }
1308 };
1309
1310 metrics.pending_connections_errors_total.with_label_values(&[&reason]).inc();
1311 }
1312 }
1313 Some(Litep2pEvent::ListDialFailures { errors }) => {
1314 log::debug!(
1315 target: LOG_TARGET,
1316 "failed to dial peer on multiple addresses {errors:?}",
1317 );
1318
1319 if let Some(metrics) = &self.metrics {
1320 metrics.pending_connections_errors_total.with_label_values(&["transport-errors"]).inc();
1321 }
1322 }
1323 None => {
1324 log::error!(
1325 target: LOG_TARGET,
1326 "Litep2p backend terminated"
1327 );
1328 return
1329 }
1330 },
1331 }
1332 }
1333 }
1334}
1335
1336#[cfg(test)]
1337mod tests {
1338 use super::*;
1339 use crate::{
1340 config::{ed25519, NetworkConfiguration, ProtocolId, Role, Secret},
1341 service::traits::NetworkStateInfo,
1342 };
1343 use sc_network_types::{
1344 multiaddr::Multiaddr as NetworkMultiaddr, multihash::Multihash as NetworkMultihash,
1345 };
1346 use sp_core::H256;
1347 use substrate_test_runtime_client::runtime::Block;
1348
1349 const WEBRTC_LISTEN_ADDRESS: &str = "/ip4/127.0.0.1/udp/0/webrtc-direct";
1351
1352 const WEBRTC_PUBLIC_ADDRESS: &str = "/ip4/203.0.113.9/udp/31234/webrtc-direct";
1355
1356 fn certhash(address: &NetworkMultiaddr) -> Option<NetworkMultihash> {
1358 address.iter().find_map(|protocol| match protocol {
1359 NetworkProtocol::Certhash(hash) => Some(hash),
1360 _ => None,
1361 })
1362 }
1363
1364 fn start_backend(
1366 network_config: &NetworkConfiguration,
1367 ) -> Result<Litep2pNetworkBackend, Error> {
1368 let config = FullNetworkConfiguration::<Block, H256, Litep2pNetworkBackend>::new(
1369 network_config,
1370 None,
1371 );
1372
1373 let (block_announce_config, _notification_service) =
1374 <Litep2pNetworkBackend as NetworkBackend<Block, H256>>::notification_config(
1375 "/block-announces/1".into(),
1376 vec![],
1377 1024,
1378 None,
1379 SetConfig::default(),
1380 NotificationMetrics::new(None),
1381 config.peer_store_handle(),
1382 );
1383
1384 <Litep2pNetworkBackend as NetworkBackend<Block, H256>>::new(Params {
1385 role: Role::Full,
1386 executor: Box::new(|future| {
1387 tokio::spawn(future);
1388 }),
1389 network_config: config,
1390 protocol_id: ProtocolId::from("test"),
1391 genesis_hash: H256::zero(),
1392 fork_id: None,
1393 metrics_registry: None,
1394 block_announce_config,
1395 ipfs_config: None,
1396 notification_metrics: NotificationMetrics::new(None),
1397 })
1398 }
1399
1400 #[tokio::test]
1403 async fn webrtc_addresses_advertised_with_certhash() {
1404 let mut network_config = NetworkConfiguration::new_local();
1405 network_config.listen_addresses = vec![WEBRTC_LISTEN_ADDRESS.parse().unwrap()];
1406 network_config.public_addresses = vec![WEBRTC_PUBLIC_ADDRESS.parse().unwrap()];
1407 network_config.node_key = NodeKeyConfig::Ed25519(Secret::Input(
1409 ed25519::SecretKey::try_from_bytes([7u8; 32]).unwrap(),
1410 ));
1411 network_config.validate_and_complete_addresses().unwrap();
1412
1413 let (keypair, _peer_id) =
1414 Litep2pNetworkBackend::get_keypair(&network_config.node_key).unwrap();
1415 let node_certhash: NetworkMultihash =
1416 webrtc::derive_certificate(keypair.secret()).unwrap().certhash().into();
1417
1418 let backend = start_backend(&network_config).unwrap();
1420 let network_service =
1421 <Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1422
1423 let advertised = network_service.listen_addresses();
1424 assert_eq!(advertised.len(), 1);
1425 assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1426
1427 let advertised = network_service.external_addresses();
1428 assert_eq!(advertised.len(), 1);
1429 assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1430 }
1431
1432 #[tokio::test]
1436 async fn webrtc_certhash_consistent_with_default_node_key() {
1437 let mut network_config = NetworkConfiguration::new_local();
1438 network_config.listen_addresses = vec![WEBRTC_LISTEN_ADDRESS.parse().unwrap()];
1439 network_config.public_addresses = vec![WEBRTC_PUBLIC_ADDRESS.parse().unwrap()];
1440 network_config.validate_and_complete_addresses().unwrap();
1441
1442 let (keypair, _peer_id) =
1444 Litep2pNetworkBackend::get_keypair(&network_config.node_key).unwrap();
1445 let node_certhash: NetworkMultihash =
1446 webrtc::derive_certificate(keypair.secret()).unwrap().certhash().into();
1447
1448 let backend = start_backend(&network_config).unwrap();
1450 let network_service =
1451 <Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1452
1453 let advertised = network_service.listen_addresses();
1454 assert_eq!(advertised.len(), 1);
1455 assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1456
1457 let advertised = network_service.external_addresses();
1458 assert_eq!(advertised.len(), 1);
1459 assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1460 }
1461
1462 #[tokio::test]
1463 async fn webrtc_public_address_completed_at_config_creation_accepted() {
1464 let mut network_config = NetworkConfiguration::new_local();
1465 network_config.listen_addresses = vec![WEBRTC_LISTEN_ADDRESS.parse().unwrap()];
1466 network_config.public_addresses = vec![WEBRTC_PUBLIC_ADDRESS.parse().unwrap()];
1467 network_config.node_key = NodeKeyConfig::Ed25519(Secret::Input(
1468 ed25519::SecretKey::try_from_bytes([7u8; 32]).unwrap(),
1469 ));
1470 network_config.validate_and_complete_addresses().unwrap();
1471
1472 let backend = start_backend(&network_config).unwrap();
1474 let network_service =
1475 <Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1476 let local_peer_id = network_service.local_peer_id();
1477
1478 assert_eq!(
1479 network_service.external_addresses(),
1480 vec![network_config.public_addresses[0]
1481 .clone()
1482 .with(NetworkProtocol::P2p(local_peer_id.into()))],
1483 );
1484 }
1485
1486 #[tokio::test]
1490 async fn webrtc_listen_address_with_certhash_refuses_to_start() {
1491 let listen_address: NetworkMultiaddr = WEBRTC_LISTEN_ADDRESS.parse().unwrap();
1492
1493 let mut network_config = NetworkConfiguration::new_local();
1494 network_config.listen_addresses =
1495 vec![listen_address.with(NetworkProtocol::Certhash(a_certhash()))];
1496
1497 assert!(matches!(start_backend(&network_config), Err(Error::InvalidWebRtcAddress { .. }),));
1498 }
1499
1500 #[tokio::test]
1501 async fn non_webrtc_public_address_untouched() {
1502 let public_address: NetworkMultiaddr = "/ip4/203.0.113.9/tcp/31234".parse().unwrap();
1503
1504 let mut network_config = NetworkConfiguration::new_local();
1505 network_config.listen_addresses = vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()];
1506 network_config.public_addresses = vec![public_address.clone()];
1507
1508 let backend = start_backend(&network_config).unwrap();
1510 let network_service =
1511 <Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1512 let local_peer_id = network_service.local_peer_id();
1513
1514 assert_eq!(
1516 network_service.external_addresses(),
1517 vec![public_address.with(NetworkProtocol::P2p(local_peer_id.into()))],
1518 );
1519 }
1520
1521 fn a_certhash() -> NetworkMultihash {
1523 sc_network_types::multihash::Code::Sha2_256.digest(b"certificate")
1524 }
1525}