Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 38 additions & 13 deletions crates/anemo/src/config.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
use crate::{
crypto::{CertVerifier, ExpectedCertVerifier},
crypto::{CertVerifier, ExpectedCertVerifier, PROBE_SERVER_NAME},
PeerId, Result,
};
use pkcs8::EncodePrivateKey;
Expand Down Expand Up @@ -505,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) => {
Expand All @@ -517,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(),
Expand Down Expand Up @@ -661,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(),
Expand All @@ -691,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()
Expand Down
9 changes: 9 additions & 0 deletions crates/anemo/src/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
7 changes: 7 additions & 0 deletions crates/anemo/src/crypto.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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>,
Expand Down
16 changes: 16 additions & 0 deletions crates/anemo/src/endpoint.rs
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,22 @@ impl Endpoint {
.map(Connecting::new_outbound)
}

/// Open a probe connection that verifies `address` is reachable and the server's identity
/// matches `peer_id`, without admitting the result as a peer.
///
/// Identity is enforced by the TLS verifier, so a successful connect means
/// the identity matched.
pub fn connect_for_probe(
&self,
address: SocketAddr,
peer_id: PeerId,
) -> Result<quinn::Connecting> {
let config = self.config.client_config_for_probe(peer_id);
self.inner
.connect_with(config, address, crate::crypto::PROBE_SERVER_NAME)
.map_err(Into::into)
}

/// Returns the socket address that this Endpoint is bound to.
pub fn local_addr(&self) -> SocketAddr {
*self.local_addr.read().unwrap()
Expand Down
2 changes: 1 addition & 1 deletion crates/anemo/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ pub mod types;

pub use config::{Config, QuicConfig};
pub use error::{Error, Result};
pub use network::{Builder, KnownPeers, Network, NetworkRef, Peer};
pub use network::{Builder, KnownPeers, Network, NetworkRef, Peer, ProbeOutcome};
pub use routing::{Router, ServicesOpen, ServicesSealed};
#[doc(inline)]
pub use types::{request::Request, response::Response, ConnectionOrigin, Direction, PeerId};
Expand Down
11 changes: 11 additions & 0 deletions crates/anemo/src/network/connection_manager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,17 @@ impl ConnectionManager {
let fut = async {
let connection = connecting.await?;

// A probe connection only needs to verify reachability and our identity at the TLS
// layer, which has now completed. Decline to admit it as a peer.
if connection.server_name().as_deref() == Some(crate::crypto::PROBE_SERVER_NAME) {
trace!(
peer_id = %connection.peer_id(),
"closing completed probe connection"
);
connection.close();
return Err(anyhow::anyhow!("completed probe connection"));
}

// TODO close the connection explicitly with a reason once we have machine
// readable errors. See https://github.com/MystenLabs/anemo/issues/13 for more info.
match known_peers.get(&connection.peer_id()) {
Expand Down
76 changes: 75 additions & 1 deletion crates/anemo/src/network/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ use tower::{
util::{BoxLayer, BoxService},
Layer, Service, ServiceBuilder, ServiceExt,
};
use tracing::warn;
use tracing::{trace, warn};

mod connection_manager;
pub use connection_manager::KnownPeers;
Expand Down Expand Up @@ -269,6 +269,35 @@ impl Builder {
}
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ProbeOutcome {
/// The address was reachable and the server's identity matched the expected [`PeerId`].
Reachable,
/// A connection could not be established (e.g. no route, nothing listening).
Unreachable,
/// A connection was reached at the transport level but TLS verification failed — for a probe
/// with an explicit expected peer id, this means the server's identity did not match.
WrongIdentity,
/// The address could not be resolved or used to initiate a connection.
BadAddress,
/// The probe did not complete within the configured connect timeout.
Timeout,
}

impl ProbeOutcome {
/// Classify a failed probe connection. A transport/crypto-level failure means we reached an
/// endpoint speaking QUIC+TLS but verification failed; since a probe presents a valid client
/// cert and an explicit expected peer id, that indicates the server's identity did not match. A
/// timeout or any other connection error indicates the endpoint could not be reached.
fn from_connection_error(error: quinn::ConnectionError) -> Self {
match error {
quinn::ConnectionError::TimedOut => ProbeOutcome::Timeout,
quinn::ConnectionError::TransportError(_) => ProbeOutcome::WrongIdentity,
_ => ProbeOutcome::Unreachable,
}
}
}

/// Handle to a network.
///
/// This handle can be cheaply cloned and shared across many threads.
Expand Down Expand Up @@ -322,6 +351,17 @@ impl Network {
self.0.connect(addr.into(), Some(peer_id)).await
}

/// Probe `addr` to verify it is reachable and that the server's cryptographic identity matches
/// `expected_peer_id`, without joining the peer set or disturbing any existing connection to
/// that peer.
pub async fn probe_address<A: Into<Address>>(
&self,
addr: A,
expected_peer_id: PeerId,
) -> ProbeOutcome {
self.0.probe_address(addr.into(), expected_peer_id).await
}

pub fn disconnect(&self, peer: PeerId) -> Result<()> {
self.0.disconnect(peer)
}
Expand Down Expand Up @@ -411,6 +451,40 @@ impl NetworkInner {
receiver.await?
}

async fn probe_address(&self, addr: Address, expected_peer_id: PeerId) -> ProbeOutcome {
let socket_addr = match addr.resolve().await {
Ok(socket_addr) => socket_addr,
Err(e) => {
trace!(?addr, "probe: failed to resolve address: {e}");
return ProbeOutcome::BadAddress;
}
};

let connecting = match self
.endpoint
.connect_for_probe(socket_addr, expected_peer_id)
{
Ok(connecting) => connecting,
Err(e) => {
trace!(%socket_addr, "probe: failed to initiate connection: {e}");
return ProbeOutcome::BadAddress;
}
};

match tokio::time::timeout(self.config.connect_timeout(), connecting).await {
// TLS completed, including verifying the server's identity against `expected_peer_id`.
Ok(Ok(connection)) => {
connection.close(0u32.into(), b"probe complete");
ProbeOutcome::Reachable
}
Ok(Err(e)) => {
trace!(%socket_addr, "probe: connection failed: {e}");
ProbeOutcome::from_connection_error(e)
}
Err(_) => ProbeOutcome::Timeout,
}
}

fn disconnect(&self, peer_id: PeerId) -> Result<()> {
let active_peers = self
.active_peers
Expand Down
107 changes: 106 additions & 1 deletion crates/anemo/src/network/tests.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
use crate::{types::PeerEvent, Config, Network, NetworkRef, Request, Response, Result};
use crate::{
types::PeerEvent, Config, Network, NetworkRef, ProbeOutcome, Request, Response, Result,
};
use bytes::{Buf, BufMut, Bytes, BytesMut};
use futures::FutureExt;
use std::{convert::Infallible, time::Duration};
Expand Down Expand Up @@ -120,6 +122,109 @@ async fn connect_with_invalid_peer_id_ensure_server_doesnt_succeed() -> Result<(
Ok(())
}

#[tokio::test]
async fn probe_address_reachable_and_identity_match() -> Result<()> {
let _guard = crate::init_tracing_for_testing();

let prober = build_network()?;
let target = build_network()?;

let outcome = prober
.probe_address(target.local_addr(), target.peer_id())
.await;
assert_eq!(outcome, ProbeOutcome::Reachable);

// A probe must never join the peer set on either side.
assert!(prober.peers().is_empty());
assert!(target.peers().is_empty());

Ok(())
}

#[tokio::test]
async fn probe_address_identity_mismatch() -> Result<()> {
let _guard = crate::init_tracing_for_testing();

let prober = build_network()?;
let target = build_network()?;
let other = build_network()?;

// Probe the target's address but expect a different peer's identity.
let outcome = prober
.probe_address(target.local_addr(), other.peer_id())
.await;
assert_eq!(outcome, ProbeOutcome::WrongIdentity);

assert!(target.peers().is_empty());

Ok(())
}

#[tokio::test]
async fn probe_address_unreachable() -> Result<()> {
let _guard = crate::init_tracing_for_testing();

// Use a short connect timeout so the probe to a dead address returns promptly.
let prober = build_network_with_config(Config {
connect_timeout_ms: Some(300),
..Default::default()
})?;

// A bound UDP socket that never speaks QUIC: packets are delivered and dropped, so the
// handshake never completes and the probe times out.
let dead_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
let dead_addr = dead_socket.local_addr().unwrap();

let outcome = prober.probe_address(dead_addr, prober.peer_id()).await;
assert_eq!(outcome, ProbeOutcome::Timeout);

Ok(())
}

#[tokio::test]
async fn probe_does_not_disrupt_existing_connection() -> Result<()> {
let _guard = crate::init_tracing_for_testing();

let prober = build_network()?;
let target = build_network()?;

// Establish a production connection from the prober to the target.
let (mut target_events, _) = target.subscribe().unwrap();
let peer = prober.connect(target.local_addr()).await?;
assert_eq!(peer, target.peer_id());
assert_eq!(
target_events.recv().await,
Ok(PeerEvent::NewPeer(prober.peer_id()))
);

// Probe the target from the same node that already holds a connection to it. This is the case
// the probe SNI marker must make safe: were the probe admitted as a peer, `add_peer` would
// tie-break against the existing connection and could drop it.
let outcome = prober
.probe_address(target.local_addr(), target.peer_id())
.await;
assert_eq!(outcome, ProbeOutcome::Reachable);

// Give the target a chance to (incorrectly) process the probe as a peer.
tokio::task::yield_now().await;

// The production connection is intact: no LostPeer event, peer still present, RPC still works.
assert!(matches!(
target_events.try_recv(),
Err(tokio::sync::broadcast::error::TryRecvError::Empty)
));
assert!(target.peers().contains(&prober.peer_id()));
assert!(prober.peers().contains(&target.peer_id()));

let msg = b"still here";
let response = prober
.rpc(target.peer_id(), Request::new(msg.as_ref().into()))
.await?;
assert_eq!(response.into_body(), msg.as_ref());

Ok(())
}

#[tokio::test]
async fn connect_with_hostname() -> Result<()> {
let _guard = crate::init_tracing_for_testing();
Expand Down
Loading