Gossipsub v1.1

protocols/gossipsub/src/error.rs

@@ -21,6 +21,7 @@
 //! Error types that can result from gossipsub.
 
 use libp2p_core::identity::error::SigningError;
+use std::fmt;
 
 /// Error associated with publishing a gossipsub message.
 #[derive(Debug)]
@@ -38,3 +39,39 @@ impl From<SigningError> for PublishError {
         PublishError::SigningError(error)
     }
 }
+
+/// Errors that can occur in the protocols handler.
+#[derive(Debug)]
+pub enum GossipsubHandlerError {
+    /// The maximum number of inbound substreams created has been exceeded.
+    MaxInboundSubstreams,
+    /// The maximum number of outbound substreams created has been exceeded.
+    MaxOutboundSubstreams,
+    /// The message exceeds the maximum transmission size.
+    MaxTransmissionSize,
+    /// The message is invalid.
+    InvalidMessage(&'static str),
+    /// IO error.
+    Io(std::io::Error),
+}
+
+impl From<std::io::Error> for GossipsubHandlerError {
+    fn from(error: std::io::Error) -> GossipsubHandlerError {
+        GossipsubHandlerError::Io(error)
+    }
+}
+
+impl fmt::Display for GossipsubHandlerError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{:?}", self)
+    }
+}
+
+impl std::error::Error for GossipsubHandlerError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            GossipsubHandlerError::Io(io) => Some(io),
+            _ => None,
+        }
+    }
+}

protocols/gossipsub/src/handler.rs

@@ -20,6 +20,7 @@
 
 use crate::behaviour::GossipsubRpc;
 use crate::config::ValidationMode;
+use crate::error::GossipsubHandlerError;
 use crate::protocol::{GossipsubCodec, ProtocolConfig};
 use futures::prelude::*;
 use futures::StreamExt;
@@ -29,14 +30,21 @@ use libp2p_swarm::protocols_handler::{
     KeepAlive, ProtocolsHandler, ProtocolsHandlerEvent, ProtocolsHandlerUpgrErr, SubstreamProtocol,
 };
 use libp2p_swarm::NegotiatedSubstream;
-use log::{debug, error, trace, warn};
+use log::{error, trace, warn};
 use smallvec::SmallVec;
 use std::{
     io,
     pin::Pin,
     task::{Context, Poll},
 };
 
+// The maximum number of substreams we accept or create before disconnecting from the peer.
+//
+// Gossipsub is supposed to have a single long-lived inbound and outbound substream. On failure we
+// attempt to recreate these. This imposes an upper bound of new substreams before we consider the
+// connection faulty and disconnect. This also prevents against potential substream creation loops.
+const MAX_SUBSTREAM_CREATION: usize = 5;
+
 /// Protocol Handler that manages a single long-lived substream with a peer.
 pub struct GossipsubHandler {
     /// Upgrade configuration for the gossipsub protocol.
@@ -55,6 +63,12 @@ pub struct GossipsubHandler {
     /// requests.
     outbound_substream_establishing: bool,
 
+    /// The number of outbound substreams we have created.
+    outbound_substreams_created: usize,
+
+    /// The number of inbound substreams that have been created by the peer.
+    inbound_substreams_created: usize,
+
     /// Flag determining whether to maintain the connection to the peer.
     keep_alive: KeepAlive,
 }
@@ -99,6 +113,8 @@ impl GossipsubHandler {
             inbound_substream: None,
             outbound_substream: None,
             outbound_substream_establishing: false,
+            outbound_substreams_created: 0,
+            inbound_substreams_created: 0,
             send_queue: SmallVec::new(),
             keep_alive: KeepAlive::Yes,
         }
@@ -108,7 +124,7 @@ impl GossipsubHandler {
 impl ProtocolsHandler for GossipsubHandler {
     type InEvent = GossipsubRpc;
     type OutEvent = GossipsubRpc;
-    type Error = io::Error;
+    type Error = GossipsubHandlerError;
     type InboundProtocol = ProtocolConfig;
     type OutboundProtocol = ProtocolConfig;
     type OutboundOpenInfo = GossipsubRpc;
@@ -121,6 +137,8 @@ impl ProtocolsHandler for GossipsubHandler {
         &mut self,
         substream: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
     ) {
+        self.inbound_substreams_created += 1;
+
         // new inbound substream. Replace the current one, if it exists.
         trace!("New inbound substream request");
         self.inbound_substream = Some(InboundSubstreamState::WaitingInput(substream));
@@ -132,6 +150,8 @@ impl ProtocolsHandler for GossipsubHandler {
         message: Self::OutboundOpenInfo,
     ) {
         self.outbound_substream_establishing = false;
+        self.outbound_substreams_created += 1;
+
         // Should never establish a new outbound substream if one already exists.
         // If this happens, an outbound message is not sent.
         if self.outbound_substream.is_some() {
@@ -150,11 +170,12 @@ impl ProtocolsHandler for GossipsubHandler {
     fn inject_dial_upgrade_error(
         &mut self,
         _: Self::OutboundOpenInfo,
-        _: ProtocolsHandlerUpgrErr<
+        e: ProtocolsHandlerUpgrErr<
             <Self::OutboundProtocol as OutboundUpgrade<NegotiatedSubstream>>::Error,
         >,
     ) {
         self.outbound_substream_establishing = false;
+        warn!("Dial upgrade error {:?}", e);
         // Ignore upgrade errors for now.
         // If a peer doesn't support this protocol, this will just ignore them, but not disconnect
         // them.
@@ -175,11 +196,23 @@ impl ProtocolsHandler for GossipsubHandler {
             Self::Error,
         >,
     > {
+        if self.inbound_substreams_created > MAX_SUBSTREAM_CREATION {
+            // Too many inbound substreams have been created, end the connection.
+            return Poll::Ready(ProtocolsHandlerEvent::Close(
+                GossipsubHandlerError::MaxInboundSubstreams,
+            ));
+        }
+
         // determine if we need to create the stream
         if !self.send_queue.is_empty()
             && self.outbound_substream.is_none()
             && !self.outbound_substream_establishing
         {
+            if self.outbound_substreams_created >= MAX_SUBSTREAM_CREATION {
+                return Poll::Ready(ProtocolsHandlerEvent::Close(
+                    GossipsubHandlerError::MaxOutboundSubstreams,
+                ));
+            }
             let message = self.send_queue.remove(0);
             self.send_queue.shrink_to_fit();
             self.outbound_substream_establishing = true;
@@ -202,15 +235,21 @@ impl ProtocolsHandler for GossipsubHandler {
                                 Some(InboundSubstreamState::WaitingInput(substream));
                             return Poll::Ready(ProtocolsHandlerEvent::Custom(message));
                         }
-                        Poll::Ready(Some(Err(e))) => {
-                            match e.kind() {
-                                std::io::ErrorKind::InvalidData => {
+                        Poll::Ready(Some(Err(error))) => {
+                            match error {
+                                GossipsubHandlerError::InvalidMessage(reason) => {
                                     // Invalid message, ignore it and reset to waiting
-                                    warn!("Invalid message received. Error: {}", e);
+                                    warn!("Invalid message received. Error: {}", reason);
+                                    self.inbound_substream =
+                                        Some(InboundSubstreamState::WaitingInput(substream));
+                                }
+                                GossipsubHandlerError::MaxTransmissionSize => {
+                                    warn!("Message exceeded the maximum transmission size");
                                     self.inbound_substream =
                                         Some(InboundSubstreamState::WaitingInput(substream));
                                 }
                                 _ => {
+                                    warn!("Inbound stream error: {}", error);
                                     // More serious errors, close this side of the stream. If the
                                     // peer is still around, they will re-establish their
                                     // connection
@@ -221,6 +260,7 @@ impl ProtocolsHandler for GossipsubHandler {
                         }
                         // peer closed the stream
                         Poll::Ready(None) => {
+                            warn!("Peer closed their outbound stream");
                             self.inbound_substream =
                                 Some(InboundSubstreamState::Closing(substream));
                         }
@@ -238,7 +278,7 @@ impl ProtocolsHandler for GossipsubHandler {
                                 // Don't close the connection but just drop the inbound substream.
                                 // In case the remote has more to send, they will open up a new
                                 // substream.
-                                debug!("Inbound substream error while closing: {:?}", e);
+                                warn!("Inbound substream error while closing: {:?}", e);
                             }
 
                             self.inbound_substream = None;
@@ -290,19 +330,19 @@ impl ProtocolsHandler for GossipsubHandler {
                                     self.outbound_substream =
                                         Some(OutboundSubstreamState::PendingFlush(substream))
                                 }
+                                Err(GossipsubHandlerError::MaxTransmissionSize) => {
+                                    error!("Message exceeded the maximum transmission size and was not sent.");
+                                    self.outbound_substream =
+                                        Some(OutboundSubstreamState::WaitingOutput(substream));
+                                }
                                 Err(e) => {
-                                    if let io::ErrorKind::PermissionDenied = e.kind() {
-                                        error!("Message over the maximum transmission limit was not sent.");
-                                        self.outbound_substream =
-                                            Some(OutboundSubstreamState::WaitingOutput(substream));
-                                    } else {
-                                        return Poll::Ready(ProtocolsHandlerEvent::Close(e));
-                                    }
+                                    error!("Error sending message: {}", e);
+                                    return Poll::Ready(ProtocolsHandlerEvent::Close(e));
                                 }
                             }
                         }
                         Poll::Ready(Err(e)) => {
-                            debug!("Outbound substream error while sending output: {:?}", e);
+                            error!("Outbound substream error while sending output: {:?}", e);
                             return Poll::Ready(ProtocolsHandlerEvent::Close(e));
                         }
                         Poll::Pending => {
@@ -337,11 +377,14 @@ impl ProtocolsHandler for GossipsubHandler {
                             break;
                         }
                         Poll::Ready(Err(e)) => {
-                            debug!("Outbound substream error while closing: {:?}", e);
-                            return Poll::Ready(ProtocolsHandlerEvent::Close(io::Error::new(
-                                io::ErrorKind::BrokenPipe,
-                                "Failed to close outbound substream",
-                            )));
+                            warn!("Outbound substream error while closing: {:?}", e);
+                            return Poll::Ready(ProtocolsHandlerEvent::Close(
+                                io::Error::new(
+                                    io::ErrorKind::BrokenPipe,
+                                    "Failed to close outbound substream",
+                                )
+                                .into(),
+                            ));
                         }
                         Poll::Pending => {
                             self.outbound_substream =

protocols/gossipsub/src/protocol.rs

@@ -20,6 +20,7 @@
 
 use crate::behaviour::GossipsubRpc;
 use crate::config::ValidationMode;
+use crate::error::GossipsubHandlerError;
 use crate::rpc_proto;
 use crate::topic::TopicHash;
 use byteorder::{BigEndian, ByteOrder};
@@ -81,7 +82,7 @@ where
     TSocket: AsyncRead + AsyncWrite + Unpin + Send + 'static,
 {
     type Output = Framed<TSocket, GossipsubCodec>;
-    type Error = io::Error;
+    type Error = GossipsubHandlerError;
     type Future = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
 
     fn upgrade_inbound(self, socket: TSocket, _: Self::Info) -> Self::Future {
@@ -99,7 +100,7 @@ where
     TSocket: AsyncWrite + AsyncRead + Unpin + Send + 'static,
 {
     type Output = Framed<TSocket, GossipsubCodec>;
-    type Error = io::Error;
+    type Error = GossipsubHandlerError;
     type Future = Pin<Box<dyn Future<Output = Result<Self::Output, Self::Error>> + Send>>;
 
     fn upgrade_outbound(self, socket: TSocket, _: Self::Info) -> Self::Future {
@@ -196,7 +197,7 @@ impl GossipsubCodec {
 
 impl Encoder for GossipsubCodec {
     type Item = GossipsubRpc;
-    type Error = io::Error;
+    type Error = GossipsubHandlerError;
 
     fn encode(&mut self, item: Self::Item, dst: &mut BytesMut) -> Result<(), Self::Error> {
         // Messages
@@ -301,21 +302,29 @@ impl Encoder for GossipsubCodec {
             .expect("Buffer has sufficient capacity");
 
         // length prefix the protobuf message, ensuring the max limit is not hit
-        self.length_codec.encode(Bytes::from(buf), dst)
+        self.length_codec
+            .encode(Bytes::from(buf), dst)
+            .map_err(|_| GossipsubHandlerError::MaxTransmissionSize)
     }
 }
 
 impl Decoder for GossipsubCodec {
     type Item = GossipsubRpc;
-    type Error = io::Error;
+    type Error = GossipsubHandlerError;
 
     fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
-        let packet = match self.length_codec.decode(src)? {
+        let packet = match self.length_codec.decode(src).map_err(|e| {
+            if let std::io::ErrorKind::PermissionDenied = e.kind() {
+                GossipsubHandlerError::MaxTransmissionSize
+            } else {
+                GossipsubHandlerError::Io(e)
+            }
+        })? {
             Some(p) => p,
             None => return Ok(None),
         };
 
-        let rpc = rpc_proto::Rpc::decode(&packet[..])?;
+        let rpc = rpc_proto::Rpc::decode(&packet[..]).map_err(|e| std::io::Error::from(e))?;
 
         let mut messages = Vec::with_capacity(rpc.publish.len());
         for message in rpc.publish.into_iter() {
@@ -375,15 +384,11 @@ impl Decoder for GossipsubCodec {
             // ensure the sequence number is a u64
             let sequence_number = if verify_sequence_no {
                 let seq_no = message.seqno.ok_or_else(|| {
-                    io::Error::new(
-                        io::ErrorKind::InvalidData,
-                        "sequence number was not provided",
-                    )
+                    GossipsubHandlerError::InvalidMessage("Empty sequence number")
                 })?;
                 if seq_no.len() != 8 {
-                    return Err(io::Error::new(
-                        io::ErrorKind::InvalidData,
-                        "sequence number has an incorrect size",
+                    return Err(GossipsubHandlerError::InvalidMessage(
+                        "Invalid sequence number",
                     ));
                 }
                 Some(BigEndian::read_u64(&seq_no))
@@ -393,9 +398,8 @@ impl Decoder for GossipsubCodec {
 
             let source = if verify_source {
                 Some(
-                    PeerId::from_bytes(message.from.unwrap_or_default()).map_err(|_| {
-                        io::Error::new(io::ErrorKind::InvalidData, "Invalid Peer Id")
-                    })?,
+                    PeerId::from_bytes(message.from.unwrap_or_default())
+                        .map_err(|_| GossipsubHandlerError::InvalidMessage("Invalid Peer Id"))?,
                 )
             } else {
                 None