NAT traversal: ICE-esque candidate selection

client/src/data_store.rs

@@ -80,7 +80,7 @@ impl DataStore {
     ///
     /// Note, however, that this does not prevent a compromised server from adding a new
     /// peer under its control, of course.
-    pub fn update_peers(&mut self, current_peers: Vec<Peer>) -> Result<(), Error> {
+    pub fn update_peers(&mut self, current_peers: &[Peer]) -> Result<(), Error> {
         let peers = match &mut self.contents {
             Contents::V1 { ref mut peers, .. } => peers,
         };
@@ -149,6 +149,7 @@ mod tests {
                 is_redeemed: true,
                 persistent_keepalive_interval: None,
                 invite_expires: None,
+                candidates: vec![],
             }
         }];
         static ref BASE_CIDRS: Vec<Cidr> = vec![Cidr {
@@ -167,7 +168,7 @@ mod tests {
         assert_eq!(0, store.peers().len());
         assert_eq!(0, store.cidrs().len());
 
-        store.update_peers(BASE_PEERS.to_owned()).unwrap();
+        store.update_peers(&BASE_PEERS).unwrap();
         store.set_cidrs(BASE_CIDRS.to_owned());
         store.write().unwrap();
     }
@@ -189,13 +190,13 @@ mod tests {
             DataStore::open_with_path(&dir.path().join("peer_store.json"), false).unwrap();
 
         // Should work, since peer is unmodified.
-        store.update_peers(BASE_PEERS.clone()).unwrap();
+        store.update_peers(&BASE_PEERS).unwrap();
 
         let mut modified = BASE_PEERS.clone();
         modified[0].contents.public_key = "foo".to_string();
 
         // Should NOT work, since peer is unmodified.
-        assert!(store.update_peers(modified).is_err());
+        assert!(store.update_peers(&modified).is_err());
     }
 
     #[test]
@@ -206,7 +207,7 @@ mod tests {
             DataStore::open_with_path(&dir.path().join("peer_store.json"), false).unwrap();
 
         // Should work, since peer is unmodified.
-        store.update_peers(vec![]).unwrap();
+        store.update_peers(&[]).unwrap();
         let new_peers = BASE_PEERS
             .iter()
             .cloned()

client/src/main.rs

@@ -4,13 +4,17 @@ use dialoguer::{Confirm, Input};
 use hostsfile::HostsBuilder;
 use indoc::eprintdoc;
 use shared::{
-    interface_config::InterfaceConfig, prompts, AddAssociationOpts, AddCidrOpts, AddPeerOpts,
-    Association, AssociationContents, Cidr, CidrTree, DeleteCidrOpts, EndpointContents,
-    InstallOpts, Interface, IoErrorContext, NetworkOpt, Peer, PeerDiff, RedeemContents,
-    RenamePeerOpts, State, WrappedIoError, CLIENT_CONFIG_DIR, REDEEM_TRANSITION_WAIT,
+    interface_config::InterfaceConfig,
+    prompts,
+    wg::{DeviceExt, PeerInfoExt},
+    AddAssociationOpts, AddCidrOpts, AddPeerOpts, Association, AssociationContents, Cidr, CidrTree,
+    DeleteCidrOpts, Endpoint, EndpointContents, InstallOpts, Interface, IoErrorContext, NetworkOpt,
+    Peer, RedeemContents, RenamePeerOpts, State, WrappedIoError, CLIENT_CONFIG_DIR,
+    REDEEM_TRANSITION_WAIT,
 };
 use std::{
     fmt, io,
+    net::SocketAddr,
     path::{Path, PathBuf},
     thread,
     time::Duration,
@@ -19,9 +23,11 @@ use structopt::{clap::AppSettings, StructOpt};
 use wgctrl::{Device, DeviceUpdate, InterfaceName, PeerConfigBuilder, PeerInfo};
 
 mod data_store;
+mod nat;
 mod util;
 
 use data_store::DataStore;
+use nat::NatTraverse;
 use shared::{wg, Error};
 use util::{human_duration, human_size, Api};
 
@@ -484,70 +490,13 @@ fn fetch(
     let mut store = DataStore::open_or_create(interface)?;
     let State { peers, cidrs } = Api::new(&config.server).http("GET", "/user/state")?;
 
-    let device_info = Device::get(interface, network.backend).with_str(interface.as_str_lossy())?;
-    let interface_public_key = device_info
-        .public_key
-        .as_ref()
-        .map(|k| k.to_base64())
-        .unwrap_or_default();
-    let existing_peers = &device_info.peers;
-
-    // Match existing peers (by pubkey) to new peer information from the server.
-    let modifications = peers.iter().filter_map(|peer| {
-        if peer.is_disabled || peer.public_key == interface_public_key {
-            None
-        } else {
-            let existing_peer = existing_peers
-                .iter()
-                .find(|p| p.config.public_key.to_base64() == peer.public_key);
-            PeerDiff::new(existing_peer, Some(peer)).unwrap()
-        }
-    });
-
-    // Remove any peers on the interface that aren't in the server's peer list any more.
-    let removals = existing_peers.iter().filter_map(|existing| {
-        let public_key = existing.config.public_key.to_base64();
-        if peers.iter().any(|p| p.public_key == public_key) {
-            None
-        } else {
-            PeerDiff::new(Some(existing), None).unwrap()
-        }
-    });
+    let device = Device::get(interface, network.backend)?;
+    let modifications = device.diff(&peers);
 
     let updates = modifications
-        .chain(removals)
-        .inspect(|diff| {
-            let public_key = diff.public_key().to_base64();
-
-            let text = match (diff.old, diff.new) {
-                (None, Some(_)) => "added".green(),
-                (Some(_), Some(_)) => "modified".yellow(),
-                (Some(_), None) => "removed".red(),
-                _ => unreachable!("PeerDiff can't be None -> None"),
-            };
-
-            // Grab the peer name from either the new data, or the historical data (if the peer is removed).
-            let peer_hostname = match diff.new {
-                Some(peer) => Some(peer.name.clone()),
-                _ => store
-                    .peers()
-                    .iter()
-                    .find(|p| p.public_key == public_key)
-                    .map(|p| p.name.clone()),
-            };
-            let peer_name = peer_hostname.as_deref().unwrap_or("[unknown]");
-
-            log::info!(
-                "  peer {} ({}...) was {}.",
-                peer_name.yellow(),
-                &public_key[..10].dimmed(),
-                text
-            );
-
-            for change in diff.changes() {
-                log::debug!("    {}", change);
-            }
-        })
+        .iter()
+        .inspect(|diff| util::print_peer_diff(&store, diff))
+        .cloned()
         .map(PeerConfigBuilder::from)
         .collect::<Vec<_>>();
 
@@ -566,10 +515,44 @@ fn fetch(
     } else {
         log::info!("{}", "peers are already up to date.".green());
     }
+
     store.set_cidrs(cidrs);
-    store.update_peers(peers)?;
+    store.update_peers(&peers)?;
     store.write().with_str(interface.to_string())?;
 
+    let candidates = wg::get_local_addrs()?
+        .into_iter()
+        .map(|addr| SocketAddr::from((addr, device.listen_port.unwrap_or(51820))).into())
+        .take(10)
+        .collect::<Vec<Endpoint>>();
+    log::info!(
+        "reporting {} interface address{} as NAT traversal candidates...",
+        candidates.len(),
+        if candidates.len() == 1 { "" } else { "es" }
+    );
+    log::debug!("candidates: {:?}", candidates);
+    match Api::new(&config.server).http_form::<_, ()>("PUT", "/user/candidates", &candidates) {
+        Err(ureq::Error::Status(404, _)) => {
+            log::warn!("your network is using an old version of innernet-server that doesn't support NAT traversal candidate reporting.")
+        },
+        Err(e) => return Err(e.into()),
+        _ => {},
+    }
+
+    log::debug!("viable ICE candidates: {:?}", candidates);
+
+    let mut nat_traverse = NatTraverse::new(interface, network.backend, &modifications)?;
+    loop {
+        if nat_traverse.is_finished() {
+            break;
+        }
+        log::info!(
+            "Attempting to establish connection with {} remaining unconnected peers...",
+            nat_traverse.remaining()
+        );
+        nat_traverse.step()?;
+    }
+
     Ok(())
 }
 
@@ -992,7 +975,9 @@ fn print_peer(peer: &PeerState, short: bool, level: usize) {
     let pad = level * 2;
     let PeerState { peer, info } = peer;
     if short {
-        let connected = PeerDiff::peer_recently_connected(info);
+        let connected = info
+            .map(|info| !info.is_recently_connected())
+            .unwrap_or_default();
 
         println_pad!(
             pad,

client/src/nat.rs

@@ -0,0 +1,125 @@
+//! ICE-like NAT traversal logic.
+//!
+//! Doesn't follow the specific ICE protocol, but takes great inspiration from RFC 8445
+//! and applies it to a protocol more specific to innernet.
+
+use std::time::{Duration, Instant};
+
+use anyhow::Error;
+use shared::{
+    wg::{DeviceExt, PeerInfoExt},
+    Endpoint, Peer, PeerDiff,
+};
+use wgctrl::{Backend, Device, DeviceUpdate, InterfaceName, Key, PeerConfigBuilder};
+
+const STEP_INTERVAL: Duration = Duration::from_secs(5);
+
+pub struct NatTraverse<'a> {
+    interface: &'a InterfaceName,
+    backend: Backend,
+    remaining: Vec<Peer>,
+}
+
+impl<'a> NatTraverse<'a> {
+    pub fn new(interface: &'a InterfaceName, backend: Backend, diffs: &[PeerDiff]) -> Result<Self, Error> {
+        let mut remaining: Vec<_> = diffs.iter().filter_map(|diff| diff.new).cloned().collect();
+
+        for peer in &mut remaining {
+            // Limit reported alternative candidates to 10.
+            peer.candidates.truncate(10);
+
+            // remove server-reported endpoint from elsewhere in the list if it existed.
+            let endpoint = peer.endpoint.clone();
+            peer.candidates
+                .retain(|addr| Some(addr) != endpoint.as_ref());
+        }
+        let mut nat_traverse = Self {
+            interface,
+            backend,
+            remaining,
+        };
+        nat_traverse.refresh_remaining()?;
+        Ok(nat_traverse)
+    }
+
+    pub fn is_finished(&self) -> bool {
+        self.remaining.is_empty()
+    }
+
+    pub fn remaining(&self) -> usize {
+        self.remaining.len()
+    }
+
+    /// Refreshes the current state of candidate traversal attempts, returning
+    /// the peers that have been exhausted of all options (not included are
+    /// peers that have successfully connected, or peers removed from the interface).
+    fn refresh_remaining(&mut self) -> Result<Vec<Peer>, Error> {
+        let device = Device::get(self.interface, self.backend)?;
+        // Remove connected and missing peers
+        self.remaining.retain(|peer| {
+            if let Some(peer_info) = device.get_peer(&peer.public_key) {
+                let recently_connected = peer_info.is_recently_connected();
+                if recently_connected {
+                    log::debug!(
+                        "peer {} removed from NAT traverser (connected!).",
+                        peer.name
+                    );
+                }
+                !recently_connected
+            } else {
+                log::debug!(
+                    "peer {} removed from NAT traverser (no longer on interface).",
+                    peer.name
+                );
+                false
+            }
+        });
+        let (exhausted, remaining): (Vec<_>, Vec<_>) = self
+            .remaining
+            .drain(..)
+            .partition(|peer| peer.candidates.is_empty());
+        self.remaining = remaining;
+        Ok(exhausted)
+    }
+
+    pub fn step(&mut self) -> Result<(), Error> {
+        let exhuasted = self.refresh_remaining()?;
+
+        // Reset peer endpoints that had no viable candidates back to the server-reported one, if it exists.
+        let reset_updates = exhuasted
+            .into_iter()
+            .filter_map(|peer| set_endpoint(&peer.public_key, peer.endpoint.as_ref()));
+
+        // Set all peers' endpoints to their next available candidate.
+        let candidate_updates = self.remaining.iter_mut().filter_map(|peer| {
+            let endpoint = peer.candidates.pop();
+            set_endpoint(&peer.public_key, endpoint.as_ref())
+        });
+
+        let updates: Vec<_> = reset_updates.chain(candidate_updates).collect();
+
+        DeviceUpdate::new()
+            .add_peers(&updates)
+            .apply(self.interface, self.backend)?;
+
+        let start = Instant::now();
+        while start.elapsed() < STEP_INTERVAL {
+            self.refresh_remaining()?;
+            if self.is_finished() {
+                log::debug!("NAT traverser is finished!");
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(100));
+        }
+        Ok(())
+    }
+}
+
+/// Return a PeerConfigBuilder if an endpoint exists and resolves successfully.
+fn set_endpoint(public_key: &str, endpoint: Option<&Endpoint>) -> Option<PeerConfigBuilder> {
+    endpoint
+        .and_then(|endpoint| endpoint.resolve().ok())
+        .map(|addr| {
+            PeerConfigBuilder::new(&Key::from_base64(public_key).unwrap()).set_endpoint(addr)
+        })
+}