feat: add validation for peerinfo

Author: varex83

Cargo.lock

@@ -61,6 +61,7 @@ tracing-subscriber = { version = "0.3.9", features = ["env-filter"] }
 tracing-loki = "0.2.6"
 vise = "0.3.2"
 vise-exporter = "0.3.2"
+parking_lot = "0.12.2"
 
 # Crates in the workspace
 charon = { path = "crates/charon" }

Cargo.toml

@@ -18,6 +18,9 @@ tracing.workspace = true
 chrono.workspace = true
 unsigned-varint = { version = "0.8", features = ["futures"] }
 vise.workspace = true
+charon-core.workspace = true
+charon-p2p.workspace = true
+parking_lot.workspace = true
 
 [build-dependencies]
 charon-build-proto.workspace = true
@@ -28,6 +31,7 @@ clap.workspace = true
 tokio.workspace = true
 tracing-subscriber.workspace = true
 hex.workspace = true
+vise-exporter.workspace = true
 
 [lints]
 workspace = true

crates/peerinfo/Cargo.toml

@@ -13,7 +13,7 @@
 //! Terminal 1: `cargo run --example peerinfo -p charon-peerinfo -- --port 4001`
 //! Terminal 2: `cargo run --example peerinfo -p charon-peerinfo -- --port 4002`
 #![allow(missing_docs)]
-use std::time::Duration;
+use std::{net::SocketAddr, time::Duration};
 
 use charon_peerinfo::{Behaviour, Config, Event, LocalPeerInfo};
 use clap::Parser;
@@ -26,6 +26,7 @@ use libp2p::{
 };
 use tokio::signal;
 use tracing_subscriber::EnvFilter;
+use vise_exporter::MetricsExporter;
 
 /// Command line arguments
 #[derive(Debug, Parser)]
@@ -73,9 +74,7 @@ fn build_swarm(
         )?
         .with_behaviour(|key| {
             Ok(CombinedBehaviour {
-                peer_info: Behaviour::new(
-                    Config::new(peerinfo_config, key.public().to_peer_id()).with_interval(interval),
-                ),
+                peer_info: Behaviour::new(Config::new(peerinfo_config).with_interval(interval)),
                 identify: identify::Behaviour::new(identify::Config::new(
                     "/peerinfo-example/1.0.0".to_string(),
                     key.public(),
@@ -175,6 +174,20 @@ async fn main() -> anyhow::Result<()> {
         .with_env_filter(EnvFilter::from_default_env().add_directive("debug".parse()?))
         .init();
 
+    // Run the metrics exporter
+    let bind_address = SocketAddr::from(([0, 0, 0, 0], 9465));
+
+    let exporter = MetricsExporter::default()
+        .bind(bind_address)
+        .await
+        .expect("Failed to bind metrics exporter");
+    tokio::spawn(async move {
+        exporter
+            .start()
+            .await
+            .expect("Failed to start metrics exporter");
+    });
+
     let args = Args::parse();
 
     // Create local peer info

crates/peerinfo/examples/peerinfo.rs

@@ -77,8 +77,6 @@ impl Behaviour {
             chrono::Utc::now().timestamp()
         };
 
-        let started_at = started_at as f64;
-
         PEERINFO_METRICS.start_time_secs[&name].set(started_at);
 
         if config.local_info().builder_api_enabled {
@@ -87,6 +85,11 @@ impl Behaviour {
             PEERINFO_METRICS.builder_api_enabled[&name].set(0);
         }
 
+        for (idx, peer) in config.peers().iter().enumerate() {
+            let peer_name = charon_p2p::name::peer_name(peer);
+            PEERINFO_METRICS.index[&peer_name].set(idx);
+        }
+
         Self {
             config,
             events: VecDeque::new(),
@@ -106,22 +109,22 @@ impl NetworkBehaviour for Behaviour {
     fn handle_established_inbound_connection(
         &mut self,
         _connection_id: ConnectionId,
-        _peer: PeerId,
+        peer: PeerId,
         _local_addr: &Multiaddr,
         _remote_addr: &Multiaddr,
     ) -> Result<THandler<Self>, ConnectionDenied> {
-        Ok(Handler::new(self.config.clone()))
+        Ok(Handler::new(self.config.clone(), peer))
     }
 
     fn handle_established_outbound_connection(
         &mut self,
         _connection_id: ConnectionId,
-        _peer: PeerId,
+        peer: PeerId,
         _addr: &Multiaddr,
         _role_override: libp2p::core::Endpoint,
         _port_use: libp2p::core::transport::PortUse,
     ) -> Result<THandler<Self>, ConnectionDenied> {
-        Ok(Handler::new(self.config.clone()))
+        Ok(Handler::new(self.config.clone(), peer))
     }
 
     fn on_swarm_event(&mut self, _event: FromSwarm) {

crates/peerinfo/src/behaviour.rs

@@ -22,8 +22,8 @@ pub struct Config {
     interval: Duration,
     /// Local peer info to send to other peers.
     local_info: LocalPeerInfo,
-    /// Peer ID of the local peer.
-    peer_id: PeerId,
+    /// Known peers.
+    peers: Vec<PeerId>,
 }
 
 /// Local peer information to be shared with other peers.
@@ -74,7 +74,7 @@ impl LocalPeerInfo {
             git_hash: self.git_hash.clone(),
             sent_at: Some(Timestamp {
                 seconds: now.timestamp(),
-                nanos: 0,
+                nanos: i32::try_from(now.timestamp_subsec_nanos()).unwrap_or(0),
             }),
             started_at: self.started_at,
             builder_api_enabled: self.builder_api_enabled,
@@ -94,12 +94,12 @@ impl Config {
     /// * A peer info request is sent every 60 seconds on a healthy connection.
     /// * Every request must yield a response within 20 seconds to be
     ///   successful.
-    pub fn new(local_info: LocalPeerInfo, peer_id: PeerId) -> Self {
+    pub fn new(local_info: LocalPeerInfo) -> Self {
         Self {
             timeout: DEFAULT_TIMEOUT,
             interval: DEFAULT_INTERVAL,
             local_info,
-            peer_id,
+            peers: Vec::new(),
         }
     }
 
@@ -121,9 +121,9 @@ impl Config {
         self
     }
 
-    /// Sets the peer ID.
-    pub fn with_peer_id(mut self, peer_id: PeerId) -> Self {
-        self.peer_id = peer_id;
+    /// Sets the known peers.
+    pub fn with_peers(mut self, peers: Vec<PeerId>) -> Self {
+        self.peers = peers;
         self
     }
 
@@ -132,11 +132,6 @@ impl Config {
         &self.local_info
     }
 
-    /// Returns the peer ID.
-    pub fn peer_id(&self) -> &PeerId {
-        &self.peer_id
-    }
-
     /// Returns the timeout.
     pub fn timeout(&self) -> Duration {
         self.timeout
@@ -146,4 +141,9 @@ impl Config {
     pub fn interval(&self) -> Duration {
         self.interval
     }
+
+    /// Returns the known peers.
+    pub fn peers(&self) -> &[PeerId] {
+        &self.peers
+    }
 }

crates/peerinfo/src/config.rs

@@ -8,13 +8,15 @@
 use std::{
     collections::VecDeque,
     convert::Infallible,
+    sync::Arc,
     task::{Context, Poll},
     time::Duration,
 };
 
 use futures::{future::BoxFuture, prelude::*};
 use futures_timer::Delay;
 use libp2p::{
+    PeerId,
     core::upgrade::ReadyUpgrade,
     swarm::{
         ConnectionHandler, ConnectionHandlerEvent, Stream, StreamProtocol, StreamUpgradeError,
@@ -26,7 +28,8 @@ use libp2p::{
 };
 
 use crate::{
-    PROTOCOL_NAME, config::Config, failure::Failure, peerinfopb::v1::peerinfo::PeerInfo, protocol,
+    PROTOCOL_NAME, config::Config, failure::Failure, peerinfopb::v1::peerinfo::PeerInfo,
+    protocol::ProtocolState,
 };
 
 /// Result of a successful peer info exchange.
@@ -55,6 +58,8 @@ pub struct Handler {
     inbound: Option<InboundFuture>,
     /// Tracks the state of our handler.
     state: State,
+    /// The protocol state.
+    protocol: Arc<ProtocolState>,
 }
 
 /// Tracks the state of the handler.
@@ -71,8 +76,9 @@ enum State {
 
 impl Handler {
     /// Builds a new [`Handler`] with the given configuration.
-    pub fn new(config: Config) -> Self {
+    pub fn new(config: Config, peer: PeerId) -> Self {
         let interval = config.interval();
+        let local_info = config.local_info().clone();
         Handler {
             config,
             interval: Delay::new(interval),
@@ -81,6 +87,7 @@ impl Handler {
             outbound: None,
             inbound: None,
             state: State::Active,
+            protocol: Arc::new(ProtocolState::new(peer, local_info)),
         }
     }
 
@@ -158,8 +165,14 @@ impl ConnectionHandler for Handler {
                 }
                 Poll::Ready(Ok((stream, _request))) => {
                     tracing::trace!("Answered inbound peerinfo request from peer");
-                    self.inbound =
-                        Some(recv_peer_info(stream, self.config.local_info().to_proto()).boxed());
+                    self.inbound = Some(
+                        recv_peer_info(
+                            self.protocol.clone(),
+                            stream,
+                            self.config.local_info().to_proto(),
+                        )
+                        .boxed(),
+                    );
                 }
             }
         }
@@ -207,6 +220,7 @@ impl ConnectionHandler for Handler {
                     Poll::Ready(_) => {
                         self.outbound = Some(OutboundState::Request(
                             send_peer_info(
+                                self.protocol.clone(),
                                 stream,
                                 self.config.local_info().to_proto(),
                                 self.config.timeout(),
@@ -246,7 +260,8 @@ impl ConnectionHandler for Handler {
             }) => {
                 stream.ignore_for_keep_alive();
                 let local_info = self.config.local_info().to_proto();
-                self.inbound = Some(recv_peer_info(stream, local_info).boxed());
+                self.inbound =
+                    Some(recv_peer_info(self.protocol.clone(), stream, local_info).boxed());
             }
             ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound {
                 protocol: mut stream,
@@ -256,7 +271,13 @@ impl ConnectionHandler for Handler {
                 self.interval.reset(Duration::new(0, 0));
                 let request = self.config.local_info().to_proto();
                 self.outbound = Some(OutboundState::Request(
-                    send_peer_info(stream, request, self.config.timeout()).boxed(),
+                    send_peer_info(
+                        self.protocol.clone(),
+                        stream,
+                        request,
+                        self.config.timeout(),
+                    )
+                    .boxed(),
                 ));
             }
             ConnectionEvent::DialUpgradeError(dial_upgrade_error) => {
@@ -282,11 +303,12 @@ enum OutboundState {
 
 /// A wrapper around [`protocol::send_peer_info`] that enforces a timeout.
 async fn send_peer_info(
+    protocol: Arc<ProtocolState>,
     stream: Stream,
     request: PeerInfo,
     timeout: Duration,
 ) -> Result<(Stream, PeerInfo), Failure> {
-    let send = protocol::send_peer_info(stream, &request);
+    let send = protocol.send_peer_info(stream, &request);
     futures::pin_mut!(send);
 
     match future::select(send, Delay::new(timeout)).await {
@@ -299,8 +321,9 @@ async fn send_peer_info(
 /// A wrapper around [`protocol::recv_peer_info`] that returns only the stream
 /// and request (for use in inbound handling).
 async fn recv_peer_info(
+    protocol: Arc<ProtocolState>,
     stream: Stream,
     local_info: PeerInfo,
 ) -> Result<(Stream, PeerInfo), std::io::Error> {
-    protocol::recv_peer_info(stream, &local_info).await
+    protocol.recv_peer_info(stream, &local_info).await
 }

crates/peerinfo/src/handler.rs

@@ -8,7 +8,7 @@ use vise::*;
 pub struct PeerInfoMetrics {
     /// Peer clock offset in seconds.
     #[metrics(labels = ["peer"])]
-    pub clock_offset_seconds: LabeledFamily<String, Gauge<f64>>,
+    pub clock_offset_seconds: LabeledFamily<String, Gauge<i64>>,
 
     /// Constant gauge with version label set to peer's charon version.
     pub version: Family<PeerVersionLabels, Gauge>,
@@ -18,11 +18,11 @@ pub struct PeerInfoMetrics {
 
     /// Constant gauge set to the peer start time of the binary in unix seconds.
     #[metrics(labels = ["peer"])]
-    pub start_time_secs: LabeledFamily<String, Gauge<f64>>,
+    pub start_time_secs: LabeledFamily<String, Gauge<i64>>,
 
     /// Constant gauge set to the peer index in the cluster definition.
     #[metrics(labels = ["peer"])]
-    pub index: LabeledFamily<String, Gauge>,
+    pub index: LabeledFamily<String, Gauge<usize>>,
 
     /// Set to 1 if the peer's version is supported by (compatible with) the
     /// current version, else 0 if unsupported.