kyle/tines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(core): fuzz harness and proptest twin

Browse Source
This commit is contained in:
kyle
2026-10-22 08:44:09 -07:00
parent ef539b494f
commit 164cf97a5b
1084 changed files with 1301 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fuzz/src/lib.rs
+592
View File
@@ -0,0 +1,592 @@
//! Shared fuzz + proptest harness that drives `tines-core`'s [`Node`] (over a
//! [`MemStore`] and a scripted [`Transport`]) through an ordered list of
//! [`Step`]s, asserting the §7 core-safety and §10 adversarial-peer properties.
//!
//! The same module is compiled into two crates:
//! * the `cargo-fuzz` target (`fuzz/fuzz_targets/node.rs`) via its package lib,
//! decoding arbitrary bytes into [`Step`]s with [`decode_steps`]; and
//! * the stable proptest twin (`tines-core/tests/fuzz_twin.rs`) via a `#[path]`
//! include, generating [`Step`]s directly with proptest strategies.
//!
//! Both entry points converge on [`Harness::step`] / [`run`], so any bug fixed in
//! this oracle is caught by either harness — that is what "same harness" means
//! here (and the reason a second copy of the code would defeat the point).
#![allow(dead_code, clippy::needless_range_loop)]
use std::cell::RefCell;
use std::collections::VecDeque;
use std::rc::Rc;
use tines_core::transport::FrameBuf;
use tines_core::{
consts::{ACK_IDS_CAP, MAX_FRAME, MAX_PAYLOAD, MAX_TOPIC, SUMMARY_BYTES},
Event, Frame, MemStore, Node, NodeId, ObservationId, SendError, Tick, Transport,
};
/// Wire address of a peer on the scripted transport. The exchange pins each
/// `Addr` to the `from`-id of its first frame (F6), so an attacker who controls
/// addresses can present many identities — exactly what adversarial mode tests.
pub type Addr = u64;
/// Which invariant suite a run exercises.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum Mode {
/// Basic: raw injected frames (possibly garbage), ticks, and our own
/// observations from an uncontrolled mix of addresses. Exercises §7 only.
Basic,
/// Adversarial: well-formed frames built from chosen bits/ids/lifetimes,
/// driven from many addresses beyond `MAX_PEERS`. Exercises §7 **and** the
/// §10 peer-table properties (flood recovery, summary idempotency).
Adversarial,
}
/// A single scripted action fed to the node. Basic mode mostly issues
/// [`Step::Frame`] / [`Step::Advance`] from an uncontrolled set of addresses;
/// adversarial mode interleaves structured frames and a hand-crafted contact
/// graph drawn from many addresses.
#[derive(Clone, Debug)]
pub enum Step {
/// A fresh `Contact` event for `addr` (pins no id yet — the first frame does).
Contact(Addr),
/// An explicit `Lost` for `addr` (peer goes offline).
Lost(Addr),
/// Feed raw bytes (possibly garbage, wrong-version) from `addr`. Decoded by
/// the harness exactly as the exchange decodes them, so the drop oracle can
/// predict their fate.
Frame(Addr, Vec<u8>),
/// A well-formed `Obs` frame built from a structured spec.
Obs(ObsSpec),
/// A well-formed `Summary` frame.
Summarize(SpecSummary),
/// An `Ack` frame referencing `ids`.
Ack(AckSpec),
/// Advance the clock by `delta` ms (may be negative; saturates at 0).
Advance(i64),
/// Our own broadcast observation at the current time.
Ourselves,
/// Our own addressed observation to `to` under `lifetime_ms`.
OurselvesTo { to: NodeId, lifetime_ms: u64 },
}
/// Structured fields of an [`Step::Obs`] frame. `from` is the peer's node id and
/// also the obs `observer` (a real peer signs neither in 1.0, but the slot
/// carries it).
#[derive(Clone, Debug)]
pub struct ObsSpec {
pub addr: Addr,
pub from: NodeId,
pub seq: u64,
pub age_ms: u64,
pub topic: Vec<u8>,
pub payload: Vec<u8>,
/// `None` = broadcast; `Some((to, lifetime_ms))` = addressed.
pub dest: Option<(NodeId, u64)>,
}
/// Structured fields of an [`Step::Summarize`] frame.
#[derive(Clone, Debug)]
pub struct SpecSummary {
pub addr: Addr,
/// The peer's declared sender id; keep it stable per `addr` unless testing
/// F6 identity-change (a changing `from` is a counted drop).
pub from: NodeId,
pub salt: u32,
pub bits: Vec<u8>,
}
/// Structured fields of an [`Step::Ack`] frame.
#[derive(Clone, Debug)]
pub struct AckSpec {
pub addr: Addr,
/// Sender id on the Ack frame; defaults to `me` (`[0u8; 16]`) unless a test
/// wants to exercise an identity mismatch (a counted drop).
pub from: NodeId,
pub ids: Vec<ObservationId>,
}
// ---------------------------------------------------------------------------
// Scripted transport (its queues are shared with the harness via `Rc<RefCell>`
// so the harness can push inbound events and read outbound frames while `Node`
// owns its own copy).
// ---------------------------------------------------------------------------
type Inbound = VecDeque<(Addr, Option<Vec<u8>>)>; // None payload == a Contact event
struct Queues {
inbound: RefCell<Inbound>,
lost: RefCell<VecDeque<Addr>>,
outbound: RefCell<Vec<Vec<u8>>>,
}
/// A [`Transport`] whose queues the harness fills and drains directly, so a run
/// is fully scripted: frames arrive in the order we push them, and everything we
/// send is buffered for inspection.
pub struct ScriptedTransport {
mtu: usize,
q: Rc<Queues>,
}
impl Transport for ScriptedTransport {
type Addr = Addr;
/// Pop one queued event per call (the [`Node`] loops until this returns
/// `None`, draining the whole queue in a single poll). A `None` payload is a
/// bare `Contact`; an oversized injected payload never becomes an event, so
/// it is dropped here before the exchange — and the drop oracle must not
/// count frames that never reach it.
fn poll(&mut self) -> Option<Event<Self::Addr>> {
if let Some((addr, raw)) = self.q.inbound.borrow_mut().pop_front() {
match raw {
None => return Some(Event::Contact(addr)),
Some(bytes) => {
// An injected payload larger than MAX_FRAME cannot fit into a
// FrameBuf, so it never reaches the exchange (dropped here).
if bytes.len() <= MAX_FRAME {
let buf: FrameBuf = heapless::Vec::<u8, MAX_FRAME>::from_slice(&bytes)
.expect("len checked below MAX_FRAME");
return Some(Event::Frame(addr, buf));
}
return None;
}
}
}
if let Some(addr) = self.q.lost.borrow_mut().pop_front() {
return Some(Event::Lost(addr));
}
None
}
/// Enforce the MTU: the exchange already clamps before calling `send`, so a
/// `TooLarge` here would signal a clamp regression. The harness's "every sent
/// frame ≤ mtu" check relies on this bound being hard.
fn send(&mut self, _to: &Self::Addr, frame: &[u8]) -> Result<(), SendError> {
if frame.len() > self.mtu {
return Err(SendError::TooLarge);
}
self.q.outbound.borrow_mut().push(frame.to_vec());
Ok(())
}
fn mtu(&self) -> usize {
self.mtu
}
}
// ---------------------------------------------------------------------------
// Harness: one node, its counters, and the post-step checks.
// ---------------------------------------------------------------------------
/// Drives a single [`Node`] over a [`MemStore`] + [`ScriptedTransport`], recording
/// the counts the §7/§10 assertions reason about.
pub struct Harness {
node: Node<MemStore, ScriptedTransport>,
mtu: usize,
q: Rc<Queues>,
now: Tick,
/// Frames that actually reached the exchange (len ≤ `MAX_FRAME`; oversized
/// injected payloads are dropped by the transport before decode). Used as the
/// upper bound for the drop-count reconciliation.
frames_seen: u64,
/// Sum of every `Dropped` count drained so far this run.
dropped_total: u64,
}
/// Serialize a frame into bytes via its public [`Frame::encode`] (version byte +
/// postcard), returning `None` if it cannot fit in a `MAX_FRAME` buffer — the
/// harness never panics on an over-large structured frame; it just skips it.
fn encode_frame(frame: Frame) -> Option<Vec<u8>> {
let mut buf = [0u8; MAX_FRAME];
match frame.encode(&mut buf) {
Ok(n) => Some(buf[..n].to_vec()),
Err(_) => None,
}
}
/// Build a well-formed `Obs` frame from an [`ObsSpec`], clamping the topic and
/// payload to the wire caps first (`broadcast`/`addressed` `.expect()`-panic on
/// oversized input — this harness must not, so it truncates rather than panics).
fn build_obs(spec: &ObsSpec) -> Option<Vec<u8>> {
let topic = &spec.topic[..spec.topic.len().min(MAX_TOPIC)];
let payload = &spec.payload[..spec.payload.len().min(MAX_PAYLOAD)];
let obs = match spec.dest {
None => {
tines_core::observation::Observation::broadcast(spec.from, spec.seq, topic, payload)
}
Some((to, lifetime_ms)) => tines_core::observation::Observation::addressed(
spec.from,
spec.seq,
topic,
payload,
to,
lifetime_ms,
),
};
encode_frame(Frame::Obs {
from: spec.from,
obs,
})
}
/// Build a well-formed `Summary` frame, clamping the bloom bits to
/// `SUMMARY_BYTES`.
fn build_summary(spec: &SpecSummary) -> Option<Vec<u8>> {
let bits: heapless::Vec<u8, SUMMARY_BYTES> =
spec.bits.iter().cloned().take(SUMMARY_BYTES).collect();
encode_frame(Frame::Summary {
from: spec.from,
salt: spec.salt,
bits,
})
}
/// Build a well-formed `Ack` frame, capping the id list at `ACK_IDS_CAP`.
fn build_ack(spec: &AckSpec) -> Option<Vec<u8>> {
let ids: heapless::Vec<ObservationId, ACK_IDS_CAP> =
spec.ids.iter().copied().take(ACK_IDS_CAP).collect();
encode_frame(Frame::Ack {
from: spec.from,
ids,
})
}
/// Push a built frame (or skip it) into the inbound queue for `addr`.
fn push_frame_step(q: &Rc<Queues>, addr: Addr, built: Option<Vec<u8>>) {
if let Some(bytes) = built {
q.inbound.borrow_mut().push_back((addr, Some(bytes)));
}
}
impl Harness {
/// Create a harness for local node id `[0u8; 16]` at `mtu`, clocked at 0.
pub fn new(mtu: usize) -> Self {
let me = [0u8; 16];
let q = Rc::new(Queues {
inbound: RefCell::new(VecDeque::new()),
lost: RefCell::new(VecDeque::new()),
outbound: RefCell::new(Vec::new()),
});
Harness {
node: Node::new(
me,
0,
MemStore::new(),
ScriptedTransport { mtu, q: q.clone() },
),
mtu,
q,
now: 0,
frames_seen: 0,
dropped_total: 0,
}
}
/// The local node id (constant across the run — a NodeId never changes).
pub fn me(&self) -> NodeId {
self.node.id()
}
/// Current simulated time in ms.
pub fn now(&self) -> Tick {
self.now
}
/// The backing store (public accessors let the twin assert on stored ids).
pub fn store(&self) -> &MemStore {
self.node.store()
}
/// Sum of all `Dropped` counts drained so far this run.
pub fn dropped_total(&self) -> u64 {
self.dropped_total
}
/// Frames that actually reached the exchange this round (only counted once a
/// frame fits in the `MAX_FRAME` buffer — oversized injected payloads are
/// dropped by the transport before decode and never reach the exchange).
pub fn frames_seen(&self) -> u64 {
self.frames_seen
}
/// Clone and clear the outbound frames emitted so far — lets a caller count
/// exactly what a phase produced, for the "repeating a summary does not
/// re-send" check.
pub fn sent_frames(&mut self) -> Vec<Vec<u8>> {
let out = self.q.outbound.borrow_mut().drain(..).collect::<Vec<_>>();
out
}
/// Advance the clock by `delta` ms (may be negative; saturates at 0). No poll
/// here — [`run`] drains after every step so each step is processed atomically.
fn advance_clock(&mut self, delta: i64) {
if delta >= 0 {
self.now = self.now.saturating_add(delta as u64);
} else {
// `saturating_neg` keeps `i64::MIN` from panicking the negation.
let d = delta.saturating_neg() as u64;
self.now = self.now.saturating_sub(d);
}
}
/// Drain the transport through one [`Node::poll`] and fold its events into the
/// counters. Runs after every injected step so that a single injected event is
/// processed atomically (the exchange pins an address on first frame and emits
/// at most one reply during that poll). Public so the proptest twin can drive
/// individual steps directly.
pub fn drain(&mut self) {
let _stored = self.node.poll(self.now);
for ev in self.node.drain_events() {
if let tines_core::NodeEvent::Dropped { count, .. } = ev {
self.dropped_total += count as u64;
}
}
}
/// Drive one step: push it onto the transport queues, then drain all queued
/// inbound events through a single [`Node::poll`] (which also emits at most
/// one outbound frame) and collect its events. No panic is caught here — a
/// panic on untrusted bytes is itself the failure the harness hunts for.
pub fn step(&mut self, step: &Step) {
match step {
Step::Contact(addr) => self.q.inbound.borrow_mut().push_back((*addr, None)),
Step::Lost(addr) => self.q.lost.borrow_mut().push_back(*addr),
Step::Frame(addr, raw) => {
if raw.len() <= MAX_FRAME {
self.frames_seen += 1;
self.q
.inbound
.borrow_mut()
.push_back((*addr, Some(raw.clone())));
}
}
Step::Obs(spec) => push_frame_step(&self.q, spec.addr, build_obs(spec)),
Step::Summarize(spec) => push_frame_step(&self.q, spec.addr, build_summary(spec)),
Step::Ack(spec) => push_frame_step(&self.q, spec.addr, build_ack(spec)),
Step::Advance(delta) => self.advance_clock(*delta),
Step::Ourselves => {
let _ = self.node.observe(&[0u8; 4], &[0u8; 16], None, self.now);
self.drain();
}
Step::OurselvesTo { to, lifetime_ms } => {
let _ =
self.node
.observe_to(*to, *lifetime_ms, &[0u8; 4], &[0u8; 8], None, self.now);
self.drain();
}
}
}
/// Drive a full scripted run, exercising the §7 core-safety invariants on
/// every step and once at the end. No panic may escape (a panic on untrusted
/// bytes / adversarial input is itself the bug this harness hunts for); every
/// emitted frame must respect the per-run MTU; and the drop bookkeeping may not
/// overflow its counter.
pub fn run(&mut self, steps: &[Step]) {
// Honest signature round-trip coverage given Observation::non_exhaustive:
// there is no public way to read a stored observation's signature bytes to
// compare against the originals (the type is #[non_exhaustive] with only
// private setters), so we cannot re-derive an expected sig here. Instead we
// exercise the store's OWN signature-serialization contract via the canonical
// conformance suite (`sig_round_trip`) — enabled for this harness through
// tines-core's `conformance` feature, which the fuzz crate turns on. When the
// feature is off (e.g. a plain `cargo test -p tines-core` without it) we fall
// back to driving observations purely through the public observe() API above;
// the lifecycle of those stored observations is asserted by the twin's
// contains()/is_released() checks, not by re-deriving signature bytes.
#[cfg(feature = "conformance")]
{
tines_core::store_conformance::sig_round_trip(MemStore::new);
}
for step in steps {
self.step(step);
self.drain();
let emitted = self.sent_frames();
for f in &emitted {
assert!(
f.len() <= self.mtu,
"fuzz: emitted frame {} exceeds mtu {}",
f.len(),
self.mtu
);
}
}
// §7 drop reconciliation guard: the exchange must never overflow or
// underflow its per-address i32 drop counter. A wrap surfaces as an absurd
// total; real runs produce at most a few thousand drops across thousands of
// steps, so 1e6 is a safe margin that still catches any integer bug.
assert!(
self.dropped_total < 1_000_000,
"fuzz: drop bookkeeping overflowed (dropped_total = {})",
self.dropped_total
);
}
}
/// Byte cursor over fuzz input with bounds-checked field reads; every read
/// returns `None` on a short buffer so decoding stops gracefully rather than
/// panicking on arbitrary bytes.
struct Cursor<'a> {
data: &'a [u8],
pos: usize,
}
impl<'a> Cursor<'a> {
fn new(data: &'a [u8]) -> Self {
Cursor { data, pos: 0 }
}
/// Read a fixed-width little-endian integer of `N` bytes.
fn int<const N: usize>(&mut self) -> Option<[u8; N]> {
let end = self.pos.checked_add(N)?;
if end > self.data.len() {
return None;
}
let mut out = [0u8; N];
out.copy_from_slice(&self.data[self.pos..end]);
self.pos = end;
Some(out)
}
fn u64(&mut self) -> Option<u64> {
let b = self.int::<8>()?;
Some(u64::from_le_bytes(b))
}
fn i64(&mut self) -> Option<i64> {
let b = self.int::<8>()?;
Some(i64::from_le_bytes(b))
}
fn node_id(&mut self) -> Option<[u8; 16]> {
self.int::<16>()
}
/// Read a length-prefixed byte slice (length in `len_ty` bytes).
fn slice_u8_len(&mut self) -> Option<Vec<u8>> {
let len = self.u8()? as usize;
let end = self.pos.checked_add(len)?;
if end > self.data.len() {
return None;
}
let out = self.data[self.pos..end].to_vec();
self.pos = end;
Some(out)
}
fn u8(&mut self) -> Option<u8> {
let b = *self.data.get(self.pos)?;
self.pos += 1;
Some(b)
}
}
/// Decode arbitrary fuzz input into an ordered list of [`Step`]s. Every field is
/// read with bounds-checked slicing, so malformed or truncated input never panics
/// — decoding simply stops at the first position it cannot satisfy. The tag set
/// covers §7 steps (raw `Frame`, `Advance` including backwards ticks) plus the
/// structured frames adversarial mode needs (`Obs` / `Summarize` / `Ack`).
pub fn decode_steps(data: &[u8]) -> Vec<Step> {
let mut steps = Vec::new();
let mut pos = 0usize;
while pos < data.len() {
let tag = data[pos];
pos += 1; // consume the tag byte itself (progress is guaranteed)
match read_step(data, &mut pos, tag) {
Some(step) => steps.push(step),
None => break, // malformed: stop decoding gracefully
}
}
steps
}
/// Read one step's payload fields. The caller has already consumed the tag byte;
/// this consumes only what it needs and returns `None` on a short buffer so that
/// [`decode_steps`] can stop without panicking.
fn read_step(data: &[u8], pos: &mut usize, tag: u8) -> Option<Step> {
let mut c = Cursor::new(data);
c.pos = *pos;
let step = match tag {
// 0 — Contact(addr u64)
0 => Step::Contact(c.u64()?),
// 1 — Lost(addr u64)
1 => Step::Lost(c.u64()?),
// 2 — Frame(addr u64, Vec<u8>)
2 => Step::Frame(c.u64()?, c.slice_u8_len()?),
// 3 — Obs(ObsSpec): addr, from[16], seq(u64), age(u64), topic, payload, dest_kind(0=broadcast,1=addressed{to,u64})
3 => {
let addr = c.u64()?;
let from = c.node_id()?;
let seq = c.u64()?;
let age_ms = c.u64()?;
let topic = c.slice_u8_len()?;
let payload = c.slice_u8_len()?.clone();
match c.u8()? {
0 => Step::Obs(ObsSpec {
addr,
from,
seq,
age_ms,
topic,
payload,
dest: None,
}),
1 => {
let to = c.node_id()?;
let lifetime_ms = c.u64()?;
Step::Obs(ObsSpec {
addr,
from,
seq,
age_ms,
topic,
payload,
dest: Some((to, lifetime_ms)),
})
}
_ => return None,
}
}
// 4 — Summarize(SpecSummary): addr, from[16], salt(u32), bits (u8-prefixed)
4 => {
let addr = c.u64()?;
let from = c.node_id()?;
let salt_bytes = c.int::<4>()?;
let salt = u32::from_le_bytes(salt_bytes);
let bits = c.slice_u8_len()?.clone();
Step::Summarize(SpecSummary {
addr,
from,
salt,
bits,
})
}
// 5 — Ack(AckSpec): addr, ids (u8 count of [u8;16])
5 => {
let addr = c.u64()?;
let n = c.u8()? as usize;
let mut ids = Vec::with_capacity(n);
for _ in 0..n {
ids.push(c.node_id()?);
}
Step::Ack(AckSpec {
addr,
from: [0u8; 16],
ids,
})
}
// 6 — Advance(i64) (negative delta goes backwards; saturates at 0)
6 => Step::Advance(c.i64()?),
// 7 — Ourselves (no fields)
7 => Step::Ourselves,
// 8 — OurselvesTo { to[16], lifetime_ms u64 }
8 => {
let to = c.node_id()?;
let lifetime_ms = c.u64()?;
Step::OurselvesTo { to, lifetime_ms }
}
_ => return None, // unknown tag: stop decoding gracefully
};
*pos = c.pos;
Some(step)
}