floresta_wire/p2p_wire/node/

conn.rs

// SPDX-License-Identifier: MIT OR Apache-2.0

use core::net::IpAddr;
use core::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;

use bitcoin::p2p::address::AddrV2;
use bitcoin::p2p::ServiceFlags;
use bitcoin::Network;
use floresta_chain::ChainBackend;
use floresta_common::service_flags;
use floresta_common::Ema;
use floresta_mempool::Mempool;
use tokio::net::tcp::WriteHalf;
use tokio::spawn;
use tokio::sync::mpsc::unbounded_channel;
use tokio::sync::mpsc::UnboundedReceiver;
use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::oneshot;
use tokio::sync::Mutex;
use tokio::time::timeout;
use tracing::debug;
use tracing::info;

use super::try_and_log;
use super::ConnectionKind;
use super::InflightRequests;
use super::LocalPeerView;
use super::NodeNotification;
use super::NodeRequest;
use super::PeerStatus;
use super::UtreexoNode;
use crate::address_man::AddressMan;
use crate::address_man::AddressState;
use crate::address_man::LocalAddress;
use crate::node_context::NodeContext;
use crate::p2p_wire::error::AddrParseError;
use crate::p2p_wire::error::WireError;
use crate::p2p_wire::peer::create_actors;
use crate::p2p_wire::peer::Peer;
use crate::p2p_wire::transport;
use crate::TransportProtocol;

/// How long before we consider using alternative ways to find addresses,
/// such as hard-coded peers
const HARDCODED_ADDRESSES_GRACE_PERIOD: Duration = Duration::from_secs(60);

/// The minimum amount of time between address fetching requests from DNS seeds (one hour).
const DNS_SEED_REQUEST_INTERVAL: Duration = Duration::from_secs(60 * 60);

impl<T, Chain> UtreexoNode<Chain, T>
where
    T: 'static + Default + NodeContext,
    Chain: ChainBackend + 'static,
    WireError: From<Chain::Error>,
{
    // === CONNECTION CREATION ===

    /// Create a new outgoing connection, selecting an appropriate peer address.
    ///
    /// If a fixed peer is set via the `--connect` CLI argument, its connection
    /// kind will always be coerced to [`ConnectionKind::Manual`]. Otherwise,
    /// an address is selected from the [`AddressMan`] based on the required
    /// [`ServiceFlags`] for the given `connection_kind`.
    ///
    /// If no address is available and the kind is not [`ConnectionKind::Manual`],
    /// hardcoded addresses are loaded into the [`AddressMan`] as a fallback.
    pub(crate) fn create_connection(
        &mut self,
        mut conn_kind: ConnectionKind,
    ) -> Result<(), WireError> {
        // Set the fixed peer's connection kind to manual, if set.
        if self.fixed_peer.is_some() {
            conn_kind = ConnectionKind::Manual;
        }

        // Get the peer's `ServiceFlags`.
        let required_services = match conn_kind {
            ConnectionKind::Regular(services) => services,
            _ => ServiceFlags::NONE,
        };

        // Get the fixed peer's `peer_id` and `LocalAddress` if set,
        // or fetch a new address from the address manager.
        let candidate_peer = self
            .fixed_peer
            .as_ref()
            .map(|addr| (0, addr.clone()))
            .or_else(|| {
                self.address_man.get_address_to_connect(
                    required_services,
                    matches!(conn_kind, ConnectionKind::Feeler),
                )
            });

        // Load hardcoded addresses to the address manager if no fixed or manual peers exist.
        let Some((peer_id, peer_address)) = candidate_peer else {
            if !matches!(conn_kind, ConnectionKind::Manual) {
                let net = self.network;
                self.address_man.add_fixed_addresses(net);
            }
            return Err(WireError::NoAddressesAvailable);
        };

        debug!("Attempting connection with address={peer_address:?} kind={conn_kind:?}",);

        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();

        // Defaults to failed, if the connection is successful, we'll update the state
        self.address_man
            .update_set_state(peer_id, AddressState::Failed(now));

        // Don't open duplicate connections to the same peer.
        let is_peer_connected = |(_, old_peer): (_, &LocalPeerView)| {
            peer_address.get_net_address() == old_peer.address
                && peer_address.get_port() == old_peer.port
        };
        if self.peers.iter().any(is_peer_connected) {
            return Err(WireError::PeerAlreadyExists(
                peer_address.get_net_address(),
                peer_address.get_port(),
            ));
        }

        // Only allow P2PV1 fallback if the peer's connection kind is manual,
        // or if the `--allow-v1-fallback` CLI argument was set.
        let allow_v1_fallback =
            matches!(conn_kind, ConnectionKind::Manual) || self.config.allow_v1_fallback;

        // Open a connection to the peer.
        self.open_connection(conn_kind, peer_id, peer_address, allow_v1_fallback)?;

        Ok(())
    }

    pub(crate) fn open_feeler_connection(&mut self) -> Result<(), WireError> {
        // No feeler if `--connect` is set
        if self.fixed_peer.is_some() {
            return Ok(());
        }

        for _ in 0..T::NEW_CONNECTIONS_BATCH_SIZE {
            self.create_connection(ConnectionKind::Feeler)?;
        }

        Ok(())
    }

    /// Creates a new outgoing connection with `address`.
    ///
    /// `kind` may or may not be a [`ConnectionKind::Feeler`], a special connection type
    /// that is used to learn about good peers, but are not kept after handshake
    /// (others are [`ConnectionKind::Regular`], [`ConnectionKind::Manual`] and [`ConnectionKind::Extra`]).
    ///
    /// We will always try to open a V2 connection first. If the `allow_v1_fallback` is set,
    /// we may retry the connection with the old V1 protocol if the V2 connection fails.
    /// We don't open the connection here, we create a [`Peer`] actor that will try to open
    /// a connection with the given address and kind. If it succeeds, it will send a
    /// [`PeerMessages::Ready`](crate::p2p_wire::peer::PeerMessages) to the node after handshaking.
    pub(crate) fn open_connection(
        &mut self,
        kind: ConnectionKind,
        peer_id: usize,
        peer_address: LocalAddress,
        allow_v1_fallback: bool,
    ) -> Result<(), WireError> {
        let (requests_tx, requests_rx) = unbounded_channel();
        if let Some(ref proxy) = self.socks5 {
            spawn(timeout(
                Duration::from_secs(10),
                Self::open_proxy_connection(
                    proxy.address,
                    kind,
                    self.mempool.clone(),
                    self.network,
                    self.node_tx.clone(),
                    peer_address.clone(),
                    requests_rx,
                    self.peer_id_count,
                    self.config.user_agent.clone(),
                    self.chain
                        .get_best_block()
                        .expect("infallible in ChainState")
                        .0,
                    allow_v1_fallback,
                ),
            ));
        } else {
            spawn(timeout(
                Duration::from_secs(10),
                Self::open_non_proxy_connection(
                    kind,
                    peer_address.clone(),
                    requests_rx,
                    self.peer_id_count,
                    self.mempool.clone(),
                    self.network,
                    self.node_tx.clone(),
                    self.config.user_agent.clone(),
                    self.chain
                        .get_best_block()
                        .expect("infallible in ChainState")
                        .0,
                    allow_v1_fallback,
                ),
            ));
        }

        let peer_count: u32 = self.peer_id_count;

        self.inflight.insert(
            InflightRequests::Connect(peer_count),
            (peer_count, Instant::now()),
        );

        self.peers.insert(
            peer_count,
            LocalPeerView {
                message_times: Ema::with_half_life_50(),
                address: peer_address.get_net_address(),
                port: peer_address.get_port(),
                user_agent: "".to_string(),
                state: PeerStatus::Awaiting,
                channel: requests_tx,
                services: ServiceFlags::NONE,
                _last_message: Instant::now(),
                kind,
                address_id: peer_id as u32,
                height: 0,
                banscore: 0,
                // Will be downgraded to V1 if the V2 handshake fails, and we allow fallback
                transport_protocol: TransportProtocol::V2,
            },
        );

        match kind {
            ConnectionKind::Feeler => self.last_feeler = Instant::now(),
            ConnectionKind::Regular(_) => self.last_connection = Instant::now(),
            // Note: Creating a manual peer intentionally doesn't affect the `last_connection`
            // timer, since they don't necessarily follow our connection logic, and we may still
            // need more utreexo/CBS peers
            //
            // Extra connections are also not taken into account here because they will probably be
            // short-lived.
            _ => {}
        }

        // Increment peer_id count and the list of peer ids
        // so we can get information about connected or
        // added peers when requesting with getpeerinfo command
        self.peer_id_count += 1;
        Ok(())
    }

    /// Opens a new connection that doesn't require a proxy and includes the functionalities of create_outbound_connection.
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn open_non_proxy_connection(
        kind: ConnectionKind,
        peer_address: LocalAddress,
        requests_rx: UnboundedReceiver<NodeRequest>,
        peer_id_count: u32,
        mempool: Arc<Mutex<Mempool>>,
        network: Network,
        node_tx: UnboundedSender<NodeNotification>,
        our_user_agent: String,
        our_best_block: u32,
        allow_v1_fallback: bool,
    ) -> Result<(), WireError> {
        let (transport_reader, transport_writer, transport_protocol) = transport::connect(
            (peer_address.get_net_address(), peer_address.get_port()),
            network,
            allow_v1_fallback,
        )
        .await?;

        let (cancellation_sender, cancellation_receiver) = oneshot::channel();
        let (actor_receiver, actor) = create_actors(transport_reader);
        tokio::spawn(async move {
            tokio::select! {
                _ = cancellation_receiver => {}
                _ = actor.run() => {}
            }
        });

        // Use create_peer function instead of manually creating the peer
        Peer::<WriteHalf>::create_peer(
            peer_id_count,
            peer_address,
            mempool,
            node_tx.clone(),
            requests_rx,
            kind,
            actor_receiver,
            transport_writer,
            our_user_agent,
            our_best_block,
            cancellation_sender,
            transport_protocol,
        );

        Ok(())
    }

    /// Opens a connection through a socks5 interface
    #[allow(clippy::too_many_arguments)]
    pub(crate) async fn open_proxy_connection(
        proxy: SocketAddr,
        kind: ConnectionKind,
        mempool: Arc<Mutex<Mempool>>,
        network: Network,
        node_tx: UnboundedSender<NodeNotification>,
        peer_address: LocalAddress,
        requests_rx: UnboundedReceiver<NodeRequest>,
        peer_id_count: u32,
        our_user_agent: String,
        our_best_block: u32,
        allow_v1_fallback: bool,
    ) -> Result<(), WireError> {
        let (transport_reader, transport_writer, transport_protocol) =
            transport::connect_proxy(proxy, peer_address.clone(), network, allow_v1_fallback)
                .await?;

        let (cancellation_sender, cancellation_receiver) = oneshot::channel();
        let (actor_receiver, actor) = create_actors(transport_reader);
        tokio::spawn(async move {
            tokio::select! {
                _ = cancellation_receiver => {}
                _ = actor.run() => {}
            }
        });

        Peer::<WriteHalf>::create_peer(
            peer_id_count,
            peer_address,
            mempool,
            node_tx,
            requests_rx,
            kind,
            actor_receiver,
            transport_writer,
            our_user_agent,
            our_best_block,
            cancellation_sender,
            transport_protocol,
        );
        Ok(())
    }

    // === BOOTSTRAPPING ===

    /// Resolves a string address into a LocalAddress
    ///
    /// This function should get an address in the format `<address>[<:port>]` and return a
    /// usable [`LocalAddress`]. It can be an ipv4, ipv6 or a hostname. In case of hostnames,
    /// we resolve them using the system's DNS resolver and return an ip address. Errors if
    /// the provided address is invalid, or we can't resolve it.
    ///
    /// TODO: Allow for non-clearnet addresses like onion services and i2p.
    pub(crate) fn resolve_connect_host(
        address: &str,
        default_port: u16,
    ) -> Result<LocalAddress, AddrParseError> {
        // ipv6
        if address.starts_with('[') {
            if !address.contains(']') {
                return Err(AddrParseError::InvalidIpv6);
            }

            let mut split = address.trim_end().split(']');
            let hostname = split.next().ok_or(AddrParseError::InvalidIpv6)?;
            let port = split
                .next()
                .filter(|x| !x.is_empty())
                .map(|port| {
                    port.trim_start_matches(':')
                        .parse()
                        .map_err(|_e| AddrParseError::InvalidPort)
                })
                .transpose()?
                .unwrap_or(default_port);

            let hostname = hostname.trim_start_matches('[');
            let ip = hostname.parse().map_err(|_e| AddrParseError::InvalidIpv6)?;
            return Ok(LocalAddress::new(
                AddrV2::Ipv6(ip),
                0,
                AddressState::NeverTried,
                ServiceFlags::NONE,
                port,
                rand::random(),
            ));
        }

        // ipv4 - it's hard to differentiate between ipv4 and hostname without an actual regex
        // simply try to parse it as an ip address and if it fails, assume it's a hostname

        // this breaks the necessity of feature gate on windows
        let mut address = address;
        if address.is_empty() {
            address = "127.0.0.1"
        }

        let mut split = address.split(':');
        let ip = split
            .next()
            .ok_or(AddrParseError::InvalidIpv4)?
            .parse()
            .map_err(|_e| AddrParseError::InvalidIpv4);

        match ip {
            Ok(ip) => {
                let port = split
                    .next()
                    .map(|port| port.parse().map_err(|_e| AddrParseError::InvalidPort))
                    .transpose()?
                    .unwrap_or(default_port);

                if split.next().is_some() {
                    return Err(AddrParseError::Inconclusive);
                }

                let id = rand::random();
                Ok(LocalAddress::new(
                    AddrV2::Ipv4(ip),
                    0,
                    AddressState::NeverTried,
                    ServiceFlags::NONE,
                    port,
                    id,
                ))
            }

            Err(_) => {
                let mut split = address.split(':');
                let hostname = split.next().ok_or(AddrParseError::InvalidHostname)?;
                let port = split
                    .next()
                    .map(|port| port.parse().map_err(|_e| AddrParseError::InvalidPort))
                    .transpose()?
                    .unwrap_or(default_port);

                if split.next().is_some() {
                    return Err(AddrParseError::Inconclusive);
                }

                let ip = dns_lookup::lookup_host(hostname)
                    .map_err(|_e| AddrParseError::InvalidHostname)?;
                let id = rand::random();
                let ip = match ip[0] {
                    IpAddr::V4(ip) => AddrV2::Ipv4(ip),
                    IpAddr::V6(ip) => AddrV2::Ipv6(ip),
                };

                Ok(LocalAddress::new(
                    ip,
                    0,
                    AddressState::NeverTried,
                    ServiceFlags::NONE,
                    port,
                    id,
                ))
            }
        }
    }

    // TODO(@luisschwab): get rid of this once
    // https://github.com/rust-bitcoin/rust-bitcoin/pull/4639 makes it into a release.
    pub(crate) fn get_port(network: Network) -> u16 {
        match network {
            Network::Bitcoin => 8333,
            Network::Signet => 38333,
            Network::Testnet => 18333,
            Network::Testnet4 => 48333,
            Network::Regtest => 18444,
        }
    }

    /// Fetch peers from DNS seeds, sending a `NodeNotification` with found ones. Returns
    /// immediately after spawning a background blocking task that performs the work.
    pub(crate) fn get_peers_from_dns(&self) -> Result<(), WireError> {
        let node_sender = self.node_tx.clone();
        let network = self.network;

        let proxy_addr = self.socks5.as_ref().map(|proxy| {
            let addr = proxy.address;
            info!("Asking for DNS peers via the SOCKS5 proxy: {addr}");
            addr
        });

        tokio::task::spawn_blocking(move || {
            let default_port = Self::get_port(network);
            let dns_seeds = floresta_chain::get_chain_dns_seeds(network);

            let mut addresses = Vec::new();
            for seed in &dns_seeds {
                if let Ok(got) = AddressMan::get_seeds_from_dns(seed, default_port, proxy_addr) {
                    addresses.extend(got);
                }
            }

            info!(
                "Fetched {} peer addresses from all DNS seeds",
                addresses.len()
            );

            node_sender
                .send(NodeNotification::DnsSeedAddresses(addresses))
                .unwrap();
        });

        Ok(())
    }

    /// Check whether it's necessary to request more addresses from DNS seeds.
    ///
    /// Perform another address request from DNS seeds if we still don't have enough addresses
    /// on the [`AddressMan`] and the last address request from DNS seeds was over 2 minutes ago.
    fn maybe_ask_dns_seed_for_addresses(&mut self) {
        let enough_addresses = self.address_man.enough_addresses();

        // Skip if address fetching from DNS seeds is disabled,
        // or if the [`AddressMan`] has enough addresses in its database.
        if self.config.disable_dns_seeds || enough_addresses {
            return;
        }

        // Don't ask for peers too often.
        let last_dns_request = self.last_dns_seed_call.elapsed();
        if last_dns_request < DNS_SEED_REQUEST_INTERVAL {
            return;
        }

        self.last_dns_seed_call = Instant::now();

        info!("Floresta has been running for a while without enough addresses, requesting more from DNS seeds");
        try_and_log!(self.get_peers_from_dns());
    }

    /// If we don't have any peers, we use the hardcoded addresses.
    ///
    ///
    /// This is only done if we don't have any peers for a long time, or we
    /// can't find a Utreexo peer in a context we need them. This function
    /// won't do anything if `--connect` was used
    fn maybe_use_hardcoded_addresses(&mut self) {
        if self.fixed_peer.is_some() {
            return;
        }

        if self.used_fixed_addresses {
            return;
        }

        if self.address_man.enough_addresses() {
            return;
        }

        let wait = HARDCODED_ADDRESSES_GRACE_PERIOD;
        if self.startup_time.elapsed() < wait {
            return;
        }

        self.used_fixed_addresses = true;

        info!("No peers found, using hardcoded addresses");
        let net = self.network;
        self.address_man.add_fixed_addresses(net);
    }

    pub(crate) fn init_peers(&mut self) -> Result<(), WireError> {
        let anchors = self.common.address_man.start_addr_man(self.datadir.clone());
        let enough_addresses = self.common.address_man.enough_addresses();

        if !self.config.disable_dns_seeds && !enough_addresses {
            self.get_peers_from_dns()?;
            self.last_dns_seed_call = Instant::now();
        }

        for address in anchors {
            self.open_connection(
                ConnectionKind::Regular(service_flags::UTREEXO.into()),
                address.id,
                address,
                // Using V1 transport fallback as utreexo nodes have limited support
                true,
            )?;
        }

        Ok(())
    }

    // === MAINTENANCE ===

    pub(crate) fn maybe_open_connection(
        &mut self,
        required_service: ServiceFlags,
    ) -> Result<(), WireError> {
        // try to connect with manually added peers
        self.maybe_open_connection_with_added_peers()?;
        if self.connected_peers() >= T::MAX_OUTGOING_PEERS {
            return Ok(());
        }

        let connection_kind = ConnectionKind::Regular(required_service);

        // If the user passes in a `--connect` cli argument, we only connect with
        // that particular peer.
        if self.fixed_peer.is_some() {
            if self.peers.is_empty() {
                self.create_connection(connection_kind)?;
            }
            return Ok(());
        }

        // If we've tried getting some connections, but the addresses we have are not
        // working. Try getting some more addresses from DNS
        self.maybe_ask_dns_seed_for_addresses();
        self.maybe_use_hardcoded_addresses();

        for _ in 0..T::NEW_CONNECTIONS_BATCH_SIZE {
            // Ignore the error so we don't break out of the loop
            let _ = self.create_connection(connection_kind);
        }

        Ok(())
    }

    pub(crate) fn maybe_open_connection_with_added_peers(&mut self) -> Result<(), WireError> {
        if self.added_peers.is_empty() {
            return Ok(());
        }
        let peers_count = self.peer_id_count;
        for added_peer in self.added_peers.clone() {
            let matching_peer = self.peers.values().find(|peer| {
                self.to_addr_v2(peer.address) == added_peer.address && peer.port == added_peer.port
            });

            if matching_peer.is_none() {
                let address = LocalAddress::new(
                    added_peer.address.clone(),
                    0,
                    AddressState::Tried(
                        SystemTime::now()
                            .duration_since(UNIX_EPOCH)
                            .unwrap()
                            .as_secs(),
                    ),
                    ServiceFlags::NONE,
                    added_peer.port,
                    peers_count as usize,
                );

                // Finally, open the connection with the node
                self.open_connection(
                    ConnectionKind::Manual,
                    peers_count as usize,
                    address,
                    added_peer.v1_fallback,
                )?
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use floresta_chain::pruned_utreexo::partial_chain::PartialChainState;

    use crate::node::running_ctx::RunningNode;
    use crate::node::UtreexoNode;

    fn check_address_resolving(address: &str, port: u16, should_succeed: bool, description: &str) {
        let result =
            UtreexoNode::<PartialChainState, RunningNode>::resolve_connect_host(address, port);
        if should_succeed {
            assert!(result.is_ok(), "Failed: {description}");
        } else {
            assert!(result.is_err(), "Unexpected success: {description}");
        }
    }

    #[test]
    fn test_parse_address() {
        // IPv6 Tests
        check_address_resolving("[::1]", 8333, true, "Valid IPv6 without port");
        check_address_resolving("[::1", 8333, false, "Invalid IPv6 format");
        check_address_resolving("[::1]:8333", 8333, true, "Valid IPv6 with port");
        check_address_resolving(
            "[::1]:8333:8333",
            8333,
            false,
            "Invalid IPv6 with multiple ports",
        );

        // IPv4 Tests
        check_address_resolving("127.0.0.1", 8333, true, "Valid IPv4 without port");
        check_address_resolving("321.321.321.321", 8333, false, "Invalid IPv4 format");
        check_address_resolving("127.0.0.1:8333", 8333, true, "Valid IPv4 with port");
        check_address_resolving(
            "127.0.0.1:8333:8333",
            8333,
            false,
            "Invalid IPv4 with multiple ports",
        );

        // Hostname Tests
        check_address_resolving("example.com", 8333, true, "Valid hostname without port");
        check_address_resolving("example", 8333, false, "Invalid hostname");
        check_address_resolving("example.com:8333", 8333, true, "Valid hostname with port");
        check_address_resolving(
            "example.com:8333:8333",
            8333,
            false,
            "Invalid hostname with multiple ports",
        );

        // Edge Cases
        // This could fail on windows but doesnt since inside `resolve_connect_host` we specificate empty addresses as localhost for all OS`s.
        check_address_resolving("", 8333, true, "Empty string address");
        check_address_resolving(
            " 127.0.0.1:8333 ",
            8333,
            false,
            "Address with leading/trailing spaces",
        );
        check_address_resolving("127.0.0.1:0", 0, true, "Valid address with port 0");
        check_address_resolving(
            "127.0.0.1:65535",
            65535,
            true,
            "Valid address with maximum port",
        );
        check_address_resolving(
            "127.0.0.1:65536",
            65535,
            false,
            "Valid address with out-of-range port",
        )
    }
}