Convert ShMem into a state-full ShMemProvider and otherwise refactor shmem/llmp (#54)

* shmeme/llmp refactor to convert ShMem into a stateful ShMemProvider
factory.

At the moment we use parking_lot::ReentrantMutex. That may not be
necessary.

* fix merge issue

* formatting

* Fix fuzzer examples for new ShMemProvider

* Fix clippy warnings

* Fix build and clippy for x86_64

* Resolve review comments

* Remove ReentrantMutex and RefCell - they are not needed

* Hopefully fix win32 build

* Fix tests, windows build

* Rename ShMemProvider to ShMem

* Revert "Rename ShMemProvider to ShMem"

This reverts commit eca07c8d7bb3d5e829fecf3f7213c763470a41e9.

* Rename ShMemMapping to ShMem; Test fixes

* Add missing trait to scope

* Fix from_int

* Fix try_into

* Move to alloc::sync::Arc and spin::Mutex to support nostd

* Fix tests

* nostd fixes; Make new() a part of the ShMemProvider trait

* Fix errant ?

* Fix windows

* Fix missing trait

* nostd remove dbg!

* Add Default and Clone to ShMemProvider

* Formatting

* Fix windows

* Get rid of ArcMutex in favor of RefCell

* Rc RefCell

* moved to refs

* SHP->SP

* Use alloc::rc::Rc instead of std::rc::Rc

* Format

* Add setup_restarting_mgr_std which selects the right ShMemProvider; changed fuzzers to use it

* Get rid of unnecessary clone

* Fix clippy error on windows

* Fix nostd

* Fix formatting

* Make StdShmemProvider include ServedShMemProvider

* Get rid of lifetime specifiers now that we are using Rc

* Get rid of unneccesary spin

* Rename ShMemProvider::Mapping to ShMemProvider::Mem

* Formatting

* fix Windows

* Rename DefaultUnixShmem* to CommonUnixShmem*

Co-authored-by: Dominik Maier <domenukk@gmail.com>

fuzzers/frida_libpng/src/fuzzer.rs

@@ -2,15 +2,12 @@
 //! The example harness is built for libpng.
 
 use libafl::{
-    bolts::{
+    bolts::tuples::{tuple_list, Named},
-        shmem::UnixShMem,
-        tuples::{tuple_list, Named},
-    },
     corpus::{
         Corpus, InMemoryCorpus, IndexesLenTimeMinimizerCorpusScheduler, OnDiskCorpus,
         QueueCorpusScheduler,
     },
-    events::{setup_restarting_mgr, EventManager},
+    events::{setup_restarting_mgr_std, EventManager},
     executors::{inprocess::InProcessExecutor, Executor, ExitKind, HasObservers},
     feedbacks::{CrashFeedback, MaxMapFeedback},
     fuzzer::{Fuzzer, StdFuzzer},
@@ -25,6 +22,7 @@ use libafl::{
     Error,
 };
 
+use core::cell::RefCell;
 #[cfg(target_arch = "x86_64")]
 use frida_gum::instruction_writer::X86Register;
 #[cfg(target_arch = "aarch64")]
@@ -34,8 +32,7 @@ use frida_gum::{
     stalker::{NoneEventSink, Stalker, Transformer},
 };
 use frida_gum::{Gum, MemoryRange, Module, NativePointer, PageProtection};
-
+use std::{env, ffi::c_void, path::PathBuf};
-use std::{cell::RefCell, env, ffi::c_void, path::PathBuf};
 
 /// An helper that feeds FridaInProcessExecutor with user-supplied instrumentation
 pub trait FridaHelper<'a> {
@@ -399,7 +396,7 @@ unsafe fn fuzz(
 
     // The restarting state will spawn the same process again as child, then restarted it each time it crashes.
     let (state, mut restarting_mgr) =
-        match setup_restarting_mgr::<_, _, UnixShMem, _>(stats, broker_port) {
+        match setup_restarting_mgr_std(stats, broker_port) {
             Ok(res) => res,
             Err(err) => match err {
                 Error::ShuttingDown => {

fuzzers/libfuzzer_libmozjpeg/src/lib.rs

@@ -4,9 +4,9 @@
 use std::{env, path::PathBuf};
 
 use libafl::{
-    bolts::{shmem::UnixShMem, tuples::tuple_list},
+    bolts::tuples::tuple_list,
     corpus::{Corpus, InMemoryCorpus, OnDiskCorpus, RandCorpusScheduler},
-    events::setup_restarting_mgr,
+    events::setup_restarting_mgr_std,
     executors::{inprocess::InProcessExecutor, ExitKind},
     feedbacks::{CrashFeedback, MaxMapFeedback},
     fuzzer::{Fuzzer, StdFuzzer},
@@ -20,9 +20,11 @@ use libafl::{
     Error,
 };
 
-use libafl_targets::{libfuzzer_initialize, libfuzzer_test_one_input, EDGES_MAP, MAX_EDGES_NUM, CMP_MAP, CMP_MAP_SIZE};
+use libafl_targets::{
+    libfuzzer_initialize, libfuzzer_test_one_input, CMP_MAP, CMP_MAP_SIZE, EDGES_MAP, MAX_EDGES_NUM,
+};
 
-const ALLOC_MAP_SIZE: usize = 16*1024;
+const ALLOC_MAP_SIZE: usize = 16 * 1024;
 extern "C" {
     static mut libafl_alloc_map: [usize; ALLOC_MAP_SIZE];
 }
@@ -53,17 +55,19 @@ fn fuzz(corpus_dirs: Vec<PathBuf>, objective_dir: PathBuf, broker_port: u16) ->
 
     // The restarting state will spawn the same process again as child, then restarted it each time it crashes.
     let (state, mut restarting_mgr) =
-        setup_restarting_mgr::<_, _, UnixShMem, _>(stats, broker_port)
+        setup_restarting_mgr_std(stats, broker_port)
             .expect("Failed to setup the restarter".into());
 
     // Create an observation channel using the coverage map
-    let edges_observer = StdMapObserver::new("edges", unsafe { &mut EDGES_MAP }, unsafe { MAX_EDGES_NUM });
+    let edges_observer =
+        StdMapObserver::new("edges", unsafe { &mut EDGES_MAP }, unsafe { MAX_EDGES_NUM });
 
     // Create an observation channel using the cmp map
     let cmps_observer = StdMapObserver::new("cmps", unsafe { &mut CMP_MAP }, CMP_MAP_SIZE);
 
     // Create an observation channel using the allocations map
-    let allocs_observer = StdMapObserver::new("allocs", unsafe { &mut libafl_alloc_map }, ALLOC_MAP_SIZE);
+    let allocs_observer =
+        StdMapObserver::new("allocs", unsafe { &mut libafl_alloc_map }, ALLOC_MAP_SIZE);
 
     // If not restarting, create a State from scratch
     let mut state = state.unwrap_or_else(|| {

fuzzers/libfuzzer_libpng/src/lib.rs

@@ -5,12 +5,12 @@ use core::time::Duration;
 use std::{env, path::PathBuf};
 
 use libafl::{
-    bolts::{shmem::StdShMem, tuples::tuple_list},
+    bolts::tuples::tuple_list,
     corpus::{
         Corpus, InMemoryCorpus, IndexesLenTimeMinimizerCorpusScheduler, OnDiskCorpus,
         QueueCorpusScheduler,
     },
-    events::setup_restarting_mgr,
+    events::setup_restarting_mgr_std,
     executors::{inprocess::InProcessExecutor, ExitKind, TimeoutExecutor},
     feedbacks::{CrashFeedback, MaxMapFeedback, TimeFeedback, TimeoutFeedback},
     fuzzer::{Fuzzer, StdFuzzer},
@@ -52,7 +52,7 @@ fn fuzz(corpus_dirs: Vec<PathBuf>, objective_dir: PathBuf, broker_port: u16) ->
 
     // The restarting state will spawn the same process again as child, then restarted it each time it crashes.
     let (state, mut restarting_mgr) =
-        match setup_restarting_mgr::<_, _, StdShMem, _>(stats, broker_port) {
+        match setup_restarting_mgr_std(stats, broker_port) {
             Ok(res) => res,
             Err(err) => match err {
                 Error::ShuttingDown => {

libafl/examples/llmp_test/main.rs

@@ -3,15 +3,19 @@ This shows how llmp can be used directly, without libafl abstractions
 */
 extern crate alloc;
 
+use alloc::rc::Rc;
 #[cfg(all(unix, feature = "std"))]
-use core::{convert::TryInto, time::Duration};
+use core::{cell::RefCell, convert::TryInto, time::Duration};
 #[cfg(all(unix, feature = "std"))]
 use std::{thread, time};
 
 use libafl::bolts::llmp::Tag;
 #[cfg(all(unix, feature = "std"))]
 use libafl::{
-    bolts::{llmp, shmem::UnixShMem},
+    bolts::{
+        llmp,
+        shmem::{ShMemProvider, StdShMemProvider},
+    },
     Error,
 };
 
@@ -21,7 +25,8 @@ const _TAG_1MEG_V1: Tag = 0xB1111161;
 
 #[cfg(all(unix, feature = "std"))]
 fn adder_loop(port: u16) -> ! {
-    let mut client = llmp::LlmpClient::<UnixShMem>::create_attach_to_tcp(port).unwrap();
+    let shmem_provider = Rc::new(RefCell::new(StdShMemProvider::new()));
+    let mut client = llmp::LlmpClient::create_attach_to_tcp(&shmem_provider, port).unwrap();
     let mut last_result: u32 = 0;
     let mut current_result: u32 = 0;
     loop {
@@ -63,7 +68,11 @@ fn adder_loop(port: u16) -> ! {
 
 #[cfg(all(unix, feature = "std"))]
 fn large_msg_loop(port: u16) -> ! {
-    let mut client = llmp::LlmpClient::<UnixShMem>::create_attach_to_tcp(port).unwrap();
+    let mut client = llmp::LlmpClient::create_attach_to_tcp(
+        &Rc::new(RefCell::new(StdShMemProvider::new())),
+        port,
+    )
+    .unwrap();
 
     let meg_buf = [1u8; 1 << 20];
 
@@ -124,7 +133,8 @@ fn main() {
 
     match mode.as_str() {
         "broker" => {
-            let mut broker = llmp::LlmpBroker::<UnixShMem>::new().unwrap();
+            let mut broker =
+                llmp::LlmpBroker::new(&Rc::new(RefCell::new(StdShMemProvider::new()))).unwrap();
             broker
                 .launch_listener(llmp::Listener::Tcp(
                     std::net::TcpListener::bind(format!("127.0.0.1:{}", port)).unwrap(),
@@ -133,7 +143,11 @@ fn main() {
             broker.loop_forever(&mut broker_message_hook, Some(Duration::from_millis(5)))
         }
         "ctr" => {
-            let mut client = llmp::LlmpClient::<UnixShMem>::create_attach_to_tcp(port).unwrap();
+            let mut client = llmp::LlmpClient::create_attach_to_tcp(
+                &Rc::new(RefCell::new(StdShMemProvider::new())),
+                port,
+            )
+            .unwrap();
             let mut counter: u32 = 0;
             loop {
                 counter = counter.wrapping_add(1);

libafl/src/bolts/os/ashmem_server.rs

@@ -5,12 +5,18 @@ and forwards them over unix domain sockets.
 */
 
 use crate::{
-    bolts::shmem::{ShMem, ShMemDescription, UnixShMem},
+    bolts::shmem::{
+        unix_shmem::ashmem::{AshmemShMem, AshmemShMemProvider},
+        ShMem, ShMemDescription, ShMemId, ShMemProvider,
+    },
     Error,
 };
 use hashbrown::HashMap;
 use serde::{Deserialize, Serialize};
-use std::io::{Read, Write};
+use std::{
+    io::{Read, Write},
+    sync::{Arc, Condvar, Mutex},
+};
 
 #[cfg(all(feature = "std", unix))]
 use nix::poll::{poll, PollFd, PollFlags};
@@ -18,8 +24,8 @@ use nix::poll::{poll, PollFd, PollFlags};
 #[cfg(all(feature = "std", unix))]
 use std::{
     os::unix::{
+        io::{AsRawFd, RawFd},
         net::{UnixListener, UnixStream},
-        {io::AsRawFd, prelude::RawFd},
     },
     thread,
 };
@@ -27,30 +33,43 @@ use std::{
 #[cfg(all(unix, feature = "std"))]
 use uds::{UnixListenerExt, UnixSocketAddr, UnixStreamExt};
 
-#[derive(Debug)]
-/// The Sharedmem backed by a `ShmemService`a
-pub struct ServedShMem {
-    stream: UnixStream,
-    shmem: Option<UnixShMem>,
-    slice: Option<[u8; 20]>,
-    fd: Option<RawFd>,
-}
 const ASHMEM_SERVER_NAME: &str = "@ashmem_server";
 
-impl ServedShMem {
+#[derive(Debug)]
-    /// Create a new ServedShMem and connect to the ashmem server.
+pub struct ServedShMemProvider {
-    pub fn connect(name: &str) -> Self {
+    stream: UnixStream,
-        Self {
+    inner: AshmemShMemProvider,
-            stream: UnixStream::connect_to_unix_addr(&UnixSocketAddr::from_abstract(name).unwrap())
+}
-                .expect("Failed to connect to the ashmem server"),
+
-            shmem: None,
+#[derive(Clone, Debug)]
-            slice: None,
+pub struct ServedShMem {
-            fd: None,
+    inner: AshmemShMem,
-        }
+    server_fd: i32,
+}
+
+impl ShMem for ServedShMem {
+    fn id(&self) -> ShMemId {
+        let client_id = self.inner.id();
+        ShMemId::from_string(&format!("{}:{}", self.server_fd, client_id.to_string()))
     }
 
+    fn len(&self) -> usize {
+        self.inner.len()
+    }
+
+    fn map(&self) -> &[u8] {
+        self.inner.map()
+    }
+
+    fn map_mut(&mut self) -> &mut [u8] {
+        self.inner.map_mut()
+    }
+}
+
+impl ServedShMemProvider {
     /// Send a request to the server, and wait for a response
-    fn send_receive(&mut self, request: AshmemRequest) -> ([u8; 20], RawFd) {
+    #[allow(clippy::similar_names)] // id and fd
+    fn send_receive(&mut self, request: AshmemRequest) -> (i32, i32) {
         let body = postcard::to_allocvec(&request).unwrap();
 
         let header = (body.len() as u32).to_be_bytes();
@@ -66,63 +85,62 @@ impl ServedShMem {
         self.stream
             .recv_fds(&mut shm_slice, &mut fd_buf)
             .expect("Did not receive a response");
-        (shm_slice, fd_buf[0])
+
+        let server_id = ShMemId::from_slice(&shm_slice);
+        let server_id_str = server_id.to_string();
+        let server_fd: i32 = server_id_str.parse().unwrap();
+        (server_fd, fd_buf[0])
     }
 }
 
-impl ShMem for ServedShMem {
+impl Default for ServedShMemProvider {
-    fn new_map(map_size: usize) -> Result<Self, crate::Error> {
+    fn default() -> Self {
-        let mut res = Self::connect(ASHMEM_SERVER_NAME);
+        Self::new()
-        let (shm_slice, fd) = res.send_receive(AshmemRequest::NewMap(map_size));
+    }
-        if fd == -1 {
+}
-            Err(Error::IllegalState(
+
-                "Could not allocate from the ashmem server".to_string(),
+impl Clone for ServedShMemProvider {
-            ))
+    fn clone(&self) -> Self {
-        } else {
+        Self::new()
-            res.slice = Some(shm_slice);
+    }
-            res.fd = Some(fd);
+}
-            res.shmem = Some(
+
-                UnixShMem::existing_from_shm_slice(&shm_slice, map_size)
+impl ShMemProvider for ServedShMemProvider {
-                    .expect("Failed to create the UnixShMem"),
+    type Mem = ServedShMem;
-            );
+
-            Ok(res)
+    /// Connect to the server and return a new ServedShMemProvider
+    fn new() -> Self {
+        Self {
+            stream: UnixStream::connect_to_unix_addr(
+                &UnixSocketAddr::new(ASHMEM_SERVER_NAME).unwrap(),
+            )
+            .expect("Unable to open connection to ashmem service"),
+            inner: AshmemShMemProvider::new(),
         }
     }
+    fn new_map(&mut self, map_size: usize) -> Result<Self::Mem, crate::Error> {
+        let (server_fd, client_fd) = self.send_receive(AshmemRequest::NewMap(map_size));
 
-    fn existing_from_shm_slice(
+        Ok(ServedShMem {
-        map_str_bytes: &[u8; 20],
+            inner: self
-        map_size: usize,
+                .inner
-    ) -> Result<Self, crate::Error> {
+                .from_id_and_size(ShMemId::from_string(&format!("{}", client_fd)), map_size)?,
-        let mut res = Self::connect(ASHMEM_SERVER_NAME);
+            server_fd,
-        let (shm_slice, fd) = res.send_receive(AshmemRequest::ExistingMap(ShMemDescription {
+        })
-            size: map_size,
-            str_bytes: *map_str_bytes,
-        }));
-        if fd == -1 {
-            Err(Error::IllegalState(
-                "Could not allocate from the ashmem server".to_string(),
-            ))
-        } else {
-            res.slice = Some(shm_slice);
-            res.fd = Some(fd);
-            res.shmem = Some(
-                UnixShMem::existing_from_shm_slice(&shm_slice, map_size)
-                    .expect("Failed to create the UnixShMem"),
-            );
-            Ok(res)
-        }
     }
 
-    fn shm_slice(&self) -> &[u8; 20] {
+    fn from_id_and_size(&mut self, id: ShMemId, size: usize) -> Result<Self::Mem, Error> {
-        self.slice.as_ref().unwrap()
+        let parts = id.to_string().split(':').collect::<Vec<&str>>();
-    }
+        let server_id_str = parts.get(0).unwrap();
-
+        let (server_fd, client_fd) = self.send_receive(AshmemRequest::ExistingMap(
-    fn map(&self) -> &[u8] {
+            ShMemDescription::from_string_and_size(server_id_str, size),
-        self.shmem.as_ref().unwrap().map()
+        ));
-    }
+        Ok(ServedShMem {
-
+            inner: self
-    fn map_mut(&mut self) -> &mut [u8] {
+                .inner
-        self.shmem.as_mut().unwrap().map_mut()
+                .from_id_and_size(ShMemId::from_string(&format!("{}", client_fd)), size)?,
+            server_fd,
+        })
     }
 }
 
@@ -138,13 +156,20 @@ pub enum AshmemRequest {
 }
 
 #[derive(Debug)]
-pub struct AshmemClient {
+struct AshmemClient {
-    unix_socket_file: String,
+    stream: UnixStream,
+}
+
+impl AshmemClient {
+    fn new(stream: UnixStream) -> Self {
+        Self { stream }
+    }
 }
 
 #[derive(Debug)]
 pub struct AshmemService {
-    maps: HashMap<[u8; 20], UnixShMem>,
+    provider: AshmemShMemProvider,
+    maps: Vec<AshmemShMem>,
 }
 
 impl AshmemService {
@@ -152,116 +177,145 @@ impl AshmemService {
     #[must_use]
     fn new() -> Self {
         AshmemService {
-            maps: HashMap::new(),
+            provider: AshmemShMemProvider::new(),
+            maps: Vec::new(),
         }
     }
 
     /// Read and handle the client request, send the answer over unix fd.
-    fn handle_client(&mut self, stream: &mut UnixStream) -> Result<(), Error> {
+    fn handle_client(&mut self, client: &mut AshmemClient) -> Result<(), Error> {
         // Always receive one be u32 of size, then the command.
         let mut size_bytes = [0u8; 4];
-        stream.read_exact(&mut size_bytes)?;
+        client.stream.read_exact(&mut size_bytes)?;
         let size = u32::from_be_bytes(size_bytes);
         let mut bytes = vec![];
         bytes.resize(size as usize, 0u8);
-        stream
+        client
+            .stream
             .read_exact(&mut bytes)
             .expect("Failed to read message body");
         let request: AshmemRequest = postcard::from_bytes(&bytes)?;
 
         // Handle the client request
-        let (shmem_slice, fd): ([u8; 20], RawFd) = match request {
+        let mapping = match request {
-            AshmemRequest::NewMap(map_size) => match UnixShMem::new(map_size) {
+            AshmemRequest::NewMap(map_size) => self.provider.new_map(map_size)?,
-                Err(e) => {
-                    println!("Error allocating shared map {:?}", e);
-                    ([0; 20], -1)
-                }
-                Ok(map) => {
-                    let res = (*map.shm_slice(), map.shm_id);
-                    self.maps.insert(*map.shm_slice(), map);
-                    res
-                }
-            },
             AshmemRequest::ExistingMap(description) => {
-                match self.maps.get(&description.str_bytes) {
+                self.provider.from_description(description)?
-                    None => {
-                        println!("Error finding shared map {:?}", description);
-                        ([0; 20], -1)
-                    }
-                    Some(map) => (*map.shm_slice(), map.shm_id),
-                }
             }
             AshmemRequest::Deregister(_) => {
                 return Ok(());
             }
         };
 
-        stream.send_fds(&shmem_slice, &[fd])?;
+        let id = mapping.id();
+        let server_fd: i32 = id.to_string().parse().unwrap();
+        client
+            .stream
+            .send_fds(&id.to_string().as_bytes(), &[server_fd])?;
+        self.maps.push(mapping);
         Ok(())
     }
 
     /// Create a new AshmemService, then listen and service incoming connections in a new thread.
-    pub fn start() -> Result<thread::JoinHandle<()>, Error> {
+    pub fn start() -> Result<thread::JoinHandle<Result<(), Error>>, Error> {
-        Ok(thread::spawn(move || {
+        #[allow(clippy::mutex_atomic)]
-            Self::new().listen(ASHMEM_SERVER_NAME).unwrap()
+        let syncpair = Arc::new((Mutex::new(false), Condvar::new()));
-        }))
+        let childsyncpair = Arc::clone(&syncpair);
+        let join_handle =
+            thread::spawn(move || Self::new().listen(ASHMEM_SERVER_NAME, childsyncpair));
+
+        let (lock, cvar) = &*syncpair;
+        let mut started = lock.lock().unwrap();
+        while !*started {
+            started = cvar.wait(started).unwrap();
+        }
+
+        Ok(join_handle)
     }
 
     /// Listen on a filename (or abstract name) for new connections and serve them. This function
     /// should not return.
-    fn listen(&mut self, filename: &str) -> Result<(), Error> {
+    fn listen(
-        let listener = UnixListener::bind_unix_addr(&UnixSocketAddr::new(filename)?)?;
+        &mut self,
-        let mut clients: HashMap<RawFd, (UnixStream, UnixSocketAddr)> = HashMap::new();
+        filename: &str,
-        let mut poll_fds: HashMap<RawFd, PollFd> = HashMap::new();
+        syncpair: Arc<(Mutex<bool>, Condvar)>,
-
+    ) -> Result<(), Error> {
-        poll_fds.insert(
+        let listener = if let Ok(listener) =
+            UnixListener::bind_unix_addr(&UnixSocketAddr::new(filename)?)
+        {
+            listener
+        } else {
+            let (lock, cvar) = &*syncpair;
+            *lock.lock().unwrap() = true;
+            cvar.notify_one();
+            return Err(Error::Unknown(
+                "The server appears to already be running. We are probably a client".to_string(),
+            ));
+        };
+        let mut clients: HashMap<RawFd, AshmemClient> = HashMap::new();
+        let mut poll_fds: Vec<PollFd> = vec![PollFd::new(
             listener.as_raw_fd(),
-            PollFd::new(listener.as_raw_fd(), PollFlags::POLLIN),
+            PollFlags::POLLIN | PollFlags::POLLRDNORM | PollFlags::POLLRDBAND,
-        );
+        )];
+
+        let (lock, cvar) = &*syncpair;
+        *lock.lock().unwrap() = true;
+        cvar.notify_one();
 
         loop {
-            let mut fds_to_poll: Vec<PollFd> = poll_fds.values().copied().collect();
+            match poll(&mut poll_fds, -1) {
-            let fd = match poll(&mut fds_to_poll, -1) {
+                Ok(num_fds) if num_fds > 0 => (),
-                Ok(fd) => fd,
+                Ok(_) => continue,
                 Err(e) => {
                     println!("Error polling for activity: {:?}", e);
                     continue;
                 }
             };
-            if fd == listener.as_raw_fd() {
+            let copied_poll_fds: Vec<PollFd> = poll_fds.iter().copied().collect();
-                let (stream, addr) = match listener.accept_unix_addr() {
+            for poll_fd in copied_poll_fds {
-                    Ok(stream_val) => stream_val,
+                let revents = poll_fd.revents().expect("revents should not be None");
-                    Err(e) => {
+                let raw_polled_fd =
-                        println!("Error accepting client: {:?}", e);
+                    unsafe { *((&poll_fd as *const PollFd) as *const libc::pollfd) }.fd;
-                        continue;
+                if revents.contains(PollFlags::POLLHUP) {
-                    }
+                    poll_fds.remove(poll_fds.iter().position(|item| *item == poll_fd).unwrap());
-                };
+                    clients.remove(&raw_polled_fd);
+                } else if revents.contains(PollFlags::POLLIN) {
+                    if clients.contains_key(&raw_polled_fd) {
+                        let mut client = clients.get_mut(&raw_polled_fd).unwrap();
+                        match self.handle_client(&mut client) {
+                            Ok(()) => (),
+                            Err(e) => {
+                                dbg!("Ignoring failed read from client", e, poll_fd);
+                                continue;
+                            }
+                        };
+                    } else {
+                        let (stream, addr) = match listener.accept_unix_addr() {
+                            Ok(stream_val) => stream_val,
+                            Err(e) => {
+                                println!("Error accepting client: {:?}", e);
+                                continue;
+                            }
+                        };
 
-                println!("Recieved connection from {:?}", addr);
+                        println!("Recieved connection from {:?}", addr);
-                let pollfd = PollFd::new(stream.as_raw_fd(), PollFlags::POLLIN);
+                        let pollfd = PollFd::new(
-                poll_fds.insert(stream.as_raw_fd(), pollfd);
+                            stream.as_raw_fd(),
-                clients
+                            PollFlags::POLLIN | PollFlags::POLLRDNORM | PollFlags::POLLRDBAND,
-                    .insert(stream.as_raw_fd(), (stream, addr))
+                        );
-                    .as_ref()
+                        poll_fds.push(pollfd);
-                    .unwrap();
+                        let mut client = AshmemClient::new(stream);
-            } else if poll_fds
+                        match self.handle_client(&mut client) {
-                .get(&fd)
+                            Ok(()) => (),
-                .unwrap()
+                            Err(e) => {
-                .revents()
+                                dbg!("Ignoring failed read from client", e);
-                .unwrap()
+                            }
-                .contains(PollFlags::POLLHUP)
+                        };
-            {
+                        clients.insert(client.stream.as_raw_fd(), client);
-                poll_fds.remove(&fd);
-                clients.remove(&fd);
-            } else {
-                let (stream, _addr) = clients.get_mut(&fd).unwrap();
-                match self.handle_client(stream) {
-                    Ok(()) => (),
-                    Err(e) => {
-                        dbg!("Ignoring failed read from client", e);
-                        continue;
                     }
-                };
+                } else {
+                    //println!("Unknown revents flags: {:?}", revents);
+                }
             }
         }
     }

libafl/src/events/llmp.rs

@@ -1,7 +1,7 @@
 //! LLMP-backed event manager for scalable multi-processed fuzzing
 
-use alloc::{string::ToString, vec::Vec};
+use alloc::{rc::Rc, string::ToString, vec::Vec};
-use core::{marker::PhantomData, time::Duration};
+use core::{cell::RefCell, marker::PhantomData, time::Duration};
 use serde::{de::DeserializeOwned, Serialize};
 
 #[cfg(feature = "std")]
@@ -17,12 +17,16 @@ use crate::utils::startable_self;
 use crate::utils::{fork, ForkResult};
 
 #[cfg(all(feature = "std", unix))]
-use crate::bolts::shmem::UnixShMem;
+use crate::bolts::shmem::UnixShMemProvider;
 
+#[cfg(all(feature = "std", target_os = "android"))]
+use crate::bolts::os::ashmem_server::AshmemService;
+#[cfg(feature = "std")]
+use crate::bolts::shmem::StdShMemProvider;
 use crate::{
     bolts::{
         llmp::{self, LlmpClient, LlmpClientDescription, LlmpSender, Tag},
-        shmem::ShMem,
+        shmem::ShMemProvider,
     },
     corpus::CorpusScheduler,
     events::{BrokerEventResult, Event, EventManager},
@@ -46,23 +50,23 @@ const LLMP_TAG_EVENT_TO_BOTH: llmp::Tag = 0x2B0741;
 const _LLMP_TAG_RESTART: llmp::Tag = 0x8357A87;
 const _LLMP_TAG_NO_RESTART: llmp::Tag = 0x57A7EE71;
 
-#[derive(Clone, Debug)]
+#[derive(Debug)]
-pub struct LlmpEventManager<I, S, SH, ST>
+pub struct LlmpEventManager<I, S, SP, ST>
 where
     I: Input,
     S: IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider + 'static,
     ST: Stats,
     //CE: CustomEvent<I>,
 {
     stats: Option<ST>,
-    llmp: llmp::LlmpConnection<SH>,
+    llmp: llmp::LlmpConnection<SP>,
     phantom: PhantomData<(I, S)>,
 }
 
 #[cfg(feature = "std")]
 #[cfg(unix)]
-impl<I, S, ST> LlmpEventManager<I, S, UnixShMem, ST>
+impl<I, S, ST> LlmpEventManager<I, S, UnixShMemProvider, ST>
 where
     I: Input,
     S: IfInteresting<I>,
@@ -72,10 +76,14 @@ where
     /// If the port is not yet bound, it will act as broker
     /// Else, it will act as client.
     #[cfg(feature = "std")]
-    pub fn new_on_port_std(stats: ST, port: u16) -> Result<Self, Error> {
+    pub fn new_on_port_std(
+        shmem_provider: &Rc<RefCell<UnixShMemProvider>>,
+        stats: ST,
+        port: u16,
+    ) -> Result<Self, Error> {
         Ok(Self {
             stats: Some(stats),
-            llmp: llmp::LlmpConnection::on_port(port)?,
+            llmp: llmp::LlmpConnection::on_port(shmem_provider, port)?,
             phantom: PhantomData,
         })
     }
@@ -83,16 +91,19 @@ where
     /// If a client respawns, it may reuse the existing connection, previously stored by LlmpClient::to_env
     /// Std uses UnixShMem.
     #[cfg(feature = "std")]
-    pub fn existing_client_from_env_std(env_name: &str) -> Result<Self, Error> {
+    pub fn existing_client_from_env_std(
-        Self::existing_client_from_env(env_name)
+        shmem_provider: &Rc<RefCell<UnixShMemProvider>>,
+        env_name: &str,
+    ) -> Result<Self, Error> {
+        Self::existing_client_from_env(shmem_provider, env_name)
     }
 }
 
-impl<I, S, SH, ST> Drop for LlmpEventManager<I, S, SH, ST>
+impl<I, S, SP, ST> Drop for LlmpEventManager<I, S, SP, ST>
 where
     I: Input,
     S: IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider,
     ST: Stats,
 {
     /// LLMP clients will have to wait until their pages are mapped by somebody.
@@ -101,32 +112,39 @@ where
     }
 }
 
-impl<I, S, SH, ST> LlmpEventManager<I, S, SH, ST>
+impl<I, S, SP, ST> LlmpEventManager<I, S, SP, ST>
 where
     I: Input,
     S: IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider,
     ST: Stats,
 {
     /// Create llmp on a port
     /// If the port is not yet bound, it will act as broker
     /// Else, it will act as client.
     #[cfg(feature = "std")]
-    pub fn new_on_port(stats: ST, port: u16) -> Result<Self, Error> {
+    pub fn new_on_port(
+        shmem_provider: &Rc<RefCell<SP>>,
+        stats: ST,
+        port: u16,
+    ) -> Result<Self, Error> {
         Ok(Self {
             stats: Some(stats),
-            llmp: llmp::LlmpConnection::on_port(port)?,
+            llmp: llmp::LlmpConnection::on_port(shmem_provider, port)?,
             phantom: PhantomData,
         })
     }
 
     /// 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(env_name: &str) -> Result<Self, Error> {
+    pub fn existing_client_from_env(
+        shmem_provider: &Rc<RefCell<SP>>,
+        env_name: &str,
+    ) -> Result<Self, Error> {
         Ok(Self {
             stats: None,
             llmp: llmp::LlmpConnection::IsClient {
-                client: LlmpClient::on_existing_from_env(env_name)?,
+                client: LlmpClient::on_existing_from_env(shmem_provider, env_name)?,
             },
             // Inserting a nop-stats element here so rust won't complain.
             // In any case, the client won't currently use it.
@@ -141,26 +159,21 @@ where
 
     /// Create an existing client from description
     pub fn existing_client_from_description(
+        shmem_provider: &Rc<RefCell<SP>>,
         description: &LlmpClientDescription,
     ) -> Result<Self, Error> {
         Ok(Self {
             stats: None,
-            llmp: llmp::LlmpConnection::existing_client_from_description(description)?,
+            llmp: llmp::LlmpConnection::existing_client_from_description(
+                shmem_provider,
+                description,
+            )?,
             // Inserting a nop-stats element here so rust won't complain.
             // In any case, the client won't currently use it.
             phantom: PhantomData,
         })
     }
 
-    /// A client on an existing map
-    pub fn for_client(client: LlmpClient<SH>) -> Self {
-        Self {
-            stats: None,
-            llmp: llmp::LlmpConnection::IsClient { client },
-            phantom: PhantomData,
-        }
-    }
-
     /// Write the config for a client eventmgr to env vars, a new client can reattach using existing_client_from_env
     #[cfg(feature = "std")]
     pub fn to_env(&self, env_name: &str) {
@@ -309,29 +322,11 @@ where
     }
 }
 
-#[cfg(all(feature = "std", unix))]
+impl<I, S, SP, ST> EventManager<I, S> for LlmpEventManager<I, S, SP, ST>
-impl<I, S, SH, ST> LlmpEventManager<I, S, SH, ST>
 where
     I: Input,
     S: IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider,
-    ST: Stats,
-{
-    #[cfg(all(feature = "std", unix))]
-    pub fn new_on_domain_socket(stats: ST, filename: &str) -> Result<Self, Error> {
-        Ok(Self {
-            stats: Some(stats),
-            llmp: llmp::LlmpConnection::on_domain_socket(filename)?,
-            phantom: PhantomData,
-        })
-    }
-}
-
-impl<I, S, SH, ST> EventManager<I, S> for LlmpEventManager<I, S, SH, ST>
-where
-    I: Input,
-    S: IfInteresting<I>,
-    SH: ShMem,
     ST: Stats, //CE: CustomEvent<I>,
 {
     /// The llmp client needs to wait until a broker mapped all pages, before shutting down.
@@ -388,14 +383,14 @@ where
 /// Serialize the current state and corpus during an executiont to bytes.
 /// On top, add the current llmp event manager instance to be restored
 /// This method is needed when the fuzzer run crashes and has to restart.
-pub fn serialize_state_mgr<I, S, SH, ST>(
+pub fn serialize_state_mgr<I, S, SP, ST>(
     state: &S,
-    mgr: &LlmpEventManager<I, S, SH, ST>,
+    mgr: &LlmpEventManager<I, S, SP, ST>,
 ) -> Result<Vec<u8>, Error>
 where
     I: Input,
     S: Serialize + IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider,
     ST: Stats,
 {
     Ok(postcard::to_allocvec(&(&state, &mgr.describe()?))?)
@@ -403,43 +398,44 @@ where
 
 /// Deserialize the state and corpus tuple, previously serialized with `serialize_state_corpus(...)`
 #[allow(clippy::type_complexity)]
-pub fn deserialize_state_mgr<I, S, SH, ST>(
+pub fn deserialize_state_mgr<I, S, SP, ST>(
+    shmem_provider: &Rc<RefCell<SP>>,
     state_corpus_serialized: &[u8],
-) -> Result<(S, LlmpEventManager<I, S, SH, ST>), Error>
+) -> Result<(S, LlmpEventManager<I, S, SP, ST>), Error>
 where
     I: Input,
     S: DeserializeOwned + IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider,
     ST: Stats,
 {
     let tuple: (S, _) = postcard::from_bytes(&state_corpus_serialized)?;
     Ok((
         tuple.0,
-        LlmpEventManager::existing_client_from_description(&tuple.1)?,
+        LlmpEventManager::existing_client_from_description(shmem_provider, &tuple.1)?,
     ))
 }
 
 /// A manager that can restart on the fly, storing states in-between (in `on_resatrt`)
-#[derive(Clone, Debug)]
+#[derive(Debug)]
-pub struct LlmpRestartingEventManager<I, S, SH, ST>
+pub struct LlmpRestartingEventManager<I, S, SP, ST>
 where
     I: Input,
     S: IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider + 'static,
     ST: Stats,
     //CE: CustomEvent<I>,
 {
     /// The embedded llmp event manager
-    llmp_mgr: LlmpEventManager<I, S, SH, ST>,
+    llmp_mgr: LlmpEventManager<I, S, SP, ST>,
     /// The sender to serialize the state for the next runner
-    sender: LlmpSender<SH>,
+    sender: LlmpSender<SP>,
 }
 
-impl<I, S, SH, ST> EventManager<I, S> for LlmpRestartingEventManager<I, S, SH, ST>
+impl<I, S, SP, ST> EventManager<I, S> for LlmpRestartingEventManager<I, S, SP, ST>
 where
     I: Input,
     S: IfInteresting<I> + Serialize,
-    SH: ShMem,
+    SP: ShMemProvider,
     ST: Stats, //CE: CustomEvent<I>,
 {
     /// The llmp client needs to wait until a broker mapped all pages, before shutting down.
@@ -484,29 +480,53 @@ const _ENV_FUZZER_RECEIVER: &str = &"_AFL_ENV_FUZZER_RECEIVER";
 /// The llmp (2 way) connection from a fuzzer to the broker (broadcasting all other fuzzer messages)
 const _ENV_FUZZER_BROKER_CLIENT_INITIAL: &str = &"_AFL_ENV_FUZZER_BROKER_CLIENT";
 
-impl<I, S, SH, ST> LlmpRestartingEventManager<I, S, SH, ST>
+impl<I, S, SP, ST> LlmpRestartingEventManager<I, S, SP, ST>
 where
     I: Input,
     S: IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider,
     ST: Stats, //CE: CustomEvent<I>,
 {
     /// Create a new runner, the executed child doing the actual fuzzing.
-    pub fn new(llmp_mgr: LlmpEventManager<I, S, SH, ST>, sender: LlmpSender<SH>) -> Self {
+    pub fn new(llmp_mgr: LlmpEventManager<I, S, SP, ST>, sender: LlmpSender<SP>) -> Self {
         Self { llmp_mgr, sender }
     }
 
     /// Get the sender
-    pub fn sender(&self) -> &LlmpSender<SH> {
+    pub fn sender(&self) -> &LlmpSender<SP> {
         &self.sender
     }
 
     /// Get the sender (mut)
-    pub fn sender_mut(&mut self) -> &mut LlmpSender<SH> {
+    pub fn sender_mut(&mut self) -> &mut LlmpSender<SP> {
         &mut self.sender
     }
 }
 
+#[cfg(feature = "std")]
+#[allow(clippy::type_complexity)]
+pub fn setup_restarting_mgr_std<I, S, ST>(
+    //mgr: &mut LlmpEventManager<I, S, SH, ST>,
+    stats: ST,
+    broker_port: u16,
+) -> Result<
+    (
+        Option<S>,
+        LlmpRestartingEventManager<I, S, StdShMemProvider, ST>,
+    ),
+    Error,
+>
+where
+    I: Input,
+    S: DeserializeOwned + IfInteresting<I>,
+    ST: Stats,
+{
+    #[cfg(target_os = "android")]
+    AshmemService::start().expect("Error starting Ashmem Service");
+
+    setup_restarting_mgr(StdShMemProvider::new(), stats, broker_port)
+}
+
 /// A restarting state is a combination of restarter and runner, that can be used on systems without `fork`.
 /// The restarter will start a new process each time the child crashes or timeouts.
 #[cfg(feature = "std")]
@@ -515,30 +535,25 @@ where
     clippy::type_complexity,
     clippy::similar_names
 )] // for { mgr = LlmpEventManager... }
-pub fn setup_restarting_mgr<I, S, SH, ST>(
+pub fn setup_restarting_mgr<I, S, SP, ST>(
+    shmem_provider: SP,
     //mgr: &mut LlmpEventManager<I, S, SH, ST>,
     stats: ST,
     broker_port: u16,
-) -> Result<(Option<S>, LlmpRestartingEventManager<I, S, SH, ST>), Error>
+) -> Result<(Option<S>, LlmpRestartingEventManager<I, S, SP, ST>), Error>
 where
     I: Input,
     S: DeserializeOwned + IfInteresting<I>,
-    SH: ShMem,
+    SP: ShMemProvider,
     ST: Stats,
 {
-    let mut mgr;
+    let shmem_provider = Rc::new(RefCell::new(shmem_provider));
+
+    let mut mgr =
+        LlmpEventManager::<I, S, SP, ST>::new_on_port(&shmem_provider, stats, broker_port)?;
 
     // We start ourself as child process to actually fuzz
-    let (sender, mut receiver) = if std::env::var(_ENV_FUZZER_SENDER).is_err() {
+    let (sender, mut receiver, shmem_provider) = if std::env::var(_ENV_FUZZER_SENDER).is_err() {
-        #[cfg(target_os = "android")]
-        {
-            mgr = LlmpEventManager::<I, S, SH, ST>::new_on_domain_socket(stats, "\x00llmp_socket")?;
-        };
-        #[cfg(not(target_os = "android"))]
-        {
-            mgr = LlmpEventManager::<I, S, SH, ST>::new_on_port(stats, broker_port)?
-        };
-
         if mgr.is_broker() {
             // Yep, broker. Just loop here.
             println!("Doing broker things. Run this tool again to start fuzzing in a client.");
@@ -550,11 +565,14 @@ where
         mgr.to_env(_ENV_FUZZER_BROKER_CLIENT_INITIAL);
 
         // First, create a channel from the fuzzer (sender) to us (receiver) to report its state for restarts.
-        let sender = LlmpSender::new(0, false)?;
+        let sender = { LlmpSender::new(&shmem_provider, 0, false)? };
-        let receiver = LlmpReceiver::on_existing_map(
+
-            SH::clone_ref(&sender.out_maps.last().unwrap().shmem)?,
+        let map = {
-            None,
+            shmem_provider
-        )?;
+                .borrow_mut()
+                .clone_ref(&sender.out_maps.last().unwrap().shmem)?
+        };
+        let receiver = LlmpReceiver::on_existing_map(shmem_provider.clone(), map, None)?;
         // Store the information to a map.
         sender.to_env(_ENV_FUZZER_SENDER)?;
         receiver.to_env(_ENV_FUZZER_RECEIVER)?;
@@ -568,7 +586,7 @@ where
             #[cfg(unix)]
             let _ = match unsafe { fork() }? {
                 ForkResult::Parent(handle) => handle.status(),
-                ForkResult::Child => break (sender, receiver),
+                ForkResult::Child => break (sender, receiver, shmem_provider),
             };
 
             // On windows, we spawn ourself again
@@ -585,20 +603,29 @@ where
     } else {
         // We are the newly started fuzzing instance, first, connect to our own restore map.
         // A sender and a receiver for single communication
+        // Clone so we get a new connection to the AshmemServer if we are using
+        // ServedShMemProvider
+        let shmem_provider = Rc::new(RefCell::new(shmem_provider.borrow_mut().clone()));
         (
-            LlmpSender::<SH>::on_existing_from_env(_ENV_FUZZER_SENDER)?,
+            LlmpSender::on_existing_from_env(&shmem_provider, _ENV_FUZZER_SENDER)?,
-            LlmpReceiver::<SH>::on_existing_from_env(_ENV_FUZZER_RECEIVER)?,
+            LlmpReceiver::on_existing_from_env(&shmem_provider, _ENV_FUZZER_RECEIVER)?,
+            shmem_provider,
         )
     };
 
     println!("We're a client, let's fuzz :)");
 
+    for (var, val) in std::env::vars() {
+        println!("ENV VARS: {:?}: {:?}", var, val);
+    }
+
     // If we're restarting, deserialize the old state.
     let (state, mut mgr) = match receiver.recv_buf()? {
         None => {
             println!("First run. Let's set it all up");
             // Mgr to send and receive msgs from/to all other fuzzer instances
-            let client_mgr = LlmpEventManager::<I, S, SH, ST>::existing_client_from_env(
+            let client_mgr = LlmpEventManager::<I, S, SP, ST>::existing_client_from_env(
+                &shmem_provider,
                 _ENV_FUZZER_BROKER_CLIENT_INITIAL,
             )?;
 
@@ -607,7 +634,8 @@ where
         // Restoring from a previous run, deserialize state and corpus.
         Some((_sender, _tag, msg)) => {
             println!("Subsequent run. Let's load all data from shmem (received {} bytes from previous instance)", msg.len());
-            let (state, mgr): (S, LlmpEventManager<I, S, SH, ST>) = deserialize_state_mgr(&msg)?;
+            let (state, mgr): (S, LlmpEventManager<I, S, SP, ST>) =
+                deserialize_state_mgr(&shmem_provider, &msg)?;
 
             (Some(state), LlmpRestartingEventManager::new(mgr, sender))
         }