SyncFromBrokerStage to sync from a broker with a different Input type (#997)

* ConverterLlmpEventManager

* docs

* SyncFromBrokerStage

* fix

* separate InputConverter

* nautilus

* nautilus_sync

* send testcases

* upd nautilus

* meta

* fix

* clippy

* fix

* Update build_and_test.yml

* fix

* fix

* Use find_libpython

* ci

* upd qemu

.github/workflows/build_and_test.yml

@@ -173,7 +173,7 @@ jobs:
       run: rustup toolchain install nightly --component rustfmt --component clippy --allow-downgrade
     - name: Add no_std toolchain
       run: rustup toolchain install nightly-x86_64-unknown-linux-gnu ; rustup component add rust-src --toolchain nightly-x86_64-unknown-linux-gnu
-    - name: Install python
+    - name: Install python (macOS)
       # Removing macOS things already installed in CI against failed linking
       if: runner.os == 'macOS'
       run: rm /usr/local/bin/2to3* /usr/local/bin/idle3* /usr/local/bin/pydoc3* /usr/local/bin/python3*; brew install --force-bottle --overwrite python
@@ -184,7 +184,7 @@ jobs:
         # update bash for macos to support `declare -A` command`
         macos: llvm libpng nasm coreutils z3 bash wget
     - name: pip install
-      run: python3 -m pip install msgpack jinja2
+      run: python3 -m pip install msgpack jinja2 find_libpython
     # Note that nproc needs to have coreutils installed on macOS, so the order of CI commands matters.
     - name: enable mult-thread for `make`
       run: export MAKEFLAGS="-j$(expr $(nproc) \+ 1)"

fuzzers/nautilus_sync/.gitignore

@@ -0,0 +1,3 @@
+fuzzer_libpng_nautilus
+grammar.json
+libpng-*

fuzzers/nautilus_sync/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "nautilus_sync"
+version = "0.9.0"
+authors = ["Andrea Fioraldi <andreafioraldi@gmail.com>", "Dominik Maier <domenukk@gmail.com>"]
+edition = "2021"
+
+[features]
+default = ["std"]
+std = []
+
+[profile.dev]
+panic = "abort"
+debug = true
+
+[profile.release]
+panic = "abort"
+lto = true
+codegen-units = 1
+opt-level = 3
+debug = true
+
+[dependencies]
+libafl = { path = "../../libafl/", features = ["default", "nautilus"] }
+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 }
+
+[lib]
+name = "nautilus_sync"
+crate-type = ["staticlib"]

fuzzers/nautilus_sync/Makefile.toml

@@ -0,0 +1,125 @@
+# Variables
+[env]
+FUZZER_NAME='fuzzer_libpng_nautilus'
+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}"
+cp ../baby_fuzzer_nautilus/grammar.json .
+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}/../libfuzzer_libpng/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
+'''
+dependencies = [ "fuzzer" ]
+
+[tasks.run_unix_sync]
+script_runner = "@shell"
+script='''
+./${FUZZER_NAME} --cores 0 -b 1337
+'''
+dependencies = [ "fuzzer" ]
+
+# Test
+[tasks.test]
+linux_alias = "test_unix"
+mac_alias = "test_unix"
+windows_alias = "unsupported"
+
+[tasks.test_unix]
+script_runner = "@shell"
+script='''
+rm -rf libafl_unix_shmem_server || true
+timeout 11s ./${FUZZER_NAME} --cores 0 2>/dev/null &
+'''
+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/nautilus_sync/rust-toolchain

@@ -0,0 +1 @@
+nightly

fuzzers/nautilus_sync/src/bin/libafl_cc.rs

@@ -0,0 +1,59 @@
+use std::{env, process::Command, str};
+
+use libafl_cc::{ClangWrapper, CompilerWrapper};
+
+fn find_libpython() -> Result<String, String> {
+    match Command::new("python3")
+        .args(&["-m", "find_libpython"])
+        .output()
+    {
+        Ok(output) => {
+            let shared_obj = str::from_utf8(&output.stdout).unwrap_or_default().trim();
+            if shared_obj.is_empty() {
+                return Err("Empty return from python3 -m find_libpython".to_string());
+            }
+            Ok(shared_obj.to_owned())
+        }
+        Err(err) => Err(format!(
+            "Could not execute python3 -m find_libpython: {err:?}"
+        )),
+    }
+}
+
+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 libpython = find_libpython().expect("Failed to find libpython");
+
+        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, "nautilus_sync")
+            .add_arg("-fsanitize-coverage=trace-pc-guard")
+            // needed by Nautilus
+            .add_link_arg(libpython)
+            .add_link_arg("-lutil")
+            .run()
+            .expect("Failed to run the wrapped compiler")
+        {
+            std::process::exit(code);
+        }
+    } else {
+        panic!("LibAFL CC: No Arguments given");
+    }
+}

fuzzers/nautilus_sync/src/bin/libafl_cxx.rs

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

fuzzers/nautilus_sync/src/lib.rs

@@ -0,0 +1,262 @@
+use mimalloc::MiMalloc;
+#[global_allocator]
+static GLOBAL: MiMalloc = MiMalloc;
+
+#[cfg(windows)]
+use std::ptr::write_volatile;
+use std::{env, net::SocketAddr, path::PathBuf, time::Duration};
+
+use clap::{self, Parser};
+use libafl::{
+    bolts::{
+        core_affinity::Cores,
+        current_nanos,
+        launcher::Launcher,
+        rands::StdRand,
+        shmem::{ShMemProvider, StdShMemProvider},
+        tuples::tuple_list,
+    },
+    corpus::{InMemoryCorpus, OnDiskCorpus},
+    events::{llmp::LlmpEventConverter, EventConfig},
+    executors::{inprocess::InProcessExecutor, ExitKind},
+    feedback_or,
+    feedbacks::{CrashFeedback, MaxMapFeedback, NautilusChunksMetadata, NautilusFeedback},
+    fuzzer::{Fuzzer, StdFuzzer},
+    generators::{NautilusContext, NautilusGenerator},
+    inputs::{NautilusInput, NautilusToBytesInputConverter},
+    monitors::SimpleMonitor,
+    mutators::{
+        NautilusRandomMutator, NautilusRecursionMutator, NautilusSpliceMutator, StdScheduledMutator,
+    },
+    none_input_converter,
+    schedulers::QueueScheduler,
+    stages::{mutational::StdMutationalStage, sync::SyncFromBrokerStage},
+    state::{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 = "1338"
+    )]
+    broker_port: u16,
+
+    #[arg(
+        short = 'b',
+        long,
+        help = "Specify a BytesInput broker TCP port",
+        name = "BYTESPORT"
+    )]
+    bytes_broker_port: Option<u16>,
+
+    #[arg(short = 'a', long, help = "Specify a remote broker", name = "REMOTE")]
+    remote_broker_addr: Option<SocketAddr>,
+
+    #[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,
+}
+
+/// The main fn, `no_mangle` as it is a C symbol
+#[no_mangle]
+pub fn libafl_main() {
+    // 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");
+
+    // The Monitor trait define how the fuzzer stats are reported to the user
+    let monitor = SimpleMonitor::new(|s| println!("{s}"));
+
+    let context = NautilusContext::from_file(15, "grammar.json");
+
+    let mut event_converter = opt.bytes_broker_port.map(|port| {
+        LlmpEventConverter::new_on_port(
+            shmem_provider.clone(),
+            port,
+            Some(NautilusToBytesInputConverter::new(&context)),
+            none_input_converter!(),
+        )
+        .unwrap()
+    });
+
+    let mut run_client = |state: Option<_>, mut mgr, _core_id| {
+        let mut bytes = vec![];
+
+        // The closure that we want to fuzz
+        let mut harness = |input: &NautilusInput| {
+            input.unparse(&context, &mut bytes);
+            libfuzzer_test_one_input(&bytes);
+            ExitKind::Ok
+        };
+
+        // Create an observation channel using the coverage map
+        let observer = unsafe { std_edges_map_observer("edges") };
+
+        // Feedback to rate the interestingness of an input
+        let mut feedback = feedback_or!(
+            MaxMapFeedback::new(&observer),
+            NautilusFeedback::new(&context)
+        );
+
+        // A feedback to choose if an input is a solution or not
+        let mut objective = CrashFeedback::new();
+
+        // 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(PathBuf::from("./crashes")).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()
+        });
+
+        if state.metadata().get::<NautilusChunksMetadata>().is_none() {
+            state.add_metadata(NautilusChunksMetadata::new("/tmp/".into()));
+        }
+
+        // A queue policy to get testcasess from the corpus
+        let scheduler = QueueScheduler::new();
+
+        // A fuzzer with feedbacks and a corpus scheduler
+        let mut fuzzer = StdFuzzer::new(scheduler, feedback, objective);
+
+        // Create the executor for an in-process function with just one observer
+        let mut executor = InProcessExecutor::new(
+            &mut harness,
+            tuple_list!(observer),
+            &mut fuzzer,
+            &mut state,
+            &mut mgr,
+        )
+        .expect("Failed to create the Executor");
+
+        // 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")
+        }
+
+        let mut generator = NautilusGenerator::new(&context);
+
+        // Generate 8 initial inputs
+        state
+            .generate_initial_inputs_forced(&mut fuzzer, &mut executor, &mut generator, &mut mgr, 8)
+            .expect("Failed to generate the initial corpus");
+
+        // Setup a mutational stage with a basic bytes mutator
+        let mutator = StdScheduledMutator::with_max_stack_pow(
+            tuple_list!(
+                NautilusRandomMutator::new(&context),
+                NautilusRandomMutator::new(&context),
+                NautilusRandomMutator::new(&context),
+                NautilusRandomMutator::new(&context),
+                NautilusRandomMutator::new(&context),
+                NautilusRandomMutator::new(&context),
+                NautilusRecursionMutator::new(&context),
+                NautilusSpliceMutator::new(&context),
+                NautilusSpliceMutator::new(&context),
+                NautilusSpliceMutator::new(&context),
+            ),
+            2,
+        );
+
+        if let Some(conv) = event_converter.take() {
+            let mut stages = tuple_list!(
+                StdMutationalStage::new(mutator),
+                SyncFromBrokerStage::new(conv)
+            );
+
+            fuzzer
+                .fuzz_loop(&mut stages, &mut executor, &mut state, &mut mgr)
+                .expect("Error in the fuzzing loop");
+        } else {
+            let mut stages = tuple_list!(StdMutationalStage::new(mutator));
+
+            fuzzer
+                .fuzz_loop(&mut stages, &mut executor, &mut state, &mut mgr)
+                .expect("Error in the fuzzing loop");
+        }
+
+        Ok(())
+    };
+
+    match Launcher::builder()
+        .shmem_provider(shmem_provider)
+        .configuration(EventConfig::from_name("nautilus"))
+        .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/Cargo.toml

@@ -100,7 +100,7 @@ concat-idents = { version = "1.1.3", optional = true }
 
 # AGPL
 # !!! this create requires nightly
-grammartec = { version = "0.2", optional = true }
+grammartec = { version = "0.3", optional = true }
 
 [target.'cfg(unix)'.dependencies]
 libc = "0.2" # For (*nix) libc

libafl/src/events/llmp.rs

@@ -47,7 +47,7 @@ use crate::{
     },
     executors::{Executor, HasObservers},
     fuzzer::{EvaluatorObservers, ExecutionProcessor},
-    inputs::{Input, UsesInput},
+    inputs::{Input, InputConverter, UsesInput},
     monitors::Monitor,
     state::{HasClientPerfMonitor, HasExecutions, HasMetadata, UsesState},
     Error,
@@ -1014,6 +1014,352 @@ where
     }
 }
 
+/// A manager-like llmp client that converts between input types
+pub struct LlmpEventConverter<IC, ICB, DI, S, SP>
+where
+    S: UsesInput,
+    SP: ShMemProvider + 'static,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    llmp: LlmpClient<SP>,
+    /// The custom buf handler
+    custom_buf_handlers: Vec<Box<CustomBufHandlerFn<S>>>,
+    #[cfg(feature = "llmp_compression")]
+    compressor: GzipCompressor,
+    converter: Option<IC>,
+    converter_back: Option<ICB>,
+    phantom: PhantomData<S>,
+}
+
+impl<IC, ICB, DI, S, SP> core::fmt::Debug for LlmpEventConverter<IC, ICB, DI, S, SP>
+where
+    SP: ShMemProvider + 'static,
+    S: UsesInput,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        let mut debug_struct = f.debug_struct("LlmpEventConverter");
+        let debug = debug_struct.field("llmp", &self.llmp);
+        //.field("custom_buf_handlers", &self.custom_buf_handlers)
+        #[cfg(feature = "llmp_compression")]
+        let debug = debug.field("compressor", &self.compressor);
+        debug
+            .field("converter", &self.converter)
+            .field("converter_back", &self.converter_back)
+            .field("phantom", &self.phantom)
+            .finish_non_exhaustive()
+    }
+}
+
+impl<IC, ICB, DI, S, SP> LlmpEventConverter<IC, ICB, DI, S, SP>
+where
+    S: UsesInput + HasExecutions + HasClientPerfMonitor,
+    SP: ShMemProvider + 'static,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    /// Create a client from a raw llmp client
+    pub fn new(
+        llmp: LlmpClient<SP>,
+        converter: Option<IC>,
+        converter_back: Option<ICB>,
+    ) -> Result<Self, Error> {
+        Ok(Self {
+            llmp,
+            #[cfg(feature = "llmp_compression")]
+            compressor: GzipCompressor::new(COMPRESS_THRESHOLD),
+            converter,
+            converter_back,
+            phantom: PhantomData,
+            custom_buf_handlers: vec![],
+        })
+    }
+
+    /// Create a client from port and the input converters
+    #[cfg(feature = "std")]
+    pub fn new_on_port(
+        shmem_provider: SP,
+        port: u16,
+        converter: Option<IC>,
+        converter_back: Option<ICB>,
+    ) -> Result<Self, Error> {
+        Ok(Self {
+            llmp: LlmpClient::create_attach_to_tcp(shmem_provider, port)?,
+            #[cfg(feature = "llmp_compression")]
+            compressor: GzipCompressor::new(COMPRESS_THRESHOLD),
+            converter,
+            converter_back,
+            phantom: PhantomData,
+            custom_buf_handlers: vec![],
+        })
+    }
+
+    /// If a client respawns, it may reuse the existing connection, previously stored by [`LlmpClient::to_env()`].
+    #[cfg(feature = "std")]
+    pub fn existing_client_from_env(
+        shmem_provider: SP,
+        env_name: &str,
+        converter: Option<IC>,
+        converter_back: Option<ICB>,
+    ) -> Result<Self, Error> {
+        Ok(Self {
+            llmp: LlmpClient::on_existing_from_env(shmem_provider, env_name)?,
+            #[cfg(feature = "llmp_compression")]
+            compressor: GzipCompressor::new(COMPRESS_THRESHOLD),
+            phantom: PhantomData,
+            converter,
+            converter_back,
+            custom_buf_handlers: vec![],
+        })
+    }
+
+    // TODO other new_* routines
+
+    /// Check if it can convert the input
+    pub fn can_convert(&self) -> bool {
+        self.converter.is_some()
+    }
+
+    /// Check if it can convert the input back
+    pub fn can_convert_back(&self) -> bool {
+        self.converter_back.is_some()
+    }
+
+    /// Describe the client event mgr's llmp parts in a restorable fashion
+    pub fn describe(&self) -> Result<LlmpClientDescription, Error> {
+        self.llmp.describe()
+    }
+
+    /// Write the config for a client [`EventManager`] to env vars, a new client can reattach using [`LlmpEventConverter::existing_client_from_env()`].
+    #[cfg(feature = "std")]
+    pub fn to_env(&self, env_name: &str) {
+        self.llmp.to_env(env_name).unwrap();
+    }
+
+    // Handle arriving events in the client
+    fn handle_in_client<E, EM, Z>(
+        &mut self,
+        fuzzer: &mut Z,
+        executor: &mut E,
+        state: &mut S,
+        manager: &mut EM,
+        _client_id: u32,
+        event: Event<DI>,
+    ) -> Result<(), Error>
+    where
+        E: Executor<EM, Z> + HasObservers<State = S>,
+        EM: UsesState<State = S> + EventFirer,
+        for<'a> E::Observers: Deserialize<'a>,
+        Z: ExecutionProcessor<E::Observers, State = S> + EvaluatorObservers<E::Observers>,
+    {
+        match event {
+            Event::NewTestcase {
+                input,
+                client_config: _,
+                exit_kind: _,
+                corpus_size: _,
+                observers_buf: _, // Useless as we are converting between types
+                time: _,
+                executions: _,
+            } => {
+                #[cfg(feature = "std")]
+                println!("Received new Testcase to convert from {_client_id}");
+
+                let Some(converter) = self.converter_back.as_mut() else {
+                    return Ok(());
+                };
+
+                let _res = fuzzer.evaluate_input_with_observers::<E, EM>(
+                    state,
+                    executor,
+                    manager,
+                    converter.convert(input)?,
+                    false,
+                )?;
+                #[cfg(feature = "std")]
+                if let Some(item) = _res.1 {
+                    println!("Added received Testcase as item #{item}");
+                }
+                Ok(())
+            }
+            Event::CustomBuf { tag, buf } => {
+                for handler in &mut self.custom_buf_handlers {
+                    if handler(state, &tag, &buf)? == CustomBufEventResult::Handled {
+                        break;
+                    }
+                }
+                Ok(())
+            }
+            _ => Err(Error::unknown(format!(
+                "Received illegal message that message should not have arrived: {:?}.",
+                event.name()
+            ))),
+        }
+    }
+
+    /// Handle arriving events in the client
+    #[allow(clippy::unused_self)]
+    pub fn process<E, EM, Z>(
+        &mut self,
+        fuzzer: &mut Z,
+        state: &mut S,
+        executor: &mut E,
+        manager: &mut EM,
+    ) -> Result<usize, Error>
+    where
+        E: Executor<EM, Z> + HasObservers<State = S>,
+        EM: UsesState<State = S> + EventFirer,
+        for<'a> E::Observers: Deserialize<'a>,
+        Z: ExecutionProcessor<E::Observers, State = S> + EvaluatorObservers<E::Observers>,
+    {
+        // TODO: Get around local event copy by moving handle_in_client
+        let self_id = self.llmp.sender.id;
+        let mut count = 0;
+        while let Some((client_id, tag, _flags, msg)) = self.llmp.recv_buf_with_flags()? {
+            assert!(
+                tag != _LLMP_TAG_EVENT_TO_BROKER,
+                "EVENT_TO_BROKER parcel should not have arrived in the client!"
+            );
+
+            if client_id == self_id {
+                continue;
+            }
+            #[cfg(not(feature = "llmp_compression"))]
+            let event_bytes = msg;
+            #[cfg(feature = "llmp_compression")]
+            let compressed;
+            #[cfg(feature = "llmp_compression")]
+            let event_bytes = if _flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
+                compressed = self.compressor.decompress(msg)?;
+                &compressed
+            } else {
+                msg
+            };
+
+            let event: Event<DI> = postcard::from_bytes(event_bytes)?;
+            self.handle_in_client(fuzzer, executor, state, manager, client_id, event)?;
+            count += 1;
+        }
+        Ok(count)
+    }
+}
+
+impl<IC, ICB, DI, S, SP> UsesState for LlmpEventConverter<IC, ICB, DI, S, SP>
+where
+    S: UsesInput,
+    SP: ShMemProvider,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    type State = S;
+}
+
+impl<IC, ICB, DI, S, SP> EventFirer for LlmpEventConverter<IC, ICB, DI, S, SP>
+where
+    S: UsesInput,
+    SP: ShMemProvider,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    #[cfg(feature = "llmp_compression")]
+    fn fire(
+        &mut self,
+        _state: &mut Self::State,
+        event: Event<<Self::State as UsesInput>::Input>,
+    ) -> Result<(), Error> {
+        if self.converter.is_none() {
+            return Ok(());
+        }
+
+        // Filter out non interestign events and convert `NewTestcase`
+        let converted_event = match event {
+            Event::NewTestcase {
+                input,
+                client_config,
+                exit_kind,
+                corpus_size,
+                observers_buf,
+                time,
+                executions,
+            } => Event::NewTestcase {
+                input: self.converter.as_mut().unwrap().convert(input)?,
+                client_config,
+                exit_kind,
+                corpus_size,
+                observers_buf,
+                time,
+                executions,
+            },
+            Event::CustomBuf { buf, tag } => Event::CustomBuf { buf, tag },
+            _ => {
+                return Ok(());
+            }
+        };
+        let serialized = postcard::to_allocvec(&converted_event)?;
+        let flags: Flags = LLMP_FLAG_INITIALIZED;
+
+        match self.compressor.compress(&serialized)? {
+            Some(comp_buf) => {
+                self.llmp.send_buf_with_flags(
+                    LLMP_TAG_EVENT_TO_BOTH,
+                    flags | LLMP_FLAG_COMPRESSED,
+                    &comp_buf,
+                )?;
+            }
+            None => {
+                self.llmp.send_buf(LLMP_TAG_EVENT_TO_BOTH, &serialized)?;
+            }
+        }
+        Ok(())
+    }
+
+    #[cfg(not(feature = "llmp_compression"))]
+    fn fire(
+        &mut self,
+        _state: &mut Self::State,
+        event: Event<<Self::State as UsesInput>::Input>,
+    ) -> Result<(), Error> {
+        if self.converter.is_none() {
+            return Ok(());
+        }
+
+        // Filter out non interestign events and convert `NewTestcase`
+        let converted_event = match event {
+            Event::NewTestcase {
+                input,
+                client_config,
+                exit_kind,
+                corpus_size,
+                observers_buf,
+                time,
+                executions,
+            } => Event::NewTestcase {
+                input: self.converter.as_mut().unwrap().convert(input)?,
+                client_config,
+                exit_kind,
+                corpus_size,
+                observers_buf,
+                time,
+                executions,
+            },
+            Event::CustomBuf { buf, tag } => Event::CustomBuf { buf, tag },
+            _ => {
+                return Ok(());
+            }
+        };
+        let serialized = postcard::to_allocvec(&converted_event)?;
+        self.llmp.send_buf(LLMP_TAG_EVENT_TO_BOTH, &serialized)?;
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 #[cfg(feature = "std")]
 mod tests {

libafl/src/inputs/mod.rs

@@ -15,10 +15,11 @@ pub use generalized::*;
 #[cfg(feature = "nautilus")]
 pub mod nautilus;
 use alloc::{
+    boxed::Box,
     string::{String, ToString},
     vec::Vec,
 };
-use core::{clone::Clone, fmt::Debug};
+use core::{clone::Clone, fmt::Debug, marker::PhantomData};
 #[cfg(feature = "std")]
 use std::{fs::File, hash::Hash, io::Read, path::Path};
 
@@ -79,6 +80,25 @@ pub trait Input: Clone + Serialize + serde::de::DeserializeOwned + Debug {
     fn wrapped_as_testcase(&mut self) {}
 }
 
+/// Convert between two input types with a state
+pub trait InputConverter: Debug {
+    /// Source type
+    type From: Input;
+    /// Destination type
+    type To: Input;
+
+    /// Convert the src type to the dest
+    fn convert(&mut self, input: Self::From) -> Result<Self::To, Error>;
+}
+
+/// `None` type to satisfy the type infearence in an `Option`
+#[macro_export]
+macro_rules! none_input_converter {
+    () => {
+        None::<$crate::inputs::ClosureInputConverter<_, _>>
+    };
+}
+
 /// An input for tests, mainly. There is no real use much else.
 #[derive(Copy, Clone, Serialize, Deserialize, Debug, Hash)]
 pub struct NopInput {}
@@ -116,3 +136,74 @@ pub trait UsesInput {
     /// Type which will be used throughout this state.
     type Input: Input;
 }
+
+#[derive(Debug)]
+/// Basic `InputConverter` with just one type that is not converting
+pub struct NopInputConverter<I> {
+    phantom: PhantomData<I>,
+}
+
+impl<I> Default for NopInputConverter<I> {
+    fn default() -> Self {
+        Self {
+            phantom: PhantomData,
+        }
+    }
+}
+
+impl<I> InputConverter for NopInputConverter<I>
+where
+    I: Input,
+{
+    type From = I;
+    type To = I;
+
+    fn convert(&mut self, input: Self::From) -> Result<Self::To, Error> {
+        Ok(input)
+    }
+}
+
+/// `InputConverter` that uses a closure to convert
+pub struct ClosureInputConverter<F, T>
+where
+    F: Input,
+    T: Input,
+{
+    convert_cb: Box<dyn FnMut(F) -> Result<T, Error>>,
+}
+
+impl<F, T> Debug for ClosureInputConverter<F, T>
+where
+    F: Input,
+    T: Input,
+{
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("ClosureInputConverter")
+            .finish_non_exhaustive()
+    }
+}
+
+impl<F, T> ClosureInputConverter<F, T>
+where
+    F: Input,
+    T: Input,
+{
+    /// Create a new converter using two closures, use None to forbid the conversion or the conversion back
+    #[must_use]
+    pub fn new(convert_cb: Box<dyn FnMut(F) -> Result<T, Error>>) -> Self {
+        Self { convert_cb }
+    }
+}
+
+impl<F, T> InputConverter for ClosureInputConverter<F, T>
+where
+    F: Input,
+    T: Input,
+{
+    type From = F;
+    type To = T;
+
+    fn convert(&mut self, input: Self::From) -> Result<Self::To, Error> {
+        (self.convert_cb)(input)
+    }
+}

libafl/src/inputs/nautilus.rs

@@ -15,7 +15,12 @@ use grammartec::{
 };
 use serde::{Deserialize, Serialize};
 
-use crate::{bolts::HasLen, generators::nautilus::NautilusContext, inputs::Input};
+use crate::{
+    bolts::HasLen,
+    generators::nautilus::NautilusContext,
+    inputs::{BytesInput, Input, InputConverter},
+    Error,
+};
 
 /// An [`Input`] implementation for `Nautilus` grammar.
 #[derive(Serialize, Deserialize, Clone, Debug)]
@@ -104,3 +109,28 @@ impl Hash for NautilusInput {
         self.tree().sizes.hash(state);
     }
 }
+
+/// `InputConverter` to convert from `NautilusInput` to `BytesInput`
+#[derive(Debug)]
+pub struct NautilusToBytesInputConverter<'a> {
+    ctx: &'a NautilusContext,
+}
+
+impl<'a> NautilusToBytesInputConverter<'a> {
+    #[must_use]
+    /// Create a new `NautilusToBytesInputConverter` from a context
+    pub fn new(ctx: &'a NautilusContext) -> Self {
+        Self { ctx }
+    }
+}
+
+impl<'a> InputConverter for NautilusToBytesInputConverter<'a> {
+    type From = NautilusInput;
+    type To = BytesInput;
+
+    fn convert(&mut self, input: Self::From) -> Result<Self::To, Error> {
+        let mut bytes = vec![];
+        input.unparse(self.ctx, &mut bytes);
+        Ok(BytesInput::new(bytes))
+    }
+}

libafl/src/observers/map.rs

@@ -884,7 +884,7 @@ where
 {
     /// Creates a new [`MapObserver`]
     ///
-    /// # Safety
+    /// # Note
     /// Will get a pointer to the map and dereference it at any point in time.
     /// The map must not move in memory!
     #[must_use]

libafl/src/stages/sync.rs

@@ -10,11 +10,14 @@ use std::{
 use serde::{Deserialize, Serialize};
 
 use crate::{
-    corpus::CorpusId,
+    bolts::{current_time, shmem::ShMemProvider},
-    fuzzer::Evaluator,
+    corpus::{Corpus, CorpusId},
-    inputs::{Input, UsesInput},
+    events::{llmp::LlmpEventConverter, Event, EventConfig, EventFirer},
+    executors::{Executor, ExitKind, HasObservers},
+    fuzzer::{Evaluator, EvaluatorObservers, ExecutionProcessor},
+    inputs::{Input, InputConverter, UsesInput},
     stages::Stage,
-    state::{HasClientPerfMonitor, HasCorpus, HasMetadata, HasRand, UsesState},
+    state::{HasClientPerfMonitor, HasCorpus, HasExecutions, HasMetadata, HasRand, UsesState},
     Error,
 };
 
@@ -186,3 +189,129 @@ where
         }
     }
 }
+
+/// Metadata used to store information about the last sent testcase with `SyncFromBrokerStage`
+#[derive(Serialize, Deserialize, Debug)]
+pub struct SyncFromBrokerMetadata {
+    /// The `CorpusId` of the last sent testcase
+    pub last_id: Option<CorpusId>,
+}
+
+crate::impl_serdeany!(SyncFromBrokerMetadata);
+
+impl SyncFromBrokerMetadata {
+    /// Create a new [`struct@SyncFromBrokerMetadata`]
+    #[must_use]
+    pub fn new(last_id: Option<CorpusId>) -> Self {
+        Self { last_id }
+    }
+}
+
+/// A stage that loads testcases from disk to sync with other fuzzers such as AFL++
+#[derive(Debug)]
+pub struct SyncFromBrokerStage<IC, ICB, DI, S, SP>
+where
+    SP: ShMemProvider + 'static,
+    S: UsesInput,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    client: LlmpEventConverter<IC, ICB, DI, S, SP>,
+}
+
+impl<IC, ICB, DI, S, SP> UsesState for SyncFromBrokerStage<IC, ICB, DI, S, SP>
+where
+    SP: ShMemProvider + 'static,
+    S: UsesInput,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    type State = S;
+}
+
+impl<E, EM, IC, ICB, DI, S, SP, Z> Stage<E, EM, Z> for SyncFromBrokerStage<IC, ICB, DI, S, SP>
+where
+    EM: UsesState<State = S> + EventFirer,
+    S: UsesInput + HasClientPerfMonitor + HasExecutions + HasCorpus + HasRand + HasMetadata,
+    SP: ShMemProvider,
+    E: HasObservers<State = S> + Executor<EM, Z>,
+    for<'a> E::Observers: Deserialize<'a>,
+    Z: EvaluatorObservers<E::Observers, State = S> + ExecutionProcessor<E::Observers, State = S>,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    #[inline]
+    fn perform(
+        &mut self,
+        fuzzer: &mut Z,
+        executor: &mut E,
+        state: &mut Z::State,
+        manager: &mut EM,
+        _corpus_idx: CorpusId,
+    ) -> Result<(), Error> {
+        if self.client.can_convert() {
+            let last_id = state
+                .metadata()
+                .get::<SyncFromBrokerMetadata>()
+                .and_then(|m| m.last_id);
+
+            let mut cur_id =
+                last_id.map_or_else(|| state.corpus().first(), |id| state.corpus().next(id));
+
+            while let Some(id) = cur_id {
+                let input = state.corpus().get(id)?.borrow_mut().load_input()?.clone();
+
+                self.client.fire(
+                    state,
+                    Event::NewTestcase {
+                        input,
+                        observers_buf: None,
+                        exit_kind: ExitKind::Ok,
+                        corpus_size: 0, // TODO choose if sending 0 or the actual real value
+                        client_config: EventConfig::AlwaysUnique,
+                        time: current_time(),
+                        executions: 0,
+                    },
+                )?;
+
+                cur_id = state.corpus().next(id);
+            }
+
+            let last = state.corpus().last();
+            if last_id.is_none() {
+                state
+                    .metadata_mut()
+                    .insert(SyncFromBrokerMetadata::new(last));
+            } else {
+                state
+                    .metadata_mut()
+                    .get_mut::<SyncFromBrokerMetadata>()
+                    .unwrap()
+                    .last_id = last;
+            }
+        }
+
+        self.client.process(fuzzer, state, executor, manager)?;
+        #[cfg(feature = "introspection")]
+        state.introspection_monitor_mut().finish_stage();
+        Ok(())
+    }
+}
+
+impl<IC, ICB, DI, S, SP> SyncFromBrokerStage<IC, ICB, DI, S, SP>
+where
+    SP: ShMemProvider + 'static,
+    S: UsesInput,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    /// Creates a new [`SyncFromBrokerStage`]
+    #[must_use]
+    pub fn new(client: LlmpEventConverter<IC, ICB, DI, S, SP>) -> Self {
+        Self { client }
+    }
+}

libafl_qemu/libafl_qemu_build/src/build.rs

@@ -8,7 +8,7 @@ use which::which;
 
 const QEMU_URL: &str = "https://github.com/AFLplusplus/qemu-libafl-bridge";
 const QEMU_DIRNAME: &str = "qemu-libafl-bridge";
-const QEMU_REVISION: &str = "d15b4d47e4f4ee3e3311e57183c2b7c1a12456f2";
+const QEMU_REVISION: &str = "f6a2e732e8e225ebb8d1a9399561af7330af31b3";
 
 fn build_dep_check(tools: &[&str]) {
     for tool in tools {