Discard non-new testcase events for multi-machine messages

libafl/src/events/broker_hooks/centralized_multi_machine.rs

@@ -176,7 +176,8 @@ where
     ) -> Result<LlmpMsgHookResult, Error> {
         let shared_state = self.shared_state.clone();
 
-        // Here, we suppose msg will never be written again and will always be available.
+        // # Safety
+        // Here, we suppose msg will *never* be written again and will always be available.
         // Thus, it is safe to handle this in a separate thread.
         let msg_lock = unsafe { NullLock::new((msg.as_ptr(), msg.len())) };
         // let flags = msg_flags.clone();

libafl/src/events/llmp/mgr.rs

@@ -15,7 +15,7 @@ use libafl_bolts::{
 };
 use libafl_bolts::{
     current_time,
-    llmp::{LlmpClient, LlmpClientDescription},
+    llmp::{LlmpClient, LlmpClientDescription, LLMP_FLAG_FROM_MM},
     shmem::{NopShMemProvider, ShMemProvider},
     tuples::Handle,
     ClientId,
@@ -605,7 +605,7 @@ where
         // 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()? {
+        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!"
@@ -619,14 +619,21 @@ where
             #[cfg(feature = "llmp_compression")]
             let compressed;
             #[cfg(feature = "llmp_compression")]
-            let event_bytes = if _flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
+            let event_bytes = if flags & LLMP_FLAG_COMPRESSED == LLMP_FLAG_COMPRESSED {
                 compressed = self.compressor.decompress(msg)?;
                 &compressed
             } else {
                 msg
             };
             let event: Event<S::Input> = postcard::from_bytes(event_bytes)?;
             log::debug!("Received event in normal llmp {}", event.name_detailed());
+
+            // If the message comes from another machine, do not
+            // consider other events than new testcase.
+            if !event.is_new_testcase() && (flags & LLMP_FLAG_FROM_MM == LLMP_FLAG_FROM_MM) {
+                continue;
+            }
+
             self.handle_in_client(fuzzer, executor, state, client_id, event)?;
             count += 1;
         }

libafl/src/events/mod.rs

@@ -394,6 +394,11 @@ where
             } => "todo",*/
         }
     }
+
+    /// Returns true if self is a new testcase, false otherwise.
+    pub fn is_new_testcase(&self) -> bool {
+        matches!(self, Event::NewTestcase { .. })
+    }
 }
 
 /// [`EventFirer`] fires an event.

libafl_bolts/src/llmp.rs

@@ -143,6 +143,8 @@ pub const LLMP_FLAG_INITIALIZED: Flags = Flags(0x0);
 pub const LLMP_FLAG_COMPRESSED: Flags = Flags(0x1);
 /// From another broker.
 pub const LLMP_FLAG_FROM_B2B: Flags = Flags(0x2);
+/// From another machine (with the `multi_machine` mode)
+pub const LLMP_FLAG_FROM_MM: Flags = Flags(0x4);
 
 /// Timt the broker 2 broker connection waits for incoming data,
 /// before checking for own data to forward again.