snowstorm/net/

swarm.rs

use super::behaviour::SnowstormBehaviour;
use crate::{
    config::Config,
    net::{get_default_interface, netif::DebouncedWatcher},
};
use futures::StreamExt;
use lazy_static::lazy_static;
use libp2p::{core::muxing::StreamMuxerBox, multiaddr::Protocol, Multiaddr, Swarm, Transport};
use libp2p_webrtc as webrtc;
use log::{debug, info, warn};
use once_cell::sync::Lazy;
use parking_lot::RwLock;
use prometheus_client::registry::Registry;
use socket2::Socket;
#[cfg(not(target_os = "windows"))]
use std::os::fd::{AsFd, AsRawFd};
use std::{
    io,
    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket},
    sync::{Arc, Mutex},
    time::Duration,
};
use tokio_util::sync::CancellationToken;
use transport::signalling::ControlHandle;
use webrtc::tokio::Certificate;

pub mod auth;
mod transport;

static OUTBOUND_ADDRS: Lazy<Arc<RwLock<Vec<IpAddr>>>> = Lazy::new(|| Arc::new(RwLock::new(vec![])));

lazy_static! {
    static ref CONFIG: Mutex<Config> = Mutex::new(Default::default());
}

pub(super) fn build_swarm(
    token: CancellationToken,
    config: Config,
    registry: &mut std::sync::MutexGuard<'_, Registry>,
    num_streams: Option<usize>,
) -> (Swarm<SnowstormBehaviour>, Arc<ControlHandle>) {
    {
        let mut config_guard = CONFIG.try_lock().unwrap();
        *config_guard = config.clone();
    }

    if !config.outbound_addrs.is_empty() {
        {
            let mut addrs = OUTBOUND_ADDRS.write();

            *addrs = config
                .outbound_addrs
                .iter()
                .map(multiaddr_to_ipaddr)
                .filter_map(Result::ok)
                .collect();
        }

        // Debounced IF watcher to make sure we have up to date addrs when network +- stabilizes
        if let Ok(mut set) = DebouncedWatcher::new(Duration::from_millis(500)) {
            tokio::spawn(async move {
                loop {
                    tokio::select! {
                        _ = token.cancelled() => {
                            return;
                        }
                        new_outbound_addrs = set.select_next_some() => {
                            info!(
                                "Got IF event: Outbound IPs changed {:?}",
                                new_outbound_addrs
                            );
                            let mut addrs = OUTBOUND_ADDRS.write();
                            *addrs = new_outbound_addrs;

                            info!("Updated outbound addresses updated");
                        }
                    }
                }
            });
        } else {
            warn!("Failed to create debounced IF watcher, using static addresses");
        }
    }

    let yamux_config = libp2p::yamux::Config::default();
    let mut quic_config = libp2p::quic::Config::new(&config.keypair.clone());
    let mut tcp_config = libp2p::tcp::Config::default();

    match config.outbound_addrs.len() {
        0 => {
            warn!("no outbound addresses configured, forced binding disabled");
        }
        _ => {
            tcp_config = tcp_config.pre_connect_cb(tcp_outgoing_cb);

            quic_config.socket_created_cb = Some(udp_protect_outgoing);
            quic_config.eligible_listener_cb = Some(quic_eligible_listener_cb);
            quic_config.outgoing_dial_addr_cb = Some(quic_outgoing_dial_addr_cb);
        }
    };

    if let Some(num_streams) = num_streams {
        // Setting this makes the mux upset on fly.io for some reason,
        // so just use the default config for now.
        //
        // yamux_config.set_max_num_streams(num_streams);
        quic_config.max_concurrent_stream_limit = num_streams as u32;
        info!("increased max concurrent streams: {}", num_streams);
    }

    let tcp_transport = libp2p::tcp::tokio::Transport::new(tcp_config.clone())
        .upgrade(libp2p::core::upgrade::Version::V1Lazy)
        .authenticate(libp2p::noise::Config::new(&config.keypair).unwrap())
        .multiplex(yamux_config.clone())
        .map(|(p, c), _| (p, StreamMuxerBox::new(c)));

    let quic_transport = libp2p::quic::tokio::Transport::new(quic_config.clone())
        .map(|(peer_id, muxer), _| (peer_id, StreamMuxerBox::new(muxer)));

    let webrtc_transport = webrtc::tokio::Transport::new(
        config.keypair.clone(),
        Certificate::generate(&mut rand::rng()).unwrap(),
        Some(webrtc_bind_addr_cb),
        Some(udp_protect_outgoing),
    )
    .map(|(peer_id, conn), _| (peer_id, StreamMuxerBox::new(conn)))
    .boxed();

    let transport_chain = tcp_transport
        .or_transport(quic_transport)
        .map(|either, _| either.into_inner())
        .or_transport(webrtc_transport)
        .map(
            |either: futures::future::Either<
                (libp2p::PeerId, StreamMuxerBox),
                (libp2p::PeerId, StreamMuxerBox),
            >,
             _| either.into_inner(),
        );

    let dns_transport = libp2p::dns::tokio::Transport::custom(
        transport_chain,
        Default::default(),
        Default::default(),
    );

    let control_handle = Arc::new(ControlHandle::new());
    let snowflake_transport = self::transport::SnowflakeTransport::new(
        control_handle.clone(),
        Some(webrtc_bind_addr_cb),
        Some(udp_protect_outgoing),
    )
    .upgrade(libp2p::core::upgrade::Version::V1Lazy)
    .authenticate(libp2p::noise::Config::new(&config.keypair).unwrap())
    .multiplex(yamux_config.clone())
    .map(|(p, c), _| (p, StreamMuxerBox::new(c)));

    // Try snowflakes first, since we need to intercept /dns4/snowflake
    let outer_transport = snowflake_transport
        .or_transport(dns_transport)
        .map(|either, _| either.into_inner());

    let measured_transport = libp2p::metrics::BandwidthTransport::new(outer_transport, registry)
        .map(|(peer_id, conn), _| (peer_id, StreamMuxerBox::new(conn)));

    let final_transport = measured_transport.boxed();

    let behaviour = SnowstormBehaviour::from(&config);
    let local_peer_id = config.keypair.public().to_peer_id();

    let swarm_config = libp2p::swarm::Config::with_tokio_executor()
        .with_idle_connection_timeout(Duration::from_secs(u64::MAX))
        .with_per_connection_event_buffer_size(128);

    let swarm = libp2p::swarm::Swarm::new(final_transport, behaviour, local_peer_id, swarm_config);

    (swarm, control_handle)
}

fn tcp_protect_outgoing(socket: &Socket) -> io::Result<()> {
    debug!("tcp_protect_outgoing: called");
    let config = CONFIG.lock().unwrap();

    let protect_fn = config.protect_socket_fn.clone();

    #[cfg(not(target_os = "windows"))]
    if let Some(protect_fn) = protect_fn {
        let fd = socket.as_fd().as_raw_fd();
        protect_fn.call(fd);
    }

    Ok(())
}

fn udp_protect_outgoing(socket: &UdpSocket) -> io::Result<()> {
    debug!("udp_protect_outgoing: called");
    let config = CONFIG.lock().unwrap();

    let protect_fn = config.protect_socket_fn.clone();

    #[cfg(target_os = "macos")]
    {
        // Something changed in the new macos, so instead of binding to a specific interface via
        // its address we bind via its index. Might be related to tethering/personal hotspot only.
        let s2 = socket2::SockRef::from(socket);
        let iface = get_default_interface();

        debug!(
            "udp_protect_outgoing: bind {}",
            iface
                .as_ref()
                .map(|iface| iface.name.clone())
                .unwrap_or("none".to_string())
        );
        let if_index = iface.map(|iface| iface.index.try_into().unwrap());

        if let Ok(local_addr) = socket.local_addr() {
            if local_addr.is_ipv4() {
                s2.bind_device_by_index_v4(if_index)?;
            } else {
                s2.bind_device_by_index_v6(if_index)?;
            }
        }
    }

    #[cfg(not(target_os = "windows"))]
    if let Some(protect_fn) = protect_fn {
        let fd = socket.as_raw_fd();
        protect_fn.call(fd);
    }

    Ok(())
}

fn tcp_outgoing_cb(socket: &Socket, dst_addr: &SocketAddr) -> io::Result<()> {
    // Protect the socket (Android)
    let _ = tcp_protect_outgoing(socket);

    let addrs = OUTBOUND_ADDRS.read().clone();

    if addrs.is_empty() {
        return Ok(());
    }

    debug!("tcp_outgoing_cb: {:?}", addrs);

    // Bind the socket to the appropriate outbound interface.
    //
    // WARNING: If we ever decide to reuse ports, this might need
    // to be more involved (though the reuse-port code calls bind,
    // so perhaps not?).

    #[cfg(target_os = "macos")]
    {
        // Something changed in the new macos, so instead of binding to a specific interface via
        // its address we bind via its index. Might be related to tethering/personal hotspot only.
        let iface = get_default_interface();
        debug!(
            "tcp_outgoing_cb: bind {}",
            iface
                .as_ref()
                .map(|iface| iface.name.clone())
                .unwrap_or("none".to_string())
        );
        let if_index = iface.map(|iface| iface.index.try_into().unwrap());

        if dst_addr.is_ipv4() {
            socket.bind_device_by_index_v4(if_index)
        } else {
            socket.bind_device_by_index_v6(if_index)
        }
    }

    #[cfg(not(target_os = "macos"))]
    {
        let bind_addr = outbound_addr_for_dst(addrs, dst_addr)?;
        debug!("tcp_outgoing_cb: bind {}", bind_addr);

        socket.bind(&bind_addr.into())
    }
}

fn quic_eligible_listener_cb(ln_addr: &SocketAddr, dst_addr: &SocketAddr) -> bool {
    let addrs = OUTBOUND_ADDRS.read().clone();

    if addrs.is_empty() {
        return true;
    }

    debug!("quic_eligible_Listener_cb: {} {}", ln_addr, dst_addr);

    match outbound_addr_for_dst(addrs, dst_addr) {
        Ok(addr) => {
            let ok = ln_addr.ip() == addr.ip();
            debug!(
                "quic_eligible_Listener_cb: got addr: {} ?= {} ({})",
                ln_addr, dst_addr, ok
            );

            return ok;
        }
        Err(_) => false,
    }
}

fn quic_outgoing_dial_addr_cb(
    ln_addr: &SocketAddr,
    dst_addr: &SocketAddr,
) -> io::Result<SocketAddr> {
    let addrs = OUTBOUND_ADDRS.read().clone();

    debug!("quic_outgoing_dial_addr_cb: called");

    match addrs.is_empty() {
        true => Ok(ln_addr.clone()),
        false => outbound_addr_for_dst(addrs, dst_addr),
    }
}

fn webrtc_bind_addr_cb(candidate: &SocketAddr) -> io::Result<SocketAddr> {
    let addrs = OUTBOUND_ADDRS.read().clone();

    debug!("webrtc_outgoing_dial_addr_cb: called");

    match addrs.is_empty() {
        true => Ok(candidate.clone()),
        false => outbound_addr_for_dst(addrs, candidate),
    }
}

fn outbound_addr_for_dst(addrs: Vec<IpAddr>, dst_addr: &SocketAddr) -> io::Result<SocketAddr> {
    // Note: This is used by both QUIC and TCP/IP.  The libp2p QUIC
    // transport only supports having one non-listener dialer per AF,
    // which is unfortunate, so this routine consistently returns the
    // first external address for now.
    //
    // As far as I can tell the transport basically will totally freak
    // out if the external address changes mid-way as well.

    let dst_is_ipv4 = dst_addr.is_ipv4();

    for ip in addrs.into_iter() {
        // Pick the first interface that has the same AF.
        if ip.is_ipv4() == dst_is_ipv4 {
            debug!("outbound_addr_for_dst: ({}), {}", dst_addr, ip);
            // Have the OS pick a port for us.
            return Ok(SocketAddr::new(ip, 0));
        }
        if ip.is_ipv6() == !dst_is_ipv4 {
            debug!("outbound_addr_for_dst: ({}), {}", dst_addr, ip);
            // Have the OS pick a port for us.
            return Ok(SocketAddr::new(ip, 0));
        }
    }

    warn!("failed to find outbound address for dest: {}", dst_addr);

    Err(io::Error::new(
        io::ErrorKind::Other,
        "failed to find eligible outbound_addr",
    ))
}

fn multiaddr_to_ipaddr(addr: &Multiaddr) -> io::Result<IpAddr> {
    let mut ma = addr.clone();

    while let Some(proto) = ma.pop() {
        match proto {
            Protocol::Ip4(ip) => {
                return Ok(IpAddr::V4(ip.into()));
            }
            Protocol::Ip6(ip) => {
                return Ok(IpAddr::V6(ip.into()));
            }
            _ => {}
        }
    }

    warn!("failed to parse Multiaddr as IP: {}", addr);

    Err(io::Error::new(
        io::ErrorKind::Other,
        "failed to parse Multiaddr as IpAddr",
    ))
}

pub fn multiaddr_is_public(addr: &Multiaddr) -> bool {
    let ip = match multiaddr_to_ipaddr(addr) {
        Ok(ip) => ip,
        Err(_) => return false,
    };

    match ip {
        IpAddr::V4(ip) => ipv4_is_global(&ip),
        IpAddr::V6(ip) => ipv6_is_global(&ip),
    }
}

pub fn multiaddr_is_loopback(addr: &Multiaddr) -> bool {
    let ip = match multiaddr_to_ipaddr(addr) {
        Ok(ip) => ip,
        Err(_) => return false,
    };

    match ip {
        IpAddr::V4(ip) => ip.is_loopback(),
        IpAddr::V6(ip) => ip.is_loopback(),
    }
}

#[inline]
pub fn ipv4_is_global(ip: &Ipv4Addr) -> bool {
    !(ip.octets()[0] == 0 // "This network"
            || ip.is_private()
            || (ip.octets()[0] == 100 && (ip.octets()[1] & 0b1100_0000 == 0b0100_0000)) //is_shared
            || ip.is_loopback()
            || ip.is_link_local()
            // addresses reserved for future protocols (`192.0.0.0/24`)
            // .9 and .10 are documented as globally reachable so they're excluded
            || (
                ip.octets()[0] == 192 && ip.octets()[1] == 0 && ip.octets()[2] == 0
                && ip.octets()[3] != 9 && ip.octets()[3] != 10
            )
            || ip.is_documentation()
            || (ip.octets()[0] == 198 && (ip.octets()[1] & 0xfe) == 18) //is_benchmarking
            || (ip.octets()[0] & 240 == 240 && !ip.is_broadcast()) // is_reserved
            || ip.is_broadcast())
}

#[inline]
pub fn ipv6_is_global(ip: &Ipv6Addr) -> bool {
    !(ip.is_unspecified()
            || ip.is_loopback()
            // IPv4-mapped Address (`::ffff:0:0/96`)
            || matches!(ip.segments(), [0, 0, 0, 0, 0, 0xffff, _, _])
            // IPv4-IPv6 Translat. (`64:ff9b:1::/48`)
            || matches!(ip.segments(), [0x64, 0xff9b, 1, _, _, _, _, _])
            // Discard-Only Address Block (`100::/64`)
            || matches!(ip.segments(), [0x100, 0, 0, 0, _, _, _, _])
            // IETF Protocol Assignments (`2001::/23`)
            || (matches!(ip.segments(), [0x2001, b, _, _, _, _, _, _] if b < 0x200)
                && !(
                    // Port Control Protocol Anycast (`2001:1::1`)
                    u128::from_be_bytes(ip.octets()) == 0x2001_0001_0000_0000_0000_0000_0000_0001
                    // Traversal Using Relays around NAT Anycast (`2001:1::2`)
                    || u128::from_be_bytes(ip.octets()) == 0x2001_0001_0000_0000_0000_0000_0000_0002
                    // AMT (`2001:3::/32`)
                    || matches!(ip.segments(), [0x2001, 3, _, _, _, _, _, _])
                    // AS112-v6 (`2001:4:112::/48`)
                    || matches!(ip.segments(), [0x2001, 4, 0x112, _, _, _, _, _])
                    // ORCHIDv2 (`2001:20::/28`)
                    // Drone Remote ID Protocol Entity Tags (DETs) Prefix (`2001:30::/28`)`
                    || matches!(ip.segments(), [0x2001, b, _, _, _, _, _, _] if b >= 0x20 && b <= 0x3F)
                ))
            // 6to4 (`2002::/16`) – it's not explicitly documented as globally reachable,
            // IANA says N/A.
            || matches!(ip.segments(), [0x2002, _, _, _, _, _, _, _])
            || ((ip.segments()[0] == 0x2001) && (ip.segments()[1] == 0xdb8)) //documentation
            || ((ip.segments()[0] & 0xfe00) == 0xfc00) //is_unique_local
            || ((ip.segments()[0] & 0xffc0) == 0xfe80)) //is_unicast_link_local
}