diff --git a/crates/app/src/node/behaviour.rs b/crates/app/src/node/behaviour.rs index ae4b1f89e..ecd72895a 100644 --- a/crates/app/src/node/behaviour.rs +++ b/crates/app/src/node/behaviour.rs @@ -96,8 +96,7 @@ pub(crate) struct WireP2PParams { } /// Composes the core behaviours and builds the libp2p [`Node`]. -// TODO(#402 part B): QUIC transport (featureset-gated off at v1.7.1) and -// bandwidth metrics. +// TODO(#402 part B): bandwidth metrics. pub(crate) async fn wire_p2p( params: WireP2PParams, ) -> Result<(Node, CoreHandles), AppError> { @@ -215,10 +214,17 @@ pub(crate) async fn wire_p2p( // checker observes the same shared peer/connection state the swarm updates. let p2p_context_for_handle = p2p_context.clone(); + // Mirrors Charon's `wireP2P`. + let node_type = if feature_set.enabled(pluto_featureset::Feature::Quic) { + NodeType::QUIC + } else { + NodeType::TCP + }; + let node = Node::new( p2p_config, key, - NodeType::TCP, + node_type, false, p2p_context, |builder, _keypair, relay_client| { diff --git a/crates/p2p/src/config.rs b/crates/p2p/src/config.rs index 91deaf029..f22023de2 100644 --- a/crates/p2p/src/config.rs +++ b/crates/p2p/src/config.rs @@ -1,16 +1,11 @@ //! # Charon P2P Configuration -use std::{ - fmt, - net::{IpAddr, SocketAddr}, - str::FromStr, - time::Duration, -}; +use std::{fmt, net::SocketAddr, str::FromStr, time::Duration}; use libp2p::{Multiaddr, multiaddr, ping}; use url::Url; -use crate::utils::TransportProtocol; +use crate::utils::{TransportProtocol, with_transport}; /// Shared default relay endpoints used by commands and P2P-facing configs. pub const DEFAULT_RELAYS: [&str; 5] = [ @@ -122,10 +117,6 @@ pub enum P2PConfigError { /// Failed to parse the UDP addresses. #[error("Failed to parse the UDP addresses")] FailedToParseUdpAddresses(std::net::AddrParseError), - - /// Failed to parse the multiaddress. - #[error("Failed to parse the multiaddress")] - FailedToParseMultiaddr(#[from] multiaddr::Error), } // Note: this is only for testing purposes! @@ -141,10 +132,6 @@ impl PartialEq for P2PConfigError { P2PConfigError::FailedToParseUdpAddresses(x), P2PConfigError::FailedToParseUdpAddresses(y), ) if x == y => true, - ( - P2PConfigError::FailedToParseMultiaddr(x), - P2PConfigError::FailedToParseMultiaddr(y), - ) if x.to_string() == y.to_string() => true, _ => false, } } @@ -202,10 +189,11 @@ impl P2PConfig { /// Returns the configured listen multiaddresses for `proto`. pub fn multiaddrs(&self, proto: TransportProtocol) -> Result> { - self.parse_addrs(proto)? + Ok(self + .parse_addrs(proto)? .into_iter() .map(|addr| multi_addr_from_socket_addr(addr, proto)) - .collect() + .collect()) } } @@ -240,27 +228,13 @@ fn resolve_listen_addr(addr: impl AsRef, proto: TransportProtocol) -> Resul } /// Renders `socket_addr` as a `proto` multiaddr. -pub(crate) fn multi_addr_from_socket_addr( - socket_addr: SocketAddr, - proto: TransportProtocol, -) -> Result { - let typ = match socket_addr.ip() { - IpAddr::V4(_) => "ip4", - IpAddr::V6(_) => "ip6", - }; - - let transport = match proto { - TransportProtocol::Tcp => format!("tcp/{}", socket_addr.port()), - TransportProtocol::Quic => format!("udp/{}/quic-v1", socket_addr.port()), - }; - - Multiaddr::from_str(&format!("/{}/{}/{}", typ, socket_addr.ip(), transport)) - .map_err(P2PConfigError::FailedToParseMultiaddr) +fn multi_addr_from_socket_addr(socket_addr: SocketAddr, proto: TransportProtocol) -> Multiaddr { + with_transport(Multiaddr::from(socket_addr.ip()), socket_addr.port(), proto) } #[cfg(test)] mod tests { - use std::net::{Ipv4Addr, Ipv6Addr}; + use std::net::{IpAddr, Ipv4Addr, Ipv6Addr}; use super::*; diff --git a/crates/p2p/src/p2p.rs b/crates/p2p/src/p2p.rs index b2896ad15..9911972a4 100644 --- a/crates/p2p/src/p2p.rs +++ b/crates/p2p/src/p2p.rs @@ -88,6 +88,7 @@ //! Client nodes may include relay client to support connecting via relays. use std::{ + net::IpAddr, pin::Pin, task::{Context, Poll}, time::Duration, @@ -291,6 +292,18 @@ pub struct Node { /// Listeners registered through [`Node::listen_on`], in registration order. listener_ids: Vec, + + /// External IP to advertise on the bound ports. + external_ip: Option, + + /// External hostname to advertise on the bound ports. + external_host: Option, + + /// Whether private listen addresses are withheld from advertisement. + filter_private_addrs: bool, + + /// Addresses this node currently registers as external on the swarm. + advertised_addrs: Vec, } impl Node { @@ -394,13 +407,25 @@ impl Node { Ok(node) } - /// Listens on and advertises the configured addresses of every transport - /// this node's [`NodeType`] installs. + /// Listens on the configured addresses of every transport this node's + /// [`NodeType`] installs. /// /// Transport and address selection are driven by the same /// [`NodeType::transports`] list, so a node can never listen on a transport /// it did not install, nor install one it never listens on. + /// + /// Nothing is advertised yet: a configured port of 0 means the kernel picks + /// one, so the advertised set is derived from the addresses libp2p reports + /// as bound (see [`Node::readvertise`]). fn apply_config(&mut self, cfg: &P2PConfig, filter_private_addrs: bool) -> Result<()> { + self.external_ip = cfg + .external_ip + .as_deref() + .map(str::parse::) + .transpose()?; + self.external_host = cfg.external_host.clone(); + self.filter_private_addrs = filter_private_addrs; + let mut addrs = Vec::new(); for &proto in self.node_type.transports() { @@ -419,48 +444,51 @@ impl Node { ); } - // Listen on internal addresses only - for addr in &addrs { - self.listen_on(addr.clone())?; + for addr in addrs { + self.listen_on(addr)?; } - self.set_advertised_addrs(cfg, filter_private_addrs, &addrs) + Ok(()) } - /// Advertises the external IP / hostname from `cfg` on the ports of - /// `listen_addrs`, together with `listen_addrs` themselves. - /// - /// Replaces everything the node advertises, including addresses added - /// through [`Node::add_external_address`]. - /// - /// Callers that listen on port 0 should call this again once libp2p has - /// reported the kernel-assigned ports: the configured addresses advertise - /// port 0, which is not dialable. - pub fn set_advertised_addrs( - &mut self, - cfg: &P2PConfig, - filter_private_addrs: bool, - listen_addrs: &[Multiaddr], - ) -> Result<()> { - let external_addrs = utils::external_multiaddrs(cfg, listen_addrs)?; + /// Re-derives the advertised set from the listen addresses libp2p reports: + /// the external IP / hostname on their ports plus the addresses themselves, + /// private ones withheld when configured. Relay circuit listeners are + /// skipped: their ports are the relay's, and identify lists them anyway. + /// Only the difference to the previous set is applied, so addresses the + /// swarm confirmed by other means (AutoNAT) survive. + fn readvertise(&mut self) { + let listen_addrs: Vec = self + .swarm + .listeners() + .filter(|addr| !utils::is_relay_addr(addr)) + .cloned() + .collect(); + + let external_addrs = utils::external_multiaddrs_on( + self.external_ip, + self.external_host.as_deref(), + &listen_addrs, + ); - // Advertise filtered addresses (external + optionally filtered - // internal) - let advertised_addrs = utils::filter_advertised_addresses( + let next = utils::filter_advertised_addresses( utils::ExternalAddresses(external_addrs), - utils::InternalAddresses(listen_addrs.to_vec()), - filter_private_addrs, + utils::InternalAddresses(listen_addrs), + self.filter_private_addrs, ); - for addr in self.swarm.external_addresses().cloned().collect::>() { - self.swarm.remove_external_address(&addr); + for addr in &self.advertised_addrs { + if !next.contains(addr) { + self.swarm.remove_external_address(addr); + } } - - for addr in advertised_addrs { - self.swarm.add_external_address(addr); + for addr in &next { + if !self.advertised_addrs.contains(addr) { + self.swarm.add_external_address(addr.clone()); + } } - Ok(()) + self.advertised_addrs = next; } fn bind_local_peer_id(p2p_context: &P2PContext, local_peer_id: PeerId) -> Result<()> { @@ -517,6 +545,10 @@ impl Node { node_type, p2p_context, listener_ids: Vec::new(), + external_ip: None, + external_host: None, + filter_private_addrs: false, + advertised_addrs: Vec::new(), }) } @@ -552,6 +584,10 @@ impl Node { node_type, p2p_context, listener_ids: Vec::new(), + external_ip: None, + external_host: None, + filter_private_addrs: false, + advertised_addrs: Vec::new(), }) } @@ -669,13 +705,16 @@ impl Node { } } - // Listen address changes + // Listener changes drive what the node advertises. SwarmEvent::NewListenAddr { address, .. } => { info!(%address, "listening on new address"); + self.readvertise(); } SwarmEvent::ExpiredListenAddr { address, .. } => { info!(%address, "listen address expired"); + self.readvertise(); } + SwarmEvent::ListenerClosed { .. } => self.readvertise(), // External address discovery SwarmEvent::ExternalAddrConfirmed { address } => { diff --git a/crates/p2p/src/quic_upgrade.rs b/crates/p2p/src/quic_upgrade.rs index a7b78346a..2f446e942 100644 --- a/crates/p2p/src/quic_upgrade.rs +++ b/crates/p2p/src/quic_upgrade.rs @@ -229,11 +229,10 @@ impl QuicUpgradeBehaviour { "already has direct QUIC connection to peer" ); + // Relayed TCP connections are redundant too, as in Charon. let tcp_conn_ids: Vec<_> = conns .iter() - .filter(|c| { - utils::is_tcp_addr(&c.remote_addr) && !utils::is_relay_addr(&c.remote_addr) - }) + .filter(|c| utils::is_tcp_addr(&c.remote_addr)) .map(|c| c.connection_id) .collect(); @@ -429,12 +428,51 @@ impl NetworkBehaviour for QuicUpgradeBehaviour { #[cfg(test)] mod tests { + use libp2p::swarm::CloseConnection; + use super::*; + use crate::p2p_context::Peer; + + const RELAY_ID: &str = "16Uiu2HAkzdQ5Y9SYT91K1ue5SxXwgmajXntfScGnLYeip5hHyWmT"; + const TCP: &str = "/ip4/10.0.0.2/tcp/3610"; + const QUIC: &str = "/ip4/10.0.0.2/udp/3630/quic-v1"; fn behaviour() -> QuicUpgradeBehaviour { QuicUpgradeBehaviour::new(P2PContext::default(), PeerId::random(), true) } + fn addr(s: &str) -> Multiaddr { + s.parse().unwrap() + } + + fn relayed(transport: &str) -> Multiaddr { + addr(&format!("{transport}/p2p/{RELAY_ID}/p2p-circuit")) + } + + /// A behaviour that knows `peer`, holds the given `(connection id, remote + /// address)` connections to it, and has learned `addrs` for it via + /// identify. + fn connected( + peer: PeerId, + conns: &[(usize, Multiaddr)], + addrs: &[Multiaddr], + ) -> QuicUpgradeBehaviour { + let local = PeerId::random(); + let ctx = P2PContext::new([local, peer]); + { + let mut store = ctx.peer_store_write_lock(); + for (id, remote_addr) in conns { + store.add_peer(Peer { + id: peer, + connection_id: ConnectionId::new_unchecked(*id), + remote_addr: remote_addr.clone(), + }); + } + store.set_peer_addresses(peer, addrs.to_vec()); + } + QuicUpgradeBehaviour::new(ctx, local, true) + } + /// The reason carried by the queued `UpgradeFailed` event for `peer`. fn failure_reason(behaviour: &QuicUpgradeBehaviour, peer: &PeerId) -> Option { behaviour @@ -449,6 +487,155 @@ mod tests { }) } + /// Peers the queued `Dial` events target. + fn dialed(behaviour: &QuicUpgradeBehaviour) -> Vec { + behaviour + .pending_events + .iter() + .filter_map(|event| match event { + ToSwarm::Dial { opts } => opts.get_peer_id(), + _ => None, + }) + .collect() + } + + /// Connections the queued `CloseConnection` events close. + fn closed(behaviour: &QuicUpgradeBehaviour) -> Vec { + behaviour + .pending_events + .iter() + .filter_map(|event| match event { + ToSwarm::CloseConnection { + connection: CloseConnection::One(id), + .. + } => Some(*id), + _ => None, + }) + .collect() + } + + /// Peers the queued `Upgraded` events report. + fn upgraded(behaviour: &QuicUpgradeBehaviour) -> Vec { + behaviour + .pending_events + .iter() + .filter_map(|event| match event { + ToSwarm::GenerateEvent(QuicUpgradeEvent::Upgraded { peer }) => Some(*peer), + _ => None, + }) + .collect() + } + + #[tokio::test] + async fn tick_leaves_relayed_only_peers_to_force_direct() { + let peer = PeerId::random(); + let mut behaviour = connected(peer, &[(1, relayed(TCP))], &[addr(QUIC)]); + + behaviour.run_upgrade_logic(); + + assert!(behaviour.pending_events.is_empty()); + assert!(behaviour.pending_upgrades.is_empty()); + } + + #[tokio::test] + async fn tick_closes_redundant_tcp_once_direct_quic_exists() { + let peer = PeerId::random(); + let mut behaviour = connected( + peer, + &[(1, addr(TCP)), (2, addr(QUIC)), (3, relayed(TCP))], + &[addr(QUIC)], + ); + + behaviour.run_upgrade_logic(); + + let mut closed = closed(&behaviour); + closed.sort(); + assert_eq!( + closed, + [1, 3].map(ConnectionId::new_unchecked), + "every TCP connection is redundant, relayed ones included" + ); + assert!(dialed(&behaviour).is_empty()); + assert!(behaviour.pending_upgrades.is_empty()); + } + + #[tokio::test] + async fn upgrade_dials_quic_once_then_closes_tcp() { + let peer = PeerId::random(); + let mut behaviour = connected( + peer, + &[(1, addr(TCP))], + &[addr(TCP), addr(QUIC), relayed(QUIC)], + ); + // An elapsed backoff left over from an earlier failure. + behaviour.backoffs.insert( + peer, + QuicUpgradeBackoff { + tickers_remaining: 0, + backoff_duration: 4, + }, + ); + + behaviour.run_upgrade_logic(); + + assert_eq!(dialed(&behaviour), vec![peer]); + assert!( + closed(&behaviour).is_empty(), + "TCP stays up until QUIC is established" + ); + let Some(UpgradeState::DialingQuic { tcp_conn_ids }) = + behaviour.pending_upgrades.get(&peer) + else { + panic!("upgrade must be armed for {peer}"); + }; + assert_eq!(tcp_conn_ids, &[ConnectionId::new_unchecked(1)]); + + // An armed upgrade is not dialed again on the next tick. + behaviour.run_upgrade_logic(); + assert_eq!(dialed(&behaviour).len(), 1); + + behaviour.handle_connection_established(peer, &addr(QUIC)); + + assert_eq!(upgraded(&behaviour), vec![peer]); + assert_eq!(closed(&behaviour), vec![ConnectionId::new_unchecked(1)]); + assert!(behaviour.pending_upgrades.is_empty()); + assert!( + !behaviour.backoffs.contains_key(&peer), + "success clears the backoff" + ); + } + + #[tokio::test] + async fn failed_upgrade_keeps_tcp_and_arms_backoff() { + let armed = || { + let peer = PeerId::random(); + let mut behaviour = connected(peer, &[(1, addr(TCP))], &[addr(QUIC)]); + behaviour.run_upgrade_logic(); + (behaviour, peer) + }; + let check = |behaviour: &QuicUpgradeBehaviour, peer: &PeerId, case: &str| { + assert!(upgraded(behaviour).is_empty(), "{case}"); + assert!( + closed(behaviour).is_empty(), + "{case}: the TCP connection is kept" + ); + assert!(failure_reason(behaviour, peer).is_some(), "{case}"); + assert!(behaviour.pending_upgrades.is_empty(), "{case}"); + assert!( + behaviour.backoffs.contains_key(peer), + "{case}: failure arms the backoff" + ); + }; + + let (mut behaviour, peer) = armed(); + behaviour.handle_connection_established(peer, &addr(TCP)); + check(&behaviour, &peer, "non-QUIC connection"); + + let (mut behaviour, peer) = armed(); + behaviour.handle_dial_failure(Some(peer)); + check(&behaviour, &peer, "dial failure"); + } + #[test] fn backoff_doubles_then_pins_at_the_cap() { let mut backoff = QuicUpgradeBackoff::new(); diff --git a/crates/p2p/src/utils.rs b/crates/p2p/src/utils.rs index fdf646cad..c33db1856 100644 --- a/crates/p2p/src/utils.rs +++ b/crates/p2p/src/utils.rs @@ -8,11 +8,7 @@ //! //! These utilities are primarily used internally by the [`crate::p2p`] module. -use std::{ - collections::HashSet, - net::{IpAddr, SocketAddr}, - time::Duration, -}; +use std::{collections::HashSet, net::IpAddr, time::Duration}; use libp2p::{ Multiaddr, @@ -22,10 +18,7 @@ use libp2p::{ use crate::metrics::{ConnectionType, Protocol}; -use crate::{ - config::{self, P2PConfig}, - manet::Manet, -}; +use crate::{config::P2PConfig, manet::Manet}; /// A transport a node can listen on and advertise. /// @@ -39,49 +32,32 @@ pub enum TransportProtocol { Quic, } -/// Returns the external IP and Hostname fields as `proto` multiaddrs on -/// `ports`. -/// -/// `ports` must be the ports the node actually listens on: a configured port of -/// 0 means the kernel picks one, so the configured value would advertise -/// nothing dialable. -fn external_proto_multiaddrs( +/// Returns the external IP and hostname from `cfg` as multiaddrs on the ports +/// of `listen_addrs`, TCP forms first. +pub fn external_multiaddrs( cfg: &P2PConfig, - ports: &[u16], - proto: TransportProtocol, + listen_addrs: &[Multiaddr], ) -> crate::p2p::Result> { - let mut resp = vec![]; - - if let Some(external_ip) = cfg.external_ip.as_ref() { - let ip = external_ip.parse::()?; - - for port in ports { - let maddr = config::multi_addr_from_socket_addr(SocketAddr::new(ip, *port), proto)?; - - resp.push(maddr); - } - } - - if let Some(external_host) = cfg.external_host.as_ref() { - for port in ports { - resp.push(match proto { - TransportProtocol::Tcp => multiaddr::multiaddr!(Dns(external_host), Tcp(*port)), - TransportProtocol::Quic => { - multiaddr::multiaddr!(Dns(external_host), Udp(*port), QuicV1) - } - }); - } - } - - Ok(resp) + let external_ip = cfg + .external_ip + .as_deref() + .map(str::parse::) + .transpose()?; + + Ok(external_multiaddrs_on( + external_ip, + cfg.external_host.as_deref(), + listen_addrs, + )) } -/// Returns the external IP and Hostname fields as multiaddrs on the ports of +/// Returns `external_ip` and `external_host` as multiaddrs on the ports of /// `listen_addrs`, TCP forms first. -pub fn external_multiaddrs( - cfg: &P2PConfig, +pub(crate) fn external_multiaddrs_on( + external_ip: Option, + external_host: Option<&str>, listen_addrs: &[Multiaddr], -) -> crate::p2p::Result> { +) -> Vec { let mut addrs = Vec::new(); for proto in [TransportProtocol::Tcp, TransportProtocol::Quic] { @@ -90,10 +66,32 @@ pub fn external_multiaddrs( .filter_map(|addr| addr_port(addr, proto)) .collect(); - addrs.extend(external_proto_multiaddrs(cfg, &ports, proto)?); + if let Some(ip) = external_ip { + addrs.extend( + ports + .iter() + .map(|&port| with_transport(Multiaddr::from(ip), port, proto)), + ); + } + + if let Some(host) = external_host { + addrs.extend( + ports + .iter() + .map(|&port| with_transport(multiaddr::multiaddr!(Dns(host)), port, proto)), + ); + } } - Ok(addrs) + addrs +} + +/// Appends the `proto` transport on `port` to `base`. +pub(crate) fn with_transport(base: Multiaddr, port: u16, proto: TransportProtocol) -> Multiaddr { + match proto { + TransportProtocol::Tcp => base.with(MaProtocol::Tcp(port)), + TransportProtocol::Quic => base.with(MaProtocol::Udp(port)).with(MaProtocol::QuicV1), + } } /// Returns the port `addr` carries for `proto`, if any. @@ -194,11 +192,6 @@ pub fn is_tcp_addr(addr: &Multiaddr) -> bool { addr.iter().any(|p| matches!(p, MaProtocol::Tcp(_))) } -/// Returns true if the node has QUIC enabled (listening on QUIC addresses). -pub fn is_quic_enabled<'a>(listen_addrs: impl Iterator) -> bool { - listen_addrs.into_iter().any(is_quic_addr) -} - /// Returns true if there is a direct (non-relay) QUIC connection among the /// peers. pub fn has_direct_quic_conn(peers: &[&crate::p2p_context::Peer]) -> bool { @@ -334,16 +327,6 @@ mod tests { assert!(filter_direct_quic_addrs(std::iter::empty()).is_empty()); } - #[test] - fn quic_is_enabled_only_while_listening_on_quic() { - let tcp = addr("/ip4/1.2.3.4/tcp/3610"); - let quic = addr("/ip4/1.2.3.4/udp/3610/quic-v1"); - - assert!(is_quic_enabled([&tcp, &quic].into_iter())); - assert!(!is_quic_enabled([&tcp].into_iter())); - assert!(!is_quic_enabled(std::iter::empty())); - } - #[test] fn direct_conn_checks_ignore_relayed_connections() { let quic = conn(addr("/ip4/1.2.3.4/udp/3610/quic-v1")); diff --git a/crates/p2p/tests/advertised_addrs.rs b/crates/p2p/tests/advertised_addrs.rs new file mode 100644 index 000000000..1c8043bf4 --- /dev/null +++ b/crates/p2p/tests/advertised_addrs.rs @@ -0,0 +1,142 @@ +//! What a node advertises, seen through identify as a peer receives it. +//! +//! A node configured to listen on port 0 must advertise the kernel-assigned +//! port, never the configured 0, with its external IP on that same port; and a +//! relay circuit listener must not leak the relay's port into that set. + +mod common; + +use common::{TEST_TIMEOUT, spawn_relay_server}; +use futures::StreamExt as _; +use libp2p::{Multiaddr, identify, multiaddr::Protocol, relay, swarm::SwarmEvent}; +use pluto_p2p::{ + behaviours::pluto::PlutoBehaviourEvent, + config::P2PConfig, + p2p::{Node, NodeType}, + p2p_context::P2PContext, + peer::peer_id_from_key, + utils::is_relay_addr, +}; +use pluto_testutil::random::generate_insecure_k1_key; +use tokio::time::timeout; + +type ClientNode = Node; + +const EXTERNAL_IP: &str = "1.2.3.4"; + +fn tcp_port(addr: &Multiaddr) -> Option { + addr.iter().find_map(|p| match p { + Protocol::Tcp(port) => Some(port), + _ => None, + }) +} + +/// Drives `node` until it has reported both a direct and a relayed listen +/// address, and returns the direct one. +async fn direct_listen_addr(node: &mut ClientNode) -> Multiaddr { + timeout(TEST_TIMEOUT, async { + let mut direct = None; + let mut relayed = false; + loop { + if let SwarmEvent::NewListenAddr { address, .. } = node.select_next_some().await { + if is_relay_addr(&address) { + relayed = true; + } else { + direct = Some(address); + } + if relayed && let Some(addr) = &direct { + return addr.clone(); + } + } + } + }) + .await + .expect("timed out waiting for the listen addresses") +} + +#[tokio::test] +async fn advertises_own_bound_ports_only() { + let (relay_peer, relay_addr, relay_handle) = + spawn_relay_server(generate_insecure_k1_key(20)).await; + + let key_a = generate_insecure_k1_key(21); + let key_b = generate_insecure_k1_key(22); + let peer_a = peer_id_from_key(key_a.public_key()).expect("peer id A"); + let peer_b = peer_id_from_key(key_b.public_key()).expect("peer id B"); + + // A listens on a kernel-assigned TCP port, reserves a relay circuit, and + // has an external IP override. + let mut node_a: ClientNode = Node::new( + P2PConfig::builder() + .with_tcp_addrs(vec!["127.0.0.1:0".to_owned()]) + .with_external_ip(EXTERNAL_IP.to_owned()) + .build(), + key_a, + NodeType::TCP, + false, + P2PContext::new(vec![peer_b, relay_peer]), + |builder, _keypair, relay_client| builder.with_inner(relay_client), + ) + .expect("build node A"); + node_a + .listen_on( + relay_addr + .with(Protocol::P2p(relay_peer)) + .with(Protocol::P2pCircuit), + ) + .expect("A listen_on circuit"); + + let mut node_b: ClientNode = Node::new( + P2PConfig::default(), + key_b, + NodeType::TCP, + false, + P2PContext::new(vec![peer_a]), + |builder, _keypair, relay_client| builder.with_inner(relay_client), + ) + .expect("build node B"); + + let bound = direct_listen_addr(&mut node_a).await; + let bound_port = tcp_port(&bound).expect("bound TCP port"); + assert!(bound_port != 0, "kernel must have assigned a port"); + + node_b.dial(bound).expect("dial A"); + + // Drive both until B has A's identify payload. + let advertised = timeout(TEST_TIMEOUT, async { + loop { + tokio::select! { + _ = node_a.select_next_some() => {} + event = node_b.select_next_some() => { + if let SwarmEvent::Behaviour(PlutoBehaviourEvent::Identify( + identify::Event::Received { peer_id, info, .. }, + )) = event + && peer_id == peer_a + { + return info.listen_addrs; + } + } + } + } + }) + .await + .expect("timed out waiting for A's identify"); + + relay_handle.abort(); + + let external: Multiaddr = format!("/ip4/{EXTERNAL_IP}/tcp/{bound_port}") + .parse() + .expect("external multiaddr"); + assert!( + advertised.contains(&external), + "external address {external} missing from {advertised:?}", + ); + + assert!( + advertised + .iter() + .filter(|addr| !is_relay_addr(addr)) + .all(|addr| tcp_port(addr) == Some(bound_port)), + "advertised addresses on a port other than {bound_port}: {advertised:?}", + ); +} diff --git a/crates/p2p/tests/common/mod.rs b/crates/p2p/tests/common/mod.rs index d24ae946d..eb7f787d9 100644 --- a/crates/p2p/tests/common/mod.rs +++ b/crates/p2p/tests/common/mod.rs @@ -64,10 +64,6 @@ pub async fn spawn_relay_server(key: SecretKey) -> (PeerId, Multiaddr, JoinHandl .await .expect("timed out waiting for the relay listen address"); - // Without a reachable advertised address, reservations are rejected - // client-side with `NoAddressesInReservation`. - node.add_external_address(addr.clone()); - let handle = tokio::spawn(async move { loop { node.select_next_some().await; diff --git a/crates/relay-server/src/p2p.rs b/crates/relay-server/src/p2p.rs index 8ab91601f..54f818c2b 100644 --- a/crates/relay-server/src/p2p.rs +++ b/crates/relay-server/src/p2p.rs @@ -249,19 +249,12 @@ pub async fn bind_relay(config: &Config, key: SecretKey) -> Result { )?; // First poll of the swarm, and so the first point at which this relay - // services anything. Every other listener is already bound. + // services anything. Every other listener is already bound, and the node + // advertises each address as libp2p reports it bound. let listen_addrs = Arc::new(RwLock::new(Vec::new())); wait_for_listen_addrs(&mut node, &listen_addrs).await?; let bound_addrs = listen_addrs.read().await.clone(); - // Advertise the ports libp2p bound rather than the configured ones, which - // carry port 0 whenever the kernel picked the port. - node.set_advertised_addrs( - &config.p2p_config, - config.filter_private_addrs, - &bound_addrs, - )?; - // Compute external multiaddrs from external_ip / external_host config so // they're advertised on `/` and folded into ENR responses on `/enr` even // when libp2p only sees private listen addresses (e.g., K8s pods behind