Add Network::probe_address for non-disruptive reachability + identity probing

crates/anemo/src/config.rs

@@ -1,5 +1,5 @@
 use crate::{
-    crypto::{CertVerifier, ExpectedCertVerifier},
+    crypto::{CertVerifier, ExpectedCertVerifier, PROBE_SERVER_NAME},
     PeerId, Result,
 };
 use pkcs8::EncodePrivateKey;
@@ -8,7 +8,7 @@ use rcgen::{CertificateParams, KeyPair};
 use rustls::pki_types::CertificateDer;
 use rustls::pki_types::PrivateKeyDer;
 use serde::{Deserialize, Serialize};
-use std::{sync::Arc, time::Duration};
+use std::{num::NonZeroU32, sync::Arc, time::Duration};
 
 /// Configuration for a [`Network`](crate::Network).
 #[derive(Clone, Debug, Default, Serialize, Deserialize)]
@@ -63,13 +63,13 @@ pub struct Config {
     ///
     /// This limit is applied in the following ways:
     ///  - Inbound connections from [`KnownPeers`] with [`PeerAffinity::High`] or
-    ///     [`PeerAffinity::Allowed`] bypass this limit. All other inbound
-    ///     connections are only accepted if the total number of inbound and outbound
-    ///     connections, irrespective of affinity, is less than this limit.
+    ///    [`PeerAffinity::Allowed`] bypass this limit. All other inbound
+    ///    connections are only accepted if the total number of inbound and outbound
+    ///    connections, irrespective of affinity, is less than this limit.
     ///  - Outbound connections explicitly made by the application via [`Network::connect`] or
-    ///     [`Network::connect_with_peer_id`] bypass this limit.
+    ///    [`Network::connect_with_peer_id`] bypass this limit.
     ///  - Outbound connections made in the background, due to configured [`KnownPeers`], to peers with
-    ///     [`PeerAffinity::High`] bypass this limit and are always attempted.
+    ///    [`PeerAffinity::High`] bypass this limit and are always attempted.
     ///
     /// If unspecified, there will be no limit on the number of concurrent connections.
     ///
@@ -81,6 +81,36 @@ pub struct Config {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub max_concurrent_connections: Option<usize>,
 
+    /// Whether to require QUIC source-address validation (via a stateless Retry
+    /// packet) before beginning the TLS handshake for an inbound connection whose
+    /// source address has not yet been validated.
+    ///
+    /// When enabled, the first attempt from an unvalidated address is answered with
+    /// a Retry instead of starting the handshake, forcing the peer to prove it can
+    /// receive traffic at its claimed address.
+    ///
+    /// If unspecified, this defaults to `true`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub require_inbound_address_validation: Option<bool>,
+
+    /// Maximum sustained rate, per source IP, of new inbound connections admitted to
+    /// the TLS handshake, as a token-bucket refill rate in connections per second.
+    /// Inbound attempts from a source IP exceeding this rate are dropped before the
+    /// handshake begins.
+    ///
+    /// If unspecified, this defaults to `10`. Setting it to `0` disables per-IP
+    /// inbound connection rate limiting.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub inbound_connection_rate_limit_per_ip: Option<u32>,
+
+    /// Token-bucket burst size for `inbound_connection_rate_limit_per_ip`, i.e. the
+    /// maximum number of inbound connections from a single source IP that may be
+    /// admitted in an instantaneous burst.
+    ///
+    /// If unspecified, this defaults to `100`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub inbound_connection_rate_limit_burst_per_ip: Option<u32>,
+
     /// Size of the broadcast channel use for subscribing to
     /// [`PeerEvent`](crate::types::PeerEvent)s via
     /// [`Network::subscribe`](crate::Network::subscribe).
@@ -91,12 +121,26 @@ pub struct Config {
 
     /// Set the maximum frame size in bytes.
     ///
-    /// This controls the maximum size of a request or response.
+    /// This controls the maximum size of a request or response. Acts as the default
+    /// for both directions when [`Config::max_request_frame_size`] or
+    /// [`Config::max_response_frame_size`] are not set.
     ///
     /// If unspecified, there will be no limit.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub max_frame_size: Option<usize>,
 
+    /// Set the maximum frame size in bytes for request messages.
+    ///
+    /// If unspecified, falls back to [`Config::max_frame_size`].
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_request_frame_size: Option<usize>,
+
+    /// Set the maximum frame size in bytes for response messages.
+    ///
+    /// If unspecified, falls back to [`Config::max_frame_size`].
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub max_response_frame_size: Option<usize>,
+
     /// Set a timeout, in milliseconds, for all inbound requests.
     ///
     /// When an inbound timeout is hit when processing a request a Response is sent to the
@@ -264,15 +308,47 @@ impl Config {
         self.max_concurrent_connections
     }
 
+    pub(crate) fn require_inbound_address_validation(&self) -> bool {
+        const DEFAULT: bool = true;
+
+        self.require_inbound_address_validation.unwrap_or(DEFAULT)
+    }
+
+    /// Sustained per-source-IP inbound connection admission rate in connections per
+    /// second. Returns `None` only when explicitly disabled by configuring `0`.
+    pub(crate) fn inbound_connection_rate_limit_per_ip(&self) -> Option<NonZeroU32> {
+        const DEFAULT_RATE_PER_SEC: u32 = 10;
+
+        // `NonZeroU32::new` yields `None` for a configured `0`, i.e. rate limiting off.
+        NonZeroU32::new(
+            self.inbound_connection_rate_limit_per_ip
+                .unwrap_or(DEFAULT_RATE_PER_SEC),
+        )
+    }
+
+    pub(crate) fn inbound_connection_rate_limit_burst_per_ip(&self) -> NonZeroU32 {
+        const DEFAULT_BURST: u32 = 100;
+
+        // A configured burst of `0` is nonsensical (it would admit nothing); fall back
+        // to the default in that case as well as when unset.
+        self.inbound_connection_rate_limit_burst_per_ip
+            .and_then(NonZeroU32::new)
+            .unwrap_or_else(|| NonZeroU32::new(DEFAULT_BURST).expect("DEFAULT_BURST is nonzero"))
+    }
+
     pub(crate) fn peer_event_broadcast_channel_capacity(&self) -> usize {
         const PEER_EVENT_BROADCAST_CHANNEL_CAPACITY: usize = 128;
 
         self.peer_event_broadcast_channel_capacity
             .unwrap_or(PEER_EVENT_BROADCAST_CHANNEL_CAPACITY)
     }
 
-    pub(crate) fn max_frame_size(&self) -> Option<usize> {
-        self.max_frame_size
+    pub(crate) fn request_frame_size(&self) -> Option<usize> {
+        self.max_request_frame_size.or(self.max_frame_size)
+    }
+
+    pub(crate) fn response_frame_size(&self) -> Option<usize> {
+        self.max_response_frame_size.or(self.max_frame_size)
     }
 
     pub(crate) fn inbound_request_timeout(&self) -> Option<Duration> {
@@ -429,6 +505,12 @@ impl EndpointConfigBuilder {
             transport_config.clone(),
         )?;
 
+        // Always register a dedicated probe certificate/server-name in the cert resolver so any node
+        // can be probed (the client selects it via SNI). It is intentionally not added to the
+        // client-cert verifier: probing clients still authenticate with their primary certificate.
+        let (probe_certificate, _) = Self::generate_cert(&keypair, PROBE_SERVER_NAME);
+        let probe_cert_entry = (PROBE_SERVER_NAME.to_owned(), probe_certificate);
+
         let alternate_server_name = self.alternate_server_name;
         let server_config = match alternate_server_name {
             Some(alternate_server_name) => {
@@ -441,14 +523,18 @@ impl EndpointConfigBuilder {
                     vec![
                         (primary_server_name.clone(), primary_certificate.clone()),
                         (alternate_server_name, alternate_certificate),
+                        probe_cert_entry,
                     ],
                     pkcs8_der.clone_key(),
                     cert_verifier,
                     transport_config.clone(),
                 )
             }
             _ => Self::server_config(
-                vec![(primary_server_name.clone(), primary_certificate.clone())],
+                vec![
+                    (primary_server_name.clone(), primary_certificate.clone()),
+                    probe_cert_entry,
+                ],
                 pkcs8_der.clone_key(),
                 cert_verifier,
                 transport_config.clone(),
@@ -585,23 +671,14 @@ impl EndpointConfig {
     // deprecated. Before that happens, we use the attribute to
     // keep clippy happy.
     #[allow(deprecated)]
-    pub fn client_config_with_expected_server_identity(
-        &self,
-        peer_id: PeerId,
-    ) -> quinn::ClientConfig {
-        let server_cert_verifier = ExpectedCertVerifier(
-            CertVerifier {
-                server_names: vec![self.server_name().into()],
-            },
-            peer_id,
-        );
+    fn client_config_with_verifier(&self, verifier: ExpectedCertVerifier) -> quinn::ClientConfig {
         let client_crypto = rustls::ClientConfig::builder_with_provider(Arc::new(
             rustls::crypto::ring::default_provider(),
         ))
         .with_safe_default_protocol_versions()
         .unwrap()
         .dangerous()
-        .with_custom_certificate_verifier(Arc::new(server_cert_verifier))
+        .with_custom_certificate_verifier(Arc::new(verifier))
         .with_client_auth_cert(
             vec![self.client_certificate.clone()],
             self.pkcs8_der.clone_key(),
@@ -615,6 +692,30 @@ impl EndpointConfig {
         client
     }
 
+    pub fn client_config_with_expected_server_identity(
+        &self,
+        peer_id: PeerId,
+    ) -> quinn::ClientConfig {
+        self.client_config_with_verifier(ExpectedCertVerifier(
+            CertVerifier {
+                server_names: vec![self.server_name().into()],
+            },
+            peer_id,
+        ))
+    }
+
+    /// Client config for a probe connection: verifies the server's identity matches `peer_id` and
+    /// uses the dedicated probe server-name so the server can recognize and decline the probe (see
+    /// [`PROBE_SERVER_NAME`]). The probe must also pass `PROBE_SERVER_NAME` as the SNI when dialing.
+    pub fn client_config_for_probe(&self, peer_id: PeerId) -> quinn::ClientConfig {
+        self.client_config_with_verifier(ExpectedCertVerifier(
+            CertVerifier {
+                server_names: vec![PROBE_SERVER_NAME.into()],
+            },
+            peer_id,
+        ))
+    }
+
     #[cfg(test)]
     pub(crate) fn random(server_name: &str) -> Self {
         Self::builder()

crates/anemo/src/connection.rs

@@ -58,6 +58,15 @@ impl Connection {
         self.origin
     }
 
+    /// The TLS server-name (SNI) negotiated for this connection, if any.
+    pub fn server_name(&self) -> Option<String> {
+        let handshake_data = self.inner.handshake_data()?;
+        let handshake_data = handshake_data
+            .downcast::<quinn::crypto::rustls::HandshakeData>()
+            .ok()?;
+        handshake_data.server_name
+    }
+
     /// Time the Connection was established
     #[allow(unused)]
     pub fn time_established(&self) -> std::time::Instant {

crates/anemo/src/crypto.rs

@@ -20,6 +20,13 @@ static SUPPORTED_ALGORITHMS: WebPkiSupportedAlgorithms = WebPkiSupportedAlgorith
     mapping: &[(rustls::SignatureScheme::ED25519, SUPPORTED_SIG_ALGS)],
 };
 
+/// Dedicated TLS server-name (SNI) used to mark probe connections.
+///
+/// A probe verifies that an address is reachable and that the server's cryptographic identity
+/// matches an expected [`PeerId`], at the TLS layer only. The dialing client selects this
+/// server-name before the handshake.
+pub(crate) const PROBE_SERVER_NAME: &str = "anemo-probe";
+
 #[derive(Clone, Debug)]
 pub(crate) struct CertVerifier {
     pub(crate) server_names: Vec<String>,
@@ -326,9 +333,12 @@ fn pki_error(error: webpki::Error) -> rustls::Error {
         BadDer | BadDerTime => {
             rustls::Error::InvalidCertificate(rustls::CertificateError::BadEncoding)
         }
+        #[allow(deprecated)]
         InvalidSignatureForPublicKey
         | UnsupportedSignatureAlgorithm
-        | UnsupportedSignatureAlgorithmForPublicKey => {
+        | UnsupportedSignatureAlgorithmForPublicKey
+        | UnsupportedSignatureAlgorithmContext(..)
+        | UnsupportedSignatureAlgorithmForPublicKeyContext(..) => {
             rustls::Error::InvalidCertificate(rustls::CertificateError::BadSignature)
         }
         e => {