hopr_transport/

config.rs

use std::{
    fmt::{Display, Formatter},
    net::ToSocketAddrs,
    num::ParseIntError,
    str::FromStr,
    time::Duration,
};

use hopr_api::Multiaddr;
pub use hopr_protocol_hopr::{HoprCodecConfig, HoprUnacknowledgedTicketProcessorConfig, SurbStoreConfig};
pub use hopr_transport_mixer::config::MixerConfig;
pub use hopr_transport_probe::config::ProbeConfig;
use hopr_transport_session::{MIN_BALANCER_SAMPLING_INTERVAL, MIN_SURB_BUFFER_DURATION};
use proc_macro_regex::regex;
use validator::{Validate, ValidationError, ValidationErrors};

use crate::{errors::HoprTransportError, protocol::PacketPipelineConfig};

const DEFAULT_COUNTER_FLUSH_INTERVAL: Duration = Duration::from_secs(15);

const DEFAULT_PER_PEER_CHANNEL_CAPACITY: usize = 5_000;

fn default_per_peer_channel_capacity() -> usize {
    DEFAULT_PER_PEER_CHANNEL_CAPACITY
}

/// Configuration of the per-peer egress stream layer.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Validate, smart_default::SmartDefault)]
#[cfg_attr(
    feature = "serde",
    derive(serde::Serialize, serde::Deserialize),
    serde(deny_unknown_fields)
)]
pub struct StreamProtocolConfig {
    /// Capacity of the per-peer outgoing mpsc channel in packets.
    ///
    /// Sized to absorb a typical SURB pre-fill burst (default SurbBalancer:
    /// target 7 000 / max 5 000/s) so the 50 ms per-peer send timeout is
    /// only reached under genuine sustained overload, not on normal pre-fill
    /// traffic. Packets that still exceed the timeout are dropped as
    /// intentional transport drops.
    ///
    /// Defaults to 5 000.
    #[validate(range(min = 1))]
    #[default(default_per_peer_channel_capacity())]
    #[cfg_attr(feature = "serde", serde(default = "default_per_peer_channel_capacity"))]
    pub per_peer_channel_capacity: usize,
}

fn default_counter_flush_interval() -> Duration {
    DEFAULT_COUNTER_FLUSH_INTERVAL
}

/// Complete configuration of the HOPR protocol stack.
#[derive(Debug, smart_default::SmartDefault, Validate, Clone, PartialEq)]
#[cfg_attr(
    feature = "serde",
    derive(serde::Serialize, serde::Deserialize),
    serde(deny_unknown_fields)
)]
pub struct HoprProtocolConfig {
    /// Libp2p-related transport configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub transport: TransportConfig,
    /// HOPR packet pipeline configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub packet: HoprPacketPipelineConfig,
    /// Probing protocol configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub probe: ProbeConfig,
    /// Session protocol global configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub session: SessionGlobalConfig,
    /// Mixer configuration.
    #[cfg_attr(feature = "serde", serde(default))]
    pub mixer: MixerConfig,
    /// Per-peer egress stream configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub stream: StreamProtocolConfig,
    /// Path planner configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(skip))]
    pub path_planner: crate::path::PathPlannerConfig,
    /// Interval at which per-peer protocol conformance counters are flushed
    /// into the network graph.
    ///
    /// Default is 15 seconds.
    #[default(default_counter_flush_interval())]
    #[cfg_attr(
        feature = "serde",
        serde(default = "default_counter_flush_interval", with = "humantime_serde")
    )]
    pub counter_flush_interval: Duration,
}

/// Configuration of the HOPR packet pipeline.
#[derive(Clone, Copy, Debug, PartialEq, Validate, smart_default::SmartDefault)]
#[cfg_attr(
    feature = "serde",
    derive(serde::Serialize, serde::Deserialize),
    serde(deny_unknown_fields)
)]
pub struct HoprPacketPipelineConfig {
    /// HOPR packet codec configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub codec: HoprCodecConfig,
    /// Configuration of unacknowledged tickets processing.
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub ack_processor: HoprUnacknowledgedTicketProcessorConfig,
    /// Single Use Reply Block (SURB) handling configuration
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub surb_store: SurbStoreConfig,
    /// Packet pipeline configuration controlling output/input concurrency and acknowledgement processing
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub pipeline: PacketPipelineConfig,
}

regex!(is_dns_address_regex "^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\\.)*[a-z0-9][a-z0-9-]{0,61}[a-z0-9]$");

/// Check whether the string looks like a valid domain.
#[inline]
pub fn looks_like_domain(s: &str) -> bool {
    is_dns_address_regex(s)
}

/// Check whether the string is an actual reachable domain.
pub fn is_reachable_domain(host: &str) -> bool {
    host.to_socket_addrs().is_ok_and(|i| i.into_iter().next().is_some())
}

/// Enumeration of possible host types.
#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub enum HostType {
    /// IPv4 based host
    IPv4(String),
    /// DNS based host
    Domain(String),
}

impl validator::Validate for HostType {
    fn validate(&self) -> Result<(), ValidationErrors> {
        match &self {
            HostType::IPv4(ip4) => validate_ipv4_address(ip4).map_err(|e| {
                let mut errs = ValidationErrors::new();
                errs.add("ipv4", e);
                errs
            }),
            HostType::Domain(domain) => validate_dns_address(domain).map_err(|e| {
                let mut errs = ValidationErrors::new();
                errs.add("domain", e);
                errs
            }),
        }
    }
}

impl Default for HostType {
    fn default() -> Self {
        HostType::IPv4("127.0.0.1".to_owned())
    }
}

/// Configuration of the listening host.
///
/// This is used for the P2P and REST API listeners.
///
/// Intentionally has no default because it depends on the use case.
#[derive(Debug, Validate, Clone, PartialEq)]
#[cfg_attr(
    feature = "serde",
    derive(serde::Serialize, serde::Deserialize),
    serde(deny_unknown_fields)
)]
pub struct HostConfig {
    /// Host on which to listen
    #[cfg_attr(feature = "serde", serde(default))]
    pub address: HostType,
    /// Listening TCP or UDP port (mandatory).
    #[validate(range(min = 1u16))]
    #[cfg_attr(feature = "serde", serde(default))]
    pub port: u16,
}

impl FromStr for HostConfig {
    type Err = String;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let (ip_or_dns, str_port) = match s.split_once(':') {
            None => return Err("Invalid host, is not in the '<host>:<port>' format".into()),
            Some(split) => split,
        };

        let port = str_port.parse().map_err(|e: ParseIntError| e.to_string())?;

        if validator::ValidateIp::validate_ipv4(&ip_or_dns) {
            Ok(Self {
                address: HostType::IPv4(ip_or_dns.to_owned()),
                port,
            })
        } else if looks_like_domain(ip_or_dns) {
            Ok(Self {
                address: HostType::Domain(ip_or_dns.to_owned()),
                port,
            })
        } else {
            Err("Not a valid IPv4 or domain host".into())
        }
    }
}

impl Display for HostConfig {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "{:?}:{}", self.address, self.port)
    }
}

fn default_multiaddr_transport(port: u16) -> String {
    cfg_if::cfg_if! {
        if #[cfg(feature = "p2p-announce-quic")] {
            // In case we run on a Dappnode-like device, presumably behind NAT, we fall back to TCP
            // to circumvent issues with QUIC in such environments. To make this work reliably,
            // we would need proper NAT traversal support.
            let on_dappnode = std::env::var("DAPPNODE")
                .map(|v| v.to_lowercase() == "true")
                .unwrap_or(false);

            // Using HOPRD_NAT a user can overwrite the default behaviour even on a Dappnode-like device
            let uses_nat = std::env::var("HOPRD_NAT")
                .map(|v| v.to_lowercase() == "true")
                .unwrap_or(on_dappnode);

            if uses_nat {
                format!("tcp/{port}")
            } else {
                format!("udp/{port}/quic-v1")
            }
        } else {
            format!("tcp/{port}")
        }
    }
}

impl TryFrom<&HostConfig> for Multiaddr {
    type Error = HoprTransportError;

    fn try_from(value: &HostConfig) -> Result<Self, Self::Error> {
        match &value.address {
            HostType::IPv4(ip) => Multiaddr::from_str(
                format!("/ip4/{}/{}", ip.as_str(), default_multiaddr_transport(value.port)).as_str(),
            )
            .map_err(|e| HoprTransportError::Api(e.to_string())),
            HostType::Domain(domain) => Multiaddr::from_str(
                format!("/dns4/{}/{}", domain.as_str(), default_multiaddr_transport(value.port)).as_str(),
            )
            .map_err(|e| HoprTransportError::Api(e.to_string())),
        }
    }
}

fn validate_ipv4_address(s: &str) -> Result<(), ValidationError> {
    if validator::ValidateIp::validate_ipv4(&s) {
        let ipv4 = std::net::Ipv4Addr::from_str(s)
            .map_err(|_| ValidationError::new("Failed to deserialize the string into an ipv4 address"))?;

        if ipv4.is_private() || ipv4.is_multicast() || ipv4.is_unspecified() {
            return Err(ValidationError::new(
                "IPv4 cannot be private, multicast or unspecified (0.0.0.0)",
            ))?;
        }
        Ok(())
    } else {
        Err(ValidationError::new("Invalid IPv4 address provided"))
    }
}

fn validate_dns_address(s: &str) -> Result<(), ValidationError> {
    if looks_like_domain(s) || is_reachable_domain(s) {
        Ok(())
    } else {
        Err(ValidationError::new("Invalid DNS address provided"))
    }
}

/// Configuration of the physical transport mechanism.
#[derive(Debug, Default, Validate, Clone, Copy, PartialEq)]
#[cfg_attr(
    feature = "serde",
    derive(serde::Serialize, serde::Deserialize),
    serde(deny_unknown_fields)
)]
pub struct TransportConfig {
    /// When true, assume that the node is running in an isolated network and does
    /// not need any connection to nodes outside the subnet
    #[cfg_attr(feature = "serde", serde(default))]
    pub announce_local_addresses: bool,
    /// When true, assume a testnet with multiple nodes running on the same machine
    /// or in the same private IPv4 network
    #[cfg_attr(feature = "serde", serde(default))]
    pub prefer_local_addresses: bool,
}

const DEFAULT_SESSION_IDLE_TIMEOUT: Duration = Duration::from_mins(3);

const SESSION_IDLE_MIN_TIMEOUT: Duration = Duration::from_secs(2);

const DEFAULT_SESSION_ESTABLISH_RETRY_DELAY: Duration = Duration::from_secs(2);

const DEFAULT_SESSION_ESTABLISH_MAX_RETRIES: usize = 3;

const DEFAULT_SESSION_BALANCER_SAMPLING: Duration = Duration::from_millis(100);

const DEFAULT_SESSION_BALANCER_BUFFER_DURATION: Duration = Duration::from_secs(5);

const DEFAULT_MAXIMUM_MANAGED_SESSIONS: usize = 100;

fn default_session_balancer_buffer_duration() -> Duration {
    DEFAULT_SESSION_BALANCER_BUFFER_DURATION
}

fn default_session_establish_max_retries() -> usize {
    DEFAULT_SESSION_ESTABLISH_MAX_RETRIES
}

fn default_session_idle_timeout() -> Duration {
    DEFAULT_SESSION_IDLE_TIMEOUT
}

fn default_session_establish_retry_delay() -> Duration {
    DEFAULT_SESSION_ESTABLISH_RETRY_DELAY
}

fn default_session_balancer_sampling() -> Duration {
    DEFAULT_SESSION_BALANCER_SAMPLING
}

fn default_max_managed_sessions() -> usize {
    DEFAULT_MAXIMUM_MANAGED_SESSIONS
}

fn validate_session_idle_timeout(value: &Duration) -> Result<(), ValidationError> {
    if SESSION_IDLE_MIN_TIMEOUT <= *value {
        Ok(())
    } else {
        Err(ValidationError::new("session idle timeout is too low"))
    }
}

fn validate_balancer_sampling(value: &Duration) -> Result<(), ValidationError> {
    if MIN_BALANCER_SAMPLING_INTERVAL <= *value {
        Ok(())
    } else {
        Err(ValidationError::new("balancer sampling interval is too low"))
    }
}

fn validate_balancer_buffer_duration(value: &Duration) -> Result<(), ValidationError> {
    if MIN_SURB_BUFFER_DURATION <= *value {
        Ok(())
    } else {
        Err(ValidationError::new("minmum SURB buffer duration is too low"))
    }
}

/// Global configuration of Sessions and the Session manager.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Validate, smart_default::SmartDefault)]
#[cfg_attr(
    feature = "serde",
    derive(serde::Serialize, serde::Deserialize),
    serde(deny_unknown_fields)
)]
pub struct SessionGlobalConfig {
    /// Maximum time before an idle Session is closed.
    ///
    /// Defaults to 3 minutes.
    #[validate(custom(function = "validate_session_idle_timeout"))]
    #[default(default_session_idle_timeout())]
    #[cfg_attr(
        feature = "serde",
        serde(default = "default_session_idle_timeout", with = "humantime_serde")
    )]
    pub idle_timeout: Duration,

    /// Maximum number of Sessions that can be managed by the Session manager.
    ///
    /// Default is 1000, minimum is 2, maximum is 100 000.
    #[validate(range(min = 2, max = 100_000))]
    #[default(default_max_managed_sessions())]
    #[cfg_attr(feature = "serde", serde(default = "default_max_managed_sessions"))]
    pub maximum_managed_sessions: usize,

    /// Maximum retries to attempt to establish the Session
    /// Set 0 for no retries.
    ///
    /// Defaults to 3, maximum is 20.
    #[validate(range(min = 0, max = 20))]
    #[default(default_session_establish_max_retries())]
    #[cfg_attr(feature = "serde", serde(default = "default_session_establish_max_retries"))]
    pub establish_max_retries: usize,

    /// Delay between Session establishment retries.
    ///
    /// Default is 2 seconds.
    #[default(default_session_establish_retry_delay())]
    #[cfg_attr(
        feature = "serde",
        serde(default = "default_session_establish_retry_delay", with = "humantime_serde")
    )]
    pub establish_retry_timeout: Duration,

    /// Sampling interval for SURB balancer in milliseconds.
    ///
    /// Default is 100 milliseconds.
    #[validate(custom(function = "validate_balancer_sampling"))]
    #[default(default_session_balancer_sampling())]
    #[cfg_attr(
        feature = "serde",
        serde(default = "default_session_balancer_sampling", with = "humantime_serde")
    )]
    pub balancer_sampling_interval: Duration,

    /// Minimum runway of received SURBs in seconds.
    ///
    /// This applies to incoming Sessions on Exit nodes only and is the main indicator of how
    /// the egress traffic will be shaped, unless the `NoRateControl` Session
    /// capability is specified during initiation.
    ///
    /// Default is 5 seconds, minimum is 1 second.
    #[validate(custom(function = "validate_balancer_buffer_duration"))]
    #[default(default_session_balancer_buffer_duration())]
    #[cfg_attr(
        feature = "serde",
        serde(default = "default_session_balancer_buffer_duration", with = "humantime_serde")
    )]
    pub balancer_minimum_surb_buffer_duration: Duration,

    /// Tag allocator partition configuration.
    #[validate(nested)]
    #[cfg_attr(feature = "serde", serde(default))]
    pub tag_allocator: hopr_transport_tag_allocator::TagAllocatorConfig,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_valid_domains_for_looks_like_a_domain() {
        assert!(looks_like_domain("localhost"));
        assert!(looks_like_domain("hoprnet.org"));
        assert!(looks_like_domain("hub.hoprnet.org"));
    }

    #[test]
    fn test_valid_domains_for_does_not_look_like_a_domain() {
        assert!(!looks_like_domain(".org"));
        assert!(!looks_like_domain("-hoprnet-.org"));
    }

    #[test]
    fn test_valid_domains_should_be_reachable() {
        assert!(!is_reachable_domain("google.com"));
    }

    #[test]
    fn test_verify_valid_ip4_addresses() {
        assert!(validate_ipv4_address("1.1.1.1").is_ok());
        assert!(validate_ipv4_address("1.255.1.1").is_ok());
        assert!(validate_ipv4_address("187.1.1.255").is_ok());
        assert!(validate_ipv4_address("127.0.0.1").is_ok());
    }

    #[test]
    fn test_verify_invalid_ip4_addresses() {
        assert!(validate_ipv4_address("1.256.1.1").is_err());
        assert!(validate_ipv4_address("-1.1.1.255").is_err());
        assert!(validate_ipv4_address("127.0.0.256").is_err());
        assert!(validate_ipv4_address("1").is_err());
        assert!(validate_ipv4_address("1.1").is_err());
        assert!(validate_ipv4_address("1.1.1").is_err());
        assert!(validate_ipv4_address("1.1.1.1.1").is_err());
    }

    #[test]
    fn test_verify_valid_dns_addresses() {
        assert!(validate_dns_address("localhost").is_ok());
        assert!(validate_dns_address("google.com").is_ok());
        assert!(validate_dns_address("hub.hoprnet.org").is_ok());
    }

    #[test]
    fn test_verify_invalid_dns_addresses() {
        assert!(validate_dns_address("-hoprnet-.org").is_err());
    }

    #[test]
    fn test_multiaddress_on_dappnode_default() {
        temp_env::with_var("DAPPNODE", Some("true"), || {
            assert_eq!(default_multiaddr_transport(1234), "tcp/1234");
        });
    }

    #[cfg(feature = "p2p-announce-quic")]
    #[test]
    fn test_multiaddress_on_non_dappnode_default() {
        temp_env::with_vars([("DAPPNODE", Some("false")), ("HOPRD_NAT", Some("false"))], || {
            assert_eq!(default_multiaddr_transport(1234), "udp/1234/quic-v1");
        });
    }

    #[cfg(not(feature = "p2p-announce-quic"))]
    #[test]
    fn test_multiaddress_on_non_dappnode_default() {
        assert_eq!(default_multiaddr_transport(1234), "tcp/1234");
    }

    #[test]
    fn test_multiaddress_on_non_dappnode_uses_nat() {
        temp_env::with_var("HOPRD_NAT", Some("true"), || {
            assert_eq!(default_multiaddr_transport(1234), "tcp/1234");
        });
    }

    #[cfg(feature = "p2p-announce-quic")]
    #[test]
    fn test_multiaddress_on_non_dappnode_not_uses_nat() {
        temp_env::with_var("HOPRD_NAT", Some("false"), || {
            assert_eq!(default_multiaddr_transport(1234), "udp/1234/quic-v1");
        });
    }

    #[cfg(not(feature = "p2p-announce-quic"))]
    #[test]
    fn test_multiaddress_on_non_dappnode_not_uses_nat() {
        temp_env::with_var("HOPRD_NAT", Some("false"), || {
            assert_eq!(default_multiaddr_transport(1234), "tcp/1234");
        });
    }

    #[cfg(feature = "p2p-announce-quic")]
    #[test]
    fn test_multiaddress_on_dappnode_not_uses_nat() {
        temp_env::with_vars([("DAPPNODE", Some("true")), ("HOPRD_NAT", Some("false"))], || {
            assert_eq!(default_multiaddr_transport(1234), "udp/1234/quic-v1");
        });
    }

    #[cfg(not(feature = "p2p-announce-quic"))]
    #[test]
    fn test_multiaddress_on_dappnode_not_uses_nat() {
        temp_env::with_vars([("DAPPNODE", Some("true")), ("HOPRD_NAT", Some("false"))], || {
            assert_eq!(default_multiaddr_transport(1234), "tcp/1234");
        });
    }

    // --- HostConfig::FromStr tests ---

    #[test]
    fn host_config_parses_ipv4_address() {
        let cfg = HostConfig::from_str("1.2.3.4:9091").unwrap();
        insta::assert_debug_snapshot!(cfg);
    }

    #[test]
    fn host_config_parses_domain() {
        let cfg = HostConfig::from_str("example.com:443").unwrap();
        insta::assert_debug_snapshot!(cfg);
    }

    #[test]
    fn host_config_rejects_missing_port() {
        assert!(HostConfig::from_str("1.2.3.4").is_err());
    }

    #[test]
    fn host_config_rejects_invalid_port() {
        assert!(HostConfig::from_str("1.2.3.4:abc").is_err());
    }

    #[test]
    fn host_config_rejects_invalid_host() {
        assert!(HostConfig::from_str("-invalid-.com:80").is_err());
    }

    #[test]
    fn host_config_display_roundtrip() {
        let cfg = HostConfig {
            address: HostType::IPv4("10.0.0.1".into()),
            port: 8080,
        };
        insta::assert_yaml_snapshot!(cfg.to_string());
    }

    // --- TryFrom<&HostConfig> for Multiaddr tests ---

    #[test]
    fn multiaddr_from_ipv4_host_config() {
        let cfg = HostConfig {
            address: HostType::IPv4("1.2.3.4".into()),
            port: 9091,
        };
        let addr = Multiaddr::try_from(&cfg).unwrap();
        insta::assert_yaml_snapshot!(addr.to_string());
    }

    #[test]
    fn multiaddr_from_domain_host_config() {
        let cfg = HostConfig {
            address: HostType::Domain("example.com".into()),
            port: 443,
        };
        let addr = Multiaddr::try_from(&cfg).unwrap();
        insta::assert_yaml_snapshot!(addr.to_string());
    }

    // --- SessionGlobalConfig validation tests ---

    #[test]
    fn session_global_config_default_is_valid() {
        let cfg = SessionGlobalConfig::default();
        assert!(cfg.validate().is_ok());
    }

    #[test]
    fn session_global_config_too_low_idle_timeout_is_rejected() {
        let cfg = SessionGlobalConfig {
            idle_timeout: Duration::from_millis(100),
            ..Default::default()
        };
        assert!(cfg.validate().is_err());
    }

    #[test]
    fn session_global_config_too_many_retries_is_rejected() {
        let cfg = SessionGlobalConfig {
            establish_max_retries: 21,
            ..Default::default()
        };
        assert!(cfg.validate().is_err());
    }

    #[test]
    fn stream_protocol_config_default_has_expected_capacity() {
        let cfg = StreamProtocolConfig::default();
        assert_eq!(cfg.per_peer_channel_capacity, DEFAULT_PER_PEER_CHANNEL_CAPACITY);
        cfg.validate().expect("default StreamProtocolConfig must be valid");
    }

    #[test]
    fn stream_protocol_config_zero_capacity_is_rejected() {
        let cfg = StreamProtocolConfig {
            per_peer_channel_capacity: 0,
        };
        assert!(cfg.validate().is_err());
    }
}