Authentication of PeerIds in authority discovery records (#10317)

* Consolidating test and production code

* Signing/verifying authority discovery records with PeerId

Unsigned records cannot be rejected yet, they just produce
a warning in the log.

* Upgrading to libp2p 0.40

* libp2p::identity and sp_core::crypto Ed25519 are compatible

* Rejecting authority records unsigned by peer id can be configured

* Fixes based on review comments

* No command-line argument needed

* info was still too much spam in the logs

* Added tests for both strict and loose validation

* Fixing based on review comments

* Pierre preferred a signing method

* Ooops, I need to slow down

* Update bin/node/cli/src/service.rs

* Reexport libp2p crypto used in sc-network

* Added proto3 compatibility tests. And import noise.

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

substrate/client/authority-discovery/src/error.rs

@@ -46,10 +46,18 @@ pub enum Error {
 	EncodingDecodingScale(codec::Error),
 	/// Failed to parse a libp2p multi address.
 	ParsingMultiaddress(libp2p::core::multiaddr::Error),
+	/// Failed to parse a libp2p key.
+	ParsingLibp2pIdentity(sc_network::DecodingError),
 	/// Failed to sign using a specific public key.
 	MissingSignature(CryptoTypePublicPair),
 	/// Failed to sign using all public keys.
 	Signing,
 	/// Failed to register Prometheus metric.
 	Prometheus(prometheus_endpoint::PrometheusError),
+	/// Received authority record that contains addresses with multiple peer ids
+	ReceivingDhtValueFoundEventWithDifferentPeerIds,
+	/// Received authority record without any addresses having a peer id
+	ReceivingDhtValueFoundEventWithNoPeerIds,
+	/// Received authority record without a valid signature for the remote peer id.
+	MissingPeerIdSignature,
 }

substrate/client/authority-discovery/src/lib.rs

@@ -78,6 +78,11 @@ pub struct WorkerConfig {
 	///
 	/// Defaults to `true` to avoid the surprise factor.
 	pub publish_non_global_ips: bool,
+
+	/// Reject authority discovery records that are not signed by their network identity (PeerId)
+	///
+	/// Defaults to `false` to provide compatibility with old versions
+	pub strict_record_validation: bool,
 }
 
 impl Default for WorkerConfig {
@@ -98,6 +103,7 @@ impl Default for WorkerConfig {
 			// `authority_discovery_dht_event_received`.
 			max_query_interval: Duration::from_secs(10 * 60),
 			publish_non_global_ips: true,
+			strict_record_validation: false,
 		}
 	}
 }

substrate/client/authority-discovery/src/tests.rs

@@ -82,3 +82,32 @@ fn get_addresses_and_authority_id() {
 		);
 	});
 }
+
+#[test]
+fn cryptos_are_compatible() {
+	use sp_core::crypto::Pair;
+
+	let libp2p_secret = sc_network::Keypair::generate_ed25519();
+	let libp2p_public = libp2p_secret.public();
+
+	let sp_core_secret = {
+		let libp2p_ed_secret = match libp2p_secret.clone() {
+			sc_network::Keypair::Ed25519(x) => x,
+			_ => panic!("generate_ed25519 should have generated an Ed25519 key ¯\\_(ツ)_/¯"),
+		};
+		sp_core::ed25519::Pair::from_seed_slice(&libp2p_ed_secret.secret().as_ref()).unwrap()
+	};
+	let sp_core_public = sp_core_secret.public();
+
+	let message = b"we are more powerful than not to be better";
+
+	let libp2p_signature = libp2p_secret.sign(message).unwrap();
+	let sp_core_signature = sp_core_secret.sign(message); // no error expected...
+
+	assert!(sp_core::ed25519::Pair::verify(
+		&sp_core::ed25519::Signature::from_slice(&libp2p_signature),
+		message,
+		&sp_core_public
+	));
+	assert!(libp2p_public.verify(message, sp_core_signature.as_ref()));
+}

substrate/client/authority-discovery/src/worker.rs

@@ -19,7 +19,7 @@
 use crate::{
 	error::{Error, Result},
 	interval::ExpIncInterval,
-	ServicetoWorkerMsg,
+	ServicetoWorkerMsg, WorkerConfig,
 };
 
 use std::{
@@ -57,7 +57,10 @@ use sp_runtime::{generic::BlockId, traits::Block as BlockT};
 mod addr_cache;
 /// Dht payload schemas generated from Protobuf definitions via Prost crate in build.rs.
 mod schema {
-	include!(concat!(env!("OUT_DIR"), "/authority_discovery.rs"));
+	#[cfg(test)]
+	mod tests;
+
+	include!(concat!(env!("OUT_DIR"), "/authority_discovery_v2.rs"));
 }
 #[cfg(test)]
 pub mod tests;
@@ -111,6 +114,7 @@ pub struct Worker<Client, Network, Block, DhtEventStream> {
 	client: Arc<Client>,
 
 	network: Arc<Network>,
+
 	/// Channel we receive Dht events on.
 	dht_event_rx: DhtEventStream,
 
@@ -124,6 +128,8 @@ pub struct Worker<Client, Network, Block, DhtEventStream> {
 	latest_published_keys: HashSet<CryptoTypePublicPair>,
 	/// Same value as in the configuration.
 	publish_non_global_ips: bool,
+	/// Same value as in the configuration.
+	strict_record_validation: bool,
 
 	/// Interval at which to request addresses of authorities, refilling the pending lookups queue.
 	query_interval: ExpIncInterval,
@@ -131,7 +137,7 @@ pub struct Worker<Client, Network, Block, DhtEventStream> {
 	/// Queue of throttled lookups pending to be passed to the network.
 	pending_lookups: Vec<AuthorityId>,
 	/// Set of in-flight lookups.
-	in_flight_lookups: HashMap<libp2p::kad::record::Key, AuthorityId>,
+	in_flight_lookups: HashMap<sc_network::KademliaKey, AuthorityId>,
 
 	addr_cache: addr_cache::AddrCache,
 
@@ -158,7 +164,7 @@ where
 		dht_event_rx: DhtEventStream,
 		role: Role,
 		prometheus_registry: Option<prometheus_endpoint::Registry>,
-		config: crate::WorkerConfig,
+		config: WorkerConfig,
 	) -> Self {
 		// When a node starts up publishing and querying might fail due to various reasons, for
 		// example due to being not yet fully bootstrapped on the DHT. Thus one should retry rather
@@ -197,6 +203,7 @@ where
 			publish_if_changed_interval,
 			latest_published_keys: HashSet::new(),
 			publish_non_global_ips: config.publish_non_global_ips,
+			strict_record_validation: config.strict_record_validation,
 			query_interval,
 			pending_lookups: Vec::new(),
 			in_flight_lookups: HashMap::new(),
@@ -313,7 +320,7 @@ where
 			return Ok(())
 		}
 
-		let addresses = self.addresses_to_publish().map(|a| a.to_vec()).collect::<Vec<_>>();
+		let addresses = serialize_addresses(self.addresses_to_publish());
 
 		if let Some(metrics) = &self.metrics {
 			metrics.publish.inc();
@@ -322,32 +329,21 @@ where
 				.set(addresses.len().try_into().unwrap_or(std::u64::MAX));
 		}
 
-		let mut serialized_addresses = vec![];
-		schema::AuthorityAddresses { addresses }
-			.encode(&mut serialized_addresses)
-			.map_err(Error::EncodingProto)?;
+		let serialized_record = serialize_authority_record(addresses)?;
+		let peer_signature = sign_record_with_peer_id(&serialized_record, self.network.as_ref())?;
 
 		let keys_vec = keys.iter().cloned().collect::<Vec<_>>();
-		let signatures = key_store
-			.sign_with_all(
-				key_types::AUTHORITY_DISCOVERY,
-				keys_vec.clone(),
-				serialized_addresses.as_slice(),
-			)
-			.await
-			.map_err(|_| Error::Signing)?;
 
-		for (sign_result, key) in signatures.into_iter().zip(keys_vec.iter()) {
-			let mut signed_addresses = vec![];
+		let kv_pairs = sign_record_with_authority_ids(
+			serialized_record,
+			Some(peer_signature),
+			key_store.as_ref(),
+			keys_vec,
+		)
+		.await?;
 
-			// Verify that all signatures exist for all provided keys.
-			let signature =
-				sign_result.ok().flatten().ok_or_else(|| Error::MissingSignature(key.clone()))?;
-			schema::SignedAuthorityAddresses { addresses: serialized_addresses.clone(), signature }
-				.encode(&mut signed_addresses)
-				.map_err(Error::EncodingProto)?;
-
-			self.network.put_value(hash_authority_id(key.1.as_ref()), signed_addresses);
+		for (key, value) in kv_pairs.into_iter() {
+			self.network.put_value(key, value);
 		}
 
 		self.latest_published_keys = keys;
@@ -471,18 +467,11 @@ where
 
 	fn handle_dht_value_found_event(
 		&mut self,
-		values: Vec<(libp2p::kad::record::Key, Vec<u8>)>,
+		values: Vec<(sc_network::KademliaKey, Vec<u8>)>,
 	) -> Result<()> {
 		// Ensure `values` is not empty and all its keys equal.
-		let remote_key = values
-			.iter()
-			.fold(Ok(None), |acc, (key, _)| match acc {
-				Ok(None) => Ok(Some(key.clone())),
-				Ok(Some(ref prev_key)) if prev_key != key =>
-					Err(Error::ReceivingDhtValueFoundEventWithDifferentKeys),
-				x @ Ok(_) => x,
-				Err(e) => Err(e),
-			})?
+		let remote_key = single(values.iter().map(|(key, _)| key.clone()))
+			.map_err(|_| Error::ReceivingDhtValueFoundEventWithDifferentKeys)?
 			.ok_or(Error::ReceivingDhtValueFoundEventWithNoRecords)?;
 
 		let authority_id: AuthorityId = self
@@ -495,18 +484,18 @@ where
 		let remote_addresses: Vec<Multiaddr> = values
 			.into_iter()
 			.map(|(_k, v)| {
-				let schema::SignedAuthorityAddresses { signature, addresses } =
-					schema::SignedAuthorityAddresses::decode(v.as_slice())
+				let schema::SignedAuthorityRecord { record, auth_signature, peer_signature } =
+					schema::SignedAuthorityRecord::decode(v.as_slice())
 						.map_err(Error::DecodingProto)?;
 
-				let signature = AuthoritySignature::decode(&mut &signature[..])
+				let auth_signature = AuthoritySignature::decode(&mut &auth_signature[..])
 					.map_err(Error::EncodingDecodingScale)?;
 
-				if !AuthorityPair::verify(&signature, &addresses, &authority_id) {
+				if !AuthorityPair::verify(&auth_signature, &record, &authority_id) {
 					return Err(Error::VerifyingDhtPayload)
 				}
 
-				let addresses = schema::AuthorityAddresses::decode(addresses.as_slice())
+				let addresses: Vec<Multiaddr> = schema::AuthorityRecord::decode(record.as_slice())
 					.map(|a| a.addresses)
 					.map_err(Error::DecodingProto)?
 					.into_iter()
@@ -514,32 +503,49 @@ where
 					.collect::<std::result::Result<_, _>>()
 					.map_err(Error::ParsingMultiaddress)?;
 
+				let get_peer_id = |a: &Multiaddr| match a.iter().last() {
+					Some(multiaddr::Protocol::P2p(key)) => PeerId::from_multihash(key).ok(),
+					_ => None,
+				};
+
+				// Ignore [`Multiaddr`]s without [`PeerId`] or with own addresses.
+				let addresses: Vec<Multiaddr> = addresses
+					.into_iter()
+					.filter(|a| get_peer_id(&a).filter(|p| *p != local_peer_id).is_some())
+					.collect();
+
+				let remote_peer_id = single(addresses.iter().map(get_peer_id))
+					.map_err(|_| Error::ReceivingDhtValueFoundEventWithDifferentPeerIds)? // different peer_id in records
+					.flatten()
+					.ok_or(Error::ReceivingDhtValueFoundEventWithNoPeerIds)?; // no records with peer_id in them
+
+				// At this point we know all the valid multiaddresses from the record, know that
+				// each of them belong to the same PeerId, we just need to check if the record is
+				// properly signed by the owner of the PeerId
+
+				if let Some(peer_signature) = peer_signature {
+					let public_key =
+						sc_network::PublicKey::from_protobuf_encoding(&peer_signature.public_key)
+							.map_err(|e| Error::ParsingLibp2pIdentity(e))?;
+					let signature =
+						sc_network::Signature { public_key, bytes: peer_signature.signature };
+
+					if !signature.verify(record, &remote_peer_id) {
+						return Err(Error::VerifyingDhtPayload)
+					}
+				} else if self.strict_record_validation {
+					return Err(Error::MissingPeerIdSignature)
+				} else {
+					debug!(
+						target: LOG_TARGET,
+						"Received unsigned authority discovery record from {}", authority_id
+					);
+				}
 				Ok(addresses)
 			})
 			.collect::<Result<Vec<Vec<Multiaddr>>>>()?
 			.into_iter()
 			.flatten()
-			// Ignore [`Multiaddr`]s without [`PeerId`] and own addresses.
-			.filter(|addr| {
-				addr.iter().any(|protocol| {
-					// Parse to PeerId first as Multihashes of old and new PeerId
-					// representation don't equal.
-					//
-					// See https://github.com/libp2p/rust-libp2p/issues/555 for
-					// details.
-					if let multiaddr::Protocol::P2p(hash) = protocol {
-						let peer_id = match PeerId::from_multihash(hash) {
-							Ok(peer_id) => peer_id,
-							Err(_) => return false, // Discard address.
-						};
-
-						// Discard if equal to local peer id, keep if it differs.
-						return !(peer_id == local_peer_id)
-					}
-
-					false // `protocol` is not a [`Protocol::P2p`], let's keep looking.
-				})
-			})
 			.take(MAX_ADDRESSES_PER_AUTHORITY)
 			.collect();
 
@@ -588,16 +594,37 @@ where
 	}
 }
 
+pub trait NetworkSigner {
+	/// Sign a message in the name of `self.local_peer_id()`
+	fn sign_with_local_identity(
+		&self,
+		msg: impl AsRef<[u8]>,
+	) -> std::result::Result<sc_network::Signature, sc_network::SigningError>;
+}
+
 /// NetworkProvider provides [`Worker`] with all necessary hooks into the
 /// underlying Substrate networking. Using this trait abstraction instead of
 /// [`sc_network::NetworkService`] directly is necessary to unit test [`Worker`].
 #[async_trait]
-pub trait NetworkProvider: NetworkStateInfo {
+pub trait NetworkProvider: NetworkStateInfo + NetworkSigner {
 	/// Start putting a value in the Dht.
-	fn put_value(&self, key: libp2p::kad::record::Key, value: Vec<u8>);
+	fn put_value(&self, key: sc_network::KademliaKey, value: Vec<u8>);
 
 	/// Start getting a value from the Dht.
-	fn get_value(&self, key: &libp2p::kad::record::Key);
+	fn get_value(&self, key: &sc_network::KademliaKey);
+}
+
+impl<B, H> NetworkSigner for sc_network::NetworkService<B, H>
+where
+	B: BlockT + 'static,
+	H: ExHashT,
+{
+	fn sign_with_local_identity(
+		&self,
+		msg: impl AsRef<[u8]>,
+	) -> std::result::Result<sc_network::Signature, sc_network::SigningError> {
+		self.sign_with_local_identity(msg)
+	}
 }
 
 #[async_trait::async_trait]
@@ -606,16 +633,87 @@ where
 	B: BlockT + 'static,
 	H: ExHashT,
 {
-	fn put_value(&self, key: libp2p::kad::record::Key, value: Vec<u8>) {
+	fn put_value(&self, key: sc_network::KademliaKey, value: Vec<u8>) {
 		self.put_value(key, value)
 	}
-	fn get_value(&self, key: &libp2p::kad::record::Key) {
+	fn get_value(&self, key: &sc_network::KademliaKey) {
 		self.get_value(key)
 	}
 }
 
-fn hash_authority_id(id: &[u8]) -> libp2p::kad::record::Key {
-	libp2p::kad::record::Key::new(&libp2p::multihash::Sha2_256::digest(id))
+fn hash_authority_id(id: &[u8]) -> sc_network::KademliaKey {
+	sc_network::KademliaKey::new(&libp2p::multihash::Sha2_256::digest(id))
+}
+
+// Makes sure all values are the same and returns it
+//
+// Returns Err(_) if not all values are equal. Returns Ok(None) if there are
+// no values.
+fn single<T>(values: impl IntoIterator<Item = T>) -> std::result::Result<Option<T>, ()>
+where
+	T: PartialEq<T>,
+{
+	values.into_iter().try_fold(None, |acc, item| match acc {
+		None => Ok(Some(item)),
+		Some(ref prev) if *prev != item => Err(()),
+		Some(x) => Ok(Some(x)),
+	})
+}
+
+fn serialize_addresses(addresses: impl Iterator<Item = Multiaddr>) -> Vec<Vec<u8>> {
+	addresses.map(|a| a.to_vec()).collect()
+}
+
+fn serialize_authority_record(addresses: Vec<Vec<u8>>) -> Result<Vec<u8>> {
+	let mut serialized_record = vec![];
+	schema::AuthorityRecord { addresses }
+		.encode(&mut serialized_record)
+		.map_err(Error::EncodingProto)?;
+	Ok(serialized_record)
+}
+
+fn sign_record_with_peer_id(
+	serialized_record: &[u8],
+	network: &impl NetworkSigner,
+) -> Result<schema::PeerSignature> {
+	let signature = network
+		.sign_with_local_identity(serialized_record)
+		.map_err(|_| Error::Signing)?;
+	let public_key = signature.public_key.to_protobuf_encoding();
+	let signature = signature.bytes;
+	Ok(schema::PeerSignature { signature, public_key })
+}
+
+async fn sign_record_with_authority_ids(
+	serialized_record: Vec<u8>,
+	peer_signature: Option<schema::PeerSignature>,
+	key_store: &dyn CryptoStore,
+	keys: Vec<CryptoTypePublicPair>,
+) -> Result<Vec<(sc_network::KademliaKey, Vec<u8>)>> {
+	let signatures = key_store
+		.sign_with_all(key_types::AUTHORITY_DISCOVERY, keys.clone(), &serialized_record)
+		.await
+		.map_err(|_| Error::Signing)?;
+
+	let mut result = vec![];
+	for (sign_result, key) in signatures.into_iter().zip(keys.iter()) {
+		let mut signed_record = vec![];
+
+		// Verify that all signatures exist for all provided keys.
+		let auth_signature =
+			sign_result.ok().flatten().ok_or_else(|| Error::MissingSignature(key.clone()))?;
+		schema::SignedAuthorityRecord {
+			record: serialized_record.clone(),
+			auth_signature,
+			peer_signature: peer_signature.clone(),
+		}
+		.encode(&mut signed_record)
+		.map_err(Error::EncodingProto)?;
+
+		result.push((hash_authority_id(key.1.as_ref()), signed_record));
+	}
+
+	Ok(result)
 }
 
 /// Prometheus metrics for a [`Worker`].

substrate/client/authority-discovery/src/worker/schema/dht-v1.proto

@@ -1,6 +1,6 @@
 syntax = "proto3";
 
-package authority_discovery;
+package authority_discovery_v1;
 
 // First we need to serialize the addresses in order to be able to sign them.
 message AuthorityAddresses {
@@ -11,4 +11,4 @@ message AuthorityAddresses {
 message SignedAuthorityAddresses {
 	bytes addresses = 1;
 	bytes signature = 2;
-}
+}

substrate/client/authority-discovery/src/worker/schema/dht-v2.proto

@@ -0,0 +1,23 @@
+syntax = "proto3";
+
+package authority_discovery_v2;
+
+// First we need to serialize the addresses in order to be able to sign them.
+message AuthorityRecord {
+	// Possibly multiple `MultiAddress`es through which the node can be 
+	repeated bytes addresses = 1;
+}
+
+message PeerSignature {
+	bytes signature = 1;
+	bytes public_key = 2;
+}
+
+// Then we need to serialize the authority record and signature to send them over the wire.
+message SignedAuthorityRecord {
+	bytes record = 1;
+	bytes auth_signature = 2;
+	// Even if there are multiple `record.addresses`, all of them have the same peer id.
+	// Old versions are missing this field, so `optional` will provide compatibility both ways.
+	optional PeerSignature peer_signature = 3;
+}

substrate/client/authority-discovery/src/worker/schema/tests.rs

@@ -0,0 +1,90 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2017-2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+mod schema_v1 {
+	include!(concat!(env!("OUT_DIR"), "/authority_discovery_v1.rs"));
+}
+
+use super::*;
+use libp2p::multiaddr::Multiaddr;
+use prost::Message;
+use sc_network::PeerId;
+
+#[test]
+fn v2_decodes_v1() {
+	let peer_id = PeerId::random();
+	let multiaddress: Multiaddr =
+		format!("/ip4/127.0.0.1/tcp/3003/p2p/{}", peer_id).parse().unwrap();
+	let vec_addresses = vec![multiaddress.to_vec()];
+	let vec_auth_signature = b"Totally valid signature, I promise!".to_vec();
+
+	let addresses_v1 = schema_v1::AuthorityAddresses { addresses: vec_addresses.clone() };
+	let mut vec_addresses_v1 = vec![];
+	addresses_v1.encode(&mut vec_addresses_v1).unwrap();
+	let signed_addresses_v1 = schema_v1::SignedAuthorityAddresses {
+		addresses: vec_addresses_v1.clone(),
+		signature: vec_auth_signature.clone(),
+	};
+	let mut vec_signed_addresses_v1 = vec![];
+	signed_addresses_v1.encode(&mut vec_signed_addresses_v1).unwrap();
+
+	let signed_record_v2_decoded =
+		SignedAuthorityRecord::decode(vec_signed_addresses_v1.as_slice()).unwrap();
+
+	assert_eq!(&signed_record_v2_decoded.record, &vec_addresses_v1);
+	assert_eq!(&signed_record_v2_decoded.auth_signature, &vec_auth_signature);
+	assert_eq!(&signed_record_v2_decoded.peer_signature, &None);
+
+	let record_v2_decoded = AuthorityRecord::decode(vec_addresses_v1.as_slice()).unwrap();
+	assert_eq!(&record_v2_decoded.addresses, &vec_addresses);
+}
+
+#[test]
+fn v1_decodes_v2() {
+	let peer_secret = sc_network::Keypair::generate_ed25519();
+	let peer_public = peer_secret.public();
+	let peer_id = peer_public.to_peer_id();
+	let multiaddress: Multiaddr =
+		format!("/ip4/127.0.0.1/tcp/3003/p2p/{}", peer_id).parse().unwrap();
+	let vec_addresses = vec![multiaddress.to_vec()];
+	let vec_auth_signature = b"Totally valid signature, I promise!".to_vec();
+	let vec_peer_signature = b"Surprisingly hard to crack crypto".to_vec();
+
+	let record_v2 = AuthorityRecord { addresses: vec_addresses.clone() };
+	let mut vec_record_v2 = vec![];
+	record_v2.encode(&mut vec_record_v2).unwrap();
+	let vec_peer_public = peer_public.to_protobuf_encoding();
+	let peer_signature_v2 =
+		PeerSignature { public_key: vec_peer_public, signature: vec_peer_signature };
+	let signed_record_v2 = SignedAuthorityRecord {
+		record: vec_record_v2.clone(),
+		auth_signature: vec_auth_signature.clone(),
+		peer_signature: Some(peer_signature_v2.clone()),
+	};
+	let mut vec_signed_record_v2 = vec![];
+	signed_record_v2.encode(&mut vec_signed_record_v2).unwrap();
+
+	let signed_addresses_v1_decoded =
+		schema_v1::SignedAuthorityAddresses::decode(vec_signed_record_v2.as_slice()).unwrap();
+
+	assert_eq!(&signed_addresses_v1_decoded.addresses, &vec_record_v2);
+	assert_eq!(&signed_addresses_v1_decoded.signature, &vec_auth_signature);
+
+	let addresses_v2_decoded = AuthorityRecord::decode(vec_record_v2.as_slice()).unwrap();
+	assert_eq!(&addresses_v2_decoded.addresses, &vec_addresses);
+}

substrate/client/authority-discovery/src/worker/tests.rs

@@ -16,8 +16,6 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-use crate::worker::schema;
-
 use std::{
 	collections::HashSet,
 	sync::{Arc, Mutex},
@@ -32,11 +30,10 @@ use futures::{
 	sink::SinkExt,
 	task::LocalSpawn,
 };
-use libp2p::{core::multiaddr, kad, PeerId};
+use libp2p::{core::multiaddr, PeerId};
 use prometheus_endpoint::prometheus::default_registry;
 
 use sp_api::{ApiRef, ProvideRuntimeApi};
-use sp_core::crypto::Public;
 use sp_keystore::{testing::KeyStore, CryptoStore};
 use sp_runtime::traits::{Block as BlockT, NumberFor, Zero};
 use substrate_test_runtime_client::runtime::Block;
@@ -114,17 +111,18 @@ sp_api::mock_impl_runtime_apis! {
 
 #[derive(Debug)]
 pub enum TestNetworkEvent {
-	GetCalled(kad::record::Key),
-	PutCalled(kad::record::Key, Vec<u8>),
+	GetCalled(sc_network::KademliaKey),
+	PutCalled(sc_network::KademliaKey, Vec<u8>),
 }
 
 pub struct TestNetwork {
 	peer_id: PeerId,
+	identity: sc_network::Keypair,
 	external_addresses: Vec<Multiaddr>,
 	// Whenever functions on `TestNetwork` are called, the function arguments are added to the
 	// vectors below.
-	pub put_value_call: Arc<Mutex<Vec<(kad::record::Key, Vec<u8>)>>>,
-	pub get_value_call: Arc<Mutex<Vec<kad::record::Key>>>,
+	pub put_value_call: Arc<Mutex<Vec<(sc_network::KademliaKey, Vec<u8>)>>>,
+	pub get_value_call: Arc<Mutex<Vec<sc_network::KademliaKey>>>,
 	event_sender: mpsc::UnboundedSender<TestNetworkEvent>,
 	event_receiver: Option<mpsc::UnboundedReceiver<TestNetworkEvent>>,
 }
@@ -138,8 +136,10 @@ impl TestNetwork {
 impl Default for TestNetwork {
 	fn default() -> Self {
 		let (tx, rx) = mpsc::unbounded();
+		let identity = sc_network::Keypair::generate_ed25519();
 		TestNetwork {
-			peer_id: PeerId::random(),
+			peer_id: identity.public().to_peer_id(),
+			identity,
 			external_addresses: vec!["/ip6/2001:db8::/tcp/30333".parse().unwrap()],
 			put_value_call: Default::default(),
 			get_value_call: Default::default(),
@@ -149,16 +149,25 @@ impl Default for TestNetwork {
 	}
 }
 
+impl NetworkSigner for TestNetwork {
+	fn sign_with_local_identity(
+		&self,
+		msg: impl AsRef<[u8]>,
+	) -> std::result::Result<sc_network::Signature, sc_network::SigningError> {
+		sc_network::Signature::sign_message(msg, &self.identity)
+	}
+}
+
 #[async_trait]
 impl NetworkProvider for TestNetwork {
-	fn put_value(&self, key: kad::record::Key, value: Vec<u8>) {
+	fn put_value(&self, key: sc_network::KademliaKey, value: Vec<u8>) {
 		self.put_value_call.lock().unwrap().push((key.clone(), value.clone()));
 		self.event_sender
 			.clone()
 			.unbounded_send(TestNetworkEvent::PutCalled(key, value))
 			.unwrap();
 	}
-	fn get_value(&self, key: &kad::record::Key) {
+	fn get_value(&self, key: &sc_network::KademliaKey) {
 		self.get_value_call.lock().unwrap().push(key.clone());
 		self.event_sender
 			.clone()
@@ -177,35 +186,35 @@ impl NetworkStateInfo for TestNetwork {
 	}
 }
 
-async fn build_dht_event(
+impl NetworkSigner for sc_network::Keypair {
+	fn sign_with_local_identity(
+		&self,
+		msg: impl AsRef<[u8]>,
+	) -> std::result::Result<sc_network::Signature, sc_network::SigningError> {
+		sc_network::Signature::sign_message(msg, self)
+	}
+}
+
+async fn build_dht_event<Signer: NetworkSigner>(
 	addresses: Vec<Multiaddr>,
 	public_key: AuthorityId,
-	key_store: &KeyStore,
-) -> (libp2p::kad::record::Key, Vec<u8>) {
-	let mut serialized_addresses = vec![];
-	schema::AuthorityAddresses { addresses: addresses.into_iter().map(|a| a.to_vec()).collect() }
-		.encode(&mut serialized_addresses)
-		.map_err(Error::EncodingProto)
-		.unwrap();
+	key_store: &dyn CryptoStore,
+	network: Option<&Signer>,
+) -> Vec<(sc_network::KademliaKey, Vec<u8>)> {
+	let serialized_record =
+		serialize_authority_record(serialize_addresses(addresses.into_iter())).unwrap();
 
-	let signature = key_store
-		.sign_with(
-			key_types::AUTHORITY_DISCOVERY,
-			&public_key.clone().into(),
-			serialized_addresses.as_slice(),
-		)
-		.await
-		.unwrap()
-		.unwrap();
-
-	let mut signed_addresses = vec![];
-	schema::SignedAuthorityAddresses { addresses: serialized_addresses.clone(), signature }
-		.encode(&mut signed_addresses)
-		.unwrap();
-
-	let key = hash_authority_id(&public_key.to_raw_vec());
-	let value = signed_addresses;
-	(key, value)
+	let peer_signature = network.map(|n| sign_record_with_peer_id(&serialized_record, n).unwrap());
+	let kv_pairs = sign_record_with_authority_ids(
+		serialized_record,
+		peer_signature,
+		key_store,
+		vec![public_key.into()],
+	)
+	.await
+	.unwrap();
+	// There is always a single item in it, because we signed it with a single key
+	kv_pairs
 }
 
 #[test]
@@ -453,13 +462,14 @@ fn dont_stop_polling_dht_event_stream_after_bogus_event() {
 
 		// Make previously triggered lookup succeed.
 		let dht_event = {
-			let (key, value) = build_dht_event(
+			let kv_pairs = build_dht_event::<TestNetwork>(
 				vec![remote_multiaddr.clone()],
 				remote_public_key.clone(),
 				&remote_key_store,
+				None,
 			)
 			.await;
-			sc_network::DhtEvent::ValueFound(vec![(key, value)])
+			sc_network::DhtEvent::ValueFound(kv_pairs)
 		};
 		dht_event_tx.send(dht_event).await.expect("Channel has capacity of 1.");
 
@@ -474,97 +484,191 @@ fn dont_stop_polling_dht_event_stream_after_bogus_event() {
 	});
 }
 
+struct DhtValueFoundTester {
+	pub remote_key_store: KeyStore,
+	pub remote_authority_public: sp_core::sr25519::Public,
+	pub remote_node_key: sc_network::Keypair,
+	pub local_worker: Option<
+		Worker<
+			TestApi,
+			TestNetwork,
+			sp_runtime::generic::Block<
+				sp_runtime::generic::Header<u64, sp_runtime::traits::BlakeTwo256>,
+				substrate_test_runtime_client::runtime::Extrinsic,
+			>,
+			std::pin::Pin<Box<futures::channel::mpsc::Receiver<sc_network::DhtEvent>>>,
+		>,
+	>,
+}
+
+impl DhtValueFoundTester {
+	fn new() -> Self {
+		let remote_key_store = KeyStore::new();
+		let remote_authority_public =
+			block_on(remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
+				.unwrap();
+
+		let remote_node_key = sc_network::Keypair::generate_ed25519();
+		Self { remote_key_store, remote_authority_public, remote_node_key, local_worker: None }
+	}
+
+	fn multiaddr_with_peer_id(&self, idx: u16) -> Multiaddr {
+		let peer_id = self.remote_node_key.public().to_peer_id();
+		let address: Multiaddr =
+			format!("/ip6/2001:db8:0:0:0:0:0:{:x}/tcp/30333", idx).parse().unwrap();
+
+		address.with(multiaddr::Protocol::P2p(peer_id.into()))
+	}
+
+	fn process_value_found(
+		&mut self,
+		strict_record_validation: bool,
+		values: Vec<(sc_network::KademliaKey, Vec<u8>)>,
+	) -> Option<&HashSet<Multiaddr>> {
+		let (_dht_event_tx, dht_event_rx) = channel(1);
+		let local_test_api =
+			Arc::new(TestApi { authorities: vec![self.remote_authority_public.clone().into()] });
+		let local_network: Arc<TestNetwork> = Arc::new(Default::default());
+		let local_key_store = KeyStore::new();
+
+		let (_to_worker, from_service) = mpsc::channel(0);
+		let mut local_worker = Worker::new(
+			from_service,
+			local_test_api,
+			local_network.clone(),
+			Box::pin(dht_event_rx),
+			Role::PublishAndDiscover(Arc::new(local_key_store)),
+			None,
+			WorkerConfig { strict_record_validation, ..Default::default() },
+		);
+
+		block_on(local_worker.refill_pending_lookups_queue()).unwrap();
+		local_worker.start_new_lookups();
+
+		drop(local_worker.handle_dht_value_found_event(values));
+
+		self.local_worker = Some(local_worker);
+
+		self.local_worker
+			.as_ref()
+			.map(|w| {
+				w.addr_cache
+					.get_addresses_by_authority_id(&self.remote_authority_public.clone().into())
+			})
+			.unwrap()
+	}
+}
+
 #[test]
 fn limit_number_of_addresses_added_to_cache_per_authority() {
-	let remote_key_store = KeyStore::new();
-	let remote_public =
-		block_on(remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
-			.unwrap();
-
-	let addresses = (0..100)
-		.map(|_| {
-			let peer_id = PeerId::random();
-			let address: Multiaddr = "/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
-			address.with(multiaddr::Protocol::P2p(peer_id.into()))
-		})
-		.collect();
-
-	let dht_event = block_on(build_dht_event(addresses, remote_public.into(), &remote_key_store));
-
-	let (_dht_event_tx, dht_event_rx) = channel(1);
-
-	let (_to_worker, from_service) = mpsc::channel(0);
-	let mut worker = Worker::new(
-		from_service,
-		Arc::new(TestApi { authorities: vec![remote_public.into()] }),
-		Arc::new(TestNetwork::default()),
-		Box::pin(dht_event_rx),
-		Role::Discover,
+	let mut tester = DhtValueFoundTester::new();
+	assert!(MAX_ADDRESSES_PER_AUTHORITY < 100);
+	let addresses = (1..100).map(|i| tester.multiaddr_with_peer_id(i)).collect();
+	let kv_pairs = block_on(build_dht_event::<TestNetwork>(
+		addresses,
+		tester.remote_authority_public.clone().into(),
+		&tester.remote_key_store,
 		None,
-		Default::default(),
-	);
+	));
 
-	block_on(worker.refill_pending_lookups_queue()).unwrap();
-	worker.start_new_lookups();
+	let cached_remote_addresses = tester.process_value_found(false, kv_pairs);
+	assert_eq!(MAX_ADDRESSES_PER_AUTHORITY, cached_remote_addresses.unwrap().len());
+}
+
+#[test]
+fn strict_accept_address_with_peer_signature() {
+	let mut tester = DhtValueFoundTester::new();
+	let addr = tester.multiaddr_with_peer_id(1);
+	let kv_pairs = block_on(build_dht_event(
+		vec![addr.clone()],
+		tester.remote_authority_public.clone().into(),
+		&tester.remote_key_store,
+		Some(&tester.remote_node_key),
+	));
+
+	let cached_remote_addresses = tester.process_value_found(true, kv_pairs);
 
-	worker.handle_dht_value_found_event(vec![dht_event]).unwrap();
 	assert_eq!(
-		MAX_ADDRESSES_PER_AUTHORITY,
-		worker
-			.addr_cache
-			.get_addresses_by_authority_id(&remote_public.into())
-			.unwrap()
-			.len(),
+		Some(&HashSet::from([addr])),
+		cached_remote_addresses,
+		"Expect worker to only cache `Multiaddr`s with `PeerId`s.",
 	);
 }
 
 #[test]
-fn do_not_cache_addresses_without_peer_id() {
-	let remote_key_store = KeyStore::new();
-	let remote_public =
-		block_on(remote_key_store.sr25519_generate_new(key_types::AUTHORITY_DISCOVERY, None))
-			.unwrap();
-
-	let multiaddr_with_peer_id = {
-		let peer_id = PeerId::random();
-		let address: Multiaddr = "/ip6/2001:db8:0:0:0:0:0:2/tcp/30333".parse().unwrap();
-
-		address.with(multiaddr::Protocol::P2p(peer_id.into()))
-	};
-
-	let multiaddr_without_peer_id: Multiaddr =
-		"/ip6/2001:db8:0:0:0:0:0:1/tcp/30333".parse().unwrap();
-
-	let dht_event = block_on(build_dht_event(
-		vec![multiaddr_with_peer_id.clone(), multiaddr_without_peer_id],
-		remote_public.into(),
-		&remote_key_store,
+fn reject_address_with_rogue_peer_signature() {
+	let mut tester = DhtValueFoundTester::new();
+	let rogue_remote_node_key = sc_network::Keypair::generate_ed25519();
+	let kv_pairs = block_on(build_dht_event(
+		vec![tester.multiaddr_with_peer_id(1)],
+		tester.remote_authority_public.clone().into(),
+		&tester.remote_key_store,
+		Some(&rogue_remote_node_key),
 	));
 
-	let (_dht_event_tx, dht_event_rx) = channel(1);
-	let local_test_api = Arc::new(TestApi { authorities: vec![remote_public.into()] });
-	let local_network: Arc<TestNetwork> = Arc::new(Default::default());
-	let local_key_store = KeyStore::new();
+	let cached_remote_addresses = tester.process_value_found(false, kv_pairs);
 
-	let (_to_worker, from_service) = mpsc::channel(0);
-	let mut local_worker = Worker::new(
-		from_service,
-		local_test_api,
-		local_network.clone(),
-		Box::pin(dht_event_rx),
-		Role::PublishAndDiscover(Arc::new(local_key_store)),
-		None,
-		Default::default(),
+	assert!(
+		cached_remote_addresses.is_none(),
+		"Expected worker to ignore record signed by a different key.",
 	);
+}
 
-	block_on(local_worker.refill_pending_lookups_queue()).unwrap();
-	local_worker.start_new_lookups();
+#[test]
+fn reject_address_with_invalid_peer_signature() {
+	let mut tester = DhtValueFoundTester::new();
+	let mut kv_pairs = block_on(build_dht_event(
+		vec![tester.multiaddr_with_peer_id(1)],
+		tester.remote_authority_public.clone().into(),
+		&tester.remote_key_store,
+		Some(&tester.remote_node_key),
+	));
+	// tamper with the signature
+	let mut record = schema::SignedAuthorityRecord::decode(kv_pairs[0].1.as_slice()).unwrap();
+	record.peer_signature.as_mut().map(|p| p.signature[1] += 1);
+	record.encode(&mut kv_pairs[0].1).unwrap();
 
-	local_worker.handle_dht_value_found_event(vec![dht_event]).unwrap();
+	let cached_remote_addresses = tester.process_value_found(false, kv_pairs);
+
+	assert!(
+		cached_remote_addresses.is_none(),
+		"Expected worker to ignore record with tampered signature.",
+	);
+}
+
+#[test]
+fn reject_address_without_peer_signature() {
+	let mut tester = DhtValueFoundTester::new();
+	let kv_pairs = block_on(build_dht_event::<TestNetwork>(
+		vec![tester.multiaddr_with_peer_id(1)],
+		tester.remote_authority_public.clone().into(),
+		&tester.remote_key_store,
+		None,
+	));
+
+	let cached_remote_addresses = tester.process_value_found(true, kv_pairs);
+
+	assert!(cached_remote_addresses.is_none(), "Expected worker to ignore unsigned record.",);
+}
+
+#[test]
+fn do_not_cache_addresses_without_peer_id() {
+	let mut tester = DhtValueFoundTester::new();
+	let multiaddr_with_peer_id = tester.multiaddr_with_peer_id(1);
+	let multiaddr_without_peer_id: Multiaddr =
+		"/ip6/2001:db8:0:0:0:0:0:2/tcp/30333".parse().unwrap();
+	let kv_pairs = block_on(build_dht_event::<TestNetwork>(
+		vec![multiaddr_with_peer_id.clone(), multiaddr_without_peer_id],
+		tester.remote_authority_public.clone().into(),
+		&tester.remote_key_store,
+		None,
+	));
+
+	let cached_remote_addresses = tester.process_value_found(false, kv_pairs);
 
 	assert_eq!(
 		Some(&HashSet::from([multiaddr_with_peer_id])),
-		local_worker.addr_cache.get_addresses_by_authority_id(&remote_public.into()),
+		cached_remote_addresses,
 		"Expect worker to only cache `Multiaddr`s with `PeerId`s.",
 	);
 }
@@ -697,10 +801,14 @@ fn lookup_throttling() {
 			let remote_hash = network.get_value_call.lock().unwrap().pop().unwrap();
 			let remote_key: AuthorityId = remote_hash_to_key.get(&remote_hash).unwrap().clone();
 			let dht_event = {
-				let (key, value) =
-					build_dht_event(vec![remote_multiaddr.clone()], remote_key, &remote_key_store)
-						.await;
-				sc_network::DhtEvent::ValueFound(vec![(key, value)])
+				let kv_pairs = build_dht_event::<TestNetwork>(
+					vec![remote_multiaddr.clone()],
+					remote_key,
+					&remote_key_store,
+					None,
+				)
+				.await;
+				sc_network::DhtEvent::ValueFound(kv_pairs)
 			};
 			dht_event_tx.send(dht_event).await.expect("Channel has capacity of 1.");