feat(p2p): add reputation management features (#2389)

crates/net/network-api/src/lib.rs

@@ -55,6 +55,7 @@ pub trait PeersInfo: Send + Sync {
 }
 
 /// Provides an API for managing the peers of the network.
+#[async_trait]
 pub trait Peers: PeersInfo {
     /// Adds a peer to the peer set.
     fn add_peer(&self, peer: PeerId, addr: SocketAddr) {
@@ -80,6 +81,9 @@ pub trait Peers: PeersInfo {
 
     /// Send a reputation change for the given peer.
     fn reputation_change(&self, peer_id: PeerId, kind: ReputationChangeKind);
+
+    /// Get the reputation of a peer.
+    async fn reputation_by_id(&self, peer_id: PeerId) -> Result<Option<Reputation>, NetworkError>;
 }
 
 /// Represents the kind of peer

crates/net/network-api/src/reputation.rs

@@ -20,6 +20,15 @@ pub enum ReputationChangeKind {
     FailedToConnect,
     /// Connection dropped by peer.
     Dropped,
+    /// Reset the reputation to the default value.
+    Reset,
     /// Apply a reputation change by value
     Other(Reputation),
 }
+
+impl ReputationChangeKind {
+    /// Returns true if the reputation change is a reset.
+    pub fn is_reset(&self) -> bool {
+        matches!(self, Self::Reset)
+    }
+}

crates/net/network-api/src/test_utils.rs

@@ -1,4 +1,6 @@
-use crate::{NetworkError, NetworkInfo, PeerKind, Peers, PeersInfo, ReputationChangeKind};
+use crate::{
+    NetworkError, NetworkInfo, PeerKind, Peers, PeersInfo, Reputation, ReputationChangeKind,
+};
 use async_trait::async_trait;
 use reth_eth_wire::{DisconnectReason, ProtocolVersion};
 use reth_primitives::{rpc::Chain::Mainnet, NodeRecord, PeerId};
@@ -49,6 +51,7 @@ impl PeersInfo for NoopNetwork {
     }
 }
 
+#[async_trait]
 impl Peers for NoopNetwork {
     fn add_peer_kind(&self, _peer: PeerId, _kind: PeerKind, _addr: SocketAddr) {}
 
@@ -59,4 +62,8 @@ impl Peers for NoopNetwork {
     fn disconnect_peer_with_reason(&self, _peer: PeerId, _reason: DisconnectReason) {}
 
     fn reputation_change(&self, _peer_id: PeerId, _kind: ReputationChangeKind) {}
+
+    async fn reputation_by_id(&self, _peer_id: PeerId) -> Result<Option<Reputation>, NetworkError> {
+        Ok(None)
+    }
 }

crates/net/network/src/manager.rs

@@ -537,6 +537,9 @@ where
             NetworkHandleMessage::ReputationChange(peer_id, kind) => {
                 self.swarm.state_mut().peers_mut().apply_reputation_change(&peer_id, kind);
             }
+            NetworkHandleMessage::GetReputationById(peer_id, tx) => {
+                let _ = tx.send(self.swarm.state_mut().peers().get_reputation(&peer_id));
+            }
             NetworkHandleMessage::FetchClient(tx) => {
                 let _ = tx.send(self.fetch_client());
             }

crates/net/network/src/network.rs

@@ -11,7 +11,7 @@ use reth_interfaces::{
 };
 use reth_net_common::bandwidth_meter::BandwidthMeter;
 use reth_network_api::{
-    NetworkError, NetworkInfo, PeerKind, Peers, PeersInfo, ReputationChangeKind,
+    NetworkError, NetworkInfo, PeerKind, Peers, PeersInfo, Reputation, ReputationChangeKind,
 };
 use reth_primitives::{Head, NodeRecord, PeerId, TransactionSigned, H256};
 use reth_rpc_types::NetworkStatus;
@@ -184,6 +184,7 @@ impl PeersInfo for NetworkHandle {
     }
 }
 
+#[async_trait]
 impl Peers for NetworkHandle {
     /// Sends a message to the [`NetworkManager`](crate::NetworkManager) to add a peer to the known
     /// set, with the given kind.
@@ -213,6 +214,12 @@ impl Peers for NetworkHandle {
     fn reputation_change(&self, peer_id: PeerId, kind: ReputationChangeKind) {
         self.send_message(NetworkHandleMessage::ReputationChange(peer_id, kind));
     }
+
+    async fn reputation_by_id(&self, peer_id: PeerId) -> Result<Option<Reputation>, NetworkError> {
+        let (tx, rx) = oneshot::channel();
+        let _ = self.manager().send(NetworkHandleMessage::GetReputationById(peer_id, tx));
+        Ok(rx.await?)
+    }
 }
 
 #[async_trait]
@@ -314,6 +321,8 @@ pub(crate) enum NetworkHandleMessage {
     GetPeerInfo(oneshot::Sender<Vec<PeerInfo>>),
     /// Get PeerInfo for a specific peer
     GetPeerInfoById(PeerId, oneshot::Sender<Option<PeerInfo>>),
+    /// Get the reputation for a specific peer
+    GetReputationById(PeerId, oneshot::Sender<Option<Reputation>>),
     /// Gracefully shutdown network
     Shutdown(oneshot::Sender<()>),
 }

crates/net/network/src/peers/manager.rs

@@ -321,11 +321,20 @@ impl PeersManager {
         }
     }
 
+    pub(crate) fn get_reputation(&self, peer_id: &PeerId) -> Option<i32> {
+        self.peers.get(peer_id).map(|peer| peer.reputation)
+    }
+
     /// Apply the corresponding reputation change to the given peer
     pub(crate) fn apply_reputation_change(&mut self, peer_id: &PeerId, rep: ReputationChangeKind) {
-        let reputation_change = self.reputation_weights.change(rep);
         let outcome = if let Some(peer) = self.peers.get_mut(peer_id) {
-            peer.apply_reputation(reputation_change.as_i32())
+            // First check if we should reset the reputation
+            if rep.is_reset() {
+                peer.reset_reputation()
+            } else {
+                let reputation_change = self.reputation_weights.change(rep);
+                peer.apply_reputation(reputation_change.as_i32())
+            }
         } else {
             return
         };
@@ -844,13 +853,21 @@ impl Peer {
         Self { kind, ..Self::new(addr) }
     }
 
+    /// Resets the reputation of the peer to the default value. This always returns
+    /// [`ReputationChangeOutcome::None`].
+    fn reset_reputation(&mut self) -> ReputationChangeOutcome {
+        self.reputation = DEFAULT_REPUTATION;
+
+        ReputationChangeOutcome::None
+    }
+
     /// Applies a reputation change to the peer and returns what action should be taken.
     fn apply_reputation(&mut self, reputation: i32) -> ReputationChangeOutcome {
         let previous = self.reputation;
         // we add reputation since negative reputation change decrease total reputation
         self.reputation = previous.saturating_add(reputation);
 
-        trace!(target: "net::peers", repuation=%self.reputation, banned=%self.is_banned(), "applied reputation change");
+        trace!(target: "net::peers", reputation=%self.reputation, banned=%self.is_banned(), "applied reputation change");
 
         if self.state.is_connected() && self.is_banned() {
             self.state.disconnect();
@@ -1695,6 +1712,21 @@ mod test {
         }
     }
 
+    #[tokio::test]
+    async fn test_reputation_management() {
+        let peer = PeerId::random();
+        let socket_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 1, 2)), 8008);
+        let mut peers = PeersManager::default();
+        peers.add_peer(peer, socket_addr, None);
+        assert_eq!(peers.get_reputation(&peer), Some(0));
+
+        peers.apply_reputation_change(&peer, ReputationChangeKind::Other(1024));
+        assert_eq!(peers.get_reputation(&peer), Some(1024));
+
+        peers.apply_reputation_change(&peer, ReputationChangeKind::Reset);
+        assert_eq!(peers.get_reputation(&peer), Some(0));
+    }
+
     #[tokio::test]
     async fn test_remove_discovered_active() {
         let peer = PeerId::random();

crates/net/network/src/peers/reputation.rs

@@ -70,6 +70,7 @@ impl ReputationChangeWeights {
             ReputationChangeKind::BadProtocol => self.bad_protocol.into(),
             ReputationChangeKind::FailedToConnect => self.failed_to_connect.into(),
             ReputationChangeKind::Dropped => self.dropped.into(),
+            ReputationChangeKind::Reset => DEFAULT_REPUTATION.into(),
             ReputationChangeKind::Other(val) => val.into(),
         }
     }