feat(network): limit connections from same ip (#388)

* feat(network): introduce connection limit for same ip

* feat(network): ban same ip peer for a while to avoid select it

* feat: expose same ip conn limit config

* refactor(network): split conn limit check from session book insert

* fix(network): save session peer even it failed conn limit check later

core/network/src/common.rs

@@ -174,7 +174,7 @@ impl Future for HeartBeat {
     }
 }
 
-#[derive(Debug, Display, PartialEq, Eq, Serialize, Deserialize, Clone)]
+#[derive(Debug, Display, PartialEq, Eq, Serialize, Deserialize, Clone, Hash)]
 #[display(fmt = "{}:{}", host, port)]
 pub struct ConnectedAddr {
     pub host: String,

core/network/src/config.rs

@@ -18,7 +18,7 @@ use crate::{
     common::socket_to_multi_addr,
     connection::ConnectionConfig,
     error::NetworkError,
-    peer_manager::{ArcPeer, PeerManagerConfig, SharedSessionsConfig, TrustMetricConfig},
+    peer_manager::{ArcPeer, PeerManagerConfig, TrustMetricConfig},
     selfcheck::SelfCheckConfig,
     traits::MultiaddrExt,
     PeerIdExt,
@@ -39,6 +39,8 @@ pub const DEFAULT_MAX_WAIT_STREAMS: usize = 256;
 // Default write timeout
 pub const DEFAULT_WRITE_TIMEOUT: u64 = 10; // seconds
 
+pub const DEFAULT_SAME_IP_CONN_LIMIT: usize = 1;
+
 // Default peer trust metric
 pub const DEFAULT_PEER_TRUST_INTERVAL_DURATION: Duration = Duration::from_secs(60);
 pub const DEFAULT_PEER_TRUST_MAX_HISTORY_DURATION: Duration =
@@ -124,6 +126,7 @@ pub struct NetworkConfig {
     pub peer_trust_max_history: Duration,
     pub peer_fatal_ban:         Duration,
     pub peer_soft_ban:          Duration,
+    pub same_ip_conn_limit:     usize,
 
     // identity and encryption
     pub secio_keypair: SecioKeyPair,
@@ -170,6 +173,7 @@ impl NetworkConfig {
             peer_trust_max_history: DEFAULT_PEER_TRUST_MAX_HISTORY_DURATION,
             peer_fatal_ban:         DEFAULT_PEER_FATAL_BAN_DURATION,
             peer_soft_ban:          DEFAULT_PEER_SOFT_BAN_DURATION,
+            same_ip_conn_limit:     DEFAULT_SAME_IP_CONN_LIMIT,
 
             secio_keypair: SecioKeyPair::secp256k1_generated(),
 
@@ -194,6 +198,14 @@ impl NetworkConfig {
         self
     }
 
+    pub fn same_ip_conn_limit(mut self, limit: Option<usize>) -> Self {
+        if let Some(limit) = limit {
+            self.same_ip_conn_limit = limit;
+        }
+
+        self
+    }
+
     pub fn max_frame_length(mut self, max: Option<usize>) -> Self {
         if let Some(max) = max {
             self.max_frame_length = max;
@@ -427,17 +439,18 @@ impl From<&NetworkConfig> for PeerManagerConfig {
             TrustMetricConfig::new(config.peer_trust_interval, config.peer_trust_max_history);
 
         PeerManagerConfig {
-            our_id:            config.secio_keypair.peer_id(),
-            pubkey:            config.secio_keypair.public_key(),
-            bootstraps:        config.bootstraps.clone(),
-            allowlist:         config.allowlist.clone(),
-            allowlist_only:    config.allowlist_only,
-            peer_trust_config: Arc::new(peer_trust_config),
-            peer_fatal_ban:    config.peer_fatal_ban,
-            peer_soft_ban:     config.peer_soft_ban,
-            max_connections:   config.max_connections,
-            routine_interval:  config.peer_manager_heart_beat_interval,
-            peer_dat_file:     config.peer_dat_file.clone(),
+            our_id:             config.secio_keypair.peer_id(),
+            pubkey:             config.secio_keypair.public_key(),
+            bootstraps:         config.bootstraps.clone(),
+            allowlist:          config.allowlist.clone(),
+            allowlist_only:     config.allowlist_only,
+            peer_trust_config:  Arc::new(peer_trust_config),
+            peer_fatal_ban:     config.peer_fatal_ban,
+            peer_soft_ban:      config.peer_soft_ban,
+            max_connections:    config.max_connections,
+            same_ip_conn_limit: config.same_ip_conn_limit,
+            routine_interval:   config.peer_manager_heart_beat_interval,
+            peer_dat_file:      config.peer_dat_file.clone(),
         }
     }
 }
@@ -462,13 +475,3 @@ impl From<&NetworkConfig> for SelfCheckConfig {
         }
     }
 }
-
-// TODO: checkout max_frame_length
-impl From<&NetworkConfig> for SharedSessionsConfig {
-    fn from(config: &NetworkConfig) -> Self {
-        SharedSessionsConfig {
-            write_timeout:          config.write_timeout,
-            max_stream_window_size: config.max_frame_length,
-        }
-    }
-}

core/network/src/connection/control.rs

@@ -15,10 +15,10 @@ use protocol::{traits::Priority, Bytes};
 use crate::{
     error::NetworkError,
     event::PeerManagerEvent,
-    traits::{MessageSender, NetworkProtocol, SessionBook},
+    traits::{MessageSender, NetworkProtocol, SharedSessionBook},
 };
 
-pub struct ConnectionServiceControl<P: NetworkProtocol, B: SessionBook> {
+pub struct ConnectionServiceControl<P: NetworkProtocol, B: SharedSessionBook> {
     inner:    ServiceControl,
     mgr_srv:  UnboundedSender<PeerManagerEvent>,
     sessions: B,
@@ -27,7 +27,7 @@ pub struct ConnectionServiceControl<P: NetworkProtocol, B: SessionBook> {
     pin_protocol: PhantomData<fn() -> P>,
 }
 
-impl<P: NetworkProtocol, B: SessionBook> ConnectionServiceControl<P, B> {
+impl<P: NetworkProtocol, B: SharedSessionBook> ConnectionServiceControl<P, B> {
     pub fn new(
         control: ServiceControl,
         mgr_srv: UnboundedSender<PeerManagerEvent>,
@@ -84,7 +84,7 @@ impl<P: NetworkProtocol, B: SessionBook> ConnectionServiceControl<P, B> {
     }
 }
 
-impl<P: NetworkProtocol, B: SessionBook + Clone> Clone for ConnectionServiceControl<P, B> {
+impl<P: NetworkProtocol, B: SharedSessionBook + Clone> Clone for ConnectionServiceControl<P, B> {
     fn clone(&self) -> Self {
         ConnectionServiceControl {
             inner:    self.inner.clone(),
@@ -100,7 +100,7 @@ impl<P: NetworkProtocol, B: SessionBook + Clone> Clone for ConnectionServiceCont
 impl<P, B> MessageSender for ConnectionServiceControl<P, B>
 where
     P: NetworkProtocol,
-    B: SessionBook + Send + Sync + Unpin + 'static,
+    B: SharedSessionBook + Send + Sync + Unpin + 'static,
 {
     fn send(&self, tar: TargetSession, msg: Bytes, pri: Priority) -> Result<(), NetworkError> {
         let proto_id = P::message_proto_id();

core/network/src/connection/mod.rs

@@ -24,7 +24,7 @@ use tentacle::{
 use crate::{
     error::NetworkError,
     event::{ConnectionEvent, PeerManagerEvent},
-    traits::{NetworkProtocol, SessionBook},
+    traits::{NetworkProtocol, SharedSessionBook},
 };
 
 pub struct ConnectionConfig {
@@ -114,7 +114,7 @@ impl<P: NetworkProtocol> ConnectionService<P> {
         Ok(())
     }
 
-    pub fn control<B: SessionBook>(
+    pub fn control<B: SharedSessionBook>(
         &self,
         mgr_tx: UnboundedSender<PeerManagerEvent>,
         book: B,

core/network/src/metrics.rs

@@ -10,7 +10,7 @@ use futures::task::AtomicWaker;
 
 use crate::{
     common::{ConnectedAddr, HeartBeat},
-    traits::SessionBook,
+    traits::SharedSessionBook,
 };
 
 const METRICS_INTERVAL: Duration = Duration::from_secs(1);
@@ -23,7 +23,7 @@ pub(crate) struct Metrics<S> {
 
 impl<S> Metrics<S>
 where
-    S: SessionBook + Send + Unpin + 'static,
+    S: SharedSessionBook + Send + Unpin + 'static,
 {
     pub fn new(sessions: S) -> Self {
         let waker = Arc::new(AtomicWaker::new());
@@ -60,7 +60,7 @@ where
 
 impl<S> Future for Metrics<S>
 where
-    S: SessionBook + Send + Unpin + 'static,
+    S: SharedSessionBook + Send + Unpin + 'static,
 {
     type Output = ();