Merge branch 'app-installation-idempotentency-des-462'

3 files changed, 233 insertions, 91 deletions

diff --git a/test/test-manager/src/network_monitor.rs b/test/test-manager/src/network_monitor.rs
index 02c1e24d9e..7f74f0f425 100644
--- a/test/test-manager/src/network_monitor.rs
+++ b/test/test-manager/src/network_monitor.rs
@@ -4,11 +4,7 @@ use std::{
     time::Duration,
 };
 
-use futures::{
-    channel::oneshot,
-    future::{select, Either},
-    pin_mut, StreamExt,
-};
+use futures::{channel::oneshot, pin_mut, StreamExt};
 pub use pcap::Direction;
 use pcap::PacketCodec;
 use pnet_packet::{
@@ -243,7 +239,7 @@ async fn start_packet_monitor_for_interface(
             no_frame: monitor_options.no_frame,
         })
         .unwrap();
-    let (stop_tx, stop_rx) = oneshot::channel();
+    let (stop_tx, mut stop_rx) = oneshot::channel();
 
     let interface = interface.to_owned();
 
@@ -261,9 +257,7 @@ async fn start_packet_monitor_for_interface(
                 futures::future::pending().await
             }
         };
-
         pin_mut!(timeout);
-        pin_mut!(stop_rx);
 
         let mut is_receiving_tx = Some(is_receiving_tx);
 
@@ -272,40 +266,42 @@ async fn start_packet_monitor_for_interface(
             let next_packet =
                 poll_fn(|ctx| poll_and_notify(ctx, &mut next_packet_fut, &mut is_receiving_tx));
 
-            match select(select(next_packet, &mut stop_rx), &mut timeout).await {
-                Either::Left((Either::Left((Some(Ok(packet)), _)), _)) => {
-                    if let Some(packet) = packet {
-                        if !filter_fn(&packet) {
-                            log::debug!(
-                                "{interface} \"{packet:?}\" does not match closure conditions"
-                            );
-                            monitor_result.discarded_packets =
-                                monitor_result.discarded_packets.saturating_add(1);
-                        } else {
-                            log::debug!("{interface} \"{packet:?}\" matches closure conditions");
+            tokio::select! {
+                _stop = &mut stop_rx => {
+                     log::trace!("stopping packet monitor");
+                     break Ok(monitor_result);
+                }
+                _timeout = &mut timeout => {
+                     log::info!("monitor timed out");
+                     break Ok(monitor_result);
+                }
+                maybe_next_packet = next_packet => {
+                    match maybe_next_packet {
+                        Some(Ok(packet))=> {
+                            if let Some(packet) = packet {
+                                if !filter_fn(&packet) {
+                                    log::debug!("{interface} \"{packet:?}\" does not match closure conditions");
+                                    monitor_result.discarded_packets =
+                                        monitor_result.discarded_packets.saturating_add(1);
+                                } else {
+                                    log::debug!("{interface} \"{packet:?}\" matches closure conditions");
 
-                            let should_continue = should_continue_fn(&packet);
+                                    let should_continue = should_continue_fn(&packet);
 
-                            monitor_result.packets.push(packet);
+                                    monitor_result.packets.push(packet);
 
-                            if !should_continue {
-                                break Ok(monitor_result);
+                                    if !should_continue {
+                                        break Ok(monitor_result);
+                                    }
+                                }
                             }
                         }
+                        _ => {
+                            log::error!("lost packet stream");
+                            break Err(MonitorUnexpectedlyStopped(()));
+                        }
                     }
                 }
-                Either::Left((Either::Left(_), _)) => {
-                    log::error!("lost packet stream");
-                    break Err(MonitorUnexpectedlyStopped(()));
-                }
-                Either::Left((Either::Right(_), _)) => {
-                    log::trace!("stopping packet monitor");
-                    break Ok(monitor_result);
-                }
-                Either::Right(_) => {
-                    log::info!("monitor timed out");
-                    break Ok(monitor_result);
-                }
             }
         }
     });
diff --git a/test/test-manager/src/tests/helpers.rs b/test/test-manager/src/tests/helpers.rs
index 5b1c0a9903..48dba20aac 100644
--- a/test/test-manager/src/tests/helpers.rs
+++ b/test/test-manager/src/tests/helpers.rs
@@ -1,5 +1,7 @@
 use super::{config::TEST_CONFIG, Error, PING_TIMEOUT, WAIT_FOR_TUNNEL_STATE_TIMEOUT};
-use crate::network_monitor::{start_packet_monitor, MonitorOptions};
+use crate::network_monitor::{
+    self, start_packet_monitor, MonitorOptions, MonitorUnexpectedlyStopped, PacketMonitor,
+};
 use futures::StreamExt;
 use mullvad_management_interface::{types, ManagementServiceClient, MullvadProxyClient};
 use mullvad_types::{
@@ -104,7 +106,7 @@ pub async fn send_guest_probes(
     let pktmon = start_packet_monitor(
         move |packet| packet.destination.ip() == destination.ip(),
         MonitorOptions {
-            direction: Some(crate::network_monitor::Direction::In),
+            direction: Some(network_monitor::Direction::In),
             timeout: Some(MONITOR_DURATION),
             ..Default::default()
         },
@@ -330,10 +332,6 @@ impl<T> AbortOnDrop<T> {
     pub fn new(inner: tokio::task::JoinHandle<T>) -> AbortOnDrop<T> {
         AbortOnDrop(Some(inner))
     }
-
-    pub fn into_inner(mut self) -> tokio::task::JoinHandle<T> {
-        self.0.take().unwrap()
-    }
 }
 
 impl<T> Drop for AbortOnDrop<T> {
@@ -523,3 +521,134 @@ pub fn into_constraint(relay: &Relay) -> Constraint<LocationConstraint> {
         .map(Constraint::Only)
         .expect("relay is missing location")
 }
+
+/// Ping monitoring made easy!
+///
+/// Continously ping some destination while monitoring to detect diverging
+/// packets.
+///
+/// To customize [`Pinger`] before the pinging and network monitoring starts,
+/// see [`PingerBuilder`]. Call [`start`](Pinger::start) to start pinging, and
+/// [`stop`](Pinger::stop) to get the leak test results.
+#[allow(dead_code)]
+pub struct Pinger {
+    // These values can be configured with [`PingerBuilder`].
+    destination: SocketAddr,
+    interval: tokio::time::Interval,
+    // Run-time specific values
+    pub guest_ip: IpAddr,
+    ping_task: AbortOnDrop<tokio::task::JoinHandle<()>>,
+    monitor: PacketMonitor,
+}
+
+impl Pinger {
+    /// Create a [`Pinger`] with a default configuration.
+    ///
+    /// See [`PingerBuilder`] for details.
+    pub async fn start(rpc: &test_rpc::ServiceClient) -> Pinger {
+        let defaults = PingerBuilder::default();
+        Self::start_with(defaults, rpc).await
+    }
+
+    /// Create a [`Pinger`] using the configuration of `builder`.
+    ///
+    /// See [`PingerBuilder`] for details on how to configure a [`Pinger`]
+    /// before starting it.
+    pub async fn start_with(builder: PingerBuilder, rpc: &test_rpc::ServiceClient) -> Pinger {
+        // Get the associated IP address of the test runner on the default, non-tunnel interface.
+        let guest_ip = obtain_guest_ip(rpc).await;
+        log::debug!("Guest IP: {guest_ip}");
+
+        // Start a network monitor
+        log::debug!("Monitoring outgoing traffic");
+        let monitor = start_packet_monitor(
+            move |packet| {
+                // NOTE: Many packets will likely be observed for API traffic. Rather than filtering all
+                // of those specifically, simply fail if our probes are observed.
+                packet.source.ip() == guest_ip
+                    && packet.destination.ip() == builder.destination.ip()
+            },
+            MonitorOptions::default(),
+        )
+        .await;
+
+        // Start pinging
+        //
+        // Create some network activity for the network monitor to sniff.
+        let ping_rpc = rpc.clone();
+        let mut interval = tokio::time::interval(builder.interval.period());
+        #[allow(clippy::async_yields_async)]
+        let ping_task = AbortOnDrop::new(tokio::spawn(async move {
+            loop {
+                send_guest_probes_without_monitor(ping_rpc.clone(), None, builder.destination)
+                    .await;
+                interval.tick().await;
+            }
+        }));
+
+        Pinger {
+            destination: builder.destination,
+            interval: builder.interval,
+            guest_ip,
+            ping_task,
+            monitor,
+        }
+    }
+
+    /// Stop pinging and extract the result of the network monitor.
+    pub async fn stop(self) -> Result<network_monitor::MonitorResult, MonitorUnexpectedlyStopped> {
+        // Abort the inner probe sender, which is accomplished by dropping the
+        // join handle to the running task.
+        drop(self.ping_task);
+        self.monitor.into_result().await
+    }
+
+    /// Return the time period determining the cadence of pings that are sent.
+    pub fn period(&self) -> tokio::time::Duration {
+        self.interval.period()
+    }
+}
+
+/// Returns the [`IpAddr`] of the default non-tunnel interface.
+async fn obtain_guest_ip(rpc: &ServiceClient) -> IpAddr {
+    let guest_iface = rpc
+        .get_default_interface()
+        .await
+        .expect("failed to obtain default interface");
+    rpc.get_interface_ip(guest_iface)
+        .await
+        .expect("failed to obtain non-tun IP")
+}
+
+/// Configure a [`Pinger`] before starting it.
+pub struct PingerBuilder {
+    destination: SocketAddr,
+    interval: tokio::time::Interval,
+}
+
+#[allow(dead_code)]
+impl PingerBuilder {
+    /// Create a default [`PingerBuilder`].
+    ///
+    /// This is probably good enough for checking network traffic leaks when the
+    /// test-runner is supposed to be blocked from sending or receiving *any*
+    /// packets outside of localhost.
+    pub fn default() -> PingerBuilder {
+        PingerBuilder {
+            destination: "1.1.1.1:1337".parse().unwrap(),
+            interval: tokio::time::interval(Duration::from_secs(1)),
+        }
+    }
+
+    /// Set the target to ping.
+    pub fn destination(mut self, destination: SocketAddr) -> Self {
+        self.destination = destination;
+        self
+    }
+
+    /// How often a ping should be sent.
+    pub fn interval(mut self, period: Duration) -> Self {
+        self.interval = tokio::time::interval(period);
+        self
+    }
+}
diff --git a/test/test-manager/src/tests/install.rs b/test/test-manager/src/tests/install.rs
index 37a89f5fc8..d2040771df 100644
--- a/test/test-manager/src/tests/install.rs
+++ b/test/test-manager/src/tests/install.rs
@@ -1,14 +1,10 @@
-use super::helpers::{get_package_desc, AbortOnDrop};
+use super::helpers::{connect_and_wait, get_package_desc};
 use super::{Error, TestContext};
 
 use super::config::TEST_CONFIG;
-use crate::network_monitor::{start_packet_monitor, MonitorOptions};
-use mullvad_management_interface::types;
-use std::{
-    collections::HashMap,
-    net::{SocketAddr, ToSocketAddrs},
-    time::Duration,
-};
+use crate::tests::helpers::{wait_for_tunnel_state, Pinger};
+use mullvad_management_interface::{types, ManagementServiceClient};
+use std::{collections::HashMap, net::ToSocketAddrs, time::Duration};
 use test_macro::test_function;
 use test_rpc::meta::Os;
 use test_rpc::{mullvad_daemon::ServiceStatus, ServiceClient};
@@ -47,8 +43,6 @@ pub async fn test_install_previous_app(_: TestContext, rpc: ServiceClient) -> Re
 /// * The VPN service is not running after the upgrade.
 #[test_function(priority = -190)]
 pub async fn test_upgrade_app(ctx: TestContext, rpc: ServiceClient) -> Result<(), Error> {
-    let inet_destination: SocketAddr = "1.1.1.1:1337".parse().unwrap();
-
     // Verify that daemon is running
     if rpc.mullvad_daemon_get_status().await? != ServiceStatus::Running {
         return Err(Error::DaemonNotRunning);
@@ -70,6 +64,7 @@ pub async fn test_upgrade_app(ctx: TestContext, rpc: ServiceClient) -> Result<()
     //
     log::debug!("Entering blocking error state");
 
+    // TODO: Update this to `rpc.exec("mullvad", ["debug", "block-connection"])` when 2023.6 is released.
     rpc.exec("mullvad", ["relay", "set", "location", "xx"])
         .await
         .expect("Failed to set relay location");
@@ -101,41 +96,7 @@ pub async fn test_upgrade_app(ctx: TestContext, rpc: ServiceClient) -> Result<()
     //
     // Begin monitoring outgoing traffic and pinging
     //
-
-    let guest_iface = rpc
-        .get_default_interface()
-        .await
-        .expect("failed to obtain default interface");
-    let guest_ip = rpc
-        .get_interface_ip(guest_iface)
-        .await
-        .expect("failed to obtain non-tun IP");
-    log::debug!("Guest IP: {guest_ip}");
-
-    log::debug!("Monitoring outgoing traffic");
-
-    let monitor = start_packet_monitor(
-        move |packet| {
-            // NOTE: Many packets will likely be observed for API traffic. Rather than filtering all
-            // of those specifically, simply fail if our probes are observed.
-            packet.source.ip() == guest_ip && packet.destination.ip() == inet_destination.ip()
-        },
-        MonitorOptions::default(),
-    )
-    .await;
-
-    let ping_rpc = rpc.clone();
-    let probe_sender = AbortOnDrop::new(tokio::spawn(async move {
-        loop {
-            super::helpers::send_guest_probes_without_monitor(
-                ping_rpc.clone(),
-                None,
-                inet_destination,
-            )
-            .await;
-            tokio::time::sleep(Duration::from_secs(1)).await;
-        }
-    }));
+    let pinger = Pinger::start(&rpc).await;
 
     // install new package
     log::debug!("Installing new app");
@@ -153,17 +114,16 @@ pub async fn test_upgrade_app(ctx: TestContext, rpc: ServiceClient) -> Result<()
     //
     // Check if any traffic was observed
     //
-    let probe_sender = probe_sender.into_inner();
-    probe_sender.abort();
-    let _ = probe_sender.await;
-
-    let monitor_result = monitor.into_result().await.unwrap();
+    let guest_ip = pinger.guest_ip;
+    let monitor_result = pinger.stop().await.unwrap();
     assert_eq!(
         monitor_result.packets.len(),
         0,
         "observed unexpected packets from {guest_ip}"
     );
 
+    // NOTE: Need to create a new `mullvad_client` here after the restart
+    // otherwise we *probably* can't communicate with the daemon.
     let mut mullvad_client = ctx.rpc_provider.new_client().await;
 
     // check if settings were (partially) preserved
@@ -321,6 +281,63 @@ pub async fn test_install_new_app(_: TestContext, rpc: ServiceClient) -> Result<
     Ok(())
 }
 
+/// Install the multiple times starting from a connected state with auto-connect
+/// disabled, failing if the app starts in a disconnected state.
+///
+/// This test is supposed to guard against regressions to this fix included in
+/// the 2021.3-beta1 release:
+/// https://github.com/mullvad/mullvadvpn-app/blob/main/CHANGELOG.md#security-10
+#[test_function(priority = -150)]
+pub async fn test_installation_idempotency(
+    _: TestContext,
+    rpc: ServiceClient,
+    mut mullvad_client: ManagementServiceClient,
+) -> Result<(), Error> {
+    // Connect to any relay
+    connect_and_wait(&mut mullvad_client).await?;
+    // Disable auto-connect
+    mullvad_client
+        .set_auto_connect(false)
+        .await
+        .expect("failed to enable auto-connect");
+    // Check for traffic leaks during the installation processes.
+    //
+    // Start continously pinging while monitoring the network traffic. No
+    // traffic should be observed going outside of the tunnel during either
+    // installation process.
+    let pinger = Pinger::start(&rpc).await;
+    for _ in 1..=2 {
+        // install package
+        log::debug!("Installing new app");
+        rpc.install_app(get_package_desc(&TEST_CONFIG.current_app_filename)?)
+            .await?;
+        // verify that daemon is running
+        wait_for_tunnel_state(mullvad_client.clone(), |state| state.is_connected())
+            .await
+            .map_err(|err| {
+                log::error!(
+                    "App did not start in the expected `Connected` state after the installation process."
+                );
+                err
+            })?;
+        // Wait for an arbitrary amount of time. The point is that the pinger
+        // should be able to ping while the newly installed app is running.
+        if let Some(delay) = pinger.period().checked_mul(3) {
+            tokio::time::sleep(delay).await;
+        }
+    }
+    // Make sure that no network leak occured during any installation process.
+    let guest_ip = pinger.guest_ip;
+    let monitor_result = pinger.stop().await.unwrap();
+    assert_eq!(
+        monitor_result.packets.len(),
+        0,
+        "observed unexpected packets from {guest_ip}"
+    );
+
+    Ok(())
+}
+
 fn get_app_env() -> HashMap<String, String> {
     let mut map = HashMap::new();