Share objectives between nodes

libafl/Cargo.toml

@@ -146,6 +146,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

@@ -285,15 +285,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
-                Event::Stop => true,
+                Event::UpdateExecStats { .. } | Event::Stop => true, // send UpdateExecStats 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,
+
                 _ => false,
             };
 
@@ -677,6 +680,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

@@ -469,6 +469,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,7 +346,136 @@ where
         }
     }
 
+    #[cfg(feature = "share_objectives")]
+    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: HasSolutions + HasCurrentTestcase,
+        S::Corpus: Corpus<Input = S::Input>,
+        S::Solutions: 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
+    #[cfg(not(feature = "share_objectives"))]
+    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)
+    }
+
     /// Handle arriving events in the client
+    #[cfg(feature = "share_objectives")]
     pub fn process<E, EM, Z>(
         &mut self,
         fuzzer: &mut Z,
@@ -351,7 +486,9 @@ where
     where
         E: Executor<EM, Z, State = S> + HasObservers,
         EM: UsesState<State = S> + EventFirer,
+        S: HasSolutions + HasCurrentTestcase,
         S::Corpus: Corpus<Input = S::Input>,
+        S::Solutions: 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>,

libafl/src/events/mod.rs

@@ -323,6 +323,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

@@ -643,6 +643,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

@@ -472,6 +472,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

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

libafl/src/stages/sync.rs

@@ -17,7 +17,10 @@ use crate::{
     fuzzer::{Evaluator, EvaluatorObservers, ExecutionProcessor},
     inputs::{Input, InputConverter, UsesInput},
     stages::{RetryCountRestartHelper, Stage},
-    state::{HasCorpus, HasExecutions, HasRand, MaybeHasClientPerfMonitor, State, Stoppable},
+    state::{
+        HasCorpus, HasExecutions, HasRand, HasSolutions, MaybeHasClientPerfMonitor, State,
+        Stoppable,
+    },
     Error, HasMetadata, HasNamedMetadata,
 };
 
@@ -232,6 +235,7 @@ where
     client: LlmpEventConverter<DI, IC, ICB, S, SP>,
 }
 
+// Do not include trait bound HasSolutions to S if share_objectives is disabled
 impl<E, EM, IC, ICB, DI, S, SP, Z> Stage<E, EM, S, Z> for SyncFromBrokerStage<DI, IC, ICB, S, SP>
 where
     EM: EventFirer<State = S>,
@@ -241,7 +245,8 @@ where
         + HasMetadata
         + Stoppable
         + UsesInput<Input = <S::Corpus as Corpus>::Input>
-        + State,
+        + State
+        + HasSolutions,
     SP: ShMemProvider,
     E: HasObservers + Executor<EM, Z, State = S>,
     for<'a> E::Observers: Deserialize<'a>,
@@ -251,6 +256,7 @@ where
     ICB: InputConverter<From = DI, To = <S::Corpus as Corpus>::Input>,
     DI: Input,
     <<S as HasCorpus>::Corpus as Corpus>::Input: Input + Clone,
+    S::Solutions: Corpus<Input = S::Input>,
 {
     #[inline]
     fn perform(