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,67 @@ 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_with_config(KademliaConfig::default());
+
+    let mut swarms = vec![client_swarm, server_swarm];
+
+    // Collect addresses and peer IDs.
+    let addrs: Vec<_> = swarms.iter().map(|(addr, _)| addr.clone()).collect();
+    let peers: Vec<_> = swarms
+        .iter()
+        .map(|(_, swarm)| swarm.local_peer_id().clone())
+        .collect();
+
+    swarms[0].1.dial_addr(addrs[1].clone()).unwrap();
+    swarms[1].1.dial_addr(addrs[0].clone()).unwrap();
+
+    block_on(poll_fn(move |ctx| {
+        // Flag variables to store the result of various checks.
+        let mut is_server_present = false;
+        let mut is_client_absent = false;
+
+        for (_, swarm) in swarms.iter_mut() {
+            loop {
+                match swarm.poll_next_unpin(ctx) {
+                    Poll::Ready(Some(SwarmEvent::Behaviour(KademliaEvent::RoutingUpdated {
+                        peer,
+                        ..
+                    }))) => {
+                        // Check if the server peer is present in the client peer's routing table.
+                        if swarm.local_peer_id().clone() == peers[0] {
+                            is_server_present = peer == peers[1];
+                        }
+                        return Poll::Ready(());
+                    }
+                    Poll::Ready(_) => {
+                        // Check if the client peer is NOT present in the server peer's routing table.
+                        if swarm.local_peer_id().clone() == peers[1] {
+                            is_client_absent = swarm.behaviour_mut().kbucket(peers[0]).is_none();
+                        }
+                        return Poll::Ready(());
+                    }
+                    Poll::Pending => break,
+                }
+            }
+        }
+        // The assert statements are present outside the match block, to make sure that all the
+        // necessary events occur.
+        assert!(
+            is_server_present,
+            "The client peer does not have the server peer in its routing table."
+        );
+        assert!(
+            is_client_absent,
+            "The server peer has the client peer in its routing table."
+        );
+        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,6 +537,9 @@ where
         self.next_connec_unique_id.0 += 1;
         self.substreams
             .push(SubstreamState::InWaitingMessage(connec_unique_id, protocol));
+        // 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
@@ -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 {