Add optional feature to share objectives between nodes

libafl/Cargo.toml

@@ -142,6 +142,9 @@ unicode = ["libafl_bolts/alloc", "ahash/std", "serde/rc", "bitvec"]
 ## Enable multi-part input formats and mutators
 multipart_inputs = ["arrayvec", "rand_trait"]
 
+## Share objectives across nodes
+share_objectives = []
+
 #! ## LibAFL-Bolts Features
 
 ## Provide the `#[derive(SerdeAny)]` macro.

libafl/src/events/centralized.rs

@@ -287,14 +287,18 @@ where
         if !self.is_main {
             // secondary node
             let mut is_tc = false;
-            // Forward to main only if new tc or heartbeat
+            // Forward to main only if new tc, heartbeat, or optionally, a new objective
             let should_be_forwarded = match &mut event {
                 Event::NewTestcase { forward_id, .. } => {
                     *forward_id = Some(ClientId(self.inner.mgr_id().0 as u32));
                     is_tc = true;
                     true
                 }
                 Event::UpdateExecStats { .. } => true, // send it but this guy won't be handled. the only purpose is to keep this client alive else the broker thinks it is dead and will dc it
+
+                #[cfg(feature = "share_objectives")]
+                Event::Objective { .. } => true,
+
                 Event::Stop => true,
                 _ => false,
             };
@@ -679,6 +683,12 @@ where
                     log::debug!("[{}] {} was discarded...)", process::id(), event_name);
                 }
             }
+
+            #[cfg(feature = "share_objectives")]
+            Event::Objective { .. } => {
+                log::debug!("Received new Objective");
+            }
+
             Event::Stop => {
                 state.request_stop();
             }

libafl/src/events/llmp/mgr.rs

@@ -461,6 +461,12 @@ where
                     }
                 }
             }
+
+            #[cfg(feature = "share_objectives")]
+            Event::Objective { .. } => {
+                log::debug!("Received new Objective");
+            }
+
             Event::CustomBuf { tag, buf } => {
                 for handler in &mut self.custom_buf_handlers {
                     if handler(state, &tag, &buf)? == CustomBufEventResult::Handled {

libafl/src/events/llmp/mod.rs

@@ -24,6 +24,11 @@ use crate::{
     state::{HasCorpus, HasExecutions, NopState, State, Stoppable, UsesState},
     Error, HasMetadata,
 };
+#[cfg(feature = "share_objectives")]
+use crate::{
+    corpus::Testcase,
+    state::{HasCurrentTestcase, HasSolutions},
+};
 
 /// The llmp event manager
 pub mod mgr;
@@ -284,6 +289,7 @@ where
     }
 
     // Handle arriving events in the client
+    #[cfg(not(feature = "share_objectives"))]
     fn handle_in_client<E, EM, Z>(
         &mut self,
         fuzzer: &mut Z,
@@ -340,6 +346,150 @@ where
         }
     }
 
+    /// Handle arriving events in the client
+    #[cfg(not(feature = "share_objectives"))]
+    #[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, State = S> + HasObservers,
+        EM: UsesState<State = S> + EventFirer,
+        S::Corpus: Corpus<Input = S::Input>,
+        for<'a> E::Observers: Deserialize<'a>,
+        Z: ExecutionProcessor<EM, <S::Corpus as Corpus>::Input, E::Observers, S>
+            + EvaluatorObservers<E, EM, <S::Corpus as Corpus>::Input, S>,
+    {
+        // 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)?;
+            log::debug!("Processor received message {}", event.name_detailed());
+            self.handle_in_client(fuzzer, executor, state, manager, client_id, event)?;
+            count += 1;
+        }
+        Ok(count)
+    }
+}
+
+#[cfg(feature = "share_objectives")]
+impl<DI, IC, ICB, S, SP> LlmpEventConverter<DI, IC, ICB, S, SP>
+where
+    S: UsesInput
+        + HasExecutions
+        + HasSolutions
+        + HasMetadata
+        + Stoppable
+        + HasCurrentTestcase
+        + HasCorpus,
+    S::Solutions: Corpus<Input = S::Input>,
+    SP: ShMemProvider,
+    IC: InputConverter<From = S::Input, To = DI>,
+    ICB: InputConverter<From = DI, To = S::Input>,
+    DI: Input,
+{
+    // 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: ClientId,
+        event: Event<DI>,
+    ) -> Result<(), Error>
+    where
+        E: Executor<EM, Z, State = S> + HasObservers,
+        EM: UsesState<State = S> + EventFirer,
+        S::Corpus: Corpus<Input = S::Input>,
+        for<'a> E::Observers: Deserialize<'a>,
+        Z: ExecutionProcessor<EM, <S::Corpus as Corpus>::Input, E::Observers, S>
+            + EvaluatorObservers<E, EM, <S::Corpus as Corpus>::Input, S>,
+    {
+        match event {
+            Event::NewTestcase {
+                input, forward_id, ..
+            } => {
+                log::debug!("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(
+                    state,
+                    executor,
+                    manager,
+                    converter.convert(input)?,
+                    false,
+                )?;
+
+                if let Some(item) = res.1 {
+                    log::info!("Added received Testcase as item #{item}");
+                }
+                Ok(())
+            }
+            Event::Objective { input, .. } => {
+                log::debug!("Received new Objective");
+
+                let Some(converter) = self.converter_back.as_mut() else {
+                    return Ok(());
+                };
+
+                let converted_input = converter.convert(input)?;
+                let mut testcase = Testcase::from(converted_input);
+                testcase.set_parent_id_optional(*state.corpus().current());
+
+                if let Ok(mut tc) = state.current_testcase_mut() {
+                    tc.found_objective();
+                }
+
+                state.solutions_mut().add(testcase)?;
+                log::info!("Added received Objective to Corpus");
+
+                Ok(())
+            }
+            Event::CustomBuf { tag, buf } => {
+                for handler in &mut self.custom_buf_handlers {
+                    if handler(state, &tag, &buf)? == CustomBufEventResult::Handled {
+                        break;
+                    }
+                }
+                Ok(())
+            }
+            Event::Stop => 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>(

libafl/src/events/mod.rs

@@ -325,6 +325,9 @@ where
     },
     /// A new objective was found
     Objective {
+        /// Input of newly found Objective
+        #[cfg(feature = "share_objectives")]
+        input: I,
         /// Objective corpus size
         objective_size: usize,
         /// The time when this event was created

libafl/src/events/tcp.rs

@@ -645,6 +645,12 @@ where
                     log::info!("Added received Testcase as item #{item}");
                 }
             }
+
+            #[cfg(feature = "share_objectives")]
+            Event::Objective { .. } => {
+                log::info!("Received new Objective");
+            }
+
             Event::CustomBuf { tag, buf } => {
                 for handler in &mut self.custom_buf_handlers {
                     if handler(state, &tag, &buf)? == CustomBufEventResult::Handled {

libafl/src/executors/inprocess/mod.rs

@@ -476,6 +476,9 @@ pub fn run_observers_and_save_state<E, EM, OF, Z>(
             .fire(
                 state,
                 Event::Objective {
+                    #[cfg(feature = "share_objectives")]
+                    input: input.clone(),
+
                     objective_size: state.solutions().count(),
                     time: libafl_bolts::current_time(),
                 },

libafl/src/fuzzer/mod.rs

@@ -426,6 +426,9 @@ where
                     manager.fire(
                         state,
                         Event::Objective {
+                            #[cfg(feature = "share_objectives")]
+                            input,
+
                             objective_size: state.solutions().count(),
                             time: current_time(),
                         },
@@ -610,6 +613,9 @@ where
             manager.fire(
                 state,
                 Event::Objective {
+                    #[cfg(feature = "share_objectives")]
+                    input,
+
                     objective_size: state.solutions().count(),
                     time: current_time(),
                 },