Merge pull request #142 from ackintosh/libp2p-0510

Upgrade libp2p to 0.51

Cargo.toml

@@ -15,7 +15,8 @@ delay_map = "0.3.0"
 discv5 = "0.2.2"
 futures = "0.3.28"
 hex = "0.4.3"
-libp2p = { version = "0.50.0", default-features = false, features = ["dns", "tokio", "ecdsa", "noise", "mplex", "secp256k1", "tcp", "yamux"] }
+libp2p = { version = "0.51.0", default-features = false, features = ["dns", "tokio", "ecdsa", "noise", "macros", "mplex", "secp256k1", "tcp", "yamux"] }
+#libp2p-identity = { version = "0.2.1", features = ["secp256k1"] }
 lru = "0.10.1"
 parking_lot = "0.12.1"
 primitive-types = "0.12.1"
@@ -60,3 +61,5 @@ mev-rs = { git = "https://github.com/ralexstokes//mev-rs", rev = "7813d4a4a564e0
 ethereum-consensus = { git = "https://github.com/ralexstokes//ethereum-consensus", rev = "9b0ee0a8a45b968c8df5e7e64ea1c094e16f053d" }
 [patch."https://github.com/ralexstokes/ssz-rs"]
 ssz-rs = { git = "https://github.com/ralexstokes//ssz-rs", rev = "adf1a0b14cef90b9536f28ef89da1fab316465e1" }
+[patch."https://github.com/ralexstokes/beacon-api-client"]
+beacon-api-client = { git = "https://github.com/ralexstokes//beacon-api-client", rev = "30679e9e25d61731cde54e14cd8a3688a39d8e5b" }

src/bootstrap.rs

@@ -4,7 +4,7 @@ use beacon_chain::BeaconChainTypes;
 use discv5::Enr;
 use libp2p::core::muxing::StreamMuxerBox;
 use libp2p::identity::Keypair;
-use libp2p::{noise, PeerId, Transport};
+use libp2p::{noise, yamux, PeerId, Transport};
 use parking_lot::RwLock;
 use std::process::exit;
 use std::sync::Arc;
@@ -22,17 +22,13 @@ pub(crate) async fn build_network_transport(
 
     // Ref: Why are we using Noise?
     // https://github.com/ethereum/consensus-specs/blob/dev/specs/phase0/p2p-interface.md#why-are-we-using-noise
-    let noise_keys = noise::Keypair::<noise::X25519Spec>::new()
-        .into_authentic(&key_pair)
-        .expect("Signing libp2p-noise static DH keypair failed.");
+    let noise_config = noise::Config::new(&key_pair).expect("noise config");
+    let yamux_config = yamux::Config::default();
 
     transport
         .upgrade(libp2p::core::upgrade::Version::V1)
-        .authenticate(noise::NoiseConfig::xx(noise_keys).into_authenticated())
-        .multiplex(libp2p::core::upgrade::SelectUpgrade::new(
-            libp2p::yamux::YamuxConfig::default(),
-            libp2p::mplex::MplexConfig::default(),
-        ))
+        .authenticate(noise_config)
+        .multiplex(yamux_config)
         .timeout(std::time::Duration::from_secs(20))
         .boxed()
 }

src/discovery/behaviour.rs

@@ -1,13 +1,13 @@
 use crate::discovery::DiscoveryEvent;
+use crate::identity::peer_id_to_node_id;
 use discv5::enr::{CombinedKey, NodeId};
 use discv5::{Discv5, Discv5ConfigBuilder, Discv5Event, Enr, QueryError};
 use futures::stream::FuturesUnordered;
 use futures::{Future, FutureExt, StreamExt};
-use libp2p::core::connection::ConnectionId;
 use libp2p::swarm::dummy::ConnectionHandler as DummyConnectionHandler;
 use libp2p::swarm::{
-    ConnectionHandler, IntoConnectionHandler, NetworkBehaviour, NetworkBehaviourAction,
-    PollParameters,
+    ConnectionId, DialError, DialFailure, FromSwarm, NetworkBehaviour, PollParameters,
+    THandlerInEvent, THandlerOutEvent, ToSwarm,
 };
 use libp2p::{Multiaddr, PeerId};
 use lru::LruCache;
@@ -16,7 +16,7 @@ use std::num::NonZeroUsize;
 use std::task::{Context, Poll};
 use std::time::Duration;
 use tokio::sync::mpsc::Receiver;
-use tracing::{error, info, trace, warn};
+use tracing::{debug, error, info, trace, warn};
 
 // ////////////////////////////////////////////////////////
 // Internal message of Discovery module
@@ -98,6 +98,35 @@ impl Behaviour {
         );
         self.active_queries.push(Box::pin(query_future));
     }
+
+    fn on_dial_failure(&self, peer_id: Option<PeerId>, dial_error: &DialError) {
+        if let Some(peer_id) = peer_id {
+            match dial_error {
+                DialError::Banned
+                | DialError::LocalPeerId { .. }
+                | DialError::NoAddresses
+                | DialError::InvalidPeerId(_)
+                | DialError::WrongPeerId { .. }
+                | DialError::Denied { .. }
+                | DialError::Transport(_) => {
+                    debug!("[{peer_id}] Marking peer disconnected in DHT. error: {dial_error}");
+                    match peer_id_to_node_id(&peer_id) {
+                        Ok(node_id) => {
+                            let _ = self.discv5.disconnect_node(&node_id);
+                        }
+                        Err(e) => {
+                            warn!(
+                                "[{peer_id}] Failed to convert from PeerId to NodeId. error: {e}"
+                            );
+                        }
+                    }
+                }
+                DialError::Aborted
+                | DialError::ConnectionLimit(_)
+                | DialError::DialPeerConditionFalse(_) => {}
+            }
+        }
+    }
 }
 
 // ************************************************
@@ -160,22 +189,46 @@ impl NetworkBehaviour for Behaviour {
         }
     }
 
-    fn inject_event(
+    fn on_connection_handler_event(
         &mut self,
         _peer_id: PeerId,
-        _connection: ConnectionId,
-        _event: <<Self::ConnectionHandler as IntoConnectionHandler>::Handler as ConnectionHandler>::OutEvent,
+        _connection_id: ConnectionId,
+        _event: THandlerOutEvent<Self>,
     ) {
-        trace!("inject_event -> nothing to do");
-        // SEE https://github.com/sigp/lighthouse/blob/73ec29c267f057e70e89856403060c4c35b5c0c8/beacon_node/eth2_libp2p/src/discovery/mod.rs#L948-L954
+        // Nothing to do.
+    }
+
+    fn on_swarm_event(&mut self, event: FromSwarm<Self::ConnectionHandler>) {
+        match event {
+            FromSwarm::DialFailure(DialFailure {
+                peer_id,
+                error,
+                connection_id: _,
+            }) => {
+                self.on_dial_failure(peer_id, error);
+            }
+            FromSwarm::ConnectionEstablished(_)
+            | FromSwarm::ConnectionClosed(_)
+            | FromSwarm::AddressChange(_)
+            | FromSwarm::ListenFailure(_)
+            | FromSwarm::NewListener(_)
+            | FromSwarm::NewListenAddr(_)
+            | FromSwarm::ExpiredListenAddr(_)
+            | FromSwarm::ListenerError(_)
+            | FromSwarm::ListenerClosed(_)
+            | FromSwarm::NewExternalAddr(_)
+            | FromSwarm::ExpiredExternalAddr(_) => {
+                // Ignore events not relevant to discovery
+            }
+        }
     }
 
     #[allow(clippy::single_match)]
     fn poll(
         &mut self,
         cx: &mut Context<'_>,
         _params: &mut impl PollParameters,
-    ) -> Poll<NetworkBehaviourAction<Self::OutEvent, Self::ConnectionHandler>> {
+    ) -> Poll<ToSwarm<Self::OutEvent, THandlerInEvent<Self>>> {
         trace!("poll");
 
         if let Poll::Ready(Some(query_result)) = self.active_queries.poll_next_unpin(cx) {
@@ -199,9 +252,7 @@ impl NetworkBehaviour for Behaviour {
                         self.cached_enrs.put(*p, e.clone());
                     }
 
-                    Poll::Ready(NetworkBehaviourAction::GenerateEvent(
-                        DiscoveryEvent::FoundPeers(peers),
-                    ))
+                    Poll::Ready(ToSwarm::GenerateEvent(DiscoveryEvent::FoundPeers(peers)))
                 }
                 Err(query_error) => {
                     error!("Discovery query failed: {}", query_error);

src/identity.rs

@@ -1,6 +1,5 @@
 use discv5::enr::CombinedPublicKey;
 use discv5::Enr;
-use libp2p::identity::PublicKey;
 use libp2p::multiaddr::Protocol;
 use libp2p::{Multiaddr, PeerId};
 use tiny_keccak::{Hasher, Keccak};
@@ -10,11 +9,10 @@ pub(crate) fn enr_to_peer_id(enr: &Enr) -> PeerId {
     match enr.public_key() {
         CombinedPublicKey::Secp256k1(pk) => {
             let pk_bytes = pk.to_bytes();
-            let libp2p_pk = libp2p::core::PublicKey::Secp256k1(
-                libp2p::core::identity::secp256k1::PublicKey::decode(&pk_bytes)
-                    .expect("valid public key"),
-            );
-            PeerId::from(libp2p_pk)
+            let libp2p_pk = libp2p::identity::secp256k1::PublicKey::try_from_bytes(&pk_bytes)
+                .expect("valid public key");
+            let public_key: libp2p::identity::PublicKey = libp2p_pk.into();
+            PeerId::from(public_key)
         }
         CombinedPublicKey::Ed25519(_) => unreachable!(), // not implemented as the ENR key is generated with secp256k1
     }
@@ -26,32 +24,34 @@ pub(crate) fn peer_id_to_node_id(peer_id: &PeerId) -> Result<discv5::enr::NodeId
     // if generated from a PublicKey with Identity multihash.
     let pk_bytes = &peer_id.to_bytes()[2..];
 
-    match PublicKey::from_protobuf_encoding(pk_bytes).map_err(|e| {
+    let public_key = libp2p::identity::PublicKey::try_decode_protobuf(pk_bytes).map_err(|e| {
         format!(
             " Cannot parse libp2p public key public key from peer id: {}",
             e
         )
-    })? {
-        PublicKey::Secp256k1(pk) => {
-            let uncompressed_key_bytes = &pk.encode_uncompressed()[1..];
-            let mut output = [0_u8; 32];
-            let mut hasher = Keccak::v256();
-            hasher.update(uncompressed_key_bytes);
-            hasher.finalize(&mut output);
-            Ok(discv5::enr::NodeId::parse(&output).expect("Must be correct length"))
-        }
-        PublicKey::Ed25519(pk) => {
-            let uncompressed_key_bytes = pk.encode();
-            let mut output = [0_u8; 32];
-            let mut hasher = Keccak::v256();
-            hasher.update(&uncompressed_key_bytes);
-            hasher.finalize(&mut output);
-            Ok(discv5::enr::NodeId::parse(&output).expect("Must be correct length"))
-        }
-        PublicKey::Ecdsa(_) => Err(format!(
+    })?;
+
+    // TODO: matching public_key.key_type() after libp2p upgrading.
+    // ref: https://github.com/sigp/lighthouse/blob/8dff926c70b7c7558e7c41316770d22608dbba4c/beacon_node/lighthouse_network/src/discovery/enr_ext.rs#L266-L293
+    if let Ok(pk) = public_key.clone().try_into_secp256k1() {
+        let uncompressed_key_bytes = &pk.to_bytes_uncompressed()[1..];
+        let mut output = [0_u8; 32];
+        let mut hasher = Keccak::v256();
+        hasher.update(uncompressed_key_bytes);
+        hasher.finalize(&mut output);
+        Ok(discv5::enr::NodeId::parse(&output).expect("Must be correct length"))
+    } else if let Ok(pk) = public_key.clone().try_into_ed25519() {
+        let uncompressed_key_bytes = pk.to_bytes();
+        let mut output = [0_u8; 32];
+        let mut hasher = Keccak::v256();
+        hasher.update(&uncompressed_key_bytes);
+        hasher.finalize(&mut output);
+        Ok(discv5::enr::NodeId::parse(&output).expect("Must be correct length"))
+    } else {
+        Err(format!(
             "Unsupported public key (Ecdsa) from peer {}",
             peer_id
-        )),
+        ))
     }
 }
 

src/main.rs

@@ -22,7 +22,7 @@ use client::ClientBuilder;
 use discv5::enr::{CombinedKey, EnrBuilder};
 use environment::{EnvironmentBuilder, LoggerConfig};
 use eth2_network_config::Eth2NetworkConfig;
-use libp2p::identity::Keypair;
+// use libp2p::identity::Keypair;
 use parking_lot::RwLock;
 use std::sync::Arc;
 use tracing::info;
@@ -36,15 +36,16 @@ fn main() {
 
     // generate private key
     let enr_key = CombinedKey::generate_secp256k1();
-    let key_pair = {
+    let key_pair: libp2p::identity::Keypair = {
         match enr_key {
             CombinedKey::Secp256k1(ref key) => {
                 let mut key_bytes = key.to_bytes();
                 let secret_key =
-                    libp2p::core::identity::secp256k1::SecretKey::from_bytes(&mut key_bytes)
+                    libp2p::identity::secp256k1::SecretKey::try_from_bytes(&mut key_bytes)
                         .expect("valid secp256k1 key");
-                let kp: libp2p::core::identity::secp256k1::Keypair = secret_key.into();
-                Keypair::Secp256k1(kp)
+
+                let kp: libp2p::identity::secp256k1::Keypair = secret_key.into();
+                kp.into()
             }
             CombinedKey::Ed25519(_) => unreachable!(), // not implemented as the ENR key is generated with secp256k1
         }