Reduce the spam from network misbehaviours (#1750)

* Reduce the spam from network misbehaviours

* Add protocol id to misbehaviour

substrate/core/network-libp2p/src/custom_proto/behaviour.rs

@@ -242,9 +242,14 @@ impl<TSubstream> CustomProtos<TSubstream> {
 	/// Disconnects the given peer if we are connected to it and disables it for a little while.
 	pub fn ban_peer(&mut self, peer_id: PeerId) {
 		// Peer is already banned
-		if self.banned_peers.iter().any(|(p, _)| p == &peer_id) {
-			return
+		if let Some(pos) = self.banned_peers.iter().position(|(p, _)| p == &peer_id) {
+			if self.banned_peers[pos].1 > Instant::now() {
+				return
+			} else {
+				self.banned_peers.remove(pos);
+			}
 		}
+
 		self.banned_peers.push((peer_id.clone(), Instant::now() + PEER_DISABLE_DURATION));
 		if self.enabled_peers.remove(&peer_id).is_some() {
 			self.events.push(NetworkBehaviourAction::SendEvent {
@@ -565,6 +570,11 @@ where
 					messages,
 				}));
 			}
+			CustomProtosHandlerOut::ProtocolError { protocol_id, error } => {
+				warn!(target: "sub-libp2p", "Network misbehaviour from {:?} with protocol \
+					{:?}: {:?}", source, protocol_id, error);
+				self.ban_peer(source);
+			}
 		}
 	}
 
@@ -589,7 +599,7 @@ where
 		}
 
 		// Clean up `banned_peers`
-		self.banned_peers.retain(|(_, end)| *end < Instant::now());
+		self.banned_peers.retain(|(_, end)| *end > Instant::now());
 		self.banned_peers.shrink_to_fit();
 
 		if !self.events.is_empty() {

substrate/core/network-libp2p/src/custom_proto/handler.rs

@@ -24,7 +24,7 @@ use libp2p::core::{
 	protocols_handler::ProtocolsHandlerUpgrErr,
 	upgrade::{InboundUpgrade, OutboundUpgrade}
 };
-use log::{trace, warn};
+use log::trace;
 use smallvec::SmallVec;
 use std::{fmt, io, time::Duration, time::Instant};
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -50,7 +50,7 @@ pub struct CustomProtosHandler<TSubstream> {
 	substreams: SmallVec<[RegisteredProtocolSubstream<TSubstream>; 6]>,
 
 	/// Queue of events to send to the outside.
-	events_queue: SmallVec<[ProtocolsHandlerEvent<RegisteredProtocol, (), CustomProtosHandlerOut>; 16]>,
+	events_queue: SmallVec<[ProtocolsHandlerEvent<RegisteredProtocol, ProtocolId, CustomProtosHandlerOut>; 16]>,
 }
 
 /// State of the handler.
@@ -128,6 +128,14 @@ pub enum CustomProtosHandlerOut {
 		/// Copy of the messages that are within the buffer, for further diagnostic.
 		messages: Vec<Bytes>,
 	},
+
+	/// An error has happened on the protocol level with this node.
+	ProtocolError {
+		/// Protocol for which the error happened.
+		protocol_id: ProtocolId,
+		/// The error that happened.
+		error: ProtocolsHandlerUpgrErr<io::Error>,
+	},
 }
 
 impl<TSubstream> CustomProtosHandler<TSubstream>
@@ -191,7 +199,7 @@ where
 	type Error = Void;
 	type InboundProtocol = RegisteredProtocols;
 	type OutboundProtocol = RegisteredProtocol;
-	type OutboundOpenInfo = ();
+	type OutboundOpenInfo = ProtocolId;
 
 	#[inline]
 	fn listen_protocol(&self) -> Self::InboundProtocol {
@@ -258,7 +266,7 @@ where
 
 						self.events_queue.push(ProtocolsHandlerEvent::OutboundSubstreamRequest {
 							upgrade: protocol.clone(),
-							info: (),
+							info: protocol.id(),
 						});
 					}
 				}
@@ -286,8 +294,11 @@ where
 	fn inject_inbound_closed(&mut self) {}
 
 	#[inline]
-	fn inject_dial_upgrade_error(&mut self, _: Self::OutboundOpenInfo, err: ProtocolsHandlerUpgrErr<io::Error>) {
-		warn!(target: "sub-libp2p", "Error while opening custom protocol: {:?}", err);
+	fn inject_dial_upgrade_error(&mut self, protocol_id: Self::OutboundOpenInfo, error: ProtocolsHandlerUpgrErr<io::Error>) {
+		if let State::Normal = self.state {
+			let event = CustomProtosHandlerOut::ProtocolError { protocol_id, error };
+			self.events_queue.push(ProtocolsHandlerEvent::Custom(event));
+		}
 
 		// Right now if the remote doesn't support one of the custom protocols, we shut down the
 		// entire connection. This is a hack-ish solution to the problem where we connect to nodes