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

@@ -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<T: BeaconChainTypes>(
         .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<T: BeaconChainTypes>(
             },
         );
 
+    // 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<T::EthSpec>,
+             network_globals: Arc<NetworkGlobals<T::EthSpec>>,
+             network_tx: UnboundedSender<NetworkMessage<T::EthSpec>>,
+             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<T: BeaconChainTypes>(
                     .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),
             ),
         )

beacon_node/lighthouse_network/src/peer_manager/mod.rs

@@ -114,6 +114,7 @@ pub struct PeerManager<E: EthSpec> {
     metrics_enabled: bool,
     /// Keeps track of whether the QUIC protocol is enabled or not.
     quic_enabled: bool,
+    trusted_peers: HashSet<Enr>,
     /// The logger associated with the `PeerManager`.
     log: slog::Logger,
 }
@@ -195,6 +196,7 @@ impl<E: EthSpec> PeerManager<E> {
             discovery_enabled,
             metrics_enabled,
             quic_enabled,
+            trusted_peers: Default::default(),
             log: log.clone(),
         })
     }
@@ -943,6 +945,13 @@ impl<E: EthSpec> PeerManager<E> {
         }
     }
 
+    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<E: EthSpec> PeerManager<E> {
     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<E: EthSpec> PeerManager<E> {
             )
         })
     }
+
+    pub fn add_trusted_peer(&mut self, enr: Enr) {
+        self.trusted_peers.insert(enr);
+    }
 }
 
 enum ConnectingType {

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<E: EthSpec> PeerDB<E> {
         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<Item = &PeerId> {
         self.peers.keys()

beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs

@@ -21,7 +21,7 @@ use PeerConnectionStatus::*;
 #[serde(bound = "E: EthSpec")]
 pub struct PeerInfo<E: EthSpec> {
     /// 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<E: EthSpec> {
     #[serde(skip)]
     min_ttl: Option<Instant>,
     /// 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<ConnectionDirection>,

beacon_node/lighthouse_network/src/service/mod.rs

@@ -1236,6 +1236,11 @@ impl<E: EthSpec> Network<E> {
         }
     }
 
+    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.

beacon_node/lighthouse_network/src/types/globals.rs

@@ -162,6 +162,10 @@ impl<E: EthSpec> NetworkGlobals<E> {
             .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

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<E: EthSpec> {
         id: PeerRequestId,
     },
     /// Publish a list of messages to the gossipsub protocol.
-    Publish { messages: Vec<PubsubMessage<E>> },
+    Publish {
+        messages: Vec<PubsubMessage<E>>,
+    },
     /// 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<E: EthSpec> {
         reason: GoodbyeReason,
         source: ReportSource,
     },
+    ConnectToPeer(Enr),
 }
 
 /// Messages triggered by validators that may trigger a subscription to a subnet.
@@ -688,6 +692,9 @@ impl<T: BeaconChainTypes> NetworkService<T> {
                 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;

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,