diff options
| author | tv <tv@krebsco.de> | 2026-10-31 04:20:41 +0100 |
|---|---|---|
| committer | tv <tv@krebsco.de> | 2026-10-31 04:20:41 +0100 |
| commit | 235411a261441df6c0ddb1c4d7f99cf06b8f3c55 (patch) | |
| tree | 8ccf4f4dfbdc985fd2196fca87ad9a2dd0d16ed3 /src/transport/libp2p | |
| parent | 803c2a528b25c5bf983130a9826bdcbfdc394ba6 (diff) | |
libp2p: explode!
Diffstat (limited to 'src/transport/libp2p')
| -rw-r--r-- | src/transport/libp2p/behaviour.rs | 364 | ||||
| -rw-r--r-- | src/transport/libp2p/book.rs | 524 | ||||
| -rw-r--r-- | src/transport/libp2p/config.rs | 117 | ||||
| -rw-r--r-- | src/transport/libp2p/driver.rs | 527 | ||||
| -rw-r--r-- | src/transport/libp2p/errors.rs | 128 | ||||
| -rw-r--r-- | src/transport/libp2p/ignore.rs | 146 | ||||
| -rw-r--r-- | src/transport/libp2p/mod.rs | 657 | ||||
| -rw-r--r-- | src/transport/libp2p/relays.rs | 234 | ||||
| -rw-r--r-- | src/transport/libp2p/tests.rs | 779 |
9 files changed, 3476 insertions, 0 deletions
diff --git a/src/transport/libp2p/behaviour.rs b/src/transport/libp2p/behaviour.rs new file mode 100644 index 0000000..4ced88a --- /dev/null +++ b/src/transport/libp2p/behaviour.rs @@ -0,0 +1,364 @@ +//! The libp2p behaviours this transport combines (`Behaviour`), hole punching with LAN +//! candidates (`HolePunching`), and building the swarm (TCP and QUIC, or in memory). + +use super::*; + +/// The protocol version identify says (informational: nothing checks it). +pub(super) const IDENTIFY_VERSION: &str = "/continuum/1"; + +#[derive(::libp2p::swarm::NetworkBehaviour)] +pub(super) struct Behaviour { + /// First: connections it denies need nothing from the others. + pub(super) ignoring: Ignoring, + pub(super) stream: libp2p_stream::Behaviour, + pub(super) identify: ::libp2p::identify::Behaviour, + pub(super) relay_client: ::libp2p::relay::client::Behaviour, + /// Only with `Libp2pConfig::relay`. + pub(super) relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>, + pub(super) dcutr: HolePunching, + /// Each connection's round trip (`Book::rtts`): of several to one peer, the fastest + /// is kept. + pub(super) ping: ::libp2p::ping::Behaviour, + pub(super) autonat: ::libp2p::autonat::Behaviour, + /// Only with `Libp2pConfig::mdns`. + pub(super) mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>, + /// Only with `Libp2pConfig::upnp`: maps our listen ports on the router, and confirms + /// the mapped addresses as ours (identify then tells them). + pub(super) upnp: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::upnp::tokio::Behaviour>, + /// Only with `Libp2pConfig::public`: the IPFS DHT. + pub(super) kad: ::libp2p::swarm::behaviour::toggle::Toggle< + ::libp2p::kad::Behaviour<::libp2p::kad::store::MemoryStore>, + >, +} + +/// libp2p's hole punching (DCUtR), offering our listen addresses too. DCUtR offers only +/// where others saw us (external address candidates, from identify): seen by a relay +/// out on the internet, two nodes on one LAN are both at their router's public address, +/// and punching through it needs the router to hairpin, which many don't. Their LAN +/// addresses work, the firewalls on the way being stateful. Only DCUtR is told: +/// AutoNAT would test every candidate, and fail on all of these. +pub(super) struct HolePunching { + pub(super) inner: ::libp2p::dcutr::Behaviour, + /// Where we listen (`offered`). + pub(super) listening: Vec<Multiaddr>, + /// `Libp2pConfig::ignore`: not offered. + pub(super) ignored: Vec<IpRange>, +} + +/// Whether DCUtR offers our listen address `address` too: neither loopback nor +/// link-local (as for publishing), nor through a relay. +pub(super) fn offered(address: &Multiaddr) -> bool { + !is_local(address) && !is_relayed(address) +} + +impl HolePunching { + pub(super) fn new(local: PeerId, ignored: Vec<IpRange>) -> Self { + Self { + inner: ::libp2p::dcutr::Behaviour::new(local), + listening: Vec::new(), + ignored, + } + } + + /// Tells DCUtR where we listen, as if others had seen us there. Again after every + /// other candidate, as DCUtR keeps the latest 20. + pub(super) fn offer_listening(&mut self) { + use ::libp2p::swarm::{FromSwarm, NetworkBehaviour, NewExternalAddrCandidate}; + for addr in &self.listening { + self.inner + .on_swarm_event(FromSwarm::NewExternalAddrCandidate( + NewExternalAddrCandidate { addr }, + )); + } + } +} + +impl ::libp2p::swarm::NetworkBehaviour for HolePunching { + type ConnectionHandler = + <::libp2p::dcutr::Behaviour as ::libp2p::swarm::NetworkBehaviour>::ConnectionHandler; + type ToSwarm = ::libp2p::dcutr::Event; + + fn handle_pending_inbound_connection( + &mut self, + connection_id: ConnectionId, + local_addr: &Multiaddr, + remote_addr: &Multiaddr, + ) -> Result<(), ::libp2p::swarm::ConnectionDenied> { + self.inner + .handle_pending_inbound_connection(connection_id, local_addr, remote_addr) + } + + fn handle_established_inbound_connection( + &mut self, + connection_id: ConnectionId, + peer: PeerId, + local_addr: &Multiaddr, + remote_addr: &Multiaddr, + ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> { + if is_relayed(local_addr) { + // DCUtR offers its candidates on this connection: where others saw us, and + // these. + debug!(peer = %Addressed(SCHEME, &peer), listening = ?self.listening, "Reached through a relay; hole punching offers where we listen too"); + } + self.inner.handle_established_inbound_connection( + connection_id, + peer, + local_addr, + remote_addr, + ) + } + + fn handle_pending_outbound_connection( + &mut self, + connection_id: ConnectionId, + maybe_peer: Option<PeerId>, + addresses: &[Multiaddr], + effective_role: ::libp2p::core::Endpoint, + ) -> Result<Vec<Multiaddr>, ::libp2p::swarm::ConnectionDenied> { + self.inner.handle_pending_outbound_connection( + connection_id, + maybe_peer, + addresses, + effective_role, + ) + } + + 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<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> { + self.inner.handle_established_outbound_connection( + connection_id, + peer, + addr, + role_override, + port_use, + ) + } + + fn on_swarm_event(&mut self, event: ::libp2p::swarm::FromSwarm) { + use ::libp2p::swarm::FromSwarm; + self.inner.on_swarm_event(event); + match event { + FromSwarm::NewListenAddr(new) + if offered(new.addr) && !ignored(&self.ignored, new.addr) => + { + if !self.listening.contains(new.addr) { + debug!(address = %new.addr, "Offering where we listen for hole punching"); + self.listening.push(new.addr.clone()); + } + self.offer_listening(); + } + FromSwarm::ExpiredListenAddr(expired) => { + self.listening.retain(|address| address != expired.addr); + } + FromSwarm::NewExternalAddrCandidate(_) => self.offer_listening(), + _ => {} + } + } + + fn on_connection_handler_event( + &mut self, + peer_id: PeerId, + connection_id: ConnectionId, + event: ::libp2p::swarm::THandlerOutEvent<Self>, + ) { + self.inner + .on_connection_handler_event(peer_id, connection_id, event) + } + + fn poll( + &mut self, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll< + ::libp2p::swarm::ToSwarm<Self::ToSwarm, ::libp2p::swarm::THandlerInEvent<Self>>, + > { + self.inner.poll(cx) + } +} + +/// What the behaviour has besides what every node has. +#[derive(Clone, Copy)] +pub(super) struct Features { + pub(super) relay: bool, + pub(super) mdns: bool, + pub(super) upnp: bool, + pub(super) public: bool, +} + +impl Features { + pub(super) fn of(config: &Libp2pConfig) -> Self { + Self { + relay: config.relay, + mdns: config.mdns, + upnp: config.upnp, + public: config.public, + } + } +} + +/// The IPFS DHT's protocol, and its bootstrap nodes (all at `IPFS_BOOTSTRAP_ADDRESS`). +pub(super) const IPFS_DHT: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0"); + +pub(super) const IPFS_BOOTSTRAP_ADDRESS: &str = "/dnsaddr/bootstrap.libp2p.io"; + +pub(super) const IPFS_BOOTSTRAP_PEERS: [&str; 4] = [ + "QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN", + "QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa", + "QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb", + "QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt", +]; + +/// How often a node on the DHT refreshes its routing table. +pub(super) const DHT_BOOTSTRAP: Duration = Duration::from_secs(300); + +impl Behaviour { + pub(super) fn new( + key: &Keypair, + relay_client: ::libp2p::relay::client::Behaviour, + features: Features, + ignored: Vec<IpRange>, + ) -> std::io::Result<Self> { + let Features { + relay, + mdns, + upnp, + public, + } = features; + let peer_id = key.public().to_peer_id(); + let kad = public.then(|| { + let mut dht = ::libp2p::kad::Behaviour::with_config( + peer_id, + ::libp2p::kad::store::MemoryStore::new(peer_id), + ::libp2p::kad::Config::new(IPFS_DHT), + ); + let bootstrap: Multiaddr = IPFS_BOOTSTRAP_ADDRESS.parse().expect("a multiaddr"); + for peer in IPFS_BOOTSTRAP_PEERS { + let peer: PeerId = peer.parse().expect("a peer ID"); + let _ = dht.add_address(&peer, bootstrap.clone()); + } + dht + }); + let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public()) + .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION"))); + let relay = relay.then(|| { + ::libp2p::relay::Behaviour::new( + peer_id, + ::libp2p::relay::Config { + max_circuits: RELAY_CIRCUITS, + max_circuits_per_peer: RELAY_CIRCUITS_PER_PEER, + max_circuit_duration: RELAY_CIRCUIT_DURATION, + max_circuit_bytes: RELAY_CIRCUIT_BYTES, + reservation_rate_limiters: Vec::new(), + circuit_src_rate_limiters: Vec::new(), + ..Default::default() + }, + ) + }); + let mdns = match mdns { + true => Some(::libp2p::mdns::tokio::Behaviour::new( + ::libp2p::mdns::Config::default(), + peer_id, + )?), + false => None, + }; + Ok(Self { + ignoring: Ignoring { + ranges: ignored.clone(), + }, + stream: libp2p_stream::Behaviour::new(), + identify: ::libp2p::identify::Behaviour::new(identify), + relay_client, + relay: relay.into(), + dcutr: HolePunching::new(peer_id, ignored), + ping: ::libp2p::ping::Behaviour::new(::libp2p::ping::Config::new()), + autonat: ::libp2p::autonat::Behaviour::new( + peer_id, + ::libp2p::autonat::Config { + // Swarms on private networks (a LAN, WireGuard) count too. + only_global_ips: false, + ..Default::default() + }, + ), + mdns: mdns.into(), + upnp: upnp.then(::libp2p::upnp::tokio::Behaviour::default).into(), + kad: kad.into(), + }) + } +} + +pub(super) fn network_swarm( + key: Keypair, + features: Features, + ignored: Vec<IpRange>, +) -> Result<Swarm<Behaviour>> { + Ok(::libp2p::SwarmBuilder::with_existing_identity(key) + .with_tokio() + .with_tcp( + ::libp2p::tcp::Config::default(), + ::libp2p::noise::Config::new, + ::libp2p::yamux::Config::default, + )? + .with_quic() + .with_dns()? + .with_relay_client( + ::libp2p::noise::Config::new, + ::libp2p::yamux::Config::default, + )? + .with_behaviour(|key, relay_client| { + Behaviour::new(key, relay_client, features, ignored.clone()) + .map_err(|error| Box::new(error) as Box<dyn std::error::Error + Send + Sync>) + })? + .with_swarm_config(|config| swarm_config(config, features)) + .build()) +} + +/// The swarm's settings. Streams are opened without waiting for the other side to agree +/// on the protocol (`V1Lazy`): the protocol and the first data go together, saving a +/// round trip per message (one stream each, module docs). That is safe only towards +/// peers sure to speak the protocol, so not with public libp2p infrastructure +/// (`Libp2pConfig::public`), whose nodes speak only some: there a protocol they don't +/// would fail late, and unclearly. +pub(super) fn swarm_config( + config: ::libp2p::swarm::Config, + features: Features, +) -> ::libp2p::swarm::Config { + let config = config + .with_idle_connection_timeout(IDLE_TIMEOUT) + .with_dial_concurrency_factor(DIAL_CONCURRENCY); + if features.public { + config + } else { + config.with_substream_upgrade_protocol_override(::libp2p::core::upgrade::Version::V1Lazy) + } +} + +pub(super) fn memory_swarm( + key: Keypair, + features: Features, + ignored: Vec<IpRange>, +) -> Result<Swarm<Behaviour>> { + use ::libp2p::Transport as _; + use ::libp2p::core::transport::{MemoryTransport, OrTransport}; + use ::libp2p::core::upgrade::Version; + + let peer_id = key.public().to_peer_id(); + let (relay_transport, relay_client) = ::libp2p::relay::client::new(peer_id); + let transport = OrTransport::new(relay_transport, MemoryTransport::default()) + .boxed() + .upgrade(Version::V1) + .authenticate(::libp2p::noise::Config::new(&key)?) + .multiplex(::libp2p::yamux::Config::default()) + .timeout(CONNECT_TIMEOUT) + .boxed(); + Ok(Swarm::new( + transport, + Behaviour::new(&key, relay_client, features, ignored)?, + peer_id, + swarm_config(::libp2p::swarm::Config::with_tokio_executor(), features), + )) +} diff --git a/src/transport/libp2p/book.rs b/src/transport/libp2p/book.rs new file mode 100644 index 0000000..5658c56 --- /dev/null +++ b/src/transport/libp2p/book.rs @@ -0,0 +1,524 @@ +//! The book: who is where (peers' addresses, connections and their round trips), the +//! relays and their slots, hole punching and upgrades, and the counts the figures show. + +use super::*; + +/// Who named a relay to pick, in the order they are picked. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Named { + /// `Libp2pConfig::relays`. + Configured, + /// The relays listed in the document (`Transport::learn_relays`). + Listed, +} + +impl Named { + pub(super) fn as_str(self) -> &'static str { + match self { + Self::Configured => "configured", + Self::Listed => "listed", + } + } +} + +/// A relay named to pick. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(super) struct NamedRelay { + /// Where, with the relay's peer ID. + pub(super) at: Multiaddr, + /// As named: without the peer ID, it is found out again when the relay doesn't + /// grant a slot (it may have a new one). + pub(super) given: Multiaddr, + pub(super) named: Named, +} + +/// Where peers are, and where we listen. +#[derive(Default)] +pub(super) struct Book { + pub(super) addresses: HashMap<PeerId, Vec<Multiaddr>>, + pub(super) listening: Vec<Multiaddr>, + /// Where the router forwards to our listen addresses (UPnP, `Libp2pConfig::upnp`). + pub(super) mapped: Vec<Multiaddr>, + /// `Libp2pConfig::announce`. + pub(super) announced: Vec<Multiaddr>, + /// `Libp2pConfig::ignore`: not published (but announced), not dialled. + pub(super) ignored: Vec<IpRange>, + /// `Libp2pConfig::relay`: this node relays, and says so. + pub(super) serving: bool, + /// Relays named to pick (configured, listed), by peer: picked first. + pub(super) named_relays: HashMap<PeerId, NamedRelay>, + /// Relays named without their peer ID, being found out (`Command::FindRelay`). + pub(super) finding: HashMap<Multiaddr, Named>, + /// The relays listed in the document, as last learned (`learn_listed`). + pub(super) listed: Vec<Multiaddr>, + /// Members that say they relay, and where (with the relay's peer ID). + pub(super) relays: HashMap<PeerId, Multiaddr>, + /// With `Libp2pConfig::public`: other peers that relay (identify says so), and + /// where. + pub(super) public_relays: HashMap<PeerId, Multiaddr>, + /// Relays not to be picked, until when (`RELAY_PAUSE`, `RELAY_RETRY`). + pub(super) failed_relays: HashMap<PeerId, tokio::time::Instant>, + /// Relays that granted us a slot we hold. + pub(super) slots: std::collections::HashSet<PeerId>, + /// Peers holding a slot on this node, as a relay. + pub(super) slots_here: std::collections::HashSet<PeerId>, + /// Peers connected only through a relay, dialled directly (`upgrade`): when to again + /// at the soonest, and how long to wait after that. + pub(super) upgrades: HashMap<PeerId, (tokio::time::Instant, Duration)>, + /// The shortest round trip measured on each open connection (libp2p's ping). + pub(super) rtts: HashMap<ConnectionId, Duration>, + /// Peers whose hole punching failed: how often in a row, and when to try again + /// (`PUNCH_RETRIES`). + pub(super) punching: HashMap<PeerId, (usize, tokio::time::Instant)>, + /// Open connections: to whom, whether through a relay, and the remote address. + pub(super) connections: HashMap<ConnectionId, (PeerId, bool, Multiaddr)>, + /// What AutoNAT says, and the address it confirmed while public. + pub(super) reachability: Reachability, + pub(super) public_address: Option<Multiaddr>, + /// Since start (`stats`). + pub(super) counts: Counts, + /// When each peer was last mentioned (`STALE_AFTER`). + pub(super) seen: HashMap<PeerId, tokio::time::Instant>, + /// Peers found on the LAN (mDNS) not known yet to speak continuum. + pub(super) lan: std::collections::HashSet<PeerId>, + /// Peers that speak continuum (identify says so). + pub(super) continuum: std::collections::HashSet<PeerId>, + /// Peers identify says don't speak continuum, and whether they relay: not sent to + /// (`send_encoded`), so that a seed that's no node says so. + pub(super) not_nodes: HashMap<PeerId, bool>, + /// Peers found on the LAN that speak continuum, with their address: seeds not + /// handed out yet (`discovered_seeds`). + pub(super) discovered: Vec<Multiaddr>, +} + +/// What `Libp2pStats` counts since start. +#[derive(Default)] +pub(super) struct Counts { + pub(super) slots_granted: u64, + pub(super) slots_lost: u64, + pub(super) slots_not_granted: u64, + pub(super) hole_punches: HolePunches, + pub(super) dial_failures: DialFailures, +} + +impl Book { + /// How we are connected to `peer`: directly if any connection is direct. + pub(super) fn connection(&self, peer: &PeerId) -> Option<Connection> { + book_connection(&self.connections, peer) + } + + /// The connections to `peer` to close, of several: all but one, the oldest direct + /// one (the oldest relayed one if none is direct). + /// + /// None between a relay and a node with a slot on it, or reserving one (`reserving`): + /// a slot lives on one connection, which libp2p doesn't tell, and closing it would + /// end the slot, and with it every circuit to the node (the relay then answers + /// `NoReservation`). + pub(super) fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec<ConnectionId> { + if reserving || self.slots.contains(peer) || self.slots_here.contains(peer) { + return Vec::new(); + } + self.ranked(peer).into_iter().skip(1).collect() + } + + /// The connections to `peer`, the one to keep first: direct ones before relayed + /// ones, measured ones by their round trip (`rtts`: over the LAN before a VPN's hub, + /// whichever opened first), then the oldest. + pub(super) fn ranked(&self, peer: &PeerId) -> Vec<ConnectionId> { + let mut connections: Vec<(bool, Duration, ConnectionId)> = self + .connections + .iter() + .filter(|(_, (connected, _, _))| connected == peer) + .map(|(id, (_, relayed, _))| { + let rtt = self.rtts.get(id).copied().unwrap_or(Duration::MAX); + (*relayed, rtt, *id) + }) + .collect(); + connections.sort(); + connections.into_iter().map(|(_, _, id)| id).collect() + } + + /// The remote address of the connection to `peer` that is kept (`ranked`). + pub(super) fn location(&self, peer: &PeerId) -> Option<&Multiaddr> { + let kept = *self.ranked(peer).first()?; + self.connections.get(&kept).map(|(_, _, address)| address) + } + + pub(super) fn stats(&self) -> Libp2pStats { + let mut connections = Connections::default(); + for (_, relayed, _) in self.connections.values() { + if *relayed { + connections.relayed += 1; + } else { + connections.direct += 1; + } + } + Libp2pStats { + reachability: self.reachability, + public_address: self.public_address.as_ref().map(Multiaddr::to_string), + connections, + relay_slots: RelaySlots { + held: self.slots.len(), + paused: self + .failed_relays + .values() + .filter(|until| tokio::time::Instant::now() < **until) + .count(), + granted: self.counts.slots_granted, + lost: self.counts.slots_lost, + not_granted: self.counts.slots_not_granted, + }, + hole_punches: self.counts.hole_punches, + dial_failures: self.counts.dial_failures, + } + } + + /// `peer` was mentioned just now. + pub(super) fn touch(&mut self, peer: PeerId) { + self.seen.insert(peer, tokio::time::Instant::now()); + } + + /// Forgets the peers not mentioned within `STALE_AFTER` before `now`, but those + /// `connected`. Returns how many. + pub(super) fn prune( + &mut self, + now: tokio::time::Instant, + connected: impl Fn(&PeerId) -> bool, + ) -> usize { + let stale: Vec<PeerId> = self + .addresses + .keys() + .chain(self.relays.keys()) + .chain(self.public_relays.keys()) + .chain(self.continuum.iter()) + .chain(self.lan.iter()) + .filter(|peer| { + !connected(peer) + && self + .seen + .get(peer) + .is_none_or(|at| now.saturating_duration_since(*at) >= STALE_AFTER) + }) + .copied() + .collect::<std::collections::HashSet<_>>() + .into_iter() + .collect(); + for peer in &stale { + self.addresses.remove(peer); + self.relays.remove(peer); + self.public_relays.remove(peer); + self.continuum.remove(peer); + self.not_nodes.remove(peer); + self.lan.remove(peer); + self.failed_relays.remove(peer); + self.upgrades.remove(peer); + self.punching.remove(peer); + self.seen.remove(peer); + } + stale.len() + } + + /// `peer`, found on the LAN, speaks continuum: a seed, at the address we know first. + pub(super) fn discover(&mut self, peer: PeerId) { + let Some(address) = self.addresses.get(&peer).and_then(|known| known.first()) else { + return; + }; + if let Ok(address) = address.clone().with_p2p(peer) + && !self.discovered.contains(&address) + && self.discovered.len() < MAX_PEERS + { + self.discovered.push(address); + } + } + + /// Where others could dial us directly: the announced addresses, or else those we + /// listen on (`published`), where the router forwards to them (`mapped`), and where + /// AutoNAT found others can dial us (`public_address`: a port forwarded by hand, say), + /// without the peer ID. + pub(super) fn direct(&self) -> Vec<Multiaddr> { + if !self.announced.is_empty() { + return self.announced.clone(); + } + let direct: Vec<Multiaddr> = self + .listening + .iter() + .filter(|address| !is_relayed(address)) + .cloned() + .collect(); + let mut direct = published(&direct); + let confirmed = self.public_address.iter().filter_map(|address| { + let mut address = address.clone(); + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + (!address.is_empty() && !is_relayed(&address)).then_some(address) + }); + // First: the member list keeps only so many (`members::MAX_ADDRESSES`), and a + // node listening everywhere has many. + let mut first: Vec<Multiaddr> = Vec::new(); + for address in self.mapped.iter().cloned().chain(confirmed) { + if !first.contains(&address) { + first.push(address); + } + } + direct.retain(|address| !first.contains(address)); + first.extend(direct); + first.retain(|address| !ignored(&self.ignored, address)); + first + } +} + +/// Whether `address` only reaches this host (or this link): no use to other hosts, and +/// dialled by one, it reaches that host itself. +pub(super) fn is_local(address: &Multiaddr) -> bool { + match address.iter().next() { + Some(Protocol::Ip4(ip)) => ip.is_loopback() || ip.is_link_local() || ip.is_unspecified(), + Some(Protocol::Ip6(ip)) => { + ip.is_loopback() || ip.is_unicast_link_local() || ip.is_unspecified() + } + _ => false, + } +} + +/// The addresses of `all` worth publishing: all but local ones, unless they're all +/// local (a test on one host). +pub(super) fn published(all: &[Multiaddr]) -> Vec<Multiaddr> { + let reachable: Vec<Multiaddr> = all + .iter() + .filter(|address| !is_local(address)) + .cloned() + .collect(); + if reachable.is_empty() { + all.to_vec() + } else { + reachable + } +} + +impl Book { + /// `named` names the relay at `given`; returns it if its peer is to be found out. + /// `local`: this node, never its own relay. + pub(super) fn name_relay( + &mut self, + given: Multiaddr, + named: Named, + local: PeerId, + ) -> Option<Multiaddr> { + match split_peer(&given) { + Some(peer) if peer == local => None, + Some(peer) => { + // Configured wins over listed. + if self + .named_relays + .get(&peer) + .is_none_or(|known| known.named == Named::Listed || named == Named::Configured) + { + self.named_relays.insert( + peer, + NamedRelay { + at: given.clone(), + given, + named, + }, + ); + } + None + } + None => { + let known = self.finding.contains_key(&given) + || self.named_relays.values().any(|relay| relay.given == given); + if known { + return None; + } + self.finding.insert(given.clone(), named); + Some(given) + } + } + } + + /// The relay named at `given` (without a peer ID) is `peer`: unless it is no longer + /// named (or is this node). Returns whether that's news. + pub(super) fn relay_found(&mut self, given: Multiaddr, peer: PeerId, local: PeerId) -> bool { + let Some(named) = self.finding.remove(&given) else { + return false; + }; + let Ok(at) = given.clone().with_p2p(peer) else { + return false; + }; + if peer == local + || (named == Named::Listed + && self + .named_relays + .get(&peer) + .is_some_and(|known| known.named == Named::Configured)) + { + return false; + } + self.named_relays + .insert(peer, NamedRelay { at, given, named }); + true + } + + /// `relay` didn't grant a slot: if it was named without its peer ID, its address is + /// to be found out again, returned. + pub(super) fn find_again(&mut self, relay: PeerId) -> Option<Multiaddr> { + if self + .named_relays + .get(&relay) + .is_none_or(|known| split_peer(&known.given).is_some()) + { + return None; + } + let known = self.named_relays.remove(&relay)?; + self.finding.insert(known.given.clone(), known.named); + Some(known.given) + } + + /// The relays listed in the document are `listed` now: returns those to find out, + /// None if nothing changed. + pub(super) fn learn_listed( + &mut self, + listed: Vec<Multiaddr>, + local: PeerId, + ) -> Option<Vec<Multiaddr>> { + if listed == self.listed { + return None; + } + self.named_relays + .retain(|_, relay| relay.named != Named::Listed || listed.contains(&relay.given)); + self.finding + .retain(|given, named| *named != Named::Listed || listed.contains(given)); + self.listed = listed.clone(); + Some( + listed + .into_iter() + .filter_map(|given| self.name_relay(given, Named::Listed, local)) + .collect(), + ) + } + + /// The named relays `named` names, by peer, where. + pub(super) fn named(&self, named: Named) -> HashMap<PeerId, Multiaddr> { + self.named_relays + .iter() + .filter(|(_, relay)| relay.named == named) + .map(|(peer, relay)| (*peer, relay.at.clone())) + .collect() + } + + /// Where `peer` may be reached through the relays we know (named, then members that + /// relay): for a peer known by its ID only (a seed `/p2p/<peer ID>`), which may hold + /// a slot on one of them. `local`: this node, never its own relay. + pub(super) fn through_relays(&self, peer: PeerId, local: PeerId) -> Vec<Multiaddr> { + let mut relays: Vec<(PeerId, Multiaddr)> = self + .named_relays + .iter() + .map(|(relay, named)| (*relay, named.at.clone())) + .collect(); + relays.sort_by_key(|(_, address)| address.to_string()); + let mut members: Vec<(PeerId, Multiaddr)> = self + .relays + .iter() + .filter(|(relay, _)| !self.named_relays.contains_key(relay)) + .map(|(relay, address)| (*relay, address.clone())) + .collect(); + members.sort_by_key(|(_, address)| address.to_string()); + relays + .into_iter() + .chain(members) + .filter(|(relay, _)| *relay != local && *relay != peer) + .map(|(_, address)| address.with(Protocol::P2pCircuit).with(Protocol::P2p(peer))) + .take(ADDRESSES_PER_PEER) + .collect() + } + + /// Whether all of `addresses` are known of `peer` (`remember`). + pub(super) fn knows(&self, peer: PeerId, addresses: &[Multiaddr]) -> bool { + let Some(known) = self.addresses.get(&peer) else { + return false; + }; + addresses.iter().all(|address| { + let mut address = address.clone(); + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + known.contains(&address) + }) + } + + /// `peer` is at `address` (without its `/p2p` part): kept first. + pub(super) fn remember(&mut self, peer: PeerId, mut address: Multiaddr) { + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + if !self.addresses.contains_key(&peer) && self.addresses.len() >= MAX_PEERS { + return; + } + self.touch(peer); + let addresses = self.addresses.entry(peer).or_default(); + addresses.retain(|known| *known != address); + addresses.insert(0, address); + addresses.truncate(ADDRESSES_PER_PEER); + } +} + +/// Whether `address` goes through a relay. +pub(super) fn is_relayed(address: &Multiaddr) -> bool { + address + .iter() + .any(|protocol| matches!(protocol, Protocol::P2pCircuit)) +} + +/// The relay `address` goes through, if it goes through one: the peer before +/// `/p2p-circuit`. +pub(super) fn relay_of(address: &Multiaddr) -> Option<PeerId> { + let mut relay = None; + for protocol in address.iter() { + match protocol { + Protocol::P2p(peer) => relay = Some(peer), + Protocol::P2pCircuit => return relay, + _ => {} + } + } + None +} + +/// How `peer` is connected, of `connections` (`Book::connections`): directly if one is +/// direct. +pub(super) fn book_connection( + connections: &HashMap<ConnectionId, (PeerId, bool, Multiaddr)>, + peer: &PeerId, +) -> Option<Connection> { + let mut found = None; + for (connected, relayed, _) in connections.values() { + if connected == peer { + if !relayed { + return Some(Connection::Direct); + } + found = Some(Connection::Relayed); + } + } + found +} + +/// The addresses of `known` this node can dial: not those through itself as the relay +/// (a relay reaches the nodes it relays for only while they are connected to it). +pub(super) fn dialable(known: &[Multiaddr], local: PeerId) -> Vec<Multiaddr> { + known + .iter() + .filter(|address| relay_of(address) != Some(local)) + .cloned() + .collect() +} + +/// `address` with the peer ID it ends in, if it ends in one. +pub(super) fn split_peer(address: &Multiaddr) -> Option<PeerId> { + match address.iter().last() { + Some(Protocol::P2p(peer)) => Some(peer), + _ => None, + } +} + +pub(super) fn lock(book: &Mutex<Book>) -> MutexGuard<'_, Book> { + book.lock_or_recover() +} diff --git a/src/transport/libp2p/config.rs b/src/transport/libp2p/config.rs new file mode 100644 index 0000000..012999d --- /dev/null +++ b/src/transport/libp2p/config.rs @@ -0,0 +1,117 @@ +//! What the options make of the libp2p transport (`Libp2pConfig`), and their checks. + +use super::*; + +/// What a libp2p transport is made with. +pub(crate) struct Libp2pConfig { + pub(crate) key: Keypair, + /// Multiaddrs to listen on. + pub(crate) listen: Vec<Multiaddr>, + /// Multiaddrs to publish instead of the listen addresses, if any. + pub(crate) announce: Vec<Multiaddr>, + /// Relays to reserve a slot on (multiaddrs; without the relay's peer ID, a connection + /// finds it out). + pub(crate) relays: Vec<Multiaddr>, + /// Relay for other nodes. + pub(crate) relay: bool, + /// Speak continuum (`PROTOCOL`): not on a node that only relays (`--relay-only`), + /// which nobody should take for a node. + pub(crate) continuum: bool, + /// Find peers on the LAN (mDNS). + pub(crate) mdns: bool, + /// Ask the router to forward our listen ports (UPnP), and publish where. + pub(crate) upnp: bool, + /// Address ranges neither published nor dialled, nor connected over (`Ignoring`). + pub(crate) ignore: Vec<IpRange>, + /// Use public infrastructure: the IPFS DHT, public relays. + pub(crate) public: bool, + /// libp2p's in-memory transport instead of TCP and QUIC: for tests. + pub(crate) memory: bool, +} + +/// The multiaddrs of option `option`. +pub(super) fn parse_addresses(option: &str, addresses: &[String]) -> Result<Vec<Multiaddr>> { + addresses + .iter() + .map(|address| { + address + .parse::<Multiaddr>() + .with_context(|| format!("{option} {address:?}")) + }) + .collect() +} + +/// The ranges of `libp2p-ignore`. +pub(super) fn parse_ranges(ranges: &[String]) -> Result<Vec<IpRange>> { + ranges + .iter() + .map(|range| range.parse().context("libp2p-ignore")) + .collect() +} + +/// This transport's options, checked when the configuration is loaded +/// (`config::Settings::check`), so `--print-config` tells a wrong one too: the +/// multiaddrs and ranges parse, and a relay that is no node (`--relay-only`) can be +/// dialled and keeps its peer ID. +pub(crate) fn check(settings: &crate::config::Settings) -> Result<()> { + parse_addresses("libp2p-listen", &settings.libp2p_listen)?; + parse_addresses("libp2p-announce", &settings.libp2p_announce)?; + parse_addresses("libp2p-relays", &settings.libp2p_relays)?; + parse_ranges(&settings.libp2p_ignore)?; + if settings.relay_only { + if settings.libp2p_listen.is_empty() { + bail!("relay-only needs libp2p-listen: where nodes reach the relay"); + } + if settings.libp2p_key_file.is_none() { + bail!( + "relay-only needs libp2p-key-file: without it, the relay's peer ID changes with every start and nodes lose it" + ); + } + } + Ok(()) +} + +impl Libp2pConfig { + /// The configuration `settings` give; `relay_only`: a relay that is no node + /// (`--relay-only`). + pub(crate) fn from_settings( + settings: &crate::config::Settings, + relay_only: bool, + ) -> Result<Self> { + let parse = parse_addresses; + Ok(Self { + key: Self::load_key(settings.libp2p_key_file.as_deref())?, + listen: parse("libp2p-listen", &settings.libp2p_listen)?, + announce: parse("libp2p-announce", &settings.libp2p_announce)?, + relays: parse("libp2p-relays", &settings.libp2p_relays)?, + relay: settings.libp2p_relay || relay_only, + continuum: !relay_only, + mdns: settings.libp2p_mdns, + upnp: settings.libp2p_upnp, + ignore: parse_ranges(&settings.libp2p_ignore)?, + public: settings.libp2p_public, + memory: false, + }) + } + + /// The key in `path` (32 raw bytes, an Ed25519 secret key), or a new one each start + /// without a file: the peer ID only says where a node is, so a new one costs nothing + /// but the WHO that finds it again (`transport::identity`). + pub(crate) fn load_key(path: Option<&Path>) -> Result<Keypair> { + let Some(path) = path else { + return Ok(Keypair::generate_ed25519()); + }; + let mut bytes = std::fs::read(path) + .with_context(|| format!("Reading the libp2p key from {}", path.display()))?; + crate::swarm::auth::warn_if_readable_by_others(path, "libp2p key"); + if bytes.len() != 32 { + bail!( + "The libp2p key in {} has {} bytes; it has 32", + path.display(), + bytes.len() + ); + } + Keypair::ed25519_from_bytes(&mut bytes) + .map_err(|error| anyhow!("The libp2p key in {}: {error}", path.display())) + } +} diff --git a/src/transport/libp2p/driver.rs b/src/transport/libp2p/driver.rs new file mode 100644 index 0000000..efcf109 --- /dev/null +++ b/src/transport/libp2p/driver.rs @@ -0,0 +1,527 @@ +//! The swarm's task (`run`): it owns the swarm, takes `Command`s from the transport, and +//! answers the swarm's events (connections, identify, relays, hole punching, AutoNAT, +//! mDNS, UPnP, the DHT), keeping the book. + +use super::*; + +pub(super) enum Command { + /// Connect to `peer` (at `addresses`) unless connected. + Connect { + peer: PeerId, + addresses: Vec<Multiaddr>, + done: oneshot::Sender<Result<(), String>>, + }, + /// Connect to `address` to find out which peer is there. + Identify { + address: Multiaddr, + done: oneshot::Sender<Result<PeerId, String>>, + }, + /// See whether relays are needed now (one was just learned of). + CheckRelays, + /// Find out which peer relays at `address`, named without the relay's peer ID + /// (`Book::finding`). + FindRelay { address: Multiaddr }, +} + +/// Drives the swarm: dials what `commands` ask for and notes where we listen, until the +/// transport is dropped. `again`: to schedule commands. +pub(super) async fn run( + mut swarm: Swarm<Behaviour>, + mut commands: mpsc::UnboundedReceiver<Command>, + again: mpsc::UnboundedSender<Command>, + book: Arc<Mutex<Book>>, + relaying: Relaying, + stopping: watch::Sender<bool>, +) { + let serving = relaying.serving; + let mut nat = ::libp2p::autonat::NatStatus::Unknown; + // Relays picked, by listener. + let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new(); + let mut check = tokio::time::interval(RELAY_CHECK); + check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP); + dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut connecting: HashMap<PeerId, Vec<oneshot::Sender<Result<(), String>>>> = HashMap::new(); + let mut identifying: HashMap< + ConnectionId, + (Multiaddr, oneshot::Sender<Result<PeerId, String>>), + > = HashMap::new(); + let mut prune = tokio::time::interval(PRUNE_EVERY); + prune.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // Peers that may have several connections, and when to cut them to one. Only the + // side with the smaller peer ID does that, so the two never close different ones. + let mut dedupe_due: HashMap<PeerId, tokio::time::Instant> = HashMap::new(); + // Connections closed as duplicates, until they are: their "Connection closed" says so. + let mut closing_duplicates: std::collections::HashSet<ConnectionId> = + std::collections::HashSet::new(); + let mut dedupe = tokio::time::interval(DEDUPE_EVERY); + dedupe.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tokio::select! { + _ = dedupe.tick(), if !dedupe_due.is_empty() => { + let now = tokio::time::Instant::now(); + let due: Vec<PeerId> = dedupe_due + .iter() + .filter(|(_, at)| **at <= now) + .map(|(peer, _)| *peer) + .collect(); + for peer in due { + dedupe_due.remove(&peer); + let reserving = picked.values().any(|relay| *relay == peer); + let duplicates = lock(&book).duplicates(&peer, reserving); + for connection in duplicates { + if swarm.close_connection(connection) { + closing_duplicates.insert(connection); + } + } + } + } + _ = prune.tick() => { + let forgotten = lock(&book).prune(tokio::time::Instant::now(), |peer| swarm.is_connected(peer)); + if forgotten > 0 { + debug!(forgotten, "Forgot peers nobody mentions any more"); + } + } + _ = dht_bootstrap.tick(), if relaying.public => { + if let Some(dht) = swarm.behaviour_mut().kad.as_mut() + && let Err(error) = dht.bootstrap() + { + warn!(error = %error, "Couldn't join the IPFS DHT"); + } + } + _ = check.tick() => { + pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked); + retry_punches(&mut swarm, &book, relaying.local); + } + command = commands.recv() => match command { + None => break, + Some(Command::CheckRelays) => { + pick_relays(&mut swarm, &book, &relaying, &nat, &mut picked); + } + Some(Command::Connect { peer, addresses, done }) => { + if swarm.is_connected(&peer) { + let _ = done.send(Ok(())); + upgrade(&mut swarm, &book, peer, relaying.local); + continue; + } + let waiting = connecting.entry(peer).or_default(); + waiting.push(done); + if waiting.len() > 1 { + continue; // Already dialing. + } + let mut opts = DialOpts::peer_id(peer) + .addresses(addresses) + .condition(PeerCondition::DisconnectedAndNotDialing); + if relaying.public { + // Addresses the DHT knows too. + opts = opts.extend_addresses_through_behaviour(); + } + if let Err(error) = swarm.dial(opts.build()) { + count_dial_failure(&mut lock(&book).counts.dial_failures, &error); + for done in connecting.remove(&peer).into_iter().flatten() { + let _ = done.send(Err(error.to_string())); + } + find_on_dht(&mut swarm, peer); + } + } + Some(Command::FindRelay { address }) => { + let again = again.clone(); + let local = relaying.local; + let book = Arc::clone(&book); + tokio::spawn(async move { + let (done, identified) = oneshot::channel(); + if again + .send(Command::Identify { address: address.clone(), done }) + .is_err() + { + return; + } + match tokio::time::timeout(CONNECT_TIMEOUT, identified).await { + Ok(Ok(Ok(peer))) => { + if lock(&book).relay_found(address.clone(), peer, local) { + debug!(relay = %Addressed(SCHEME, &address), peer = %Addressed(SCHEME, &peer), "Found a named relay"); + let _ = again.send(Command::CheckRelays); + } + } + failed => { + if !lock(&book).finding.contains_key(&address) { + return; // No longer named. + } + let error = match failed { + Ok(Ok(Err(error))) => error, + Ok(_) => "the libp2p swarm stopped".to_owned(), + Err(_) => format!("no answer within {CONNECT_TIMEOUT:?}"), + }; + warn!(relay = %Addressed(SCHEME, &address), error = %error, retry = ?RELAY_RETRY, "Couldn't reach a named relay; trying again later"); + tokio::time::sleep(RELAY_RETRY).await; + let _ = again.send(Command::FindRelay { address }); + } + } + }); + } + Some(Command::Identify { address, done }) => { + let opts = DialOpts::unknown_peer_id().address(address.clone()).build(); + let id = opts.connection_id(); + match swarm.dial(opts) { + Ok(()) => { + identifying.insert(id, (address, done)); + } + Err(error) => { + let _ = done.send(Err(error.to_string())); + } + } + } + }, + event = swarm.select_next_some() => match event { + SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, .. } => { + { + let mut book = lock(&book); + book.touch(peer_id); + let relayed = endpoint.is_relayed(); + book.connections.insert( + connection_id, + (peer_id, relayed, endpoint.get_remote_address().clone()), + ); + if !relayed { + book.upgrades.remove(&peer_id); + book.punching.remove(&peer_id); + } + if relaying.local < peer_id + && book.connections.values().filter(|(connected, _, _)| *connected == peer_id).count() > 1 + { + // Only through relays: maybe a punch under way. + let grace = match book.connection(&peer_id) { + Some(Connection::Relayed) => PUNCH_GRACE, + _ => DUPLICATE_GRACE, + }; + dedupe_due.insert(peer_id, tokio::time::Instant::now() + grace); + } + debug!( + peer = %Addressed(SCHEME, &peer_id), + connection = %connection_id, + dialled = endpoint.is_dialer(), + address = %endpoint.get_remote_address(), + relayed, + "Connection opened" + ); + } + for done in connecting.remove(&peer_id).into_iter().flatten() { + let _ = done.send(Ok(())); + } + if let Some((address, done)) = identifying.remove(&connection_id) { + lock(&book).remember(peer_id, address); + let _ = done.send(Ok(peer_id)); + } + } + SwarmEvent::ConnectionClosed { peer_id, connection_id, cause, .. } => { + { + let mut book = lock(&book); + book.connections.remove(&connection_id); + book.rtts.remove(&connection_id); + } + let cause = match cause { + _ if closing_duplicates.remove(&connection_id) => "a duplicate".to_owned(), + // By us, or idle: libp2p doesn't tell which. + None => "closed here".to_owned(), + Some(error) => error.to_string(), + }; + debug!( + peer = %Addressed(SCHEME, &peer_id), + connection = %connection_id, + cause = %cause, + "Connection closed" + ); + } + SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => { + count_dial_failure(&mut lock(&book).counts.dial_failures, &error); + if let Some((_, done)) = identifying.remove(&connection_id) { + let _ = done.send(Err(describe_dial_error(&error))); + } + if let Some(peer) = peer_id + && !swarm.is_connected(&peer) + && let Some(waiting) = connecting.remove(&peer) + { + for done in waiting { + let _ = done.send(Err(describe_dial_error(&error))); + } + find_on_dht(&mut swarm, peer); + } + } + SwarmEvent::NewListenAddr { address, .. } => { + info!(address = %address, "Listening"); + // A relay tells those reserving where it is: its external addresses. + if serving && !is_local(&address) && !is_relayed(&address) { + swarm.add_external_address(address.clone()); + } + let mut book = lock(&book); + if !book.listening.contains(&address) { + book.listening.push(address); + } + } + SwarmEvent::ExpiredListenAddr { address, .. } => { + info!(address = %address, "No longer listening"); + if serving { + swarm.remove_external_address(&address); + } + lock(&book).listening.retain(|known| *known != address); + } + SwarmEvent::ListenerClosed { listener_id, addresses, reason } => { + lock(&book) + .listening + .retain(|known| !addresses.contains(known)); + // A relay picked: the next check picks again, another one if this one + // never granted the slot. + if let Some(relay) = picked.remove(&listener_id) { + let refused = match &reason { + Ok(()) => false, + Err(error) => relay_refused(&format!("{error:?}"), &innermost_causes(error)), + }; + let (lost, pause) = { + let mut book = lock(&book); + let lost = book.slots.remove(&relay); + let pause = match (refused, lost) { + (true, _) => Some(RELAY_PAUSE), + (false, true) => None, + (false, false) => Some(RELAY_RETRY), + }; + if lost { + book.counts.slots_lost += 1; + } else { + book.counts.slots_not_granted += 1; + // Named without its peer ID: maybe another one now. + if let Some(address) = book.find_again(relay) { + let _ = again.send(Command::FindRelay { address }); + } + } + if let Some(pause) = pause { + book.failed_relays.insert(relay, tokio::time::Instant::now() + pause); + } + (lost, pause) + }; + // Again at once, or others meanwhile. + let _ = again.send(Command::CheckRelays); + let reason = relay_failure(&reason); + match (lost, pause) { + (true, None) => { + info!(relay = %Addressed(SCHEME, &relay), reason = %reason, "Lost the slot on a relay; reserving again"); + } + (true, Some(pause)) => { + info!(relay = %Addressed(SCHEME, &relay), pause = ?pause, reason = %reason, "Lost the slot on a relay; picking others for a while"); + } + (false, pause) => { + info!( + relay = %Addressed(SCHEME, &relay), + pause = ?pause.unwrap_or_default(), + reason = %reason, + "A relay didn't grant a slot; picking others for a while" + ); + } + } + } + } + SwarmEvent::ListenerError { error, .. } => { + warn!(error = %error, "Listener failed"); + } + SwarmEvent::Behaviour(BehaviourEvent::RelayClient( + ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. }, + )) => { + { + let mut book = lock(&book); + book.slots.insert(relay_peer_id); + book.failed_relays.remove(&relay_peer_id); + if !renewal { + book.counts.slots_granted += 1; + } + } + if !renewal { + info!(relay = %Addressed(SCHEME, &relay_peer_id), "Reachable through a relay"); + } + } + SwarmEvent::Behaviour(BehaviourEvent::Autonat( + ::libp2p::autonat::Event::StatusChanged { old, new }, + )) => { + info!(old = ?old, new = ?new, "Reachability (AutoNAT)"); + { + use ::libp2p::autonat::NatStatus; + let mut book = lock(&book); + (book.reachability, book.public_address) = match &new { + NatStatus::Unknown => (Reachability::Unknown, None), + NatStatus::Private => (Reachability::Private, None), + NatStatus::Public(address) => { + (Reachability::Public, Some(address.clone())) + } + }; + } + nat = new; + } + SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => { + { + use ::libp2p::relay::Event; + let mut book = lock(&book); + match &event { + Event::ReservationReqAccepted { src_peer_id, .. } => { + book.slots_here.insert(*src_peer_id); + } + Event::ReservationClosed { src_peer_id } + | Event::ReservationTimedOut { src_peer_id } => { + book.slots_here.remove(src_peer_id); + } + _ => {} + } + } + debug!(event = ?event, "Relay event"); + } + SwarmEvent::Behaviour(BehaviourEvent::Ping(::libp2p::ping::Event { + peer, + connection, + result, + })) => match result { + Ok(rtt) => { + let mut book = lock(&book); + if book.connections.contains_key(&connection) { + let shortest = book.rtts.entry(connection).or_insert(rtt); + *shortest = (*shortest).min(rtt); + } + trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, rtt = ?rtt, "Connection's round trip"); + } + Err(error) => { + trace!(peer = %Addressed(SCHEME, &peer), connection = %connection, error = %error, "Connection's ping failed"); + } + }, + SwarmEvent::Behaviour(BehaviourEvent::Upnp(event)) => match event { + ::libp2p::upnp::Event::NewExternalAddr { + local_addr, + external_addr: address, + } => { + info!(address = %address, to = %local_addr, "The router forwards to us (UPnP)"); + let mut book = lock(&book); + if !book.mapped.contains(&address) { + book.mapped.push(address); + } + } + ::libp2p::upnp::Event::ExpiredExternalAddr { + local_addr, + external_addr: address, + } => { + info!(address = %address, to = %local_addr, "The router no longer forwards to us (UPnP)"); + lock(&book).mapped.retain(|mapped| *mapped != address); + } + ::libp2p::upnp::Event::GatewayNotFound => { + info!("No router that maps ports (UPnP) answered"); + } + ::libp2p::upnp::Event::NonRoutableGateway => { + warn!("The router that maps ports (UPnP) is behind another NAT itself; its mappings don't reach us from outside"); + } + }, + SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result { + Ok(_) => { + let mut book = lock(&book); + book.counts.hole_punches.succeeded += 1; + book.punching.remove(&event.remote_peer_id); + drop(book); + info!(peer = %Addressed(SCHEME, &event.remote_peer_id), "Connected directly to a peer behind a relay"); + } + Err(error) => { + let retry = { + let mut book = lock(&book); + book.counts.hole_punches.failed += 1; + let failed = book + .punching + .get(&event.remote_peer_id) + .map_or(0, |(failed, _)| *failed); + let retry = PUNCH_RETRIES[failed.min(PUNCH_RETRIES.len() - 1)]; + book.punching.insert( + event.remote_peer_id, + (failed + 1, tokio::time::Instant::now() + retry), + ); + retry + }; + debug!(peer = %Addressed(SCHEME, &event.remote_peer_id), error = %error, retry = ?retry, "Couldn't connect directly to a peer behind a relay; the relay carries on"); + } + }, + SwarmEvent::Behaviour(BehaviourEvent::Identify( + ::libp2p::identify::Event::Received { connection_id, peer_id, info }, + )) => { + { + let mut book = lock(&book); + let listen = published(&info.listen_addrs); + // Every new connection identifies again: logged when it says something new. + if book.knows(peer_id, &listen) { + trace!(peer = %Addressed(SCHEME, &peer_id), connection = %connection_id, "Peer identified again"); + } else { + debug!( + peer = %Addressed(SCHEME, &peer_id), + connection = %connection_id, + listen = ?info.listen_addrs, + observed = %info.observed_addr, + agent = %info.agent_version, + "Peer identified" + ); + } + // Last first: `remember` puts each in front. + for address in listen.into_iter().rev() { + book.remember(peer_id, address); + } + if relaying.public + && info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME) + && (book.public_relays.len() < MAX_PEERS + || book.public_relays.contains_key(&peer_id)) + && let Some(address) = book + .addresses + .get(&peer_id) + .and_then(|known| known.first()) + .and_then(|address| address.clone().with_p2p(peer_id).ok()) + { + book.public_relays.insert(peer_id, address); + } + if !info.protocols.contains(&PROTOCOL) + && (book.not_nodes.len() < MAX_PEERS + || book.not_nodes.contains_key(&peer_id)) + { + let relays = info.protocols.contains(&::libp2p::relay::HOP_PROTOCOL_NAME); + book.not_nodes.insert(peer_id, relays); + } + if info.protocols.contains(&PROTOCOL) { + book.not_nodes.remove(&peer_id); + book.continuum.insert(peer_id); + if book.lan.remove(&peer_id) { + book.discover(peer_id); + } + } + } + upgrade(&mut swarm, &book, peer_id, relaying.local); + } + SwarmEvent::Behaviour(BehaviourEvent::Mdns( + ::libp2p::mdns::Event::Discovered(found), + )) => { + for (peer, address) in found { + if peer == relaying.local { + continue; + } + debug!(peer = %Addressed(SCHEME, &peer), address = %address, "Peer on the LAN"); + let continuum = { + let mut book = lock(&book); + book.remember(peer, address.clone()); + book.continuum.contains(&peer) + }; + upgrade(&mut swarm, &book, peer, relaying.local); + if continuum { + lock(&book).discover(peer); + } else if lock(&book).lan.len() < MAX_PEERS { + // Connected, identify says whether it speaks continuum. + lock(&book).lan.insert(peer); + let _ = swarm.dial( + DialOpts::peer_id(peer) + .addresses(vec![address]) + .condition(PeerCondition::DisconnectedAndNotDialing) + .build(), + ); + } + } + } + _ => {} + }, + } + } + let _ = stopping.send(true); +} diff --git a/src/transport/libp2p/errors.rs b/src/transport/libp2p/errors.rs new file mode 100644 index 0000000..7eb4551 --- /dev/null +++ b/src/transport/libp2p/errors.rs @@ -0,0 +1,128 @@ +//! What libp2p's errors say, for the log and the figures: dial failures, relay +//! reservations refused or lost. + +use super::*; + +/// A dial error, shortly: for each address tried (without the peer ID the caller names +/// already), the innermost cause, once. libp2p's own text nests each transport's errors +/// in the next ("Multiple dial errors occurred: ..."), repeating the cause at each level. +pub(super) fn describe_dial_error(error: &::libp2p::swarm::DialError) -> String { + use ::libp2p::swarm::DialError; + let DialError::Transport(errors) = error else { + return error.to_string(); + }; + let described: Vec<String> = errors + .iter() + .map(|(address, error)| { + let mut address = address.clone(); + if let Some(Protocol::P2p(_)) = address.iter().last() { + address.pop(); + } + format!("{address}: {}", innermost_causes(error)) + }) + .collect(); + if described.is_empty() { + return error.to_string(); + } + described.join("; ") +} + +/// Why a slot on a relay ended, or was never granted, shortly: libp2p nests the +/// relay's answer several types deep (`Left(Left(Left(Reservation(...))))`), and the +/// common cases deserve a sentence. +pub(super) fn relay_failure(reason: &std::io::Result<()>) -> String { + let Err(error) = reason else { + return "closed".to_owned(); + }; + relay_failure_text(&format!("{error:?}"), &innermost_causes(error)) +} + +/// `relay_failure` on the error's debug form, and its innermost cause as text. +pub(super) fn relay_failure_text(debug: &str, innermost: &str) -> String { + let known = [ + ( + "NoAddressesInReservation", + "the relay gave no address to be reached through it", + ), + ("ResourceLimitExceeded", "the relay has no slot left"), + ("PermissionDenied", "the relay doesn't let us reserve"), + ("NoReservation", "the relay holds no slot for us"), + ( + "connection gated", + "the relay refused the connection (connection gated)", + ), + ( + "UnsupportedProtocol", + "the relay doesn't relay (no circuit relay protocol)", + ), + ("Timeout", "the relay didn't answer in time"), + ]; + if let Some((_, said)) = known.iter().find(|(pattern, _)| debug.contains(pattern)) { + return (*said).to_owned(); + } + // The protocol refused, but only after its first data (`swarm_config`). + if innermost == "Failed" { + return "the relay refused the circuit relay protocol (is it a relay?)".to_owned(); + } + innermost.to_owned() +} + +/// Whether a slot ended, or wasn't granted, because the relay refused it (its error's +/// debug form and innermost cause, as for `relay_failure_text`), rather than because it +/// didn't answer or the connection closed: only then is it worth not asking again for a +/// while. +pub(super) fn relay_refused(debug: &str, innermost: &str) -> bool { + let refusals = [ + "NoAddressesInReservation", + "ResourceLimitExceeded", + "PermissionDenied", + "connection gated", + "UnsupportedProtocol", + "WrongPeerId", + ]; + refusals.iter().any(|pattern| debug.contains(pattern)) || innermost == "Failed" +} + +/// The innermost of `error`'s causes, each distinct one once: those libp2p lists in a +/// "Multiple dial errors occurred:" text (one per line, ` - <cause>: <its cause>...`), +/// or the last of the source chain. +pub(super) fn innermost_causes(error: &(dyn std::error::Error + 'static)) -> String { + let mut innermost = error; + while let Some(source) = innermost.source() { + innermost = source; + } + let text = innermost.to_string(); + let mut causes: Vec<&str> = Vec::new(); + for line in text.lines() { + let Some(listed) = line.trim().strip_prefix("- ") else { + continue; + }; + // A line ends in the next level's heading when that level lists more. + let cause = listed + .split(": ") + .filter(|part| !part.starts_with("Multiple dial errors")) + .last() + .unwrap_or(listed) + .trim(); + if !cause.is_empty() && !causes.contains(&cause) { + causes.push(cause); + } + } + if causes.is_empty() { + let cause = text.rsplit(": ").next().unwrap_or(&text).trim(); + return cause.to_owned(); + } + causes.join(", ") +} + +/// Counts a dial that failed with `error` (not one that wasn't needed). +pub(super) fn count_dial_failure(failures: &mut DialFailures, error: &::libp2p::swarm::DialError) { + use ::libp2p::swarm::DialError; + match error { + DialError::DialPeerConditionFalse(_) => {} + DialError::Transport(_) => failures.unreachable += 1, + DialError::WrongPeerId { .. } => failures.wrong_peer_id += 1, + DialError::NoAddresses => failures.no_addresses += 1, + _ => failures.other += 1, + } +} diff --git a/src/transport/libp2p/ignore.rs b/src/transport/libp2p/ignore.rs new file mode 100644 index 0000000..b14235e --- /dev/null +++ b/src/transport/libp2p/ignore.rs @@ -0,0 +1,146 @@ +//! `--libp2p-ignore`: address ranges neither published nor dialled, nor connected over +//! (`Ignoring`, a behaviour that denies connections there). + +use super::*; + +/// An address range (`10.244.0.0/16`, `42::/16`; a bare address is a range of one), for +/// `Libp2pConfig::ignore`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct IpRange { + pub(super) network: std::net::IpAddr, + pub(super) prefix: u8, +} + +impl FromStr for IpRange { + type Err = anyhow::Error; + + fn from_str(text: &str) -> Result<Self> { + let (address, prefix) = match text.split_once('/') { + Some((address, prefix)) => (address, Some(prefix)), + None => (text, None), + }; + let network: std::net::IpAddr = address + .parse() + .with_context(|| format!("{text:?} is no address range (e.g. 10.244.0.0/16)"))?; + let bits = if network.is_ipv4() { 32 } else { 128 }; + let prefix = match prefix { + Some(prefix) => prefix + .parse::<u8>() + .ok() + .filter(|prefix| *prefix <= bits) + .with_context(|| format!("{text:?}: the prefix is 0 to {bits}"))?, + None => bits, + }; + Ok(Self { network, prefix }) + } +} + +impl IpRange { + pub(super) fn contains(&self, ip: std::net::IpAddr) -> bool { + use std::net::IpAddr; + match (self.network, ip) { + (IpAddr::V4(network), IpAddr::V4(ip)) => { + let mask = u32::MAX + .checked_shl(32 - u32::from(self.prefix)) + .unwrap_or(0); + u32::from(network) & mask == u32::from(ip) & mask + } + (IpAddr::V6(network), IpAddr::V6(ip)) => { + let mask = u128::MAX + .checked_shl(128 - u32::from(self.prefix)) + .unwrap_or(0); + u128::from(network) & mask == u128::from(ip) & mask + } + _ => false, + } + } +} + +/// Whether `address` is in one of `ranges`: a direct address there (one through a relay +/// is judged by the connection to the relay). +pub(super) fn ignored(ranges: &[IpRange], address: &Multiaddr) -> bool { + if ranges.is_empty() || is_relayed(address) { + return false; + } + let ip = match address.iter().next() { + Some(Protocol::Ip4(ip)) => std::net::IpAddr::V4(ip), + Some(Protocol::Ip6(ip)) => std::net::IpAddr::V6(ip), + _ => return false, + }; + ranges.iter().any(|range| range.contains(ip)) +} + +/// Keeps connections off the address ranges `libp2p-ignore` names, whoever dialled: +/// hole punching and identify bring addresses there too, and a peer may dial us there. +pub(super) struct Ignoring { + pub(super) ranges: Vec<IpRange>, +} + +#[derive(Debug, thiserror::Error)] +#[error("{0} is in a range left out (libp2p-ignore)")] +pub(super) struct IgnoredAddress(Multiaddr); + +impl ::libp2p::swarm::NetworkBehaviour for Ignoring { + type ConnectionHandler = ::libp2p::swarm::dummy::ConnectionHandler; + type ToSwarm = std::convert::Infallible; + + fn handle_pending_inbound_connection( + &mut self, + _connection_id: ConnectionId, + _local_addr: &Multiaddr, + remote_addr: &Multiaddr, + ) -> Result<(), ::libp2p::swarm::ConnectionDenied> { + if ignored(&self.ranges, remote_addr) { + return Err(::libp2p::swarm::ConnectionDenied::new(IgnoredAddress( + remote_addr.clone(), + ))); + } + Ok(()) + } + + fn handle_established_inbound_connection( + &mut self, + _connection_id: ConnectionId, + _peer: PeerId, + _local_addr: &Multiaddr, + _remote_addr: &Multiaddr, + ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> { + Ok(::libp2p::swarm::dummy::ConnectionHandler) + } + + 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<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> { + if ignored(&self.ranges, addr) { + return Err(::libp2p::swarm::ConnectionDenied::new(IgnoredAddress( + addr.clone(), + ))); + } + Ok(::libp2p::swarm::dummy::ConnectionHandler) + } + + fn on_swarm_event(&mut self, _event: ::libp2p::swarm::FromSwarm) {} + + fn on_connection_handler_event( + &mut self, + _peer_id: PeerId, + _connection_id: ConnectionId, + event: ::libp2p::swarm::THandlerOutEvent<Self>, + ) { + match event {} + } + + fn poll( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll< + ::libp2p::swarm::ToSwarm<Self::ToSwarm, ::libp2p::swarm::THandlerInEvent<Self>>, + > { + std::task::Poll::Pending + } +} diff --git a/src/transport/libp2p/mod.rs b/src/transport/libp2p/mod.rs new file mode 100644 index 0000000..86e7734 --- /dev/null +++ b/src/transport/libp2p/mod.rs @@ -0,0 +1,657 @@ +//! libp2p as a transport (`--transports libp2p`, in builds with the `libp2p` feature): +//! direct connections over TCP (Noise, Yamux) and QUIC, to addresses this node knows. +//! +//! A node's transport identity is its libp2p peer ID; where it is, its listen addresses +//! (multiaddrs ending in the peer ID, `locations`). Seeds are such multiaddrs, or ones +//! without the peer ID, which a connection finds out. Messages go one per stream of the +//! `/continuum/1` protocol: their length (4 bytes, big endian), then the message as +//! `wire` encodes it. A send is done once the stream is written and closed; a connection +//! that can't be made, or a stream that can't be written, refuses it. A stream's protocol +//! goes with its first data, without waiting for the other side to agree +//! (`swarm_config`): opening a stream costs no round trip. +//! +//! Connected peers tell each other where they listen (libp2p's identify protocol), so a +//! node learns the addresses of peers that dialled it. A node publishes where it can be +//! reached (`local_addresses`): the addresses it announces (`--libp2p-announce`, e.g. a +//! public address forwarded to it), or else those it listens on, loopback left out. +//! +//! A node nobody can dial (behind NAT, or a closed port) can still be reached through a +//! relay: a node of the swarm that relays (`--libp2p-relay`), or one configured +//! (`--libp2p-relays`), on which it reserves a slot; its address through the relay +//! (`<relay>/p2p-circuit/p2p/<it>`) is published with the others. Over a relayed connection, both sides then try to +//! connect directly (libp2p's DCUtR, hole punching); the relay carries the traffic until +//! that works, or for good if it doesn't (punched again now and then, `PUNCH_RETRIES`), +//! within its limits (`RELAY_CIRCUIT_DURATION`, +//! `RELAY_CIRCUIT_BYTES`; a circuit that ends is made again on the next send). Hole +//! punching needs both sides to listen (`HolePunching`: at their LAN addresses too); +//! where only one does, the other dials it directly once it knows where (`upgrade`). +//! +//! Of several connections to one peer (both dialled at once, or AutoNAT dialled back), +//! all but one are closed a few seconds later (`DUPLICATE_GRACE`, `Book::duplicates`), +//! by the side with the smaller peer ID: it keeps the direct one with the shortest round +//! trip (libp2p's ping, on each connection), so a path over the LAN wins over one +//! through a VPN, whichever opened first. +//! +//! A relay says so where it publishes its addresses: next to each, the same address +//! ending in `/p2p-circuit` (where to reserve a slot), which other nodes find in the +//! member list. A node that needs a relay picks up to `AUTO_RELAYS` relays +//! (`relay_check`): those configured first, then those listed in the document +//! (`learn_relays`), then members that relay. It needs one when +//! it has no address others could dial, or when AutoNAT says it can't be dialled: +//! AutoNAT asks connected peers (nodes of the swarm) to dial it back. Once AutoNAT says +//! it can be dialled, it lets the relays it picked go. +//! +//! With `--libp2p-ignore`, address ranges are left out (VPNs): not published, not +//! offered for hole punching, not dialled, and connections there refused (`Ignoring`). +//! +//! With `--libp2p-upnp`, a node asks its router to forward its listen ports (UPnP), and +//! publishes the router's public address for them with its own (`Book::direct`): then +//! it can be dialled from elsewhere, whatever the other side's NAT. +//! +//! With `--libp2p-mdns`, a node finds libp2p peers on its LAN (multicast DNS). Their +//! LAN addresses are noted, and those that turn out to speak continuum (identify says +//! so) are seeds while the node joins (`discovered_seeds`): nodes on one LAN find each +//! other without configured seeds. +//! +//! It uses public libp2p infrastructure only with `--libp2p-public`: then it joins the +//! IPFS DHT (Kademlia, through the IPFS bootstrap nodes), where it finds addresses of +//! peers the member list doesn't have, and public peers that relay; those are picked +//! after members that relay. Public relays keep circuits short (by default 2 minutes and +//! 128 KiB): enough to punch holes through NATs, not to carry a swarm's traffic. Without +//! it, relays are those configured, or members that say they relay. + +use std::collections::HashMap; +use std::path::Path; +use std::str::FromStr; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::time::Duration; + +use ::libp2p::multiaddr::Protocol; +use ::libp2p::swarm::dial_opts::{DialOpts, PeerCondition}; +use ::libp2p::swarm::{ConnectionId, SwarmEvent}; +use ::libp2p::{Multiaddr, PeerId, StreamProtocol, Swarm, identity::Keypair}; +use anyhow::{Context, Result, anyhow, bail}; +use async_trait::async_trait; +use futures::{AsyncReadExt, AsyncWriteExt, StreamExt}; +use tokio::sync::{Semaphore, mpsc, oneshot, watch}; +use tracing::{debug, info, trace, warn}; + +use crate::locks::MutexExt; +use crate::transport::stats::{ + Connections, DialFailures, HolePunches, Libp2pStats, Reachability, RelaySlots, TransportStats, +}; +use crate::transport::{ + Addressed, Connection, EncodedMessage, HandlerFn, MessageHandler, ParsedMessage, Transport, + hand_on_probe, is_probe, wire, +}; + +mod behaviour; +mod book; +mod config; +mod driver; +mod errors; +mod ignore; +mod relays; + +use behaviour::*; +use book::*; +pub(crate) use config::{Libp2pConfig, check}; +use driver::*; +use errors::*; +use ignore::*; +use relays::*; + +/// The scheme of this transport's addresses (`transport::paths`). +pub(crate) const SCHEME: &str = "libp2p"; + +const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1"); +/// Sends taking this long are logged with where the time went (at debug). +const SLOW_SEND: Duration = Duration::from_millis(5); +/// The largest message read: anyone who can connect can send anything, so this bounds +/// what one stream costs. Sync messages above 16 chunks travel as blobs. +const MAX_MESSAGE: usize = 128 << 20; +/// How long a connection may take to come up. +const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// A connection nobody uses is closed after this; neighbours keep theirs busy. +const IDLE_TIMEOUT: Duration = Duration::from_secs(300); +/// Addresses of one dial tried at once (libp2p's default: 8). Hole punching (`HolePunching`) +/// dials every address both sides offer, where they listen (LAN, VPNs, public IPv6) and +/// where others saw them, often more than 8, while the other side punches towards all of +/// them at the same moment: an address dialled only after others failed (each after +/// QUIC's 5 s handshake timeout) misses its punch, and the LAN's is often among the +/// last. Dials of the member list's addresses (`ADDRESSES_PER_PEER`) are unchanged. +const DIAL_CONCURRENCY: std::num::NonZeroU8 = std::num::NonZeroU8::new(32).unwrap(); +/// How long a stream may take to open and to carry a message, before what its size adds +/// (`stream_timeout`). +const STREAM_TIMEOUT: Duration = Duration::from_secs(10); +/// The slowest a message is expected to go, in bytes per second. +const MIN_RATE: u64 = 1 << 20; +/// Inbound streams read at once. +const READS: usize = 32; +/// Addresses kept per peer, and peers with addresses at most. +const ADDRESSES_PER_PEER: usize = 8; +const MAX_PEERS: usize = 4096; + +/// A relayed connection (circuit) through this node lasts at most this long and carries +/// at most this much: generous, since the swarm's own relays may carry sync traffic for +/// good where hole punching fails (libp2p's defaults, 2 minutes and 128 KiB, are for +/// the punching only). +const RELAY_CIRCUIT_DURATION: Duration = Duration::from_secs(3600); +const RELAY_CIRCUIT_BYTES: u64 = 1 << 30; +/// Circuits through this node at once, and per peer (either end: a node behind the +/// relay is an end of every circuit to it). libp2p's rate limits on circuits and slots +/// (a burst, then one every minute or two, per peer and per IP address: every node +/// behind one NAT shares one) are left out: they're for public relays serving +/// strangers, and refused circuits used them up too. +const RELAY_CIRCUITS: usize = 1024; +const RELAY_CIRCUITS_PER_PEER: usize = 64; +/// A configured relay that couldn't be reached to find out its peer ID is tried again +/// after this. +const RELAY_RETRY: Duration = Duration::from_secs(30); +/// Relays picked among the members, at most, and how often a node checks whether it +/// needs them. +const AUTO_RELAYS: usize = 2; +const RELAY_CHECK: Duration = Duration::from_secs(30); +/// A relay that refused a slot (`relay_refused`: none left, not allowed, not a relay, +/// another peer at its address: a relay restarted without its libp2p key) isn't picked +/// again for this long. One that didn't grant it for another reason (it doesn't answer, +/// the connection closed) isn't for `RELAY_RETRY`; a slot lost is reserved again at +/// once. +const RELAY_PAUSE: Duration = Duration::from_secs(600); +/// A peer connected only through a relay is dialled directly where it can be reached +/// (`upgrade`): first at once, then after a failure no sooner than this, doubling up to +/// `RELAY_PAUSE`. +const UPGRADE_RETRY: Duration = Duration::from_secs(30); +/// A peer nobody mentioned for this long (the member list, identify, mDNS, a +/// connection) is forgotten, unless connected: mostly old peer IDs of nodes restarted +/// without their libp2p key, whose new ones the member list has since. Members are +/// mentioned every few seconds (`swarm::topology`). Checked every `PRUNE_EVERY`. +const STALE_AFTER: Duration = Duration::from_secs(600); +const PRUNE_EVERY: Duration = Duration::from_secs(60); +/// Several connections to one peer (both dialled at once, or AutoNAT dialled back to +/// see whether it can) are cut to one, the fastest (`Book::duplicates`), this long after +/// the last one opened: long enough for what it was dialled for, and for each to be +/// pinged (at once, as it opens). Looked for every `DEDUPE_EVERY`. +const DUPLICATE_GRACE: Duration = Duration::from_secs(10); +/// The same while a peer is connected only through relays: a new relayed connection is +/// a hole punch under way (`PUNCH_RETRIES`), which takes its tries (up to a QUIC +/// handshake's 5 s each, three times) on that connection. +const PUNCH_GRACE: Duration = Duration::from_secs(60); +/// After libp2p's hole punching gave up on a peer (three tries, as the relayed +/// connection opened), it tries again after these, the last one over and over: what +/// failed may have been passing (a NAT's mapping, a busy link). Each time through a new +/// connection through the relay, on which the peer starts the punch (`retry_punches`). +const PUNCH_RETRIES: [Duration; 3] = [ + Duration::from_secs(60), + Duration::from_secs(300), + Duration::from_secs(1800), +]; +const DEDUPE_EVERY: Duration = Duration::from_secs(5); + +/// How long a message of `bytes` may take on a stream. +fn stream_timeout(bytes: usize) -> Duration { + STREAM_TIMEOUT + Duration::from_millis(bytes as u64 * 1000 / MIN_RATE) +} + +pub(crate) struct Libp2pTransport { + local_id: String, + peer_id: PeerId, + control: libp2p_stream::Control, + commands: mpsc::UnboundedSender<Command>, + book: Arc<Mutex<Book>>, + incoming: tokio::sync::Mutex<Option<libp2p_stream::IncomingStreams>>, + stopped: watch::Receiver<bool>, +} + +impl Libp2pTransport { + /// Builds the swarm, listens, and runs it in a task of its own. + pub(crate) fn start(config: Libp2pConfig) -> Result<Self> { + let peer_id = config.key.public().to_peer_id(); + let features = Features::of(&config); + let mut swarm = if config.memory { + memory_swarm(config.key, features, config.ignore.clone())? + } else { + network_swarm(config.key, features, config.ignore.clone())? + }; + for address in config.listen { + swarm + .listen_on(address.clone()) + .with_context(|| format!("Listening on {address}"))?; + } + for address in &config.announce { + // Identify tells peers these. + swarm.add_external_address(address.clone()); + } + let mut control = swarm.behaviour().stream.new_control(); + let incoming = if config.continuum { + Some( + control + .accept(PROTOCOL) + .map_err(|error| anyhow!("Accepting {PROTOCOL}: {error}"))?, + ) + } else { + None + }; + let (commands, receiver) = mpsc::unbounded_channel(); + let mut book = Book { + announced: config.announce, + ignored: config.ignore.clone(), + serving: config.relay, + ..Book::default() + }; + for relay in config.relays { + if let Some(address) = book.name_relay(relay, Named::Configured, peer_id) { + let _ = commands.send(Command::FindRelay { address }); + } + } + let book = Arc::new(Mutex::new(book)); + let (stopping, stopped) = watch::channel(false); + tokio::spawn(run( + swarm, + receiver, + commands.clone(), + Arc::clone(&book), + Relaying { + serving: config.relay, + public: config.public, + local: peer_id, + }, + stopping, + )); + info!(peer_id = %peer_id, "Transport started"); + Ok(Self { + local_id: peer_id.to_string(), + peer_id, + control, + commands, + book, + incoming: tokio::sync::Mutex::new(incoming), + stopped, + }) + } + + /// Connected to `peer`, if it can be had in time. + async fn connect(&self, peer: PeerId) -> Result<()> { + let addresses = { + let book = lock(&self.book); + let mut known = book + .addresses + .get(&peer) + .map(|known| dialable(known, self.peer_id)) + .unwrap_or_default(); + known.retain(|address| !ignored(&book.ignored, address)); + if known.is_empty() { + book.through_relays(peer, self.peer_id) + } else { + known + } + }; + let (done, connected) = oneshot::channel(); + self.commands + .send(Command::Connect { + peer, + addresses, + done, + }) + .map_err(|_| anyhow!("the libp2p swarm stopped"))?; + match tokio::time::timeout(CONNECT_TIMEOUT, connected).await { + Ok(Ok(Ok(()))) => Ok(()), + Ok(Ok(Err(error))) => bail!("couldn't connect to {peer}: {error}"), + Ok(Err(_)) => bail!("the libp2p swarm stopped"), + Err(_) => bail!("connecting to {peer} took longer than {CONNECT_TIMEOUT:?}"), + } + } +} + +/// One message from an inbound stream. +async fn read_message(mut stream: ::libp2p::Stream) -> Result<Vec<u8>> { + let mut length = [0u8; 4]; + tokio::time::timeout(STREAM_TIMEOUT, stream.read_exact(&mut length)) + .await + .map_err(|_| anyhow!("no length within {STREAM_TIMEOUT:?}"))??; + let length = u32::from_be_bytes(length) as usize; + if length > MAX_MESSAGE { + bail!("a message of {length} bytes; at most {MAX_MESSAGE} are read"); + } + let mut body = vec![0u8; length]; + tokio::time::timeout(stream_timeout(length), stream.read_exact(&mut body)) + .await + .map_err(|_| { + anyhow!( + "{length} bytes took longer than {:?}", + stream_timeout(length) + ) + })??; + Ok(body) +} + +#[async_trait] +impl Transport for Libp2pTransport { + fn local_id(&self) -> &str { + &self.local_id + } + + fn is_peer_id(&self, id: &str) -> bool { + PeerId::from_str(id).is_ok() + } + + /// The longest a message may take on a stream. + fn delivery_limit(&self) -> Duration { + stream_timeout(MAX_MESSAGE) + } + + /// Where we can be reached, each with our peer ID: the announced addresses, or else + /// those we listen on (`published`), and those through relays. A relay adds each + /// direct one again, ending in `/p2p-circuit`: where to reserve a slot. + fn local_addresses(&self) -> Vec<String> { + let book = lock(&self.book); + let direct: Vec<Multiaddr> = book + .direct() + .into_iter() + .filter_map(|address| address.with_p2p(self.peer_id).ok()) + .collect(); + let relayed = book + .listening + .iter() + .filter(|address| is_relayed(address)) + .filter_map(|address| address.clone().with_p2p(self.peer_id).ok()); + let offers: Vec<Multiaddr> = if book.serving { + direct + .iter() + .map(|address| address.clone().with(Protocol::P2pCircuit)) + .collect() + } else { + Vec::new() + }; + direct + .into_iter() + .chain(relayed) + .chain(offers) + .map(|address| address.to_string()) + .collect() + } + + /// A peer ID, or a multiaddr ending in one (noted as where that peer is). + fn locate(&self, target: &str) -> Option<String> { + if let Ok(peer) = PeerId::from_str(target) { + return Some(peer.to_string()); + } + let mut address = Multiaddr::from_str(target).ok()?; + // A relay saying where to reserve a slot: `<address>/p2p/<relay>/p2p-circuit`. + if let Some(Protocol::P2pCircuit) = address.iter().last() { + address.pop(); + let relay = split_peer(&address)?; + if relay != self.peer_id { + let new = { + let mut book = lock(&self.book); + let new = book.relays.get(&relay) != Some(&address); + if new && (book.relays.len() < MAX_PEERS || book.relays.contains_key(&relay)) { + book.relays.insert(relay, address.clone()); + } + book.remember(relay, address); + new + }; + if new { + let _ = self.commands.send(Command::CheckRelays); + } + } + return Some(relay.to_string()); + } + let peer = split_peer(&address)?; + address.pop(); + // A peer ID only says nothing about where. + if peer != self.peer_id && !address.is_empty() { + lock(&self.book).remember(peer, address); + } + Some(peer.to_string()) + } + + /// Multiaddrs, with or without the relay's peer ID; others are ignored. + fn learn_relays(&self, targets: &[String]) { + let listed = targets + .iter() + .filter_map(|target| Multiaddr::from_str(target).ok()) + .collect(); + let Some(found) = lock(&self.book).learn_listed(listed, self.peer_id) else { + return; + }; + for address in found { + let _ = self.commands.send(Command::FindRelay { address }); + } + let _ = self.commands.send(Command::CheckRelays); + } + + fn connection(&self, target: &str) -> Option<Connection> { + let peer = PeerId::from_str(target).ok()?; + lock(&self.book).connection(&peer) + } + + fn punching(&self, target: &str) -> Option<crate::transport::Punching> { + let peer = PeerId::from_str(target).ok()?; + let (failed, next) = *lock(&self.book).punching.get(&peer)?; + Some(crate::transport::Punching { + failed, + next_in_secs: next + .saturating_duration_since(tokio::time::Instant::now()) + .as_secs(), + }) + } + + fn location(&self, target: &str) -> Option<String> { + let peer = PeerId::from_str(target).ok()?; + let address = lock(&self.book).location(&peer)?.clone(); + Some( + address + .with_p2p(peer) + .unwrap_or_else(|address| address) + .to_string(), + ) + } + + fn stats(&self) -> TransportStats { + TransportStats { + libp2p: Some(lock(&self.book).stats()), + ..TransportStats::default() + } + } + + /// Nodes found on the LAN (mDNS) that speak continuum. + fn discovered_seeds(&self) -> Vec<String> { + std::mem::take(&mut lock(&self.book).discovered) + .iter() + .map(Multiaddr::to_string) + .collect() + } + + /// Where `target` is, each address with its peer ID. + fn locations(&self, target: &str) -> Vec<String> { + let Ok(peer) = PeerId::from_str(target) else { + return Vec::new(); + }; + lock(&self.book) + .addresses + .get(&peer) + .into_iter() + .flatten() + .filter_map(|address| address.clone().with_p2p(peer).ok()) + .map(|address| address.to_string()) + .collect() + } + + /// A multiaddr ending in a peer ID is that peer; one without is connected to, to + /// find out. A bare peer ID resolves if this transport knows where it is. + async fn resolve_seed(&self, seed: &str) -> Result<String> { + let address = match PeerId::from_str(seed) { + Ok(peer) => Multiaddr::empty().with(Protocol::P2p(peer)), + Err(_) => { + Multiaddr::from_str(seed).with_context(|| format!("{seed} is no multiaddr"))? + } + }; + // A peer ID only (`/p2p/<peer ID>`): reached where it's known, or through the + // relays we know. + if let [Protocol::P2p(peer)] = address.iter().collect::<Vec<_>>()[..] { + let book = lock(&self.book); + if book.addresses.contains_key(&peer) + || !book.through_relays(peer, self.peer_id).is_empty() + { + return Ok(peer.to_string()); + } + bail!( + "no address known for libp2p peer {peer}, and no relay to reach it through; \ + give a multiaddr, or a relay (libp2p-relays)" + ); + } + if let Some(peer) = self.locate(seed) { + return Ok(peer); + } + let (done, identified) = oneshot::channel(); + self.commands + .send(Command::Identify { address, done }) + .map_err(|_| anyhow!("the libp2p swarm stopped"))?; + match tokio::time::timeout(CONNECT_TIMEOUT, identified).await { + Ok(Ok(Ok(peer))) => Ok(peer.to_string()), + Ok(Ok(Err(error))) => bail!("couldn't connect to {seed}: {error}"), + Ok(Err(_)) => bail!("the libp2p swarm stopped"), + Err(_) => bail!("connecting to {seed} took longer than {CONNECT_TIMEOUT:?}"), + } + } + + fn encode(&self, peer: &str, message: &ParsedMessage) -> Result<EncodedMessage> { + let body = wire::encode(message)?; + Ok(EncodedMessage { + peer: peer.to_owned(), + address: peer.to_owned(), + payload_len: body.len(), + body: body.into(), + kind: message.kind(), + message: None, + }) + } + + async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { + let peer = PeerId::from_str(&message.peer) + .with_context(|| format!("{} is no libp2p peer ID", message.peer))?; + let started = std::time::Instant::now(); + self.connect(peer).await?; + // The protocol goes with the first data (`swarm_config`), so a peer that doesn't + // speak it seems to take the message: once identify has told, say so. + if let Some(relays) = lock(&self.book).not_nodes.get(&peer).copied() { + if relays { + bail!( + "{peer} is a relay, not a node: give a node behind it as the seed \ + (<relay>/p2p-circuit/p2p/<peer ID>, or /p2p/<peer ID> with libp2p-relays)" + ); + } + bail!("{peer} doesn't speak continuum"); + } + let connected = started.elapsed(); + let mut control = self.control.clone(); + let body = message.body.clone(); + let write = async move { + let mut stream = control.open_stream(peer, PROTOCOL).await?; + let opened = started.elapsed(); + stream.write_all(&(body.len() as u32).to_be_bytes()).await?; + stream.write_all(&body).await?; + stream.close().await?; + // Kept until the peer has closed its side too. Over QUIC, dropping a stream + // stops its reading half (STOP_SENDING), and the peer's answer to the + // protocol (sent with the first data, `swarm_config`) then fails, and with it + // the stream, before the peer has read the message: identify, which reads + // its answers, works, while every message is lost. Yamux (TCP) doesn't mind. + // Off the send, which is done. + tokio::spawn(async move { + let mut rest = [0u8; 64]; + let _ = tokio::time::timeout(STREAM_TIMEOUT, async { + while let Ok(read) = stream.read(&mut rest).await { + if read == 0 { + break; + } + } + }) + .await; + }); + Ok::<_, anyhow::Error>(opened) + }; + let timeout = stream_timeout(message.body.len()); + let opened = tokio::time::timeout(timeout, write) + .await + .map_err(|_| anyhow!("sending to {peer} took longer than {timeout:?}"))? + .with_context(|| format!("Sending to {peer}"))?; + let took = started.elapsed(); + if took >= SLOW_SEND { + // Where a send's time goes: the swarm task (connecting), the stream's protocol + // negotiation (opening), or the connection (writing). + debug!( + peer = %Addressed(SCHEME, &peer), + kind = ?message.kind, + bytes = message.body.len(), + connect_ms = connected.as_secs_f64() * 1000.0, + open_ms = (opened - connected).as_secs_f64() * 1000.0, + write_ms = (took - opened).as_secs_f64() * 1000.0, + "Slow send" + ); + } + Ok(()) + } + + /// Reads inbound streams (`READS` at once) and handles their messages one at a + /// time, in the order they were read; PINGs and PONGs at once (`is_probe`). + async fn start_consumer(&self, handler: MessageHandler) -> Result<()> { + let Some(mut incoming) = self.incoming.lock().await.take() else { + bail!("libp2p consumer already started, or this node only relays"); + }; + let handler: Arc<HandlerFn> = Arc::from(handler); + let (sender, mut received) = mpsc::channel::<(PeerId, ParsedMessage, usize)>(READS); + let reads = Arc::new(Semaphore::new(READS)); + let probes = Arc::clone(&handler); + tokio::spawn(async move { + while let Some((peer, stream)) = incoming.next().await { + let Ok(permit) = Arc::clone(&reads).acquire_owned().await else { + break; + }; + let sender = sender.clone(); + let probes = Arc::clone(&probes); + tokio::spawn(async move { + match read_message(stream).await { + Ok(body) => match wire::decode(&body) { + Ok(message) if is_probe(&message) => { + hand_on_probe( + SCHEME, + &probes, + peer.to_string(), + message, + body.len(), + ); + } + Ok(message) => { + let _ = sender.send((peer, message, body.len())).await; + } + Err(error) => { + debug!(peer = %Addressed(SCHEME, &peer), error = %error, "Dropping a message that doesn't decode"); + } + }, + Err(error) => { + debug!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Dropping an unreadable stream"); + } + } + drop(permit); + }); + } + }); + while let Some((peer, message, payload_len)) = received.recv().await { + if let Err(error) = handler(peer.to_string(), message, payload_len).await { + warn!(peer = %Addressed(SCHEME, &peer), error = %format!("{error:#}"), "Failed to process inbound message"); + } + } + Ok(()) + } + + async fn wait_for_exit(&self) -> std::io::Result<std::process::ExitStatus> { + let mut stopped = self.stopped.clone(); + let _ = stopped.wait_for(|stopped| *stopped).await; + Err(std::io::Error::other("the libp2p swarm stopped")) + } +} + +#[cfg(test)] +mod tests; diff --git a/src/transport/libp2p/relays.rs b/src/transport/libp2p/relays.rs new file mode 100644 index 0000000..3463408 --- /dev/null +++ b/src/transport/libp2p/relays.rs @@ -0,0 +1,234 @@ +//! Reaching peers past NATs: picking relays and keeping slots on them, upgrading +//! relayed connections to direct ones, retrying hole punches, and finding peers on the +//! DHT. + +use super::*; + +/// How a node relates to relays. +pub(super) struct Relaying { + /// It relays for others. + pub(super) serving: bool, + /// Public relays may be picked, after members. + pub(super) public: bool, + pub(super) local: PeerId, +} + +/// The relays to pick among `offers`, if this node needs relays, keeping `using`: a +/// node needs them when it has no address others could dial (`direct`), or AutoNAT says +/// it can't be dialled; up to `AUTO_RELAYS`, in an order of its own (the lowest hash of +/// both peer IDs), so that not every node picks the same. +pub(super) fn relay_check( + local: PeerId, + direct: bool, + nat: &::libp2p::autonat::NatStatus, + offers: &HashMap<PeerId, Multiaddr>, + using: &[PeerId], +) -> Vec<(PeerId, Multiaddr)> { + use ::libp2p::autonat::NatStatus; + let needed = match nat { + NatStatus::Private => true, + NatStatus::Public(_) => false, + NatStatus::Unknown => !direct, + }; + if !needed || using.len() >= AUTO_RELAYS { + return Vec::new(); + } + let score = |relay: &PeerId| { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + (local, relay).hash(&mut hasher); + hasher.finish() + }; + let mut candidates: Vec<(PeerId, Multiaddr)> = offers + .iter() + .filter(|(relay, _)| **relay != local && !using.contains(relay)) + .map(|(relay, address)| (*relay, address.clone())) + .collect(); + candidates.sort_by_key(|(relay, _)| score(relay)); + candidates.truncate(AUTO_RELAYS - using.len()); + candidates +} + +/// Reserves slots on the relays `relay_check` picks (configured ones first, then listed +/// ones, members, public ones), and lets those picked go once AutoNAT says this node can +/// be dialled. +pub(super) fn pick_relays( + swarm: &mut Swarm<Behaviour>, + book: &Mutex<Book>, + relaying: &Relaying, + nat: &::libp2p::autonat::NatStatus, + picked: &mut HashMap<::libp2p::core::transport::ListenerId, PeerId>, +) { + if nat.is_public() && !picked.is_empty() { + info!("Reachable directly (AutoNAT); letting the relays picked go"); + let mut book = lock(book); + for (listener, relay) in std::mem::take(picked) { + swarm.remove_listener(listener); + book.slots.remove(&relay); + } + return; + } + let (direct, configured, listed, members, public) = { + let mut book = lock(book); + let now = tokio::time::Instant::now(); + book.failed_relays.retain(|_, until| now < *until); + let failed = book.failed_relays.clone(); + let usable = |offers: &HashMap<PeerId, Multiaddr>| -> HashMap<PeerId, Multiaddr> { + offers + .iter() + .filter(|(relay, _)| !failed.contains_key(relay)) + .map(|(relay, address)| (*relay, address.clone())) + .collect() + }; + let public = if relaying.public { + usable(&book.public_relays) + } else { + HashMap::new() + }; + ( + !book.direct().is_empty(), + usable(&book.named(Named::Configured)), + usable(&book.named(Named::Listed)), + usable(&book.relays), + public, + ) + }; + let mut using: Vec<PeerId> = picked.values().copied().collect(); + // Each kind for what the ones before left. + for (kind, offers) in [ + (Named::Configured.as_str(), configured), + (Named::Listed.as_str(), listed), + ("member", members), + ("public", public), + ] { + for (relay, address) in relay_check(relaying.local, direct, nat, &offers, &using) { + using.push(relay); + match swarm.listen_on(address.clone().with(Protocol::P2pCircuit)) { + Ok(listener) => { + info!(relay = %Addressed(SCHEME, &relay), address = %address, kind, "Picked a relay"); + picked.insert(listener, relay); + } + Err(error) => { + debug!(relay = %Addressed(SCHEME, &relay), kind, error = %error, "Couldn't use a relay"); + } + } + } + } +} + +/// Dials `peer` directly if it's connected only through a relay and is known elsewhere +/// too (the member list, identify, mDNS), since the first connection, made through a +/// relay (a relayed seed), would otherwise be kept for good: a peer connected isn't +/// dialled again, and hole punching needs both sides to listen. The direct connection, +/// once open, wins over the relayed one (`Book::duplicates`). +pub(super) fn upgrade( + swarm: &mut Swarm<Behaviour>, + book: &Mutex<Book>, + peer: PeerId, + local: PeerId, +) { + let direct: Vec<Multiaddr> = { + let mut book = lock(book); + if book.connection(&peer) != Some(Connection::Relayed) { + return; + } + let direct: Vec<Multiaddr> = book + .addresses + .get(&peer) + .map(|known| dialable(known, local)) + .unwrap_or_default() + .into_iter() + .filter(|address| !is_relayed(address) && !ignored(&book.ignored, address)) + .collect(); + if direct.is_empty() { + return; + } + let now = tokio::time::Instant::now(); + let (due, wait) = book + .upgrades + .get(&peer) + .copied() + .unwrap_or((now, UPGRADE_RETRY)); + if now < due { + return; + } + book.upgrades + .insert(peer, (now + wait, (wait * 2).min(RELAY_PAUSE))); + direct + }; + debug!(peer = %Addressed(SCHEME, &peer), addresses = ?direct, "Connected only through a relay; dialling directly"); + let _ = swarm.dial( + DialOpts::peer_id(peer) + .addresses(direct) + .condition(PeerCondition::NotDialing) + .build(), + ); +} + +/// Tries hole punching again with the peers it failed with that are due (`punching`). +/// libp2p punches as a relayed connection opens, started by the side that was dialled +/// through the relay; that side is where a punch's failure is told, so it is this one. +/// A new relayed connection starts a new punch: we dial the peer through a relay if it +/// has a slot on one; else we close our relayed connections to it, and its next message +/// (a probe, within a minute) dials us through ours again. Peers connected directly by +/// now are dropped; those not connected now are kept, failures and all. +pub(super) fn retry_punches(swarm: &mut Swarm<Behaviour>, book: &Mutex<Book>, local: PeerId) { + let now = tokio::time::Instant::now(); + let mut close: Vec<ConnectionId> = Vec::new(); + let due: Vec<(PeerId, usize, Vec<Multiaddr>)> = { + let mut book = lock(book); + let book = &mut *book; + let mut due = Vec::new(); + book.punching.retain(|peer, (failed, next)| { + match book_connection(&book.connections, peer) { + Some(Connection::Direct) => return false, + // Kept, failures and all: one we closed to have it dial anew comes back. + None => return true, + Some(Connection::Relayed) if now < *next => return true, + Some(Connection::Relayed) => {} + } + let through: Vec<Multiaddr> = book + .addresses + .get(peer) + .map(|known| dialable(known, local)) + .unwrap_or_default() + .into_iter() + .filter(is_relayed) + .collect(); + // Until its punch fails again (or the peer tries from its side). + *next = now + PUNCH_RETRIES[(*failed).min(PUNCH_RETRIES.len() - 1)]; + if through.is_empty() { + close.extend( + book.connections + .iter() + .filter(|(_, (connected, relayed, _))| connected == peer && *relayed) + .map(|(id, _)| *id), + ); + } else { + due.push((*peer, *failed, through)); + } + true + }); + due + }; + for connection in close { + debug!(connection = %connection, "Hole punching again: closing a relayed connection, for the peer to dial anew"); + swarm.close_connection(connection); + } + for (peer, failed, through) in due { + debug!(peer = %Addressed(SCHEME, &peer), failed, "Hole punching again, through a new relayed connection"); + let _ = swarm.dial( + DialOpts::peer_id(peer) + .addresses(through) + .condition(PeerCondition::Always) + .build(), + ); + } +} + +/// Asks the DHT (with `--libp2p-public`) where `peer` is, for the next attempt. +pub(super) fn find_on_dht(swarm: &mut Swarm<Behaviour>, peer: PeerId) { + if let Some(dht) = swarm.behaviour_mut().kad.as_mut() { + dht.get_closest_peers(peer); + } +} diff --git a/src/transport/libp2p/tests.rs b/src/transport/libp2p/tests.rs new file mode 100644 index 0000000..f02668a --- /dev/null +++ b/src/transport/libp2p/tests.rs @@ -0,0 +1,779 @@ +use std::sync::atomic::{AtomicU64, Ordering}; + +use super::*; +use crate::document::replica::Epoch; +use crate::transport::{HeadsSummary, Status}; + +/// Memory ports are process-wide. +static NEXT_PORT: AtomicU64 = AtomicU64::new(40_000); + +/// A node on the in-memory transport, listening on a port of its own (or nowhere), +/// and its address with its peer ID (or just its peer ID). +fn memory_node_with( + listen: bool, + relays: Vec<Multiaddr>, + relay: bool, +) -> (Libp2pTransport, Multiaddr) { + let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed); + let key = Keypair::generate_ed25519(); + let address: Multiaddr = format!("/memory/{port}").parse().unwrap(); + let peer = key.public().to_peer_id(); + let transport = Libp2pTransport::start(Libp2pConfig { + key, + listen: if listen { + vec![address.clone()] + } else { + Vec::new() + }, + announce: Vec::new(), + relays, + relay, + continuum: true, + mdns: false, + upnp: false, + ignore: Vec::new(), + public: false, + memory: true, + }) + .unwrap(); + let address = if listen { + address.with_p2p(peer).unwrap() + } else { + Multiaddr::empty().with(Protocol::P2p(peer)) + }; + (transport, address) +} + +fn memory_node() -> (Libp2pTransport, Multiaddr) { + memory_node_with(true, Vec::new(), false) +} + +fn status() -> ParsedMessage { + ParsedMessage::Status(Status { + heads: HeadsSummary::of(Vec::new()), + epoch: Epoch::GENESIS, + epoch_seq: 3, + ready: true, + syncs_with_you: false, + leaving: false, + }) +} + +/// Starts `transport`'s consumer, forwarding what it receives. +fn received(transport: Arc<Libp2pTransport>) -> mpsc::UnboundedReceiver<(String, ParsedMessage)> { + let (sender, receiver) = mpsc::unbounded_channel(); + tokio::spawn(async move { + transport + .start_consumer(Box::new(move |from, message, _| { + let sender = sender.clone(); + Box::pin(async move { + let _ = sender.send((from, message)); + Ok(()) + }) + })) + .await + }); + receiver +} + +#[tokio::test] +async fn messages_both_ways_over_one_connection() { + let (a, a_address) = memory_node(); + let (b, _) = memory_node(); + let (a, b) = (Arc::new(a), Arc::new(b)); + let mut at_a = received(Arc::clone(&a)); + let mut at_b = received(Arc::clone(&b)); + + assert_eq!( + b.resolve_seed(&a_address.to_string()).await.unwrap(), + a.local_id() + ); + b.send_message(a.local_id(), &status()).await.unwrap(); + let (from, message) = at_a.recv().await.unwrap(); + assert_eq!(from, b.local_id()); + assert!(matches!(message, ParsedMessage::Status(status) if status.epoch_seq == 3)); + + // a knows no address of b, but b's connection is there. + a.send_message(b.local_id(), &status()).await.unwrap(); + let (from, _) = at_b.recv().await.unwrap(); + assert_eq!(from, a.local_id()); + + assert_eq!(a.connection(b.local_id()), Some(Connection::Direct)); + assert_eq!(b.connection(a.local_id()), Some(Connection::Direct)); + let stats = a.stats().libp2p.unwrap(); + assert!(stats.connections.direct >= 1); + assert_eq!(stats.connections.relayed, 0); +} + +#[tokio::test] +async fn a_seed_without_a_peer_id_is_connected_to_to_find_out() { + let (a, a_address) = memory_node(); + let (b, _) = memory_node(); + let mut bare = a_address.clone(); + bare.pop(); + assert_eq!( + b.resolve_seed(&bare.to_string()).await.unwrap(), + a.local_id() + ); + // And b now knows where a is. + assert_eq!(b.locations(a.local_id()), [a_address.to_string()]); +} + +#[tokio::test] +async fn listen_addresses_carry_the_peer_id() { + let (a, a_address) = memory_node(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while a.local_addresses().is_empty() && tokio::time::Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(a.local_addresses(), [a_address.to_string()]); + assert_eq!( + a.locate(&a_address.to_string()).as_deref(), + Some(a.local_id()) + ); +} + +/// Waits up to 5 s for `condition`. +async fn eventually(mut condition: impl FnMut() -> bool) -> bool { + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + while !condition() { + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + true +} + +/// A node learns where a peer that dialled it listens (identify), and can then +/// pass it on. +#[tokio::test] +async fn a_dialled_node_learns_where_the_dialler_listens() { + let (a, a_address) = memory_node(); + let (b, b_address) = memory_node(); + let (a, b) = (Arc::new(a), Arc::new(b)); + let _at_a = received(Arc::clone(&a)); + b.resolve_seed(&a_address.to_string()).await.unwrap(); + b.send_message(a.local_id(), &status()).await.unwrap(); + let b_id = b.local_id().to_owned(); + assert!( + eventually(|| a.locations(&b_id) == [b_address.to_string()]).await, + "a knows {:?} of b", + a.locations(&b_id) + ); +} + +#[tokio::test] +async fn announced_addresses_are_published_instead() { + let key = Keypair::generate_ed25519(); + let peer = key.public().to_peer_id(); + let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed); + let announced: Multiaddr = "/ip4/203.0.113.7/tcp/9661".parse().unwrap(); + let transport = Libp2pTransport::start(Libp2pConfig { + key, + listen: vec![format!("/memory/{port}").parse().unwrap()], + announce: vec![announced.clone()], + relays: Vec::new(), + relay: false, + continuum: true, + mdns: false, + upnp: false, + ignore: Vec::new(), + public: false, + memory: true, + }) + .unwrap(); + assert_eq!( + transport.local_addresses(), + [announced.with_p2p(peer).unwrap().to_string()] + ); +} + +#[test] +fn address_ranges() { + let range = |text: &str| text.parse::<IpRange>().unwrap(); + let address = |text: &str| text.parse::<Multiaddr>().unwrap(); + let ranges = [range("10.244.0.0/16"), range("42::/16"), range("192.0.2.7")]; + for left_out in [ + "/ip4/10.244.3.2/udp/9661/quic-v1", + "/ip6/42:1:a46e:be4a:cb58:e460:bf0:142c/tcp/9661", + "/ip4/192.0.2.7/tcp/1", + ] { + assert!(ignored(&ranges, &address(left_out)), "{left_out}"); + } + for kept in [ + "/ip4/10.23.1.144/udp/9661/quic-v1", + "/ip4/10.245.0.1/tcp/1", + "/ip6/2a0a:4cc0:c1:5eb0::1/udp/9661/quic-v1", + "/ip4/192.0.2.8/tcp/1", + "/dns4/example.org/tcp/1", + ] { + assert!(!ignored(&ranges, &address(kept)), "{kept}"); + } + // Through a relay: judged by the connection to the relay, not here. + let relayed = format!("/ip4/10.244.0.1/tcp/1/p2p/{}/p2p-circuit", peer()); + assert!(!ignored(&ranges, &address(&relayed))); + assert!(range("0.0.0.0/0").contains("203.0.113.1".parse().unwrap())); + assert!(!range("0.0.0.0/0").contains("::1".parse().unwrap())); + for wrong in ["10.244.0.0/33", "42::/129", "nowhere", "10.0.0.0/x"] { + assert!(wrong.parse::<IpRange>().is_err(), "{wrong}"); + } +} + +/// Ranges left out aren't published; announced addresses are, whatever they are. +#[test] +fn ignored_addresses_are_not_published() { + let parse = |text: &str| text.parse::<Multiaddr>().unwrap(); + let lan = parse("/ip4/10.23.1.5/udp/9663/quic-v1"); + let vpn = parse("/ip4/10.244.3.2/udp/9663/quic-v1"); + let mut book = Book { + listening: vec![lan.clone(), vpn.clone()], + ignored: vec!["10.244.0.0/16".parse().unwrap()], + ..Book::default() + }; + assert_eq!(book.direct(), [lan]); + book.announced = vec![vpn.clone()]; + assert_eq!(book.direct(), [vpn]); +} + +/// A port forwarded by hand: once AutoNAT found the node dialable at the router's +/// address, that is published with the listen addresses; announced ones still win. +#[test] +fn the_address_autonat_confirmed_is_published() { + let parse = |text: &str| text.parse::<Multiaddr>().unwrap(); + let lan = parse("/ip4/10.23.1.5/udp/9663/quic-v1"); + let public = parse("/ip4/91.65.180.254/udp/9663/quic-v1"); + let mut book = Book { + listening: vec![lan.clone()], + ..Book::default() + }; + assert_eq!(book.direct(), [lan.clone()]); + book.public_address = Some(public.clone().with_p2p(peer()).unwrap()); + assert_eq!(book.direct(), [public.clone(), lan.clone()]); + let announced = parse("/dns4/example.org/udp/9663/quic-v1"); + book.announced = vec![announced.clone()]; + assert_eq!(book.direct(), [announced]); +} + +#[test] +fn local_addresses_are_left_out_unless_there_are_only_local_ones() { + let parse = |text: &str| text.parse::<Multiaddr>().unwrap(); + let loopback = parse("/ip4/127.0.0.1/tcp/1"); + let link = parse("/ip6/fe80::1/tcp/1"); + let lan = parse("/ip4/192.168.1.2/tcp/1"); + let wireguard = parse("/ip6/fd00::2/tcp/1"); + assert_eq!( + published(&[ + loopback.clone(), + link.clone(), + lan.clone(), + wireguard.clone() + ]), + [lan, wireguard] + ); + assert_eq!( + published(&[loopback.clone(), link.clone()]), + [loopback, link] + ); +} + +/// b listens nowhere but through the relay r; a reaches it there. +#[tokio::test] +async fn a_node_nobody_can_dial_is_reached_through_a_relay() { + let (_r, r_address) = memory_node_with(true, Vec::new(), true); + let (b, _) = memory_node_with(false, vec![r_address.clone()], false); + let b = Arc::new(b); + let mut at_b = received(Arc::clone(&b)); + assert!( + eventually(|| !b.local_addresses().is_empty()).await, + "b got no slot on the relay" + ); + let relayed = b.local_addresses()[0].clone(); + assert!( + relayed.starts_with(&format!("{r_address}/p2p-circuit/")), + "{relayed}" + ); + + assert!(eventually(|| b.stats().libp2p.unwrap().relay_slots.held == 1).await); + let slots = b.stats().libp2p.unwrap().relay_slots; + assert_eq!((slots.granted, slots.not_granted), (1, 0)); + + let (a, _) = memory_node(); + assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); + a.send_message(b.local_id(), &status()).await.unwrap(); + let (from, _) = at_b.recv().await.unwrap(); + assert_eq!(from, a.local_id()); + assert!(a.connection(b.local_id()).is_some()); +} + +/// A seed that's a peer ID only is reached through the relays the node knows. +#[tokio::test] +async fn a_peer_id_seed_is_reached_through_a_relay() { + let (_r, r_address) = memory_node_with(true, Vec::new(), true); + let (b, _) = memory_node_with(false, vec![r_address.clone()], false); + let b = Arc::new(b); + let mut at_b = received(Arc::clone(&b)); + assert!( + eventually(|| !b.local_addresses().is_empty()).await, + "b got no slot on the relay" + ); + let (a, _) = memory_node_with(true, vec![r_address.clone()], false); + let seed = format!("/p2p/{}", b.local_id()); + assert_eq!(a.resolve_seed(&seed).await.unwrap(), b.local_id()); + assert_eq!(a.resolve_seed(b.local_id()).await.unwrap(), b.local_id()); + // Nothing to dial is remembered. + assert!(!lock(&a.book).addresses.contains_key(&b.peer_id)); + a.send_message(b.local_id(), &status()).await.unwrap(); + let (from, _) = at_b.recv().await.unwrap(); + assert_eq!(from, a.local_id()); + + // Without a relay, there's nowhere to look. + let (c, _) = memory_node(); + assert!(c.resolve_seed(&seed).await.is_err()); +} + +/// b listens nowhere; a, which can be dialled, reaches it through the relay r: b +/// then dials a directly (identify told it where), and both keep that. +#[tokio::test] +async fn a_relayed_connection_is_upgraded_where_one_side_can_be_dialled() { + let (_r, r_address) = memory_node_with(true, Vec::new(), true); + let (b, _) = memory_node_with(false, vec![r_address.clone()], false); + let b = Arc::new(b); + let mut at_b = received(Arc::clone(&b)); + assert!( + eventually(|| !b.local_addresses().is_empty()).await, + "b got no slot on the relay" + ); + let (a, _) = memory_node(); + let relayed = b.local_addresses()[0].clone(); + assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); + a.send_message(b.local_id(), &status()).await.unwrap(); + at_b.recv().await.unwrap(); + assert!( + eventually(|| b.connection(a.local_id()) == Some(Connection::Direct)).await, + "b didn't dial a directly" + ); + assert_eq!(a.connection(b.local_id()), Some(Connection::Direct)); + // The relayed one goes (`DUPLICATE_GRACE`); the direct one stays. + let deadline = tokio::time::Instant::now() + DUPLICATE_GRACE * 2; + while a.stats().libp2p.unwrap().connections.relayed > 0 { + assert!( + tokio::time::Instant::now() < deadline, + "the relayed connection stayed" + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + b.send_message(a.local_id(), &status()).await.unwrap(); + assert_eq!(a.connection(b.local_id()), Some(Connection::Direct)); +} + +/// A relay that is no node (`--relay-only`) relays, and takes no messages itself. +#[tokio::test] +async fn a_relay_that_is_no_node() { + let port = NEXT_PORT.fetch_add(1, Ordering::Relaxed); + let key = Keypair::generate_ed25519(); + let r_peer = key.public().to_peer_id(); + let r_listen: Multiaddr = format!("/memory/{port}").parse().unwrap(); + let r = Libp2pTransport::start(Libp2pConfig { + key, + listen: vec![r_listen.clone()], + announce: Vec::new(), + relays: Vec::new(), + relay: true, + continuum: false, + mdns: false, + upnp: false, + ignore: Vec::new(), + public: false, + memory: true, + }) + .unwrap(); + let r_address = r_listen.with_p2p(r_peer).unwrap(); + let (b, _) = memory_node_with(false, vec![r_address.clone()], false); + let b = Arc::new(b); + let mut at_b = received(Arc::clone(&b)); + assert!( + eventually(|| !b.local_addresses().is_empty()).await, + "b got no slot on the relay" + ); + let (a, _) = memory_node(); + let relayed = b.local_addresses()[0].clone(); + assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); + a.send_message(b.local_id(), &status()).await.unwrap(); + let (from, _) = at_b.recv().await.unwrap(); + assert_eq!(from, a.local_id()); + + // The relay itself doesn't speak continuum: it takes no messages, and a send to + // it is refused once identify has told (the relay's own refusal comes too late: + // the protocol goes with the first data, `swarm_config`), saying what to give + // as a seed instead. + assert_eq!( + a.resolve_seed(&r_address.to_string()).await.unwrap(), + r.local_id() + ); + assert!( + eventually(|| lock(&a.book).not_nodes.get(&r.peer_id) == Some(&true)).await, + "identify didn't tell" + ); + let refused = a.send_message(r.local_id(), &status()).await.unwrap_err(); + assert!( + refused.to_string().contains("is a relay, not a node"), + "{refused:#}" + ); + assert!( + r.start_consumer(Box::new(|_, _, _| Box::pin(async { Ok(()) }))) + .await + .is_err() + ); +} + +/// A relay configured without its peer ID is found out, and used. +#[tokio::test] +async fn a_relay_configured_without_its_peer_id() { + let (_r, r_address) = memory_node_with(true, Vec::new(), true); + let mut bare = r_address.clone(); + bare.pop(); + let (b, _) = memory_node_with(false, vec![bare], false); + assert!( + eventually(|| !b.local_addresses().is_empty()).await, + "b got no slot on the relay" + ); + assert!(b.local_addresses()[0].starts_with(&format!("{r_address}/p2p-circuit/"))); +} + +fn peer() -> PeerId { + Keypair::generate_ed25519().public().to_peer_id() +} + +#[test] +fn relays_are_picked_when_needed() { + use ::libp2p::autonat::NatStatus; + let local = peer(); + let offers: HashMap<PeerId, Multiaddr> = (0..4) + .map(|port| (peer(), format!("/memory/{port}").parse().unwrap())) + .collect(); + let check = + |direct, nat: &NatStatus, using: &[PeerId]| relay_check(local, direct, nat, &offers, using); + // Dialable, or not known yet with an address to dial: none. + assert!(check(true, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty()); + assert!(check(false, &NatStatus::Public(Multiaddr::empty()), &[]).is_empty()); + assert!(check(true, &NatStatus::Unknown, &[]).is_empty()); + // No address to dial, or AutoNAT says it can't be dialled: some. + assert_eq!(check(false, &NatStatus::Unknown, &[]).len(), AUTO_RELAYS); + let picked = check(true, &NatStatus::Private, &[]); + assert_eq!(picked.len(), AUTO_RELAYS); + // One in use: the next in the same order. + let more = check(true, &NatStatus::Private, &[picked[0].0]); + assert_eq!(more.len(), AUTO_RELAYS - 1); + assert_eq!(more[0].0, picked[1].0); + let all: Vec<PeerId> = picked.iter().map(|(relay, _)| *relay).collect(); + assert!(check(true, &NatStatus::Private, &all).is_empty()); + // Never itself. + let only_us = HashMap::from([(local, Multiaddr::empty())]); + assert!(relay_check(local, false, &NatStatus::Private, &only_us, &[]).is_empty()); +} + +/// A relay says so in its addresses; b, which nobody can dial, learns of it (as from +/// the member list) and becomes reachable through it. +#[tokio::test] +async fn a_member_that_relays_is_picked() { + let (r, _) = memory_node_with(true, Vec::new(), true); + assert!(eventually(|| !r.local_addresses().is_empty()).await); + let offer = r + .local_addresses() + .into_iter() + .find(|address| address.ends_with("/p2p-circuit")) + .expect("the relay offers itself"); + let (b, _) = memory_node_with(false, Vec::new(), false); + assert_eq!(b.locate(&offer).as_deref(), Some(r.local_id())); + assert!( + eventually(|| b + .local_addresses() + .iter() + .any(|address| address.contains("/p2p-circuit/"))) + .await, + "b isn't reachable through r: {:?}", + b.local_addresses() + ); +} + +/// A peer on the LAN that speaks continuum becomes a seed once, at its address. +#[test] +fn lan_peers_become_seeds_once() { + let mut book = Book::default(); + let lan = peer(); + let address: Multiaddr = "/ip4/192.168.1.5/tcp/9661".parse().unwrap(); + book.discover(lan); // No address known: nothing. + assert!(book.discovered.is_empty()); + book.remember(lan, address.clone()); + book.discover(lan); + book.discover(lan); + assert_eq!(book.discovered, [address.with_p2p(lan).unwrap()]); +} + +#[test] +fn named_relays() { + let (local, r1, r2) = (peer(), peer(), peer()); + let at = |port: u16| -> Multiaddr { format!("/ip6/fd00::1/tcp/{port}").parse().unwrap() }; + let mut book = Book::default(); + // With a peer ID: known at once; without: to be found out, once. + assert_eq!( + book.name_relay(at(1).with_p2p(r1).unwrap(), Named::Configured, local), + None + ); + assert_eq!( + book.name_relay(at(2), Named::Configured, local), + Some(at(2)) + ); + assert_eq!(book.name_relay(at(2), Named::Configured, local), None); + assert_eq!( + book.name_relay(at(3).with_p2p(local).unwrap(), Named::Configured, local), + None + ); + assert!(book.relay_found(at(2), r2, local)); + assert_eq!(book.named(Named::Configured).len(), 2); + // A relay named with its peer ID isn't found out again; one without is. + assert_eq!(book.find_again(r1), None); + assert_eq!(book.find_again(r2), Some(at(2))); + assert!(!book.named(Named::Configured).contains_key(&r2)); + + // Listed: configured wins; unlisted ones go, unless also configured. + let r3 = peer(); + let listed = vec![ + at(1).with_p2p(r1).unwrap(), + at(4).with_p2p(r3).unwrap(), + at(5), + ]; + assert_eq!(book.learn_listed(listed.clone(), local), Some(vec![at(5)])); + assert_eq!(book.learn_listed(listed, local), None); + assert_eq!(book.named_relays[&r1].named, Named::Configured); + assert_eq!(book.named(Named::Listed).keys().collect::<Vec<_>>(), [&r3]); + assert_eq!(book.learn_listed(Vec::new(), local), Some(Vec::new())); + assert!(book.named(Named::Listed).is_empty()); + assert!(!book.finding.contains_key(&at(5))); + assert!(!book.relay_found(at(5), peer(), local)); // No longer listed. + assert_eq!(book.named(Named::Configured).len(), 1); +} + +#[test] +fn known_addresses() { + let mut book = Book::default(); + let (relay, us) = (peer(), peer()); + let direct: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap(); + let relayed: Multiaddr = format!("/ip6/fd00::2/tcp/9661/p2p/{relay}/p2p-circuit/p2p/{us}") + .parse() + .unwrap(); + assert!(!book.knows(us, &[direct.clone()])); + book.remember(us, relayed.clone()); + book.remember(us, direct.clone()); + // In any order, with or without the peer ID. + assert!(book.knows(us, &[relayed.clone(), direct.clone()])); + assert!(book.knows(us, &[direct.clone().with_p2p(us).unwrap()])); + assert!(!book.knows(us, &["/ip6/fd00::3/tcp/9661".parse().unwrap()])); +} + +/// Peers nobody mentions any more are forgotten, but connected ones. +#[test] +fn stale_peers_are_forgotten() { + let (old, current, connected) = (peer(), peer(), peer()); + let address: Multiaddr = "/ip6/fd00::1/tcp/9661".parse().unwrap(); + let mut book = Book::default(); + for peer in [old, current, connected] { + book.remember(peer, address.clone()); + } + book.relays.insert(old, address.clone()); + book.continuum.insert(old); + // From after the mentions above (taken at `Instant::now()`), however fast or + // slow the clock moved meanwhile. + let later = tokio::time::Instant::now() + STALE_AFTER; + // `current` was mentioned again since. + book.seen.insert(current, later); + assert_eq!(book.prune(later, |peer| *peer == connected), 1); + assert!(!book.addresses.contains_key(&old)); + assert!(!book.relays.contains_key(&old) && !book.continuum.contains(&old)); + assert!(book.addresses.contains_key(¤t)); + assert!(book.addresses.contains_key(&connected)); + assert_eq!(book.prune(later, |_| false), 1); // `connected`, once it isn't. +} + +#[test] +fn relay_failures_are_short() { + assert_eq!( + relay_failure_text( + "Custom { kind: Other, error: Other(Left(Left(Left(Reservation(Protocol(NoAddressesInReservation)))))) }", + "x" + ), + "the relay gave no address to be reached through it" + ); + assert_eq!( + relay_failure_text( + "Codec(Error(Custom { kind: Other, error: Failed }))", + "Failed" + ), + "the relay refused the circuit relay protocol (is it a relay?)" + ); + assert_eq!( + relay_failure_text("Io(…)", "connection reset"), + "connection reset" + ); + assert_eq!(relay_failure(&Ok(())), "closed"); +} + +/// Only refusals pause a relay; a slot that ended otherwise is asked for again. +#[test] +fn only_refusals_pause_relays() { + let refused = [ + ("Reservation(Protocol(NoAddressesInReservation))", "x"), + ("Reservation(Status(ResourceLimitExceeded))", "x"), + ("Reservation(Status(PermissionDenied))", "x"), + ("Denied { cause: \"connection gated\" }", "x"), + ("Upgrade(UnsupportedProtocol)", "x"), + ("Dial(WrongPeerId { obtained: … })", "x"), + ( + "Codec(Error(Custom { kind: Other, error: Failed }))", + "Failed", + ), + ]; + for (debug, innermost) in refused { + assert!(relay_refused(debug, innermost), "{debug}"); + } + let not = [ + ("Io(Custom { kind: Other, error: Timeout })", "Timeout"), + ( + "Io(Custom { kind: Other, error: Right(Closed) })", + "connection is closed", + ), + ("Io(…)", "connection reset"), + ]; + for (debug, innermost) in not { + assert!(!relay_refused(debug, innermost), "{debug}"); + } +} + +#[test] +fn listen_addresses_offered_for_hole_punching() { + let parse = |text: &str| text.parse::<Multiaddr>().unwrap(); + assert!(offered(&parse("/ip4/192.168.1.2/udp/9661/quic-v1"))); + assert!(offered(&parse("/ip6/fd00::2/tcp/9661"))); + assert!(!offered(&parse("/ip4/127.0.0.1/udp/9661/quic-v1"))); + assert!(!offered(&parse("/ip6/fe80::1/tcp/9661"))); + let relay = PeerId::random(); + assert!(!offered(&parse(&format!( + "/ip4/192.168.1.2/tcp/9661/p2p/{relay}/p2p-circuit" + )))); +} + +/// libp2p's nested dial errors come out as each address's cause, once. +#[test] +fn dial_errors_are_short() { + use ::libp2p::core::transport::TransportError; + let peer = peer(); + let address: Multiaddr = format!("/ip6/fd00::1/tcp/9661/p2p/{peer}").parse().unwrap(); + let nested = std::io::Error::other( + "Multiple dial errors occurred:\n - Network is unreachable (os error 101): Multiple dial errors occurred:\n - Network is unreachable (os error 101): Network is unreachable (os error 101)", + ); + let error = + ::libp2p::swarm::DialError::Transport(vec![(address, TransportError::Other(nested))]); + assert_eq!( + describe_dial_error(&error), + "/ip6/fd00::1/tcp/9661: Network is unreachable (os error 101)" + ); + let refused = ::libp2p::swarm::DialError::Transport(vec![( + "/ip4/10.0.0.1/tcp/1".parse().unwrap(), + TransportError::Other(std::io::Error::from(std::io::ErrorKind::ConnectionRefused)), + )]); + assert_eq!( + describe_dial_error(&refused), + "/ip4/10.0.0.1/tcp/1: connection refused" + ); +} + +/// A relay doesn't dial the nodes it relays for through itself. +#[test] +fn no_circuits_through_ourselves() { + let (us, other_relay, target) = (peer(), peer(), peer()); + let through = |relay: PeerId| -> Multiaddr { + format!("/ip6/fd00::1/tcp/9661/p2p/{relay}/p2p-circuit") + .parse() + .unwrap() + }; + let direct: Multiaddr = "/ip6/fd00::2/tcp/9661".parse().unwrap(); + assert_eq!(relay_of(&through(us)), Some(us)); + assert_eq!(relay_of(&direct), None); + assert_eq!(relay_of(&direct.clone().with_p2p(target).unwrap()), None); + assert_eq!( + dialable(&[through(us), through(other_relay), direct.clone()], us), + [through(other_relay), direct] + ); +} + +#[tokio::test] +async fn unknown_peers_are_refused() { + let (b, _) = memory_node(); + let stranger = Keypair::generate_ed25519() + .public() + .to_peer_id() + .to_string(); + assert!(b.send_message(&stranger, &status()).await.is_err()); + assert!(b.resolve_seed(&stranger).await.is_err()); + assert_eq!(b.connection(&stranger), None); +} + +#[test] +fn a_direct_connection_wins_over_a_relayed_one() { + let mut book = Book::default(); + let (relayed_only, both) = (peer(), peer()); + let mut id = 0; + let mut connect = |book: &mut Book, peer: PeerId, relayed: bool| { + id += 1; + let address: Multiaddr = format!("/ip4/10.0.0.{id}/tcp/1").parse().unwrap(); + book.connections + .insert(ConnectionId::new_unchecked(id), (peer, relayed, address)); + }; + connect(&mut book, relayed_only, true); + connect(&mut book, both, true); + connect(&mut book, both, false); + assert_eq!(book.connection(&relayed_only), Some(Connection::Relayed)); + assert_eq!(book.connection(&both), Some(Connection::Direct)); + assert_eq!(book.connection(&peer()), None); + assert_eq!( + book.location(&both).map(|address| address.to_string()), + Some("/ip4/10.0.0.3/tcp/1".to_owned()) + ); + let connections = book.stats().connections; + assert_eq!((connections.direct, connections.relayed), (1, 2)); + // Of several, the oldest direct one stays; a relayed one only without a direct. + connect(&mut book, both, false); + assert_eq!( + book.duplicates(&both, false), + [ + ConnectionId::new_unchecked(4), + ConnectionId::new_unchecked(2) + ] + ); + assert!(book.duplicates(&relayed_only, false).is_empty()); + // Not while one of them carries a slot, ours or the peer's, or we reserve one. + assert!(book.duplicates(&both, true).is_empty()); + book.slots.insert(both); + assert!(book.duplicates(&both, false).is_empty()); + book.slots.clear(); + book.slots_here.insert(both); + assert!(book.duplicates(&both, false).is_empty()); + book.slots_here.clear(); + // Once measured, the fastest direct one stays, however young. + book.rtts + .insert(ConnectionId::new_unchecked(3), Duration::from_millis(40)); + book.rtts + .insert(ConnectionId::new_unchecked(4), Duration::from_millis(2)); + assert_eq!( + book.duplicates(&both, false), + [ + ConnectionId::new_unchecked(3), + ConnectionId::new_unchecked(2) + ] + ); + assert_eq!( + book.location(&both).map(|address| address.to_string()), + Some("/ip4/10.0.0.4/tcp/1".to_owned()) + ); +} |
