Gossipsub message signing

core/src/identity.rs

@@ -29,7 +29,7 @@ pub mod secp256k1;
 pub mod error;
 
 use self::error::*;
-use crate::{PeerId, keys_proto};
+use crate::{keys_proto, PeerId};
 
 /// Identity keypair of a node.
 ///
@@ -57,7 +57,7 @@ pub enum Keypair {
     Rsa(rsa::Keypair),
     /// A Secp256k1 keypair.
     #[cfg(feature = "secp256k1")]
-    Secp256k1(secp256k1::Keypair)
+    Secp256k1(secp256k1::Keypair),
 }
 
 impl Keypair {
@@ -100,7 +100,7 @@ impl Keypair {
             #[cfg(not(target_arch = "wasm32"))]
             Rsa(ref pair) => pair.sign(msg),
             #[cfg(feature = "secp256k1")]
-            Secp256k1(ref pair) => pair.secret().sign(msg)
+            Secp256k1(ref pair) => pair.secret().sign(msg),
         }
     }
 
@@ -127,7 +127,7 @@ pub enum PublicKey {
     Rsa(rsa::PublicKey),
     #[cfg(feature = "secp256k1")]
     /// A public Secp256k1 key.
-    Secp256k1(secp256k1::PublicKey)
+    Secp256k1(secp256k1::PublicKey),
 }
 
 impl PublicKey {
@@ -142,7 +142,7 @@ impl PublicKey {
             #[cfg(not(target_arch = "wasm32"))]
             Rsa(pk) => pk.verify(msg, sig),
             #[cfg(feature = "secp256k1")]
-            Secp256k1(pk) => pk.verify(msg, sig)
+            Secp256k1(pk) => pk.verify(msg, sig),
         }
     }
 
@@ -152,27 +152,26 @@ impl PublicKey {
         use prost::Message;
 
         let public_key = match self {
-            PublicKey::Ed25519(key) =>
-                keys_proto::PublicKey {
-                    r#type: keys_proto::KeyType::Ed25519 as i32,
-                    data: key.encode().to_vec()
-                },
+            PublicKey::Ed25519(key) => keys_proto::PublicKey {
+                r#type: keys_proto::KeyType::Ed25519 as i32,
+                data: key.encode().to_vec(),
+            },
             #[cfg(not(target_arch = "wasm32"))]
-            PublicKey::Rsa(key) =>
-                keys_proto::PublicKey {
-                    r#type: keys_proto::KeyType::Rsa as i32,
-                    data: key.encode_x509()
-                },
+            PublicKey::Rsa(key) => keys_proto::PublicKey {
+                r#type: keys_proto::KeyType::Rsa as i32,
+                data: key.encode_x509(),
+            },
             #[cfg(feature = "secp256k1")]
-            PublicKey::Secp256k1(key) =>
-                keys_proto::PublicKey {
-                    r#type: keys_proto::KeyType::Secp256k1 as i32,
-                    data: key.encode().to_vec()
-                }
+            PublicKey::Secp256k1(key) => keys_proto::PublicKey {
+                r#type: keys_proto::KeyType::Secp256k1 as i32,
+                data: key.encode().to_vec(),
+            },
         };
 
         let mut buf = Vec::with_capacity(public_key.encoded_len());
-        public_key.encode(&mut buf).expect("Vec<u8> provides capacity as needed");
+        public_key
+            .encode(&mut buf)
+            .expect("Vec<u8> provides capacity as needed");
         buf
     }
 
@@ -191,7 +190,7 @@ impl PublicKey {
         match key_type {
             keys_proto::KeyType::Ed25519 => {
                 ed25519::PublicKey::decode(&pubkey.data).map(PublicKey::Ed25519)
-            },
+            }
             #[cfg(not(target_arch = "wasm32"))]
             keys_proto::KeyType::Rsa => {
                 rsa::PublicKey::decode_x509(&pubkey.data).map(PublicKey::Rsa)
@@ -200,7 +199,7 @@ impl PublicKey {
             keys_proto::KeyType::Rsa => {
                 log::debug!("support for RSA was disabled at compile-time");
                 Err(DecodingError::new("Unsupported"))
-            },
+            }
             #[cfg(feature = "secp256k1")]
             keys_proto::KeyType::Secp256k1 => {
                 secp256k1::PublicKey::decode(&pubkey.data).map(PublicKey::Secp256k1)
@@ -218,4 +217,3 @@ impl PublicKey {
         self.into()
     }
 }
-

core/src/identity/error.rs

@@ -27,16 +27,22 @@ use std::fmt;
 #[derive(Debug)]
 pub struct DecodingError {
     msg: String,
-    source: Option<Box<dyn Error + Send + Sync>>
+    source: Option<Box<dyn Error + Send + Sync>>,
 }
 
 impl DecodingError {
     pub(crate) fn new<S: ToString>(msg: S) -> Self {
-        Self { msg: msg.to_string(), source: None }
+        Self {
+            msg: msg.to_string(),
+            source: None,
+        }
     }
 
     pub(crate) fn source(self, source: impl Error + Send + Sync + 'static) -> Self {
-        Self { source: Some(Box::new(source)), .. self }
+        Self {
+            source: Some(Box::new(source)),
+            ..self
+        }
     }
 }
 
@@ -56,17 +62,23 @@ impl Error for DecodingError {
 #[derive(Debug)]
 pub struct SigningError {
     msg: String,
-    source: Option<Box<dyn Error + Send + Sync>>
+    source: Option<Box<dyn Error + Send + Sync>>,
 }
 
 /// An error during encoding of key material.
 impl SigningError {
     pub(crate) fn new<S: ToString>(msg: S) -> Self {
-        Self { msg: msg.to_string(), source: None }
+        Self {
+            msg: msg.to_string(),
+            source: None,
+        }
     }
 
     pub(crate) fn source(self, source: impl Error + Send + Sync + 'static) -> Self {
-        Self { source: Some(Box::new(source)), .. self }
+        Self {
+            source: Some(Box::new(source)),
+            ..self
+        }
     }
 }
 
@@ -81,4 +93,3 @@ impl Error for SigningError {
         self.source.as_ref().map(|s| &**s as &dyn Error)
     }
 }
-

examples/gossipsub-chat.rs

@@ -50,15 +50,15 @@ use async_std::{io, task};
 use env_logger::{Builder, Env};
 use futures::prelude::*;
 use libp2p::gossipsub::protocol::MessageId;
-use libp2p::gossipsub::{GossipsubEvent, GossipsubMessage, Topic};
-use libp2p::{
-    gossipsub, identity,
-    PeerId,
-};
+use libp2p::gossipsub::{GossipsubEvent, GossipsubMessage, Signing, Topic};
+use libp2p::{gossipsub, identity, PeerId};
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
 use std::time::Duration;
-use std::{error::Error, task::{Context, Poll}};
+use std::{
+    error::Error,
+    task::{Context, Poll},
+};
 
 fn main() -> Result<(), Box<dyn Error>> {
     Builder::from_env(Env::default().default_filter_or("info")).init();
@@ -69,7 +69,7 @@ fn main() -> Result<(), Box<dyn Error>> {
     println!("Local peer id: {:?}", local_peer_id);
 
     // Set up an encrypted TCP Transport over the Mplex and Yamux protocols
-    let transport = libp2p::build_development_transport(local_key)?;
+    let transport = libp2p::build_development_transport(local_key.clone())?;
 
     // Create a Gossipsub topic
     let topic = Topic::new("test-net".into());
@@ -93,7 +93,8 @@ fn main() -> Result<(), Box<dyn Error>> {
             //same content will be propagated.
             .build();
         // build a gossipsub network behaviour
-        let mut gossipsub = gossipsub::Gossipsub::new(local_peer_id.clone(), gossipsub_config);
+        let mut gossipsub =
+            gossipsub::Gossipsub::new(Signing::Enabled(local_key), gossipsub_config);
         gossipsub.subscribe(topic.clone());
         libp2p::Swarm::new(transport, gossipsub, local_peer_id)
     };
@@ -120,11 +121,13 @@ fn main() -> Result<(), Box<dyn Error>> {
     let mut listening = false;
     task::block_on(future::poll_fn(move |cx: &mut Context| {
         loop {
-            match stdin.try_poll_next_unpin(cx)? {
+            if let Err(e) = match stdin.try_poll_next_unpin(cx)? {
                 Poll::Ready(Some(line)) => swarm.publish(&topic, line.as_bytes()),
                 Poll::Ready(None) => panic!("Stdin closed"),
                 Poll::Pending => break,
-            };
+            } {
+                println!("Publish error: {:?}", e);
+            }
         }
 
         loop {

examples/ipfs-private.rs

@@ -35,7 +35,7 @@ use async_std::{io, task};
 use futures::{future, prelude::*};
 use libp2p::{
     core::{either::EitherTransport, transport::upgrade::Version, StreamMuxer},
-    gossipsub::{self, Gossipsub, GossipsubConfigBuilder, GossipsubEvent},
+    gossipsub::{self, Gossipsub, GossipsubConfigBuilder, GossipsubEvent, Signing},
     identify::{Identify, IdentifyEvent},
     identity,
     multiaddr::Protocol,
@@ -178,18 +178,14 @@ fn main() -> Result<(), Box<dyn Error>> {
         ping: Ping,
     }
 
-    impl NetworkBehaviourEventProcess<IdentifyEvent>
-        for MyBehaviour
-    {
+    impl NetworkBehaviourEventProcess<IdentifyEvent> for MyBehaviour {
         // Called when `identify` produces an event.
         fn inject_event(&mut self, event: IdentifyEvent) {
             println!("identify: {:?}", event);
         }
     }
 
-    impl NetworkBehaviourEventProcess<GossipsubEvent>
-        for MyBehaviour
-    {
+    impl NetworkBehaviourEventProcess<GossipsubEvent> for MyBehaviour {
         // Called when `gossipsub` produces an event.
         fn inject_event(&mut self, event: GossipsubEvent) {
             match event {
@@ -204,9 +200,7 @@ fn main() -> Result<(), Box<dyn Error>> {
         }
     }
 
-    impl NetworkBehaviourEventProcess<PingEvent>
-        for MyBehaviour
-    {
+    impl NetworkBehaviourEventProcess<PingEvent> for MyBehaviour {
         // Called when `ping` produces an event.
         fn inject_event(&mut self, event: PingEvent) {
             use ping::handler::{PingFailure, PingSuccess};
@@ -245,11 +239,11 @@ fn main() -> Result<(), Box<dyn Error>> {
 
     // Create a Swarm to manage peers and events
     let mut swarm = {
-        let gossipsub_config = GossipsubConfigBuilder::default()
+        let gossipsub_config = GossipsubConfigBuilder::new()
             .max_transmit_size(262144)
             .build();
         let mut behaviour = MyBehaviour {
-            gossipsub: Gossipsub::new(local_peer_id.clone(), gossipsub_config),
+            gossipsub: Gossipsub::new(Signing::Enabled(local_key.clone()), gossipsub_config),
             identify: Identify::new(
                 "/ipfs/0.1.0".into(),
                 "rust-ipfs-example".into(),
@@ -280,12 +274,14 @@ fn main() -> Result<(), Box<dyn Error>> {
     let mut listening = false;
     task::block_on(future::poll_fn(move |cx: &mut Context| {
         loop {
-            match stdin.try_poll_next_unpin(cx)? {
+            if let Err(e) = match stdin.try_poll_next_unpin(cx)? {
                 Poll::Ready(Some(line)) => {
-                    swarm.gossipsub.publish(&gossipsub_topic, line.as_bytes());
+                    swarm.gossipsub.publish(&gossipsub_topic, line.as_bytes())
                 }
                 Poll::Ready(None) => panic!("Stdin closed"),
                 Poll::Pending => break,
+            } {
+                println!("Publish error: {:?}", e);
             }
         }
         loop {

protocols/gossipsub/Cargo.toml

@@ -33,6 +33,7 @@ env_logger = "0.7.1"
 libp2p-plaintext = { path = "../plaintext" }
 libp2p-yamux = { path = "../../muxers/yamux" }
 quickcheck = "0.9.2"
+hex = "0.4.2"
 
 [build-dependencies]
 prost-build = "0.6"