libp2p: explode!

1 files changed, 364 insertions, 0 deletions

diff --git a/src/transport/libp2p/behaviour.rs b/src/transport/libp2p/behaviour.rs
new file mode 100644
index 0000000..4ced88a
--- /dev/null
+++ b/src/transport/libp2p/behaviour.rs
@@ -0,0 +1,364 @@
+//! The libp2p behaviours this transport combines (`Behaviour`), hole punching with LAN
+//! candidates (`HolePunching`), and building the swarm (TCP and QUIC, or in memory).
+
+use super::*;
+
+/// The protocol version identify says (informational: nothing checks it).
+pub(super) const IDENTIFY_VERSION: &str = "/continuum/1";
+
+#[derive(::libp2p::swarm::NetworkBehaviour)]
+pub(super) struct Behaviour {
+    /// First: connections it denies need nothing from the others.
+    pub(super) ignoring: Ignoring,
+    pub(super) stream: libp2p_stream::Behaviour,
+    pub(super) identify: ::libp2p::identify::Behaviour,
+    pub(super) relay_client: ::libp2p::relay::client::Behaviour,
+    /// Only with `Libp2pConfig::relay`.
+    pub(super) relay: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::relay::Behaviour>,
+    pub(super) dcutr: HolePunching,
+    /// Each connection's round trip (`Book::rtts`): of several to one peer, the fastest
+    /// is kept.
+    pub(super) ping: ::libp2p::ping::Behaviour,
+    pub(super) autonat: ::libp2p::autonat::Behaviour,
+    /// Only with `Libp2pConfig::mdns`.
+    pub(super) mdns: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::mdns::tokio::Behaviour>,
+    /// Only with `Libp2pConfig::upnp`: maps our listen ports on the router, and confirms
+    /// the mapped addresses as ours (identify then tells them).
+    pub(super) upnp: ::libp2p::swarm::behaviour::toggle::Toggle<::libp2p::upnp::tokio::Behaviour>,
+    /// Only with `Libp2pConfig::public`: the IPFS DHT.
+    pub(super) kad: ::libp2p::swarm::behaviour::toggle::Toggle<
+        ::libp2p::kad::Behaviour<::libp2p::kad::store::MemoryStore>,
+    >,
+}
+
+/// libp2p's hole punching (DCUtR), offering our listen addresses too. DCUtR offers only
+/// where others saw us (external address candidates, from identify): seen by a relay
+/// out on the internet, two nodes on one LAN are both at their router's public address,
+/// and punching through it needs the router to hairpin, which many don't. Their LAN
+/// addresses work, the firewalls on the way being stateful. Only DCUtR is told:
+/// AutoNAT would test every candidate, and fail on all of these.
+pub(super) struct HolePunching {
+    pub(super) inner: ::libp2p::dcutr::Behaviour,
+    /// Where we listen (`offered`).
+    pub(super) listening: Vec<Multiaddr>,
+    /// `Libp2pConfig::ignore`: not offered.
+    pub(super) ignored: Vec<IpRange>,
+}
+
+/// Whether DCUtR offers our listen address `address` too: neither loopback nor
+/// link-local (as for publishing), nor through a relay.
+pub(super) fn offered(address: &Multiaddr) -> bool {
+    !is_local(address) && !is_relayed(address)
+}
+
+impl HolePunching {
+    pub(super) fn new(local: PeerId, ignored: Vec<IpRange>) -> Self {
+        Self {
+            inner: ::libp2p::dcutr::Behaviour::new(local),
+            listening: Vec::new(),
+            ignored,
+        }
+    }
+
+    /// Tells DCUtR where we listen, as if others had seen us there. Again after every
+    /// other candidate, as DCUtR keeps the latest 20.
+    pub(super) fn offer_listening(&mut self) {
+        use ::libp2p::swarm::{FromSwarm, NetworkBehaviour, NewExternalAddrCandidate};
+        for addr in &self.listening {
+            self.inner
+                .on_swarm_event(FromSwarm::NewExternalAddrCandidate(
+                    NewExternalAddrCandidate { addr },
+                ));
+        }
+    }
+}
+
+impl ::libp2p::swarm::NetworkBehaviour for HolePunching {
+    type ConnectionHandler =
+        <::libp2p::dcutr::Behaviour as ::libp2p::swarm::NetworkBehaviour>::ConnectionHandler;
+    type ToSwarm = ::libp2p::dcutr::Event;
+
+    fn handle_pending_inbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        local_addr: &Multiaddr,
+        remote_addr: &Multiaddr,
+    ) -> Result<(), ::libp2p::swarm::ConnectionDenied> {
+        self.inner
+            .handle_pending_inbound_connection(connection_id, local_addr, remote_addr)
+    }
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        peer: PeerId,
+        local_addr: &Multiaddr,
+        remote_addr: &Multiaddr,
+    ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> {
+        if is_relayed(local_addr) {
+            // DCUtR offers its candidates on this connection: where others saw us, and
+            // these.
+            debug!(peer = %Addressed(SCHEME, &peer), listening = ?self.listening, "Reached through a relay; hole punching offers where we listen too");
+        }
+        self.inner.handle_established_inbound_connection(
+            connection_id,
+            peer,
+            local_addr,
+            remote_addr,
+        )
+    }
+
+    fn handle_pending_outbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        maybe_peer: Option<PeerId>,
+        addresses: &[Multiaddr],
+        effective_role: ::libp2p::core::Endpoint,
+    ) -> Result<Vec<Multiaddr>, ::libp2p::swarm::ConnectionDenied> {
+        self.inner.handle_pending_outbound_connection(
+            connection_id,
+            maybe_peer,
+            addresses,
+            effective_role,
+        )
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        peer: PeerId,
+        addr: &Multiaddr,
+        role_override: ::libp2p::core::Endpoint,
+        port_use: ::libp2p::core::transport::PortUse,
+    ) -> Result<::libp2p::swarm::THandler<Self>, ::libp2p::swarm::ConnectionDenied> {
+        self.inner.handle_established_outbound_connection(
+            connection_id,
+            peer,
+            addr,
+            role_override,
+            port_use,
+        )
+    }
+
+    fn on_swarm_event(&mut self, event: ::libp2p::swarm::FromSwarm) {
+        use ::libp2p::swarm::FromSwarm;
+        self.inner.on_swarm_event(event);
+        match event {
+            FromSwarm::NewListenAddr(new)
+                if offered(new.addr) && !ignored(&self.ignored, new.addr) =>
+            {
+                if !self.listening.contains(new.addr) {
+                    debug!(address = %new.addr, "Offering where we listen for hole punching");
+                    self.listening.push(new.addr.clone());
+                }
+                self.offer_listening();
+            }
+            FromSwarm::ExpiredListenAddr(expired) => {
+                self.listening.retain(|address| address != expired.addr);
+            }
+            FromSwarm::NewExternalAddrCandidate(_) => self.offer_listening(),
+            _ => {}
+        }
+    }
+
+    fn on_connection_handler_event(
+        &mut self,
+        peer_id: PeerId,
+        connection_id: ConnectionId,
+        event: ::libp2p::swarm::THandlerOutEvent<Self>,
+    ) {
+        self.inner
+            .on_connection_handler_event(peer_id, connection_id, event)
+    }
+
+    fn poll(
+        &mut self,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<
+        ::libp2p::swarm::ToSwarm<Self::ToSwarm, ::libp2p::swarm::THandlerInEvent<Self>>,
+    > {
+        self.inner.poll(cx)
+    }
+}
+
+/// What the behaviour has besides what every node has.
+#[derive(Clone, Copy)]
+pub(super) struct Features {
+    pub(super) relay: bool,
+    pub(super) mdns: bool,
+    pub(super) upnp: bool,
+    pub(super) public: bool,
+}
+
+impl Features {
+    pub(super) fn of(config: &Libp2pConfig) -> Self {
+        Self {
+            relay: config.relay,
+            mdns: config.mdns,
+            upnp: config.upnp,
+            public: config.public,
+        }
+    }
+}
+
+/// The IPFS DHT's protocol, and its bootstrap nodes (all at `IPFS_BOOTSTRAP_ADDRESS`).
+pub(super) const IPFS_DHT: StreamProtocol = StreamProtocol::new("/ipfs/kad/1.0.0");
+
+pub(super) const IPFS_BOOTSTRAP_ADDRESS: &str = "/dnsaddr/bootstrap.libp2p.io";
+
+pub(super) const IPFS_BOOTSTRAP_PEERS: [&str; 4] = [
+    "QmNnooDu7bfjPFoTZYxMNLWUQJyrVwtbZg5gBMjTezGAJN",
+    "QmQCU2EcMqAqQPR2i9bChDtGNJchTbq5TbXJJ16u19uLTa",
+    "QmbLHAnMoJPWSCR5Zhtx6BHJX9KiKNN6tpvbUcqanj75Nb",
+    "QmcZf59bWwK5XFi76CZX8cbJ4BhTzzA3gU1ZjYZcYW3dwt",
+];
+
+/// How often a node on the DHT refreshes its routing table.
+pub(super) const DHT_BOOTSTRAP: Duration = Duration::from_secs(300);
+
+impl Behaviour {
+    pub(super) fn new(
+        key: &Keypair,
+        relay_client: ::libp2p::relay::client::Behaviour,
+        features: Features,
+        ignored: Vec<IpRange>,
+    ) -> std::io::Result<Self> {
+        let Features {
+            relay,
+            mdns,
+            upnp,
+            public,
+        } = features;
+        let peer_id = key.public().to_peer_id();
+        let kad = public.then(|| {
+            let mut dht = ::libp2p::kad::Behaviour::with_config(
+                peer_id,
+                ::libp2p::kad::store::MemoryStore::new(peer_id),
+                ::libp2p::kad::Config::new(IPFS_DHT),
+            );
+            let bootstrap: Multiaddr = IPFS_BOOTSTRAP_ADDRESS.parse().expect("a multiaddr");
+            for peer in IPFS_BOOTSTRAP_PEERS {
+                let peer: PeerId = peer.parse().expect("a peer ID");
+                let _ = dht.add_address(&peer, bootstrap.clone());
+            }
+            dht
+        });
+        let identify = ::libp2p::identify::Config::new(IDENTIFY_VERSION.to_owned(), key.public())
+            .with_agent_version(format!("continuum/{}", env!("CARGO_PKG_VERSION")));
+        let relay = relay.then(|| {
+            ::libp2p::relay::Behaviour::new(
+                peer_id,
+                ::libp2p::relay::Config {
+                    max_circuits: RELAY_CIRCUITS,
+                    max_circuits_per_peer: RELAY_CIRCUITS_PER_PEER,
+                    max_circuit_duration: RELAY_CIRCUIT_DURATION,
+                    max_circuit_bytes: RELAY_CIRCUIT_BYTES,
+                    reservation_rate_limiters: Vec::new(),
+                    circuit_src_rate_limiters: Vec::new(),
+                    ..Default::default()
+                },
+            )
+        });
+        let mdns = match mdns {
+            true => Some(::libp2p::mdns::tokio::Behaviour::new(
+                ::libp2p::mdns::Config::default(),
+                peer_id,
+            )?),
+            false => None,
+        };
+        Ok(Self {
+            ignoring: Ignoring {
+                ranges: ignored.clone(),
+            },
+            stream: libp2p_stream::Behaviour::new(),
+            identify: ::libp2p::identify::Behaviour::new(identify),
+            relay_client,
+            relay: relay.into(),
+            dcutr: HolePunching::new(peer_id, ignored),
+            ping: ::libp2p::ping::Behaviour::new(::libp2p::ping::Config::new()),
+            autonat: ::libp2p::autonat::Behaviour::new(
+                peer_id,
+                ::libp2p::autonat::Config {
+                    // Swarms on private networks (a LAN, WireGuard) count too.
+                    only_global_ips: false,
+                    ..Default::default()
+                },
+            ),
+            mdns: mdns.into(),
+            upnp: upnp.then(::libp2p::upnp::tokio::Behaviour::default).into(),
+            kad: kad.into(),
+        })
+    }
+}
+
+pub(super) fn network_swarm(
+    key: Keypair,
+    features: Features,
+    ignored: Vec<IpRange>,
+) -> Result<Swarm<Behaviour>> {
+    Ok(::libp2p::SwarmBuilder::with_existing_identity(key)
+        .with_tokio()
+        .with_tcp(
+            ::libp2p::tcp::Config::default(),
+            ::libp2p::noise::Config::new,
+            ::libp2p::yamux::Config::default,
+        )?
+        .with_quic()
+        .with_dns()?
+        .with_relay_client(
+            ::libp2p::noise::Config::new,
+            ::libp2p::yamux::Config::default,
+        )?
+        .with_behaviour(|key, relay_client| {
+            Behaviour::new(key, relay_client, features, ignored.clone())
+                .map_err(|error| Box::new(error) as Box<dyn std::error::Error + Send + Sync>)
+        })?
+        .with_swarm_config(|config| swarm_config(config, features))
+        .build())
+}
+
+/// The swarm's settings. Streams are opened without waiting for the other side to agree
+/// on the protocol (`V1Lazy`): the protocol and the first data go together, saving a
+/// round trip per message (one stream each, module docs). That is safe only towards
+/// peers sure to speak the protocol, so not with public libp2p infrastructure
+/// (`Libp2pConfig::public`), whose nodes speak only some: there a protocol they don't
+/// would fail late, and unclearly.
+pub(super) fn swarm_config(
+    config: ::libp2p::swarm::Config,
+    features: Features,
+) -> ::libp2p::swarm::Config {
+    let config = config
+        .with_idle_connection_timeout(IDLE_TIMEOUT)
+        .with_dial_concurrency_factor(DIAL_CONCURRENCY);
+    if features.public {
+        config
+    } else {
+        config.with_substream_upgrade_protocol_override(::libp2p::core::upgrade::Version::V1Lazy)
+    }
+}
+
+pub(super) fn memory_swarm(
+    key: Keypair,
+    features: Features,
+    ignored: Vec<IpRange>,
+) -> Result<Swarm<Behaviour>> {
+    use ::libp2p::Transport as _;
+    use ::libp2p::core::transport::{MemoryTransport, OrTransport};
+    use ::libp2p::core::upgrade::Version;
+
+    let peer_id = key.public().to_peer_id();
+    let (relay_transport, relay_client) = ::libp2p::relay::client::new(peer_id);
+    let transport = OrTransport::new(relay_transport, MemoryTransport::default())
+        .boxed()
+        .upgrade(Version::V1)
+        .authenticate(::libp2p::noise::Config::new(&key)?)
+        .multiplex(::libp2p::yamux::Config::default())
+        .timeout(CONNECT_TIMEOUT)
+        .boxed();
+    Ok(Swarm::new(
+        transport,
+        Behaviour::new(&key, relay_client, features, ignored)?,
+        peer_id,
+        swarm_config(::libp2p::swarm::Config::with_tokio_executor(), features),
+    ))
+}