path: root/src/transport/libp2p/behaviour.rs

//! 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: Identifying,
    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,
    /// How long a connection may idle (`idle_timeout`).
    pub(super) idle_timeout: Duration,
}

impl Features {
    pub(super) fn of(config: &Libp2pConfig) -> Self {
        Self {
            relay: config.relay,
            mdns: config.mdns,
            upnp: config.upnp,
            public: config.public,
            idle_timeout: idle_timeout(config.keepalive),
        }
    }
}

/// 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,
            idle_timeout: _,
        } = 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")))
            // Connected peers hear of a new interface at once (`explore`).
            .with_push_listen_addr_updates(true);
        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: Identifying {
                inner: ::libp2p::identify::Behaviour::new(identify),
                ranges: ignored.clone(),
                // Set by `Libp2pTransport::start`.
                announced: Vec::new(),
            },
            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(features.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),
    ))
}