referrerpolicy=no-referrer-when-downgrade

sc_network/litep2p/
mod.rs

1// This file is part of Substrate.
2
3// Copyright (C) Parity Technologies (UK) Ltd.
4// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
5
6// This program is free software: you can redistribute it and/or modify
7// it under the terms of the GNU General Public License as published by
8// the Free Software Foundation, either version 3 of the License, or
9// (at your option) any later version.
10
11// This program is distributed in the hope that it will be useful,
12// but WITHOUT ANY WARRANTY; without even the implied warranty of
13// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14// GNU General Public License for more details.
15
16// You should have received a copy of the GNU General Public License
17// along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19//! `NetworkBackend` implementation for `litep2p`.
20
21use 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
102/// Litep2p bandwidth sink.
103struct 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
117/// Litep2p task executor.
118struct Litep2pExecutor {
119	/// Executor.
120	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
133/// Logging target for the file.
134const LOG_TARGET: &str = "sub-libp2p";
135
136/// Peer context.
137struct ConnectionContext {
138	/// Peer endpoints.
139	endpoints: HashMap<ConnectionId, Endpoint>,
140
141	/// Number of active connections.
142	num_connections: usize,
143}
144
145/// Kademlia query we are tracking.
146#[derive(Debug)]
147enum KadQuery {
148	/// `FIND_NODE` query for target and when it was initiated.
149	FindNode(PeerId, Instant),
150	/// `GET_VALUE` query for key and when it was initiated.
151	GetValue(RecordKey, Instant),
152	/// `PUT_VALUE` query for key and when it was initiated.
153	PutValue(RecordKey, Instant),
154	/// `GET_PROVIDERS` query for key and when it was initiated.
155	GetProviders(RecordKey, Instant),
156	/// `ADD_PROVIDER` query for key and when it was initiated.
157	AddProvider(RecordKey, Instant),
158}
159
160/// Networking backend for `litep2p`.
161pub struct Litep2pNetworkBackend {
162	/// Main `litep2p` object.
163	litep2p: Litep2p,
164
165	/// `NetworkService` implementation for `Litep2pNetworkBackend`.
166	network_service: Arc<dyn NetworkService>,
167
168	/// RX channel for receiving commands from `Litep2pNetworkService`.
169	cmd_rx: TracingUnboundedReceiver<NetworkServiceCommand>,
170
171	/// `Peerset` handles to notification protocols.
172	peerset_handles: HashMap<ProtocolName, ProtocolControlHandle>,
173
174	/// Pending Kademlia queries.
175	pending_queries: HashMap<QueryId, KadQuery>,
176
177	/// Discovery.
178	discovery: Discovery,
179
180	/// Number of connected peers.
181	num_connected: Arc<AtomicUsize>,
182
183	/// Connected peers.
184	peers: HashMap<litep2p::PeerId, ConnectionContext>,
185
186	/// Peerstore.
187	peerstore_handle: Arc<dyn PeerStoreProvider>,
188
189	/// Block announce protocol name.
190	block_announce_protocol: ProtocolName,
191
192	/// Sender for DHT events.
193	event_streams: out_events::OutChannels,
194
195	/// Prometheus metrics.
196	metrics: Option<Metrics>,
197}
198
199impl Litep2pNetworkBackend {
200	/// From an iterator of multiaddress(es), parse and group all addresses of peers
201	/// so that litep2p can consume the information easily.
202	fn parse_addresses(
203		addresses: impl Iterator<Item = Multiaddr>,
204	) -> HashMap<PeerId, Vec<Multiaddr>> {
205		addresses
206			.into_iter()
207			.filter_map(|address| match address.iter().next() {
208				Some(
209					Protocol::Dns(_) |
210					Protocol::Dns4(_) |
211					Protocol::Dns6(_) |
212					Protocol::Ip6(_) |
213					Protocol::Ip4(_),
214				) => match address.iter().find(|protocol| std::matches!(protocol, Protocol::P2p(_)))
215				{
216					Some(Protocol::P2p(peer_id)) => Some((peer_id.into(), Some(address))),
217					_ => None,
218				},
219				Some(Protocol::P2p(peer_id)) => Some((peer_id.into(), None)),
220				_ => None,
221			})
222			.fold(HashMap::new(), |mut acc, (peer, maybe_address)| {
223				let entry = acc.entry(peer).or_default();
224				maybe_address.map(|address| entry.push(address));
225
226				acc
227			})
228	}
229
230	/// Add new known addresses to `litep2p` and return the parsed peer IDs.
231	fn add_addresses(&mut self, peers: impl Iterator<Item = Multiaddr>) -> HashSet<PeerId> {
232		Self::parse_addresses(peers.into_iter())
233			.into_iter()
234			.filter_map(|(peer, addresses)| {
235				// `peers` contained multiaddress in the form `/p2p/<peer ID>`
236				if addresses.is_empty() {
237					return Some(peer);
238				}
239
240				if self.litep2p.add_known_address(peer.into(), addresses.clone().into_iter()) == 0 {
241					log::warn!(
242						target: LOG_TARGET,
243						"couldn't add any addresses for {peer:?} and it won't be added as reserved peer",
244					);
245					return None;
246				}
247
248				self.peerstore_handle.add_known_peer(peer);
249				Some(peer)
250			})
251			.collect()
252	}
253}
254
255impl Litep2pNetworkBackend {
256	/// Get `litep2p` keypair from `NodeKeyConfig`.
257	fn get_keypair(node_key: &NodeKeyConfig) -> Result<(Keypair, litep2p::PeerId), Error> {
258		let secret: litep2p::crypto::ed25519::SecretKey =
259			node_key.clone().into_keypair()?.secret().into();
260
261		let local_identity = Keypair::from(secret);
262		let local_public = local_identity.public();
263		let local_peer_id = local_public.to_peer_id();
264
265		Ok((local_identity, local_peer_id))
266	}
267
268	/// Configure transport protocols for `Litep2pNetworkBackend`.
269	fn configure_transport<B: BlockT + 'static, H: ExHashT>(
270		config: &FullNetworkConfiguration<B, H, Self>,
271		keypair: Keypair,
272	) -> Result<ConfigBuilder, Error> {
273		let _ = match config.network_config.transport {
274			TransportConfig::MemoryOnly => panic!("memory transport not supported"),
275			TransportConfig::Normal { .. } => false,
276		};
277		let config_builder = ConfigBuilder::new();
278
279		let listen_addr_len = config.network_config.listen_addresses.len();
280		let mut tcp_addresses = Vec::with_capacity(listen_addr_len);
281		let mut websocket_addresses = Vec::with_capacity(listen_addr_len);
282		let mut webrtc_addresses = Vec::with_capacity(listen_addr_len);
283
284		for addr in &config.network_config.listen_addresses {
285			let mut iter = addr.iter();
286
287			let ip_version = iter.next();
288			let Some(NetworkProtocol::Ip4(_) | NetworkProtocol::Ip6(_)) = ip_version else {
289				log::error!(
290					target: LOG_TARGET,
291					"unknown protocol {ip_version:?}, ignoring {addr:?}",
292				);
293				continue;
294			};
295
296			let transport_layer = iter.next();
297			let protocol_type = iter.next();
298
299			match (&transport_layer, &protocol_type) {
300				// Plain TCP address.
301				(Some(NetworkProtocol::Tcp(_)), Some(NetworkProtocol::P2p(_)) | None) => {
302					tcp_addresses.push(addr.clone());
303				},
304				// Websocket address.
305				(
306					Some(NetworkProtocol::Tcp(_)),
307					Some(NetworkProtocol::Ws(_) | NetworkProtocol::Wss(_)),
308				) => {
309					websocket_addresses.push(addr.clone());
310				},
311				// WebRTC-Direct address.
312				(Some(NetworkProtocol::Udp(_)), Some(NetworkProtocol::WebRTCDirect)) => {
313					// An address carrying anything past `webrtc-direct` is rejected.
314					webrtc::validate_listen_address(addr)?;
315					webrtc_addresses.push(addr.clone());
316				},
317				_ => {
318					log::error!(
319						target: LOG_TARGET,
320						"unknown transport layer {transport_layer:?} and protocol type {protocol_type:?}, ignoring {addr:?}",
321					);
322				},
323			};
324		}
325
326		let mut config_builder = config_builder
327			.with_websocket(WebSocketTransportConfig {
328				listen_addresses: websocket_addresses.into_iter().map(Into::into).collect(),
329				yamux_config: litep2p::yamux::Config::default(),
330				nodelay: true,
331				..Default::default()
332			})
333			.with_tcp(TcpTransportConfig {
334				listen_addresses: tcp_addresses.into_iter().map(Into::into).collect(),
335				yamux_config: litep2p::yamux::Config::default(),
336				nodelay: true,
337				..Default::default()
338			});
339
340		if !webrtc_addresses.is_empty() {
341			let certificate =
342				webrtc::derive_certificate(keypair.secret()).map_err(Error::Litep2p)?;
343			log::info!(target: LOG_TARGET, "WebRTC certhash: {}", certificate.certhash_b64());
344			config_builder = config_builder.with_webrtc(WebRtcTransportConfig {
345				listen_addresses: webrtc_addresses.into_iter().map(Into::into).collect(),
346				certificate: Some(certificate),
347				..Default::default()
348			});
349		}
350
351		Ok(config_builder.with_keypair(keypair))
352	}
353}
354
355#[async_trait::async_trait]
356impl<B: BlockT + 'static, H: ExHashT> NetworkBackend<B, H> for Litep2pNetworkBackend {
357	const SUPPORTS_IPFS: bool = true;
358
359	type NotificationProtocolConfig = NotificationProtocolConfig;
360	type RequestResponseProtocolConfig = RequestResponseConfig;
361	type NetworkService<Block, Hash> = Arc<Litep2pNetworkService>;
362	type PeerStore = Peerstore;
363
364	fn new(mut params: Params<B, H, Self>) -> Result<Self, Error>
365	where
366		Self: Sized,
367	{
368		// Install the ring CryptoProvider for rustls before any TLS connections are made.
369		if let Err(err) = rustls::crypto::ring::default_provider().install_default() {
370			log::warn!(
371				target: LOG_TARGET,
372				"failed to install ring CryptoProvider for rustls, another provider might be installed: {err:?}",
373			);
374		}
375
376		let (keypair, local_peer_id) =
377			Self::get_keypair(&params.network_config.network_config.node_key)?;
378		let (cmd_tx, cmd_rx) = tracing_unbounded("mpsc_network_worker", 100_000);
379
380		params.network_config.network_config.boot_nodes = params
381			.network_config
382			.network_config
383			.boot_nodes
384			.into_iter()
385			.filter(|boot_node| boot_node.peer_id != local_peer_id.into())
386			.collect();
387		params.network_config.network_config.default_peers_set.reserved_nodes = params
388			.network_config
389			.network_config
390			.default_peers_set
391			.reserved_nodes
392			.into_iter()
393			.filter(|reserved_node| {
394				if reserved_node.peer_id == local_peer_id.into() {
395					log::warn!(
396						target: LOG_TARGET,
397						"Local peer ID used in reserved node, ignoring: {reserved_node}",
398					);
399					false
400				} else {
401					true
402				}
403			})
404			.collect();
405
406		if let Some(path) = &params.network_config.network_config.net_config_path {
407			fs::create_dir_all(path)?;
408		}
409
410		log::info!(target: LOG_TARGET, "Local node identity is: {local_peer_id}");
411		log::info!(target: LOG_TARGET, "Running litep2p network backend");
412
413		params.network_config.sanity_check_addresses()?;
414		params.network_config.sanity_check_bootnodes()?;
415
416		let mut config_builder =
417			Self::configure_transport(&params.network_config, keypair.clone())?;
418		let known_addresses = params.network_config.known_addresses();
419		let peer_store_handle = params.network_config.peer_store_handle();
420		let executor = Arc::new(Litep2pExecutor { executor: params.executor });
421
422		let FullNetworkConfiguration {
423			notification_protocols,
424			request_response_protocols,
425			network_config,
426			..
427		} = params.network_config;
428
429		// initialize notification protocols
430		//
431		// pass the protocol configuration to `Litep2pConfigBuilder` and save the TX channel
432		// to the protocol's `Peerset` together with the protocol name to allow other subsystems
433		// of Polkadot SDK to control connectivity of the notification protocol
434		let block_announce_protocol = params.block_announce_config.protocol_name().clone();
435		let mut notif_protocols = HashMap::from_iter([(
436			params.block_announce_config.protocol_name().clone(),
437			params.block_announce_config.handle,
438		)]);
439
440		// handshake for all but the syncing protocol is set to node role
441		config_builder = notification_protocols
442			.into_iter()
443			.fold(config_builder, |config_builder, mut config| {
444				config.config.set_handshake(Roles::from(&params.role).encode());
445				notif_protocols.insert(config.protocol_name, config.handle);
446
447				config_builder.with_notification_protocol(config.config)
448			})
449			.with_notification_protocol(params.block_announce_config.config);
450
451		// initialize request-response protocols
452		let metrics = match &params.metrics_registry {
453			Some(registry) => Some(register_without_sources(registry)?),
454			None => None,
455		};
456
457		// create channels that are used to send request before initializing protocols so the
458		// senders can be passed onto all request-response protocols
459		//
460		// all protocols must have each others' senders so they can send the fallback request in
461		// case the main protocol is not supported by the remote peer and user specified a fallback
462		let (mut request_response_receivers, request_response_senders): (
463			HashMap<_, _>,
464			HashMap<_, _>,
465		) = request_response_protocols
466			.iter()
467			.map(|config| {
468				let (tx, rx) = tracing_unbounded("outbound-requests", 10_000);
469				((config.protocol_name.clone(), rx), (config.protocol_name.clone(), tx))
470			})
471			.unzip();
472
473		config_builder = request_response_protocols.into_iter().fold(
474			config_builder,
475			|config_builder, config| {
476				let (protocol_config, handle) = RequestResponseConfigBuilder::new(
477					Litep2pProtocolName::from(config.protocol_name.clone()),
478				)
479				.with_max_size(cmp::max(config.max_request_size, config.max_response_size) as usize)
480				.with_fallback_names(config.fallback_names.into_iter().map(From::from).collect())
481				.with_timeout(config.request_timeout)
482				.build();
483
484				let protocol = RequestResponseProtocol::new(
485					config.protocol_name.clone(),
486					handle,
487					Arc::clone(&peer_store_handle),
488					config.inbound_queue,
489					request_response_receivers
490						.remove(&config.protocol_name)
491						.expect("receiver exists as it was just added and there are no duplicate protocols; qed"),
492					request_response_senders.clone(),
493					metrics.clone(),
494				);
495
496				executor.run(Box::pin(async move {
497					protocol.run().await;
498				}));
499
500				config_builder.with_request_response_protocol(protocol_config)
501			},
502		);
503
504		// collect known addresses
505		let known_addresses: HashMap<litep2p::PeerId, Vec<Multiaddr>> =
506			known_addresses.into_iter().fold(HashMap::new(), |mut acc, (peer, address)| {
507				let address = match address.iter().last() {
508					Some(
509						NetworkProtocol::Ws(_) | NetworkProtocol::Wss(_) | NetworkProtocol::Tcp(_),
510					) => address.with(NetworkProtocol::P2p(peer.into())),
511					Some(NetworkProtocol::WebRTCDirect | NetworkProtocol::Certhash(_)) => {
512						address.with(NetworkProtocol::P2p(peer.into()))
513					},
514					Some(NetworkProtocol::P2p(_)) => address,
515					_ => return acc,
516				};
517
518				acc.entry(peer.into()).or_default().push(address.into());
519				peer_store_handle.add_known_peer(peer);
520
521				acc
522			});
523
524		// enable ipfs ping, identify and kademlia, and potentially mdns if user enabled it
525		let listen_addresses = Arc::new(Default::default());
526		let (discovery, ping_config, identify_config, kademlia_config, maybe_mdns_config) =
527			Discovery::new(
528				local_peer_id,
529				&network_config,
530				params.genesis_hash,
531				params.fork_id.as_deref(),
532				&params.protocol_id,
533				known_addresses.clone(),
534				Arc::clone(&listen_addresses),
535				Arc::clone(&peer_store_handle),
536			);
537
538		if let Some(ipfs) = params.ipfs_config {
539			config_builder = config_builder.with_libp2p_bitswap(ipfs.litep2p_bitswap_config);
540
541			if !ipfs.bootnodes.is_empty() {
542				let (ipfs_dht, kad_config) = IpfsDht::new(ipfs.bootnodes, ipfs.block_provider);
543				config_builder = config_builder.with_libp2p_kademlia(kad_config);
544				executor.run(Box::pin(ipfs_dht.run()));
545			} else {
546				log::warn!(
547					target: LOG_TARGET,
548					"Not starting IPFS DHT publisher because no IPFS bootnodes are configured. \
549					 Only direct Bitswap requests will be handled.",
550				);
551			}
552		}
553
554		config_builder = config_builder
555			.with_known_addresses(known_addresses.clone().into_iter())
556			.with_libp2p_ping(ping_config)
557			.with_libp2p_identify(identify_config)
558			.with_libp2p_kademlia(kademlia_config)
559			.with_connection_limits(ConnectionLimitsConfig::default().max_incoming_connections(
560				Some(crate::MAX_CONNECTIONS_ESTABLISHED_INCOMING as usize),
561			))
562			.with_keep_alive_timeout(network_config.idle_connection_timeout)
563			// Use system DNS resolver to enable intranet domain resolution and administrator
564			// control over DNS lookup.
565			.with_system_resolver()
566			.with_executor(executor);
567
568		if let Some(config) = maybe_mdns_config {
569			config_builder = config_builder.with_mdns(config);
570		}
571
572		let litep2p =
573			Litep2p::new(config_builder.build()).map_err(|error| Error::Litep2p(error))?;
574
575		litep2p.listen_addresses().for_each(|address| {
576			log::debug!(target: LOG_TARGET, "listening on: {address}");
577
578			listen_addresses.write().insert(address.clone());
579		});
580
581		let public_addresses = litep2p.public_addresses();
582		for address in network_config.public_addresses.iter() {
583			if let Err(err) = public_addresses.add_address(address.clone().into()) {
584				log::warn!(
585					target: LOG_TARGET,
586					"failed to add public address {address:?}: {err:?}",
587				);
588			}
589		}
590
591		let network_service = Arc::new(Litep2pNetworkService::new(
592			local_peer_id,
593			keypair.clone(),
594			cmd_tx,
595			Arc::clone(&peer_store_handle),
596			notif_protocols.clone(),
597			block_announce_protocol.clone(),
598			request_response_senders,
599			Arc::clone(&listen_addresses),
600			public_addresses,
601		));
602
603		// register rest of the metrics now that `Litep2p` has been created
604		let num_connected = Arc::new(Default::default());
605		let bandwidth: Arc<dyn BandwidthSink> =
606			Arc::new(Litep2pBandwidthSink { sink: litep2p.bandwidth_sink() });
607
608		if let Some(registry) = &params.metrics_registry {
609			MetricSources::register(registry, bandwidth, Arc::clone(&num_connected))?;
610		}
611
612		Ok(Self {
613			network_service,
614			cmd_rx,
615			metrics,
616			peerset_handles: notif_protocols,
617			num_connected,
618			discovery,
619			pending_queries: HashMap::new(),
620			peerstore_handle: peer_store_handle,
621			block_announce_protocol,
622			event_streams: out_events::OutChannels::new(None)?,
623			peers: HashMap::new(),
624			litep2p,
625		})
626	}
627
628	fn network_service(&self) -> Arc<dyn NetworkService> {
629		Arc::clone(&self.network_service)
630	}
631
632	fn peer_store(
633		bootnodes: Vec<sc_network_types::PeerId>,
634		metrics_registry: Option<Registry>,
635	) -> Self::PeerStore {
636		Peerstore::new(bootnodes, metrics_registry)
637	}
638
639	fn register_notification_metrics(registry: Option<&Registry>) -> NotificationMetrics {
640		NotificationMetrics::new(registry)
641	}
642
643	/// Create notification protocol configuration for `protocol`.
644	fn notification_config(
645		protocol_name: ProtocolName,
646		fallback_names: Vec<ProtocolName>,
647		max_notification_size: u64,
648		handshake: Option<NotificationHandshake>,
649		set_config: SetConfig,
650		metrics: NotificationMetrics,
651		peerstore_handle: Arc<dyn PeerStoreProvider>,
652	) -> (Self::NotificationProtocolConfig, Box<dyn NotificationService>) {
653		Self::NotificationProtocolConfig::new(
654			protocol_name,
655			fallback_names,
656			max_notification_size as usize,
657			handshake,
658			set_config,
659			metrics,
660			peerstore_handle,
661		)
662	}
663
664	/// Create request-response protocol configuration.
665	fn request_response_config(
666		protocol_name: ProtocolName,
667		fallback_names: Vec<ProtocolName>,
668		max_request_size: u64,
669		max_response_size: u64,
670		request_timeout: Duration,
671		inbound_queue: Option<async_channel::Sender<IncomingRequest>>,
672	) -> Self::RequestResponseProtocolConfig {
673		Self::RequestResponseProtocolConfig::new(
674			protocol_name,
675			fallback_names,
676			max_request_size,
677			max_response_size,
678			request_timeout,
679			inbound_queue,
680		)
681	}
682
683	/// Start [`Litep2pNetworkBackend`] event loop.
684	async fn run(mut self) {
685		log::debug!(target: LOG_TARGET, "starting litep2p network backend");
686
687		loop {
688			let num_connected_peers = self
689				.peerset_handles
690				.get(&self.block_announce_protocol)
691				.map_or(0usize, |handle| handle.connected_peers.load(Ordering::Relaxed));
692			self.num_connected.store(num_connected_peers, Ordering::Relaxed);
693
694			tokio::select! {
695				command = self.cmd_rx.next() => match command {
696					None => return,
697					Some(command) => match command {
698						NetworkServiceCommand::FindClosestPeers { target } => {
699							let query_id = self.discovery.find_node(target.into()).await;
700							self.pending_queries.insert(query_id, KadQuery::FindNode(target, Instant::now()));
701						}
702						NetworkServiceCommand::GetValue{ key } => {
703							let query_id = self.discovery.get_value(key.clone()).await;
704							self.pending_queries.insert(query_id, KadQuery::GetValue(key, Instant::now()));
705						}
706						NetworkServiceCommand::PutValue { key, value } => {
707							let query_id = self.discovery.put_value(key.clone(), value).await;
708							self.pending_queries.insert(query_id, KadQuery::PutValue(key, Instant::now()));
709						}
710						NetworkServiceCommand::PutValueTo { record, peers, update_local_storage} => {
711							let kademlia_key = record.key.clone();
712							let query_id = self.discovery.put_value_to_peers(record.into(), peers, update_local_storage).await;
713							self.pending_queries.insert(query_id, KadQuery::PutValue(kademlia_key, Instant::now()));
714						}
715						NetworkServiceCommand::StoreRecord { key, value, publisher, expires } => {
716							self.discovery.store_record(key, value, publisher.map(Into::into), expires).await;
717						}
718						NetworkServiceCommand::StartProviding { key } => {
719							let query_id = self.discovery.start_providing(key.clone()).await;
720							self.pending_queries.insert(query_id, KadQuery::AddProvider(key, Instant::now()));
721						}
722						NetworkServiceCommand::StopProviding { key } => {
723							self.discovery.stop_providing(key).await;
724						}
725						NetworkServiceCommand::GetProviders { key } => {
726							let query_id = self.discovery.get_providers(key.clone()).await;
727							self.pending_queries.insert(query_id, KadQuery::GetProviders(key, Instant::now()));
728						}
729						NetworkServiceCommand::EventStream { tx } => {
730							self.event_streams.push(tx);
731						}
732						NetworkServiceCommand::Status { tx } => {
733							let _ = tx.send(NetworkStatus {
734								num_connected_peers: self
735									.peerset_handles
736									.get(&self.block_announce_protocol)
737									.map_or(0usize, |handle| handle.connected_peers.load(Ordering::Relaxed)),
738								total_bytes_inbound: self.litep2p.bandwidth_sink().inbound() as u64,
739								total_bytes_outbound: self.litep2p.bandwidth_sink().outbound() as u64,
740							});
741						}
742						NetworkServiceCommand::AddPeersToReservedSet {
743							protocol,
744							peers,
745						} => {
746							let peers = self.add_addresses(peers.into_iter().map(Into::into));
747
748							match self.peerset_handles.get(&protocol) {
749								Some(handle) => {
750									let _ = handle.tx.unbounded_send(PeersetCommand::AddReservedPeers { peers });
751								}
752								None => log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist"),
753							};
754						}
755						NetworkServiceCommand::AddKnownAddress { peer, address } => {
756							let mut address: Multiaddr = address.into();
757
758							if !address.iter().any(|protocol| std::matches!(protocol, Protocol::P2p(_))) {
759								address.push(Protocol::P2p(litep2p::PeerId::from(peer).into()));
760							}
761
762							if self.litep2p.add_known_address(peer.into(), iter::once(address.clone())) > 0 {
763								// libp2p backend generates `DiscoveryOut::Discovered(peer_id)`
764								// event when a new address is added for a peer, which leads to the
765								// peer being added to peerstore. Do the same directly here.
766								self.peerstore_handle.add_known_peer(peer);
767							} else {
768								log::debug!(
769									target: LOG_TARGET,
770									"couldn't add known address ({address}) for {peer:?}, unsupported transport"
771								);
772							}
773						},
774						NetworkServiceCommand::SetReservedPeers { protocol, peers } => {
775							let peers = self.add_addresses(peers.into_iter().map(Into::into));
776
777							match self.peerset_handles.get(&protocol) {
778								Some(handle) => {
779									let _ = handle.tx.unbounded_send(PeersetCommand::SetReservedPeers { peers });
780								}
781								None => log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist"),
782							}
783
784						},
785						NetworkServiceCommand::DisconnectPeer {
786							protocol,
787							peer,
788						} => {
789							let Some(handle) = self.peerset_handles.get(&protocol) else {
790								log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist");
791								continue
792							};
793
794							let _ = handle.tx.unbounded_send(PeersetCommand::DisconnectPeer { peer });
795						}
796						NetworkServiceCommand::SetReservedOnly {
797							protocol,
798							reserved_only,
799						} => {
800							let Some(handle) = self.peerset_handles.get(&protocol) else {
801								log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist");
802								continue
803							};
804
805							let _ = handle.tx.unbounded_send(PeersetCommand::SetReservedOnly { reserved_only });
806						}
807						NetworkServiceCommand::RemoveReservedPeers {
808							protocol,
809							peers,
810						} => {
811							let Some(handle) = self.peerset_handles.get(&protocol) else {
812								log::warn!(target: LOG_TARGET, "protocol {protocol} doens't exist");
813								continue
814							};
815
816							let _ = handle.tx.unbounded_send(PeersetCommand::RemoveReservedPeers { peers });
817						}
818					}
819				},
820				event = self.discovery.next() => match event {
821					None => return,
822					Some(DiscoveryEvent::Discovered { addresses }) => {
823						// if at least one address was added for the peer, report the peer to `Peerstore`
824						for (peer, addresses) in Litep2pNetworkBackend::parse_addresses(addresses.into_iter()) {
825							if self.litep2p.add_known_address(peer.into(), addresses.clone().into_iter()) > 0 {
826								self.peerstore_handle.add_known_peer(peer);
827							}
828						}
829					}
830					Some(DiscoveryEvent::RoutingTableUpdate { peers }) => {
831						for peer in peers {
832							self.peerstore_handle.add_known_peer(peer.into());
833						}
834					}
835					Some(DiscoveryEvent::FindNodeSuccess { query_id, target, peers }) => {
836						match self.pending_queries.remove(&query_id) {
837							Some(KadQuery::FindNode(_, started)) => {
838								log::trace!(
839									target: LOG_TARGET,
840									"`FIND_NODE` for {target:?} ({query_id:?}) succeeded",
841								);
842
843								self.event_streams.send(
844									Event::Dht(
845										DhtEvent::ClosestPeersFound(
846											target.into(),
847											peers
848												.into_iter()
849												.map(|(peer, addrs)| (
850													peer.into(),
851													addrs.into_iter().map(Into::into).collect(),
852												))
853												.collect(),
854										)
855									)
856								);
857
858								if let Some(ref metrics) = self.metrics {
859									metrics
860										.kademlia_query_duration
861										.with_label_values(&["node-find"])
862										.observe(started.elapsed().as_secs_f64());
863								}
864							},
865							query => {
866								log::error!(
867									target: LOG_TARGET,
868									"Missing/invalid pending query for `FIND_NODE`: {query:?}"
869								);
870								debug_assert!(false);
871							}
872						}
873					},
874					Some(DiscoveryEvent::GetRecordPartialResult { query_id, record }) => {
875						if !self.pending_queries.contains_key(&query_id) {
876							log::error!(
877								target: LOG_TARGET,
878								"Missing/invalid pending query for `GET_VALUE` partial result: {query_id:?}"
879							);
880
881							continue
882						}
883
884						let peer_id: sc_network_types::PeerId = record.peer.into();
885						let record = PeerRecord {
886							record: P2PRecord {
887								key: record.record.key.to_vec().into(),
888								value: record.record.value,
889								publisher: record.record.publisher.map(|peer_id| {
890									let peer_id: sc_network_types::PeerId = peer_id.into();
891									peer_id.into()
892								}),
893								expires: record.record.expires,
894							},
895							peer: Some(peer_id.into()),
896						};
897
898						self.event_streams.send(
899							Event::Dht(
900								DhtEvent::ValueFound(
901									record.into()
902								)
903							)
904						);
905					}
906					Some(DiscoveryEvent::GetRecordSuccess { query_id }) => {
907						match self.pending_queries.remove(&query_id) {
908							Some(KadQuery::GetValue(key, started)) => {
909								log::trace!(
910									target: LOG_TARGET,
911									"`GET_VALUE` for {key:?} ({query_id:?}) succeeded",
912								);
913
914								if let Some(ref metrics) = self.metrics {
915									metrics
916										.kademlia_query_duration
917										.with_label_values(&["value-get"])
918										.observe(started.elapsed().as_secs_f64());
919								}
920							},
921							query => {
922								log::error!(
923									target: LOG_TARGET,
924									"Missing/invalid pending query for `GET_VALUE`: {query:?}"
925								);
926								debug_assert!(false);
927							},
928						}
929					}
930					Some(DiscoveryEvent::PutRecordSuccess { query_id }) => {
931						match self.pending_queries.remove(&query_id) {
932							Some(KadQuery::PutValue(key, started)) => {
933								log::trace!(
934									target: LOG_TARGET,
935									"`PUT_VALUE` for {key:?} ({query_id:?}) succeeded",
936								);
937
938								self.event_streams.send(Event::Dht(
939									DhtEvent::ValuePut(key)
940								));
941
942								if let Some(ref metrics) = self.metrics {
943									metrics
944										.kademlia_query_duration
945										.with_label_values(&["value-put"])
946										.observe(started.elapsed().as_secs_f64());
947								}
948							},
949							query => {
950								log::error!(
951									target: LOG_TARGET,
952									"Missing/invalid pending query for `PUT_VALUE`: {query:?}"
953								);
954								debug_assert!(false);
955							}
956						}
957					}
958					Some(DiscoveryEvent::GetProvidersSuccess { query_id, providers }) => {
959						match self.pending_queries.remove(&query_id) {
960							Some(KadQuery::GetProviders(key, started)) => {
961								log::trace!(
962									target: LOG_TARGET,
963									"`GET_PROVIDERS` for {key:?} ({query_id:?}) succeeded",
964								);
965
966								// We likely requested providers to connect to them,
967								// so let's add their addresses to litep2p's transport manager.
968								// Consider also looking the addresses of providers up with `FIND_NODE`
969								// query, as it can yield more up to date addresses.
970								providers.iter().for_each(|p| {
971									self.litep2p.add_known_address(p.peer, p.addresses.clone().into_iter());
972								});
973
974								self.event_streams.send(Event::Dht(
975									DhtEvent::ProvidersFound(
976										key.clone().into(),
977										providers.into_iter().map(|p| p.peer.into()).collect()
978									)
979								));
980
981								// litep2p returns all providers in a single event, so we let
982								// subscribers know no more providers will be yielded.
983								self.event_streams.send(Event::Dht(
984									DhtEvent::NoMoreProviders(key.into())
985								));
986
987								if let Some(ref metrics) = self.metrics {
988									metrics
989										.kademlia_query_duration
990										.with_label_values(&["providers-get"])
991										.observe(started.elapsed().as_secs_f64());
992								}
993							},
994							query => {
995								log::error!(
996									target: LOG_TARGET,
997									"Missing/invalid pending query for `GET_PROVIDERS`: {query:?}"
998								);
999								debug_assert!(false);
1000							}
1001						}
1002					}
1003					Some(DiscoveryEvent::AddProviderSuccess { query_id, provided_key }) => {
1004						match self.pending_queries.remove(&query_id) {
1005							Some(KadQuery::AddProvider(key, started)) => {
1006								debug_assert_eq!(key, provided_key.into());
1007
1008								log::trace!(
1009									target: LOG_TARGET,
1010									"`ADD_PROVIDER` for {key:?} ({query_id:?}) succeeded",
1011								);
1012
1013								self.event_streams.send(Event::Dht(
1014									DhtEvent::StartedProviding(key.into())
1015								));
1016
1017								if let Some(ref metrics) = self.metrics {
1018									metrics
1019										.kademlia_query_duration
1020										.with_label_values(&["provider-add"])
1021										.observe(started.elapsed().as_secs_f64());
1022								}
1023							}
1024							Some(_) => {
1025								log::error!(
1026									target: LOG_TARGET,
1027									"Invalid pending query for `ADD_PROVIDER`: {query_id:?}"
1028								);
1029								debug_assert!(false);
1030							}
1031							None => {
1032								log::trace!(
1033									target: LOG_TARGET,
1034									"`ADD_PROVIDER` for key {provided_key:?} ({query_id:?}) succeeded (republishing)",
1035								);
1036							}
1037						}
1038					}
1039					Some(DiscoveryEvent::QueryFailed { query_id }) => {
1040						match self.pending_queries.remove(&query_id) {
1041							Some(KadQuery::FindNode(peer_id, started)) => {
1042								log::debug!(
1043									target: LOG_TARGET,
1044									"`FIND_NODE` ({query_id:?}) failed for target {peer_id:?}",
1045								);
1046
1047								self.event_streams.send(Event::Dht(
1048									DhtEvent::ClosestPeersNotFound(peer_id.into())
1049								));
1050
1051								if let Some(ref metrics) = self.metrics {
1052									metrics
1053										.kademlia_query_duration
1054										.with_label_values(&["node-find-failed"])
1055										.observe(started.elapsed().as_secs_f64());
1056								}
1057							},
1058							Some(KadQuery::GetValue(key, started)) => {
1059								log::debug!(
1060									target: LOG_TARGET,
1061									"`GET_VALUE` ({query_id:?}) failed for key {key:?}",
1062								);
1063
1064								self.event_streams.send(Event::Dht(
1065									DhtEvent::ValueNotFound(key)
1066								));
1067
1068								if let Some(ref metrics) = self.metrics {
1069									metrics
1070										.kademlia_query_duration
1071										.with_label_values(&["value-get-failed"])
1072										.observe(started.elapsed().as_secs_f64());
1073								}
1074							},
1075							Some(KadQuery::PutValue(key, started)) => {
1076								log::debug!(
1077									target: LOG_TARGET,
1078									"`PUT_VALUE` ({query_id:?}) failed for key {key:?}",
1079								);
1080
1081								self.event_streams.send(Event::Dht(
1082									DhtEvent::ValuePutFailed(key)
1083								));
1084
1085								if let Some(ref metrics) = self.metrics {
1086									metrics
1087										.kademlia_query_duration
1088										.with_label_values(&["value-put-failed"])
1089										.observe(started.elapsed().as_secs_f64());
1090								}
1091							},
1092							Some(KadQuery::GetProviders(key, started)) => {
1093								log::debug!(
1094									target: LOG_TARGET,
1095									"`GET_PROVIDERS` ({query_id:?}) failed for key {key:?}"
1096								);
1097
1098								self.event_streams.send(Event::Dht(
1099									DhtEvent::ProvidersNotFound(key)
1100								));
1101
1102								if let Some(ref metrics) = self.metrics {
1103									metrics
1104										.kademlia_query_duration
1105										.with_label_values(&["providers-get-failed"])
1106										.observe(started.elapsed().as_secs_f64());
1107								}
1108							},
1109							Some(KadQuery::AddProvider(key, started)) => {
1110								log::debug!(
1111									target: LOG_TARGET,
1112									"`ADD_PROVIDER` ({query_id:?}) failed with key {key:?}",
1113								);
1114
1115								self.event_streams.send(Event::Dht(
1116									DhtEvent::StartProvidingFailed(key)
1117								));
1118
1119								if let Some(ref metrics) = self.metrics {
1120									metrics
1121										.kademlia_query_duration
1122										.with_label_values(&["provider-add-failed"])
1123										.observe(started.elapsed().as_secs_f64());
1124								}
1125							},
1126							None => {
1127								log::debug!(
1128									target: LOG_TARGET,
1129									"non-existent query (likely republishing a provider) failed ({query_id:?})",
1130								);
1131							}
1132						}
1133					}
1134					Some(DiscoveryEvent::Identified { peer, listen_addresses, supported_protocols, .. }) => {
1135						self.discovery.add_self_reported_address(peer, supported_protocols, listen_addresses).await;
1136					}
1137					Some(DiscoveryEvent::ExternalAddressDiscovered { address }) => {
1138						match self.litep2p.public_addresses().add_address(address.clone().into()) {
1139							Ok(inserted) => if inserted {
1140								log::info!(target: LOG_TARGET, "๐Ÿ” Discovered new external address for our node: {address}");
1141							},
1142							Err(err) => {
1143								log::warn!(
1144									target: LOG_TARGET,
1145									"๐Ÿ” Failed to add discovered external address {address:?}: {err:?}",
1146								);
1147							},
1148						}
1149					}
1150					Some(DiscoveryEvent::ExternalAddressExpired{ address }) => {
1151						let local_peer_id = self.litep2p.local_peer_id();
1152
1153						// Litep2p requires the peer ID to be present in the address.
1154						let address = if !std::matches!(address.iter().last(), Some(Protocol::P2p(_))) {
1155							address.with(Protocol::P2p((*local_peer_id).into()))
1156						} else {
1157							address
1158						};
1159
1160						if self.litep2p.public_addresses().remove_address(&address) {
1161							log::info!(target: LOG_TARGET, "๐Ÿ” Expired external address for our node: {address}");
1162						} else {
1163							log::warn!(
1164								target: LOG_TARGET,
1165								"๐Ÿ” Failed to remove expired external address {address:?}"
1166							);
1167						}
1168					}
1169					Some(DiscoveryEvent::Ping { peer, rtt }) => {
1170						log::trace!(
1171							target: LOG_TARGET,
1172							"ping time with {peer:?}: {rtt:?}",
1173						);
1174					}
1175					Some(DiscoveryEvent::IncomingRecord { record: Record { key, value, publisher, expires }} ) => {
1176						self.event_streams.send(Event::Dht(
1177							DhtEvent::PutRecordRequest(
1178								key.into(),
1179								value,
1180								publisher.map(Into::into),
1181								expires,
1182							)
1183						));
1184					},
1185
1186					Some(DiscoveryEvent::RandomKademliaStarted) => {
1187						if let Some(metrics) = self.metrics.as_ref() {
1188							metrics.kademlia_random_queries_total.inc();
1189						}
1190					}
1191				},
1192				event = self.litep2p.next_event() => match event {
1193					Some(Litep2pEvent::ConnectionEstablished { peer, endpoint }) => {
1194						let Some(metrics) = &self.metrics else {
1195							continue;
1196						};
1197
1198						let direction = match endpoint {
1199							Endpoint::Dialer { .. } => "out",
1200							Endpoint::Listener { .. } => {
1201								// Increment incoming connections counter.
1202								//
1203								// Note: For litep2p these are represented by established negotiated connections,
1204								// while for libp2p (legacy) these represent not-yet-negotiated connections.
1205								metrics.incoming_connections_total.inc();
1206
1207								"in"
1208							},
1209						};
1210						metrics.connections_opened_total.with_label_values(&[direction]).inc();
1211
1212						match self.peers.entry(peer) {
1213							Entry::Vacant(entry) => {
1214								entry.insert(ConnectionContext {
1215									endpoints: HashMap::from_iter([(endpoint.connection_id(), endpoint)]),
1216									num_connections: 1usize,
1217								});
1218								metrics.distinct_peers_connections_opened_total.inc();
1219							}
1220							Entry::Occupied(entry) => {
1221								let entry = entry.into_mut();
1222								entry.num_connections += 1;
1223								entry.endpoints.insert(endpoint.connection_id(), endpoint);
1224							}
1225						}
1226					}
1227					Some(Litep2pEvent::ConnectionClosed { peer, connection_id }) => {
1228						let Some(metrics) = &self.metrics else {
1229							continue;
1230						};
1231
1232						let Some(context) = self.peers.get_mut(&peer) else {
1233							log::debug!(target: LOG_TARGET, "unknown peer disconnected: {peer:?} ({connection_id:?})");
1234							continue
1235						};
1236
1237						let direction = match context.endpoints.remove(&connection_id) {
1238							None => {
1239								log::debug!(target: LOG_TARGET, "connection {connection_id:?} doesn't exist for {peer:?} ");
1240								continue
1241							}
1242							Some(endpoint) => {
1243								context.num_connections -= 1;
1244
1245								match endpoint {
1246									Endpoint::Dialer { .. } => "out",
1247									Endpoint::Listener { .. } => "in",
1248								}
1249							}
1250						};
1251
1252						metrics.connections_closed_total.with_label_values(&[direction, "actively-closed"]).inc();
1253
1254						if context.num_connections == 0 {
1255							self.peers.remove(&peer);
1256							metrics.distinct_peers_connections_closed_total.inc();
1257						}
1258					}
1259					Some(Litep2pEvent::DialFailure { address, error }) => {
1260						log::debug!(
1261							target: LOG_TARGET,
1262							"failed to dial peer at {address:?}: {error:?}",
1263						);
1264
1265						if let Some(metrics) = &self.metrics {
1266							let reason = match error {
1267								DialError::Timeout => "timeout",
1268								DialError::AddressError(_) => "invalid-address",
1269								DialError::DnsError(_) => "cannot-resolve-dns",
1270								DialError::NegotiationError(error) => match error {
1271									NegotiationError::Timeout => "timeout",
1272									NegotiationError::PeerIdMissing => "missing-peer-id",
1273									NegotiationError::StateMismatch => "state-mismatch",
1274									NegotiationError::PeerIdMismatch(_,_) => "peer-id-missmatch",
1275									NegotiationError::MultistreamSelectError(_) => "multistream-select-error",
1276									NegotiationError::SnowError(_) => "noise-error",
1277									NegotiationError::ParseError(_) => "parse-error",
1278									NegotiationError::IoError(_) => "io-error",
1279									NegotiationError::WebSocket(_) => "webscoket-error",
1280									NegotiationError::BadSignature => "bad-signature",
1281								}
1282							};
1283
1284							metrics.pending_connections_errors_total.with_label_values(&[&reason]).inc();
1285						}
1286					}
1287					Some(Litep2pEvent::ListDialFailures { errors }) => {
1288						log::debug!(
1289							target: LOG_TARGET,
1290							"failed to dial peer on multiple addresses {errors:?}",
1291						);
1292
1293						if let Some(metrics) = &self.metrics {
1294							metrics.pending_connections_errors_total.with_label_values(&["transport-errors"]).inc();
1295						}
1296					}
1297					None => {
1298						log::error!(
1299								target: LOG_TARGET,
1300								"Litep2p backend terminated"
1301						);
1302						return
1303					}
1304				},
1305			}
1306		}
1307	}
1308}
1309
1310#[cfg(test)]
1311mod tests {
1312	use super::*;
1313	use crate::{
1314		config::{ed25519, NetworkConfiguration, ProtocolId, Role, Secret},
1315		service::traits::NetworkStateInfo,
1316	};
1317	use sc_network_types::{
1318		multiaddr::Multiaddr as NetworkMultiaddr, multihash::Multihash as NetworkMultihash,
1319	};
1320	use sp_core::H256;
1321	use substrate_test_runtime_client::runtime::Block;
1322
1323	/// `--listen-addr` of a node behind a NAT. Port `0` so concurrent tests don't collide.
1324	const WEBRTC_LISTEN_ADDRESS: &str = "/ip4/127.0.0.1/udp/0/webrtc-direct";
1325
1326	/// `--public-addr` of that node: the routable coordinate peers are told to dial.
1327	/// This is the shape an operator supplies when the node sits behind a proxy.
1328	const WEBRTC_PUBLIC_ADDRESS: &str = "/ip4/203.0.113.9/udp/31234/webrtc-direct";
1329
1330	/// The `/certhash` component of `address`.
1331	fn certhash(address: &NetworkMultiaddr) -> Option<NetworkMultihash> {
1332		address.iter().find_map(|protocol| match protocol {
1333			NetworkProtocol::Certhash(hash) => Some(hash),
1334			_ => None,
1335		})
1336	}
1337
1338	/// Bring up the backend from `network_config` through its real entry point.
1339	fn start_backend(
1340		network_config: &NetworkConfiguration,
1341	) -> Result<Litep2pNetworkBackend, Error> {
1342		let config = FullNetworkConfiguration::<Block, H256, Litep2pNetworkBackend>::new(
1343			network_config,
1344			None,
1345		);
1346
1347		let (block_announce_config, _notification_service) =
1348			<Litep2pNetworkBackend as NetworkBackend<Block, H256>>::notification_config(
1349				"/block-announces/1".into(),
1350				vec![],
1351				1024,
1352				None,
1353				SetConfig::default(),
1354				NotificationMetrics::new(None),
1355				config.peer_store_handle(),
1356			);
1357
1358		<Litep2pNetworkBackend as NetworkBackend<Block, H256>>::new(Params {
1359			role: Role::Full,
1360			executor: Box::new(|future| {
1361				tokio::spawn(future);
1362			}),
1363			network_config: config,
1364			protocol_id: ProtocolId::from("test"),
1365			genesis_hash: H256::zero(),
1366			fork_id: None,
1367			metrics_registry: None,
1368			block_announce_config,
1369			ipfs_config: None,
1370			notification_metrics: NotificationMetrics::new(None),
1371		})
1372	}
1373
1374	/// Both the address the node binds and the one it tells peers to dial must carry the node's
1375	/// `/certhash`: a `webrtc-direct` dialer has no other way to verify the DTLS handshake.
1376	#[tokio::test]
1377	async fn webrtc_addresses_advertised_with_certhash() {
1378		let mut network_config = NetworkConfiguration::new_local();
1379		network_config.listen_addresses = vec![WEBRTC_LISTEN_ADDRESS.parse().unwrap()];
1380		network_config.public_addresses = vec![WEBRTC_PUBLIC_ADDRESS.parse().unwrap()];
1381		// Fixed, so the certificate the node will present can be derived here as well.
1382		network_config.node_key = NodeKeyConfig::Ed25519(Secret::Input(
1383			ed25519::SecretKey::try_from_bytes([7u8; 32]).unwrap(),
1384		));
1385		network_config.validate_and_complete_webrtc_addresses().unwrap();
1386
1387		let (keypair, _peer_id) =
1388			Litep2pNetworkBackend::get_keypair(&network_config.node_key).unwrap();
1389		let node_certhash: NetworkMultihash =
1390			webrtc::derive_certificate(keypair.secret()).unwrap().certhash().into();
1391
1392		// Held for the duration of the test: dropping it closes the node's sockets.
1393		let backend = start_backend(&network_config).unwrap();
1394		let network_service =
1395			<Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1396
1397		let advertised = network_service.listen_addresses();
1398		assert_eq!(advertised.len(), 1);
1399		assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1400
1401		let advertised = network_service.external_addresses();
1402		assert_eq!(advertised.len(), 1);
1403		assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1404	}
1405
1406	/// The default node key (`Secret::New`) generates a fresh key per `into_keypair()` call:
1407	/// completing the addresses must pin the resolved key, or the backend would serve a
1408	/// certificate matching neither the advertised `/certhash` nor each other's.
1409	#[tokio::test]
1410	async fn webrtc_certhash_consistent_with_default_node_key() {
1411		let mut network_config = NetworkConfiguration::new_local();
1412		network_config.listen_addresses = vec![WEBRTC_LISTEN_ADDRESS.parse().unwrap()];
1413		network_config.public_addresses = vec![WEBRTC_PUBLIC_ADDRESS.parse().unwrap()];
1414		network_config.validate_and_complete_webrtc_addresses().unwrap();
1415
1416		// Completing the addresses pinned the key, so this is the key the backend serves.
1417		let (keypair, _peer_id) =
1418			Litep2pNetworkBackend::get_keypair(&network_config.node_key).unwrap();
1419		let node_certhash: NetworkMultihash =
1420			webrtc::derive_certificate(keypair.secret()).unwrap().certhash().into();
1421
1422		// Held for the duration of the test: dropping it closes the node's sockets.
1423		let backend = start_backend(&network_config).unwrap();
1424		let network_service =
1425			<Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1426
1427		let advertised = network_service.listen_addresses();
1428		assert_eq!(advertised.len(), 1);
1429		assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1430
1431		let advertised = network_service.external_addresses();
1432		assert_eq!(advertised.len(), 1);
1433		assert_eq!(certhash(&advertised[0]), Some(node_certhash));
1434	}
1435
1436	#[tokio::test]
1437	async fn webrtc_public_address_completed_at_config_creation_accepted() {
1438		let mut network_config = NetworkConfiguration::new_local();
1439		network_config.listen_addresses = vec![WEBRTC_LISTEN_ADDRESS.parse().unwrap()];
1440		network_config.public_addresses = vec![WEBRTC_PUBLIC_ADDRESS.parse().unwrap()];
1441		network_config.node_key = NodeKeyConfig::Ed25519(Secret::Input(
1442			ed25519::SecretKey::try_from_bytes([7u8; 32]).unwrap(),
1443		));
1444		network_config.validate_and_complete_webrtc_addresses().unwrap();
1445
1446		// Held for the duration of the test: dropping it closes the node's sockets.
1447		let backend = start_backend(&network_config).unwrap();
1448		let network_service =
1449			<Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1450		let local_peer_id = network_service.local_peer_id();
1451
1452		assert_eq!(
1453			network_service.external_addresses(),
1454			vec![network_config.public_addresses[0]
1455				.clone()
1456				.with(NetworkProtocol::P2p(local_peer_id.into()))],
1457		);
1458	}
1459
1460	#[tokio::test]
1461	async fn webrtc_listen_address_with_certhash_refuses_to_start() {
1462		let listen_address: NetworkMultiaddr = WEBRTC_LISTEN_ADDRESS.parse().unwrap();
1463
1464		let mut network_config = NetworkConfiguration::new_local();
1465		network_config.listen_addresses =
1466			vec![listen_address.with(NetworkProtocol::Certhash(a_certhash()))];
1467
1468		assert!(matches!(start_backend(&network_config), Err(Error::InvalidWebRtcAddress { .. }),));
1469	}
1470
1471	#[tokio::test]
1472	async fn non_webrtc_public_address_untouched() {
1473		let public_address: NetworkMultiaddr = "/ip4/203.0.113.9/tcp/31234".parse().unwrap();
1474
1475		let mut network_config = NetworkConfiguration::new_local();
1476		network_config.listen_addresses = vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()];
1477		network_config.public_addresses = vec![public_address.clone()];
1478
1479		// Held for the duration of the test: dropping it closes the node's sockets.
1480		let backend = start_backend(&network_config).unwrap();
1481		let network_service =
1482			<Litep2pNetworkBackend as NetworkBackend<Block, H256>>::network_service(&backend);
1483		let local_peer_id = network_service.local_peer_id();
1484
1485		// litep2p appends the local `/p2p` to every public address; nothing else is added.
1486		assert_eq!(
1487			network_service.external_addresses(),
1488			vec![public_address.with(NetworkProtocol::P2p(local_peer_id.into()))],
1489		);
1490	}
1491
1492	/// A certhash standing in for the node's own.
1493	fn a_certhash() -> NetworkMultihash {
1494		sc_network_types::multihash::Code::Sha2_256.digest(b"certificate")
1495	}
1496}