Merge branch 'main' into extra-rw-container

Author: Porges

src/agent/Cargo.lock

@@ -8,7 +8,7 @@ license = "MIT"
 anyhow = { version = "1.0", features = ["backtrace"] }
 cobertura = { path = "../cobertura" }
 debuggable-module = { path = "../debuggable-module" }
-iced-x86 = "1.18"
+iced-x86 = "1.19"
 log = "0.4.17"
 regex = "1.8"
 symbolic = { version = "10.1", features = [
@@ -22,6 +22,7 @@ thiserror = "1.0"
 debugger = { path = "../debugger" }
 
 [target.'cfg(target_os = "linux")'.dependencies]
+nix = "0.26"
 pete = "0.10"
 # For procfs, opt out of the `chrono` freature; it pulls in an old version
 # of `time`. We do not use the methods that the `chrono` feature enables.

src/agent/coverage/Cargo.toml

@@ -58,19 +58,58 @@ impl CoverageRecorder {
 
     #[cfg(target_os = "linux")]
     pub fn record(self) -> Result<Recorded> {
+        use std::sync::Mutex;
+
+        use anyhow::bail;
+
+        use crate::timer;
         use linux::debugger::Debugger;
         use linux::LinuxRecorder;
 
         let loader = self.loader.clone();
 
-        crate::timer::timed(self.timeout, move || {
-            let mut recorder = LinuxRecorder::new(&loader, self.allowlist);
-            let dbg = Debugger::new(&mut recorder);
-            let output = dbg.run(self.cmd)?;
-            let coverage = recorder.coverage;
+        let child_pid: Arc<Mutex<Option<u32>>> = Arc::new(Mutex::new(None));
 
-            Ok(Recorded { coverage, output })
-        })?
+        let recorded = {
+            let child_pid = child_pid.clone();
+
+            timer::timed(self.timeout, move || {
+                let mut recorder = LinuxRecorder::new(&loader, self.allowlist);
+                let mut dbg = Debugger::new(&mut recorder);
+                let child = dbg.spawn(self.cmd)?;
+
+                // Save child PID so we can send SIGKILL on timeout.
+                if let Ok(mut pid) = child_pid.lock() {
+                    *pid = Some(child.id());
+                } else {
+                    bail!("couldn't lock mutex to save child PID ");
+                }
+
+                let output = dbg.wait(child)?;
+                let coverage = recorder.coverage;
+
+                Ok(Recorded { coverage, output })
+            })
+        };
+
+        if let Err(timer::TimerError::Timeout(..)) = &recorded {
+            let Ok(pid) = child_pid.lock() else {
+                bail!("couldn't lock mutex to kill child PID");
+            };
+
+            if let Some(pid) = *pid {
+                use nix::sys::signal::{kill, SIGKILL};
+
+                let pid = pete::Pid::from_raw(pid as i32);
+
+                // Try to clean up, ignore errors due to earlier exits.
+                let _ = kill(pid, SIGKILL);
+            } else {
+                warn!("timeout before PID set for child process");
+            }
+        }
+
+        recorded?
     }
 
     #[cfg(target_os = "windows")]

src/agent/coverage/src/record.rs

@@ -3,7 +3,7 @@
 
 use std::collections::BTreeMap;
 use std::io::Read;
-use std::process::Command;
+use std::process::{Child, Command};
 
 use anyhow::{bail, format_err, Result};
 use debuggable_module::path::FilePath;
@@ -39,9 +39,11 @@ impl<'eh> Debugger<'eh> {
         }
     }
 
-    pub fn run(mut self, cmd: Command) -> Result<Output> {
-        let mut child = self.context.tracer.spawn(cmd)?;
+    pub fn spawn(&mut self, cmd: Command) -> Result<Child> {
+        Ok(self.context.tracer.spawn(cmd)?)
+    }
 
+    pub fn wait(self, mut child: Child) -> Result<Output> {
         if let Err(err) = self.wait_on_stops() {
             // Ignore error if child already exited.
             let _ = child.kill();
@@ -52,22 +54,29 @@ impl<'eh> Debugger<'eh> {
         // Currently unavailable on Linux.
         let status = None;
 
-        let stdout = if let Some(mut pipe) = child.stdout {
+        let stdout = if let Some(pipe) = &mut child.stdout {
             let mut stdout = Vec::new();
             pipe.read_to_end(&mut stdout)?;
             String::from_utf8_lossy(&stdout).into_owned()
         } else {
             "".into()
         };
 
-        let stderr = if let Some(mut pipe) = child.stderr {
+        let stderr = if let Some(pipe) = &mut child.stderr {
             let mut stderr = Vec::new();
             pipe.read_to_end(&mut stderr)?;
             String::from_utf8_lossy(&stderr).into_owned()
         } else {
             "".into()
         };
 
+        // Clean up, ignoring output that we've already gathered.
+        //
+        // These calls should also be unnecessary no-ops, but we really want to avoid any dangling
+        // or zombie child processes.
+        let _ = child.kill();
+        let _ = child.wait();
+
         let output = Output {
             status,
             stderr,

src/agent/coverage/src/record/linux/debugger.rs

@@ -5,9 +5,9 @@ use std::sync::mpsc;
 use std::thread;
 use std::time::Duration;
 
-use anyhow::{bail, Result};
+use thiserror::Error;
 
-pub fn timed<F, T>(timeout: Duration, function: F) -> Result<T>
+pub fn timed<F, T>(timeout: Duration, function: F) -> Result<T, TimerError>
 where
     T: Send + 'static,
     F: FnOnce() -> T + Send + 'static,
@@ -25,13 +25,23 @@ where
         let _ = timer_sender.send(Timed::Timeout);
     });
 
-    match receiver.recv()? {
-        Timed::Done(out) => Ok(out),
-        Timed::Timeout => bail!("function exceeded timeout of {:?}", timeout),
+    match receiver.recv() {
+        Ok(Timed::Done(out)) => Ok(out),
+        Ok(Timed::Timeout) => Err(TimerError::Timeout(timeout)),
+        Err(recv) => Err(TimerError::Recv(recv)),
     }
 }
 
 enum Timed<T> {
     Done(T),
     Timeout,
 }
+
+#[derive(Debug, Error)]
+pub enum TimerError {
+    #[error("timer threads exited without sending messages")]
+    Recv(mpsc::RecvError),
+
+    #[error("function exceeded timeout of {0:?}")]
+    Timeout(Duration),
+}

src/agent/coverage/src/timer.rs

@@ -9,7 +9,7 @@ anyhow = "1.0"
 elsa = "1.8.1"
 gimli = "0.27.2"
 goblin = "0.6.0"
-iced-x86 = "1.18"
+iced-x86 = "1.19"
 log = "0.4.17"
 pdb = "0.8.0"
 regex = "1.8"

src/agent/debuggable-module/Cargo.toml

@@ -9,7 +9,7 @@ license = "MIT"
 anyhow = "1.0"
 fnv = "1.0"
 goblin = "0.5"
-iced-x86 = "1.18"
+iced-x86 = "1.19"
 log = "0.4"
 memmap2 = "0.5"
 rand = "0.8"

src/agent/debugger/Cargo.toml

@@ -43,7 +43,7 @@ strum = "0.24"
 strum_macros = "0.24"
 stacktrace-parser = { path = "../stacktrace-parser" }
 storage-queue = { path = "../storage-queue" }
-tempfile = "3.5.0"
+tempfile = "3.6.0"
 thiserror = "1.0"
 tokio = { version = "1.28", features = ["full"] }
 tokio-util = { version = "0.7", features = ["full"] }

src/agent/onefuzz-task/Cargo.toml

@@ -5,6 +5,7 @@ use chrono::DateTime;
 use serde::{Deserialize, Serialize};
 use std::fmt;
 use std::sync::{LockResult, RwLockReadGuard, RwLockWriteGuard};
+use std::time::Duration;
 use uuid::Uuid;
 
 pub use chrono::Utc;
@@ -15,6 +16,8 @@ use tokio::sync::broadcast::{self, Receiver};
 #[macro_use]
 extern crate lazy_static;
 
+const DEAFAULT_CHANNEL_CLOSING_TIMEOUT: Duration = Duration::from_secs(30);
+
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(transparent)]
 pub struct MicrosoftTelemetryKey(Uuid);
@@ -355,18 +358,22 @@ pub async fn set_appinsights_clients(
     global::set_clients(instance_client, microsoft_client);
 }
 
+pub async fn try_flush_and_close() {
+    _try_flush_and_close(DEAFAULT_CHANNEL_CLOSING_TIMEOUT).await
+}
+
 /// Try to submit any pending telemetry with a blocking call.
 ///
 /// Meant for a final attempt at flushing pending items before an abnormal exit.
 /// After calling this function, any existing telemetry client will be dropped,
 /// and subsequent telemetry submission will be a silent no-op.
-pub async fn try_flush_and_close() {
+pub async fn _try_flush_and_close(timeout: Duration) {
     let clients = global::take_clients();
-
     for client in clients {
-        client.close_channel().await;
+        if let Err(e) = tokio::time::timeout(timeout, client.close_channel()).await {
+            log::warn!("Failed to close telemetry client: {}", e);
+        }
     }
-
     // dropping the broadcast sender to make sure all pending events are sent
     let _global_event_source = global::EVENT_SOURCE.write().unwrap().take();
 }
@@ -468,11 +475,15 @@ pub fn try_broadcast_trace(timestamp: DateTime<Utc>, msg: String, level: log::Le
 }
 
 pub fn subscribe_to_events() -> Result<Receiver<LoggingEvent>> {
-    let global_event_source = global::EVENT_SOURCE.read().unwrap();
-    if let Some(evs) = global_event_source.clone() {
-        Ok(evs.subscribe())
-    } else {
-        bail!("Event source not initialized");
+    match global::EVENT_SOURCE.read() {
+        Ok(global_event_source) => {
+            if let Some(evs) = global_event_source.clone() {
+                Ok(evs.subscribe())
+            } else {
+                bail!("Event source not initialized");
+            }
+        }
+        Err(e) => bail!("failed to acquire event source lock: {}", e),
     }
 }
 

src/agent/onefuzz-telemetry/src/lib.rs

@@ -40,7 +40,7 @@ url-escape = "0.1.0"
 storage-queue = { path = "../storage-queue" }
 strum = "0.24"
 strum_macros = "0.24"
-tempfile = "3.5.0"
+tempfile = "3.6.0"
 process_control = "4.0"
 reqwest-retry = { path = "../reqwest-retry" }
 onefuzz-telemetry = { path = "../onefuzz-telemetry" }