Support client mode in kademlia

protocols/kad/src/behaviour.rs

@@ -29,7 +29,7 @@ use crate::handler::{
 };
 use crate::jobs::*;
 use crate::kbucket::{self, Distance, KBucketsTable, NodeStatus};
-use crate::protocol::{KadConnectionType, KadPeer, KademliaProtocolConfig};
+use crate::protocol::{KadConnectionType, KadPeer, KademliaProtocolConfig, Mode};
 use crate::query::{Query, QueryConfig, QueryId, QueryPool, QueryPoolState};
 use crate::record::{
     self,
@@ -147,6 +147,8 @@ pub struct KademliaConfig {
     connection_idle_timeout: Duration,
     kbucket_inserts: KademliaBucketInserts,
     caching: KademliaCaching,
+    // Set the peer mode.
+    mode: Mode,
 }
 
 /// The configuration for Kademlia "write-back" caching after successful
@@ -180,6 +182,8 @@ impl Default for KademliaConfig {
             connection_idle_timeout: Duration::from_secs(10),
             kbucket_inserts: KademliaBucketInserts::OnConnected,
             caching: KademliaCaching::Enabled { max_peers: 1 },
+            // By default, a peer will not be in client mode.
+            mode: Mode::default(),
         }
     }
 }
@@ -352,6 +356,21 @@ impl KademliaConfig {
         self.caching = c;
         self
     }
+
+    /// Sets the [`Mode`] the node is operating in.
+    ///
+    /// The default is [`Mode::Server`].
+    ///
+    /// In [`Mode::Client`] the node does not advertise support for the Kademlia
+    /// protocol, nor does it accept incoming Kademlia streams. Peers will thus not
+    /// include the local node in their routing table. The node can still make
+    /// outbound requests.
+    ///
+    /// Use [`Mode::Client`] when not publicly reachable.
+    pub fn set_mode(&mut self, mode: Mode) -> &mut Self {
+        self.mode = mode;
+        self
+    }
 }
 
 impl<TStore> Kademlia<TStore>
@@ -1645,6 +1664,7 @@ where
             protocol_config: self.protocol_config.clone(),
             allow_listening: true,
             idle_timeout: self.connection_idle_timeout,
+            mode: Mode::default(),
         })
     }
 

protocols/kad/src/behaviour/test.rs

@@ -1287,3 +1287,87 @@ fn network_behaviour_inject_address_change() {
         kademlia.addresses_of_peer(&remote_peer_id),
     );
 }
+
+#[test]
+fn client_mode() {
+    let mut cfg = KademliaConfig::default();
+    cfg.set_mode(Mode::Client);
+
+    // Create a server and client peer.
+    let client_swarm = build_node_with_config(cfg);
+    let server_swarm = build_node();
+
+    let mut swarms = Vec::new();
+    swarms.push(client_swarm.1);
+    swarms.push(server_swarm.1);
+
+    let mut swarm_addrs = Vec::new();
+    swarm_addrs.push(client_swarm.0);
+    swarm_addrs.push(server_swarm.0);
+
+    let swarm_ids: Vec<_> = swarms
+        .iter()
+        .map(|swarm| swarm.local_peer_id())
+        .cloned()
+        .collect();
+
+    block_on(poll_fn(move |ctx| {
+        // flags to make sure both peers are listening
+        let mut client_listening = false;
+        let mut server_listening = false;
+
+        for swarm in swarms.iter_mut() {
+            loop {
+                match swarm.poll_next_unpin(ctx) {
+                    Poll::Ready(Some(SwarmEvent::NewListenAddr {
+                        listener_id: _,
+                        address,
+                    })) => {
+                        if address == swarm_addrs[0] {
+                            client_listening = true;
+                            return Poll::Ready(());
+                        }
+                        if address == swarm_addrs[1] {
+                            server_listening = true;
+                            return Poll::Ready(());
+                        }
+                    }
+                    Poll::Ready(_) => {
+                        if swarm.local_peer_id() == &swarm_ids[0] && client_listening == true {
+                            let bucket = swarm.behaviour_mut().kbucket(swarm_ids[1].clone());
+                            assert!(bucket.is_some());
+                            // Check if the client peer has the server peer.
+                            assert!(
+                                bucket
+                                    .unwrap()
+                                    .iter()
+                                    .any(|p| p.node.key.preimage() == &swarm_ids[1]),
+                                "The client peer does not have the server node"
+                            );
+
+                            return Poll::Ready(());
+                        }
+
+                        if swarm.local_peer_id() == &swarm_ids[1] && server_listening == true {
+                            let bucket =
+                                swarm.behaviour_mut().kbucket(swarm_ids[0].clone()).unwrap();
+                            // Check if the server peer doesn't have the client peer.
+                            assert!(
+                                bucket
+                                    .iter()
+                                    .all(|p| p.node.key.preimage() != &swarm_ids[0]),
+                                "The server peer has the client node"
+                            );
+
+                            return Poll::Ready(());
+                        }
+                    }
+                    Poll::Pending => break,
+                }
+            }
+        }
+        assert!(client_listening);
+        assert!(server_listening);
+        Poll::Pending
+    }));
+}

protocols/kad/src/handler.rs

@@ -20,7 +20,7 @@
 
 use crate::protocol::{
     KadInStreamSink, KadOutStreamSink, KadPeer, KadRequestMsg, KadResponseMsg,
-    KademliaProtocolConfig,
+    KademliaProtocolConfig, Mode,
 };
 use crate::record::{self, Record};
 use futures::prelude::*;
@@ -62,10 +62,15 @@ impl<T: Clone + fmt::Debug + Send + 'static> IntoProtocolsHandler for KademliaHa
     }
 
     fn inbound_protocol(&self) -> <Self::Handler as ProtocolsHandler>::InboundProtocol {
-        if self.config.allow_listening {
-            upgrade::EitherUpgrade::A(self.config.protocol_config.clone())
-        } else {
-            upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade)
+        match self.config.mode {
+            Mode::Client => upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade),
+            Mode::Server => {
+                if self.config.allow_listening {
+                    upgrade::EitherUpgrade::A(self.config.protocol_config.clone())
+                } else {
+                    upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade)
+                }
+            }
         }
     }
 }
@@ -123,6 +128,9 @@ pub struct KademliaHandlerConfig {
 
     /// Time after which we close an idle connection.
     pub idle_timeout: Duration,
+
+    /// [`Mode`] the handler is operating in.
+    pub mode: Mode,
 }
 
 /// State of an active substream, opened either by us or by the remote.
@@ -480,11 +488,20 @@ where
     type InboundOpenInfo = ();
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
-        if self.config.allow_listening {
-            SubstreamProtocol::new(self.config.protocol_config.clone(), ())
-                .map_upgrade(upgrade::EitherUpgrade::A)
-        } else {
-            SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
+        match self.config.mode {
+            Mode::Server => {
+                if self.config.allow_listening {
+                    SubstreamProtocol::new(self.config.protocol_config.clone(), ())
+                        .map_upgrade(upgrade::EitherUpgrade::A)
+                } else {
+                    SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
+                }
+            }
+            // If we are in client mode, I don't want to advertise Kademlia so that other peers will
+            // not send me Kademlia requests.
+            Mode::Client => {
+                SubstreamProtocol::new(upgrade::EitherUpgrade::B(upgrade::DeniedUpgrade), ())
+            }
         }
     }
 
@@ -520,12 +537,15 @@ where
         self.next_connec_unique_id.0 += 1;
         self.substreams
             .push(SubstreamState::InWaitingMessage(connec_unique_id, protocol));
-        if let ProtocolStatus::Unconfirmed = self.protocol_status {
-            // Upon the first successfully negotiated substream, we know that the
-            // remote is configured with the same protocol name and we want
-            // the behaviour to add this peer to the routing table, if possible.
-            self.protocol_status = ProtocolStatus::Confirmed;
-        }
+        // Just because another peer is sending us Kademlia requests, doesn't necessarily
+        // mean that it will answer the Kademlia requests that we send to it.
+        // Thus, Commenting this code out for now.
+        // if let ProtocolStatus::Unconfirmed = self.protocol_status {
+        //     // Upon the first successfully negotiated substream, we know that the
+        //     // remote is configured with the same protocol name and we want
+        //     // the behaviour to add this peer to the routing table, if possible.
+        //     self.protocol_status = ProtocolStatus::Confirmed;
+        // }
     }
 
     fn inject_event(&mut self, message: KademliaHandlerIn<TUserData>) {
@@ -763,6 +783,7 @@ impl Default for KademliaHandlerConfig {
             protocol_config: Default::default(),
             allow_listening: true,
             idle_timeout: Duration::from_secs(10),
+            mode: Mode::default(),
         }
     }
 }

protocols/kad/src/protocol.rs

@@ -46,6 +46,19 @@ pub const DEFAULT_PROTO_NAME: &[u8] = b"/ipfs/kad/1.0.0";
 /// The default maximum size for a varint length-delimited packet.
 pub const DEFAULT_MAX_PACKET_SIZE: usize = 16 * 1024;
 
+#[derive(Debug, Clone)]
+/// See [`crate::KademliaConfig::set_mode`].
+pub enum Mode {
+    Client,
+    Server,
+}
+
+impl Default for Mode {
+    fn default() -> Self {
+        Mode::Server
+    }
+}
+
 /// Status of our connection to a node reported by the Kademlia protocol.
 #[derive(Copy, Clone, PartialEq, Eq, Debug, Hash)]
 pub enum KadConnectionType {