Listen on netlink socket to monitor offline status

3 files changed, 147 insertions, 5 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 4a642f31b6..52030fc501 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1770,6 +1770,7 @@ dependencies = [
  "libc 0.2.44 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "mnl 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "netlink-socket 0.0.2 (git+https://github.com/mullvad/netlink?branch=best-effort-nla-parsing)",
  "nftnl 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "nix 0.12.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "notify 4.0.6 (registry+https://github.com/rust-lang/crates.io-index)",
diff --git a/talpid-core/Cargo.toml b/talpid-core/Cargo.toml
index ee3b17ad20..a393cb846d 100644
--- a/talpid-core/Cargo.toml
+++ b/talpid-core/Cargo.toml
@@ -34,6 +34,7 @@ nix = "0.12"
 dbus = "0.6"
 failure = "0.1"
 iproute2 = { git = "https://github.com/mullvad/netlink", branch = "best-effort-nla-parsing" }
+netlink-socket = { git = "https://github.com/mullvad/netlink", branch = "best-effort-nla-parsing" }
 notify = "4.0"
 resolv-conf = "0.6.1"
 rtnetlink = { git = "https://github.com/mullvad/netlink", branch = "best-effort-nla-parsing" }
diff --git a/talpid-core/src/offline/linux.rs b/talpid-core/src/offline/linux.rs
index 4b76120fb4..68792f39bb 100644
--- a/talpid-core/src/offline/linux.rs
+++ b/talpid-core/src/offline/linux.rs
@@ -1,12 +1,20 @@
 extern crate iproute2;
+extern crate netlink_socket;
 extern crate rtnetlink;
 
+use std::collections::BTreeSet;
+use std::thread;
+
 use error_chain::ChainedError;
-use futures::{future::Either, sync::mpsc::UnboundedSender, Future};
-use log::warn;
+use futures::{future::Either, sync::mpsc::UnboundedSender, Future, Stream};
+use log::{error, trace, warn};
 
 use self::iproute2::Link;
-use self::rtnetlink::LinkLayerType;
+use self::netlink_socket::{Protocol, SocketAddr, TokioSocket};
+use self::rtnetlink::{
+    LinkFlags, LinkHeader, LinkLayerType, LinkMessage, NetlinkCodec, NetlinkFramed, NetlinkMessage,
+    RtnlMessage,
+};
 
 use tunnel_state_machine::TunnelCommand;
 
@@ -18,6 +26,9 @@ error_chain! {
         NetlinkConnectionError {
             description("Failed to connect to netlink socket")
         }
+        NetlinkBindError {
+            description("Failed to start listening on netlink socket")
+        }
         NetlinkError {
             description("Error while communicating on the netlink socket")
         }
@@ -27,9 +38,28 @@ error_chain! {
     }
 }
 
+const RTMGRP_NOTIFY: u32 = 1;
+const RTMGRP_LINK: u32 = 2;
+
 pub struct MonitorHandle;
 
-pub fn spawn_monitor(_sender: UnboundedSender<TunnelCommand>) -> Result<MonitorHandle> {
+pub fn spawn_monitor(sender: UnboundedSender<TunnelCommand>) -> Result<MonitorHandle> {
+    let mut socket =
+        TokioSocket::new(Protocol::Route).chain_err(|| ErrorKind::NetlinkConnectionError)?;
+    socket
+        .bind(&SocketAddr::new(0, RTMGRP_NOTIFY | RTMGRP_LINK))
+        .chain_err(|| ErrorKind::NetlinkBindError)?;
+
+    let channel = NetlinkFramed::new(socket, NetlinkCodec::<NetlinkMessage>::new());
+    let link_monitor = LinkMonitor::new(sender)?;
+
+    thread::spawn(|| {
+        if let Err(error) = monitor_event_loop(channel, link_monitor) {
+            let chained_error = error.chain_err(|| "Error running link monitor event loop");
+            error!("{}", chained_error.display_chain());
+        }
+    });
+
     Ok(MonitorHandle)
 }
 
@@ -49,7 +79,7 @@ fn list_links_providing_connectivity() -> Result<impl Iterator<Item = Link>> {
     Ok(list_links()?.into_iter().filter(link_provides_connectivity))
 }
 
-fn link_provides_connectivity(link: &Link) -> bool {
+fn link_provides_connectivity(link: &impl BasicLinkInfo) -> bool {
     // Some tunnels have the link layer type set to None
     link.link_layer_type() != LinkLayerType::Loopback
         && link.link_layer_type() != LinkLayerType::None
@@ -71,3 +101,113 @@ fn list_links() -> Result<Vec<Link>> {
         )),
     }
 }
+
+fn monitor_event_loop(
+    channel: NetlinkFramed<NetlinkCodec<NetlinkMessage>>,
+    mut link_monitor: LinkMonitor,
+) -> Result<()> {
+    channel
+        .for_each(|(message, _address)| {
+            let (_header, payload) = message.into_parts();
+
+            match payload {
+                RtnlMessage::NewLink(link_message) => link_monitor.new_link(link_message),
+                RtnlMessage::DelLink(link_message) => link_monitor.del_link(link_message),
+                _ => trace!("Ignoring unknown link message"),
+            }
+
+            Ok(())
+        })
+        .wait()
+        .map_err(|error| {
+            Error::with_chain(failure::Fail::compat(error), ErrorKind::NetlinkError)
+        })?;
+
+    Ok(())
+}
+
+struct LinkMonitor {
+    is_offline: bool,
+    running_links: BTreeSet<u32>,
+    sender: UnboundedSender<TunnelCommand>,
+}
+
+impl LinkMonitor {
+    pub fn new(sender: UnboundedSender<TunnelCommand>) -> Result<Self> {
+        let links: Vec<Link> = list_links_providing_connectivity()?.collect();
+        let is_offline = links.is_empty();
+        let running_links = links.into_iter().map(|link| link.index()).collect();
+
+        Ok(LinkMonitor {
+            is_offline,
+            running_links,
+            sender,
+        })
+    }
+
+    pub fn new_link(&mut self, link_message: LinkMessage) {
+        if self.is_offline && link_provides_connectivity(link_message.header()) {
+            self.set_is_offline(false);
+        }
+
+        if let Ok(link) = Link::from_link_message(link_message) {
+            if link_provides_connectivity(&link) {
+                self.insert_link(link.index());
+            } else {
+                self.remove_link(link.index());
+            }
+        }
+    }
+
+    pub fn del_link(&mut self, link_message: LinkMessage) {
+        if let Ok(link) = Link::from_link_message(link_message) {
+            self.remove_link(link.index());
+        }
+    }
+
+    fn set_is_offline(&mut self, is_offline: bool) {
+        if self.is_offline != is_offline {
+            self.is_offline = is_offline;
+            let _ = self
+                .sender
+                .unbounded_send(TunnelCommand::IsOffline(is_offline));
+        }
+    }
+
+    fn insert_link(&mut self, link_index: u32) {
+        self.running_links.insert(link_index);
+        self.set_is_offline(false);
+    }
+
+    fn remove_link(&mut self, link_index: u32) {
+        self.running_links.remove(&link_index);
+        if self.running_links.is_empty() {
+            self.set_is_offline(true);
+        }
+    }
+}
+
+trait BasicLinkInfo {
+    fn flags(&self) -> LinkFlags;
+    fn link_layer_type(&self) -> LinkLayerType;
+}
+
+impl BasicLinkInfo for Link {
+    fn flags(&self) -> LinkFlags {
+        self.flags()
+    }
+
+    fn link_layer_type(&self) -> LinkLayerType {
+        self.link_layer_type()
+    }
+}
+
+impl BasicLinkInfo for LinkHeader {
+    fn flags(&self) -> LinkFlags {
+        self.flags()
+    }
+
+    fn link_layer_type(&self) -> LinkLayerType {
+        self.link_layer_type()
+    }
+}