p2pnet: outbound transports configuration implemented

bin/darkfid/darkfid_config.toml

@@ -45,6 +45,9 @@
 # Peers JSON-RPC listen URL for clock synchronization
 #consensus_peer_rpc = []
 
+# Prefered transports of outbound connections for the consensus protocol
+#consensus_p2p_transports = ["tls", "tcp"]
+
 # P2P accept addresses for the syncing protocol
 #sync_p2p_accept = ["tls://127.0.0.1:8342"]
 
@@ -60,6 +63,9 @@
 # Peers to connect to for the syncing protocol
 #sync_p2p_peer = []
 
+# Prefered transports of outbound connections for the syncing protocol
+#sync_p2p_transports = ["tls", "tcp"]
+
 # Whitelisted cashier addresses
 #cashier_pub = []
 

bin/darkfid/src/main.rs

@@ -109,6 +109,10 @@ struct Args {
     /// Seed nodes JSON-RPC listen URL for clock synchronization (repeatable flag)
     consensus_seed_rpc: Vec<Url>,
 
+    #[structopt(long)]
+    /// Prefered transports of outbound connections for the consensus protocol (repeatable flag)
+    consensus_p2p_transports: Vec<String>,
+
     #[structopt(long)]
     /// P2P accept addresses for the syncing protocol (repeatable flag)
     sync_p2p_accept: Vec<Url>,
@@ -129,6 +133,10 @@ struct Args {
     /// Connect to seed for the syncing protocol (repeatable flag)
     sync_p2p_seed: Vec<Url>,
 
+    #[structopt(long)]
+    /// Prefered transports of outbound connections for the syncing protocol (repeatable flag)
+    sync_p2p_transports: Vec<String>,
+
     #[structopt(long)]
     /// Whitelisted cashier address (repeatable flag)
     cashier_pub: Vec<String>,
@@ -298,6 +306,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
             external_addr: args.sync_p2p_external,
             peers: args.sync_p2p_peer.clone(),
             seeds: args.sync_p2p_seed.clone(),
+            outbound_transports: net::settings::get_outbound_transports(args.sync_p2p_transports),
             ..Default::default()
         };
 
@@ -339,6 +348,9 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
                 external_addr: args.consensus_p2p_external,
                 peers: args.consensus_p2p_peer.clone(),
                 seeds: args.consensus_p2p_seed.clone(),
+                outbound_transports: net::settings::get_outbound_transports(
+                    args.consensus_p2p_transports,
+                ),
                 ..Default::default()
             };
             let p2p = net::P2p::new(consensus_network_settings).await;

bin/darkwikid/darkwiki.toml

@@ -27,6 +27,9 @@ outbound_connections=5
 ## Seed nodes to connect to
 seeds = ["tls://wiki0.dark.fi:24331", "tls://wiki1.dark.fi:24331"]
 
+# Prefered transports for outbound connections
+#transports = ["tls", "tcp"]
+
 ## these are the default configuration for the p2p network
 #manual_attempt_limit=0
 #seed_query_timeout_seconds=8

bin/faucetd/faucetd_config.toml

@@ -31,10 +31,13 @@
 #sync_slots = 8
 
 # Seed nodes to connect to for the syncing protocol
-#sync_seed = []
+#sync_p2p_seed = []
 
 # Peers to connect to for the syncing protocol
-#sync_peer = []
+#sync_p2p_peer = []
+
+# Prefered transports of outbound connections for the syncing protocol
+#sync_p2p_transports = ["tls", "tcp"]
 
 # Airdrop timeout limit in seconds
 #airdrop_timeout = 600

bin/faucetd/src/main.rs

@@ -96,6 +96,10 @@ struct Args {
     /// Connect to peer for the syncing protocol (repeatable flag)
     sync_p2p_peer: Vec<Url>,
 
+    #[structopt(long)]
+    /// Prefered transports of outbound connections for the syncing protocol (repeatable flag)
+    sync_p2p_transports: Vec<String>,
+
     #[structopt(long)]
     /// Whitelisted cashier address (repeatable flag)
     cashier_pub: Vec<String>,
@@ -360,6 +364,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
         external_addr: args.sync_p2p_external,
         peers: args.sync_p2p_peer.clone(),
         seeds: args.sync_p2p_seed.clone(),
+        outbound_transports: net::settings::get_outbound_transports(args.sync_p2p_transports),
         ..Default::default()
     };
 

bin/irc-raft/ircd_config.toml

@@ -27,6 +27,9 @@ outbound_connections=5
 ## Seed nodes to connect to 
 seeds=["tls://irc0.dark.fi:11001", "tls://irc1.dark.fi:11001"]
 
+# Prefered transports for outbound connections
+#transports = ["tls", "tcp"]
+
 ## these are the default configuration for the p2p network
 #manual_attempt_limit=0
 #seed_query_timeout_seconds=8

bin/ircd/ircd_config.toml

@@ -42,6 +42,9 @@ outbound_connections=5
 ## Seed nodes to connect to 
 seeds = ["tls://lilith0.dark.fi:25551", "tls://lilith1.dark.fi:25551"]
 
+# Prefered transports for outbound connections
+#transports = ["tls", "tcp"]
+
 ## Only used for debugging. Compromises privacy when set.
 #node_id = "foo"
 

bin/tau/taud_config.toml

@@ -24,6 +24,9 @@ outbound_connections=5
 ## Seed nodes to connect to 
 seeds=["tls://lilith0.dark.fi:23331", "tls://lilith1.dark.fi:23331"]
 
+# Prefered transports for outbound connections
+#transports = ["tls", "tcp"]
+
 ## these are the default configuration for the p2p network
 #manual_attempt_limit=0
 #seed_query_timeout_seconds=8

contrib/localnet/darkfid0.toml

@@ -39,6 +39,10 @@ consensus_p2p_seed = ["tcp://127.0.0.1:33033"]
 # Peers to connect to for the consensus protocol
 #consensus_p2p_peer = []
 
+
+# Prefered transports of outbound connections for the consensus protocol
+#consensus_p2p_transports = ["tls", "tcp"]
+
 # P2P accept addresses for the syncing protocol
 sync_p2p_accept = ["tls://127.0.0.1:8342"]
 
@@ -54,6 +58,9 @@ sync_p2p_seed = ["tcp://127.0.0.1:33032"]
 # Peers to connect to for the syncing protocol
 #sync_p2p_peer = []
 
+# Prefered transports of outbound connections for the syncing protocol
+#sync_p2p_transports = ["tls", "tcp"]
+
 # Whitelisted faucet public key
 faucet_pub = ["1z3JuN456MieHtpoF4cr5a1ntFF87L8Z3m2E5ACppFodFcjnLS"]
 

contrib/localnet/darkfid1.toml

@@ -45,6 +45,9 @@ consensus_seed_rpc = ["tcp://127.0.0.1:8340"]
 # Peers JSON-RPC listen URL for clock synchronization
 #consensus_peer_rpc = []
 
+# Prefered transports of outbound connections for the consensus protocol
+#consensus_p2p_transports = ["tls", "tcp"]
+
 # P2P accept addresses for the syncing protocol
 sync_p2p_accept = ["tls://127.0.0.1:8442"]
 
@@ -60,6 +63,9 @@ sync_p2p_seed = ["tcp://127.0.0.1:33032"]
 # Peers to connect to for the syncing protocol
 #sync_p2p_peer = []
 
+# Prefered transports of outbound connections for the syncing protocol
+#sync_p2p_transports = ["tls", "tcp"]
+
 # Whitelisted faucet public key
 faucet_pub = ["1z3JuN456MieHtpoF4cr5a1ntFF87L8Z3m2E5ACppFodFcjnLS"]
 

contrib/localnet/darkfid2.toml

@@ -39,6 +39,9 @@ consensus = false
 # Peers to connect to for the consensus protocol
 #consensus_p2p_peer = []
 
+# Prefered transports of outbound connections for the consensus protocol
+#consensus_p2p_transports = ["tls", "tcp"]
+
 # P2P accept addresses for the syncing protocol
 sync_p2p_accept = ["tls://127.0.0.1:8542"]
 
@@ -54,6 +57,9 @@ sync_p2p_seed = ["tcp://127.0.0.1:33032"]
 # Peers to connect to for the syncing protocol
 #sync_p2p_peer = []
 
+# Prefered transports of outbound connections for the syncing protocol
+#sync_p2p_transports = ["tls", "tcp"]
+
 # Whitelisted faucet public key
 faucet_pub = ["1z3JuN456MieHtpoF4cr5a1ntFF87L8Z3m2E5ACppFodFcjnLS"]
 

contrib/localnet/faucetd.toml

@@ -36,6 +36,9 @@ sync_p2p_seed = ["tcp://127.0.0.1:33032"]
 # Peers to connect to for the syncing protocol
 #sync_p2p_peer = []
 
+# Prefered transports of outbound connections for the syncing protocol
+#sync_p2p_transports = ["tls", "tcp"]
+
 # Airdrop timeout limit in seconds
 airdrop_timeout = 10
 

script/research/fud/fud/fud_config.toml

@@ -26,3 +26,6 @@ seeds = ["tls://lilith0.dark.fi:13337", "tls://lilith1.dark.fi:13337"]
 
 # Peers to connect to
 #peers = []
+
+# Prefered transports for outbound connections
+#transports = ["tls", "tcp"]

script/research/fud/fud/src/main.rs

@@ -71,6 +71,10 @@ struct Args {
     /// Connect to peer (repeatable flag)
     peers: Vec<Url>,
 
+    #[structopt(long)]
+    /// Prefered transports for outbound connections (repeatable flag)
+    transports: Vec<String>,
+
     #[structopt(short, parse(from_occurrences))]
     /// Increase verbosity (-vvv supported)
     verbose: u8,
@@ -378,6 +382,7 @@ async fn realmain(args: Args, ex: Arc<Executor<'_>>) -> Result<()> {
         external_addr: args.p2p_external,
         peers: args.peers.clone(),
         seeds: args.seeds.clone(),
+        outbound_transports: net::settings::get_outbound_transports(args.transports),
         ..Default::default()
     };
 

script/research/fud/localnet/fud_config0.toml

@@ -26,3 +26,6 @@ seeds = ["tcp://127.0.0.1:13337"]
 
 # Peers to connect to
 #peers = []
+
+# Prefered transports for outbound connections
+#transports = ["tls", "tcp"]

script/research/fud/localnet/fud_config1.toml

@@ -26,3 +26,6 @@ seeds = ["tcp://127.0.0.1:13337"]
 
 # Peers to connect to
 #peers = []
+
+# Prefered transports for outbound connections
+#transports = ["tls", "tcp"]

src/net/session/outbound_session.rs

@@ -3,12 +3,13 @@ use std::fmt;
 
 use async_executor::Executor;
 use async_trait::async_trait;
-use log::{debug, info};
+use log::{debug, info, warn};
 use rand::seq::SliceRandom;
 use serde_json::json;
 use url::Url;
 
 use crate::{
+    net::TransportName,
     system::{StoppableTask, StoppableTaskPtr, Subscriber, SubscriberPtr, Subscription},
     util::async_util,
     Error, Result,
@@ -145,61 +146,78 @@ impl OutboundSession {
 
         let connector = Connector::new(self.p2p().settings(), Arc::new(parent));
 
+        // Retrieve preferent outbound transports
+        let outbound_transports = &self.p2p().settings().outbound_transports;
+
         loop {
             let addr = self.load_address(slot_number).await?;
-            info!(target: "net", "#{} connecting to outbound [{}]", slot_number, addr);
+            info!(target: "net", "#{} processing outbound [{}]", slot_number, addr);
             {
                 let info = &mut self.slot_info.lock().await[slot_number as usize];
                 info.addr = Some(addr.clone());
                 info.state = OutboundState::Pending;
             }
 
-            match connector.connect(addr.clone()).await {
-                Ok(channel) => {
-                    // Blacklist goes here
-
-                    info!(target: "net", "#{} connected to outbound [{}]", slot_number, addr);
-
-                    let stop_sub = channel.subscribe_stop().await;
-
-                    if stop_sub.is_err() {
-                        continue
+            // Check that addr transport is in configured outbound transport
+            let addr_transport = TransportName::try_from(addr.clone())?;
+            let transports = if outbound_transports.contains(&addr_transport) {
+                vec![addr_transport]
+            } else {
+                warn!(target: "net", "#{} address {} transport is not in accepted outbound transports, will try with: {:?}", slot_number, addr, outbound_transports);
+                outbound_transports.clone()
+            };
+
+            for transport in transports {
+                // Replace addr transport
+                let mut transport_addr = addr.clone();
+                transport_addr.set_scheme(&transport.to_scheme())?;
+                info!(target: "net", "#{} connecting to outbound [{}]", slot_number, transport_addr);
+                match connector.connect(transport_addr.clone()).await {
+                    Ok(channel) => {
+                        // Blacklist goes here
+                        info!(target: "net", "#{} connected to outbound [{}]", slot_number, transport_addr);
+
+                        let stop_sub = channel.subscribe_stop().await;
+                        if stop_sub.is_err() {
+                            continue
+                        }
+
+                        self.clone().register_channel(channel.clone(), executor.clone()).await?;
+
+                        // Channel is now connected but not yet setup
+
+                        // Remove pending lock since register_channel will add the channel to p2p
+                        self.p2p().remove_pending(&addr).await;
+                        {
+                            let info = &mut self.slot_info.lock().await[slot_number as usize];
+                            info.channel = Some(channel.clone());
+                            info.state = OutboundState::Connected;
+                        }
+
+                        // Notify that channel processing has been finished
+                        if *self.notify.lock().await {
+                            self.channel_subscriber.notify(Ok(channel)).await;
+                        }
+
+                        // Wait for channel to close
+                        stop_sub.unwrap().receive().await;
                     }
-
-                    self.clone().register_channel(channel.clone(), executor.clone()).await?;
-
-                    // Channel is now connected but not yet setup
-
-                    // Remove pending lock since register_channel will add the channel to p2p
-                    self.p2p().remove_pending(&addr).await;
-                    {
-                        let info = &mut self.slot_info.lock().await[slot_number as usize];
-                        info.channel = Some(channel.clone());
-                        info.state = OutboundState::Connected;
-                    }
-
-                    // Notify that channel processing has been finished
-                    if *self.notify.lock().await {
-                        self.channel_subscriber.notify(Ok(channel)).await;
+                    Err(err) => {
+                        info!(target: "net", "Unable to connect to outbound [{}]: {}", &transport_addr, err);
                     }
-
-                    // Wait for channel to close
-                    stop_sub.unwrap().receive().await;
                 }
-                Err(err) => {
-                    info!(target: "net", "Unable to connect to outbound [{}]: {}", &addr, err);
-                    {
-                        let info = &mut self.slot_info.lock().await[slot_number as usize];
-                        info.addr = None;
-                        info.channel = None;
-                        info.state = OutboundState::Open;
-                    }
+            }
 
-                    // Notify that channel processing has been finished
-                    if *self.notify.lock().await {
-                        self.channel_subscriber.notify(Err(err)).await;
-                    }
-                }
+            {
+                let info = &mut self.slot_info.lock().await[slot_number as usize];
+                info.addr = None;
+                info.channel = None;
+                info.state = OutboundState::Open;
+            }
+
+            // Notify that channel processing has been finished (failed)
+            if *self.notify.lock().await {
+                self.channel_subscriber.notify(Err(Error::ConnectFailed)).await;
             }
         }
     }

src/net/settings.rs

@@ -5,6 +5,8 @@ use structopt::StructOpt;
 use structopt_toml::StructOptToml;
 use url::Url;
 
+use crate::net::TransportName;
+
 /// Atomic pointer to network settings.
 pub type SettingsPtr = Arc<Settings>;
 
@@ -23,6 +25,7 @@ pub struct Settings {
     pub peers: Vec<Url>,
     pub seeds: Vec<Url>,
     pub node_id: String,
+    pub outbound_transports: Vec<TransportName>,
 }
 
 impl Default for Settings {
@@ -40,6 +43,7 @@ impl Default for Settings {
             peers: Vec::new(),
             seeds: Vec::new(),
             node_id: String::new(),
+            outbound_transports: get_outbound_transports(vec![]),
         }
     }
 }
@@ -88,6 +92,11 @@ pub struct SettingsOpt {
     #[serde(default)]
     #[structopt(skip)]
     pub node_id: String,
+
+    /// Prefered transports for outbound connections
+    #[serde(default)]
+    #[structopt(long = "transports")]
+    pub outbound_transports: Vec<String>,
 }
 
 impl From<SettingsOpt> for Settings {
@@ -105,6 +114,26 @@ impl From<SettingsOpt> for Settings {
             peers: settings_opt.peers,
             seeds: settings_opt.seeds,
             node_id: settings_opt.node_id,
+            outbound_transports: get_outbound_transports(settings_opt.outbound_transports),
         }
     }
 }
+
+/// Auxiliary function to convert outbound transport Vec<String>
+/// to Vec<TransportName>, using defaults if empty.
+pub fn get_outbound_transports(opt_outbound_transports: Vec<String>) -> Vec<TransportName> {
+    let mut outbound_transports = vec![];
+    for transport in opt_outbound_transports {
+        let transport_name = TransportName::try_from(transport.as_str()).unwrap();
+        outbound_transports.push(transport_name);
+    }
+
+    if outbound_transports.is_empty() {
+        let tls = TransportName::Tcp(Some("tls".into()));
+        outbound_transports.push(tls);
+        let tcp = TransportName::Tcp(None);
+        outbound_transports.push(tcp);
+    }
+
+    outbound_transports
+}

src/net/transport.rs

@@ -34,7 +34,7 @@ pub trait TransportListener: Send + Sync + Unpin {
     async fn next(&self) -> Result<(Box<dyn TransportStream>, Url)>;
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug, PartialEq)]
 pub enum TransportName {
     Tcp(Option<String>),
     Tor(Option<String>),
@@ -42,11 +42,25 @@ pub enum TransportName {
     Unix,
 }
 
-impl TryFrom<Url> for TransportName {
+impl TransportName {
+    pub fn to_scheme(&self) -> String {
+        match self {
+            Self::Tcp(None) => "tcp".into(),
+            Self::Tcp(Some(opt)) => format!("tcp+{}", opt),
+            Self::Tor(None) => "tor".into(),
+            Self::Tor(Some(opt)) => format!("tor+{}", opt),
+            Self::Nym(None) => "nym".into(),
+            Self::Nym(Some(opt)) => format!("nym+{}", opt),
+            Self::Unix => "unix".into(),
+        }
+    }
+}
+
+impl TryFrom<&str> for TransportName {
     type Error = crate::Error;
 
-    fn try_from(url: Url) -> Result<Self> {
-        let transport_name = match url.scheme() {
+    fn try_from(scheme: &str) -> Result<Self> {
+        let transport_name = match scheme {
             "tcp" => Self::Tcp(None),
             "tcp+tls" | "tls" => Self::Tcp(Some("tls".into())),
             "tor" => Self::Tor(None),
@@ -60,6 +74,14 @@ impl TryFrom<Url> for TransportName {
     }
 }
 
+impl TryFrom<Url> for TransportName {
+    type Error = crate::Error;
+
+    fn try_from(url: Url) -> Result<Self> {
+        Self::try_from(url.scheme())
+    }
+}
+
 /// The `Transport` trait serves as a base for implementing transport protocols.
 /// Base transports can optionally be upgraded with TLS in order to support encryption.
 /// The implementation of our TLS authentication can be found in the