diff --git a/Cargo.lock b/Cargo.lock index ddd4dea..cf5ce0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -159,6 +159,7 @@ dependencies = [ "chaff-datasets", "chaff-machines", "chaff-sim", + "iterstats", "mac_address", "pcap", "rand", @@ -344,6 +345,15 @@ dependencies = [ "hybrid-array", ] +[[package]] +name = "iterstats" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea8ea10912cb6c0e34dfb7f2c5fbb9d03d1611b1651dfc9754e08c8185ec9d3" +dependencies = [ + "num-traits", +] + [[package]] name = "itoa" version = "1.0.18" diff --git a/crates/chaff-capture/src/trace.rs b/crates/chaff-capture/src/trace.rs index 8e6d532..ab87b95 100644 --- a/crates/chaff-capture/src/trace.rs +++ b/crates/chaff-capture/src/trace.rs @@ -29,11 +29,14 @@ impl TryFrom for Direction { Event::ReceiveNormal | Event::ReceiveDecoy => Ok(Self::Receive), Event::QueuePopped(_) | Event::QueueFull(_) + | Event::QueueFilled(_) + | Event::QueuePushed(_) | Event::StateBudgetExhausted | Event::MachineBudgetExhausted | Event::MachineBudgetReached | Event::MachineBudgetRecovered - | Event::QueueEmpty(_) => Err(CaptureError::CantConvert), + | Event::QueueEmpty(_) + | Event::SendBlocked => Err(CaptureError::CantConvert), } } } diff --git a/crates/chaff-cli/Cargo.toml b/crates/chaff-cli/Cargo.toml index e93cc0c..c1b21e8 100644 --- a/crates/chaff-cli/Cargo.toml +++ b/crates/chaff-cli/Cargo.toml @@ -28,3 +28,4 @@ bpaf = { version = "0.9.24", features = ["derive"] } rand = "0.10.1" mac_address = "1.1.8" pcap = "2.4.0" +iterstats = "0.7.0" diff --git a/crates/chaff-cli/src/main.rs b/crates/chaff-cli/src/main.rs index 9a31ada..ec5c7b2 100644 --- a/crates/chaff-cli/src/main.rs +++ b/crates/chaff-cli/src/main.rs @@ -13,7 +13,7 @@ use chaff_cli::{ errors::CliError, subcommands::{cap_convert, capture, dataset_convert, dataset_stats, simulate, trace_stats}, }; -use chaff_machines::test::construct_test_machine; +use chaff_machines::constant; /// Command-line interface options #[derive(Debug, Clone, Bpaf)] @@ -150,7 +150,7 @@ fn run() -> Result<(), CliError> { let mut file = File::open(path)?; Machine::deserialize_reader(&mut file)? } else { - construct_test_machine() + constant::construct() }; match action { diff --git a/crates/chaff-cli/src/subcommands/trace_stats.rs b/crates/chaff-cli/src/subcommands/trace_stats.rs index 1f065d0..2422e83 100644 --- a/crates/chaff-cli/src/subcommands/trace_stats.rs +++ b/crates/chaff-cli/src/subcommands/trace_stats.rs @@ -6,24 +6,32 @@ use chaff_capture::trace::{Direction, Trace}; use crate::errors::CliError; +use iterstats::Iterstats as _; + /// Run the trace stats subcommand with the given trace. /// /// # Errors /// /// - If deserialising the trace fails [`Trace::deserialise`]. -#[expect(clippy::cast_precision_loss)] #[expect(clippy::similar_names)] pub fn run(input: &PathBuf) -> Result<(), CliError> { let trace = Trace::deserialise(&input)?; let deltas = trace.timing_deltas(); + let sizes = trace.sizes(); let total = trace.len(); + let sent = trace .directions() .iter() .filter(|direction| **direction == Direction::Send) .count(); - let avg_delta = f64::from(deltas.iter().sum::()) / total as f64; - let avg_size = f64::from(trace.sizes().iter().sum::()) / total as f64; + + let avg_delta = deltas.iter().map(|delta| f64::from(*delta)).mean(); + let std_delta = deltas.iter().map(|delta| f64::from(*delta)).stddev(); + + let avg_size = sizes.iter().map(|size| f64::from(*size)).mean(); + let std_size = sizes.iter().map(|size| f64::from(*size)).stddev(); + let largest_burst_0mus = deltas .split(|&delta| delta != 0) .map(<[u32]>::len) @@ -61,8 +69,8 @@ pub fn run(input: &PathBuf) -> Result<(), CliError> { println!("Packets: {total}"); println!("Sent: {sent}"); println!("Received: {}", total - sent); - println!("Average packet size: {avg_size:.2} bytes"); - println!("Average time delta: {avg_delta:.2}μs"); + println!("Average packet size: {avg_size:.2}±{std_size:.2} bytes"); + println!("Average time delta: {avg_delta:.2}±{std_delta:.2}μs"); println!("Largest time delta: {largest_delta}μs"); println!("Largest burst (0μs): {largest_burst_0mus} packets"); println!("Largest burst (100μs): {largest_burst_100mus} packets"); diff --git a/crates/chaff-machines/src/constant.rs b/crates/chaff-machines/src/constant.rs new file mode 100644 index 0000000..16744f7 --- /dev/null +++ b/crates/chaff-machines/src/constant.rs @@ -0,0 +1,87 @@ +//! An attempted implementation of a machine that tries to make packets flow at a constant rate. + +#![cfg(not(tarpaulin_include))] + +use std::time::Duration; + +use chaff::{ + action::{FrameworkAction, IntegratorAction}, + distr::Distr, + event::Event, + machine, + machine::Machine, +}; + +/// Construct the constant machine. +/// +/// # Panics +/// +/// Should not panic unless modified. +#[must_use] +#[expect(clippy::expect_used)] +pub fn construct() -> Machine { + let hundred_ms: Distr = Duration::from_millis(100) + .try_into() + .expect("constant shouldn't cause validation issue"); + let max: Distr = Duration::from_nanos(u64::MAX) + .try_into() + .expect("constant shouldn't cause validation issue"); + let zero: Distr = Duration::ZERO + .try_into() + .expect("constant shouldn't cause validation issue"); + + machine! { + queues: [Some(1), Some(1), Some(1)], + budget: Absolute(500), + state init { + action: IntegratorAction::BlockOutgoing(max), + transitions: [ + Event::SendNormal => init, + Event::SendBlocked => schedule_release, + Event::ReceiveNormal => schedule_release, + ], + }, + state do_block { + action: FrameworkAction::schedule( + IntegratorAction::BlockOutgoing(max), + 1, + zero, + ), + transitions: [ + Event::QueueFilled(1) => schedule_release, + ], + }, + state send_decoy { + action: FrameworkAction::schedule( + IntegratorAction::SendDecoy, + 2, + zero, + ), + transitions: [ + Event::QueueFilled(2) => do_block, + ] + }, + state schedule_release { + action: FrameworkAction::schedule( + IntegratorAction::ReleaseBlock, + 0, + hundred_ms, + ), + transitions: [ + Event::QueueFilled(0) => blocked, + ], + }, + state blocked { + transitions: [ + Event::QueueEmpty(0) => send_decoy, + Event::SendBlocked => wait, + ] + }, + state wait { + transitions: [ + Event::QueueEmpty(0) => do_block, + ] + }, + } + .expect("predefined machine should be valid") +} diff --git a/crates/chaff-machines/src/lib.rs b/crates/chaff-machines/src/lib.rs index 7684047..d0bae2c 100644 --- a/crates/chaff-machines/src/lib.rs +++ b/crates/chaff-machines/src/lib.rs @@ -1,3 +1,4 @@ //! Machines written in the Chaff framework. +pub mod constant; pub mod test; diff --git a/crates/chaff-sim/src/lib.rs b/crates/chaff-sim/src/lib.rs index cfc9d2b..55c484f 100644 --- a/crates/chaff-sim/src/lib.rs +++ b/crates/chaff-sim/src/lib.rs @@ -1,4 +1,5 @@ //! The Chaff simulator for creating defended traces with machines. + use std::{ cmp::Ordering, collections::{BTreeMap, VecDeque}, @@ -314,6 +315,7 @@ impl Simulator { for event in &events_now { if event.event == Event::SendNormal && block_state.is_active_at(event.time) { block_state.buffer(event.clone()); + buffered_events.push(Event::SendBlocked); continue; } diff --git a/crates/chaff/src/action.rs b/crates/chaff/src/action.rs index 683f840..1711fdb 100644 --- a/crates/chaff/src/action.rs +++ b/crates/chaff/src/action.rs @@ -45,7 +45,7 @@ impl FrameworkAction { feature = "borsh", derive(borsh::BorshSerialize, borsh::BorshDeserialize) )] -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, Copy, PartialEq)] pub enum IntegratorAction { /// Send a decoy packet. SendDecoy, diff --git a/crates/chaff/src/event.rs b/crates/chaff/src/event.rs index d8923c3..220197d 100644 --- a/crates/chaff/src/event.rs +++ b/crates/chaff/src/event.rs @@ -22,10 +22,19 @@ pub enum Event { /// Decoy packet received (ingress). Emitted by integrator. ReceiveDecoy, + /// Packet blocked from sending. Emitted by framework. + SendBlocked, + /// Given queue was popped. Emitted by framework. QueuePopped(u8), - /// Given queue has reached capacity. Emitted by framework. + /// Given queue was pushed to. Emitted by framework. + QueuePushed(u8), + + /// Given queue was pushed to but has been filled. Emitted by framework. + QueueFilled(u8), + + /// Given queue is already at capacity. Emitted by framework. QueueFull(u8), /// Given queue has emptied. Emitted by framework. @@ -53,11 +62,14 @@ impl Event { Self::SendNormal | Self::ReceiveNormal | Self::SendDecoy | Self::ReceiveDecoy => false, Self::QueuePopped(_) | Self::QueueFull(_) + | Self::QueueFilled(_) + | Self::QueuePushed(_) | Self::StateBudgetExhausted | Self::MachineBudgetExhausted | Self::MachineBudgetReached | Self::MachineBudgetRecovered - | Self::QueueEmpty(_) => true, + | Self::QueueEmpty(_) + | Self::SendBlocked => true, } } } diff --git a/crates/chaff/src/framework.rs b/crates/chaff/src/framework.rs index 6fa129b..ec81987 100644 --- a/crates/chaff/src/framework.rs +++ b/crates/chaff/src/framework.rs @@ -9,7 +9,7 @@ use crate::{ action::{Action, FrameworkAction, IntegratorAction}, event::Event, machine::{Machine, MachineDecoyBudget, MachineRuntime}, - queue::{TimedAction, TimedQueue}, + queue::{QueuePushStatus, TimedAction, TimedQueue}, state::TransitionProbs, }; @@ -57,11 +57,19 @@ impl Framework { queue, delay, } => { - if !self.runtime.queues[queue as usize].push(TimedAction { + match self.runtime.queues[queue as usize].push(TimedAction { execute_at: now + delay.sample(&mut self.rng), action: int_action.into(), }) { - self.runtime.deferred_events.push(Event::QueueFull(queue)); + QueuePushStatus::Pushed => { + self.runtime.deferred_events.push(Event::QueuePushed(queue)); + } + QueuePushStatus::PushedButFull => { + self.runtime.deferred_events.push(Event::QueueFilled(queue)); + } + QueuePushStatus::Full => { + self.runtime.deferred_events.push(Event::QueueFull(queue)); + } } } FrameworkAction::CancelQueue(queue) => self.runtime.queues[queue as usize].cancel(), diff --git a/crates/chaff/src/machine.rs b/crates/chaff/src/machine.rs index 93b0f4c..0a0e7c5 100644 --- a/crates/chaff/src/machine.rs +++ b/crates/chaff/src/machine.rs @@ -106,6 +106,9 @@ impl Machine { errors.push(ValidationError::NegativeProportion(percent)); } + // TODO: maybe validate transitions too, i.e. no point in having a QueueFilled(10) + // transition when there's only 10 queues. + #[expect(clippy::expect_used)] match errors.len() { 0 => Ok(()), diff --git a/crates/chaff/src/queue.rs b/crates/chaff/src/queue.rs index cdb5715..abe560c 100644 --- a/crates/chaff/src/queue.rs +++ b/crates/chaff/src/queue.rs @@ -64,6 +64,18 @@ impl Timed for TimedAction { } } +/// Status from pushing to a [`TimedQueue`] +pub enum QueuePushStatus { + /// Push successful. + Pushed, + + /// Push successful but queue now full. + PushedButFull, + + /// Push unsuccessful as queue full. + Full, +} + /// A priority queue of [`Scheduled`] objects, ordered by their [`Timed::execute_at`]. Under the hood /// this is just a [`BinaryHeap>`]. #[derive(Debug, Clone, Default)] @@ -84,14 +96,20 @@ impl TimedQueue { /// Push an item onto the [`TimedQueue`]. #[must_use] - pub fn push(&mut self, item: T) -> bool { + pub fn push(&mut self, item: T) -> QueuePushStatus { if let Some(capacity) = self.capacity && self.queue.len() >= capacity { - false + QueuePushStatus::Full } else { self.queue.push(Scheduled(item)); - true + if let Some(capacity) = self.capacity + && self.queue.len() >= capacity + { + QueuePushStatus::PushedButFull + } else { + QueuePushStatus::Pushed + } } }