Centralized Testcase evaluation EventManager (#1216)

* template

* moar

* merge

* compiles

* fuzzer

* forward event newtestcase

* clippy

fuzzers/libfuzzer_libpng_centralized/.gitignore

@@ -0,0 +1,2 @@
+fuzzer_stats.toml
+libpng-*

fuzzers/libfuzzer_libpng_centralized/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "libfuzzer_libpng_launcher"
+version = "0.10.0"
+authors = ["Andrea Fioraldi <andreafioraldi@gmail.com>", "Dominik Maier <domenukk@gmail.com>"]
+edition = "2021"
+
+[features]
+default = ["std"]
+std = []
+
+[profile.release]
+lto = true
+codegen-units = 1
+opt-level = 3
+debug = true
+
+[build-dependencies]
+cc = { version = "1.0", features = ["parallel"] }
+which = { version = "4.0.2" }
+
+[dependencies]
+libafl = { path = "../../libafl/", features = ["std", "derive", "rand_trait", "fork", "prelude", "gzip", "regex"] }
+libafl_targets = { path = "../../libafl_targets/", features = ["sancov_pcguard_hitcounts", "libfuzzer"] }
+# TODO Include it only when building cc
+libafl_cc = { path = "../../libafl_cc/" }
+clap = { version = "4.0", features = ["derive"] }
+mimalloc = { version = "*", default-features = false }
+env_logger = "0.10"
+
+[lib]
+name = "libfuzzer_libpng"
+crate-type = ["staticlib"]

fuzzers/libfuzzer_libpng_centralized/Makefile.toml

@@ -0,0 +1,131 @@
+# Variables
+[env]
+FUZZER_NAME='fuzzer_libpng_launcher'
+CARGO_TARGET_DIR = { value = "${PROJECT_DIR}/target", condition = { env_not_set = ["CARGO_TARGET_DIR"] } }
+LIBAFL_CC = '${CARGO_TARGET_DIR}/release/libafl_cc'
+LIBAFL_CXX = '${CARGO_TARGET_DIR}/release/libafl_cxx'
+FUZZER = '${CARGO_TARGET_DIR}/release/${FUZZER_NAME}'
+PROJECT_DIR = { script = ["pwd"] }
+
+[tasks.unsupported]
+script_runner="@shell"
+script='''
+echo "Cargo-make not integrated yet on this platform"
+'''
+
+# libpng
+[tasks.libpng]
+linux_alias = "libpng_unix"
+mac_alias = "libpng_unix"
+windows_alias = "unsupported"
+
+[tasks.libpng_unix]
+condition = { files_not_exist = ["./libpng-1.6.37"]}
+script_runner="@shell"
+script='''
+wget https://deac-fra.dl.sourceforge.net/project/libpng/libpng16/1.6.37/libpng-1.6.37.tar.xz
+tar -xvf libpng-1.6.37.tar.xz
+'''
+
+# Compilers
+[tasks.cxx]
+linux_alias = "cxx_unix"
+mac_alias = "cxx_unix"
+windows_alias = "unsupported"
+
+[tasks.cxx_unix]
+command = "cargo"
+args = ["build" , "--release"]
+
+[tasks.cc]
+linux_alias = "cc_unix"
+mac_alias = "cc_unix"
+windows_alias = "unsupported"
+
+[tasks.cc_unix]
+command = "cargo"
+args = ["build" , "--release"]
+
+# Library
+[tasks.lib]
+linux_alias = "lib_unix"
+mac_alias = "lib_unix"
+windows_alias = "unsupported"
+
+[tasks.lib_unix]
+script_runner="@shell"
+script='''
+cd libpng-1.6.37 && ./configure --enable-shared=no --with-pic=yes --enable-hardware-optimizations=yes
+cd "${PROJECT_DIR}"
+make -C libpng-1.6.37 CC="${CARGO_TARGET_DIR}/release/libafl_cc" CXX="${CARGO_TARGET_DIR}/release/libafl_cxx"
+'''
+dependencies = [ "libpng", "cxx", "cc" ]
+
+
+# Harness
+[tasks.fuzzer]
+linux_alias = "fuzzer_unix"
+mac_alias = "fuzzer_unix"
+windows_alias = "unsupported"
+
+[tasks.fuzzer_unix]
+command = "${CARGO_TARGET_DIR}/release/libafl_cxx"
+args = ["${PROJECT_DIR}/harness.cc", "${PROJECT_DIR}/libpng-1.6.37/.libs/libpng16.a", "-I", "${PROJECT_DIR}/libpng-1.6.37/", "-o", "${FUZZER_NAME}", "-lm", "-lz"]
+dependencies = [ "lib", "cxx", "cc" ]
+
+# Run the fuzzer
+[tasks.run]
+linux_alias = "run_unix"
+mac_alias = "run_unix"
+windows_alias = "unsupported"
+
+[tasks.run_unix]
+script_runner = "@shell"
+script='''
+./${FUZZER_NAME} --cores 0 --input ./corpus
+'''
+dependencies = [ "fuzzer" ]
+
+# Test
+[tasks.test]
+linux_alias = "test_unix"
+mac_alias = "test_mac"
+windows_alias = "unsupported"
+
+[tasks.test_unix]
+script_runner = "@shell"
+script='''
+rm -rf libafl_unix_shmem_server || true
+timeout 11s ./${FUZZER_NAME} --cores 0 --input ./corpus 2>/dev/null >fuzz_stdout.log || true
+if [ -z "$(grep "corpus: 30" fuzz_stdout.log)" ]; then
+    echo "Fuzzer does not generate any testcases or any crashes"
+    exit 1
+else
+    echo "Fuzzer is working"
+fi
+'''
+dependencies = [ "fuzzer" ]
+
+[tasks.test_mac]
+script_runner = "@shell"
+script='''
+rm -rf libafl_unix_shmem_server || true
+timeout 11s ./${FUZZER_NAME} --cores 0 --input ./corpus 2>/dev/null >fuzz_stdout.log || true
+'''
+dependencies = [ "fuzzer" ]
+
+# Clean up
+[tasks.clean]
+linux_alias = "clean_unix"
+mac_alias = "clean_unix"
+windows_alias = "unsupported"
+
+[tasks.clean_unix]
+# Disable default `clean` definition
+clear = true
+script_runner="@shell"
+script='''
+rm -f ./${FUZZER_NAME}
+make -C libpng-1.6.37 clean
+cargo clean
+'''

fuzzers/libfuzzer_libpng_centralized/README.md

@@ -0,0 +1,47 @@
+# Libfuzzer for libpng, with launcher
+
+This folder contains an example fuzzer for libpng, using LLMP for fast multi-process fuzzing and crash detection.
+To show off crash detection, we added a `ud2` instruction to the harness, edit harness.cc if you want a non-crashing example.
+It has been tested on Linux.
+
+In contrast to the normal libfuzzer libpng example, this uses the `launcher` feature, that automatically spawns `n` child processes, and binds them to a free core.
+
+## Build
+
+To build this example, run
+
+```bash
+cargo build --release
+```
+
+This will build the library with the fuzzer (src/lib.rs) with the libfuzzer compatibility layer and the SanitizerCoverage runtime functions for coverage feedback.
+In addition, it will also build two C and C++ compiler wrappers (bin/libafl_c(libafl_c/xx).rs) that you must use to compile the target.
+
+Then download libpng, and unpack the archive:
+```bash
+wget https://deac-fra.dl.sourceforge.net/project/libpng/libpng16/1.6.37/libpng-1.6.37.tar.xz
+tar -xvf libpng-1.6.37.tar.xz
+```
+
+Now compile libpng, using the libafl_cc compiler wrapper:
+
+```bash
+cd libpng-1.6.37
+./configure
+make CC=../target/release/libafl_cc CXX=../target/release/libafl_cxx -j `nproc`
+```
+
+You can find the static lib at `libpng-1.6.37/.libs/libpng16.a`.
+
+Now, we have to build the libfuzzer harness and link all together to create our fuzzer binary.
+
+```
+cd ..
+./target/release/libafl_cxx ./harness.cc libpng-1.6.37/.libs/libpng16.a -I libpng-1.6.37/ -o fuzzer_libpng -lz -lm
+```
+
+Afterwards, the fuzzer will be ready to run.
+
+## Run
+
+Just run once, the launcher feature should do the rest.

fuzzers/libfuzzer_libpng_centralized/harness.cc

@@ -0,0 +1,188 @@
+// libpng_read_fuzzer.cc
+// Copyright 2017-2018 Glenn Randers-Pehrson
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that may
+// be found in the LICENSE file https://cs.chromium.org/chromium/src/LICENSE
+
+// Last changed in libpng 1.6.35 [July 15, 2018]
+
+// The modifications in 2017 by Glenn Randers-Pehrson include
+// 1. addition of a PNG_CLEANUP macro,
+// 2. setting the option to ignore ADLER32 checksums,
+// 3. adding "#include <string.h>" which is needed on some platforms
+//    to provide memcpy().
+// 4. adding read_end_info() and creating an end_info structure.
+// 5. adding calls to png_set_*() transforms commonly used by browsers.
+
+#include <stddef.h>
+#include <stdint.h>
+#include <string.h>
+
+#include <vector>
+
+#define PNG_INTERNAL
+#include "png.h"
+
+#define PNG_CLEANUP                                                        \
+  if (png_handler.png_ptr) {                                               \
+    if (png_handler.row_ptr)                                               \
+      png_free(png_handler.png_ptr, png_handler.row_ptr);                  \
+    if (png_handler.end_info_ptr)                                          \
+      png_destroy_read_struct(&png_handler.png_ptr, &png_handler.info_ptr, \
+                              &png_handler.end_info_ptr);                  \
+    else if (png_handler.info_ptr)                                         \
+      png_destroy_read_struct(&png_handler.png_ptr, &png_handler.info_ptr, \
+                              nullptr);                                    \
+    else                                                                   \
+      png_destroy_read_struct(&png_handler.png_ptr, nullptr, nullptr);     \
+    png_handler.png_ptr = nullptr;                                         \
+    png_handler.row_ptr = nullptr;                                         \
+    png_handler.info_ptr = nullptr;                                        \
+    png_handler.end_info_ptr = nullptr;                                    \
+  }
+
+struct BufState {
+  const uint8_t *data;
+  size_t         bytes_left;
+};
+
+struct PngObjectHandler {
+  png_infop   info_ptr = nullptr;
+  png_structp png_ptr = nullptr;
+  png_infop   end_info_ptr = nullptr;
+  png_voidp   row_ptr = nullptr;
+  BufState   *buf_state = nullptr;
+
+  ~PngObjectHandler() {
+    if (row_ptr) { png_free(png_ptr, row_ptr); }
+    if (end_info_ptr)
+      png_destroy_read_struct(&png_ptr, &info_ptr, &end_info_ptr);
+    else if (info_ptr)
+      png_destroy_read_struct(&png_ptr, &info_ptr, nullptr);
+    else
+      png_destroy_read_struct(&png_ptr, nullptr, nullptr);
+    delete buf_state;
+  }
+};
+
+void user_read_data(png_structp png_ptr, png_bytep data, size_t length) {
+  BufState *buf_state = static_cast<BufState *>(png_get_io_ptr(png_ptr));
+  if (length > buf_state->bytes_left) { png_error(png_ptr, "read error"); }
+  memcpy(data, buf_state->data, length);
+  buf_state->bytes_left -= length;
+  buf_state->data += length;
+}
+
+static const int kPngHeaderSize = 8;
+
+// Entry point for LibFuzzer.
+// Roughly follows the libpng book example:
+// http://www.libpng.org/pub/png/book/chapter13.html
+extern "C" int LLVMFuzzerTestOneInput(const uint8_t *data, size_t size) {
+  if (size < kPngHeaderSize) { return 0; }
+
+  std::vector<unsigned char> v(data, data + size);
+  if (png_sig_cmp(v.data(), 0, kPngHeaderSize)) {
+    // not a PNG.
+    return 0;
+  }
+
+  PngObjectHandler png_handler;
+  png_handler.png_ptr = nullptr;
+  png_handler.row_ptr = nullptr;
+  png_handler.info_ptr = nullptr;
+  png_handler.end_info_ptr = nullptr;
+
+  png_handler.png_ptr =
+      png_create_read_struct(PNG_LIBPNG_VER_STRING, nullptr, nullptr, nullptr);
+  if (!png_handler.png_ptr) { return 0; }
+
+  png_handler.info_ptr = png_create_info_struct(png_handler.png_ptr);
+  if (!png_handler.info_ptr) {
+    PNG_CLEANUP
+    return 0;
+  }
+
+  png_handler.end_info_ptr = png_create_info_struct(png_handler.png_ptr);
+  if (!png_handler.end_info_ptr) {
+    PNG_CLEANUP
+    return 0;
+  }
+
+  png_set_crc_action(png_handler.png_ptr, PNG_CRC_QUIET_USE, PNG_CRC_QUIET_USE);
+#ifdef PNG_IGNORE_ADLER32
+  png_set_option(png_handler.png_ptr, PNG_IGNORE_ADLER32, PNG_OPTION_ON);
+#endif
+
+  // Setting up reading from buffer.
+  png_handler.buf_state = new BufState();
+  png_handler.buf_state->data = data + kPngHeaderSize;
+  png_handler.buf_state->bytes_left = size - kPngHeaderSize;
+  png_set_read_fn(png_handler.png_ptr, png_handler.buf_state, user_read_data);
+  png_set_sig_bytes(png_handler.png_ptr, kPngHeaderSize);
+
+  if (setjmp(png_jmpbuf(png_handler.png_ptr))) {
+    PNG_CLEANUP
+    return 0;
+  }
+
+  // Reading.
+  png_read_info(png_handler.png_ptr, png_handler.info_ptr);
+
+  // reset error handler to put png_deleter into scope.
+  if (setjmp(png_jmpbuf(png_handler.png_ptr))) {
+    PNG_CLEANUP
+    return 0;
+  }
+
+  png_uint_32 width, height;
+  int         bit_depth, color_type, interlace_type, compression_type;
+  int         filter_type;
+
+  if (!png_get_IHDR(png_handler.png_ptr, png_handler.info_ptr, &width, &height,
+                    &bit_depth, &color_type, &interlace_type, &compression_type,
+                    &filter_type)) {
+    PNG_CLEANUP
+    return 0;
+  }
+
+  // This is going to be too slow.
+  if (width && height > 100000000 / width) {
+    PNG_CLEANUP
+#ifdef HAS_DUMMY_CRASH
+  #ifdef __aarch64__
+    asm volatile(".word 0xf7f0a000\n");
+  #else
+    asm("ud2");
+  #endif
+#endif
+    return 0;
+  }
+
+  // Set several transforms that browsers typically use:
+  png_set_gray_to_rgb(png_handler.png_ptr);
+  png_set_expand(png_handler.png_ptr);
+  png_set_packing(png_handler.png_ptr);
+  png_set_scale_16(png_handler.png_ptr);
+  png_set_tRNS_to_alpha(png_handler.png_ptr);
+
+  int passes = png_set_interlace_handling(png_handler.png_ptr);
+
+  png_read_update_info(png_handler.png_ptr, png_handler.info_ptr);
+
+  png_handler.row_ptr =
+      png_malloc(png_handler.png_ptr,
+                 png_get_rowbytes(png_handler.png_ptr, png_handler.info_ptr));
+
+  for (int pass = 0; pass < passes; ++pass) {
+    for (png_uint_32 y = 0; y < height; ++y) {
+      png_read_row(png_handler.png_ptr,
+                   static_cast<png_bytep>(png_handler.row_ptr), nullptr);
+    }
+  }
+
+  png_read_end(png_handler.png_ptr, png_handler.end_info_ptr);
+
+  PNG_CLEANUP
+  return 0;
+}

fuzzers/libfuzzer_libpng_centralized/src/bin/libafl_cc.rs

@@ -0,0 +1,36 @@
+use std::env;
+
+use libafl_cc::{ClangWrapper, CompilerWrapper};
+
+pub fn main() {
+    let args: Vec<String> = env::args().collect();
+    if args.len() > 1 {
+        let mut dir = env::current_exe().unwrap();
+        let wrapper_name = dir.file_name().unwrap().to_str().unwrap();
+
+        let is_cpp = match wrapper_name[wrapper_name.len()-2..].to_lowercase().as_str() {
+            "cc" => false,
+            "++" | "pp" | "xx" => true,
+            _ => panic!("Could not figure out if c or c++ wrapper was called. Expected {dir:?} to end with c or cxx"),
+        };
+
+        dir.pop();
+
+        let mut cc = ClangWrapper::new();
+        if let Some(code) = cc
+            .cpp(is_cpp)
+            // silence the compiler wrapper output, needed for some configure scripts.
+            .silence(true)
+            .parse_args(&args)
+            .expect("Failed to parse the command line")
+            .link_staticlib(&dir, "libfuzzer_libpng")
+            .add_arg("-fsanitize-coverage=trace-pc-guard")
+            .run()
+            .expect("Failed to run the wrapped compiler")
+        {
+            std::process::exit(code);
+        }
+    } else {
+        panic!("LibAFL CC: No Arguments given");
+    }
+}

fuzzers/libfuzzer_libpng_centralized/src/bin/libafl_cxx.rs

@@ -0,0 +1,5 @@
+pub mod libafl_cc;
+
+fn main() {
+    libafl_cc::main();
+}

fuzzers/libfuzzer_libpng_centralized/src/lib.rs

@@ -0,0 +1,292 @@
+//! A libfuzzer-like fuzzer with llmp-multithreading support and restarts
+//! The example harness is built for libpng.
+//! In this example, you will see the use of the `launcher` feature.
+//! The `launcher` will spawn new processes for each cpu core.
+use mimalloc::MiMalloc;
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
+use core::time::Duration;
+use std::{env, net::SocketAddr, path::PathBuf};
+
+use clap::{self, Parser};
+use libafl::{
+    bolts::{
+        core_affinity::{CoreId, Cores},
+        current_nanos,
+        launcher::Launcher,
+        llmp::{LlmpReceiver, LlmpSender},
+        rands::StdRand,
+        shmem::{ShMemProvider, StdShMemProvider},
+        tuples::{tuple_list, Merge},
+        AsSlice, ClientId,
+    },
+    corpus::{Corpus, InMemoryCorpus, OnDiskCorpus},
+    events::{CentralizedEventManager, EventConfig},
+    executors::{inprocess::InProcessExecutor, ExitKind, TimeoutExecutor},
+    feedback_or, feedback_or_fast,
+    feedbacks::{CrashFeedback, MaxMapFeedback, TimeFeedback, TimeoutFeedback},
+    fuzzer::{Fuzzer, StdFuzzer},
+    inputs::{BytesInput, HasTargetBytes},
+    monitors::MultiMonitor,
+    mutators::{
+        scheduled::{havoc_mutations, tokens_mutations, StdScheduledMutator},
+        token_mutations::Tokens,
+    },
+    observers::{HitcountsMapObserver, TimeObserver},
+    schedulers::{IndexesLenTimeMinimizerScheduler, QueueScheduler},
+    stages::mutational::StdMutationalStage,
+    state::{HasCorpus, HasMetadata, StdState},
+    Error,
+};
+use libafl_targets::{libfuzzer_initialize, libfuzzer_test_one_input, std_edges_map_observer};
+
+/// Parse a millis string to a [`Duration`]. Used for arg parsing.
+fn timeout_from_millis_str(time: &str) -> Result<Duration, Error> {
+    Ok(Duration::from_millis(time.parse()?))
+}
+
+/// The commandline args this fuzzer accepts
+#[derive(Debug, Parser)]
+#[command(
+    name = "libfuzzer_libpng_launcher",
+    about = "A libfuzzer-like fuzzer for libpng with llmp-multithreading support and a launcher",
+    author = "Andrea Fioraldi <andreafioraldi@gmail.com>, Dominik Maier <domenukk@gmail.com>"
+)]
+struct Opt {
+    #[arg(
+        short,
+        long,
+        value_parser = Cores::from_cmdline,
+        help = "Spawn a client in each of the provided cores. Broker runs in the 0th core. 'all' to select all available cores. 'none' to run a client without binding to any core. eg: '1,2-4,6' selects the cores 1,2,3,4,6.",
+        name = "CORES"
+    )]
+    cores: Cores,
+
+    #[arg(
+        short = 'p',
+        long,
+        help = "Choose the broker TCP port, default is 1337",
+        name = "PORT",
+        default_value = "1337"
+    )]
+    broker_port: u16,
+
+    #[arg(short = 'a', long, help = "Specify a remote broker", name = "REMOTE")]
+    remote_broker_addr: Option<SocketAddr>,
+
+    #[arg(short, long, help = "Set an initial corpus directory", name = "INPUT")]
+    input: Vec<PathBuf>,
+
+    #[arg(
+        short,
+        long,
+        help = "Set the output directory, default is ./out",
+        name = "OUTPUT",
+        default_value = "./out"
+    )]
+    output: PathBuf,
+
+    #[arg(
+        value_parser = timeout_from_millis_str,
+        short,
+        long,
+        help = "Set the exeucution timeout in milliseconds, default is 10000",
+        name = "TIMEOUT",
+        default_value = "10000"
+    )]
+    timeout: Duration,
+    /*
+    /// This fuzzer has hard-coded tokens
+    #[arg(
+
+        short = "x",
+        long,
+        help = "Feed the fuzzer with an user-specified list of tokens (often called \"dictionary\"",
+        name = "TOKENS",
+        multiple = true
+    )]
+    tokens: Vec<PathBuf>,
+    */
+}
+
+/// The main fn, `no_mangle` as it is a C symbol
+#[no_mangle]
+pub fn libafl_main() {
+    env_logger::init();
+
+    // Registry the metadata types used in this fuzzer
+    // Needed only on no_std
+    //RegistryBuilder::register::<Tokens>();
+    let opt = Opt::parse();
+
+    let broker_port = opt.broker_port;
+    let cores = opt.cores;
+
+    println!(
+        "Workdir: {:?}",
+        env::current_dir().unwrap().to_string_lossy().to_string()
+    );
+
+    let shmem_provider = StdShMemProvider::new().expect("Failed to init shared memory");
+
+    let mut senders = vec![];
+    let mut receivers = vec![];
+    let mut main_core_id = None;
+    let mut core_id_map = std::collections::HashMap::<CoreId, usize>::default();
+    for core_id in &cores.ids {
+        if main_core_id.is_none() {
+            main_core_id = Some(core_id.clone());
+            continue;
+        }
+        let sender =
+            LlmpSender::new(shmem_provider.clone(), ClientId(core_id.0 as u32), false).unwrap();
+        let receiver = LlmpReceiver::on_existing_shmem(
+            shmem_provider.clone(),
+            sender.out_shmems[0].shmem.clone(),
+            None,
+        )
+        .unwrap();
+
+        core_id_map.insert(core_id.clone(), senders.len());
+        senders.push(Some(sender));
+        receivers.push(receiver);
+    }
+
+    eprintln!("Main is {main_core_id:?}");
+
+    let mut receivers = Some(receivers);
+
+    let monitor = MultiMonitor::new(|s| println!("{s}"));
+
+    let mut run_client = |state: Option<_>, restarting_mgr, core_id: CoreId| {
+        // Create an observation channel using the coverage map
+        let edges_observer = HitcountsMapObserver::new(unsafe { std_edges_map_observer("edges") });
+
+        // Create an observation channel to keep track of the execution time
+        let time_observer = TimeObserver::new("time");
+
+        // Feedback to rate the interestingness of an input
+        // This one is composed by two Feedbacks in OR
+        let mut feedback = feedback_or!(
+            // New maximization map feedback linked to the edges observer and the feedback state
+            MaxMapFeedback::tracking(&edges_observer, true, false),
+            // Time feedback, this one does not need a feedback state
+            TimeFeedback::with_observer(&time_observer)
+        );
+
+        // A feedback to choose if an input is a solution or not
+        let mut objective = feedback_or_fast!(CrashFeedback::new(), TimeoutFeedback::new());
+
+        // If not restarting, create a State from scratch
+        let mut state = state.unwrap_or_else(|| {
+            StdState::new(
+                // RNG
+                StdRand::with_seed(current_nanos()),
+                // Corpus that will be evolved, we keep it in memory for performance
+                InMemoryCorpus::new(),
+                // Corpus in which we store solutions (crashes in this example),
+                // on disk so the user can get them after stopping the fuzzer
+                OnDiskCorpus::new(&opt.output).unwrap(),
+                // States of the feedbacks.
+                // The feedbacks can report the data that should persist in the State.
+                &mut feedback,
+                // Same for objective feedbacks
+                &mut objective,
+            )
+            .unwrap()
+        });
+
+        let mut mgr = if main_core_id.unwrap() == core_id {
+            CentralizedEventManager::new_main(restarting_mgr, receivers.take().unwrap())
+        } else {
+            let idx = *core_id_map.get(&core_id).unwrap();
+            CentralizedEventManager::new_secondary(restarting_mgr, senders[idx].take().unwrap())
+        };
+
+        // let mut mgr = restarting_mgr;
+
+        println!("We're a client, let's fuzz :)");
+
+        // Create a PNG dictionary if not existing
+        if state.metadata_map().get::<Tokens>().is_none() {
+            state.add_metadata(Tokens::from([
+                vec![137, 80, 78, 71, 13, 10, 26, 10], // PNG header
+                "IHDR".as_bytes().to_vec(),
+                "IDAT".as_bytes().to_vec(),
+                "PLTE".as_bytes().to_vec(),
+                "IEND".as_bytes().to_vec(),
+            ]));
+        }
+
+        // Setup a basic mutator with a mutational stage
+        let mutator = StdScheduledMutator::new(havoc_mutations().merge(tokens_mutations()));
+        let mut stages = tuple_list!(StdMutationalStage::new(mutator));
+
+        // A minimization+queue policy to get testcasess from the corpus
+        let scheduler = IndexesLenTimeMinimizerScheduler::new(QueueScheduler::new());
+
+        // A fuzzer with feedbacks and a corpus scheduler
+        let mut fuzzer = StdFuzzer::new(scheduler, feedback, objective);
+
+        // The wrapped harness function, calling out to the LLVM-style harness
+        let mut harness = |input: &BytesInput| {
+            let target = input.target_bytes();
+            let buf = target.as_slice();
+            libfuzzer_test_one_input(buf);
+            ExitKind::Ok
+        };
+
+        // Create the executor for an in-process function with one observer for edge coverage and one for the execution time
+        let executor = InProcessExecutor::new(
+            &mut harness,
+            tuple_list!(edges_observer, time_observer),
+            &mut fuzzer,
+            &mut state,
+            &mut mgr,
+        )?;
+
+        // Wrap the executor with a timeout
+        #[cfg(target_os = "linux")]
+        let mut executor = TimeoutExecutor::batch_mode(executor, opt.timeout);
+
+        // Wrap the executor with a timeout
+        #[cfg(not(target_os = "linux"))]
+        let mut executor = TimeoutExecutor::new(executor, opt.timeout);
+
+        // The actual target run starts here.
+        // Call LLVMFUzzerInitialize() if present.
+        let args: Vec<String> = env::args().collect();
+        if libfuzzer_initialize(&args) == -1 {
+            println!("Warning: LLVMFuzzerInitialize failed with -1");
+        }
+
+        // In case the corpus is empty (on first run), reset
+        if state.must_load_initial_inputs() {
+            state
+                .load_initial_inputs(&mut fuzzer, &mut executor, &mut mgr, &opt.input)
+                .unwrap_or_else(|_| panic!("Failed to load initial corpus at {:?}", &opt.input));
+            println!("We imported {} inputs from disk.", state.corpus().count());
+        }
+
+        fuzzer.fuzz_loop(&mut stages, &mut executor, &mut state, &mut mgr)?;
+        Ok(())
+    };
+
+    match Launcher::builder()
+        .shmem_provider(shmem_provider)
+        .configuration(EventConfig::from_name("default"))
+        .monitor(monitor)
+        .run_client(&mut run_client)
+        .cores(&cores)
+        .broker_port(broker_port)
+        .remote_broker_addr(opt.remote_broker_addr)
+        .stdout_file(Some("/dev/null"))
+        .build()
+        .launch()
+    {
+        Ok(()) => (),
+        Err(Error::ShuttingDown) => println!("Fuzzing stopped by user. Good bye."),
+        Err(err) => panic!("Failed to run launcher: {err:?}"),
+    }
+}

libafl/src/events/centralized.rs

@@ -0,0 +1,283 @@
+//! A wrapper manager to implement a main-secondary architecture with point-to-point channels
+
+use alloc::{boxed::Box, string::String, vec::Vec};
+
+use serde::{Deserialize, Serialize};
+
+use super::{CustomBufEventResult, HasCustomBufHandlers, ProgressReporter};
+use crate::{
+    bolts::{
+        llmp::{LlmpReceiver, LlmpSender, Tag},
+        shmem::ShMemProvider,
+        ClientId,
+    },
+    events::{
+        Event, EventConfig, EventFirer, EventManager, EventManagerId, EventProcessor,
+        EventRestarter, HasEventManagerId, LogSeverity,
+    },
+    executors::{Executor, HasObservers},
+    fuzzer::{EvaluatorObservers, ExecutionProcessor},
+    inputs::UsesInput,
+    observers::ObserversTuple,
+    state::{HasClientPerfMonitor, HasExecutions, HasMetadata, UsesState},
+    Error,
+};
+
+const _LLMP_TAG_TO_MAIN: Tag = Tag(0x3453453);
+
+/// A wrapper manager to implement a main-secondary architecture with point-to-point channels
+#[derive(Debug)]
+pub struct CentralizedEventManager<EM, SP>
+where
+    EM: UsesState,
+    SP: ShMemProvider,
+{
+    inner: EM,
+    sender_to_main: Option<LlmpSender<SP>>,
+    receivers_from_secondary: Option<Vec<LlmpReceiver<SP>>>,
+}
+
+impl<EM, SP> UsesState for CentralizedEventManager<EM, SP>
+where
+    EM: UsesState,
+    SP: ShMemProvider,
+{
+    type State = EM::State;
+}
+
+impl<EM, SP> EventFirer for CentralizedEventManager<EM, SP>
+where
+    EM: EventFirer + HasEventManagerId,
+    SP: ShMemProvider,
+{
+    fn fire(
+        &mut self,
+        state: &mut Self::State,
+        mut event: Event<<Self::State as UsesInput>::Input>,
+    ) -> Result<(), Error> {
+        if let Some(sender) = self.sender_to_main.as_mut() {
+            // secondary node
+            let is_nt = match &mut event {
+                Event::NewTestcase {
+                    input: _,
+                    client_config: _,
+                    exit_kind: _,
+                    corpus_size: _,
+                    observers_buf: _,
+                    time: _,
+                    executions: _,
+                    forward_id,
+                } => {
+                    *forward_id = Some(ClientId(self.inner.mgr_id().0 as u32));
+                    true
+                }
+                _ => false,
+            };
+            if is_nt {
+                // TODO use copression when llmp_compression is enabled
+                let serialized = postcard::to_allocvec(&event)?;
+                return sender.send_buf(_LLMP_TAG_TO_MAIN, &serialized);
+            }
+        }
+        self.inner.fire(state, event)
+    }
+
+    fn log(
+        &mut self,
+        state: &mut Self::State,
+        severity_level: LogSeverity,
+        message: String,
+    ) -> Result<(), Error> {
+        self.inner.log(state, severity_level, message)
+    }
+
+    fn serialize_observers<OT>(&mut self, observers: &OT) -> Result<Vec<u8>, Error>
+    where
+        OT: ObserversTuple<Self::State> + Serialize,
+    {
+        self.inner.serialize_observers(observers)
+    }
+
+    fn configuration(&self) -> EventConfig {
+        self.inner.configuration()
+    }
+}
+
+impl<EM, SP> EventRestarter for CentralizedEventManager<EM, SP>
+where
+    EM: EventRestarter,
+    SP: ShMemProvider,
+{
+    #[inline]
+    fn on_restart(&mut self, state: &mut Self::State) -> Result<(), Error> {
+        self.inner.on_restart(state)
+    }
+
+    fn send_exiting(&mut self) -> Result<(), Error> {
+        self.inner.send_exiting()
+    }
+
+    #[inline]
+    fn await_restart_safe(&mut self) {
+        self.inner.await_restart_safe();
+    }
+}
+
+impl<E, EM, SP, Z> EventProcessor<E, Z> for CentralizedEventManager<EM, SP>
+where
+    EM: EventProcessor<E, Z> + EventFirer + HasEventManagerId,
+    SP: ShMemProvider,
+    E: HasObservers<State = Self::State> + Executor<Self, Z>,
+    for<'a> E::Observers: Deserialize<'a>,
+    Z: EvaluatorObservers<E::Observers, State = Self::State>
+        + ExecutionProcessor<E::Observers, State = Self::State>,
+{
+    fn process(
+        &mut self,
+        fuzzer: &mut Z,
+        state: &mut Self::State,
+        executor: &mut E,
+    ) -> Result<usize, Error> {
+        if self.receivers_from_secondary.is_some() {
+            // main node
+            let mut receivers = self.receivers_from_secondary.take().unwrap();
+            // TODO in case of error, this is discarded, that is a bug ATM
+
+            for (idx, receiver) in receivers.iter_mut().enumerate() {
+                while let Some((_client_id, tag, _flags, msg)) = receiver.recv_buf_with_flags()? {
+                    assert!(
+                        tag == _LLMP_TAG_TO_MAIN,
+                        "Only the TO_MAIN parcel should have arrived in the main node!"
+                    );
+
+                    // TODO handle compression
+                    let event: Event<<Self::State as UsesInput>::Input> =
+                        postcard::from_bytes(msg)?;
+                    match event {
+                        Event::NewTestcase {
+                            input,
+                            client_config,
+                            exit_kind,
+                            corpus_size,
+                            observers_buf,
+                            time,
+                            executions,
+                            forward_id,
+                        } => {
+                            log::info!(
+                                "Received new Testcase to evaluate from secondary node {idx:?}"
+                            );
+
+                            // TODO check the config and use the serialized observers
+
+                            let res = fuzzer.evaluate_input_with_observers::<E, Self>(
+                                state,
+                                executor,
+                                self,
+                                input.clone(),
+                                false,
+                            )?;
+                            if let Some(item) = res.1 {
+                                log::info!("Added received Testcase as item #{item}");
+
+                                self.inner.fire(
+                                    state,
+                                    Event::NewTestcase {
+                                        input,
+                                        observers_buf,
+                                        exit_kind,
+                                        corpus_size,
+                                        client_config,
+                                        time,
+                                        executions,
+                                        forward_id,
+                                    },
+                                )?;
+                            }
+                        }
+                        _ => panic!(
+                            "Only the NewTestcase event should have arrived to the main node!"
+                        ),
+                    };
+                }
+            }
+
+            self.receivers_from_secondary = Some(receivers);
+
+            Ok(0) // TODO is 0 ok?
+        } else {
+            // The main node does not process incoming events from the broker ATM
+            self.inner.process(fuzzer, state, executor)
+        }
+    }
+}
+
+impl<E, EM, SP, Z> EventManager<E, Z> for CentralizedEventManager<EM, SP>
+where
+    EM: EventManager<E, Z>,
+    EM::State: HasClientPerfMonitor + HasExecutions + HasMetadata,
+    SP: ShMemProvider,
+    E: HasObservers<State = Self::State> + Executor<Self, Z>,
+    for<'a> E::Observers: Deserialize<'a>,
+    Z: EvaluatorObservers<E::Observers, State = Self::State>
+        + ExecutionProcessor<E::Observers, State = Self::State>,
+{
+}
+
+impl<EM, SP> HasCustomBufHandlers for CentralizedEventManager<EM, SP>
+where
+    EM: HasCustomBufHandlers,
+    SP: ShMemProvider,
+{
+    /// Adds a custom buffer handler that will run for each incoming `CustomBuf` event.
+    fn add_custom_buf_handler(
+        &mut self,
+        handler: Box<
+            dyn FnMut(&mut Self::State, &String, &[u8]) -> Result<CustomBufEventResult, Error>,
+        >,
+    ) {
+        self.inner.add_custom_buf_handler(handler);
+    }
+}
+
+impl<EM, SP> ProgressReporter for CentralizedEventManager<EM, SP>
+where
+    EM: ProgressReporter + HasEventManagerId,
+    EM::State: HasClientPerfMonitor + HasMetadata + HasExecutions,
+    SP: ShMemProvider,
+{
+}
+
+impl<EM, SP> HasEventManagerId for CentralizedEventManager<EM, SP>
+where
+    EM: HasEventManagerId + UsesState,
+    SP: ShMemProvider,
+{
+    fn mgr_id(&self) -> EventManagerId {
+        self.inner.mgr_id()
+    }
+}
+
+impl<EM, SP> CentralizedEventManager<EM, SP>
+where
+    EM: UsesState,
+    SP: ShMemProvider,
+{
+    /// Creates a new [`CentralizedEventManager`].
+    pub fn new_main(inner: EM, receivers_from_secondary: Vec<LlmpReceiver<SP>>) -> Self {
+        Self {
+            inner,
+            sender_to_main: None,
+            receivers_from_secondary: Some(receivers_from_secondary),
+        }
+    }
+
+    /// Creates a new [`CentralizedEventManager`].
+    pub fn new_secondary(inner: EM, sender_to_main: LlmpSender<SP>) -> Self {
+        Self {
+            inner,
+            sender_to_main: Some(sender_to_main),
+            receivers_from_secondary: None,
+        }
+    }
+}

libafl/src/events/llmp.rs

@@ -228,11 +228,17 @@ where
                 observers_buf: _,
                 time,
                 executions,
+                forward_id,
             } => {
-                let client = monitor.client_stats_mut_for(client_id);
+                let id = if let Some(id) = *forward_id {
+                    id
+                } else {
+                    client_id
+                };
+                let client = monitor.client_stats_mut_for(id);
                 client.update_corpus_size(*corpus_size as u64);
                 client.update_executions(*executions as u64, *time);
-                monitor.display(event.name().to_string(), client_id);
+                monitor.display(event.name().to_string(), id);
                 Ok(BrokerEventResult::Forward)
             }
             Event::UpdateExecStats {
@@ -458,8 +464,9 @@ where
                 observers_buf,
                 time: _,
                 executions: _,
+                forward_id,
             } => {
-                log::info!("Received new Testcase from {client_id:?} ({client_config:?})");
+                log::info!("Received new Testcase from {client_id:?} ({client_config:?}, forward {forward_id:?})");
 
                 let _res = if client_config.match_with(&self.configuration)
                     && observers_buf.is_some()
@@ -1284,21 +1291,22 @@ where
                 observers_buf: _, // Useless as we are converting between types
                 time: _,
                 executions: _,
+                forward_id,
             } => {
-                log::info!("Received new Testcase to convert from {_client_id:?}");
+                log::info!("Received new Testcase to convert from {_client_id:?} (forward {forward_id:?}, forward {forward_id:?})");
 
                 let Some(converter) = self.converter_back.as_mut() else {
                     return Ok(());
                 };
 
-                let _res = fuzzer.evaluate_input_with_observers::<E, EM>(
+                let res = fuzzer.evaluate_input_with_observers::<E, EM>(
                     state,
                     executor,
                     manager,
                     converter.convert(input)?,
                     false,
                 )?;
-                if let Some(item) = _res.1 {
+                if let Some(item) = res.1 {
                     log::info!("Added received Testcase as item #{item}");
                 }
                 Ok(())
@@ -1404,6 +1412,7 @@ where
                 observers_buf,
                 time,
                 executions,
+                forward_id,
             } => Event::NewTestcase {
                 input: self.converter.as_mut().unwrap().convert(input)?,
                 client_config,
@@ -1412,6 +1421,7 @@ where
                 observers_buf,
                 time,
                 executions,
+                forward_id,
             },
             Event::CustomBuf { buf, tag } => Event::CustomBuf { buf, tag },
             _ => {
@@ -1456,6 +1466,7 @@ where
                 observers_buf,
                 time,
                 executions,
+                forward_id,
             } => Event::NewTestcase {
                 input: self.converter.as_mut().unwrap().convert(input)?,
                 client_config,
@@ -1464,6 +1475,7 @@ where
                 observers_buf,
                 time,
                 executions,
+                forward_id,
             },
             Event::CustomBuf { buf, tag } => Event::CustomBuf { buf, tag },
             _ => {

libafl/src/events/mod.rs

@@ -2,6 +2,8 @@
 
 pub mod simple;
 pub use simple::*;
+pub mod centralized;
+pub use centralized::*;
 pub mod llmp;
 use alloc::{boxed::Box, string::String, vec::Vec};
 #[cfg(all(unix, feature = "std"))]
@@ -24,7 +26,7 @@ use crate::bolts::os::unix_signals::{siginfo_t, ucontext_t, Handler, Signal};
 #[cfg(all(unix, feature = "std"))]
 use crate::bolts::{shmem::ShMemProvider, staterestore::StateRestorer};
 use crate::{
-    bolts::current_time,
+    bolts::{current_time, ClientId},
     executors::ExitKind,
     inputs::Input,
     monitors::UserStats,
@@ -286,6 +288,8 @@ where
         time: Duration,
         /// The executions of this client
         executions: usize,
+        /// The original sender if, if forwarded
+        forward_id: Option<ClientId>,
     },
     /// New stats event to monitor.
     UpdateExecStats {
@@ -360,6 +364,7 @@ where
                 observers_buf: _,
                 time: _,
                 executions: _,
+                forward_id: _,
             } => "Testcase",
             Event::UpdateExecStats {
                 time: _,
@@ -669,6 +674,7 @@ mod tests {
             client_config: EventConfig::AlwaysUnique,
             time: current_time(),
             executions: 0,
+            forward_id: None,
         };
 
         let serialized = postcard::to_allocvec(&e).unwrap();
@@ -683,6 +689,7 @@ mod tests {
                 client_config: _,
                 time: _,
                 executions: _,
+                forward_id: _,
             } => {
                 let o: tuple_list_type!(StdMapObserver::<u32, false>) =
                     postcard::from_bytes(observers_buf.as_ref().unwrap()).unwrap();

libafl/src/events/simple.rs

@@ -212,6 +212,7 @@ where
                 observers_buf: _,
                 time,
                 executions,
+                forward_id: _,
             } => {
                 monitor
                     .client_stats_mut_for(ClientId(0))

libafl/src/fuzzer/mod.rs

@@ -406,6 +406,7 @@ where
                             client_config: manager.configuration(),
                             time: current_time(),
                             executions: *state.executions(),
+                            forward_id: None,
                         },
                     )?;
                 }
@@ -534,6 +535,7 @@ where
                 client_config: manager.configuration(),
                 time: current_time(),
                 executions: *state.executions(),
+                forward_id: None,
             },
         )?;
         Ok(idx)

libafl/src/stages/sync.rs

@@ -280,6 +280,7 @@ where
                         client_config: EventConfig::AlwaysUnique,
                         time: current_time(),
                         executions: 0,
+                        forward_id: None,
                     },
                 )?;