Merge branch 'test-cleanup-probes'

7 files changed, 239 insertions, 105 deletions

diff --git a/.github/workflows/desktop-e2e.yml b/.github/workflows/desktop-e2e.yml
index b28e70232d..86cf45345e 100644
--- a/.github/workflows/desktop-e2e.yml
+++ b/.github/workflows/desktop-e2e.yml
@@ -15,6 +15,8 @@ jobs:
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
+        with:
+          fetch-tags: true
       - name: Run end-to-end tests
         shell: bash -ieo pipefail {0}
         run: |
@@ -30,6 +32,8 @@ jobs:
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
+        with:
+          fetch-tags: true
       - name: Run end-to-end tests
         shell: bash -ieo pipefail {0}
         run: |
@@ -45,6 +49,8 @@ jobs:
     steps:
       - name: Checkout repository
         uses: actions/checkout@v4
+        with:
+          fetch-tags: true
       - name: Run end-to-end tests
         shell: bash -ieo pipefail {0}
         run: |
diff --git a/test/ci-runtests.sh b/test/ci-runtests.sh
index 95e8340309..c794b75381 100755
--- a/test/ci-runtests.sh
+++ b/test/ci-runtests.sh
@@ -211,7 +211,7 @@ echo "* Building test runner"
 echo "**********************************"
 
 # Clean up packages. Try to keep ones that match the versions we're testing
-find "$PACKAGES_DIR/" -type f ! \( -name "*${OLD_APP_VERSION}_*" -o -name "*${OLD_APP_VERSION}.*" -o -name "*${NEW_APP_VERSION}*" \) -delete || true
+find "$PACKAGES_DIR/" -type f ! \( -name "*${OLD_APP_VERSION}*" -o -name "*${commit}*" \) -delete || true
 
 function build_test_runner {
     local target=""
diff --git a/test/test-manager/src/tests/dns.rs b/test/test-manager/src/tests/dns.rs
index e87da24db0..69de367b88 100644
--- a/test/test-manager/src/tests/dns.rs
+++ b/test/test-manager/src/tests/dns.rs
@@ -212,44 +212,50 @@ async fn leak_test_dns(
     // We should observe 2 outgoing packets to the whitelisted destination
     // on port 53, and only inside the desired interface.
 
-    spoof_packets(
-        rpc,
-        Some(Interface::Tunnel),
-        tun_bind_addr,
-        whitelisted_dest,
-    );
-    spoof_packets(
-        rpc,
-        Some(Interface::NonTunnel),
-        guest_bind_addr,
-        whitelisted_dest,
-    );
-
-    spoof_packets(
-        rpc,
-        Some(Interface::Tunnel),
-        tun_bind_addr,
-        blocked_dest_local,
-    );
-    spoof_packets(
-        rpc,
-        Some(Interface::NonTunnel),
-        guest_bind_addr,
-        blocked_dest_local,
-    );
-
-    spoof_packets(
-        rpc,
-        Some(Interface::Tunnel),
-        tun_bind_addr,
-        blocked_dest_public,
-    );
-    spoof_packets(
-        rpc,
-        Some(Interface::NonTunnel),
-        guest_bind_addr,
-        blocked_dest_public,
-    );
+    let rpc = rpc.clone();
+    let probes = tokio::spawn(async move {
+        tokio::join!(
+            // send to allowed dest
+            spoof_packets(
+                &rpc,
+                Some(Interface::Tunnel),
+                tun_bind_addr,
+                whitelisted_dest,
+            ),
+            spoof_packets(
+                &rpc,
+                Some(Interface::NonTunnel),
+                guest_bind_addr,
+                whitelisted_dest,
+            ),
+            // send to blocked local dest
+            spoof_packets(
+                &rpc,
+                Some(Interface::Tunnel),
+                tun_bind_addr,
+                blocked_dest_local,
+            ),
+            spoof_packets(
+                &rpc,
+                Some(Interface::NonTunnel),
+                guest_bind_addr,
+                blocked_dest_local,
+            ),
+            // send to blocked public dest
+            spoof_packets(
+                &rpc,
+                Some(Interface::Tunnel),
+                tun_bind_addr,
+                blocked_dest_public,
+            ),
+            spoof_packets(
+                &rpc,
+                Some(Interface::NonTunnel),
+                guest_bind_addr,
+                blocked_dest_public,
+            ),
+        )
+    });
 
     if use_tun {
         //
@@ -257,6 +263,10 @@ async fn leak_test_dns(
         //
 
         let tunnel_result = tunnel_monitor.wait().await.unwrap();
+
+        probes.abort();
+        let _ = probes.await;
+
         assert!(
             tunnel_result.packets.len() >= 2,
             "expected at least 2 in-tunnel packets to allowed destination only"
@@ -282,6 +292,9 @@ async fn leak_test_dns(
     } else {
         let non_tunnel_result = non_tunnel_monitor.wait().await.unwrap();
 
+        probes.abort();
+        let _ = probes.await;
+
         //
         // Examine tunnel traffic
         //
@@ -605,6 +618,7 @@ async fn run_dns_config_test<
     );
 
     handle.abort();
+    let _ = handle.await;
 
     Ok(())
 }
@@ -647,22 +661,23 @@ async fn connect_local_wg_relay(mullvad_client: &mut ManagementServiceClient) ->
     Ok(())
 }
 
-fn spoof_packets(
+async fn spoof_packets(
     rpc: &ServiceClient,
     interface: Option<Interface>,
     bind_addr: SocketAddr,
     dest: SocketAddr,
 ) {
-    let rpc1 = rpc.clone();
-    let rpc2 = rpc.clone();
-    tokio::spawn(async move {
+    let tcp_rpc = rpc.clone();
+    let tcp_send = async move {
         log::debug!("sending to {}/tcp from {}", dest, bind_addr);
-        let _ = rpc1.send_tcp(interface, bind_addr, dest).await;
-    });
-    tokio::spawn(async move {
+        let _ = tcp_rpc.send_tcp(interface, bind_addr, dest).await;
+    };
+    let udp_rpc = rpc.clone();
+    let udp_send = async move {
         log::debug!("sending to {}/udp from {}", dest, bind_addr);
-        let _ = rpc2.send_udp(interface, bind_addr, dest).await;
-    });
+        let _ = udp_rpc.send_udp(interface, bind_addr, dest).await;
+    };
+    let _ = tokio::join!(tcp_send, udp_send);
 }
 
 type ShouldContinue = bool;
diff --git a/test/test-manager/src/tests/helpers.rs b/test/test-manager/src/tests/helpers.rs
index daef783247..f73e6f6e79 100644
--- a/test/test-manager/src/tests/helpers.rs
+++ b/test/test-manager/src/tests/helpers.rs
@@ -79,33 +79,16 @@ pub async fn send_guest_probes(
     )
     .await;
 
-    let bind_addr = if let Some(interface) = interface {
-        SocketAddr::new(
-            rpc.get_interface_ip(interface)
-                .await
-                .expect("failed to obtain interface IP"),
-            0,
-        )
-    } else {
-        "0.0.0.0:0".parse().unwrap()
-    };
-
-    let send_handle = tokio::spawn(async move {
-        let tcp_rpc = rpc.clone();
-        let udp_rpc = rpc.clone();
-        tokio::spawn(async move {
-            let _ = tcp_rpc.send_tcp(interface, bind_addr, destination).await;
-        });
-        tokio::spawn(async move {
-            let _ = udp_rpc.send_udp(interface, bind_addr, destination).await;
-        });
-        ping_with_timeout(&rpc, destination.ip(), interface).await?;
-        Ok::<(), Error>(())
-    });
+    let send_handle = tokio::spawn(send_guest_probes_without_monitor(
+        rpc,
+        interface,
+        destination,
+    ));
 
     let monitor_result = pktmon.wait().await.unwrap();
 
     send_handle.abort();
+    let _ = send_handle.await;
 
     let mut result = ProbeResult::default();
 
@@ -127,6 +110,31 @@ pub async fn send_guest_probes(
     Ok(result)
 }
 
+/// Send one probe per transport protocol to `destination` without running a packet monitor
+pub async fn send_guest_probes_without_monitor(
+    rpc: ServiceClient,
+    interface: Option<Interface>,
+    destination: SocketAddr,
+) {
+    let bind_addr = if let Some(interface) = interface {
+        SocketAddr::new(
+            rpc.get_interface_ip(interface)
+                .await
+                .expect("failed to obtain interface IP"),
+            0,
+        )
+    } else {
+        "0.0.0.0:0".parse().unwrap()
+    };
+
+    let tcp_rpc = rpc.clone();
+    let tcp_send = async move { tcp_rpc.send_tcp(interface, bind_addr, destination).await };
+    let udp_rpc = rpc.clone();
+    let udp_send = async move { udp_rpc.send_udp(interface, bind_addr, destination).await };
+    let icmp = async move { ping_with_timeout(&rpc, destination.ip(), interface).await };
+    let _ = tokio::join!(tcp_send, udp_send, icmp);
+}
+
 pub async fn ping_with_timeout(
     rpc: &ServiceClient,
     dest: IpAddr,
@@ -274,11 +282,23 @@ pub async fn geoip_lookup_with_retries(rpc: &ServiceClient) -> Result<AmIMullvad
     }
 }
 
-pub struct AbortOnDrop<T>(pub tokio::task::JoinHandle<T>);
+pub struct AbortOnDrop<T>(Option<tokio::task::JoinHandle<T>>);
+
+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> {
     fn drop(&mut self) {
-        self.0.abort();
+        if let Some(task) = self.0.take() {
+            task.abort();
+        }
     }
 }
 
diff --git a/test/test-manager/src/tests/install.rs b/test/test-manager/src/tests/install.rs
index abd105d4dd..04d401a22d 100644
--- a/test/test-manager/src/tests/install.rs
+++ b/test/test-manager/src/tests/install.rs
@@ -1,4 +1,4 @@
-use super::helpers::{get_package_desc, ping_with_timeout, AbortOnDrop};
+use super::helpers::{get_package_desc, AbortOnDrop};
 use super::{Error, TestContext};
 
 use super::config::TEST_CONFIG;
@@ -48,7 +48,6 @@ pub async fn test_install_previous_app(_: TestContext, rpc: ServiceClient) -> Re
 #[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();
-    let bind_addr: SocketAddr = "0.0.0.0:0".parse().unwrap();
 
     // Verify that daemon is running
     if rpc.mullvad_daemon_get_status().await? != ServiceStatus::Running {
@@ -122,11 +121,14 @@ pub async fn test_upgrade_app(ctx: TestContext, rpc: ServiceClient) -> Result<()
     .await;
 
     let ping_rpc = rpc.clone();
-    let abort_on_drop = AbortOnDrop(tokio::spawn(async move {
+    let probe_sender = AbortOnDrop::new(tokio::spawn(async move {
         loop {
-            let _ = ping_rpc.send_tcp(None, bind_addr, inet_destination).await;
-            let _ = ping_rpc.send_udp(None, bind_addr, inet_destination).await;
-            let _ = ping_with_timeout(&ping_rpc, inet_destination.ip(), None).await;
+            super::helpers::send_guest_probes_without_monitor(
+                ping_rpc.clone(),
+                None,
+                inet_destination,
+            )
+            .await;
             tokio::time::sleep(Duration::from_secs(1)).await;
         }
     }));
@@ -147,7 +149,10 @@ pub async fn test_upgrade_app(ctx: TestContext, rpc: ServiceClient) -> Result<()
     //
     // Check if any traffic was observed
     //
-    drop(abort_on_drop);
+    let probe_sender = probe_sender.into_inner();
+    probe_sender.abort();
+    let _ = probe_sender.await;
+
     let monitor_result = monitor.into_result().await.unwrap();
     assert_eq!(
         monitor_result.packets.len(),
diff --git a/test/test-manager/src/vm/network/macos.rs b/test/test-manager/src/vm/network/macos.rs
index 5e4bdae786..ef815f203a 100644
--- a/test/test-manager/src/vm/network/macos.rs
+++ b/test/test-manager/src/vm/network/macos.rs
@@ -1,11 +1,11 @@
 use std::net::{Ipv4Addr, SocketAddrV4};
 
 use anyhow::{anyhow, Context, Result};
+use futures::future::{self, Either};
 use tokio::{io::AsyncWriteExt, process::Command};
 
 /// Pingable dummy LAN interface (IP)
-/// TODO: This should probably be a different host, not the gateway
-pub const DUMMY_LAN_INTERFACE_IP: Ipv4Addr = Ipv4Addr::new(192, 168, 64, 1);
+pub const DUMMY_LAN_INTERFACE_IP: Ipv4Addr = Ipv4Addr::new(192, 168, 64, 254);
 
 // Private key of the wireguard remote peer on host.
 const CUSTOM_TUN_REMOTE_PRIVKEY: &str = "gLvQuyqazziyf+pUCAFUgTnWIwn6fPE5MOReOqPEGHU=";
@@ -36,6 +36,11 @@ pub const NON_TUN_GATEWAY: Ipv4Addr = Ipv4Addr::new(192, 168, 64, 1);
 /// Name of the wireguard interface on the host
 pub const CUSTOM_TUN_INTERFACE_NAME: &str = "utun123";
 
+use std::time::Duration;
+
+/// Timeout for wireguard-go to create an interface
+const INTERFACE_SETUP_TIMEOUT: Duration = Duration::from_secs(5);
+
 /// Set up WireGuard relay and dummy hosts.
 pub async fn setup_test_network() -> Result<()> {
     log::debug!("Setting up test network");
@@ -45,9 +50,41 @@ pub async fn setup_test_network() -> Result<()> {
         .await
         .context("Failed to create WireGuard interface")?;
 
+    // A bit of trickery to detect when the bridge is available.
+    tokio::spawn(async move {
+        for _ in 0..30 {
+            let Ok(interface) = find_vm_bridge() else {
+                tokio::time::sleep(Duration::from_secs(1)).await;
+                continue;
+            };
+            match create_dummy_interface(interface).await {
+                Ok(_) => log::debug!("Created dummy interface"),
+                Err(error) => log::error!("Failed to create dummy interface: {error}"),
+            }
+            return;
+        }
+        log::error!("Failed to create dummy interface: timed out");
+    });
+
     Ok(())
 }
 
+async fn create_dummy_interface(interface: String) -> Result<()> {
+    let mut cmd = Command::new("/usr/bin/sudo");
+    cmd.args([
+        "/sbin/ifconfig",
+        &interface,
+        "alias",
+        &DUMMY_LAN_INTERFACE_IP.to_string(),
+    ]);
+    let output = cmd.output().await.context("Create dummy interface")?;
+    if output.status.success() {
+        Ok(())
+    } else {
+        Err(anyhow!("ifconfig failed: {:?}", output.status.code()))
+    }
+}
+
 /// A hack to find the Tart bridge interface using `NON_TUN_GATEWAY`.
 /// It should be possible to retrieve this using the virtualization framework instead,
 /// but that requires an entitlement.
@@ -84,28 +121,52 @@ async fn enable_forwarding() -> Result<()> {
 async fn create_wireguard_interface() -> Result<()> {
     log::debug!("Creating custom WireGuard tunnel");
 
-    // Check if the tunnel already exists
-    let mut cmd = Command::new("/sbin/ifconfig");
-    cmd.arg(CUSTOM_TUN_INTERFACE_NAME);
-    let output = cmd
-        .output()
-        .await
-        .context("Check if wireguard tunnel exists")?;
-    if output.status.success() {
-        log::debug!("Tunnel {CUSTOM_TUN_INTERFACE_NAME} already exists");
-    } else {
+    let mut go_proc = tokio::spawn(async move {
         let mut cmd = Command::new("/usr/bin/sudo");
-        cmd.args(["wireguard-go", CUSTOM_TUN_INTERFACE_NAME]);
+        cmd.kill_on_drop(true);
+        cmd.args(["wireguard-go", "-f", CUSTOM_TUN_INTERFACE_NAME]);
         let output = cmd.output().await.context("Run wireguard-go")?;
-        if !output.status.success() {
-            return Err(anyhow!(
-                "wireguard-go failed: {}",
-                output.status.code().unwrap()
-            ));
+        if output.status.success() {
+            Ok(())
+        } else {
+            Err(anyhow!(
+                "wireguard-go failed with status {:?}",
+                output.status.code()
+            ))
         }
-    }
+    });
 
-    Ok(())
+    let mut tunnel_check: tokio::task::JoinHandle<Result<()>> = tokio::spawn(async move {
+        loop {
+            // Check if the tunnel already exists
+            let mut cmd = Command::new("/sbin/ifconfig");
+            cmd.arg(CUSTOM_TUN_INTERFACE_NAME);
+            let output = cmd
+                .output()
+                .await
+                .context("Check if wireguard tunnel exists")?;
+            if output.status.success() {
+                log::debug!("Created custom WireGuard tunnel interface");
+                return Ok(());
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    });
+
+    let result = tokio::time::timeout(
+        INTERFACE_SETUP_TIMEOUT,
+        future::select(&mut go_proc, &mut tunnel_check),
+    )
+    .await
+    .context("WireGuard interface setup timed out")?;
+
+    let result = match result {
+        Either::Left((result, _)) | Either::Right((result, _)) => result,
+    };
+
+    tunnel_check.abort();
+
+    result?
 }
 
 pub async fn configure_tunnel() -> Result<()> {
diff --git a/test/test-runner/src/logging.rs b/test/test-runner/src/logging.rs
index 3c9aad9d15..224c329775 100644
--- a/test/test-runner/src/logging.rs
+++ b/test/test-runner/src/logging.rs
@@ -1,10 +1,12 @@
 use log::{Level, LevelFilter, Metadata, Record, SetLoggerError};
 use once_cell::sync::Lazy;
+use std::ffi::OsStr;
 use std::path::{Path, PathBuf};
 use test_rpc::logging::Error;
 use test_rpc::logging::{LogFile, LogOutput, Output};
 use tokio::{
-    fs::read_to_string,
+    fs::File,
+    io::{self, AsyncBufReadExt, BufReader},
     sync::{
         broadcast::{channel, Receiver, Sender},
         Mutex,
@@ -12,6 +14,12 @@ use tokio::{
 };
 
 const MAX_OUTPUT_BUFFER: usize = 10_000;
+/// Only consider files that end with ".log"
+const INCLUDE_LOG_FILE_EXT: &str = "log";
+/// Ignore log files that contain ".old"
+const EXCLUDE_LOG_FILE_CONTAIN: &str = ".old";
+/// Maximum number of lines that each log file may contain
+const TRUNCATE_LOG_FILE_LINES: usize = 100;
 
 pub static LOGGER: Lazy<StdOutBuffer> = Lazy::new(|| {
     let (sender, listener) = channel(MAX_OUTPUT_BUFFER);
@@ -69,7 +77,7 @@ async fn read_settings_file() -> Result<String, Error> {
     let mut settings_path = mullvad_paths::get_default_settings_dir()
         .map_err(|error| Error::Logs(format!("{}", error)))?;
     settings_path.push("settings.json");
-    read_to_string(&settings_path)
+    read_truncated(&settings_path)
         .await
         .map_err(|error| Error::Logs(format!("{}: {}", settings_path.display(), error)))
 }
@@ -82,7 +90,7 @@ async fn read_log_files() -> Result<Vec<Result<LogFile, Error>>, Error> {
         .map_err(|error| Error::Logs(format!("{}", error)))?;
     let mut log_files = Vec::new();
     for path in paths {
-        let log_file = read_to_string(&path)
+        let log_file = read_truncated(&path)
             .await
             .map_err(|error| Error::Logs(format!("{}: {}", path.display(), error)))
             .map(|content| LogFile {
@@ -98,14 +106,33 @@ async fn list_logs<T: AsRef<Path>>(log_dir: T) -> Result<Vec<PathBuf>, Error> {
     let mut dir_entries = tokio::fs::read_dir(&log_dir)
         .await
         .map_err(|e| Error::Logs(format!("{}: {}", log_dir.as_ref().display(), e)))?;
-    let log_extension = Some(std::ffi::OsStr::new("log"));
 
     let mut paths = Vec::new();
     while let Ok(Some(entry)) = dir_entries.next_entry().await {
         let path = entry.path();
-        if path.extension() == log_extension {
+        if let Some(u8_path) = path.to_str() {
+            if u8_path.contains(EXCLUDE_LOG_FILE_CONTAIN) {
+                continue;
+            }
+        }
+        if path.extension() == Some(OsStr::new(INCLUDE_LOG_FILE_EXT)) {
             paths.push(path);
         }
     }
     Ok(paths)
 }
+
+async fn read_truncated<T: AsRef<Path>>(path: T) -> io::Result<String> {
+    let mut output = vec![];
+    let reader = BufReader::new(File::open(path).await?);
+    let mut lines = reader.lines();
+    while let Some(line) = lines.next_line().await? {
+        output.push(line);
+    }
+    if output.len() > TRUNCATE_LOG_FILE_LINES {
+        let drop_count = output.len() - TRUNCATE_LOG_FILE_LINES;
+        // not the most efficient
+        output.drain(0..drop_count);
+    }
+    Ok(output.join("\n"))
+}