Allow setting custom protocol in mplex

Cargo.toml

@@ -77,7 +77,7 @@ libp2p-gossipsub = { version = "0.34.0", path = "./protocols/gossipsub", optiona
 libp2p-identify = { version = "0.32.0", path = "protocols/identify", optional = true }
 libp2p-kad = { version = "0.33.0", path = "protocols/kad", optional = true }
 libp2p-metrics = { version = "0.2.0", path = "misc/metrics", optional = true }
-libp2p-mplex = { version = "0.30.0", path = "muxers/mplex", optional = true }
+libp2p-mplex = { version = "0.30.1", path = "muxers/mplex", optional = true }
 libp2p-noise = { version = "0.33.0", path = "transports/noise", optional = true }
 libp2p-ping = { version = "0.32.0", path = "protocols/ping", optional = true }
 libp2p-plaintext = { version = "0.30.0", path = "transports/plaintext", optional = true }

muxers/mplex/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.30.1 [unreleased]
+
+- Add `fn set_protocol_name(&mut self, protocol_name: &'static [u8])` to MplexConfig
+
 # 0.30.0 [2021-11-01]
 
 - Make default features of `libp2p-core` optional.

muxers/mplex/Cargo.toml

@@ -2,7 +2,7 @@
 name = "libp2p-mplex"
 edition = "2018"
 description = "Mplex multiplexing protocol for libp2p"
-version = "0.30.0"
+version = "0.30.1"
 authors = ["Parity Technologies <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"

muxers/mplex/src/config.rs

@@ -21,6 +21,8 @@
 use crate::codec::MAX_FRAME_SIZE;
 use std::cmp;
 
+pub(crate) const DEFAULT_MPLEX_PROTOCOL_NAME: &'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 name, defaults to b"/mplex/6.7.0"
+    pub(crate) protocol_name: &'static [u8],
 }
 
 impl MplexConfig {
@@ -84,6 +88,18 @@ impl MplexConfig {
         self.split_send_size = size;
         self
     }
+
+    /// Set the protocol name.
+    ///
+    /// ```rust
+    /// use libp2p_mplex::MplexConfig;
+    /// let mut muxer_config = MplexConfig::new();
+    /// muxer_config.set_protocol_name(b"/mplex/6.7.0");
+    /// ```
+    pub fn set_protocol_name(&mut self, protocol_name: &'static [u8]) -> &mut Self {
+        self.protocol_name = protocol_name;
+        self
+    }
 }
 
 /// Behaviour when the maximum length of the buffer is reached.
@@ -120,6 +136,7 @@ impl Default for MplexConfig {
             max_buffer_len: 32,
             max_buffer_behaviour: MaxBufferBehaviour::Block,
             split_send_size: 8 * 1024,
+            protocol_name: DEFAULT_MPLEX_PROTOCOL_NAME,
         }
     }
 }

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_name: crate::config::DEFAULT_MPLEX_PROTOCOL_NAME,
             }
         }
     }

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_name)
     }
 }
 

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_name(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")
+            }
+            _ => {}
+        }
+    });
+}