add support for fully self-contained fuzzers (#454)

src/agent/onefuzz-agent/src/local/generic_crash_report.rs

@@ -29,7 +29,7 @@ pub fn build_report_config(args: &clap::ArgMatches<'_>) -> Result<Config> {
     } else {
         None
     };
-    let unique_reports = value_t!(args, UNIQUE_REPORTS_DIR, PathBuf)?.into();
+    let unique_reports = Some(value_t!(args, UNIQUE_REPORTS_DIR, PathBuf)?.into());
 
     let target_timeout = value_t!(args, TARGET_TIMEOUT, u64).ok();
 

src/agent/onefuzz-agent/src/local/libfuzzer_crash_report.rs

@@ -29,7 +29,7 @@ pub fn build_report_config(args: &clap::ArgMatches<'_>) -> Result<Config> {
     } else {
         None
     };
-    let unique_reports = value_t!(args, UNIQUE_REPORTS_DIR, PathBuf)?.into();
+    let unique_reports = Some(value_t!(args, UNIQUE_REPORTS_DIR, PathBuf)?.into());
 
     let target_timeout = value_t!(args, TARGET_TIMEOUT, u64).ok();
 

src/agent/onefuzz-agent/src/tasks/fuzz/supervisor.rs

@@ -5,6 +5,7 @@
 use crate::tasks::{
     config::{CommonConfig, ContainerType},
     heartbeat::*,
+    report::crash_report::monitor_reports,
     stats::common::{monitor_stats, StatsFormat},
     utils::CheckNotify,
 };
@@ -24,12 +25,13 @@ use std::{
     process::Stdio,
     time::Duration,
 };
+use tempfile::tempdir;
 use tokio::{
     process::{Child, Command},
     sync::Notify,
 };
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Default)]
 pub struct SupervisorConfig {
     pub inputs: SyncedDir,
     pub crashes: SyncedDir,
@@ -37,13 +39,16 @@ pub struct SupervisorConfig {
     pub supervisor_env: HashMap<String, String>,
     pub supervisor_options: Vec<String>,
     pub supervisor_input_marker: Option<String>,
-    pub target_exe: PathBuf,
+    pub target_exe: Option<PathBuf>,
-    pub target_options: Vec<String>,
+    pub target_options: Option<Vec<String>>,
-    pub tools: SyncedDir,
+    pub tools: Option<SyncedDir>,
     pub wait_for_files: Option<ContainerType>,
     pub stats_file: Option<String>,
     pub stats_format: Option<StatsFormat>,
     pub ensemble_sync_delay: Option<u64>,
+    pub reports: Option<SyncedDir>,
+    pub unique_reports: Option<SyncedDir>,
+    pub no_repro: Option<SyncedDir>,
     #[serde(flatten)]
     pub common: CommonConfig,
 }
@@ -54,27 +59,43 @@ pub async fn spawn(config: SupervisorConfig) -> Result<(), Error> {
     let runtime_dir = OwnedDir::new(config.common.task_id.to_string());
     runtime_dir.create_if_missing().await?;
 
-    config.tools.init_pull().await?;
+    // setup tools
-    set_executable(&config.tools.path).await?;
+    if let Some(tools) = &config.tools {
-
+        tools.init_pull().await?;
-    let supervisor_path = Expand::new()
+        set_executable(&tools.path).await?;
-        .tools_dir(&config.tools.path)
+    }
-        .evaluate_value(&config.supervisor_exe)?;
 
+    // setup crashes
     let crashes = SyncedDir {
         path: runtime_dir.path().join("crashes"),
         url: config.crashes.url.clone(),
     };
-
     crashes.init().await?;
     let monitor_crashes = crashes.monitor_results(new_result);
 
+    // setup reports
+    let reports_dir = tempdir()?;
+    if let Some(unique_reports) = &config.unique_reports {
+        unique_reports.init().await?;
+    }
+    if let Some(reports) = &config.reports {
+        reports.init().await?;
+    }
+    if let Some(no_repro) = &config.no_repro {
+        no_repro.init().await?;
+    }
+    let monitor_reports_future = monitor_reports(
+        reports_dir.path(),
+        &config.unique_reports,
+        &config.reports,
+        &config.no_repro,
+    );
+
     let inputs = SyncedDir {
         path: runtime_dir.path().join("inputs"),
         url: config.inputs.url.clone(),
     };
     inputs.init().await?;
-
     if let Some(context) = &config.wait_for_files {
         let dir = match context {
             ContainerType::Inputs => &inputs,
@@ -94,16 +115,10 @@ pub async fn spawn(config: SupervisorConfig) -> Result<(), Error> {
 
     let process = start_supervisor(
         &runtime_dir.path(),
-        &supervisor_path,
+        &config,
-        &config.target_exe,
+        &crashes,
-        &crashes.path,
+        &inputs,
-        &inputs.path,
+        reports_dir.path().to_path_buf(),
-        &config.target_options,
-        &config.supervisor_options,
-        &config.supervisor_env,
-        &config.supervisor_input_marker,
-        &config.common.setup_dir,
-        &config.tools.path,
     )
     .await?;
 
@@ -133,6 +148,7 @@ pub async fn spawn(config: SupervisorConfig) -> Result<(), Error> {
         monitor_stats,
         monitor_crashes,
         continuous_sync_task,
+        monitor_reports_future,
     )?;
 
     Ok(())
@@ -150,45 +166,49 @@ async fn heartbeat_process(
 
 async fn start_supervisor(
     runtime_dir: impl AsRef<Path>,
-    supervisor_path: impl AsRef<Path>,
+    config: &SupervisorConfig,
-    target_exe: impl AsRef<Path>,
+    crashes: &SyncedDir,
-    fault_dir: impl AsRef<Path>,
+    inputs: &SyncedDir,
-    inputs_dir: impl AsRef<Path>,
+    reports_dir: PathBuf,
-    target_options: &[String],
-    supervisor_options: &[String],
-    supervisor_env: &HashMap<String, String>,
-    supervisor_input_marker: &Option<String>,
-    setup_dir: impl AsRef<Path>,
-    tools_dir: impl AsRef<Path>,
 ) -> Result<Child> {
-    let mut cmd = Command::new(supervisor_path.as_ref());
+    let mut expand = Expand::new();
+    expand
+        .supervisor_exe(&config.supervisor_exe)
+        .supervisor_options(&config.supervisor_options)
+        .runtime_dir(&runtime_dir)
+        .crashes(&crashes.path)
+        .input_corpus(&inputs.path)
+        .reports_dir(&reports_dir)
+        .setup_dir(&config.common.setup_dir);
 
+    if let Some(tools) = &config.tools {
+        expand.tools_dir(&tools.path);
+    }
+
+    if let Some(target_exe) = &config.target_exe {
+        expand.target_exe(target_exe);
+    }
+
+    if let Some(target_options) = &config.target_options {
+        expand.target_options(target_options);
+    }
+
+    if let Some(input_marker) = &config.supervisor_input_marker {
+        expand.input_marker(input_marker);
+    }
+
+    let supervisor_path = expand.evaluate_value(&config.supervisor_exe)?;
+    let mut cmd = Command::new(supervisor_path);
     let cmd = cmd
         .kill_on_drop(true)
         .env_remove("RUST_LOG")
         .stdout(Stdio::piped())
         .stderr(Stdio::piped());
 
-    let mut expand = Expand::new();
+    let args = expand.evaluate(&config.supervisor_options)?;
-    expand
-        .supervisor_exe(supervisor_path)
-        .supervisor_options(supervisor_options)
-        .crashes(fault_dir)
-        .runtime_dir(runtime_dir)
-        .target_exe(target_exe)
-        .target_options(target_options)
-        .input_corpus(inputs_dir)
-        .setup_dir(setup_dir)
-        .tools_dir(tools_dir);
-
-    if let Some(input_marker) = supervisor_input_marker {
-        expand.input_marker(input_marker);
-    }
-
-    let args = expand.evaluate(supervisor_options)?;
     cmd.args(&args);
 
-    for (k, v) in supervisor_env {
+    for (k, v) in &config.supervisor_env {
         cmd.env(k, expand.evaluate_value(v)?);
     }
 
@@ -228,26 +248,39 @@ mod tests {
         use std::env;
 
         let runtime_dir = tempfile::tempdir().unwrap();
-        let afl_fuzz_exe = if let Ok(x) = env::var("ONEFUZZ_TEST_AFL_LINUX_FUZZER") {
+
-            x
+        let supervisor_exe = if let Ok(x) = env::var("ONEFUZZ_TEST_AFL_LINUX_FUZZER") {
+            x.into()
         } else {
             warn!("Unable to test AFL integration");
             return;
         };
 
-        let afl_test_binary = if let Ok(x) = env::var("ONEFUZZ_TEST_AFL_LINUX_TEST_BINARY") {
+        let target_exe = if let Ok(x) = env::var("ONEFUZZ_TEST_AFL_LINUX_TEST_BINARY") {
-            x
+            Some(x.into())
         } else {
             warn!("Unable to test AFL integration");
             return;
         };
 
+        let reports_dir_temp = tempfile::tempdir().unwrap();
+        let reports_dir = reports_dir_temp.path().into();
+
         let fault_dir_temp = tempfile::tempdir().unwrap();
-        let fault_dir = fault_dir_temp.path();
+        let crashes = SyncedDir {
+            path: fault_dir_temp.path().into(),
+            url: None,
+        };
+
         let corpus_dir_temp = tempfile::tempdir().unwrap();
-        let corpus_dir = corpus_dir_temp.into_path();
+        let corpus_dir = SyncedDir {
-        let seed_file_name = corpus_dir.clone().join("seed.txt");
+            path: corpus_dir_temp.path().into(),
-        let target_options = vec!["{input}".to_owned()];
+            url: None,
+        };
+        let seed_file_name = corpus_dir.path.join("seed.txt");
+        tokio::fs::write(seed_file_name, "xyz").await.unwrap();
+
+        let target_options = Some(vec!["{input}".to_owned()]);
         let supervisor_env = HashMap::new();
         let supervisor_options: Vec<_> = vec![
             "-d",
@@ -266,32 +299,24 @@ mod tests {
         // AFL input marker
         let supervisor_input_marker = Some("@@".to_owned());
 
-        println!(
+        let config = SupervisorConfig {
-            "testing 2: corpus_dir {:?} -- fault_dir {:?} -- seed_file_name {:?}",
+            supervisor_exe,
-            corpus_dir, fault_dir, seed_file_name
+            supervisor_env,
-        );
+            supervisor_options,
+            supervisor_input_marker,
+            target_exe,
+            target_options,
+            ..Default::default()
+        };
 
-        tokio::fs::write(seed_file_name, "xyz").await.unwrap();
+        let process = start_supervisor(runtime_dir, &config, &crashes, &corpus_dir, reports_dir)
-        let process = start_supervisor(
-            runtime_dir,
-            PathBuf::from(afl_fuzz_exe),
-            PathBuf::from(afl_test_binary),
-            fault_dir,
-            corpus_dir,
-            &target_options,
-            &supervisor_options,
-            &supervisor_env,
-            &supervisor_input_marker,
-            PathBuf::default(),
-            PathBuf::default(),
-        )
             .await
             .unwrap();
 
         let notify = Notify::new();
         let _fuzzing_monitor =
             monitor_process(process, "supervisor".to_string(), false, Some(&notify));
-        let stat_output = fault_dir.join("fuzzer_stats");
+        let stat_output = crashes.path.join("fuzzer_stats");
         let start = Instant::now();
         loop {
             if has_stats(&stat_output).await {

src/agent/onefuzz-agent/src/tasks/report/crash_report.rs

@@ -1,11 +1,13 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
-use anyhow::Result;
+use anyhow::{Context, Result};
+use futures::StreamExt;
 use onefuzz::{
     asan::AsanLog,
     blob::{BlobClient, BlobUrl},
     fs::exists,
+    monitor::DirectoryMonitor,
     syncdir::SyncedDir,
     telemetry::{
         Event::{new_report, new_unable_to_reproduce, new_unique_report},
@@ -15,7 +17,7 @@ use onefuzz::{
 use reqwest::{StatusCode, Url};
 use reqwest_retry::SendRetry;
 use serde::{Deserialize, Serialize};
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 use tokio::fs;
 use uuid::Uuid;
 
@@ -104,17 +106,19 @@ async fn upload_or_save_local<T: Serialize>(
 impl CrashTestResult {
     pub async fn save(
         &self,
-        unique_reports: &SyncedDir,
+        unique_reports: &Option<SyncedDir>,
         reports: &Option<SyncedDir>,
         no_repro: &Option<SyncedDir>,
     ) -> Result<()> {
         match self {
             Self::CrashReport(report) => {
                 // Use SHA-256 of call stack as dedupe key.
+                if let Some(unique_reports) = unique_reports {
                     let name = report.unique_blob_name();
                     if upload_or_save_local(&report, &name, unique_reports).await? {
                         event!(new_unique_report; EventData::Path = name);
                     }
+                }
 
                 if let Some(reports) = reports {
                     let name = report.blob_name();
@@ -193,3 +197,42 @@ impl NoCrash {
         format!("{}.json", self.input_sha256)
     }
 }
+
+async fn parse_report_file(path: PathBuf) -> Result<CrashTestResult> {
+    let raw = std::fs::read_to_string(&path)
+        .with_context(|| format_err!("unable to open crash report: {}", path.display()))?;
+
+    let json: serde_json::Value = serde_json::from_str(&raw)
+        .with_context(|| format_err!("invalid json: {} - {:?}", path.display(), raw))?;
+
+    let report: Result<CrashReport, serde_json::Error> = serde_json::from_value(json.clone());
+    if let Ok(report) = report {
+        return Ok(CrashTestResult::CrashReport(report));
+    }
+    let no_repro: Result<NoCrash, serde_json::Error> = serde_json::from_value(json);
+    if let Ok(no_repro) = no_repro {
+        return Ok(CrashTestResult::NoRepro(no_repro));
+    }
+
+    bail!("unable to parse report: {} - {:?}", path.display(), raw)
+}
+
+pub async fn monitor_reports(
+    base_dir: &Path,
+    unique_reports: &Option<SyncedDir>,
+    reports: &Option<SyncedDir>,
+    no_crash: &Option<SyncedDir>,
+) -> Result<()> {
+    if unique_reports.is_none() && reports.is_none() && no_crash.is_none() {
+        verbose!("no report directories configured");
+        return Ok(());
+    }
+
+    let mut monitor = DirectoryMonitor::new(base_dir);
+    monitor.start()?;
+    while let Some(file) = monitor.next().await {
+        let result = parse_report_file(file).await?;
+        result.save(unique_reports, reports, no_crash).await?;
+    }
+    Ok(())
+}

src/agent/onefuzz-agent/src/tasks/report/generic.rs

@@ -35,7 +35,7 @@ pub struct Config {
     pub input_queue: Option<Url>,
     pub crashes: Option<SyncedDir>,
     pub reports: Option<SyncedDir>,
-    pub unique_reports: SyncedDir,
+    pub unique_reports: Option<SyncedDir>,
     pub no_repro: Option<SyncedDir>,
 
     pub target_timeout: Option<u64>,

src/agent/onefuzz-agent/src/tasks/report/libfuzzer_report.rs

@@ -30,7 +30,7 @@ pub struct Config {
     pub input_queue: Option<Url>,
     pub crashes: Option<SyncedDir>,
     pub reports: Option<SyncedDir>,
-    pub unique_reports: SyncedDir,
+    pub unique_reports: Option<SyncedDir>,
     pub no_repro: Option<SyncedDir>,
 
     #[serde(default = "default_bool_true")]
@@ -67,7 +67,9 @@ impl ReportTask {
         };
         crashes.init().await?;
 
-        self.config.unique_reports.init().await?;
+        if let Some(unique_reports) = &self.config.unique_reports {
+            unique_reports.init().await?;
+        }
         if let Some(reports) = &self.config.reports {
             reports.init().await?;
         }

src/agent/onefuzz/src/expand.rs

@@ -34,6 +34,7 @@ pub enum PlaceHolder {
     SupervisorExe,
     SupervisorOptions,
     SetupDir,
+    ReportsDir,
 }
 
 impl PlaceHolder {
@@ -57,6 +58,7 @@ impl PlaceHolder {
             Self::SupervisorExe => "{supervisor_exe}",
             Self::SupervisorOptions => "{supervisor_options}",
             Self::SetupDir => "{setup_dir}",
+            Self::ReportsDir => "{reports_dir}",
         }
         .to_string()
     }
@@ -203,6 +205,13 @@ impl<'a> Expand<'a> {
         self
     }
 
+    pub fn reports_dir(&mut self, arg: impl AsRef<Path>) -> &mut Self {
+        let arg = arg.as_ref();
+        let path = String::from(arg.to_string_lossy());
+        self.set_value(PlaceHolder::ReportsDir, ExpandedValue::Path(path));
+        self
+    }
+
     pub fn tools_dir(&mut self, arg: impl AsRef<Path>) -> &mut Self {
         let arg = arg.as_ref();
         let path = String::from(arg.to_string_lossy());

src/agent/onefuzz/src/syncdir.rs

@@ -23,7 +23,7 @@ pub enum SyncOperation {
 const DELAY: Duration = Duration::from_secs(10);
 const DEFAULT_CONTINUOUS_SYNC_DELAY_SECONDS: u64 = 60;
 
-#[derive(Debug, Deserialize, Clone, PartialEq)]
+#[derive(Debug, Deserialize, Clone, PartialEq, Default)]
 pub struct SyncedDir {
     pub path: PathBuf,
     pub url: Option<BlobContainerUrl>,