Use route monitor to drive the offline monitor

5 files changed, 87 insertions, 186 deletions

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b46b3caa44..b31b2df964 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -81,6 +81,9 @@ Line wrap the file at 100 chars.                                              Th
 #### macOS
 - Resolve issues with the app blocking internet connectivity after sleep or when connecting to new
   wireless networks.
+- Fix issue where the app would get stuck in offline state after a reboot or a reinstall by using
+  `route monitor` instead of relying on `SCNetworkReachability` API to infer whether the host is
+  offline.
 
 #### Windows
 - Fix app size after changing display scale.
diff --git a/talpid-core/src/offline/macos.rs b/talpid-core/src/offline/macos.rs
index 884a7f1d38..46329f5585 100644
--- a/talpid-core/src/offline/macos.rs
+++ b/talpid-core/src/offline/macos.rs
@@ -1,40 +1,11 @@
-use futures::channel::mpsc::UnboundedSender;
-use std::{
-    net::{Ipv4Addr, SocketAddr},
-    sync::{
-        atomic::{AtomicBool, Ordering},
-        mpsc, Arc, Weak,
-    },
-    thread,
-};
-use system_configuration::{
-    core_foundation::{
-        array::CFArray,
-        base::{CFType, TCFType, ToVoid},
-        boolean::CFBoolean,
-        dictionary::CFDictionary,
-        runloop::{kCFRunLoopCommonModes, CFRunLoop},
-        string::CFString,
-    },
-    dynamic_store::{SCDynamicStore, SCDynamicStoreBuilder, SCDynamicStoreCallBackContext},
-    network_configuration::{self, SCNetworkInterface, SCNetworkInterfaceType},
-    network_reachability::{
-        ReachabilityFlags, SCNetworkReachability, SchedulingError, SetCallbackError,
-    },
-};
-
-const PRIMARY_INTERFACE_KEY: &str = "State:/Network/Global/IPv4";
+use futures::{channel::mpsc::UnboundedSender, Future, StreamExt};
+use std::sync::{Arc, Weak};
+use talpid_types::ErrorExt;
 
 #[derive(err_derive::Error, Debug)]
 pub enum Error {
-    #[error(display = "Failed to initialize dynamic store")]
-    DynamicStoreInitError,
-    #[error(display = "Failed to schedule reachability callback")]
-    ScheduleReachabilityCallbackError(#[error(source)] SchedulingError),
-    #[error(display = "Failed to set reachability callback")]
-    SetCallbackError(#[error(source)] SetCallbackError),
-    #[error(display = "Panic during initialization")]
-    InitializationError,
+    #[error(display = "Failed to initialize route monitor")]
+    StartMonitorError(#[error(source)] crate::routing::PlatformError),
 }
 
 pub struct MonitorHandle {
@@ -45,177 +16,83 @@ impl MonitorHandle {
     /// Host is considered to be offline if the IPv4 internet is considered to be unreachable by the
     /// given reachability flags *or* there are no active physical interfaces.
     pub async fn is_offline(&self) -> bool {
-        let reachability = SCNetworkReachability::from(ipv4_internet());
-        let store = SCDynamicStoreBuilder::new("talpid-offline-check").build();
-        reachability
-            .reachability()
-            .map(|flags| check_offline_state(&store, flags))
-            .unwrap_or(false)
+        !exists_non_tunnel_default_route().await
     }
 }
 
+async fn exists_non_tunnel_default_route() -> bool {
+    match crate::routing::get_default_routes().await {
+        Ok((Some(node), _)) | Ok((None, Some(node))) => {
+            let route_exists = node
+                .get_device()
+                .map(|iface_name| !iface_name.contains("tun"))
+                .unwrap_or(true);
+            log::debug!("Assuming non-tunnel default route exists due to {:?}", node);
+            route_exists
+        }
+        Ok((None, None)) => {
+            log::debug!("No default routes exist, assuming machine is offline");
+            false
+        }
+        Err(err) => {
+            log::error!(
+                "{}",
+                err.display_chain_with_msg(
+                    "Failed to obtain default routes, assuming machine is online."
+                )
+            );
+            true
+        }
+    }
+}
 pub async fn spawn_monitor(notify_tx: UnboundedSender<bool>) -> Result<MonitorHandle, Error> {
-    let (result_tx, result_rx) = mpsc::channel();
     let notify_tx = Arc::new(notify_tx);
-    let sender = Arc::downgrade(&notify_tx);
-    thread::spawn(move || {
-        let mut reachability_ref = SCNetworkReachability::from(ipv4_internet());
-        let store = SCDynamicStoreBuilder::new("talpid-offline-watcher").build();
-
-        let is_currently_offline = match reachability_ref.reachability() {
-            Ok(flags) => check_offline_state(&store, flags),
-            Err(_) => {
-                log::error!("Failed to obtain current connectivity, assuming machine is online");
-                false
-            }
-        };
-
-        let context = OfflineStateContext {
-            sender,
-            is_offline: Arc::new(AtomicBool::new(is_currently_offline)),
-        };
-
-        let result = || -> Result<SCDynamicStore, Error> {
-            let dynamic_store = create_dynamic_store(context.clone())?;
-            CFRunLoop::get_current().add_source(&dynamic_store.create_run_loop_source(), unsafe {
-                kCFRunLoopCommonModes
-            });
 
-            reachability_ref.set_callback(move |flags| {
-                let store = SCDynamicStoreBuilder::new("talpid-offline-watcher").build();
-                context.new_state(check_offline_state(&store, flags));
-            })?;
-
-            reachability_ref.schedule_with_runloop(&CFRunLoop::get_current(), unsafe {
-                kCFRunLoopCommonModes
-            })?;
-
-            Ok(dynamic_store)
-        };
-
-        match result() {
-            Ok(_dynamic_store) => {
-                let _ = result_tx.send(Ok(()));
-                CFRunLoop::run_current()
-            }
-            Err(err) => {
-                let _ = result_tx.send(Err(err));
-            }
-        }
-    });
+    let context = OfflineStateContext {
+        sender: Arc::downgrade(&notify_tx),
+        is_offline: !exists_non_tunnel_default_route().await,
+    };
 
-    let _ = result_rx.recv().map_err(|_| Error::InitializationError)??;
+    let route_monitor = watch_route_monitor(context)?;
+    tokio::spawn(route_monitor);
     Ok(MonitorHandle {
         _notify_tx: notify_tx,
     })
 }
 
-fn ipv4_internet() -> SocketAddr {
-    SocketAddr::new(Ipv4Addr::UNSPECIFIED.into(), 0)
-}
-
-fn check_offline_state(store: &SCDynamicStore, flags: ReachabilityFlags) -> bool {
-    let is_offline =
-        !flags.contains(ReachabilityFlags::REACHABLE) || !exists_active_physical_iface(store);
-    is_offline
-}
+fn watch_route_monitor(
+    mut context: OfflineStateContext,
+) -> Result<impl Future<Output = ()>, Error> {
+    let mut monitor = crate::routing::listen_for_default_route_changes()?;
 
-fn exists_active_physical_iface(store: &SCDynamicStore) -> bool {
-    network_configuration::get_interfaces().iter().any(|iface| {
-        let is_physical = iface_is_physical(&*iface);
-        let is_active = iface_is_active(&*iface, store);
-        let is_valid = is_physical && is_active;
-        if is_valid {
-            log::trace!(
-                "Considering interface {:?} {:?} to be active and physical",
-                iface.bsd_name(),
-                iface.display_name()
-            );
+    Ok(async move {
+        while let Some(_route_change) = monitor.next().await {
+            context.new_state(!exists_non_tunnel_default_route().await);
+            if context.should_shut_down() {
+                break;
+            }
         }
-        is_valid
+        log::debug!("Stopping offline monitor");
     })
 }
 
-fn iface_is_active(iface: &SCNetworkInterface, store: &SCDynamicStore) -> bool {
-    || -> Option<bool> {
-        let path = format!("State:/Network/Interface/{}/Link", iface.bsd_name()?);
-        let link_properties = store
-            .get(CFString::from(path.as_ref()))?
-            .downcast::<CFDictionary>()?;
-
-        let active_ptr = link_properties.find(CFString::from("Active").to_void())?;
-        if active_ptr.is_null() {
-            return None;
-        }
-
-        unsafe { CFType::wrap_under_get_rule(*active_ptr) }
-            .downcast::<CFBoolean>()
-            .map(Into::into)
-    }()
-    .unwrap_or(false)
-}
-
-fn iface_is_physical(iface: &SCNetworkInterface) -> bool {
-    use SCNetworkInterfaceType::*;
-    match iface.interface_type() {
-        Some(iface_type) => match iface_type {
-            Bluetooth | Modem | Serial | IrDA | Ethernet | FireWire | WWAN | IEEE80211 => true,
-            _ => false,
-        },
-        // if interface type is unknown, have to assume it provides internet
-        None => true,
-    }
-}
-
 #[derive(Clone)]
 struct OfflineStateContext {
     sender: Weak<UnboundedSender<bool>>,
-    is_offline: Arc<AtomicBool>,
+    is_offline: bool,
 }
 
 impl OfflineStateContext {
-    fn no_primary_interface(&self) {
-        self.new_state(true);
+    fn should_shut_down(&self) -> bool {
+        self.sender.upgrade().is_none()
     }
 
-    fn new_state(&self, is_offline: bool) {
-        if self.is_offline.swap(is_offline, Ordering::SeqCst) != is_offline {
+    fn new_state(&mut self, is_offline: bool) {
+        if self.is_offline != is_offline {
+            self.is_offline = is_offline;
             if let Some(sender) = self.sender.upgrade() {
                 let _ = sender.unbounded_send(is_offline);
             }
         }
     }
 }
-
-fn create_dynamic_store(context: OfflineStateContext) -> Result<SCDynamicStore, Error> {
-    let callback_context = SCDynamicStoreCallBackContext {
-        callout: primary_interface_change_callback,
-        info: context,
-    };
-
-    let store = SCDynamicStoreBuilder::new("talpid-primary-interface")
-        .callback_context(callback_context)
-        .build();
-
-    let watch_keys = CFArray::from_CFTypes(&[CFString::new(PRIMARY_INTERFACE_KEY)]);
-    let watch_patterns: CFArray<CFString> = CFArray::from_CFTypes(&[]);
-
-    if store.set_notification_keys(&watch_keys, &watch_patterns) {
-        log::trace!("Registered for dynamic store notifications");
-        Ok(store)
-    } else {
-        Err(Error::DynamicStoreInitError)
-    }
-}
-
-fn primary_interface_change_callback(
-    store: SCDynamicStore,
-    _changed_keys: CFArray<CFString>,
-    state: &mut OfflineStateContext,
-) {
-    let is_offline = store.get(CFString::new(PRIMARY_INTERFACE_KEY)).is_none();
-    if is_offline {
-        log::debug!("No primary interface, considering host to be offline");
-        state.no_primary_interface();
-    }
-}
diff --git a/talpid-core/src/routing/macos.rs b/talpid-core/src/routing/macos.rs
index 072ba6acfd..f191551f42 100644
--- a/talpid-core/src/routing/macos.rs
+++ b/talpid-core/src/routing/macos.rs
@@ -12,6 +12,7 @@ use std::{
     net::IpAddr,
     process::{ExitStatus, Stdio},
 };
+use talpid_types::net::IpVersion;
 use tokio::{io::AsyncBufReadExt, process::Command};
 use tokio_stream::wrappers::LinesStream;
 
@@ -64,10 +65,10 @@ pub struct RouteManagerImpl {
 
 impl RouteManagerImpl {
     pub async fn new(required_routes: HashSet<RequiredRoute>) -> Result<Self> {
-        let v4_gateway = Self::get_default_node_cmd("-inet").await?;
-        let v6_gateway = Self::get_default_node_cmd("-inet6").await?;
+        let v4_gateway = Self::get_default_node(IpVersion::V4).await?;
+        let v6_gateway = Self::get_default_node(IpVersion::V6).await?;
 
-        let monitor = listen_for_default_route_changes().await?;
+        let monitor = listen_for_default_route_changes()?;
 
         let mut manager = Self {
             default_destinations: HashSet::new(),
@@ -110,8 +111,8 @@ impl RouteManagerImpl {
                 },
 
                 _result = connectivity_change.select_next_some() => {
-                    let v4_gateway = Self::get_default_node_cmd("-inet").await.unwrap_or(None);
-                    let v6_gateway = Self::get_default_node_cmd("-inet6").await.unwrap_or(None);
+                    let v4_gateway = Self::get_default_node(IpVersion::V4).await.unwrap_or(None);
+                    let v6_gateway = Self::get_default_node(IpVersion::V6).await.unwrap_or(None);
 
                     if v4_gateway != self.v4_gateway {
                         self.v4_gateway = v4_gateway;
@@ -167,10 +168,13 @@ impl RouteManagerImpl {
     }
 
     // Retrieves the node that's currently used to reach 0.0.0.0/0
-    // Arguments can be either -inet or -inet6
-    async fn get_default_node_cmd(if_family: &'static str) -> Result<Option<Node>> {
+    pub(crate) async fn get_default_node(ip_version: IpVersion) -> Result<Option<Node>> {
+        let ip_version_arg = match ip_version {
+            IpVersion::V4 => "-inet",
+            IpVersion::V6 => "-inet6",
+        };
         let mut cmd = Command::new("route");
-        cmd.arg("-n").arg("get").arg(if_family).arg("default");
+        cmd.arg("-n").arg("get").arg(ip_version_arg).arg("default");
 
         let output = cmd.output().await.map_err(Error::FailedToRunRoute)?;
         let output = String::from_utf8(output.stdout).map_err(|e| {
@@ -296,7 +300,8 @@ fn ip_vers(prefix: IpNetwork) -> &'static str {
 
 /// Returns a stream that produces an item whenever a default route is either added or deleted from
 /// the routing table.
-async fn listen_for_default_route_changes() -> Result<impl Stream<Item = std::io::Result<()>>> {
+pub(crate) fn listen_for_default_route_changes() -> Result<impl Stream<Item = std::io::Result<()>>>
+{
     let mut cmd = Command::new("route");
     cmd.arg("-n")
         .arg("monitor")
diff --git a/talpid-core/src/routing/mod.rs b/talpid-core/src/routing/mod.rs
index 8dc8dcd778..9e247859b4 100644
--- a/talpid-core/src/routing/mod.rs
+++ b/talpid-core/src/routing/mod.rs
@@ -15,6 +15,9 @@ mod imp;
 #[cfg(target_os = "linux")]
 use netlink_packet_route::rtnl::constants::RT_TABLE_MAIN;
 
+#[cfg(target_os = "macos")]
+pub(crate) use imp::{get_default_routes, listen_for_default_route_changes, PlatformError};
+
 pub use imp::{Error, RouteManager};
 
 pub use imp::RouteManagerHandle;
diff --git a/talpid-core/src/routing/unix.rs b/talpid-core/src/routing/unix.rs
index 8ec1de888b..73de23d378 100644
--- a/talpid-core/src/routing/unix.rs
+++ b/talpid-core/src/routing/unix.rs
@@ -10,6 +10,8 @@ use futures::channel::{
     oneshot,
 };
 use std::{collections::HashSet, io};
+#[cfg(target_os = "macos")]
+use talpid_types::net::IpVersion;
 
 #[cfg(target_os = "linux")]
 use futures::stream::Stream;
@@ -20,6 +22,8 @@ use std::net::IpAddr;
 #[cfg(target_os = "macos")]
 #[path = "macos.rs"]
 mod imp;
+#[cfg(target_os = "macos")]
+pub(crate) use imp::listen_for_default_route_changes;
 
 #[cfg(target_os = "linux")]
 #[path = "linux.rs"]
@@ -263,3 +267,12 @@ impl Drop for RouteManager {
         self.runtime.clone().block_on(self.stop());
     }
 }
+
+/// Returns a tuple containing a IPv4 and IPv6 default route nodes.
+#[cfg(target_os = "macos")]
+pub(crate) async fn get_default_routes() -> Result<(Option<super::Node>, Option<super::Node>), Error>
+{
+    let v4 = imp::RouteManagerImpl::get_default_node(IpVersion::V4).await?;
+    let v6 = imp::RouteManagerImpl::get_default_node(IpVersion::V6).await?;
+    Ok((v4, v6))
+}