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	/// Protocols the peer reported over identify, replayed to an event stream that subscribes
145	/// after the peer was identified.
146	supported_protocols: Option<HashSet<ProtocolName>>,
147}
148
149/// Kademlia query we are tracking.
150#[derive(Debug)]
151enum KadQuery {
152	/// `FIND_NODE` query for target and when it was initiated.
153	FindNode(PeerId, Instant),
154	/// `GET_VALUE` query for key and when it was initiated.
155	GetValue(RecordKey, Instant),
156	/// `PUT_VALUE` query for key and when it was initiated.
157	PutValue(RecordKey, Instant),
158	/// `GET_PROVIDERS` query for key and when it was initiated.
159	GetProviders(RecordKey, Instant),
160	/// `ADD_PROVIDER` query for key and when it was initiated.
161	AddProvider(RecordKey, Instant),
162}
163
164/// Networking backend for `litep2p`.
165pub struct Litep2pNetworkBackend {
166	/// Main `litep2p` object.
167	litep2p: Litep2p,
168
169	/// `NetworkService` implementation for `Litep2pNetworkBackend`.
170	network_service: Arc<dyn NetworkService>,
171
172	/// RX channel for receiving commands from `Litep2pNetworkService`.
173	cmd_rx: TracingUnboundedReceiver<NetworkServiceCommand>,
174
175	/// `Peerset` handles to notification protocols.
176	peerset_handles: HashMap<ProtocolName, ProtocolControlHandle>,
177
178	/// Pending Kademlia queries.
179	pending_queries: HashMap<QueryId, KadQuery>,
180
181	/// Discovery.
182	discovery: Discovery,
183
184	/// Number of connected peers.
185	num_connected: Arc<AtomicUsize>,
186
187	/// Connected peers.
188	peers: HashMap<litep2p::PeerId, ConnectionContext>,
189
190	/// Peerstore.
191	peerstore_handle: Arc<dyn PeerStoreProvider>,
192
193	/// Block announce protocol name.
194	block_announce_protocol: ProtocolName,
195
196	/// Sender for DHT events.
197	event_streams: out_events::OutChannels,
198
199	/// Prometheus metrics.
200	metrics: Option<Metrics>,
201}
202
203impl Litep2pNetworkBackend {
204	/// From an iterator of multiaddress(es), parse and group all addresses of peers
205	/// so that litep2p can consume the information easily.
206	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	/// Add new known addresses to `litep2p` and return the parsed peer IDs.
235	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				// `peers` contained multiaddress in the form `/p2p/<peer ID>`
240				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	/// Get `litep2p` keypair from `NodeKeyConfig`.
261	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	/// Configure transport protocols for `Litep2pNetworkBackend`.
273	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				// Plain TCP address.
305				(Some(NetworkProtocol::Tcp(_)), Some(NetworkProtocol::P2p(_)) | None) => {
306					tcp_addresses.push(addr.clone());
307				},
308				// Websocket address.
309				(
310					Some(NetworkProtocol::Tcp(_)),
311					Some(NetworkProtocol::Ws(_) | NetworkProtocol::Wss(_)),
312				) => {
313					websocket_addresses.push(addr.clone());
314				},
315				// WebRTC-Direct address.
316				(Some(NetworkProtocol::Udp(_)), Some(NetworkProtocol::WebRTCDirect)) => {
317					// An address carrying anything past `webrtc-direct` is rejected.
318					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		// Install the ring CryptoProvider for rustls before any TLS connections are made.
373		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(&params.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) = &params.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(&params.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		// initialize notification protocols
434		//
435		// pass the protocol configuration to `Litep2pConfigBuilder` and save the TX channel
436		// to the protocol's `Peerset` together with the protocol name to allow other subsystems
437		// of Polkadot SDK to control connectivity of the notification protocol
438		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		// handshake for all but the syncing protocol is set to node role
445		config_builder = notification_protocols
446			.into_iter()
447			.fold(config_builder, |config_builder, mut config| {
448				config.config.set_handshake(Roles::from(&params.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		// initialize request-response protocols
456		let metrics = match &params.metrics_registry {
457			Some(registry) => Some(register_without_sources(registry)?),
458			None => None,
459		};
460
461		// create channels that are used to send request before initializing protocols so the
462		// senders can be passed onto all request-response protocols
463		//
464		// all protocols must have each others' senders so they can send the fallback request in
465		// case the main protocol is not supported by the remote peer and user specified a fallback
466		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		// collect known addresses
509		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		// enable ipfs ping, identify and kademlia, and potentially mdns if user enabled it
529		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				&params.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			// Use system DNS resolver to enable intranet domain resolution and administrator
568			// control over DNS lookup.
569			.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		// register rest of the metrics now that `Litep2p` has been created
608		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) = &params.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	/// Create notification protocol configuration for `protocol`.
648	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	/// Create request-response protocol configuration.
669	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	/// Start [`Litep2pNetworkBackend`] event loop.
688	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								// libp2p backend generates `DiscoveryOut::Discovered(peer_id)`
772								// event when a new address is added for a peer, which leads to the
773								// peer being added to peerstore. Do the same directly here.
774								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						// if at least one address was added for the peer, report the peer to `Peerstore`
832						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								// We likely requested providers to connect to them,
981								// so let's add their addresses to litep2p's transport manager.
982								// Consider also looking the addresses of providers up with `FIND_NODE`
983								// query, as it can yield more up to date addresses.
984								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								// litep2p returns all providers in a single event, so we let
996								// subscribers know no more providers will be yielded.
997								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						// Litep2p requires the peer ID to be present in the address.
1177						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								// Increment incoming connections counter.
1221								//
1222								// Note: For litep2p these are represented by established negotiated connections,
1223								// while for libp2p (legacy) these represent not-yet-negotiated connections.
1224								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	/// `--listen-addr` of a node behind a NAT. Port `0` so concurrent tests don't collide.
1350	const WEBRTC_LISTEN_ADDRESS: &str = "/ip4/127.0.0.1/udp/0/webrtc-direct";
1351
1352	/// `--public-addr` of that node: the routable coordinate peers are told to dial.
1353	/// This is the shape an operator supplies when the node sits behind a proxy.
1354	const WEBRTC_PUBLIC_ADDRESS: &str = "/ip4/203.0.113.9/udp/31234/webrtc-direct";
1355
1356	/// The `/certhash` component of `address`.
1357	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	/// Bring up the backend from `network_config` through its real entry point.
1365	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	/// Both the address the node binds and the one it tells peers to dial must carry the node's
1401	/// `/certhash`: a `webrtc-direct` dialer has no other way to verify the DTLS handshake.
1402	#[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		// Fixed, so the certificate the node will present can be derived here as well.
1408		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		// Held for the duration of the test: dropping it closes the node's sockets.
1419		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	/// The default node key (`Secret::New`) generates a fresh key per `into_keypair()` call:
1433	/// completing the addresses must pin the resolved key, or the backend would serve a
1434	/// certificate matching neither the advertised `/certhash` nor each other's.
1435	#[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		// Completing the addresses pinned the key, so this is the key the backend serves.
1443		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		// Held for the duration of the test: dropping it closes the node's sockets.
1449		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		// Held for the duration of the test: dropping it closes the node's sockets.
1473		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	/// A configured `/certhash` is checked against the node's own and removed by
1487	/// `NetworkConfiguration::validate_and_complete_addresses`, which this test deliberately skips:
1488	/// the backend is the guard for a configuration built without it.
1489	#[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		// Held for the duration of the test: dropping it closes the node's sockets.
1509		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		// litep2p appends the local `/p2p` to every public address; nothing else is added.
1515		assert_eq!(
1516			network_service.external_addresses(),
1517			vec![public_address.with(NetworkProtocol::P2p(local_peer_id.into()))],
1518		);
1519	}
1520
1521	/// A certhash standing in for the node's own.
1522	fn a_certhash() -> NetworkMultihash {
1523		sc_network_types::multihash::Code::Sha2_256.digest(b"certificate")
1524	}
1525}