From 235411a261441df6c0ddb1c4d7f99cf06b8f3c55 Mon Sep 17 00:00:00 2001 From: tv Date: Sat, 31 Oct 2026 04:20:41 +0100 Subject: libp2p: explode! --- src/transport/libp2p.rs | 3411 ----------------------------------------------- 1 file changed, 3411 deletions(-) delete mode 100644 src/transport/libp2p.rs (limited to 'src/transport/libp2p.rs') diff --git a/src/transport/libp2p.rs b/src/transport/libp2p.rs deleted file mode 100644 index b9f54a9..0000000 --- a/src/transport/libp2p.rs +++ /dev/null @@ -1,3411 +0,0 @@ -//! 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 -//! (`/p2p-circuit/p2p/`) 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, -}; - -/// 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) -} - -/// What a libp2p transport is made with. -pub(crate) struct Libp2pConfig { - pub(crate) key: Keypair, - /// Multiaddrs to listen on. - pub(crate) listen: Vec, - /// Multiaddrs to publish instead of the listen addresses, if any. - pub(crate) announce: Vec, - /// Relays to reserve a slot on (multiaddrs; without the relay's peer ID, a connection - /// finds it out). - pub(crate) relays: Vec, - /// 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, - /// 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`. -fn parse_addresses(option: &str, addresses: &[String]) -> Result> { - addresses - .iter() - .map(|address| { - address - .parse::() - .with_context(|| format!("{option} {address:?}")) - }) - .collect() -} - -/// The ranges of `libp2p-ignore`. -fn parse_ranges(ranges: &[String]) -> Result> { - 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 { - 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 { - 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())) - } -} - -enum Command { - /// Connect to `peer` (at `addresses`) unless connected. - Connect { - peer: PeerId, - addresses: Vec, - done: oneshot::Sender>, - }, - /// Connect to `address` to find out which peer is there. - Identify { - address: Multiaddr, - done: oneshot::Sender>, - }, - /// 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 }, -} - -/// Who named a relay to pick, in the order they are picked. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] -enum Named { - /// `Libp2pConfig::relays`. - Configured, - /// The relays listed in the document (`Transport::learn_relays`). - Listed, -} - -impl Named { - fn as_str(self) -> &'static str { - match self { - Self::Configured => "configured", - Self::Listed => "listed", - } - } -} - -/// A relay named to pick. -#[derive(Clone, Debug, PartialEq, Eq)] -struct NamedRelay { - /// Where, with the relay's peer ID. - 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). - given: Multiaddr, - named: Named, -} - -/// Where peers are, and where we listen. -#[derive(Default)] -struct Book { - addresses: HashMap>, - listening: Vec, - /// Where the router forwards to our listen addresses (UPnP, `Libp2pConfig::upnp`). - mapped: Vec, - /// `Libp2pConfig::announce`. - announced: Vec, - /// `Libp2pConfig::ignore`: not published (but announced), not dialled. - ignored: Vec, - /// `Libp2pConfig::relay`: this node relays, and says so. - serving: bool, - /// Relays named to pick (configured, listed), by peer: picked first. - named_relays: HashMap, - /// Relays named without their peer ID, being found out (`Command::FindRelay`). - finding: HashMap, - /// The relays listed in the document, as last learned (`learn_listed`). - listed: Vec, - /// Members that say they relay, and where (with the relay's peer ID). - relays: HashMap, - /// With `Libp2pConfig::public`: other peers that relay (identify says so), and - /// where. - public_relays: HashMap, - /// Relays not to be picked, until when (`RELAY_PAUSE`, `RELAY_RETRY`). - failed_relays: HashMap, - /// Relays that granted us a slot we hold. - slots: std::collections::HashSet, - /// Peers holding a slot on this node, as a relay. - slots_here: std::collections::HashSet, - /// Peers connected only through a relay, dialled directly (`upgrade`): when to again - /// at the soonest, and how long to wait after that. - upgrades: HashMap, - /// The shortest round trip measured on each open connection (libp2p's ping). - rtts: HashMap, - /// Peers whose hole punching failed: how often in a row, and when to try again - /// (`PUNCH_RETRIES`). - punching: HashMap, - /// Open connections: to whom, whether through a relay, and the remote address. - connections: HashMap, - /// What AutoNAT says, and the address it confirmed while public. - reachability: Reachability, - public_address: Option, - /// Since start (`stats`). - counts: Counts, - /// When each peer was last mentioned (`STALE_AFTER`). - seen: HashMap, - /// Peers found on the LAN (mDNS) not known yet to speak continuum. - lan: std::collections::HashSet, - /// Peers that speak continuum (identify says so). - continuum: std::collections::HashSet, - /// 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. - not_nodes: HashMap, - /// Peers found on the LAN that speak continuum, with their address: seeds not - /// handed out yet (`discovered_seeds`). - discovered: Vec, -} - -/// What `Libp2pStats` counts since start. -#[derive(Default)] -struct Counts { - slots_granted: u64, - slots_lost: u64, - slots_not_granted: u64, - hole_punches: HolePunches, - dial_failures: DialFailures, -} - -/// Counts a dial that failed with `error` (not one that wasn't needed). -/// 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. -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 = 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. -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. -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. -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, ` - : ...`), -/// or the last of the source chain. -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(", ") -} - -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, - } -} - -impl Book { - /// How we are connected to `peer`: directly if any connection is direct. - fn connection(&self, peer: &PeerId) -> Option { - 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`). - fn duplicates(&self, peer: &PeerId, reserving: bool) -> Vec { - 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. - fn ranked(&self, peer: &PeerId) -> Vec { - 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`). - fn location(&self, peer: &PeerId) -> Option<&Multiaddr> { - let kept = *self.ranked(peer).first()?; - self.connections.get(&kept).map(|(_, _, address)| address) - } - - 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. - 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. - fn prune(&mut self, now: tokio::time::Instant, connected: impl Fn(&PeerId) -> bool) -> usize { - let stale: Vec = 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::>() - .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. - 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. - fn direct(&self) -> Vec { - if !self.announced.is_empty() { - return self.announced.clone(); - } - let direct: Vec = 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 = 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. -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). -fn published(all: &[Multiaddr]) -> Vec { - let reachable: Vec = 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. - fn name_relay(&mut self, given: Multiaddr, named: Named, local: PeerId) -> Option { - 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. - 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. - fn find_again(&mut self, relay: PeerId) -> Option { - 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. - fn learn_listed(&mut self, listed: Vec, local: PeerId) -> Option> { - 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. - fn named(&self, named: Named) -> HashMap { - 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/`), which may hold - /// a slot on one of them. `local`: this node, never its own relay. - fn through_relays(&self, peer: PeerId, local: PeerId) -> Vec { - 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`). - 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. - 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. -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`. -fn relay_of(address: &Multiaddr) -> Option { - let mut relay = None; - for protocol in address.iter() { - match protocol { - Protocol::P2p(peer) => relay = Some(peer), - Protocol::P2pCircuit => return relay, - _ => {} - } - } - None -} - -/// 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). -/// 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 { - network: std::net::IpAddr, - prefix: u8, -} - -impl FromStr for IpRange { - type Err = anyhow::Error; - - fn from_str(text: &str) -> Result { - 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::() - .ok() - .filter(|prefix| *prefix <= bits) - .with_context(|| format!("{text:?}: the prefix is 0 to {bits}"))?, - None => bits, - }; - Ok(Self { network, prefix }) - } -} - -impl IpRange { - 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). -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. -struct Ignoring { - ranges: Vec, -} - -#[derive(Debug, thiserror::Error)] -#[error("{0} is in a range left out (libp2p-ignore)")] -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, ::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, ::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, - ) { - match event {} - } - - fn poll( - &mut self, - _cx: &mut std::task::Context<'_>, - ) -> std::task::Poll< - ::libp2p::swarm::ToSwarm>, - > { - std::task::Poll::Pending - } -} - -/// How `peer` is connected, of `connections` (`Book::connections`): directly if one is -/// direct. -fn book_connection( - connections: &HashMap, - peer: &PeerId, -) -> Option { - let mut found = None; - for (connected, relayed, _) in connections.values() { - if connected == peer { - if !relayed { - return Some(Connection::Direct); - } - found = Some(Connection::Relayed); - } - } - found -} - -fn dialable(known: &[Multiaddr], local: PeerId) -> Vec { - known - .iter() - .filter(|address| relay_of(address) != Some(local)) - .cloned() - .collect() -} - -/// `address` with the peer ID it ends in, if it ends in one. -fn split_peer(address: &Multiaddr) -> Option { - match address.iter().last() { - Some(Protocol::P2p(peer)) => Some(peer), - _ => None, - } -} - -pub(crate) struct Libp2pTransport { - local_id: String, - peer_id: PeerId, - control: libp2p_stream::Control, - commands: mpsc::UnboundedSender, - book: Arc>, - incoming: tokio::sync::Mutex>, - stopped: watch::Receiver, -} - -fn lock(book: &Mutex) -> MutexGuard<'_, Book> { - book.lock_or_recover() -} - -impl Libp2pTransport { - /// Builds the swarm, listens, and runs it in a task of its own. - pub(crate) fn start(config: Libp2pConfig) -> Result { - 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:?}"), - } - } -} - -/// The protocol version identify says (informational: nothing checks it). -const IDENTIFY_VERSION: &str = "/continuum/1"; - -#[derive(::libp2p::swarm::NetworkBehaviour)] -struct Behaviour { - /// First: connections it denies need nothing from the others. - ignoring: Ignoring, - stream: libp2p_stream::Behaviour, - identify: ::libp2p::identify::Behaviour, - relay_client: ::libp2p::relay::client::Behaviour, - /// Only with `Libp2pConfig::relay`. - relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>, - dcutr: HolePunching, - /// Each connection's round trip (`Book::rtts`): of several to one peer, the fastest - /// is kept. - ping: ::libp2p::ping::Behaviour, - autonat: ::libp2p::autonat::Behaviour, - /// Only with `Libp2pConfig::mdns`. - 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). - upnp: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::upnp::tokio::Behaviour>, - /// Only with `Libp2pConfig::public`: the IPFS DHT. - 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. -struct HolePunching { - inner: ::libp2p::dcutr::Behaviour, - /// Where we listen (`offered`). - listening: Vec, - /// `Libp2pConfig::ignore`: not offered. - ignored: Vec, -} - -/// Whether DCUtR offers our listen address `address` too: neither loopback nor -/// link-local (as for publishing), nor through a relay. -fn offered(address: &Multiaddr) -> bool { - !is_local(address) && !is_relayed(address) -} - -impl HolePunching { - fn new(local: PeerId, ignored: Vec) -> 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. - 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, ::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, - addresses: &[Multiaddr], - effective_role: ::libp2p::core::Endpoint, - ) -> Result, ::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, ::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.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.inner.poll(cx) - } -} - -/// What the behaviour has besides what every node has. -#[derive(Clone, Copy)] -struct Features { - relay: bool, - mdns: bool, - upnp: bool, - public: bool, -} - -impl Features { - 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`). -const IPFS_DHT: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0"); -const IPFS_BOOTSTRAP_ADDRESS: &str = "/dnsaddr/bootstrap.libp2p.io"; -const IPFS_BOOTSTRAP_PEERS: [&str; 4] = [ - "QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN", - "QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa", - "QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb", - "QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt", -]; -/// How often a node on the DHT refreshes its routing table. -const DHT_BOOTSTRAP: Duration = Duration::from_secs(300); - -impl Behaviour { - fn new( - key: &Keypair, - relay_client: ::libp2p::relay::client::Behaviour, - features: Features, - ignored: Vec, - ) -> std::io::Result { - 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(), - }) - } -} - -fn network_swarm( - key: Keypair, - features: Features, - ignored: Vec, -) -> Result> { - 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) - })? - .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. -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) - } -} - -fn memory_swarm( - key: Keypair, - features: Features, - ignored: Vec, -) -> Result> { - 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), - )) -} - -/// How a node relates to relays. -struct Relaying { - /// It relays for others. - serving: bool, - /// Public relays may be picked, after members. - public: bool, - 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. -fn relay_check( - local: PeerId, - direct: bool, - nat: &::libp2p::autonat::NatStatus, - offers: &HashMap, - 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. -fn pick_relays( - swarm: &mut Swarm, - book: &Mutex, - 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| -> HashMap { - 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 = 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`). -fn upgrade(swarm: &mut Swarm, book: &Mutex, peer: PeerId, local: PeerId) { - let direct: Vec = { - let mut book = lock(book); - if book.connection(&peer) != Some(Connection::Relayed) { - return; - } - let direct: Vec = 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. -fn retry_punches(swarm: &mut Swarm, book: &Mutex, local: PeerId) { - let now = tokio::time::Instant::now(); - let mut close: Vec = Vec::new(); - let due: Vec<(PeerId, usize, Vec)> = { - 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 = 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. -fn find_on_dht(swarm: &mut Swarm, peer: PeerId) { - if let Some(dht) = swarm.behaviour_mut().kad.as_mut() { - dht.get_closest_peers(peer); - } -} - -/// Drives the swarm: dials what `commands` ask for and notes where we listen, until the -/// transport is dropped. `again`: to schedule commands. -async fn run( - mut swarm: Swarm, - mut commands: mpsc::UnboundedReceiver, - again: mpsc::UnboundedSender, - book: Arc>, - relaying: Relaying, - stopping: watch::Sender, -) { - 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>>> = HashMap::new(); - let mut identifying: HashMap< - ConnectionId, - (Multiaddr, oneshot::Sender>), - > = 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 = HashMap::new(); - // Connections closed as duplicates, until they are: their "Connection closed" says so. - let mut closing_duplicates: std::collections::HashSet = - 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 = 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); -} - -/// One message from an inbound stream. -async fn read_message(mut stream: ::libp2p::Stream) -> Result> { - 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 { - let book = lock(&self.book); - let direct: Vec = 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 = 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 { - 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: `
/p2p//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 { - let peer = PeerId::from_str(target).ok()?; - lock(&self.book).connection(&peer) - } - - fn punching(&self, target: &str) -> Option { - 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 { - 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 { - 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 { - 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 { - 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/`): reached where it's known, or through the - // relays we know. - if let [Protocol::P2p(peer)] = address.iter().collect::>()[..] { - 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 { - 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 \ - (/p2p-circuit/p2p/, or /p2p/ 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 = 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 { - 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 { - 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, - 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, - ) -> 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::().unwrap(); - let address = |text: &str| text.parse::().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::().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::().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::().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::().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 = (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 = 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::>(), [&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::().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()) - ); - } -} -- cgit v1.3.1