Merge remote-tracking branch 'upstream/master' into no-secp256k1

.github/dependabot.yml

@@ -4,6 +4,7 @@ updates:
     directory: "/"
     schedule:
       interval: "daily"
+    open-pull-requests-limit: 9999
 
   - package-ecosystem: "github-actions"
     directory: "/"

CHANGELOG.md

@@ -40,7 +40,12 @@
 
 # `libp2p` facade crate
 
-## Version 0.39.0 [unreleased]
+## Version 0.39.1 [2021-07-12]
+
+- Update individual crates.
+    - `libp2p-swarm-derive`
+
+## Version 0.39.0 [2021-07-12]
 
 - Update individual crates.
     - `libp2p-core`

Cargo.toml

@@ -2,7 +2,7 @@
 name = "libp2p"
 edition = "2018"
 description = "Peer-to-peer networking library"
-version = "0.39.0"
+version = "0.39.1"
 authors = ["Parity Technologies <admin@parity.io>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"
@@ -55,11 +55,11 @@ libp2p-pnet = { version = "0.21.0", path = "transports/pnet", optional = true }
 libp2p-relay = { version = "0.3.0", path = "protocols/relay", optional = true }
 libp2p-request-response = { version = "0.12.0", path = "protocols/request-response", optional = true }
 libp2p-swarm = { version = "0.30.0", path = "swarm" }
-libp2p-swarm-derive = { version = "0.23.0", path = "swarm-derive" }
+libp2p-swarm-derive = { version = "0.24.0", path = "swarm-derive" }
 libp2p-uds = { version = "0.29.0", path = "transports/uds", optional = true }
 libp2p-wasm-ext = { version = "0.29.0", path = "transports/wasm-ext", default-features = false, optional = true }
 libp2p-yamux = { version = "0.33.0", path = "muxers/yamux", optional = true }
-multiaddr = { version = "0.12.0" }
+multiaddr = { version = "0.13.0" }
 parking_lot = "0.11.0"
 pin-project = "1.0.0"
 smallvec = "1.6.1"
@@ -74,7 +74,7 @@ libp2p-websocket = { version = "0.30.0", path = "transports/websocket", optional
 
 [dev-dependencies]
 async-std = { version = "1.6.2", features = ["attributes"] }
-env_logger = "0.8.1"
+env_logger = "0.9.0"
 tokio = { version = "1.0.1", features = ["io-util", "io-std", "macros", "rt", "rt-multi-thread"] }
 
 [dev-dependencies.libp2p]

core/CHANGELOG.md

@@ -1,11 +1,17 @@
-# 0.29.0 [unreleased]
+# 0.29.0 [2021-07-12]
 
 - Switch from `parity-multiaddr` to upstream `multiaddr`.
 
 - Update dependencies.
 
 - Implement `Keypair::from_protobuf_encoding` for ed25519 keys (see [PR 2090]).
 
+- Deprecate `upgrade::write_one`.
+  Deprecate `upgrade::write_with_len_prefix`.
+  Deprecate `upgrade::read_one`.
+  Introduce `upgrade::read_length_prefixed` and `upgrade::write_length_prefixed`.
+  See [PR 2111](https://github.com/libp2p/rust-libp2p/pull/2111).
+
 [PR 2090]: https://github.com/libp2p/rust-libp2p/pull/2090
 
 # 0.28.3 [2021-04-26]

core/Cargo.toml

@@ -20,12 +20,12 @@ futures-timer = "3"
 lazy_static = "1.2"
 libsecp256k1 = { version = "0.5.0", optional = true }
 log = "0.4"
-multiaddr = { version = "0.12.0" }
-multihash = { version = "0.13", default-features = false, features = ["std", "multihash-impl", "identity", "sha2"] }
+multiaddr = { version = "0.13.0" }
+multihash = { version = "0.14", default-features = false, features = ["std", "multihash-impl", "identity", "sha2"] }
 multistream-select = { version = "0.10", path = "../misc/multistream-select" }
 parking_lot = "0.11.0"
 pin-project = "1.0.0"
-prost = "0.7"
+prost = "0.8"
 rand = "0.7"
 rw-stream-sink = "0.2.0"
 sha2 = "0.9.1"
@@ -45,12 +45,12 @@ criterion = "0.3"
 libp2p-mplex = { path = "../muxers/mplex" }
 libp2p-noise = { path = "../transports/noise" }
 libp2p-tcp = { path = "../transports/tcp" }
-multihash = { version = "0.13", default-features = false, features = ["arb"] }
+multihash = { version = "0.14", default-features = false, features = ["arb"] }
 quickcheck = "0.9.0"
 wasm-timer = "0.2"
 
 [build-dependencies]
-prost-build = "0.7"
+prost-build = "0.8"
 
 [features]
 default = ["secp256k1"]

core/src/peer_id.rs

@@ -68,8 +68,8 @@ impl PeerId {
 
     /// Parses a `PeerId` from bytes.
     pub fn from_bytes(data: &[u8]) -> Result<PeerId, Error> {
-        Ok(PeerId::from_multihash(Multihash::from_bytes(&data)?)
-            .map_err(|mh| Error::UnsupportedCode(mh.code()))?)
+        PeerId::from_multihash(Multihash::from_bytes(&data)?)
+            .map_err(|mh| Error::UnsupportedCode(mh.code()))
     }
 
     /// Tries to turn a `Multihash` into a `PeerId`.

core/src/upgrade.rs

@@ -80,8 +80,10 @@ pub use self::{
     map::{MapInboundUpgrade, MapOutboundUpgrade, MapInboundUpgradeErr, MapOutboundUpgradeErr},
     optional::OptionalUpgrade,
     select::SelectUpgrade,
-    transfer::{write_one, write_with_len_prefix, write_varint, read_one, ReadOneError, read_varint},
+    transfer::{write_length_prefixed, write_varint, read_length_prefixed, read_varint},
 };
+#[allow(deprecated)]
+pub use self::transfer::ReadOneError;
 
 /// Types serving as protocol names.
 ///

core/src/upgrade/from_fn.rs

@@ -28,18 +28,20 @@ use std::iter;
 /// # Example
 ///
 /// ```
-/// # use libp2p_core::transport::{Transport, MemoryTransport};
-/// # use libp2p_core::upgrade;
+/// # use libp2p_core::transport::{Transport, MemoryTransport, memory::Channel};
+/// # use libp2p_core::{upgrade, Negotiated};
 /// # use std::io;
+/// # use futures::AsyncWriteExt;
 /// let _transport = MemoryTransport::default()
 ///     .and_then(move |out, cp| {
-///         upgrade::apply(out, upgrade::from_fn("/foo/1", move |mut sock, endpoint| async move {
+///         upgrade::apply(out, upgrade::from_fn("/foo/1", move |mut sock: Negotiated<Channel<Vec<u8>>>, endpoint| async move {
 ///             if endpoint.is_dialer() {
-///                 upgrade::write_one(&mut sock, "some handshake data").await?;
+///                 upgrade::write_length_prefixed(&mut sock, "some handshake data").await?;
+///                 sock.close().await?;
 ///             } else {
-///                 let handshake_data = upgrade::read_one(&mut sock, 1024).await?;
+///                 let handshake_data = upgrade::read_length_prefixed(&mut sock, 1024).await?;
 ///                 if handshake_data != b"some handshake data" {
-///                     return Err(upgrade::ReadOneError::from(io::Error::from(io::ErrorKind::Other)));
+///                     return Err(io::Error::new(io::ErrorKind::Other, "bad handshake"));
 ///                 }
 ///             }
 ///             Ok(sock)

core/src/upgrade/transfer.rs

@@ -25,10 +25,27 @@ use std::{error, fmt, io};
 
 // TODO: these methods could be on an Ext trait to AsyncWrite
 
+/// Writes a message to the given socket with a length prefix appended to it. Also flushes the socket.
+///
+/// > **Note**: Prepends a variable-length prefix indicate the length of the message. This is
+/// >           compatible with what [`read_length_prefixed`] expects.
+pub async fn write_length_prefixed(socket: &mut (impl AsyncWrite + Unpin), data: impl AsRef<[u8]>)
+    -> Result<(), io::Error>
+{
+    write_varint(socket, data.as_ref().len()).await?;
+    socket.write_all(data.as_ref()).await?;
+    socket.flush().await?;
+
+    Ok(())
+}
+
 /// Send a message to the given socket, then shuts down the writing side.
 ///
 /// > **Note**: Prepends a variable-length prefix indicate the length of the message. This is
 /// >           compatible with what `read_one` expects.
+///
+#[deprecated(since = "0.29.0", note = "Use `write_length_prefixed` instead. You will need to manually close the stream using `socket.close().await`.")]
+#[allow(dead_code)]
 pub async fn write_one(socket: &mut (impl AsyncWrite + Unpin), data: impl AsRef<[u8]>)
     -> Result<(), io::Error>
 {
@@ -42,6 +59,8 @@ pub async fn write_one(socket: &mut (impl AsyncWrite + Unpin), data: impl AsRef<
 ///
 /// > **Note**: Prepends a variable-length prefix indicate the length of the message. This is
 /// >           compatible with what `read_one` expects.
+#[deprecated(since = "0.29.0", note = "Use `write_length_prefixed` instead.")]
+#[allow(dead_code)]
 pub async fn write_with_len_prefix(socket: &mut (impl AsyncWrite + Unpin), data: impl AsRef<[u8]>)
     -> Result<(), io::Error>
 {
@@ -60,6 +79,7 @@ pub async fn write_varint(socket: &mut (impl AsyncWrite + Unpin), len: usize)
     let mut len_data = unsigned_varint::encode::usize_buffer();
     let encoded_len = unsigned_varint::encode::usize(len, &mut len_data).len();
     socket.write_all(&len_data[..encoded_len]).await?;
+
     Ok(())
 }
 
@@ -106,6 +126,27 @@ pub async fn read_varint(socket: &mut (impl AsyncRead + Unpin)) -> Result<usize,
     }
 }
 
+/// Reads a length-prefixed message from the given socket.
+///
+/// The `max_size` parameter is the maximum size in bytes of the message that we accept. This is
+/// necessary in order to avoid DoS attacks where the remote sends us a message of several
+/// gigabytes.
+///
+/// > **Note**: Assumes that a variable-length prefix indicates the length of the message. This is
+/// >           compatible with what [`write_length_prefixed`] does.
+pub async fn read_length_prefixed(socket: &mut (impl AsyncRead + Unpin), max_size: usize) -> io::Result<Vec<u8>>
+{
+    let len = read_varint(socket).await?;
+    if len > max_size {
+        return Err(io::Error::new(io::ErrorKind::InvalidData, format!("Received data size ({} bytes) exceeds maximum ({} bytes)", len, max_size)))
+    }
+
+    let mut buf = vec![0; len];
+    socket.read_exact(&mut buf).await?;
+
+    Ok(buf)
+}
+
 /// Reads a length-prefixed message from the given socket.
 ///
 /// The `max_size` parameter is the maximum size in bytes of the message that we accept. This is
@@ -114,6 +155,8 @@ pub async fn read_varint(socket: &mut (impl AsyncRead + Unpin)) -> Result<usize,
 ///
 /// > **Note**: Assumes that a variable-length prefix indicates the length of the message. This is
 /// >           compatible with what `write_one` does.
+#[deprecated(since = "0.29.0", note = "Use `read_length_prefixed` instead.")]
+#[allow(dead_code, deprecated)]
 pub async fn read_one(socket: &mut (impl AsyncRead + Unpin), max_size: usize)
     -> Result<Vec<u8>, ReadOneError>
 {
@@ -132,6 +175,7 @@ pub async fn read_one(socket: &mut (impl AsyncRead + Unpin), max_size: usize)
 
 /// Error while reading one message.
 #[derive(Debug)]
+#[deprecated(since = "0.29.0", note = "Use `read_length_prefixed` instead of `read_one` to avoid depending on this type.")]
 pub enum ReadOneError {
     /// Error on the socket.
     Io(std::io::Error),
@@ -144,12 +188,14 @@ pub enum ReadOneError {
     },
 }
 
+#[allow(deprecated)]
 impl From<std::io::Error> for ReadOneError {
     fn from(err: std::io::Error) -> ReadOneError {
         ReadOneError::Io(err)
     }
 }
 
+#[allow(deprecated)]
 impl fmt::Display for ReadOneError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match *self {
@@ -159,6 +205,7 @@ impl fmt::Display for ReadOneError {
     }
 }
 
+#[allow(deprecated)]
 impl error::Error for ReadOneError {
     fn source(&self) -> Option<&(dyn error::Error + 'static)> {
         match *self {
@@ -173,15 +220,18 @@ mod tests {
     use super::*;
 
     #[test]
-    fn write_one_works() {
+    fn write_length_prefixed_works() {
         let data = (0..rand::random::<usize>() % 10_000)
             .map(|_| rand::random::<u8>())
             .collect::<Vec<_>>();
-
         let mut out = vec![0; 10_000];
-        futures::executor::block_on(
-            write_one(&mut futures::io::Cursor::new(&mut out[..]), data.clone())
-        ).unwrap();
+
+        futures::executor::block_on(async {
+            let mut socket = futures::io::Cursor::new(&mut out[..]);
+
+            write_length_prefixed(&mut socket, &data).await.unwrap();
+            socket.close().await.unwrap();
+        });
 
         let (out_len, out_data) = unsigned_varint::decode::usize(&out).unwrap();
         assert_eq!(out_len, data.len());

examples/chat-tokio.rs

@@ -156,8 +156,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
                 swarm.behaviour_mut().floodsub.publish(floodsub_topic.clone(), line.as_bytes());
             }
             event = swarm.select_next_some() => {
-                if let SwarmEvent::NewListenAddr(addr) = event {
-                    println!("Listening on {:?}", addr);
+                if let SwarmEvent::NewListenAddr { address, .. } = event {
+                    println!("Listening on {:?}", address);
                 }
             }
         }

examples/chat.rs

@@ -160,8 +160,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(event)) => {
-                    if let SwarmEvent::NewListenAddr(addr) = event {
-                        println!("Listening on {:?}", addr);
+                    if let SwarmEvent::NewListenAddr { address, .. } = event {
+                        println!("Listening on {:?}", address);
                     }
                 }
                 Poll::Ready(None) => return Poll::Ready(Ok(())),

examples/distributed-key-value-store.rs

@@ -98,7 +98,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
         // Called when `kademlia` produces an event.
         fn inject_event(&mut self, message: KademliaEvent) {
             match message {
-                KademliaEvent::QueryResult { result, .. } => match result {
+                KademliaEvent::OutboundQueryCompleted { result, .. } => match result {
                     QueryResult::GetProviders(Ok(ok)) => {
                         for peer in ok.providers {
                             println!(
@@ -175,8 +175,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(event)) => {
-                    if let SwarmEvent::NewListenAddr(addr) = event {
-                        println!("Listening on {:?}", addr);
+                    if let SwarmEvent::NewListenAddr { address, .. } = event {
+                        println!("Listening on {:?}", address);
                     }
                 }
                 Poll::Ready(None) => return Poll::Ready(Ok(())),

examples/gossipsub-chat.rs

@@ -162,8 +162,8 @@ async fn main() -> Result<(), Box<dyn Error>> {
                         id,
                         peer_id
                     ),
-                    SwarmEvent::NewListenAddr(addr) => {
-                        println!("Listening on {:?}", addr);
+                    SwarmEvent::NewListenAddr { address, .. } => {
+                        println!("Listening on {:?}", address);
                     }
                     _ => {}
                 },

examples/ipfs-kad.rs

@@ -93,7 +93,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
     task::block_on(async move {
         loop {
             let event = swarm.select_next_some().await;
-            if let SwarmEvent::Behaviour(KademliaEvent::QueryResult {
+            if let SwarmEvent::Behaviour(KademliaEvent::OutboundQueryCompleted {
                 result: QueryResult::GetClosestPeers(result),
                 ..
             }) = event {

examples/ipfs-private.rs

@@ -287,8 +287,8 @@ fn main() -> Result<(), Box<dyn Error>> {
         loop {
             match swarm.poll_next_unpin(cx) {
                 Poll::Ready(Some(event)) => {
-                    if let SwarmEvent::NewListenAddr(addr) = event {
-                        println!("Listening on {:?}", addr);
+                    if let SwarmEvent::NewListenAddr { address, .. } = event {
+                        println!("Listening on {:?}", address);
                     }
                 }
                 Poll::Ready(None) => return Poll::Ready(Ok(())),

examples/ping.rs

@@ -79,7 +79,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     block_on(future::poll_fn(move |cx| loop {
         match swarm.poll_next_unpin(cx) {
             Poll::Ready(Some(event)) => match event {
-                SwarmEvent::NewListenAddr(addr) => println!("Listening on {:?}", addr),
+                SwarmEvent::NewListenAddr{ address, .. } => println!("Listening on {:?}", address),
                 SwarmEvent::Behaviour(event) => println!("{:?}", event),
                 _ => {}
             },

misc/multistream-select/Cargo.toml

@@ -19,7 +19,7 @@ unsigned-varint = "0.7"
 
 [dev-dependencies]
 async-std = "1.6.2"
-env_logger = "0.8"
+env_logger = "0.9"
 libp2p-core = { path = "../../core" }
 libp2p-mplex = { path = "../../muxers/mplex" }
 libp2p-plaintext = { path = "../../transports/plaintext" }

muxers/mplex/CHANGELOG.md

@@ -1,7 +1,11 @@
-# 0.29.0 [unreleased]
+# 0.29.0 [2021-07-12]
 
 - Update dependencies.
 
+- Support stream IDs of up to 60 bit length. See [PR 2094] for details.
+
+[PR 2094]: https://github.com/libp2p/rust-libp2p/pull/2094
+
 # 0.28.0 [2021-03-17]
 
 - Update dependencies.