Add support for Ygddrasil 0.5

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "yggdrasil-jumper"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2021"
 authors = [ "one-d-wide <one-d-wide@protonmail.com>" ]
 license = "GPL-3.0"
@@ -16,17 +16,17 @@ name = "stun-tcp"
 [dependencies]
 atty = "0.2"
 bytecodec = "0.4"
-bytes = { version = "1.4", features = [ "serde" ] }
-clap = { version = "4.2", features = [ "derive" ] }
+bytes = { version = "1.5", features = [ "serde" ] }
+clap = { version = "4.4", features = [ "derive" ] }
 futures = { version = "0.3" }
 rand = { version = "0.8", features = [ "std_rng" ] }
 serde_json = "1.0"
 serde = { version = "1.0", features = [ "derive" ] }
 socket2 = { version = "0.5", features = [ "all" ] }
 stun_codec = "0.3"
 tokio-util = { version = "0.7", features = [ "full" ] }
-tokio = { version = "1.27", features = [ "full" ] }
-toml = "0.7"
+tokio = { version = "1.33", features = [ "full" ] }
+toml = "0.8"
 tracing = "0.1"
 tracing-subscriber = "0.3"
 yggdrasilctl = { version = "1", default-features = false, features = [ "use_tokio" ] }

README.md

@@ -80,7 +80,7 @@ $ yggdrasil-jumper --config <path> # standard input will be read if path is "-"
 
 In order to know what address to use with [NAT traversal], jumper must know self external internet address and port. This task is performed using [STUN] protocol with TCP extension, hence not every [STUN] server is supported. [STUN] standard is quite broad, but jumper utilities only address lookup feature.
 
-You can check compatibility with `stun-tcp` binary from this repository.
+You can check compatibility using `stun-tcp` binary from this repository.
 
 ```shell
 $ cargo build --bin stun-tcp --release

src/admin_api.rs

@@ -38,7 +38,7 @@ pub async fn connect(config: Config) -> Result<Endpoint<util::RWSocket>, ()> {
                     let build_version = info.build_version;
                     let versions: Vec<u64> = build_version
                         .as_str()
-                        .split(".")
+                        .split('.')
                         .filter_map(|v| v.parse().ok())
                         .collect();
                     if versions.len() != 3 {
@@ -47,15 +47,14 @@ pub async fn connect(config: Config) -> Result<Endpoint<util::RWSocket>, ()> {
                     }
 
                     // If router version is lower then v0.4.5
-                    if versions[0] == 0 && versions[1] <= 4 && (versions[1] < 4 || versions[2] < 5)
+                    if versions[0] == 0
+                        && versions[1] <= 4
+                        && (versions[1] < 4 || versions[2] < 5)
+                        && config.yggdrasil_listen.is_empty()
                     {
-                        if config.yggdrasil_listen.is_empty() {
-                            warn!("Direct bridges can't be connected to the router of version {build_version} at {uri}");
-                            warn!(
-                                "Routers prior to v0.4.5 (Oct 2022) don't support addpeer/removepeer commands"
-                            );
-                            warn!("Help: Specify `ygdrasil_addresses` in the config or update your router");
-                        }
+                        warn!("Direct bridges can't be connected to the router of version {build_version} at {uri}");
+                        warn!("Routers prior to v0.4.5 (Oct 2022) don't support addpeer/removepeer commands");
+                        warn!("Help: Specify `yggdrasil_addresses` in the config or update your router");
                     }
 
                     return Ok(endpoint);

src/bin/stun-tcp.rs

@@ -1,7 +1,5 @@
 use yggdrasil_jumper::*;
 
-use SocketAddr::*;
-
 #[derive(clap::Parser)]
 #[command(name = "stun-tcp", version)]
 pub struct CliArgs {
@@ -97,14 +95,13 @@ async fn start() -> Result<(), ()> {
         let address = lookup_host(server.as_str())
             .await
             .map_err(map_error!("Failed to lookup host"))?
-            .filter(|a| (a.is_ipv4() && cli_args.ipv4) || (a.is_ipv6() && cli_args.ipv6))
-            .next()
+            .find(|a| (a.is_ipv4() && cli_args.ipv4) || (a.is_ipv6() && cli_args.ipv6))
             .ok_or_else(|| error!("No address resolved"))?;
 
         // Connect to server
         let port = match &address {
-            V6(_) => port_v6,
-            V4(_) => port_v4,
+            SocketAddr::V6(_) => port_v6,
+            SocketAddr::V4(_) => port_v4,
         };
         let socket = util::new_socket_in_domain(&address, port)?;
         let stream = select! {
@@ -127,8 +124,8 @@ async fn start() -> Result<(), ()> {
             }
 
             let last_address = match &address {
-                V6(_) => &mut last_address_v6,
-                V4(_) => &mut last_address_v4,
+                SocketAddr::V6(_) => &mut last_address_v6,
+                SocketAddr::V4(_) => &mut last_address_v4,
             };
 
             if let Some(ref last_address) = last_address {

src/bridge.rs

@@ -24,9 +24,9 @@ async fn bridge(
         loop {
             if ! recorded {
                 if let Some(ref addr) = address {
-                    let old = state.active_sessions.write().await.insert(addr.clone(), Bridge);
-                    use sessions::SessionType::*;
-                    if let Some(Bridge) = old {
+                    use sessions::SessionType;
+                    let old = state.active_sessions.write().await.insert(*addr, SessionType::Bridge);
+                    if let Some(SessionType::Bridge) = old {
                         // Multiple connections with the same identifiers are not allowed by the OS.
                         warn!("Bridge already exists. Implementation bug");
                         return Err(());
@@ -77,23 +77,22 @@ async fn bridge(
                     }
 
                     // Return if peer is not connected
-                    if delay_shutdown.is_none() {
-                        if ! peers.iter().any(|peer| &peer.remote == &uri) {
-                            return Err(info!("Bridge is not connected as peer"));
-                        }
+                    if delay_shutdown.is_none() && !peers.iter().filter_map(|peer| peer.remote.as_ref()).any(|remote| remote == &uri) {
+                        return Err(info!("Bridge is not connected as peer"));
                     }
-                    // Retreive address
+                    // Retrieve address
                     if address.is_none() {
                         for peer in peers.iter() {
-                            if &peer.remote == &uri {
-                                let addr = peer.address;
-                                *address = Some(addr);
-                                info!("Peer address instantiated");
-                                if let Some(false) = config.whitelist.as_ref().map(|w| w.contains(&addr)) {
-                                    info!("Peer misses whitelist");
-                                    return Ok(())
+                            if peer.remote.as_deref() == Some(&uri) {
+                                if let Some(addr) = peer.address {
+                                    *address = Some(addr);
+                                    info!("Peer address instantiated");
+                                    if let Some(false) = config.whitelist.as_ref().map(|w| w.contains(&addr)) {
+                                        info!("Peer misses whitelist");
+                                        return Ok(())
+                                    }
+                                    break;
                                 }
-                                break;
                             }
                         }
                     }
@@ -132,11 +131,11 @@ pub async fn run_bridge(
     let uri = |local_addr: std::io::Result<SocketAddr>| -> Result<String, ()> {
         Ok(format!(
             "tcp://{}",
-            local_addr.map_err(map_warn!("Failed to retreive local inbound socket address"))?
+            local_addr.map_err(map_warn!("Failed to retrieve local inbound socket address"))?
         ))
     };
 
-    // Try connect self to ygdrasil listen addresses directly
+    // Try connect self to yggdrasil listen addresses directly
     for addr in &config.yggdrasil_listen {
         if let Ok(ygg) = TcpStream::connect(addr).await.map_err(map_warn!(
             "Failed to connect to yggdrasil listen socket at {addr}"
@@ -147,10 +146,10 @@ pub async fn run_bridge(
         }
     }
 
-    // Fallback. Try connect ygdrasil to self
+    // Fallback. Try connect yggdrasil to self
     let socket_address = socket
         .local_addr()
-        .map_err(map_warn!("Failed to retreive local socket address"))?;
+        .map_err(map_warn!("Failed to retrieve local socket address"))?;
     let ygg = util::new_socket_in_domain(&socket_address, 0)?
         .listen(1)
         .map_err(map_warn!("Failed to create local inbound socket"))?;

src/config.rs

@@ -89,7 +89,7 @@ impl ConfigInner {
         };
         let config: Self =
             toml::from_str(config.as_str()).map_err(map_error!("Failed to parse config"))?;
-        Ok(config.verify()?)
+        config.verify()
     }
 
     fn verify(self) -> Result<Self, ()> {
@@ -107,8 +107,7 @@ impl ConfigInner {
 
 fn parse_duration<'de, D: serde::Deserializer<'de>>(deserializer: D) -> Result<Duration, D::Error> {
     use serde::de::Error;
-    Duration::try_from_secs_f64(Deserialize::deserialize(deserializer)?)
-        .map_err(|e| D::Error::custom(e))
+    Duration::try_from_secs_f64(Deserialize::deserialize(deserializer)?).map_err(D::Error::custom)
 }
 
 #[cfg(test)]

src/internet.rs

@@ -51,7 +51,7 @@ pub async fn listen(config: Config, state: State, listeners: Vec<TcpListener>) -
     }
 
     // Create await pool
-    let mut pool: FuturesUnordered<_> = listeners.into_iter().map(|l| listen(l)).collect();
+    let mut pool: FuturesUnordered<_> = listeners.into_iter().map(listen).collect();
 
     loop {
         // Accept every incoming connection
@@ -74,3 +74,38 @@ pub async fn listen(config: Config, state: State, listeners: Vec<TcpListener>) -
         ));
     }
 }
+
+pub async fn traverse(
+    config: Config,
+    state: State,
+    local: u16,
+    remote: SocketAddr,
+    monitor_addr: Option<Ipv6Addr>,
+) -> IoResult<TcpStream> {
+    let mut last_err: Option<std::io::Result<TcpStream>> = None;
+    for _ in 0..config.nat_traversal_retry_count {
+        // Check if bridge was already instantiated
+        if let Some(ref monitor_addr) = monitor_addr {
+            if let Some(sessions::SessionType::Bridge) =
+                state.active_sessions.read().await.get(monitor_addr)
+            {
+                break;
+            }
+        }
+        {
+            select! {
+                err = util::new_socket_in_domain(&remote, local)
+                    .map_err(|_| IoError::last_os_error())?.connect(remote) => { last_err = Some(err); },
+                _ = sleep(config.nat_traversal_connection_timeout) => {},
+            }
+            if let Some(Ok(_)) = last_err {
+                break;
+            }
+        }
+        sleep(config.nat_traversal_connection_delay).await;
+    }
+    match last_err {
+        Some(res) => res,
+        None => Err(IoError::new(IoErrorKind::TimedOut, "Timeout")),
+    }
+}

src/lib.rs

@@ -4,6 +4,8 @@ pub use {
     socket2::{Domain, Protocol, Socket, Type},
     std::{
         collections::{HashMap, HashSet},
+        future::Future,
+        io::{Error as IoError, ErrorKind as IoErrorKind, Result as IoResult},
         mem::drop,
         net::{Ipv6Addr, SocketAddr, SocketAddrV6},
         path::{Path, PathBuf},

src/protocol.rs

@@ -42,20 +42,12 @@ pub async fn try_session(
         let addresses = state.watch_external.borrow();
         (
             if config.allow_ipv6 {
-                addresses
-                    .iter()
-                    .map(|a| a.external)
-                    .filter(|a| a.is_ipv6())
-                    .next()
+                addresses.iter().map(|a| a.external).find(|a| a.is_ipv6())
             } else {
                 None
             },
             if config.allow_ipv4 {
-                addresses
-                    .iter()
-                    .map(|a| a.external)
-                    .filter(|a| a.is_ipv4())
-                    .next()
+                addresses.iter().map(|a| a.external).find(|a| a.is_ipv4())
             } else {
                 None
             },
@@ -131,10 +123,9 @@ pub async fn try_session(
     .map_err(map_info!("Failed to prarse peer's external addresses"))?;
 
     // 6. Validate external addresses.
-    use SocketAddr::*;
     match (external, remote_external) {
-        (V6(_), V6(_)) => (),
-        (V4(_), V4(_)) => (),
+        (SocketAddr::V6(_), SocketAddr::V6(_)) => (),
+        (SocketAddr::V4(_), SocketAddr::V4(_)) => (),
         _ => {
             info!("External addresses have incompatible ranges: self {external:?}, remote {remote_external:?}");
             return Err(());
@@ -149,39 +140,22 @@ pub async fn try_session(
         .watch_external
         .borrow()
         .iter()
-        .filter(|addr| &addr.external == &external)
-        .next()
+        .find(|addr| addr.external == external)
         .ok_or_else(|| info!("Expected external address unavailible: {external}"))?
         .local;
     let remote = remote_external;
 
-    info!("Trying NAT traversal from {local} to {remote}");
-    let mut last_err: Option<std::io::Result<TcpStream>> = None;
-    for _ in 0..config.nat_traversal_retry_count {
-        // Check if bridge was already instantiated
-        if let Some(sessions::SessionType::Bridge) =
-            state.active_sessions.read().await.get(address.ip())
-        {
-            break;
-        }
-        {
-            select! {
-                err = util::new_socket_in_domain(&local, local.port())?.connect(remote) => { last_err = Some(err); },
-                _ = sleep(config.nat_traversal_connection_timeout) => {},
-            }
-            if let Some(Ok(_)) = last_err {
-                break;
-            }
-        }
-        sleep(config.nat_traversal_connection_delay).await;
-    }
-    match last_err {
-        Some(Ok(socket)) => {
-            return bridge::run_bridge(config, state, remote, Some(address.ip().clone()), socket)
-                .await
-        }
-        Some(Err(err)) => info!("Failed: {err}"),
-        None => info!("Failed: Timeout"),
+    if let Ok(socket) = internet::traverse(
+        config.clone(),
+        state.clone(),
+        local.port(),
+        remote,
+        Some(*address.ip()),
+    )
+    .await
+    .map_err(map_info!("NAT traversal failed"))
+    {
+        return bridge::run_bridge(config, state, remote, Some(*address.ip()), socket).await;
     }
     Err(())
 }