Configure ST tunnel addresses before setting up routes

9 files changed, 110 insertions, 59 deletions

diff --git a/talpid-core/src/tunnel/mod.rs b/talpid-core/src/tunnel/mod.rs
index 3ee6bf1886..570cb64983 100644
--- a/talpid-core/src/tunnel/mod.rs
+++ b/talpid-core/src/tunnel/mod.rs
@@ -71,7 +71,7 @@ pub enum Error {
 pub enum TunnelEvent {
     /// Sent when the tunnel fails to connect due to an authentication error.
     AuthFailed(Option<String>),
-    /// Sent when the tunnel interface has been created.
+    /// Sent when the tunnel interface has been created, before routes are set up.
     InterfaceUp(TunnelMetadata),
     /// Sent when the tunnel comes up and is ready for traffic.
     Up(TunnelMetadata),
@@ -112,7 +112,7 @@ impl TunnelMonitor {
         route_manager: &mut RouteManager,
     ) -> Result<Self>
     where
-        L: (Fn(TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+        L: (Fn(TunnelEvent) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
             + Send
             + Clone
             + Sync
@@ -169,7 +169,7 @@ impl TunnelMonitor {
         route_manager: &mut RouteManager,
     ) -> Result<Self>
     where
-        L: (Fn(TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+        L: (Fn(TunnelEvent) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
             + Send
             + Sync
             + Clone
@@ -198,7 +198,7 @@ impl TunnelMonitor {
         route_manager: &mut RouteManager,
     ) -> Result<Self>
     where
-        L: (Fn(TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+        L: (Fn(TunnelEvent) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
             + Send
             + Sync
             + 'static,
diff --git a/talpid-core/src/tunnel/openvpn/mod.rs b/talpid-core/src/tunnel/openvpn/mod.rs
index 87dbc55101..f25f2624da 100644
--- a/talpid-core/src/tunnel/openvpn/mod.rs
+++ b/talpid-core/src/tunnel/openvpn/mod.rs
@@ -334,7 +334,7 @@ impl OpenVpnMonitor<OpenVpnCommand> {
         #[cfg(not(target_os = "linux"))] _route_manager: &mut routing::RouteManager,
     ) -> Result<Self>
     where
-        L: (Fn(TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+        L: (Fn(TunnelEvent) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
             + Send
             + Sync
             + 'static,
@@ -945,9 +945,11 @@ impl ProcessHandle for OpenVpnProcHandle {
 
 
 mod event_server {
+    use crate::tunnel::TunnelMetadata;
     use futures::stream::TryStreamExt;
     use parity_tokio_ipc::Endpoint as IpcEndpoint;
     use std::{
+        collections::HashMap,
         pin::Pin,
         task::{Context, Poll},
     };
@@ -981,7 +983,9 @@ mod event_server {
 
     /// Implements a gRPC service used to process events sent to by OpenVPN.
     pub struct OpenvpnEventProxyImpl<
-        L: (Fn(super::TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+        L: (Fn(
+                super::TunnelEvent,
+            ) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
             + Send
             + Sync
             + 'static,
@@ -997,12 +1001,27 @@ mod event_server {
     }
 
     impl<
-            L: (Fn(super::TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+            L: (Fn(
+                    super::TunnelEvent,
+                )
+                    -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
                 + Send
                 + Sync
                 + 'static,
         > OpenvpnEventProxyImpl<L>
     {
+        async fn up_inner(
+            &self,
+            request: Request<EventDetails>,
+        ) -> std::result::Result<Response<()>, tonic::Status> {
+            let env = request.into_inner().env;
+            (self.on_event)(super::TunnelEvent::InterfaceUp(Self::get_tunnel_metadata(
+                &env,
+            )?))
+            .await;
+            Ok(Response::new(()))
+        }
+
         async fn route_up_inner(
             &self,
             request: Request<EventDetails>,
@@ -1038,14 +1057,11 @@ mod event_server {
                 }
             }
 
-            let tunnel_alias = env
-                .get("dev")
-                .ok_or(tonic::Status::invalid_argument("missing tunnel alias"))?
-                .to_string();
+            let metadata = Self::get_tunnel_metadata(&env)?;
 
             #[cfg(windows)]
             {
-                let tunnel_device = tunnel_alias.clone();
+                let tunnel_device = metadata.interface.clone();
                 tokio::task::spawn_blocking(move || super::wait_for_ready_device(&tunnel_device))
                     .await
                     .map_err(|_| tonic::Status::internal("task failed to complete"))?
@@ -1058,6 +1074,19 @@ mod event_server {
                     })?;
             }
 
+            (self.on_event)(super::TunnelEvent::Up(metadata)).await;
+
+            Ok(Response::new(()))
+        }
+
+        fn get_tunnel_metadata(
+            env: &HashMap<String, String>,
+        ) -> std::result::Result<TunnelMetadata, tonic::Status> {
+            let tunnel_alias = env
+                .get("dev")
+                .ok_or(tonic::Status::invalid_argument("missing tunnel alias"))?
+                .to_string();
+
             let mut ips = vec![env
                 .get("ifconfig_local")
                 .ok_or(tonic::Status::invalid_argument(
@@ -1089,21 +1118,21 @@ mod event_server {
                 None
             };
 
-            (self.on_event)(super::TunnelEvent::Up(crate::tunnel::TunnelMetadata {
+            Ok(TunnelMetadata {
                 interface: tunnel_alias,
                 ips,
                 ipv4_gateway,
                 ipv6_gateway,
-            }))
-            .await;
-
-            Ok(Response::new(()))
+            })
         }
     }
 
     #[tonic::async_trait]
     impl<
-            L: (Fn(super::TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+            L: (Fn(
+                    super::TunnelEvent,
+                )
+                    -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
                 + Send
                 + Sync
                 + 'static,
@@ -1121,6 +1150,16 @@ mod event_server {
             Ok(Response::new(()))
         }
 
+        async fn up(
+            &self,
+            request: Request<EventDetails>,
+        ) -> std::result::Result<Response<()>, tonic::Status> {
+            self.up_inner(request).await.map_err(|error| {
+                self.abort_server_tx.trigger();
+                error
+            })
+        }
+
         async fn route_up(
             &self,
             request: Request<EventDetails>,
@@ -1341,6 +1380,12 @@ mod tests {
         ) -> std::result::Result<tonic::Response<()>, tonic::Status> {
             Ok(tonic::Response::new(()))
         }
+        async fn up(
+            &self,
+            _request: tonic::Request<event_server::EventDetails>,
+        ) -> std::result::Result<tonic::Response<()>, tonic::Status> {
+            Ok(tonic::Response::new(()))
+        }
         async fn route_up(
             &self,
             _request: tonic::Request<event_server::EventDetails>,
diff --git a/talpid-core/src/tunnel/wireguard/mod.rs b/talpid-core/src/tunnel/wireguard/mod.rs
index 47c86e4e5f..034cabd316 100644
--- a/talpid-core/src/tunnel/wireguard/mod.rs
+++ b/talpid-core/src/tunnel/wireguard/mod.rs
@@ -79,7 +79,7 @@ pub struct WireguardMonitor {
     tunnel: Arc<Mutex<Option<Box<dyn Tunnel>>>>,
     /// Callback to signal tunnel events
     event_callback: Box<
-        dyn (Fn(TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+        dyn (Fn(TunnelEvent) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
             + Send
             + Sync
             + 'static,
@@ -158,7 +158,7 @@ impl Drop for TcpProxy {
 impl WireguardMonitor {
     /// Starts a WireGuard tunnel with the given config
     pub fn start<
-        F: (Fn(TunnelEvent) -> Box<dyn std::future::Future<Output = ()> + Unpin + Send>)
+        F: (Fn(TunnelEvent) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>>)
             + Send
             + Sync
             + Clone
@@ -189,9 +189,6 @@ impl WireguardMonitor {
         #[cfg(windows)]
         let iface_luid = tunnel.get_interface_luid();
 
-        let metadata = Self::tunnel_metadata(&iface_name, &config);
-        runtime.block_on((on_event)(TunnelEvent::InterfaceUp(metadata.clone())));
-
         #[cfg(target_os = "windows")]
         let callback_handle = route_manager
             .add_default_route_callback(Some(WgGoTunnel::default_route_changed_callback), ())
@@ -232,7 +229,11 @@ impl WireguardMonitor {
 
         let route_handle = route_manager.handle().map_err(Error::SetupRoutingError)?;
 
+        let metadata = Self::tunnel_metadata(&iface_name, &config);
+
         std::thread::spawn(move || {
+            runtime.block_on((on_event)(TunnelEvent::InterfaceUp(metadata.clone())));
+
             #[cfg(windows)]
             {
                 let iface_close_sender = close_sender.clone();
diff --git a/talpid-core/src/tunnel_state_machine/connected_state.rs b/talpid-core/src/tunnel_state_machine/connected_state.rs
index 821cee59b7..49365d379e 100644
--- a/talpid-core/src/tunnel_state_machine/connected_state.rs
+++ b/talpid-core/src/tunnel_state_machine/connected_state.rs
@@ -8,7 +8,11 @@ use crate::{
     tunnel::{CloseHandle, TunnelEvent, TunnelMetadata},
 };
 use cfg_if::cfg_if;
-use futures::{channel::mpsc, stream::Fuse, StreamExt};
+use futures::{
+    channel::{mpsc, oneshot},
+    stream::Fuse,
+    StreamExt,
+};
 use std::net::IpAddr;
 use talpid_types::{
     net::TunnelParameters,
@@ -21,7 +25,8 @@ use crate::tunnel::TunnelMonitor;
 
 use super::connecting_state::TunnelCloseEvent;
 
-pub(crate) type TunnelEventsReceiver = Fuse<mpsc::UnboundedReceiver<TunnelEvent>>;
+pub(crate) type TunnelEventsReceiver =
+    Fuse<mpsc::UnboundedReceiver<(TunnelEvent, oneshot::Sender<()>)>>;
 
 
 pub struct ConnectedStateBootstrap {
@@ -265,13 +270,13 @@ impl ConnectedState {
 
     fn handle_tunnel_events(
         self,
-        event: Option<TunnelEvent>,
+        event: Option<(TunnelEvent, oneshot::Sender<()>)>,
         shared_values: &mut SharedTunnelStateValues,
     ) -> EventConsequence {
         use self::EventConsequence::*;
 
         match event {
-            Some(TunnelEvent::Down) | None => {
+            Some((TunnelEvent::Down, _)) | None => {
                 self.disconnect(shared_values, AfterDisconnect::Reconnect(0))
             }
             Some(_) => SameState(self.into()),
@@ -309,28 +314,6 @@ impl TunnelState for ConnectedState {
         let connected_state = ConnectedState::from(bootstrap);
         let tunnel_endpoint = connected_state.tunnel_parameters.get_tunnel_endpoint();
 
-        #[cfg(target_os = "windows")]
-        if let Err(error) = shared_values
-            .split_tunnel
-            .set_tunnel_addresses(Some(&connected_state.metadata))
-        {
-            log::error!(
-                "{}",
-                error.display_chain_with_msg(
-                    "Failed to register addresses with split tunnel driver"
-                )
-            );
-
-            return DisconnectingState::enter(
-                shared_values,
-                (
-                    connected_state.close_handle,
-                    connected_state.tunnel_close_event,
-                    AfterDisconnect::Block(ErrorStateCause::SplitTunnelError),
-                ),
-            );
-        }
-
         if let Err(error) = connected_state.set_firewall_policy(shared_values) {
             DisconnectingState::enter(
                 shared_values,
diff --git a/talpid-core/src/tunnel_state_machine/connecting_state.rs b/talpid-core/src/tunnel_state_machine/connecting_state.rs
index 6eb9891a9b..ebe4cf68b2 100644
--- a/talpid-core/src/tunnel_state_machine/connecting_state.rs
+++ b/talpid-core/src/tunnel_state_machine/connecting_state.rs
@@ -100,9 +100,12 @@ impl ConnectingState {
     ) -> crate::tunnel::Result<Self> {
         let (event_tx, event_rx) = mpsc::unbounded();
         let on_tunnel_event =
-            move |event| -> Box<dyn std::future::Future<Output = ()> + Unpin + Send> {
-                let _ = event_tx.unbounded_send(event);
-                Box::new(futures::future::ready(()))
+            move |event| -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
+                let (tx, rx) = oneshot::channel();
+                let _ = event_tx.unbounded_send((event, tx));
+                Box::pin(async move {
+                    let _ = rx.await;
+                })
             };
 
         let monitor = TunnelMonitor::start(
@@ -322,18 +325,34 @@ impl ConnectingState {
 
     fn handle_tunnel_events(
         mut self,
-        event: Option<tunnel::TunnelEvent>,
+        event: Option<(tunnel::TunnelEvent, oneshot::Sender<()>)>,
         shared_values: &mut SharedTunnelStateValues,
     ) -> EventConsequence {
         use self::EventConsequence::*;
 
         match event {
-            Some(TunnelEvent::AuthFailed(reason)) => self.disconnect(
+            Some((TunnelEvent::AuthFailed(reason), _)) => self.disconnect(
                 shared_values,
                 AfterDisconnect::Block(ErrorStateCause::AuthFailed(reason)),
             ),
-            Some(TunnelEvent::InterfaceUp(tunnel_metadata)) => {
-                self.tunnel_metadata = Some(tunnel_metadata);
+            Some((TunnelEvent::InterfaceUp(metadata), _done_tx)) => {
+                #[cfg(windows)]
+                if let Err(error) = shared_values
+                    .split_tunnel
+                    .set_tunnel_addresses(Some(&metadata))
+                {
+                    log::error!(
+                        "{}",
+                        error.display_chain_with_msg(
+                            "Failed to register addresses with split tunnel driver"
+                        )
+                    );
+                    return self.disconnect(
+                        shared_values,
+                        AfterDisconnect::Block(ErrorStateCause::SplitTunnelError),
+                    );
+                }
+                self.tunnel_metadata = Some(metadata);
                 match Self::set_firewall_policy(
                     shared_values,
                     &self.tunnel_parameters,
@@ -346,11 +365,11 @@ impl ConnectingState {
                     ),
                 }
             }
-            Some(TunnelEvent::Up(metadata)) => NewState(ConnectedState::enter(
+            Some((TunnelEvent::Up(metadata), _)) => NewState(ConnectedState::enter(
                 shared_values,
                 self.into_connected_state_bootstrap(metadata),
             )),
-            Some(TunnelEvent::Down) => SameState(self.into()),
+            Some((TunnelEvent::Down, _)) => SameState(self.into()),
             None => {
                 // The channel was closed
                 debug!("The tunnel disconnected unexpectedly");
diff --git a/talpid-core/src/tunnel_state_machine/mod.rs b/talpid-core/src/tunnel_state_machine/mod.rs
index 5fabfb0daf..97324e3484 100644
--- a/talpid-core/src/tunnel_state_machine/mod.rs
+++ b/talpid-core/src/tunnel_state_machine/mod.rs
@@ -189,7 +189,7 @@ type TunnelCommandReceiver = stream::Fuse<mpsc::UnboundedReceiver<TunnelCommand>
 
 enum EventResult {
     Command(Option<TunnelCommand>),
-    Event(Option<TunnelEvent>),
+    Event(Option<(TunnelEvent, oneshot::Sender<()>)>),
     Close(Result<Option<ErrorStateCause>, oneshot::Canceled>),
 }
 
diff --git a/talpid-openvpn-plugin/proto/openvpn_plugin.proto b/talpid-openvpn-plugin/proto/openvpn_plugin.proto
index 156caa8881..7795803e35 100644
--- a/talpid-openvpn-plugin/proto/openvpn_plugin.proto
+++ b/talpid-openvpn-plugin/proto/openvpn_plugin.proto
@@ -6,6 +6,7 @@ import "google/protobuf/empty.proto";
 
 service OpenvpnEventProxy {
     rpc AuthFailed(EventDetails) returns (google.protobuf.Empty) {}
+    rpc Up(EventDetails) returns (google.protobuf.Empty) {}
     rpc RouteUp(EventDetails) returns (google.protobuf.Empty) {}
     rpc RoutePredown(EventDetails) returns (google.protobuf.Empty) {}
 }
diff --git a/talpid-openvpn-plugin/src/lib.rs b/talpid-openvpn-plugin/src/lib.rs
index 09316d05df..85c7e6a921 100644
--- a/talpid-openvpn-plugin/src/lib.rs
+++ b/talpid-openvpn-plugin/src/lib.rs
@@ -38,6 +38,7 @@ pub enum Error {
 /// events.
 pub static INTERESTING_EVENTS: &'static [EventType] = &[
     EventType::AuthFailed,
+    EventType::Up,
     EventType::RouteUp,
     EventType::RoutePredown,
 ];
diff --git a/talpid-openvpn-plugin/src/processing.rs b/talpid-openvpn-plugin/src/processing.rs
index c266c2d1b5..e6d2a77349 100644
--- a/talpid-openvpn-plugin/src/processing.rs
+++ b/talpid-openvpn-plugin/src/processing.rs
@@ -68,6 +68,7 @@ impl EventProcessor {
             openvpn_plugin::EventType::AuthFailed => {
                 self.runtime.block_on(self.ipc_client.auth_failed(details))
             }
+            openvpn_plugin::EventType::Up => self.runtime.block_on(self.ipc_client.up(details)),
             openvpn_plugin::EventType::RouteUp => {
                 self.runtime.block_on(self.ipc_client.route_up(details))
             }