snowstorm/net/

mod.rs

pub mod behaviour;
mod discovery;

pub mod connected_exits;
mod connections;
mod control_peer;
pub mod endpoint;
pub mod event;
pub mod metrics;
pub mod netif;
pub mod protocols;
mod roles;
pub mod swarm;

use connected_exits::ConnectedExits;
use connections::ConnectionsMap;
use futures::StreamExt;
use libp2p::metrics::Recorder;
pub use libp2p::metrics::Registry;
use libp2p::swarm::dial_opts::DialOpts;
use std::net::Ipv4Addr;
use std::sync::Mutex;
use tokio_util::sync::CancellationToken;
// use libp2p::core::ConnectedPoint;
use libp2p::multiaddr::Protocol;
use serde::Deserialize;
use std::fmt;
use std::{net::TcpListener, sync::Arc};
use tokio::sync::{broadcast, mpsc};

use log::*;

pub use libp2p::swarm::ConnectionError;
use libp2p::swarm::SwarmEvent;
pub use libp2p::PeerId;

use self::{event::Event, swarm::build_swarm};
use crate::net::control_peer::ControlPeerEvent;
use crate::net::discovery::role_to_namespace;
use crate::{
    config::{Config, Multiaddr},
    control,
    identity::Role,
    net::{
        behaviour::SnowstormBehaviourEvent,
        discovery::{register, search_role},
    },
};

pub type Multihash = multihash::Multihash<64>;
pub const MULTIHASH_IDENTITY_CODE: u64 = 0;

const DEFAULT_TUN_IPV4: Ipv4Addr = Ipv4Addr::new(192, 18, 0, 1);

pub fn get_tun_interface() -> Option<netdev::Interface> {
    netdev::get_interfaces()
        .iter()
        .find(|iface| iface.ipv4.iter().any(|net| net.addr() == DEFAULT_TUN_IPV4))
        .cloned()
}

pub fn get_default_interface() -> Option<netdev::Interface> {
    let mut interfaces = netdev::get_interfaces()
        .into_iter()
        .filter(|iface| {
            // filtering out by name and not .is_tun(), some cell interfaces are point to point.
            let is_tun = iface.name.contains("tun");
            let is_snowstorm = iface.ipv4.iter().any(|net| net.addr() == DEFAULT_TUN_IPV4);
            let is_up = iface.is_up();
            
            // On iOS there's often a bunch of interfaces, some of en* are not meant for internet traffic
            // so for the time being ignoring them manually...
            #[cfg(target_os = "ios")]
            if iface.if_type == netdev::interface::InterfaceType::Ethernet {
                if !(iface.name == "en0" || iface.name == "en1") {
                    return false;
                }
            }

            let has_ipv4 = iface.ipv4.iter().any(|net| !net.addr().is_loopback());
            let has_ipv6 = iface
                .ipv6
                .iter()
                .any(|net| !net.addr().is_unicast_link_local());

            return is_up && !iface.is_loopback() && !is_snowstorm && !is_tun && (has_ipv4 || has_ipv6);
        })
        .collect::<Vec<_>>();

    interfaces.sort_by_key(|iface| {
        // On windows we don't have gateway feature enabled, so not getting info about default iface.
        #[cfg(not(target_os = "windows"))]
        {
            if iface.default {
                return 0;
            }
        }
        
        // Friendly name is useful on windows, since norlam name is a GUID.
        let names = [
            iface.name.to_lowercase(),
            iface.friendly_name.as_ref().map(|s| s.to_lowercase()).unwrap_or_default(),
        ];

        let is_en = iface.if_type == netdev::interface::InterfaceType::Ethernet;
        let is_wifi = iface.if_type == netdev::interface::InterfaceType::Wireless80211;
        let has_wifi_name = names.iter().any(|name| name.contains("wlan") || name.contains("wlp") || name.contains("wi-fi"));
        if is_en || is_wifi || has_wifi_name {
            return 1;
        }

        return 2;
    });

    let names = interfaces.iter().map(|iface| iface.name.clone()).collect::<Vec<_>>();
    let friendly_names = interfaces
        .iter()
        .filter_map(|iface| iface.friendly_name.as_ref())
        .map(|s| s.to_string())
        .collect::<Vec<_>>();
    log::info!("Available interfaces: {:?}", names);
    log::info!("Available friendly interfaces: {:?}", friendly_names);

    return interfaces.first().cloned();
}

pub fn get_outbound_ipv4() -> Option<Ipv4Addr> {
    let default_interface = get_default_interface();

    default_interface
        .and_then(|iface| iface.ipv4.first().cloned())
        .map(|net| net.addr())
}

pub fn get_outbound_ips() -> Vec<Multiaddr> {
    let default_interface = get_default_interface();

    info!("default interface: {:?}", default_interface);

    match default_interface {
        Some(iface) => {
            let mut addrs = Vec::<Multiaddr>::new();
            if let Some(ipv4) = iface.ipv4.first() {
                addrs.push(Multiaddr::from(ipv4.addr()));
            }
            if let Some(ipv6) = iface
                .ipv6
                .into_iter()
                .find(|net| !net.addr().is_unicast_link_local())
            {
                addrs.push(Multiaddr::from(ipv6.addr()));
            }
            addrs
        }
        None => Vec::new(),
    }
}

pub fn get_available_local_port() -> Option<u16> {
    (1024..65535).find(|port| port_is_available(*port))
}

fn port_is_available(port: u16) -> bool {
    match TcpListener::bind(("127.0.0.1", port)) {
        Ok(_) => true,
        Err(_) => false,
    }
}

#[derive(Clone)]
pub struct ProtectSocketFn(pub Arc<dyn Fn(i32) -> () + Send + Sync>);

impl fmt::Debug for ProtectSocketFn {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str("protect_socket_fn")
    }
}

impl ProtectSocketFn {
    pub fn call(&self, fd: i32) {
        (self.0)(fd);
    }
}

pub struct Handle {
    pub control_tx: mpsc::Sender<Event>,
    pub swarm_rx: broadcast::Receiver<Event>,
    pub swarm_tx: broadcast::Sender<Event>,
    pub sserver_port: Option<u16>,
    pub connected_exits: Arc<ConnectedExits>,
}

impl Handle {
    pub fn subscribe(&self) -> broadcast::Receiver<Event> {
        self.swarm_tx.subscribe()
    }
}

// Main entry point into the SDK start method.
pub async fn start(
    config: &Config,
    registry: Arc<Mutex<Registry>>,
) -> Result<Handle, Box<dyn std::error::Error>> {
    let (swarm_tx, swarm_rx) = broadcast::channel::<Event>(8);
    let (control_tx, mut control_rx) = mpsc::channel::<Event>(8);
    let cancellation_token = CancellationToken::new();
    let connections_map = Arc::new(ConnectionsMap::new());
    let is_client = config.roles.contains(&Role::Client);
    let is_exit = config.roles.contains(&Role::Exit);

    let _ = tracing_subscriber::fmt()
        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
        .try_init();

    let mut max_streams: usize = 512;

    // File descriptor limit should be increased to the extent which makes sense.
    // Please verify config fdlimit heuristics
    if config.fdlimit {
        match fdlimit::raise_fd_limit() {
            Ok(fdlimit::Outcome::LimitRaised { from, to }) => {
                info!("raised fd limit from {:?} to {:?}", from, to);
                // TODO: HACK HACK HACK HACK
                //
                // For some reason to was much less than from, and this was causing
                // a panic for a negative unsigned... so we skipped it for now
                //
                // PLEASE VERIFY AND FIX
                println!("raised fd limit from {:?} to {:?}", from, to);
                // max_streams = (to - from) as usize;
                max_streams = to as usize;
            }
            _ => warn!("failed to raise fd limit"),
        }
    }

    let (mut swarm, snowstorm_metrics, swarm_metrics, control_handle) = {
        let mut metrics_registry = registry.lock().unwrap();

        // Create and register our customer metric types
        let metrics = metrics::Metrics::new(&mut metrics_registry);

        // Some built-in libp2p metrics
        let swarm_metrics = libp2p::metrics::Metrics::new(&mut metrics_registry);

        // Build the swarm, enable & register generic bandwidth metrics
        let (s, control_handle) = build_swarm(
            cancellation_token.clone(),
            config.clone(),
            &mut metrics_registry,
            Some(max_streams),
        );

        (s, metrics, swarm_metrics, control_handle)
    };

    // Safe to unwrap for the first time. Afterwards - None.
    let mut control_stream_requests_rx = control_handle.next_stream_request().unwrap();

    let snowstorm_metrics = Arc::new(snowstorm_metrics);

    // Start the listener(s)
    for addr in config.listen_on.iter() {
        warn!("attempting to listen on {:?}", addr);
        let result = swarm.listen_on(addr.clone());
        if let Err(err) = result {
            warn!("Failed to listen on multiaddr {:?}: {:?}", addr, err);
        }
    }

    for addr in config.external_addrs.iter() {
        swarm.add_external_address(addr.clone());
    }

    for addr in config.entry_peers.iter() {
        let dial_result = swarm.dial(addr.clone());

        match dial_result {
            Ok(_) => info!("connected to entry peer: {:?}", addr),
            Err(err) => warn!("failed to dial entry peer: {:?}", err),
        }
    }

    let mut cookie: Option<libp2p::rendezvous::Cookie> = None;

    if is_exit {
        // TODO: Should we do proper error handling here?
        roles::exit::serve(
            &mut swarm,
            &config,
            snowstorm_metrics.clone(),
            cancellation_token.clone(),
            connections_map.clone(),
        );
    }

    let mut sserver_port = None;
    let (connected_exits, mut exit_dials) = ConnectedExits::new(config.token.as_deref().unwrap_or("").to_string(), cancellation_token.clone());
    let connected_exits = Arc::new(connected_exits);
    let connected_exits2 = connected_exits.clone();

    // NOTE(2025.03.04): This was the source of an annoying MacOS panic by directly unwrapping,
    // and has since been updated to not crash, but probably needs much more thinking / reworking.
    if is_client {
        // Client serve can sometimes result in an error, but unwrapping it directly
        // will cause an unnecessary crash. So it must be properly handled...
        match roles::client::serve(
            &mut swarm,
            &config,
            snowstorm_metrics.clone(),
            cancellation_token.clone(),
            connections_map.clone(),
            connected_exits.clone(),
        )
        .await
        {
            Ok(port) => {
                info!("[SDK] client role started on port: {}", port);
                sserver_port = Some(port);
            }
            Err(e) => {
                info!(
                    "[SDK] Error while starting client role - but continuing: {:?}",
                    e
                );
                // TODO: Figure out if we should fix something here??
                // Or update some state somewhere?
            }
        }
    }

    let config = config.clone();
    let swarm_tx_clone = swarm_tx.clone();
    let mut stream_control = swarm.behaviour_mut().stream().unwrap().new_control();
    let mut control_peer = control_peer::ControlPeer::new(stream_control.clone());

    tokio::spawn(async move {
        let config = config.clone();

        loop {
            tokio::select! {
                event = swarm.select_next_some() => {
                    swarm_metrics.record(&event);
                    control_peer.on_swarm_event(&event);

                    // We only care about connected exits if we are a client
                    if is_client {
                        connected_exits.on_swarm_event(&event);
                    }

                    match event {
                        SwarmEvent::ConnectionEstablished { peer_id, endpoint, .. } => {
                            debug!("connection established: {:?}", peer_id);
                            let _ = swarm_tx_clone.send(Event::ConnectionEstablished(peer_id.clone()));
                            connections_map.connection_established(peer_id.clone(), &endpoint);

                            if endpoint.is_listener() {
                                let _ = swarm_tx_clone.send(Event::ListenerConnectionEstablished(peer_id.clone()));
                            } else if endpoint.is_dialer() {
                                let mad = endpoint.get_remote_address();
                                log::info!("dialer connection established: {:?}", mad);
                                let _ = swarm_tx_clone.send(Event::DialerConnectionEstablished(peer_id.clone(), is_snowflake(mad)));
                            }

                            if config.has_entry(&peer_id) {
                                debug!("peer is in entry peer list, sending rendezvous request");
                                register(&config, swarm.behaviour_mut(), peer_id.clone());

                                _ = endpoint;
                                // let multiaddr = match endpoint {
                                //     ConnectedPoint::Dialer { address, .. } => address,
                                //     ConnectedPoint::Listener { send_back_addr, .. } => send_back_addr,
                                // };
                                // swarm.behaviour_mut().autonat().unwrap().add_server(peer_id, Some(multiaddr));
                            }

                            search_role(&config, swarm.behaviour_mut(), Role::Volunteer, peer_id, cookie.clone());
                            search_role(&config, swarm.behaviour_mut(), Role::Exit, peer_id, cookie.clone());
                        },
                        SwarmEvent::ConnectionClosed { peer_id, cause, .. } => {
                            debug!("connection closed: {:?}", peer_id);
                            connections_map.connection_closed(&peer_id);
                            let _ = swarm_tx_clone.send(Event::ConnectionClosed(peer_id, cause.map(Into::into)));
                        },
                        SwarmEvent::NewListenAddr { address, .. } => {
                            debug!("new listen addr: {:?}", address);

                            if let Err(err) = swarm_tx_clone.send(Event::NewListenAddr(address)) {
                                error!("Failed to send event: {:?}", err);
                            }

                            if let Some(token) = &config.token {
                                if !is_client {
                                    let device_id = config.device_id.clone();
                                    let pid = swarm.local_peer_id();
                                    let mas: Vec<&libp2p::Multiaddr> = swarm.external_addresses().map(|addr| addr).collect();
                                    let _ = control::update_multiaddrs(token.to_string(), pid.clone().to_string(), device_id, mas, true).await;
                                }
                            }
                        },
                        SwarmEvent::Behaviour(SnowstormBehaviourEvent::Identify(libp2p::identify::Event::Received { info, .. })) => {
                            debug!("identify observed addr {:?}", info.observed_addr);

                            if let Some(token) = &config.token {
                                // This logic is also triggered on our exits, so we only update based on external observations as a fallback
                                // It should be ok to update addresses like that.
                                if !is_client {
                                    let mut mas: Vec<Multiaddr> = swarm.listeners().map(Clone::clone).collect();
                                    mas.push(info.observed_addr.clone());
                                    let pid = swarm.local_peer_id();
                                    let device_id = config.device_id.clone();
                                    let _ = control::update_multiaddrs(token.to_string(), pid.clone().to_string(), device_id, mas.iter().collect(), false).await;
                                }
                            }
                            let event = Event::NewObservedAddr(info.observed_addr);
                            let _ = swarm_tx_clone.send(event);
                        },
                        SwarmEvent::Behaviour(SnowstormBehaviourEvent::RendezvousClient(libp2p::rendezvous::client::Event::Discovered {
                            registrations,
                            cookie: new_cookie,
                            ..
                        })) => {
                            cookie.replace(new_cookie);
                            debug!("rendezvous discovered: {:?}/{:?}", cookie.clone().unwrap().namespace(), registrations);
                            for registration in registrations {
                                info!("discovered rendezvous peer: {:?}", registration.record.peer_id());
                                let event = match cookie.clone().unwrap().namespace() {
                                    Some(namespace) => {
                                        if namespace.to_string() == role_to_namespace(&config, Role::Exit).to_string() {
                                            Event::ExitDiscovered(registration.record.peer_id().clone())
                                        } else {
                                            Event::Discovered(registration.record.peer_id().clone())
                                        }
                                    },
                                    None => Event::Discovered(registration.record.peer_id().clone()),
                                };

                              let _ = swarm_tx_clone.send(event);
                            }
                        },
                        _ => {}
                    }
                }
                ctrl_event = control_rx.recv() => {
                    match ctrl_event {
                        Some(Event::Stop) => {
                            debug!("stop received, breaking swarm loop");

                            // Cancel sub-tasks
                            cancellation_token.cancel();

                            // Disconnect from all peers.
                            let connected_peers: Vec<PeerId> = swarm.connected_peers().map(Clone::clone).collect();
                            for peer_id in connected_peers {
                                let _ = swarm.disconnect_peer_id(peer_id);
                            }

                            if let Some(token) = &config.token {
                                if !is_client {
                                    let pid = swarm.local_peer_id();
                                    let _ = control::unregister(token, Some(&pid.to_string()), None).await;
                                }
                            }

                            let _ = swarm_tx_clone.send(Event::Stop);
                            break;
                        }
                        _ => {}
                    }
                }

                Some(request) = control_stream_requests_rx.recv() => {
                    info!("ControlHandle: control stream request");
                    request.process(&mut stream_control).await;
                }

                Some((peer_id, addrs)) = exit_dials.recv() => {
                    log::info!("Gotta dial a peer! {:?} {:?}", peer_id, addrs);
                    let dial_opts = DialOpts::peer_id(peer_id).addresses(addrs).build();
                    let _ = swarm.dial(dial_opts);
                }

                event = control_peer.select_next_some() => {
                    match event {
                        ControlPeerEvent::DialNeeded { peer_id, address: mad} => {
                            debug!("control peer id: {:?}", peer_id);

                            let dial_opts = DialOpts::peer_id(peer_id)
                                .addresses(vec![
                                    mad.clone(),
                                ])
                                .build();

                            match swarm.dial(dial_opts) {
                                Ok(_) => {
                                    info!("dialed control peer: {:?}", mad);
                                }
                                Err(err) => {
                                    error!("failed to dial control peer: {:?}", err);
                                }
                            }
                        }
                        ControlPeerEvent::Connected(peer_id) => {
                            control_handle.add_control_peer(peer_id);
                        },
                        ControlPeerEvent::Disconnected(peer_id) => {
                            control_handle.remove_control_peer(peer_id);
                        },
                        ControlPeerEvent::Notification(msg) => {
                            #[derive(Debug, Deserialize)]
                            struct SnowflakeConnectionRequest {
                                #[serde(rename = "ID")]
                                id: String,

                                #[serde(rename = "Offer")]
                                offer: String,
                            }


                            let command = match msg.get("Command") {
                                Some(cmd) => {
                                    cmd.as_i64().unwrap()
                                }
                                None => {
                                    error!("control peer notification missing command");
                                    continue;
                                }
                            };

                            let _ = swarm_tx_clone.send(Event::NotificationReceived(command));

                            if command == 0 {
                                let payload = msg.get("Msg").unwrap().clone();
                                let request = serde_json::from_value::<SnowflakeConnectionRequest>(payload).unwrap();
                                info!("snowflake connection request: {:?}", request);

                                let mad = control_handle.register_offer(&request.id, &request.offer);

                                let r = swarm.listen_on(mad.clone());

                                match r {
                                    Ok(_) => {
                                        info!("listening on snowflake connection request: {:?}", mad);
                                    }
                                    Err(err) => {
                                        error!("failed to listen on snowflake connection request: {:?}", err);
                                    }
                                }
                            }
                        }
                    }
                },
            }
        }
    });

    Ok(Handle {
        control_tx,
        swarm_rx,
        swarm_tx,
        sserver_port,
        connected_exits: connected_exits2,
    })
}

pub async fn stop(tx: mpsc::Sender<Event>) {
    let _ = tx.send(Event::Stop).await;
}

pub fn try_peerid_from_multiaddr(ma: &Multiaddr) -> Option<PeerId> {
    let mut peer_id = None;
    for part in ma.iter() {
        if let Protocol::P2p(id) = part {
            peer_id = Some(id.clone());
            break;
        }
    }
    peer_id
}

pub fn is_snowflake(mad: &Multiaddr) -> bool {
    match mad.iter().next() {
        Some(Protocol::Dns4(hostname)) => {
            hostname.ends_with(".snowflake") || hostname == "snowflake"
        }
        _ => false,
    }
}

pub fn get_snowflake_id(mad: &Multiaddr) -> Option<String> {
    match mad.iter().next() {
        Some(Protocol::Dns4(hostname)) => {
            if hostname.ends_with(".snowflake") {
                let parts: Vec<&str> = hostname.split('.').collect();
                return Some(parts[0].to_string());
            }
        }
        _ => {}
    }

    None
}

pub fn snowflake_enrich_route_id(route_id: Option<&str>, mad: &Multiaddr) -> Multiaddr {
    // If not a snowflake, return the original mad
    if !is_snowflake(mad) {
        return mad.clone();
    }

    match route_id {
        None => return mad.clone(),

        Some(route_id) => {
            let addr = format!("/dns4/{}.snowflake", route_id);

            addr.parse().unwrap()
        }
    }
}