feat(echo_service): find if peer is externally reachable if external IP

and port is provided manually

Cargo.toml

@@ -21,7 +21,6 @@ serde_json = "1.0.59"
 structopt = "~0.3.15"
 thiserror = "1.0.23"
 webpki = "~0.21.3"
-flexi_logger = "~0.16.1"
 
   [dependencies.bytes]
   version = "1.0.1"
@@ -49,8 +48,8 @@ flexi_logger = "~0.16.1"
   features = [ "rt-threaded", "sync", "time", "macros", "io-std", "io-util" ]
 
 [dev-dependencies]
+flexi_logger = "~0.16.1"
 anyhow = "1.0.36"
-assert_matches = "1.3"
 rand = "~0.7.3"
 
 [target."cfg(any(all(unix, not(any(target_os = \"android\", target_os = \"androideabi\", target_os = \"ios\"))), windows))".dependencies]

src/api.rs

@@ -65,7 +65,6 @@ impl QuicP2p {
         bootstrap_nodes: &[SocketAddr],
         use_bootstrap_cache: bool,
     ) -> Result<Self> {
-        crate::utils::init_logging();
         let cfg = unwrap_config_or_default(cfg)?;
         debug!("Config passed in to qp2p: {:?}", cfg);
 
@@ -202,13 +201,10 @@ impl QuicP2p {
             .iter()
             .map(|addr| Box::pin(endpoint.connect_to(addr)));
 
-        future::select_ok(tasks)
-            .await
-            .map_err(|err| {
-                error!("Failed to bootstrap to the network: {}", err);
-                Error::BootstrapFailure
-            })?
-            .0;
+        future::select_ok(tasks).await.map_err(|err| {
+            error!("Failed to bootstrap to the network: {}", err);
+            Error::BootstrapFailure
+        })?;
 
         Ok((
             endpoint,

src/connection_pool.rs

@@ -60,7 +60,7 @@ impl ConnectionPool {
             .map
             .range_mut(Key::min(*addr)..=Key::max(*addr))
             .into_iter()
-            .map(|(key, _)| key.clone())
+            .map(|(key, _)| *key)
             .collect::<Vec<_>>();
 
         keys_to_remove

src/connections.rs

@@ -7,6 +7,8 @@
 // specific language governing permissions and limitations relating to use of the SAFE Network
 // Software.
 
+use crate::QuicP2p;
+
 use super::{
     connection_pool::{ConnectionPool, ConnectionRemover},
     error::{Error, Result},
@@ -16,8 +18,11 @@ use bytes::Bytes;
 use futures::stream::StreamExt;
 use log::{error, trace};
 use std::net::SocketAddr;
-use tokio::select;
 use tokio::sync::mpsc::UnboundedSender;
+use tokio::{
+    select,
+    time::{timeout, Duration},
+};
 
 /// Connection instance to a node which can be used to send messages to it
 #[derive(Clone)]
@@ -216,9 +221,7 @@ pub(super) fn listen_for_incoming_messages(
 }
 
 // Returns next message sent by peer in an unidirectional stream.
-async fn next_on_uni_streams(
-    uni_streams: &mut quinn::IncomingUniStreams,
-) -> Option<Bytes> {
+async fn next_on_uni_streams(uni_streams: &mut quinn::IncomingUniStreams) -> Option<Bytes> {
     match uni_streams.next().await {
         None => None,
         Some(Err(quinn::ConnectionError::ApplicationClosed { .. })) => {
@@ -273,7 +276,28 @@ async fn next_on_bi_streams(
                     address_sent,
                     peer_addr
                 );
-                let message = WireMsg::EndpointVerficationResp(address_sent == peer_addr);
+                // Verify if the peer's endpoint is reachable via EchoServiceReq
+                let qp2p = QuicP2p::with_config(Default::default(), &[], false).ok()?;
+                let (temporary_endpoint, _, _, _) = qp2p.new_endpoint().await.ok()?;
+                let (mut temp_send, mut temp_recv) = temporary_endpoint
+                    .open_bidirectional_stream(&address_sent)
+                    .await
+                    .ok()?;
+                let message = WireMsg::EndpointEchoReq;
+                message
+                    .write_to_stream(&mut temp_send.quinn_send_stream)
+                    .await
+                    .ok()?;
+                let verified = matches!(
+                    timeout(
+                        Duration::from_secs(30),
+                        WireMsg::read_from_stream(&mut temp_recv.quinn_recv_stream)
+                    )
+                    .await,
+                    Ok(Ok(WireMsg::EndpointEchoResp(_)))
+                );
+
+                let message = WireMsg::EndpointVerficationResp(verified);
                 message.write_to_stream(&mut send).await.ok()?;
                 trace!("Responded to Endpoint verification request");
                 Some(Bytes::new())
@@ -325,7 +349,7 @@ mod tests {
 
         let connection = peer1
             .get_connection(&peer2_addr)
-            .ok_or_else(|| Error::MissingConnection)?;
+            .ok_or(Error::MissingConnection)?;
         let (mut send_stream, mut recv_stream) = connection.open_bi().await?;
         let message = WireMsg::EndpointEchoReq;
         message

src/endpoint.rs

@@ -14,7 +14,10 @@ use super::{
     api::DEFAULT_UPNP_LEASE_DURATION_SEC,
     connection_deduplicator::ConnectionDeduplicator,
     connection_pool::ConnectionPool,
-    connections::{listen_for_incoming_connections, listen_for_incoming_messages, Connection, RecvStream, SendStream},
+    connections::{
+        listen_for_incoming_connections, listen_for_incoming_messages, Connection, RecvStream,
+        SendStream,
+    },
     error::Result,
     Config,
 };
@@ -124,12 +127,12 @@ impl Endpoint {
             // External IP and port number is provided
             // This means that the user has performed manual port-forwarding
             // Verify that the given socket address is reachable
-            if let Some(contact) = endpoint.bootstrap_nodes.iter().next() {
+            if let Some(contact) = endpoint.bootstrap_nodes.get(0) {
                 info!("Verifying provided public IP address");
                 endpoint.connect_to(contact).await?;
                 let connection = endpoint
                     .get_connection(&contact)
-                    .ok_or_else(|| Error::BootstrapFailure)?; //FIXME
+                    .ok_or(Error::MissingConnection)?;
                 let (mut send, mut recv) = connection.open_bi().await?;
                 send.send(WireMsg::EndpointVerificationReq(addr)).await?;
                 let response = WireMsg::read_from_stream(&mut recv.quinn_recv_stream).await?;
@@ -374,11 +377,14 @@ impl Endpoint {
     }
 
     /// Open a bi-directional peer with a given peer
-    pub async fn open_bidirectional_stream(&self, peer_addr: &SocketAddr) -> Result<(SendStream, RecvStream)> {
+    pub async fn open_bidirectional_stream(
+        &self,
+        peer_addr: &SocketAddr,
+    ) -> Result<(SendStream, RecvStream)> {
         self.connect_to(peer_addr).await?;
         let connection = self
             .get_connection(peer_addr)
-            .ok_or_else(|| Error::MissingConnection)?; // will never be None
+            .ok_or(Error::MissingConnection)?;
         connection.open_bi().await
     }
 
@@ -387,9 +393,7 @@ impl Endpoint {
     pub async fn send_message(&self, msg: Bytes, dest: &SocketAddr) -> Result<()> {
         self.connect_to(dest).await?;
 
-        let connection = self
-            .get_connection(dest)
-            .ok_or_else(|| Error::MissingConnection)?; // will never be None
+        let connection = self.get_connection(dest).ok_or(Error::MissingConnection)?;
         connection.send_uni(msg).await?;
         Ok(())
     }
@@ -410,9 +414,7 @@ impl Endpoint {
         for node in self.bootstrap_nodes.iter().cloned() {
             debug!("Connecting to {:?}", &node);
             self.connect_to(&node).await?;
-            let connection = self
-                .get_connection(&node)
-                .ok_or_else(|| Error::MissingConnection)?;
+            let connection = self.get_connection(&node).ok_or(Error::MissingConnection)?;
             let task_handle = tokio::spawn(async move {
                 let (mut send_stream, mut recv_stream) = connection.open_bi().await?;
                 send_stream.send(WireMsg::EndpointEchoReq).await?;

src/utils.rs

@@ -93,7 +93,7 @@ where
     Ok(())
 }
 
-// #[cfg(test)]
+#[cfg(test)]
 pub(crate) fn init_logging() {
     use flexi_logger::{DeferredNow, Logger};
     use log::Record;

src/wire_msg.rs

@@ -25,7 +25,7 @@ pub enum WireMsg {
     EndpointEchoReq,
     EndpointEchoResp(SocketAddr),
     EndpointVerificationReq(SocketAddr),
-    EndpointVerficationResp(bool), // Use result here?
+    EndpointVerficationResp(bool),
     UserMsg(Bytes),
 }