From 9436f238431d61ee0b921f0aaed99468c0540566 Mon Sep 17 00:00:00 2001 From: Pawan Dhananjay Date: Mon, 3 Mar 2025 19:34:14 -0800 Subject: [PATCH] Add http endpoint to add trusted peer (#7068) * Add a trusted_peers endpoint * Maintain trusted peers in heartbeat --- beacon_node/http_api/src/lib.rs | 40 ++++++++++++++++++- .../src/peer_manager/mod.rs | 14 +++++++ .../src/peer_manager/peerdb.rs | 15 ++++++- .../src/peer_manager/peerdb/peer_info.rs | 4 +- .../lighthouse_network/src/service/mod.rs | 5 +++ .../lighthouse_network/src/types/globals.rs | 4 ++ beacon_node/network/src/service.rs | 9 ++++- common/eth2/src/types.rs | 5 +++ 8 files changed, 90 insertions(+), 6 deletions(-) diff --git a/beacon_node/http_api/src/lib.rs b/beacon_node/http_api/src/lib.rs index d0e257ce7e..53eead4855 100644 --- a/beacon_node/http_api/src/lib.rs +++ b/beacon_node/http_api/src/lib.rs @@ -53,7 +53,7 @@ use eth2::types::{ use eth2::{CONSENSUS_VERSION_HEADER, CONTENT_TYPE_HEADER, SSZ_CONTENT_TYPE_HEADER}; use health_metrics::observe::Observe; use lighthouse_network::rpc::methods::MetaData; -use lighthouse_network::{types::SyncState, EnrExt, NetworkGlobals, PeerId, PubsubMessage}; +use lighthouse_network::{types::SyncState, Enr, EnrExt, NetworkGlobals, PeerId, PubsubMessage}; use lighthouse_version::version_with_platform; use logging::SSELoggingComponents; use network::{NetworkMessage, NetworkSenders, ValidatorSubscriptionMessage}; @@ -72,6 +72,7 @@ use std::future::Future; use std::net::{IpAddr, Ipv4Addr, SocketAddr}; use std::path::PathBuf; use std::pin::Pin; +use std::str::FromStr; use std::sync::Arc; use sysinfo::{System, SystemExt}; use system_health::{observe_nat, observe_system_health_bn}; @@ -3700,7 +3701,7 @@ pub fn serve( .and(task_spawner_filter.clone()) .and(chain_filter.clone()) .and(warp_utils::json::json()) - .and(network_tx_filter) + .and(network_tx_filter.clone()) .and(log_filter.clone()) .then( |not_synced_filter: Result<(), Rejection>, @@ -4162,6 +4163,40 @@ pub fn serve( }, ); + // POST lighthouse/add_peer + let post_lighthouse_add_peer = warp::path("lighthouse") + .and(warp::path("add_peer")) + .and(warp::path::end()) + .and(warp_utils::json::json()) + .and(task_spawner_filter.clone()) + .and(network_globals.clone()) + .and(network_tx_filter.clone()) + .and(log_filter.clone()) + .then( + |request_data: api_types::AddPeer, + task_spawner: TaskSpawner, + network_globals: Arc>, + network_tx: UnboundedSender>, + log: Logger| { + task_spawner.blocking_json_task(Priority::P0, move || { + let enr = Enr::from_str(&request_data.enr).map_err(|e| { + warp_utils::reject::custom_bad_request(format!("invalid enr error {}", e)) + })?; + info!( + log, + "Adding trusted peer"; + "peer_id" => %enr.peer_id(), + "multiaddr" => ?enr.multiaddr() + ); + network_globals.add_trusted_peer(enr.clone()); + + publish_network_message(&network_tx, NetworkMessage::ConnectToPeer(enr))?; + + Ok(api_types::GenericResponse::from(())) + }) + }, + ); + // POST lighthouse/liveness let post_lighthouse_liveness = warp::path("lighthouse") .and(warp::path("liveness")) @@ -4962,6 +4997,7 @@ pub fn serve( .uor(post_lighthouse_ui_validator_metrics) .uor(post_lighthouse_ui_validator_info) .uor(post_lighthouse_finalize) + .uor(post_lighthouse_add_peer) .recover(warp_utils::reject::handle_rejection), ), ) diff --git a/beacon_node/lighthouse_network/src/peer_manager/mod.rs b/beacon_node/lighthouse_network/src/peer_manager/mod.rs index 07c4be7959..b97a5cb8ce 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/mod.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/mod.rs @@ -114,6 +114,7 @@ pub struct PeerManager { metrics_enabled: bool, /// Keeps track of whether the QUIC protocol is enabled or not. quic_enabled: bool, + trusted_peers: HashSet, /// The logger associated with the `PeerManager`. log: slog::Logger, } @@ -195,6 +196,7 @@ impl PeerManager { discovery_enabled, metrics_enabled, quic_enabled, + trusted_peers: Default::default(), log: log.clone(), }) } @@ -943,6 +945,13 @@ impl PeerManager { } } + fn maintain_trusted_peers(&mut self) { + let trusted_peers = self.trusted_peers.clone(); + for trusted_peer in &trusted_peers { + self.dial_peer(trusted_peer.clone()); + } + } + /// This function checks the status of our current peers and optionally requests a discovery /// query if we need to find more peers to maintain the current number of peers fn maintain_peer_count(&mut self, dialing_peers: usize) { @@ -1234,6 +1243,7 @@ impl PeerManager { fn heartbeat(&mut self) { // Optionally run a discovery query if we need more peers. self.maintain_peer_count(0); + self.maintain_trusted_peers(); // Cleans up the connection state of dialing peers. // Libp2p dials peer-ids, but sometimes the response is from another peer-id or libp2p @@ -1470,6 +1480,10 @@ impl PeerManager { ) }) } + + pub fn add_trusted_peer(&mut self, enr: Enr) { + self.trusted_peers.insert(enr); + } } enum ConnectingType { diff --git a/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs b/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs index 37cb5df6ea..e0218647b7 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/peerdb.rs @@ -9,7 +9,7 @@ use std::net::IpAddr; use std::time::Instant; use std::{cmp::Ordering, fmt::Display}; use std::{ - collections::{HashMap, HashSet}, + collections::{hash_map::Entry, HashMap, HashSet}, fmt::Formatter, }; use sync_status::SyncStatus; @@ -79,6 +79,19 @@ impl PeerDB { self.peers.iter() } + pub fn set_trusted_peer(&mut self, enr: Enr) { + match self.peers.entry(enr.peer_id()) { + Entry::Occupied(mut info) => { + let entry = info.get_mut(); + entry.score = Score::max_score(); + entry.is_trusted = true; + } + Entry::Vacant(entry) => { + entry.insert(PeerInfo::trusted_peer_info()); + } + } + } + /// Gives the ids of all known peers. pub fn peer_ids(&self) -> impl Iterator { self.peers.keys() diff --git a/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs b/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs index 2e8f462565..05a29ac47e 100644 --- a/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs +++ b/beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs @@ -21,7 +21,7 @@ use PeerConnectionStatus::*; #[serde(bound = "E: EthSpec")] pub struct PeerInfo { /// The peers reputation - score: Score, + pub(crate) score: Score, /// Client managing this peer client: Client, /// Connection status of this peer @@ -50,7 +50,7 @@ pub struct PeerInfo { #[serde(skip)] min_ttl: Option, /// Is the peer a trusted peer. - is_trusted: bool, + pub(crate) is_trusted: bool, /// Direction of the first connection of the last (or current) connected session with this peer. /// None if this peer was never connected. connection_direction: Option, diff --git a/beacon_node/lighthouse_network/src/service/mod.rs b/beacon_node/lighthouse_network/src/service/mod.rs index 8586fd9cd3..f1d856be80 100644 --- a/beacon_node/lighthouse_network/src/service/mod.rs +++ b/beacon_node/lighthouse_network/src/service/mod.rs @@ -1236,6 +1236,11 @@ impl Network { } } + pub fn dial_trusted_peer(&mut self, enr: Enr) { + self.peer_manager_mut().add_trusted_peer(enr.clone()); + self.peer_manager_mut().dial_peer(enr); + } + /* Sub-behaviour event handling functions */ /// Handle a gossipsub event. diff --git a/beacon_node/lighthouse_network/src/types/globals.rs b/beacon_node/lighthouse_network/src/types/globals.rs index 2800b75133..9da3d547e3 100644 --- a/beacon_node/lighthouse_network/src/types/globals.rs +++ b/beacon_node/lighthouse_network/src/types/globals.rs @@ -162,6 +162,10 @@ impl NetworkGlobals { .unwrap_or_default() } + pub fn add_trusted_peer(&self, enr: Enr) { + self.peers.write().set_trusted_peer(enr); + } + /// Updates the syncing state of the node. /// /// The old state is returned diff --git a/beacon_node/network/src/service.rs b/beacon_node/network/src/service.rs index 1b2a681c64..4b288bfe2b 100644 --- a/beacon_node/network/src/service.rs +++ b/beacon_node/network/src/service.rs @@ -14,6 +14,7 @@ use futures::StreamExt; use lighthouse_network::rpc::{RequestId, RequestType}; use lighthouse_network::service::Network; use lighthouse_network::types::GossipKind; +use lighthouse_network::Enr; use lighthouse_network::{prometheus_client::registry::Registry, MessageAcceptance}; use lighthouse_network::{ rpc::{GoodbyeReason, RpcErrorResponse}, @@ -78,7 +79,9 @@ pub enum NetworkMessage { id: PeerRequestId, }, /// Publish a list of messages to the gossipsub protocol. - Publish { messages: Vec> }, + Publish { + messages: Vec>, + }, /// Validates a received gossipsub message. This will propagate the message on the network. ValidationResult { /// The peer that sent us the message. We don't send back to this peer. @@ -101,6 +104,7 @@ pub enum NetworkMessage { reason: GoodbyeReason, source: ReportSource, }, + ConnectToPeer(Enr), } /// Messages triggered by validators that may trigger a subscription to a subnet. @@ -688,6 +692,9 @@ impl NetworkService { reason, source, } => self.libp2p.goodbye_peer(&peer_id, reason, source), + NetworkMessage::ConnectToPeer(enr) => { + self.libp2p.dial_trusted_peer(enr); + } NetworkMessage::SubscribeCoreTopics => { if self.subscribed_core_topics() { return; diff --git a/common/eth2/src/types.rs b/common/eth2/src/types.rs index 7d70d242be..5f91c0e519 100644 --- a/common/eth2/src/types.rs +++ b/common/eth2/src/types.rs @@ -1431,6 +1431,11 @@ pub struct ManualFinalizationRequestData { pub block_root: Hash256, } +#[derive(Debug, Serialize, Deserialize)] +pub struct AddPeer { + pub enr: String, +} + #[derive(Debug, Serialize, Deserialize)] pub struct LivenessRequestData { pub epoch: Epoch,