Allow setting custom protocol in mplex

muxers/mplex/src/config.rs

@@ -21,6 +21,8 @@
 use crate::codec::MAX_FRAME_SIZE;
 use std::cmp;
 
+pub(crate) const DEFAULT_MPLEX_PROTOCOL: &'static [u8] = b"/mplex/6.7.0";
+
 /// Configuration for the multiplexer.
 #[derive(Debug, Clone)]
 pub struct MplexConfig {
@@ -33,6 +35,8 @@ pub struct MplexConfig {
     /// When sending data, split it into frames whose maximum size is this value
     /// (max 1MByte, as per the Mplex spec).
     pub(crate) split_send_size: usize,
+    /// Protocol defaults to b"/mplex/6.7.0"
+    pub(crate) protocol: &'static [u8],
 }
 
 impl MplexConfig {
@@ -84,6 +88,17 @@ impl MplexConfig {
         self.split_send_size = size;
         self
     }
+
+    /// Gets protocol
+    pub fn protocol(&self) -> &[u8] {
+        self.protocol
+    }
+
+    /// Sets protocol, muxer.set_protocol(b"/mplex/6.7.0")
+    pub fn set_protocol(&mut self, protocol: &'static [u8]) -> &mut Self {
+        self.protocol = protocol;
+        self
+    }
 }
 
 /// Behaviour when the maximum length of the buffer is reached.
@@ -120,6 +135,7 @@ impl Default for MplexConfig {
             max_buffer_len: 32,
             max_buffer_behaviour: MaxBufferBehaviour::Block,
             split_send_size: 8 * 1024,
+            protocol: DEFAULT_MPLEX_PROTOCOL,
         }
     }
 }

muxers/mplex/src/io.rs

@@ -1126,6 +1126,7 @@ mod tests {
                 max_buffer_len: g.gen_range(1, 1000),
                 max_buffer_behaviour: MaxBufferBehaviour::arbitrary(g),
                 split_send_size: g.gen_range(1, 10000),
+                protocol: crate::config::DEFAULT_MPLEX_PROTOCOL,
             }
         }
     }

muxers/mplex/src/lib.rs

@@ -40,7 +40,7 @@ impl UpgradeInfo for MplexConfig {
     type InfoIter = iter::Once<Self::Info>;
 
     fn protocol_info(&self) -> Self::InfoIter {
-        iter::once(b"/mplex/6.7.0")
+        iter::once(self.protocol)
     }
 }
 

muxers/mplex/tests/two_peers.rs

@@ -160,3 +160,62 @@ fn client_to_server_inbound() {
         bg_thread.await;
     });
 }
+
+#[test]
+fn protocol_not_match() {
+    let (tx, rx) = oneshot::channel();
+
+    let _bg_thread = async_std::task::spawn(async move {
+        let mplex = libp2p_mplex::MplexConfig::new();
+
+        let transport = TcpConfig::new()
+            .and_then(move |c, e| upgrade::apply(c, mplex, e, upgrade::Version::V1));
+
+        let mut listener = transport
+            .listen_on("/ip4/127.0.0.1/tcp/0".parse().unwrap())
+            .unwrap();
+
+        let addr = listener
+            .next()
+            .await
+            .expect("some event")
+            .expect("no error")
+            .into_new_address()
+            .expect("listen address");
+
+        tx.send(addr).unwrap();
+
+        let client = listener
+            .next()
+            .await
+            .unwrap()
+            .unwrap()
+            .into_upgrade()
+            .unwrap()
+            .0
+            .await
+            .unwrap();
+
+        let mut outbound = muxing::outbound_from_ref_and_wrap(Arc::new(client))
+            .await
+            .unwrap();
+
+        let mut buf = Vec::new();
+        outbound.read_to_end(&mut buf).await.unwrap();
+        assert_eq!(buf, b"hello world");
+    });
+
+    async_std::task::block_on(async {
+        // Make sure they do not connect when protocols do not match
+        let mut mplex = libp2p_mplex::MplexConfig::new();
+        mplex.set_protocol(b"/mplextest/1.0.0");
+        let transport = TcpConfig::new()
+            .and_then(move |c, e| upgrade::apply(c, mplex, e, upgrade::Version::V1));
+        match transport.dial(rx.await.unwrap()).unwrap().await {
+            Ok(_) => {
+                assert!(false, "Dialing should fail here as protocols do not match")
+            }
+            _ => {}
+        }
+    });
+}