PeerManager: move the check for banned peers from connection_established (#4569)

## Issue Addressed
https://github.com/sigp/lighthouse/issues/4543
## Proposed Changes
- Removes `NotBanned` from `BanResult`, implements `Display` and `std::error::Error` for `BanResult` and changes `ban_result` return type to `Option<BanResult>` which helps returning `BanResult` on `handle_established_inbound_connection`  
- moves the check from for banned peers from `on_connection_established` to `handle_established_inbound_connection` to start addressing #4543.
- Removes `allow_block_list` as it's now redundant? Not sure about this one but if `PeerManager` keeps track of the banned peers, no need to send a `Swarm` event for `alow_block_list` to also keep that list right? 
 
## Questions

-  #4543 refers:
>  More specifically, implement the connection limit behaviour inside the peer manager.

@AgeManning do you mean copying `libp2p::connection_limits::Behaviour`'s code into `PeerManager`/ having it as an inner `NetworkBehaviour` of `PeerManager`/other? If it's the first two, I think it probably makes more sense to have it as it is as it's less code to maintain.

> Also implement the banning of peers inside the behaviour, rather than passing messages back up to the swarm.

I tried to achieve this, but we still need to pass the `PeerManagerEvent::Banned` swarm event as `DiscV5` handles it's node and ip management internally and I did not find a method to query if a peer is banned. Is there anything else we can do from here?

3397612160/beacon_node/lighthouse_network/src/discovery/mod.rs (L931-L940)

Same as the question above, I did not find a way to check if `DiscV5` has the peer banned, so that we could check here and avoid sending `Swarm` events

3397612160/beacon_node/lighthouse_network/src/peer_manager/network_behaviour.rs (L168-L178)

Is there a chance we try to dial a peer that has been banned previously? 

Thanks!
This commit is contained in:
João Oliveira 2023-10-03 23:59:32 +00:00
parent 7605494791
commit 0dc95a1d37
6 changed files with 107 additions and 113 deletions

View File

@ -567,8 +567,6 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
if let Ok(node_id) = peer_id_to_node_id(peer_id) { if let Ok(node_id) = peer_id_to_node_id(peer_id) {
// If we could convert this peer id, remove it from the DHT and ban it from discovery. // If we could convert this peer id, remove it from the DHT and ban it from discovery.
self.discv5.ban_node(&node_id, None); self.discv5.ban_node(&node_id, None);
// Remove the node from the routing table.
self.discv5.remove_node(&node_id);
} }
for ip_address in ip_addresses { for ip_address in ip_addresses {

View File

@ -415,7 +415,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
/// Reports if a peer is banned or not. /// Reports if a peer is banned or not.
/// ///
/// This is used to determine if we should accept incoming connections. /// This is used to determine if we should accept incoming connections.
pub fn ban_status(&self, peer_id: &PeerId) -> BanResult { pub fn ban_status(&self, peer_id: &PeerId) -> Option<BanResult> {
self.network_globals.peers.read().ban_status(peer_id) self.network_globals.peers.read().ban_status(peer_id)
} }
@ -803,7 +803,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
) -> bool { ) -> bool {
{ {
let mut peerdb = self.network_globals.peers.write(); let mut peerdb = self.network_globals.peers.write();
if !matches!(peerdb.ban_status(peer_id), BanResult::NotBanned) { if peerdb.ban_status(peer_id).is_some() {
// don't connect if the peer is banned // don't connect if the peer is banned
error!(self.log, "Connection has been allowed to a banned peer"; "peer_id" => %peer_id); error!(self.log, "Connection has been allowed to a banned peer"; "peer_id" => %peer_id);
} }

View File

@ -1,5 +1,6 @@
//! Implementation of [`NetworkBehaviour`] for the [`PeerManager`]. //! Implementation of [`NetworkBehaviour`] for the [`PeerManager`].
use std::net::IpAddr;
use std::task::{Context, Poll}; use std::task::{Context, Poll};
use futures::StreamExt; use futures::StreamExt;
@ -8,17 +9,17 @@ use libp2p::identity::PeerId;
use libp2p::swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm}; use libp2p::swarm::behaviour::{ConnectionClosed, ConnectionEstablished, DialFailure, FromSwarm};
use libp2p::swarm::dial_opts::{DialOpts, PeerCondition}; use libp2p::swarm::dial_opts::{DialOpts, PeerCondition};
use libp2p::swarm::dummy::ConnectionHandler; use libp2p::swarm::dummy::ConnectionHandler;
use libp2p::swarm::{ConnectionId, NetworkBehaviour, PollParameters, ToSwarm}; use libp2p::swarm::{ConnectionDenied, ConnectionId, NetworkBehaviour, PollParameters, ToSwarm};
use slog::{debug, error}; use slog::{debug, error, trace};
use types::EthSpec; use types::EthSpec;
use crate::discovery::enr_ext::EnrExt; use crate::discovery::enr_ext::EnrExt;
use crate::peer_manager::peerdb::BanResult;
use crate::rpc::GoodbyeReason; use crate::rpc::GoodbyeReason;
use crate::types::SyncState; use crate::types::SyncState;
use crate::{metrics, ClearDialError}; use crate::{metrics, ClearDialError};
use super::peerdb::BanResult; use super::{ConnectingType, PeerManager, PeerManagerEvent};
use super::{ConnectingType, PeerManager, PeerManagerEvent, ReportSource};
impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> { impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
type ConnectionHandler = ConnectionHandler; type ConnectionHandler = ConnectionHandler;
@ -169,26 +170,64 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
} }
} }
fn handle_pending_inbound_connection(
&mut self,
_connection_id: ConnectionId,
_local_addr: &libp2p::Multiaddr,
remote_addr: &libp2p::Multiaddr,
) -> Result<(), ConnectionDenied> {
// get the IP address to verify it's not banned.
let ip = match remote_addr.iter().next() {
Some(libp2p::multiaddr::Protocol::Ip6(ip)) => IpAddr::V6(ip),
Some(libp2p::multiaddr::Protocol::Ip4(ip)) => IpAddr::V4(ip),
_ => {
return Err(ConnectionDenied::new(format!(
"Connection to peer rejected: invalid multiaddr: {remote_addr}"
)))
}
};
if self.network_globals.peers.read().is_ip_banned(&ip) {
return Err(ConnectionDenied::new(format!(
"Connection to peer rejected: peer {ip} is banned"
)));
}
Ok(())
}
fn handle_established_inbound_connection( fn handle_established_inbound_connection(
&mut self, &mut self,
_connection_id: ConnectionId, _connection_id: ConnectionId,
_peer: PeerId, peer_id: PeerId,
_local_addr: &libp2p::Multiaddr, _local_addr: &libp2p::Multiaddr,
_remote_addr: &libp2p::Multiaddr, remote_addr: &libp2p::Multiaddr,
) -> Result<libp2p::swarm::THandler<Self>, libp2p::swarm::ConnectionDenied> { ) -> Result<libp2p::swarm::THandler<Self>, ConnectionDenied> {
// TODO: we might want to check if we accept this peer or not in the future. trace!(self.log, "Inbound connection"; "peer_id" => %peer_id, "multiaddr" => %remote_addr);
// We already checked if the peer was banned on `handle_pending_inbound_connection`.
if let Some(BanResult::BadScore) = self.ban_status(&peer_id) {
return Err(ConnectionDenied::new(
"Connection to peer rejected: peer has a bad score",
));
}
Ok(ConnectionHandler) Ok(ConnectionHandler)
} }
fn handle_established_outbound_connection( fn handle_established_outbound_connection(
&mut self, &mut self,
_connection_id: ConnectionId, _connection_id: ConnectionId,
_peer: PeerId, peer_id: PeerId,
_addr: &libp2p::Multiaddr, addr: &libp2p::Multiaddr,
_role_override: libp2p::core::Endpoint, _role_override: libp2p::core::Endpoint,
) -> Result<libp2p::swarm::THandler<Self>, libp2p::swarm::ConnectionDenied> { ) -> Result<libp2p::swarm::THandler<Self>, libp2p::swarm::ConnectionDenied> {
// TODO: we might want to check if we accept this peer or not in the future. trace!(self.log, "Outbound connection"; "peer_id" => %peer_id, "multiaddr" => %addr);
Ok(ConnectionHandler) match self.ban_status(&peer_id) {
Some(cause) => {
error!(self.log, "Connected a banned peer. Rejecting connection"; "peer_id" => %peer_id);
Err(ConnectionDenied::new(cause))
}
None => Ok(ConnectionHandler),
}
} }
} }
@ -215,10 +254,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
// increment prometheus metrics // increment prometheus metrics
if self.metrics_enabled { if self.metrics_enabled {
let remote_addr = match endpoint { let remote_addr = endpoint.get_remote_address();
ConnectedPoint::Dialer { address, .. } => address,
ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
};
match remote_addr.iter().find(|proto| { match remote_addr.iter().find(|proto| {
matches!( matches!(
proto, proto,
@ -241,28 +277,6 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT); metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT);
} }
// Check to make sure the peer is not supposed to be banned
match self.ban_status(&peer_id) {
// TODO: directly emit the ban event?
BanResult::BadScore => {
// This is a faulty state
error!(self.log, "Connected to a banned peer. Re-banning"; "peer_id" => %peer_id);
// Disconnect the peer.
self.goodbye_peer(&peer_id, GoodbyeReason::Banned, ReportSource::PeerManager);
// Re-ban the peer to prevent repeated errors.
self.events.push(PeerManagerEvent::Banned(peer_id, vec![]));
return;
}
BanResult::BannedIp(ip_addr) => {
// A good peer has connected to us via a banned IP address. We ban the peer and
// prevent future connections.
debug!(self.log, "Peer connected via banned IP. Banning"; "peer_id" => %peer_id, "banned_ip" => %ip_addr);
self.goodbye_peer(&peer_id, GoodbyeReason::BannedIP, ReportSource::PeerManager);
return;
}
BanResult::NotBanned => {}
}
// Count dialing peers in the limit if the peer dialed us. // Count dialing peers in the limit if the peer dialed us.
let count_dialing = endpoint.is_listener(); let count_dialing = endpoint.is_listener();
// Check the connection limits // Check the connection limits
@ -326,11 +340,7 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
// reference so that peer manager can track this peer. // reference so that peer manager can track this peer.
self.inject_disconnect(&peer_id); self.inject_disconnect(&peer_id);
let remote_addr = match endpoint { let remote_addr = endpoint.get_remote_address();
ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
ConnectedPoint::Dialer { address, .. } => address,
};
// Update the prometheus metrics // Update the prometheus metrics
if self.metrics_enabled { if self.metrics_enabled {
match remote_addr.iter().find(|proto| { match remote_addr.iter().find(|proto| {

View File

@ -3,10 +3,13 @@ use peer_info::{ConnectionDirection, PeerConnectionStatus, PeerInfo};
use rand::seq::SliceRandom; use rand::seq::SliceRandom;
use score::{PeerAction, ReportSource, Score, ScoreState}; use score::{PeerAction, ReportSource, Score, ScoreState};
use slog::{crit, debug, error, trace, warn}; use slog::{crit, debug, error, trace, warn};
use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::net::IpAddr; use std::net::IpAddr;
use std::time::Instant; use std::time::Instant;
use std::{cmp::Ordering, fmt::Display};
use std::{
collections::{HashMap, HashSet},
fmt::Formatter,
};
use sync_status::SyncStatus; use sync_status::SyncStatus;
use types::EthSpec; use types::EthSpec;
@ -136,26 +139,18 @@ impl<TSpec: EthSpec> PeerDB<TSpec> {
} }
} }
/// Returns the current [`BanResult`] of the peer. This doesn't check the connection state, rather the /// Returns the current [`BanResult`] of the peer if banned. This doesn't check the connection state, rather the
/// underlying score of the peer. A peer may be banned but still in the connected state /// underlying score of the peer. A peer may be banned but still in the connected state
/// temporarily. /// temporarily.
/// ///
/// This is used to determine if we should accept incoming connections or not. /// This is used to determine if we should accept incoming connections or not.
pub fn ban_status(&self, peer_id: &PeerId) -> BanResult { pub fn ban_status(&self, peer_id: &PeerId) -> Option<BanResult> {
if let Some(peer) = self.peers.get(peer_id) { self.peers
match peer.score_state() { .get(peer_id)
ScoreState::Banned => BanResult::BadScore, .and_then(|peer| match peer.score_state() {
_ => { ScoreState::Banned => Some(BanResult::BadScore),
if let Some(ip) = self.ip_is_banned(peer) { _ => self.ip_is_banned(peer).map(BanResult::BannedIp),
BanResult::BannedIp(ip) })
} else {
BanResult::NotBanned
}
}
}
} else {
BanResult::NotBanned
}
} }
/// Checks if the peer's known addresses are currently banned. /// Checks if the peer's known addresses are currently banned.
@ -1183,22 +1178,24 @@ pub enum BanOperation {
} }
/// When checking if a peer is banned, it can be banned for multiple reasons. /// When checking if a peer is banned, it can be banned for multiple reasons.
#[derive(Copy, Clone, Debug)]
pub enum BanResult { pub enum BanResult {
/// The peer's score is too low causing it to be banned. /// The peer's score is too low causing it to be banned.
BadScore, BadScore,
/// The peer should be banned because it is connecting from a banned IP address. /// The peer should be banned because it is connecting from a banned IP address.
BannedIp(IpAddr), BannedIp(IpAddr),
/// The peer is not banned.
NotBanned,
} }
// Helper function for unit tests impl Display for BanResult {
#[cfg(test)] fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
impl BanResult { match self {
pub fn is_banned(&self) -> bool { BanResult::BadScore => write!(f, "Peer has a bad score"),
!matches!(self, BanResult::NotBanned) BanResult::BannedIp(addr) => write!(f, "Peer address: {} is banned", addr),
} }
} }
}
impl std::error::Error for BanResult {}
#[derive(Default)] #[derive(Default)]
pub struct BannedPeersCount { pub struct BannedPeersCount {
@ -1852,11 +1849,11 @@ mod tests {
} }
//check that ip1 and ip2 are banned but ip3-5 not //check that ip1 and ip2 are banned but ip3-5 not
assert!(pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_some());
assert!(!pdb.ban_status(&p3).is_banned()); assert!(pdb.ban_status(&p3).is_none());
assert!(!pdb.ban_status(&p4).is_banned()); assert!(pdb.ban_status(&p4).is_none());
assert!(!pdb.ban_status(&p5).is_banned()); assert!(pdb.ban_status(&p5).is_none());
//ban also the last peer in peers //ban also the last peer in peers
let _ = pdb.report_peer( let _ = pdb.report_peer(
@ -1868,11 +1865,11 @@ mod tests {
pdb.inject_disconnect(&peers[BANNED_PEERS_PER_IP_THRESHOLD + 1]); pdb.inject_disconnect(&peers[BANNED_PEERS_PER_IP_THRESHOLD + 1]);
//check that ip1-ip4 are banned but ip5 not //check that ip1-ip4 are banned but ip5 not
assert!(pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_some());
assert!(pdb.ban_status(&p3).is_banned()); assert!(pdb.ban_status(&p3).is_some());
assert!(pdb.ban_status(&p4).is_banned()); assert!(pdb.ban_status(&p4).is_some());
assert!(!pdb.ban_status(&p5).is_banned()); assert!(pdb.ban_status(&p5).is_none());
//peers[0] gets unbanned //peers[0] gets unbanned
reset_score(&mut pdb, &peers[0]); reset_score(&mut pdb, &peers[0]);
@ -1880,11 +1877,11 @@ mod tests {
let _ = pdb.shrink_to_fit(); let _ = pdb.shrink_to_fit();
//nothing changed //nothing changed
assert!(pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_some());
assert!(pdb.ban_status(&p3).is_banned()); assert!(pdb.ban_status(&p3).is_some());
assert!(pdb.ban_status(&p4).is_banned()); assert!(pdb.ban_status(&p4).is_some());
assert!(!pdb.ban_status(&p5).is_banned()); assert!(pdb.ban_status(&p5).is_none());
//peers[1] gets unbanned //peers[1] gets unbanned
reset_score(&mut pdb, &peers[1]); reset_score(&mut pdb, &peers[1]);
@ -1892,11 +1889,11 @@ mod tests {
let _ = pdb.shrink_to_fit(); let _ = pdb.shrink_to_fit();
//all ips are unbanned //all ips are unbanned
assert!(!pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_none());
assert!(!pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_none());
assert!(!pdb.ban_status(&p3).is_banned()); assert!(pdb.ban_status(&p3).is_none());
assert!(!pdb.ban_status(&p4).is_banned()); assert!(pdb.ban_status(&p4).is_none());
assert!(!pdb.ban_status(&p5).is_banned()); assert!(pdb.ban_status(&p5).is_none());
} }
#[test] #[test]
@ -1921,8 +1918,8 @@ mod tests {
} }
// check ip is banned // check ip is banned
assert!(pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_some());
assert!(!pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_none());
// unban a peer // unban a peer
reset_score(&mut pdb, &peers[0]); reset_score(&mut pdb, &peers[0]);
@ -1930,8 +1927,8 @@ mod tests {
let _ = pdb.shrink_to_fit(); let _ = pdb.shrink_to_fit();
// check not banned anymore // check not banned anymore
assert!(!pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_none());
assert!(!pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_none());
// unban all peers // unban all peers
for p in &peers { for p in &peers {
@ -1950,8 +1947,8 @@ mod tests {
} }
// both IP's are now banned // both IP's are now banned
assert!(pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_some());
// unban all peers // unban all peers
for p in &peers { for p in &peers {
@ -1967,16 +1964,16 @@ mod tests {
} }
// nothing is banned // nothing is banned
assert!(!pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_none());
assert!(!pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_none());
// reban last peer // reban last peer
let _ = pdb.report_peer(&peers[0], PeerAction::Fatal, ReportSource::PeerManager, ""); let _ = pdb.report_peer(&peers[0], PeerAction::Fatal, ReportSource::PeerManager, "");
pdb.inject_disconnect(&peers[0]); pdb.inject_disconnect(&peers[0]);
//Ip's are banned again //Ip's are banned again
assert!(pdb.ban_status(&p1).is_banned()); assert!(pdb.ban_status(&p1).is_some());
assert!(pdb.ban_status(&p2).is_banned()); assert!(pdb.ban_status(&p2).is_some());
} }
#[test] #[test]

View File

@ -20,8 +20,6 @@ where
AppReqId: ReqId, AppReqId: ReqId,
TSpec: EthSpec, TSpec: EthSpec,
{ {
/// Peers banned.
pub banned_peers: libp2p::allow_block_list::Behaviour<libp2p::allow_block_list::BlockedPeers>,
/// Keep track of active and pending connections to enforce hard limits. /// Keep track of active and pending connections to enforce hard limits.
pub connection_limits: libp2p::connection_limits::Behaviour, pub connection_limits: libp2p::connection_limits::Behaviour,
/// The routing pub-sub mechanism for eth2. /// The routing pub-sub mechanism for eth2.

View File

@ -337,11 +337,8 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
libp2p::connection_limits::Behaviour::new(limits) libp2p::connection_limits::Behaviour::new(limits)
}; };
let banned_peers = libp2p::allow_block_list::Behaviour::default();
let behaviour = { let behaviour = {
Behaviour { Behaviour {
banned_peers,
gossipsub, gossipsub,
eth2_rpc, eth2_rpc,
discovery, discovery,
@ -1402,15 +1399,10 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
Some(NetworkEvent::PeerDisconnected(peer_id)) Some(NetworkEvent::PeerDisconnected(peer_id))
} }
PeerManagerEvent::Banned(peer_id, associated_ips) => { PeerManagerEvent::Banned(peer_id, associated_ips) => {
self.swarm.behaviour_mut().banned_peers.block_peer(peer_id);
self.discovery_mut().ban_peer(&peer_id, associated_ips); self.discovery_mut().ban_peer(&peer_id, associated_ips);
None None
} }
PeerManagerEvent::UnBanned(peer_id, associated_ips) => { PeerManagerEvent::UnBanned(peer_id, associated_ips) => {
self.swarm
.behaviour_mut()
.banned_peers
.unblock_peer(peer_id);
self.discovery_mut().unban_peer(&peer_id, associated_ips); self.discovery_mut().unban_peer(&peer_id, associated_ips);
None None
} }
@ -1459,7 +1451,6 @@ impl<AppReqId: ReqId, TSpec: EthSpec> Network<AppReqId, TSpec> {
let maybe_event = match swarm_event { let maybe_event = match swarm_event {
SwarmEvent::Behaviour(behaviour_event) => match behaviour_event { SwarmEvent::Behaviour(behaviour_event) => match behaviour_event {
// Handle sub-behaviour events. // Handle sub-behaviour events.
BehaviourEvent::BannedPeers(void) => void::unreachable(void),
BehaviourEvent::Gossipsub(ge) => self.inject_gs_event(ge), BehaviourEvent::Gossipsub(ge) => self.inject_gs_event(ge),
BehaviourEvent::Eth2Rpc(re) => self.inject_rpc_event(re), BehaviourEvent::Eth2Rpc(re) => self.inject_rpc_event(re),
// Inform the peer manager about discovered peers. // Inform the peer manager about discovered peers.