diff --git a/substrate/core/network-libp2p/src/custom_proto/behaviour.rs b/substrate/core/network-libp2p/src/custom_proto/behaviour.rs index 66af5e048c560f98b745d0be9b6fbf2daa1df03b..92bc725afeabd2e4e2f6e7e75ff393fd411d7b82 100644 --- a/substrate/core/network-libp2p/src/custom_proto/behaviour.rs +++ b/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() { diff --git a/substrate/core/network-libp2p/src/custom_proto/handler.rs b/substrate/core/network-libp2p/src/custom_proto/handler.rs index 51ff2408ad86c6ac3fcbdccd5cea6615cda4770c..b44407ff7f7f898b841612a6db61d6bbda3bc203 100644 --- a/substrate/core/network-libp2p/src/custom_proto/handler.rs +++ b/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