Soundness issue: Input<R> can be misused to create data race to an object

[JOE1994]

Hello 🦀 , we (Rust group @sslab-gatech) found a soundness issue in this crate while scanning Rust code on crates.io for potential vulnerabilities.

Soundness Issue

Send is unconditionally implemented for Input<R>,
so that it is possible to send Input<R> to other threads even when R is not Send.

// location = 'src/pcap.rs:22:1: 22:55'
unsafe impl<R: Read> std::marker::Send for Input<R> {}

When Input<R> is misused, it is possible to create a data race to a non-Sync object.

Proof of Concept

• The example below program creates a data race to a Cell using Input<R>
  • modified from https://docs.rs/eventio/0.5.0/src/eventio/pcap.rs.html#148
• Multiple threads concurrently update the same Cell that counts the number of read events,
  making the Cell to contain incorrect statistics.
• fn main() compares the value contained inside Cell with the exact number of reads that happened.
• CustomRead is a Read object that contains a non Send object (Rc<Cell<usize>>)
• Program output is shown after the program below

use crossbeam_channel::unbounded;
use eventio::{pcap, Input};
use pcap_parser::{LegacyPcapBlock, PcapHeader, ToVec};
use std::{
    io::{self, Cursor},
    cell::Cell,
    rc::Rc,
    sync::atomic::{AtomicUsize, Ordering},
    thread,
};

/// Non-Send object implementing `Read` trait.
struct CustomRead {
    read: Cursor<Vec<u8>>,
    non_send_counter: Rc<Cell<usize>>,
    atomic_cnt: bool,
}
impl CustomRead {
    fn new(
        non_send_counter: Rc<Cell<usize>>,
        atomic_cnt: bool,
    ) -> Self {
        CustomRead {
            read: Self::create_pcap(),
            non_send_counter,
            atomic_cnt,
        }
    }
    fn create_pcap() -> Cursor<Vec<u8>> {
        let fake_content = b"fake packet";
        let pkt = LegacyPcapBlock {
            ts_sec: 0,
            ts_usec: 0,
            caplen: fake_content.len() as u32,
            origlen: fake_content.len() as u32,
            data: fake_content,
        }
        .to_vec_raw()
        .unwrap();
        let mut buf = PcapHeader::new().to_vec_raw().unwrap();
        for _ in 0..N_PACKETS {
            buf.extend(pkt.iter());
        }
        Cursor::new(buf)
    }
    fn update_read_cnt(&self) {
        if self.atomic_cnt {
            REAL_CNT.fetch_add(1, Ordering::Release);
        } else {
            self.non_send_counter.set(self.non_send_counter.get() + 1);
        }
    }
}
impl std::io::Read for CustomRead {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        // Increment counter to keep track of '# of read events'.
        self.update_read_cnt();
        self.read.read(buf)
    }
}

const N_PACKETS: usize = 9_000_000;
const NTHREADS: usize = 8;
static REAL_CNT: AtomicUsize = AtomicUsize::new(0);
fn experiment(atomic_cnt: bool) -> usize {
    let (data_tx, data_rx) = unbounded();
    let (ack_tx, ack_rx) = unbounded();

    // Non-Sync counter that counts the # of read events that happen within this thread.
    // This counter should only be updated within a single thread.
    // Updating this counter concurrently from multiple threads will result in incorrect counts.
    let read_cnt_in_thread= Rc::new(Cell::new(0_usize));

    let mut children = Vec::with_capacity(NTHREADS);
    for _ in 0..NTHREADS {
        let input = pcap::Input::with_read(
            data_tx.clone(),
            ack_rx.clone(),
            CustomRead::new(Rc::clone(&read_cnt_in_thread), atomic_cnt),
        );
        children.push(thread::spawn(move || {
            // `input` is moved from parent thread to child thread.
            input.run().unwrap()
        }));
    }

    std::mem::drop(data_tx);
    for ev in data_rx.iter() {
        ack_tx.send(ev.seq_no).unwrap();
    }
    std::mem::drop(ack_tx);

    for child in children.into_iter() {
        child.join().unwrap();
    }
    if atomic_cnt {
        REAL_CNT.load(Ordering::Acquire)
    } else {
        read_cnt_in_thread.get()
    }
}

fn main() {
    // Check that `read_cnt_in_thread` maintains incorrect count of events.
    assert_eq!(
        experiment(true),
        experiment(false),
    );
}

Program output

When compiled with rustc 1.49.0-nightly (release mode) & run on Ubuntu 18.04,
outputs from 3 executions of the program was as below.

thread 'main' panicked at 'assertion failed: `(left == right)`
  left: `29688`,
 right: `29672`', examples/eventio.rs:105:5
-------------------------------------------------------------------------------------
thread 'main' panicked at 'assertion failed: `(left == right)`
  left: `29688`,
 right: `29679`', examples/eventio.rs:105:5
-------------------------------------------------------------------------------------
thread 'main' panicked at 'assertion failed: `(left == right)`
  left: `29688`,
 right: `29680`', examples/eventio.rs:105:5

Suggested Solution

Simply adding trait bound R: Send to the Send impl for Input<R> will allow the compiler to revoke programs like the above.
After the change, Input<R> can no longer carry non-Send objects when moving across thread boundaries.

use std::marker::Send;

unsafe impl<R: Read + Send> Send for Input<R> {}

Thank you for very much for checking out this issue 👍

[minshao]

Hello 🦀 , we (Rust group @sslab-gatech) found a soundness issue in this crate while scanning Rust code on crates.io for potential vulnerabilities.

Soundness Issue

Send is unconditionally implemented for Input<R>,
so that it is possible to send Input<R> to other threads even when R is not Send.

// location = 'src/pcap.rs:22:1: 22:55'
unsafe impl<R: Read> std::marker::Send for Input<R> {}

When Input<R> is misused, it is possible to create a data race to a non-Sync object.

Proof of Concept

• The example below program creates a data race to a Cell using Input<R>

  • modified from https://docs.rs/eventio/0.5.0/src/eventio/pcap.rs.html#148

• Multiple threads concurrently update the same Cell that counts the number of read events,
  making the Cell to contain incorrect statistics.

• fn main() compares the value contained inside Cell with the exact number of reads that happened.

• CustomRead is a Read object that contains a non Send object (Rc<Cell<usize>>)

• Program output is shown after the program below

use crossbeam_channel::unbounded;
use eventio::{pcap, Input};
use pcap_parser::{LegacyPcapBlock, PcapHeader, ToVec};
use std::{
    io::{self, Cursor},
    cell::Cell,
    rc::Rc,
    sync::atomic::{AtomicUsize, Ordering},
    thread,
};

/// Non-Send object implementing `Read` trait.
struct CustomRead {
    read: Cursor<Vec<u8>>,
    non_send_counter: Rc<Cell<usize>>,
    atomic_cnt: bool,
}
impl CustomRead {
    fn new(
        non_send_counter: Rc<Cell<usize>>,
        atomic_cnt: bool,
    ) -> Self {
        CustomRead {
            read: Self::create_pcap(),
            non_send_counter,
            atomic_cnt,
        }
    }
    fn create_pcap() -> Cursor<Vec<u8>> {
        let fake_content = b"fake packet";
        let pkt = LegacyPcapBlock {
            ts_sec: 0,
            ts_usec: 0,
            caplen: fake_content.len() as u32,
            origlen: fake_content.len() as u32,
            data: fake_content,
        }
        .to_vec_raw()
        .unwrap();
        let mut buf = PcapHeader::new().to_vec_raw().unwrap();
        for _ in 0..N_PACKETS {
            buf.extend(pkt.iter());
        }
        Cursor::new(buf)
    }
    fn update_read_cnt(&self) {
        if self.atomic_cnt {
            REAL_CNT.fetch_add(1, Ordering::Release);
        } else {
            self.non_send_counter.set(self.non_send_counter.get() + 1);
        }
    }
}
impl std::io::Read for CustomRead {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        // Increment counter to keep track of '# of read events'.
        self.update_read_cnt();
        self.read.read(buf)
    }
}

const N_PACKETS: usize = 9_000_000;
const NTHREADS: usize = 8;
static REAL_CNT: AtomicUsize = AtomicUsize::new(0);
fn experiment(atomic_cnt: bool) -> usize {
    let (data_tx, data_rx) = unbounded();
    let (ack_tx, ack_rx) = unbounded();

    // Non-Sync counter that counts the # of read events that happen within this thread.
    // This counter should only be updated within a single thread.
    // Updating this counter concurrently from multiple threads will result in incorrect counts.
    let read_cnt_in_thread= Rc::new(Cell::new(0_usize));

    let mut children = Vec::with_capacity(NTHREADS);
    for _ in 0..NTHREADS {
        let input = pcap::Input::with_read(
            data_tx.clone(),
            ack_rx.clone(),
            CustomRead::new(Rc::clone(&read_cnt_in_thread), atomic_cnt),
        );
        children.push(thread::spawn(move || {
            // `input` is moved from parent thread to child thread.
            input.run().unwrap()
        }));
    }

    std::mem::drop(data_tx);
    for ev in data_rx.iter() {
        ack_tx.send(ev.seq_no).unwrap();
    }
    std::mem::drop(ack_tx);

    for child in children.into_iter() {
        child.join().unwrap();
    }
    if atomic_cnt {
        REAL_CNT.load(Ordering::Acquire)
    } else {
        read_cnt_in_thread.get()
    }
}

fn main() {
    // Check that `read_cnt_in_thread` maintains incorrect count of events.
    assert_eq!(
        experiment(true),
        experiment(false),
    );
}

Program output

When compiled with rustc 1.49.0-nightly (release mode) & run on Ubuntu 18.04,
outputs from 3 executions of the program was as below.

thread 'main' panicked at 'assertion failed: `(left == right)`
  left: `29688`,
 right: `29672`', examples/eventio.rs:105:5
-------------------------------------------------------------------------------------
thread 'main' panicked at 'assertion failed: `(left == right)`
  left: `29688`,
 right: `29679`', examples/eventio.rs:105:5
-------------------------------------------------------------------------------------
thread 'main' panicked at 'assertion failed: `(left == right)`
  left: `29688`,
 right: `29680`', examples/eventio.rs:105:5

Suggested Solution

Simply adding trait bound R: Send to the Send impl for Input<R> will allow the compiler to revoke programs like the above.
After the change, Input<R> can no longer carry non-Send objects when moving across thread boundaries.

use std::marker::Send;

unsafe impl<R: Read + Send> Send for Input<R> {}

Thank you for very much for checking out this issue 👍

Thanks for the help, really appreciate it.

[JOE1994]

Could you publish a new release to crates.io as a follow-up to this bug fix?

[minshao]

Released~ @JOE1994