feat: add a hote punching protocol try use on nat traversal

Author: driftluo

Cargo.lock

@@ -30,7 +30,7 @@ use futures::{Future, channel::mpsc::Sender};
 use ipnetwork::IpNetwork;
 use p2p::{
     SessionId, async_trait,
-    builder::ServiceBuilder,
+    builder::{MetaBuilder, ServiceBuilder},
     bytes::Bytes,
     context::{ServiceContext, SessionContext},
     error::{DialerErrorKind, HandshakeErrorKind, ProtocolHandleErrorKind, SendErrorKind},
@@ -74,10 +74,9 @@ pub struct NetworkState {
     /// Node listened addresses
     pub(crate) listened_addrs: RwLock<Vec<Multiaddr>>,
     dialing_addrs: RwLock<HashMap<PeerId, Instant>>,
-    /// Node public addresses,
-    /// includes manually public addrs and remote peer observed addrs
-    public_addrs: RwLock<HashSet<Multiaddr>>,
-    pending_observed_addrs: RwLock<HashSet<Multiaddr>>,
+    /// Node public addresses by config
+    public_addrs: HashSet<Multiaddr>,
+    observed_addrs: RwLock<HashMap<PeerIndex, Multiaddr>>,
     local_private_key: secio::SecioKeyPair,
     local_peer_id: PeerId,
     pub(crate) bootnodes: Vec<Multiaddr>,
@@ -139,9 +138,9 @@ impl NetworkState {
             bootnodes,
             peer_registry: RwLock::new(peer_registry),
             dialing_addrs: RwLock::new(HashMap::default()),
-            public_addrs: RwLock::new(public_addrs),
+            public_addrs,
             listened_addrs: RwLock::new(Vec::new()),
-            pending_observed_addrs: RwLock::new(HashSet::default()),
+            observed_addrs: RwLock::new(HashMap::default()),
             local_private_key,
             local_peer_id,
             active: AtomicBool::new(true),
@@ -187,9 +186,9 @@ impl NetworkState {
             bootnodes,
             peer_registry: RwLock::new(peer_registry),
             dialing_addrs: RwLock::new(HashMap::default()),
-            public_addrs: RwLock::new(public_addrs),
+            public_addrs,
             listened_addrs: RwLock::new(Vec::new()),
-            pending_observed_addrs: RwLock::new(HashSet::default()),
+            observed_addrs: RwLock::new(HashMap::default()),
             local_private_key,
             local_peer_id,
             active: AtomicBool::new(true),
@@ -338,12 +337,14 @@ impl NetworkState {
     }
 
     pub(crate) fn public_addrs(&self, count: usize) -> Vec<Multiaddr> {
-        self.public_addrs
-            .read()
-            .iter()
-            .take(count)
-            .cloned()
-            .collect()
+        if self.public_addrs.len() <= count {
+            return self.public_addrs.iter().cloned().collect();
+        } else {
+            self.public_addrs
+                .iter()
+                .cloned()
+                .choose_multiple(&mut rand::thread_rng(), count)
+        }
     }
 
     pub(crate) fn connection_status(&self) -> ConnectionStatus {
@@ -392,7 +393,7 @@ impl NetworkState {
             trace!("Do not dial self: {:?}, {}", peer_id, addr);
             return false;
         }
-        if self.public_addrs.read().contains(addr) {
+        if self.public_addrs.contains(addr) {
             trace!(
                 "Do not dial listened address(self): {:?}, {}",
                 peer_id, addr
@@ -500,40 +501,28 @@ impl NetworkState {
         }
     }
 
-    /// this method is intent to check observed addr by dial to self
-    pub(crate) fn try_dial_observed_addrs(&self, p2p_control: &ServiceControl) {
-        let mut pending_observed_addrs = self.pending_observed_addrs.write();
-        if pending_observed_addrs.is_empty() {
-            let addrs = self.public_addrs.read();
-            if addrs.is_empty() {
-                return;
-            }
-            // random get addr
-            if let Some(addr) = addrs.iter().choose(&mut rand::thread_rng()) {
-                if let Err(err) = p2p_control.dial(
-                    addr.clone(),
-                    TargetProtocol::Single(SupportProtocols::Identify.protocol_id()),
-                ) {
-                    trace!("try_dial_observed_addrs {err} failed in public address")
-                }
-            }
-        } else {
-            for addr in pending_observed_addrs.drain() {
-                trace!("try dial observed addr: {:?}", addr);
-                if let Err(err) = p2p_control.dial(
-                    addr,
-                    TargetProtocol::Single(SupportProtocols::Identify.protocol_id()),
-                ) {
-                    trace!("try_dial_observed_addrs {err} failed in pending observed addresses")
-                }
-            }
-        }
+    /// add observed address for identify protocol
+    pub(crate) fn add_observed_addr(&self, session_id: SessionId, addr: Multiaddr) {
+        let mut pending_observed_addrs = self.observed_addrs.write();
+        pending_observed_addrs.insert(session_id, addr);
     }
 
-    /// add observed address for identify protocol
-    pub(crate) fn add_observed_addrs(&self, iter: impl Iterator<Item = Multiaddr>) {
-        let mut pending_observed_addrs = self.pending_observed_addrs.write();
-        pending_observed_addrs.extend(iter)
+    // randomly select count addresses from observed_addrs
+    #[cfg(not(target_family = "wasm"))]
+    pub(crate) fn observed_addrs(&self, count: usize) -> Vec<Multiaddr> {
+        let observed_addrs = self
+            .observed_addrs
+            .read()
+            .values()
+            .cloned()
+            .collect::<HashSet<_>>();
+        if observed_addrs.len() <= count {
+            return observed_addrs.into_iter().collect();
+        } else {
+            observed_addrs
+                .into_iter()
+                .choose_multiple(&mut rand::thread_rng(), count)
+        }
     }
 
     /// Network message processing controller, default is true, if false, discard any received messages
@@ -613,19 +602,11 @@ impl ServiceHandle for EventHandler {
     async fn handle_error(&mut self, context: &mut ServiceContext, error: ServiceError) {
         match error {
             ServiceError::DialerError { address, error } => {
-                let mut public_addrs = self.network_state.public_addrs.write();
-
                 match error {
                     DialerErrorKind::HandshakeError(HandshakeErrorKind::SecioError(
                         SecioError::ConnectSelf,
                     )) => {
                         debug!("dial observed address success: {:?}", address);
-                        if let Some(ip) = multiaddr_to_socketaddr(&address) {
-                            if is_reachable(ip.ip()) {
-                                public_addrs.insert(address);
-                            }
-                        }
-                        return;
                     }
                     DialerErrorKind::IoError(e)
                         if e.kind() == std::io::ErrorKind::AddrNotAvailable =>
@@ -636,12 +617,6 @@ impl ServiceHandle for EventHandler {
                         debug!("DialerError({}) {}", address, error);
                     }
                 }
-                if public_addrs.remove(&address) {
-                    info!(
-                        "Dial {} failed, remove it from network_state.public_addrs",
-                        address
-                    );
-                }
                 self.network_state.dial_failed(&address);
             }
             ServiceError::ProtocolError {
@@ -815,6 +790,10 @@ impl ServiceHandle for EventHandler {
                         peer_store.remove_disconnected_peer(&session_context.address);
                     });
                 }
+                self.network_state
+                    .observed_addrs
+                    .write()
+                    .remove(&session_context.id);
             }
             _ => {
                 info!("p2p service event: {:?}", event);
@@ -937,6 +916,26 @@ impl NetworkService {
             protocol_metas.push(disconnect_message_meta);
         }
 
+        // HolePunching protocol
+        #[cfg(not(target_family = "wasm"))]
+        if config
+            .support_protocols
+            .contains(&SupportProtocol::HolePunching)
+        {
+            let hole_punching_state = Arc::clone(&network_state);
+            let hole_punching_meta_builder: MetaBuilder = SupportProtocols::HolePunching.into();
+            let hole_punching_meta = hole_punching_meta_builder
+                .before_send(crate::compress::compress)
+                .before_receive(|| Some(Box::new(crate::compress::decompress)))
+                .service_handle(move || {
+                    ProtocolHandle::Callback(Box::new(
+                        crate::protocols::hole_punching::HolePunching::new(hole_punching_state),
+                    ))
+                })
+                .build();
+            protocol_metas.push(hole_punching_meta);
+        }
+
         let mut service_builder = ServiceBuilder::default();
         let yamux_config = YamuxConfig {
             max_stream_count: protocol_metas.len(),

network/src/network.rs

@@ -217,6 +217,25 @@ impl PeerStore {
         self.addr_manager.fetch_random(count, filter)
     }
 
+    /// Return address that we never connected to, used for penetration.
+    pub fn fetch_nat_addrs(&mut self, count: usize, required_flags: Flags) -> Vec<AddrInfo> {
+        // Get info:
+        // 1. Never connected
+        // 2. Not already connected
+
+        let peers = &self.connected_peers;
+
+        let filter = |peer_addr: &AddrInfo| {
+            required_flags_filter(required_flags, Flags::from_bits_truncate(peer_addr.flags))
+                && extract_peer_id(&peer_addr.addr)
+                    .map(|peer_id| !peers.contains_key(&peer_id))
+                    .unwrap_or_default()
+                && peer_addr.connected(|t| t == 0)
+        };
+
+        self.addr_manager.fetch_random(count, filter)
+    }
+
     /// Return valid addrs that success connected, used for discovery.
     pub fn fetch_random_addrs(&mut self, count: usize, required_flags: Flags) -> Vec<AddrInfo> {
         // Get info:

network/src/peer_store/peer_store_impl.rs

@@ -9,7 +9,7 @@ use p2p::{
     bytes::Bytes,
     context::{ProtocolContext, ProtocolContextMutRef, SessionContext},
     multiaddr::{Multiaddr, Protocol},
-    service::{SessionType, TargetProtocol},
+    service::TargetProtocol,
     traits::ServiceProtocol,
     utils::{extract_peer_id, is_reachable, multiaddr_to_socketaddr},
 };
@@ -77,7 +77,7 @@ pub trait Callback: Clone + Send {
     /// Add remote peer's listen addresses
     fn add_remote_listen_addrs(&mut self, session: &SessionContext, addrs: Vec<Multiaddr>);
     /// Add our address observed by remote peer
-    fn add_observed_addr(&mut self, addr: Multiaddr, ty: SessionType) -> MisbehaveResult;
+    fn add_observed_addr(&mut self, addr: Multiaddr, sessioin_id: SessionId) -> MisbehaveResult;
     /// Report misbehavior
     fn misbehave(&mut self, session: &SessionContext, kind: Misbehavior) -> MisbehaveResult;
 }
@@ -173,7 +173,7 @@ impl<T: Callback> IdentifyProtocol<T> {
             return MisbehaveResult::Continue;
         }
 
-        self.callback.add_observed_addr(observed, info.session.ty)
+        self.callback.add_observed_addr(observed, info.session.id)
     }
 }
 
@@ -528,42 +528,14 @@ impl Callback for IdentifyCallback {
         })
     }
 
-    fn add_observed_addr(&mut self, mut addr: Multiaddr, ty: SessionType) -> MisbehaveResult {
-        if ty.is_inbound() {
-            // The address already been discovered by other peer
-            return MisbehaveResult::Continue;
-        }
-
-        // observed addr is not a reachable ip
-        if !multiaddr_to_socketaddr(&addr)
-            .map(|socket_addr| is_reachable(socket_addr.ip()))
-            .unwrap_or(false)
-        {
-            return MisbehaveResult::Continue;
-        }
-
+    fn add_observed_addr(&mut self, mut addr: Multiaddr, session_id: SessionId) -> MisbehaveResult {
         if extract_peer_id(&addr).is_none() {
             addr.push(Protocol::P2P(Cow::Borrowed(
                 self.network_state.local_peer_id().as_bytes(),
             )))
         }
 
-        let source_addr = addr.clone();
-        let observed_addrs_iter = self
-            .listen_addrs()
-            .into_iter()
-            .filter_map(|listen_addr| multiaddr_to_socketaddr(&listen_addr))
-            .map(|socket_addr| {
-                addr.iter()
-                    .map(|proto| match proto {
-                        Protocol::Tcp(_) => Protocol::Tcp(socket_addr.port()),
-                        value => value,
-                    })
-                    .collect::<Multiaddr>()
-            })
-            .chain(::std::iter::once(source_addr));
-
-        self.network_state.add_observed_addrs(observed_addrs_iter);
+        self.network_state.add_observed_addr(session_id, addr);
         // NOTE: for future usage
         MisbehaveResult::Continue
     }

network/src/protocols/identify/mod.rs

@@ -5,6 +5,9 @@ pub(crate) mod identify;
 pub(crate) mod ping;
 pub(crate) mod support_protocols;
 
+#[cfg(not(target_family = "wasm"))]
+pub(crate) mod hole_punching;
+
 #[cfg(test)]
 mod tests;
 

network/src/protocols/mod.rs

@@ -56,6 +56,8 @@ pub enum SupportProtocols {
     LightClient,
     /// Filter: A protocol used for client side block data filtering.
     Filter,
+    /// HolePunching: A protocol used to connect peers behind firewalls or NAT routers.
+    HolePunching,
 }
 
 impl SupportProtocols {
@@ -74,6 +76,7 @@ impl SupportProtocols {
             SupportProtocols::Alert => 110,
             SupportProtocols::LightClient => 120,
             SupportProtocols::Filter => 121,
+            SupportProtocols::HolePunching => 130,
         }
         .into()
     }
@@ -93,6 +96,7 @@ impl SupportProtocols {
             SupportProtocols::Alert => "/ckb/alt",
             SupportProtocols::LightClient => "/ckb/lightclient",
             SupportProtocols::Filter => "/ckb/filter",
+            SupportProtocols::HolePunching => "/ckb/HolePunching",
         }
         .to_owned()
     }
@@ -117,6 +121,7 @@ impl SupportProtocols {
             SupportProtocols::RelayV3 => vec!["2".to_owned(), LASTEST_VERSION.to_owned()],
             SupportProtocols::LightClient => vec!["2".to_owned(), LASTEST_VERSION.to_owned()],
             SupportProtocols::Filter => vec!["2".to_owned(), LASTEST_VERSION.to_owned()],
+            SupportProtocols::HolePunching => vec!["2".to_owned(), LASTEST_VERSION.to_owned()],
         }
     }
 
@@ -134,6 +139,7 @@ impl SupportProtocols {
             SupportProtocols::Alert => 128 * 1024,       // 128 KB
             SupportProtocols::LightClient => 2 * 1024 * 1024, // 2 MB
             SupportProtocols::Filter => 2 * 1024 * 1024, // 2   MB
+            SupportProtocols::HolePunching => 512 * 1024, // 512 KB
         }
     }
 

network/src/protocols/support_protocols.rs

@@ -196,11 +196,6 @@ impl OutboundPeerService {
         }
     }
 
-    fn try_dial_observed(&self) {
-        self.network_state
-            .try_dial_observed_addrs(&self.p2p_control);
-    }
-
     fn update_outbound_connected_ms(&mut self) {
         if self.update_outbound_connected_count > 10 {
             let connected_outbounds: Vec<p2p::multiaddr::Multiaddr> =
@@ -256,8 +251,6 @@ impl Future for OutboundPeerService {
             self.dial_feeler();
             // keep outbound peer is enough
             self.try_dial_peers();
-            // try dial observed addrs
-            self.try_dial_observed();
             // Keep connected nodes up to date in the peer store
             self.update_outbound_connected_ms();
         }

network/src/services/outbound_peer.rs

@@ -105,7 +105,7 @@ discovery_local_address = false # {{
 bootnode_mode = false
 
 # Supported protocols list, only "Sync" and "Identify" are mandatory, others are optional
-support_protocols = ["Ping", "Discovery", "Identify", "Feeler", "DisconnectMessage", "Sync", "Relay", "Time", "Alert", "LightClient", "Filter"]
+support_protocols = ["Ping", "Discovery", "Identify", "Feeler", "DisconnectMessage", "Sync", "Relay", "Time", "Alert", "LightClient", "Filter", "HolePunching"]
 
 # [network.sync.header_map]
 # memory_limit = "256MB"