Use the route manager for the Linux offline monitor

5 files changed, 51 insertions, 177 deletions

diff --git a/talpid-core/src/dns/linux/mod.rs b/talpid-core/src/dns/linux/mod.rs
index 676ad0f0f0..4b046035aa 100644
--- a/talpid-core/src/dns/linux/mod.rs
+++ b/talpid-core/src/dns/linux/mod.rs
@@ -51,8 +51,8 @@ impl super::DnsMonitorT for DnsMonitor {
 
     fn new(
         handle: tokio::runtime::Handle,
-        route_manager: RouteManagerHandle,
         _cache_dir: impl AsRef<Path>,
+        route_manager: RouteManagerHandle,
     ) -> Result<Self> {
         Ok(DnsMonitor {
             route_manager,
diff --git a/talpid-core/src/dns/mod.rs b/talpid-core/src/dns/mod.rs
index 9eb1ef1c13..229896cf98 100644
--- a/talpid-core/src/dns/mod.rs
+++ b/talpid-core/src/dns/mod.rs
@@ -32,11 +32,16 @@ impl DnsMonitor {
     /// Returns a new `DnsMonitor` that can set and monitor the system DNS.
     pub fn new(
         handle: tokio::runtime::Handle,
-        route_manager: RouteManagerHandle,
         cache_dir: impl AsRef<Path>,
+        #[cfg(target_os = "linux")] route_manager: RouteManagerHandle,
     ) -> Result<Self, Error> {
         Ok(DnsMonitor {
-            inner: imp::DnsMonitor::new(handle, route_manager, cache_dir)?,
+            inner: imp::DnsMonitor::new(
+                handle,
+                cache_dir,
+                #[cfg(target_os = "linux")]
+                route_manager,
+            )?,
         })
     }
 
@@ -66,8 +71,8 @@ trait DnsMonitorT: Sized {
 
     fn new(
         handle: tokio::runtime::Handle,
-        route_manager: RouteManagerHandle,
         cache_dir: impl AsRef<Path>,
+        #[cfg(target_os = "linux")] route_manager: RouteManagerHandle,
     ) -> Result<Self, Self::Error>;
 
     fn set(&mut self, interface: &str, servers: &[IpAddr]) -> Result<(), Self::Error>;
diff --git a/talpid-core/src/offline/linux.rs b/talpid-core/src/offline/linux.rs
index cf255b0541..ceaa864cc7 100644
--- a/talpid-core/src/offline/linux.rs
+++ b/talpid-core/src/offline/linux.rs
@@ -1,21 +1,12 @@
-use crate::tunnel_state_machine::TunnelCommand;
-use futures::{
-    channel::{mpsc::UnboundedSender, oneshot},
-    FutureExt, StreamExt, TryStream, TryStreamExt,
+use crate::{
+    routing::{self, RouteManagerHandle},
+    tunnel_state_machine::TunnelCommand,
 };
-use netlink_packet_core::{NetlinkPayload, NLM_F_REQUEST};
-use netlink_packet_route::{
-    rtnl::route::nlas::Nla as RouteNla, NetlinkMessage, RouteFlags, RouteMessage, RtnlMessage,
+use futures::{channel::mpsc::UnboundedSender, StreamExt};
+use std::{
+    net::{IpAddr, Ipv4Addr},
+    sync::Weak,
 };
-use rtnetlink::{
-    constants::{
-        RTMGRP_IPV4_IFADDR, RTMGRP_IPV4_ROUTE, RTMGRP_IPV6_IFADDR, RTMGRP_IPV6_ROUTE, RTMGRP_LINK,
-        RTMGRP_NOTIFY,
-    },
-    sys::SocketAddr,
-    Handle, IpVersion,
-};
-use std::{io, net::Ipv4Addr, sync::Weak};
 use talpid_types::ErrorExt;
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -23,52 +14,21 @@ pub type Result<T> = std::result::Result<T, Error>;
 #[derive(err_derive::Error, Debug)]
 #[error(no_from)]
 pub enum Error {
-    #[error(display = "Failed to resolve output interface index")]
-    GetLinkError(#[error(source)] failure::Compat<rtnetlink::Error>),
-
-    #[error(display = "No netlink response for output interface query")]
-    NoLinkError,
-
-    #[error(display = "Failed to get list of IP addresses")]
-    GetAddressesError(#[error(source)] failure::Compat<rtnetlink::Error>),
-
-    #[error(display = "Failed to get a route for an arbitrary IP address")]
-    GetRouteError(#[error(source)] failure::Compat<rtnetlink::Error>),
-
-    #[error(display = "No netlink response for route query")]
-    NoRouteError,
-
-    #[error(display = "Failed to connect to netlink socket")]
-    NetlinkConnectionError(#[error(source)] io::Error),
-
-    #[error(display = "Failed to connect to bind to netlink socket")]
-    BindError(#[error(source)] io::Error),
-
-    #[error(display = "Failed to start listening on netlink socket")]
-    NetlinkBindError(#[error(source)] io::Error),
-
-    #[error(display = "Error while processing netlink messages")]
-    MonitorNetlinkError,
-
-    #[error(display = "Netlink connection has unexpectedly disconnected")]
-    NetlinkDisconnected,
-
-    #[error(display = "Failed to initialize event loop")]
-    EventLoopError(#[error(source)] io::Error),
+    #[error(display = "The route manager returned an error")]
+    RouteManagerError(#[error(source)] routing::Error),
 }
 
 pub struct MonitorHandle {
-    handle: rtnetlink::Handle,
-    _stop_connection_tx: oneshot::Sender<()>,
+    route_manager: RouteManagerHandle,
 }
 
 // Mullvad API's public IP address, correct at the time of writing, but any public IP address will
 // work.
-const PUBLIC_INTERNET_ADDRESS: Ipv4Addr = Ipv4Addr::new(193, 138, 218, 78);
+const PUBLIC_INTERNET_ADDRESS: IpAddr = IpAddr::V4(Ipv4Addr::new(193, 138, 218, 78));
 
 impl MonitorHandle {
     pub async fn is_offline(&mut self) -> bool {
-        match public_ip_unreachable(&self.handle).await {
+        match public_ip_unreachable(&self.route_manager).await {
             Ok(is_offline) => is_offline,
             Err(err) => {
                 log::error!(
@@ -81,46 +41,28 @@ impl MonitorHandle {
     }
 }
 
-pub async fn spawn_monitor(sender: Weak<UnboundedSender<TunnelCommand>>) -> Result<MonitorHandle> {
-    let (mut connection, handle, mut messages) =
-        rtnetlink::new_connection().map_err(Error::NetlinkConnectionError)?;
-
-    let mgroup_flags = RTMGRP_IPV4_IFADDR
-        | RTMGRP_IPV4_ROUTE
-        | RTMGRP_IPV6_IFADDR
-        | RTMGRP_IPV6_ROUTE
-        | RTMGRP_LINK
-        | RTMGRP_NOTIFY;
-    let addr = SocketAddr::new(0, mgroup_flags);
-
-    connection
-        .socket_mut()
-        .bind(&addr)
-        .map_err(Error::BindError)?;
+pub async fn spawn_monitor(
+    sender: Weak<UnboundedSender<TunnelCommand>>,
+    route_manager: RouteManagerHandle,
+) -> Result<MonitorHandle> {
+    let mut is_offline = public_ip_unreachable(&route_manager).await?;
 
-    let (stop_connection_tx, stop_rx) = oneshot::channel();
-
-    // Connection will be closed once the channel is dropped
-    tokio::spawn(async {
-        futures::select! {
-            _ = connection.fuse() => (),
-            _ = stop_rx.fuse() => (),
-        }
-    });
-    let mut is_offline = public_ip_unreachable(&handle).await?;
+    let mut listener = route_manager
+        .change_listener()
+        .await
+        .map_err(Error::RouteManagerError)?;
 
     let monitor_handle = MonitorHandle {
-        handle: handle.clone(),
-        _stop_connection_tx: stop_connection_tx,
+        route_manager: route_manager.clone(),
     };
 
-
     tokio::spawn(async move {
-        while let Some(_new_message) = messages.next().await {
+        while let Some(_event) = listener.next().await {
             match sender.upgrade() {
                 Some(sender) => {
-                    let new_offline_state =
-                        public_ip_unreachable(&handle).await.unwrap_or_else(|err| {
+                    let new_offline_state = public_ip_unreachable(&route_manager)
+                        .await
+                        .unwrap_or_else(|err| {
                             log::error!(
                                 "{}",
                                 err.display_chain_with_msg("Failed to infer offline state")
@@ -141,89 +83,10 @@ pub async fn spawn_monitor(sender: Weak<UnboundedSender<TunnelCommand>>) -> Resu
 }
 
 
-async fn public_ip_unreachable(handle: &Handle) -> Result<bool> {
-    let mut request = handle.route().get(IpVersion::V4);
-    let message = request.message_mut();
-    message
-        .nlas
-        .push(RouteNla::Mark(crate::linux::TUNNEL_FW_MARK));
-    message.nlas.push(RouteNla::Destination(
-        PUBLIC_INTERNET_ADDRESS.octets().to_vec(),
-    ));
-    message.header.destination_prefix_length = 32;
-    message.header.flags = RouteFlags::RTM_F_LOOKUP_TABLE;
-    let mut stream = execute_route_get_request(handle.clone(), message.clone());
-    match stream.try_next().await {
-        // Presance of any route implies connectivity, even if it's a loopback route
-        Ok(Some(_)) => Ok(false),
-        Ok(None) => Err(Error::NoRouteError),
-        // ENETUNREACH implies that there exists no route that'd reach our random API address,
-        // as such, the host is assumed to be offline
-        Err(rtnetlink::Error::NetlinkError(nl_err)) if nl_err.code == -libc::ENETUNREACH => {
-            Ok(true)
-        }
-        Err(err) => Err(Error::GetRouteError(failure::Fail::compat(err))),
-    }
-}
-
-pub fn execute_route_get_request(
-    mut handle: Handle,
-    message: RouteMessage,
-) -> impl TryStream<Ok = RouteMessage, Error = rtnetlink::Error> {
-    use futures::future::{self, Either};
-    use rtnetlink::Error;
-
-    let mut req = NetlinkMessage::from(RtnlMessage::GetRoute(message));
-    req.header.flags = NLM_F_REQUEST;
-
-    match handle.request(req) {
-        Ok(response) => Either::Left(response.map(move |msg| {
-            let (header, payload) = msg.into_parts();
-            match payload {
-                NetlinkPayload::InnerMessage(RtnlMessage::NewRoute(msg)) => Ok(msg),
-                NetlinkPayload::Error(err) => Err(Error::NetlinkError(err)),
-                _ => Err(Error::UnexpectedMessage(NetlinkMessage::new(
-                    header, payload,
-                ))),
-            }
-        })),
-        Err(e) => Either::Right(future::err::<RouteMessage, Error>(e).into_stream()),
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-    use rtnetlink::{
-        constants::{
-            RTMGRP_IPV4_IFADDR, RTMGRP_IPV4_ROUTE, RTMGRP_IPV6_IFADDR, RTMGRP_IPV6_ROUTE,
-            RTMGRP_LINK, RTMGRP_NOTIFY,
-        },
-        sys::SocketAddr,
-    };
-
-    #[test]
-    fn test_route_table_query() {
-        let mut runtime = tokio::runtime::Runtime::new().expect("failed to initialize runtime");
-        let (mut connection, handle, _) = runtime.block_on(async {
-            rtnetlink::new_connection()
-                .map_err(Error::NetlinkConnectionError)
-                .expect("Failed to create a netlink connection")
-        });
-
-        let mgroup_flags = RTMGRP_IPV4_IFADDR
-            | RTMGRP_IPV4_ROUTE
-            | RTMGRP_IPV6_IFADDR
-            | RTMGRP_IPV6_ROUTE
-            | RTMGRP_LINK
-            | RTMGRP_NOTIFY;
-        let addr = SocketAddr::new(0, mgroup_flags);
-
-        connection.socket_mut().bind(&addr).unwrap();
-        runtime.spawn(connection);
-
-        runtime
-            .block_on(public_ip_unreachable(&handle))
-            .expect("Failed to query routing table");
-    }
+async fn public_ip_unreachable(handle: &RouteManagerHandle) -> Result<bool> {
+    Ok(handle
+        .get_destination_route(PUBLIC_INTERNET_ADDRESS, true)
+        .await
+        .map_err(Error::RouteManagerError)?
+        .is_none())
 }
diff --git a/talpid-core/src/offline/mod.rs b/talpid-core/src/offline/mod.rs
index 413646a24f..ac8b10e222 100644
--- a/talpid-core/src/offline/mod.rs
+++ b/talpid-core/src/offline/mod.rs
@@ -1,4 +1,6 @@
-use crate::{routing::RouteManagerHandle, tunnel_state_machine::TunnelCommand};
+#[cfg(target_os = "linux")]
+use crate::routing::RouteManagerHandle;
+use crate::tunnel_state_machine::TunnelCommand;
 use futures::channel::mpsc::UnboundedSender;
 use std::sync::Weak;
 #[cfg(target_os = "android")]
@@ -41,14 +43,16 @@ impl MonitorHandle {
 }
 
 pub async fn spawn_monitor(
-    route_manager: RouteManagerHandle,
     sender: Weak<UnboundedSender<TunnelCommand>>,
+    #[cfg(target_os = "linux")] route_manager: RouteManagerHandle,
     #[cfg(target_os = "android")] android_context: AndroidContext,
 ) -> Result<MonitorHandle, Error> {
     let monitor = if !*FORCE_DISABLE_OFFLINE_MONITOR {
         Some(
             imp::spawn_monitor(
                 sender,
+                #[cfg(target_os = "linux")]
+                route_manager,
                 #[cfg(target_os = "android")]
                 android_context,
             )
diff --git a/talpid-core/src/tunnel_state_machine/mod.rs b/talpid-core/src/tunnel_state_machine/mod.rs
index c6acb0ed46..ee458db680 100644
--- a/talpid-core/src/tunnel_state_machine/mod.rs
+++ b/talpid-core/src/tunnel_state_machine/mod.rs
@@ -220,17 +220,19 @@ impl TunnelStateMachine {
             .map_err(Error::InitRouteManagerError)?;
         let dns_monitor = DnsMonitor::new(
             runtime.clone(),
+            cache_dir,
+            #[cfg(target_os = "linux")]
             route_manager
                 .handle()
                 .map_err(Error::InitRouteManagerError)?,
-            cache_dir,
         )
         .map_err(Error::InitDnsMonitorError)?;
         let mut offline_monitor = offline::spawn_monitor(
+            command_tx,
+            #[cfg(target_os = "linux")]
             route_manager
                 .handle()
                 .map_err(Error::InitRouteManagerError)?,
-            command_tx,
             #[cfg(target_os = "android")]
             android_context,
         )