feat: add connection gater (#99)

* feat: add tracing

* fix: linting, cargo-deny

* feat: add p2p config

* fix: remove comment

* feat: add p2p metrics

* fix: rename label in example

* fix: linting

* feat: [wip] add relay/ping

* feat: add behaviours module, add relay client support

* fix: linter warnings

* fix: remove wildcard dependencies

* feat: add connection gater

* fix: linter warnings

* fix: linter warnings

Author: varex83

crates/charon-p2p/examples/p2p.rs

@@ -40,7 +40,7 @@ async fn main() -> Result<()> {
     let mut p2p: Node<_> = Node::new(
         P2PConfig::default(),
         key.clone(),
-        ConnGater,
+        ConnGater::new_open_gater(),
         false,
         NodeType::QUIC,
         PlutoMdnsBehaviour::new,

crates/charon-p2p/src/behaviours/pluto.rs

@@ -4,8 +4,12 @@ use std::time::Duration;
 
 use libp2p::{identify, identity::Keypair, ping, relay, swarm::NetworkBehaviour};
 
+use crate::gater::ConnGater;
+
 #[derive(NetworkBehaviour)]
 pub struct PlutoBehaviour {
+    /// Connection gater behaviour.
+    pub gater: ConnGater,
     /// Relay client behaviour.
     pub relay: relay::client::Behaviour,
     /// Identify behaviour.
@@ -28,6 +32,7 @@ impl PlutoBehaviour {
                     .with_interval(Duration::from_secs(1))
                     .with_timeout(Duration::from_secs(2)),
             ),
+            gater: ConnGater::new_conn_gater(vec![], vec![]),
         }
     }
 }

crates/charon-p2p/src/gater.rs

@@ -0,0 +1,104 @@
+//! Connection handler for the gater behaviour.
+//!
+//! This is a dummy handler since the gater doesn't need to negotiate any
+//! protocols or handle any connection-level events. The actual gating logic
+//! happens at the connection establishment phase in the `NetworkBehaviour`
+//! implementation.
+
+use std::{
+    convert::Infallible,
+    task::{Context, Poll},
+};
+
+use libp2p::swarm::{
+    ConnectionHandler, ConnectionHandlerEvent, Stream, SubstreamProtocol, handler::ConnectionEvent,
+};
+
+/// Dummy connection handler for the gater.
+///
+/// This handler doesn't negotiate any protocols or handle any events.
+/// It exists only to satisfy the `NetworkBehaviour` trait requirements.
+#[derive(Debug, Clone, Default)]
+pub struct Handler {
+    _private: (),
+}
+
+impl Handler {
+    /// Creates a new handler.
+    pub fn new() -> Self {
+        Self { _private: () }
+    }
+}
+
+impl ConnectionHandler for Handler {
+    type FromBehaviour = Infallible;
+    type InboundOpenInfo = ();
+    type InboundProtocol = DeniedUpgrade;
+    type OutboundOpenInfo = Infallible;
+    type OutboundProtocol = DeniedUpgrade;
+    type ToBehaviour = Infallible;
+
+    fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
+        SubstreamProtocol::new(DeniedUpgrade, ())
+    }
+
+    fn poll(
+        &mut self,
+        _cx: &mut Context<'_>,
+    ) -> Poll<
+        ConnectionHandlerEvent<Self::OutboundProtocol, Self::OutboundOpenInfo, Self::ToBehaviour>,
+    > {
+        Poll::Pending
+    }
+
+    fn on_behaviour_event(&mut self, event: Self::FromBehaviour) {
+        match event {}
+    }
+
+    fn on_connection_event(
+        &mut self,
+        _event: ConnectionEvent<
+            Self::InboundProtocol,
+            Self::OutboundProtocol,
+            Self::InboundOpenInfo,
+            Self::OutboundOpenInfo,
+        >,
+    ) {
+        // No events to handle
+    }
+}
+
+/// A protocol upgrade that always denies the upgrade.
+///
+/// This is used because the gater doesn't need to negotiate any protocols.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct DeniedUpgrade;
+
+impl libp2p::core::UpgradeInfo for DeniedUpgrade {
+    type Info = &'static str;
+    type InfoIter = std::iter::Empty<Self::Info>;
+
+    fn protocol_info(&self) -> Self::InfoIter {
+        std::iter::empty()
+    }
+}
+
+impl libp2p::core::upgrade::InboundUpgrade<Stream> for DeniedUpgrade {
+    type Error = Infallible;
+    type Future = std::future::Pending<Result<Self::Output, Self::Error>>;
+    type Output = Infallible;
+
+    fn upgrade_inbound(self, _: Stream, _: Self::Info) -> Self::Future {
+        std::future::pending()
+    }
+}
+
+impl libp2p::core::upgrade::OutboundUpgrade<Stream> for DeniedUpgrade {
+    type Error = Infallible;
+    type Future = std::future::Pending<Result<Self::Output, Self::Error>>;
+    type Output = Infallible;
+
+    fn upgrade_outbound(self, _: Stream, _: Self::Info) -> Self::Future {
+        std::future::pending()
+    }
+}

crates/charon-p2p/src/gater/handler.rs

@@ -0,0 +1,207 @@
+//! Gater is responsible for whitelisting / blacklisting peers.
+//!
+//! This module provides connection gating functionality that limits access to
+//! cluster peers and relays. In Rust libp2p, connection gating is implemented
+//! via the `NetworkBehaviour` trait, specifically through the
+//! `handle_established_inbound_connection` and
+//! `handle_established_outbound_connection` methods which can reject
+//! connections by returning `ConnectionDenied`.
+
+use std::{
+    collections::{HashSet, VecDeque},
+    sync::Arc,
+    task::{Context, Poll},
+};
+
+use libp2p::{
+    Multiaddr, PeerId,
+    swarm::{
+        ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, THandler, THandlerInEvent,
+        THandlerOutEvent, ToSwarm,
+    },
+};
+
+use crate::peer::MutablePeer;
+
+mod handler;
+
+/// Configuration for the connection gater.
+#[derive(Clone, Default)]
+pub struct Config {
+    peer_ids: HashSet<PeerId>,
+    relays: Vec<Arc<MutablePeer>>,
+    open: bool,
+}
+
+impl Config {
+    /// Creates a new open gater configuration that does not gate any
+    /// connections.
+    pub fn open() -> Self {
+        Self {
+            peer_ids: HashSet::new(),
+            relays: Vec::new(),
+            open: true,
+        }
+    }
+
+    /// Creates a new closed gater configuration that gates all connections
+    /// except those explicitly allowed.
+    pub fn closed() -> Self {
+        Self {
+            peer_ids: HashSet::new(),
+            relays: Vec::new(),
+            open: false,
+        }
+    }
+
+    /// Sets the allowed peer IDs.
+    pub fn with_peer_ids(mut self, peer_ids: Vec<PeerId>) -> Self {
+        self.peer_ids = peer_ids.into_iter().collect();
+        self
+    }
+
+    /// Sets the relay peers.
+    pub fn with_relays(mut self, relays: Vec<Arc<MutablePeer>>) -> Self {
+        self.relays = relays;
+        self
+    }
+}
+
+/// ConnGater filters incoming and outgoing connections by the cluster peers.
+#[derive(Clone, Default)]
+pub struct ConnGater {
+    config: Config,
+    events: VecDeque<Event>,
+}
+
+impl ConnGater {
+    /// Creates a new connection gater with the given configuration.
+    pub fn new(config: Config) -> Self {
+        Self {
+            config,
+            events: VecDeque::new(),
+        }
+    }
+
+    /// Creates a new connection gater that limits access to the cluster peers
+    /// and relays.
+    pub fn new_conn_gater(peers: Vec<PeerId>, relays: Vec<Arc<MutablePeer>>) -> Self {
+        Self {
+            config: Config::closed().with_peer_ids(peers).with_relays(relays),
+            events: VecDeque::new(),
+        }
+    }
+
+    /// Creates a new open gater that does not gate any connections.
+    pub fn new_open_gater() -> Self {
+        Self {
+            config: Config::open(),
+            events: VecDeque::new(),
+        }
+    }
+
+    /// Returns true if the gater is open (not gating any connections).
+    pub fn is_open(&self) -> bool {
+        self.config.open
+    }
+
+    /// Checks if a peer is allowed to connect.
+    fn is_peer_allowed(&self, peer_id: &PeerId) -> bool {
+        if self.config.open {
+            return true;
+        }
+
+        // Check if peer is in the allowed set
+        if self.config.peer_ids.contains(peer_id) {
+            return true;
+        }
+
+        // Check if peer is a relay
+        for relay in &self.config.relays {
+            if let Ok(Some(peer)) = relay.peer()
+                && peer.id == *peer_id
+            {
+                return true;
+            }
+        }
+
+        false
+    }
+}
+
+/// Event emitted by the connection gater behaviour.
+#[derive(Debug, Clone)]
+pub enum Event {
+    /// A peer was blocked from connecting.
+    PeerBlocked(PeerId),
+}
+
+impl NetworkBehaviour for ConnGater {
+    type ConnectionHandler = handler::Handler;
+    type ToSwarm = Event;
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        _connection_id: ConnectionId,
+        peer: PeerId,
+        _local_addr: &Multiaddr,
+        _remote_addr: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        if self.is_peer_allowed(&peer) {
+            Ok(handler::Handler::new())
+        } else {
+            self.events.push_back(Event::PeerBlocked(peer));
+            Err(ConnectionDenied::new(PeerNotAllowed(peer)))
+        }
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _connection_id: ConnectionId,
+        _peer: PeerId,
+        _addr: &Multiaddr,
+        _role_override: libp2p::core::Endpoint,
+        _port_use: libp2p::core::transport::PortUse,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        // Allow all outbound connections
+        Ok(handler::Handler::new())
+    }
+
+    fn on_swarm_event(&mut self, _event: FromSwarm) {
+        // No special handling needed for swarm events
+    }
+
+    fn on_connection_handler_event(
+        &mut self,
+        _peer_id: PeerId,
+        _connection_id: ConnectionId,
+        _event: THandlerOutEvent<Self>,
+    ) {
+        // Handler events are Void, so this is unreachable
+    }
+
+    fn poll(
+        &mut self,
+        _cx: &mut Context<'_>,
+    ) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
+        // Emit any blocked events
+        if !self.events.is_empty() {
+            let event = self.events.pop_front().expect("events is not empty");
+            return Poll::Ready(ToSwarm::GenerateEvent(event));
+        }
+
+        Poll::Pending
+    }
+}
+
+/// Error indicating a peer is not allowed to connect.
+#[derive(Debug, Clone)]
+pub struct PeerNotAllowed(pub PeerId);
+
+impl std::fmt::Display for PeerNotAllowed {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "peer {} is not in the allowed list", self.0)
+    }
+}
+
+impl std::error::Error for PeerNotAllowed {}