From 0fc6333c17fdeafb5b994df4457bd2be7a8af7b3 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Wed, 21 Oct 2026 22:52:47 -0700 Subject: [PATCH] =?UTF-8?q?test(fuzz):=20make=20the=20harness=20check=20wh?= =?UTF-8?q?at=20spec=20=C2=A77=20says=20it=20checks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fuzz oracle and its proptest twin checked much less than §7 claims (S8). The harness now: - records every frame the node attempts to send, before the transport can refuse it, so the "every frame fits mtu()" check can fail; - reads the MTU from its own u16 (input bytes 1-2), mapped into MIN_MTU..=MAX_FRAME, instead of reusing the first step tag; - reconciles drained drop counts per address: VersionMismatch/Malformed drops equal the frames that fail decode (the overflow slot makes up any difference), IdentityChanged drops never exceed frames whose `from` changed, and nothing is counted beyond the frames received; - checks no addressed observation is stored, or forwarded in an Obs frame, past its lifetime; - checks signatures round-trip from the input (an observation stored from a whole Obs frame keeps exactly its signature, and a stored signature never changes), replacing the input-independent sig_round_trip call; - checks every acked id is stored (F1) and bounds the broadcast ids offered per received summary by SEND_QUEUE_CAP (F9); - has a real adversarial Mode (input byte 0): well-formed Obs, Frag, Summary and Ack frames from 3 x MAX_PEERS addresses, with signed and max-size observations (u16 payload lengths), chosen observers (including the node's own id), ages, lifetimes and addressing, structured fragment streams that conflict, repeat or arrive out of order, replayed summaries, `from` values that change within a contact, and Acks from the peer's real id naming ids the node really sent; - adds the §10 peer-table recovery probe, which the fuzz target runs after every adversarial input. fuzz_twin.rs now runs the fuzz entry point on random bytes, generated adversarial steps, and two rewritten properties that were trivially true: summary_repeat_does_not_resend seeds broadcasts and checks a replay re-sends nothing while the next on-time summary re-sends all; peer_table_recovers_after_flood floods the table, checks it is full, and probes recovery after PEER_IDLE_MS. The recovery tests are #[ignore]d until the S1 idle sweep lands, and the fuzz target fails on the first adversarial input until then. A 180 s nightly run with the probe disabled found nothing else. Co-Authored-By: Claude Opus 5.5 --- fuzz/fuzz_targets/node.rs | 21 +- fuzz/src/lib.rs | 1256 ++++++++++++++++++++++----------- tines-core/tests/fuzz_twin.rs | 426 ++++++----- 3 files changed, 1108 insertions(+), 595 deletions(-) diff --git a/fuzz/fuzz_targets/node.rs b/fuzz/fuzz_targets/node.rs index 31968af5..94fc17e0 100644 --- a/fuzz/fuzz_targets/node.rs +++ b/fuzz/fuzz_targets/node.rs @@ -1,19 +1,16 @@ //! cargo-fuzz entry point for the `tines-core` node harness. //! -//! Decodes arbitrary bytes into [`fuzz::Step`]s and drives the shared -//! [`fuzz::Harness`] — the same oracle the proptest twin in -//! `tines-core/tests/fuzz_twin.rs` runs, so any bug caught here is caught there -//! too (and vice versa). +//! [`fuzz::run_input`] reads the mode and MTU from the first three bytes and +//! drives the shared [`fuzz::Harness`] — the same oracle the proptest twin in +//! `tines-core/tests/fuzz_twin.rs` runs, so a property checked here is +//! checked there too. Adversarial-mode inputs also get the §10 peer-table +//! recovery probe. #![no_main] libfuzzer_sys::fuzz_target!(|data: &[u8]| { - let steps = fuzz::decode_steps(data); - // Randomize the MTU per corpus entry so different runs probe different wire - // bounds (the §7 "every sent frame ≤ mtu" check then catches clamps that - // fail at a specific MTU). First byte seeds it; an empty input falls back - // to 128. - let mtu = 128usize + (data.first().copied().unwrap_or(0) as usize) % 512; - if !steps.is_empty() { - fuzz::Harness::new(mtu).run(&steps); + if let Some(mut h) = fuzz::run_input(data) { + if h.mode() == fuzz::Mode::Adversarial { + h.check_peer_table_recovers(); + } } }); diff --git a/fuzz/src/lib.rs b/fuzz/src/lib.rs index f711a863..2399e4e8 100644 --- a/fuzz/src/lib.rs +++ b/fuzz/src/lib.rs @@ -1,127 +1,248 @@ //! 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. +//! [`Step`]s, checking the spec §7 properties after every step. //! //! 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. +//! * the `cargo-fuzz` target (`fuzz/fuzz_targets/node.rs`) via this package's +//! lib, through [`run_input`]; and +//! * the stable proptest twin (`tines-core/tests/fuzz_twin.rs`) via a +//! `#[path]` include, through [`run_input`] on random bytes and through +//! [`Harness`] on generated [`Step`]s. //! -//! 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)] +//! Both entry points converge on [`Harness::run_step`], so a property checked +//! here is checked by both — that is what "same harness" means. +//! +//! # Input format ([`run_input`]) +//! +//! `[mode, mtu_lo, mtu_hi, steps...]`: the low bit of `mode` picks +//! [`Mode::Basic`] or [`Mode::Adversarial`]; the little-endian `u16` maps into +//! `MIN_MTU..=MAX_FRAME` (core's sweep bound up to the wire maximum); the rest +//! decodes into [`Step`]s with [`decode_steps`]. +//! +//! # What is checked after every step +//! +//! * No panic (a panic on untrusted input is itself the failure). +//! * Every frame the node *attempts* to send is no larger than `mtu()` — +//! recorded before the transport can refuse it, so the check can fail. +//! * No addressed observation is stored past its lifetime: each newly stored +//! addressed id has `age_ms <= lifetime_ms`, and every addressed `Obs` +//! frame the node sends does too. +//! * Every rejected frame is counted: per address, the drained +//! `VersionMismatch`/`Malformed` drops equal the frames from that address +//! that fail decode (exactly, unless the overflow counter was used; then +//! the overflow makes up the difference), `IdentityChanged` drops never +//! exceed the frames whose `from` differs from one the address sent +//! before, and nothing is counted beyond the frames an address sent. +//! * Stored signatures round-trip: an observation stored from a whole `Obs` +//! frame (or `observe`) carries exactly the signature it arrived with, and +//! a stored observation's signature never changes afterwards. +//! * Acks only for stored observations (fleet-fixes F1): every id in an Ack +//! the node sends is in its store. +//! * Bounded output per received summary (§10, F9): between two summaries +//! from an address, the node offers it at most `SEND_QUEUE_CAP` distinct +//! broadcast ids. +//! +//! [`Harness::check_peer_table_recovers`] adds the §10 peer-table-recovery +//! probe; the fuzz target runs it after every adversarial-mode input. +// `dead_code`: the twin includes this file as a module and uses only part of +// its API (the fuzz target uses the rest). +#![allow(dead_code, clippy::new_without_default)] use std::cell::RefCell; -use std::collections::VecDeque; +use std::collections::{BTreeMap, BTreeSet, VecDeque}; use std::rc::Rc; use tines_core::conformance; use tines_core::transport::FrameBuf; use tines_core::{ - consts::{ACK_IDS_CAP, MAX_FRAME, MAX_PAYLOAD, MAX_TOPIC, SUMMARY_BYTES}, - Event, MemStore, Node, NodeId, ObservationId, SendError, Tick, Transport, + consts::{ACK_IDS_CAP, MAX_FRAME, MAX_PAYLOAD, MAX_SIG, MAX_TOPIC, SUMMARY_BYTES}, + DropReason, Event, MemStore, Node, NodeEvent, NodeId, Observation, ObservationId, SendError, + SigAlgo, Signature, Store, 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 +/// `Addr` to the `from` 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. +/// The node under test's id. +pub const ME: NodeId = [0u8; 16]; + +/// Polls run after each step, at most: each `Node::poll` sends at most one +/// frame, and the harness stops early once a poll sends nothing. +const POLLS_PER_STEP: usize = 8; + +/// Which input vocabulary a run uses. #[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. + /// Contacts, losses, raw byte frames (mostly garbage), clock steps that + /// may go backwards, and the node's own observations, from a handful of + /// addresses. 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). + /// Well-formed frames built from the input — summaries with chosen bits, + /// Obs and Frag frames with chosen ids, observers (including [`ME`]), + /// seqs, ages, lifetimes, addressing and signatures, structured fragment + /// streams that conflict, repeat or arrive out of order, Acks naming ids + /// the node really sent — from more addresses than `MAX_PEERS`, with + /// `from` values that change within a contact. 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), - /// 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 }, +/// How a step names a node id. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Who { + /// The sending address's identity: the `from` of the first frame the + /// harness sent from it, or a default derived from the address. + Peer, + /// The node under test. + Me, + /// An explicit id. + Id(NodeId), } -/// 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). +/// A signature to put on an observation, not necessarily valid: version `0`, +/// algorithm `0`, or an Ed25519 signature that isn't 64 bytes all fail +/// decode. +#[derive(Clone, Debug)] +pub struct SigSpec { + /// Signature format version. + pub version: u8, + /// Raw algorithm byte (`1` is Ed25519). + pub algo: u8, + /// Signature bytes (truncated to `MAX_SIG`). + pub bytes: Vec, +} + +/// A well-formed observation sent from `addr`. #[derive(Clone, Debug)] pub struct ObsSpec { + /// The sending address. pub addr: Addr, - pub from: NodeId, + /// The frame's `from`. + pub from: Who, + /// The observation's `observer`. + pub observer: Who, + /// Observation sequence number. pub seq: u64, + /// Age the observation arrives with. pub age_ms: u64, + /// Observer wall-clock time. + pub observed_at: Option, + /// Topic bytes (truncated to `MAX_TOPIC`). pub topic: Vec, + /// Payload bytes (truncated to `MAX_PAYLOAD`). pub payload: Vec, /// `None` = broadcast; `Some((to, lifetime_ms))` = addressed. - pub dest: Option<(NodeId, u64)>, + pub dest: Option<(Who, u64)>, + /// Signature, if any. + pub sig: Option, } -/// Structured fields of an [`Step::Summarize`] frame. +/// Fragment order for a [`FragSpec`]. +#[derive(Clone, Debug)] +pub enum FragOrder { + /// Index 0, 1, ... count-1. + InOrder, + /// Index count-1 down to 0. + Reverse, + /// Exactly these indices (may repeat, skip, or exceed the count). + Explicit(Vec), +} + +/// A fragment stream of one observation, cut like a sender would cut it, +/// with optional lies. +#[derive(Clone, Debug)] +pub struct FragSpec { + /// The observation (its `addr`/`from` are the stream's). + pub obs: ObsSpec, + /// Data bytes per fragment (at least 1). + pub chunk: usize, + /// Fragment order. + pub order: FragOrder, + /// Claim this id instead of the observation's real one. + pub id: Option, + /// Claim this fragment count instead of the real one. + pub count: Option, +} + +/// A well-formed `Summary` frame. #[derive(Clone, Debug)] pub struct SpecSummary { + /// The sending address. 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, + /// The frame's `from`. + pub from: Who, + /// Bloom salt. pub salt: u32, + /// Bloom bits (truncated to `SUMMARY_BYTES`). pub bits: Vec, + /// How many identical copies to send back to back (replay). + pub repeat: u8, } -/// Structured fields of an [`Step::Ack`] frame. +/// A well-formed `Ack` frame. #[derive(Clone, Debug)] pub struct AckSpec { + /// The sending address. 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, + /// The frame's `from`. + pub from: Who, + /// Indices into the ids the node has sent to `addr` so far (taken + /// modulo their number; ignored while there are none). + pub sent: Vec, + /// Ids named outright (sent or not). + pub raw: Vec, +} + +/// A single scripted action. +#[derive(Clone, Debug)] +pub enum Step { + /// A `Contact` event for `addr`. + Contact(Addr), + /// A `Lost` event for `addr`. + Lost(Addr), + /// Raw bytes from `addr` (anything; usually not a valid frame). + Frame(Addr, Vec), + /// One well-formed `Obs` frame. + Obs(ObsSpec), + /// A stream of well-formed `Frag` frames. + Frags(FragSpec), + /// One or more identical well-formed `Summary` frames. + Summarize(SpecSummary), + /// One well-formed `Ack` frame. + Ack(AckSpec), + /// Move the clock by `delta` ms (negative goes backwards; saturates at 0). + Advance(i64), + /// The node observes a broadcast with a `size`-byte payload (mod + /// `MAX_PAYLOAD + 1`). + Ourselves { + /// Payload length. + size: u16, + }, + /// The node observes an addressed message. + OurselvesTo { + /// Recipient. + to: Who, + /// Lifetime in ms. + lifetime_ms: u64, + /// Payload length (mod `MAX_PAYLOAD + 1`). + size: u16, + }, } // --------------------------------------------------------------------------- -// Scripted transport (its queues are shared with the harness via `Rc` -// so the harness can push inbound events and read outbound frames while `Node` -// owns its own copy). +// Scripted transport. Its queues are shared with the harness through +// `Rc`, so the harness pushes inbound events and reads every +// attempted send while `Node` owns the transport. // --------------------------------------------------------------------------- -type Inbound = VecDeque<(Addr, Option>)>; // None payload == a Contact event - struct Queues { - inbound: RefCell, - lost: RefCell>, - outbound: RefCell>>, + inbound: RefCell>>, + /// Every frame `send` was called with, oversized ones included. + attempted: RefCell)>>, } -/// 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. +/// A [`Transport`] whose queues the harness fills and drains directly. pub struct ScriptedTransport { mtu: usize, q: Rc, @@ -130,41 +251,17 @@ pub struct ScriptedTransport { 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> { - 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::::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 + self.q.inbound.borrow_mut().pop_front() } - /// 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> { + /// Records the frame *before* checking it, so the harness's MTU check + /// sees an oversized frame even though the send is refused. + fn send(&mut self, to: &Self::Addr, frame: &[u8]) -> Result<(), SendError> { + self.q.attempted.borrow_mut().push((*to, frame.to_vec())); if frame.len() > self.mtu { return Err(SendError::TooLarge); } - self.q.outbound.borrow_mut().push(frame.to_vec()); Ok(()) } @@ -174,102 +271,96 @@ impl Transport for ScriptedTransport { } // --------------------------------------------------------------------------- -// Harness: one node, its counters, and the post-step checks. +// Frame builders (through core's hidden `conformance` encoders). // --------------------------------------------------------------------------- -/// Drives a single [`Node`] over a [`MemStore`] + [`ScriptedTransport`], recording -/// the counts the §7/§10 assertions reason about. -pub struct Harness { - node: Node, - mtu: usize, - q: Rc, - 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 one of core's hidden `conformance` -/// encoders (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(enc: impl FnOnce(&mut [u8]) -> Option) -> Option> { +fn encode_with(enc: impl FnOnce(&mut [u8]) -> Option) -> Option> { let mut buf = [0u8; MAX_FRAME]; enc(&mut buf).map(|n| buf[..n].to_vec()) } -/// 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> { - 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(|out| conformance::encode_obs(spec.from, &obs, out)) -} - -/// Build a well-formed `Summary` frame, clamping the bloom bits to -/// `SUMMARY_BYTES`. -fn build_summary(spec: &SpecSummary) -> Option> { - let bits = &spec.bits[..spec.bits.len().min(SUMMARY_BYTES)]; - encode_frame(|out| conformance::encode_summary(spec.from, spec.salt, bits, out)) -} - -/// Build a well-formed `Ack` frame, capping the id list at `ACK_IDS_CAP`. -fn build_ack(spec: &AckSpec) -> Option> { - let ids: Vec = spec.ids.iter().copied().take(ACK_IDS_CAP).collect(); - encode_frame(|out| conformance::encode_ack(spec.from, &ids, out)) -} - -/// Push a built frame (or skip it) into the inbound queue for `addr`. -fn push_frame_step(q: &Rc, addr: Addr, built: Option>) { - if let Some(bytes) = built { - q.inbound.borrow_mut().push_back((addr, Some(bytes))); +fn sig_algo(raw: u8) -> SigAlgo { + if raw == 1 { + SigAlgo::Ed25519 + } else { + SigAlgo::Unknown(raw) } } +/// Per-address counters and memory the checks reason about. +#[derive(Default)] +struct PeerBook { + /// Frames pushed from this address (all reach the exchange). + frames: u64, + /// Of those, frames that fail decode. + expect_decode_drops: u64, + /// Of those, decodable frames whose `from` differs from one this address + /// sent earlier — an upper bound on `IdentityChanged` drops. + identity_candidates: u64, + /// Distinct decoded `from`s seen from this address. + froms: Vec, + /// Drained `VersionMismatch`/`Malformed` counts. + seen_decode_drops: u64, + /// Drained `IdentityChanged` counts. + seen_identity_drops: u64, + /// Drained counts for any other reason. + seen_other_drops: u64, + /// Ids the node has sent this address (Obs or Frag), in first-send order. + sent: Vec, + /// Broadcast ids offered since this address's last summary. + offered: BTreeSet, +} + +/// Drives a single [`Node`] and checks it after every step. +pub struct Harness { + node: Node, + mode: Mode, + mtu: usize, + q: Rc, + now: Tick, + peers: BTreeMap, + /// Drops counted in the overflow slot (`addr: None`). + overflow_drops: u64, + /// Ids in the store after the last step, with the signature each was + /// first stored with. + stored: BTreeMap>, + /// Observations this step injected whole (one `Obs` frame or an + /// `observe` call): if one of these ids is newly stored, it must carry + /// exactly this signature. + whole: Vec<(ObservationId, Option)>, + /// `Delivered` events drained so far. + delivered: u64, +} + 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]; + /// A harness for node [`ME`] at `mtu`, clocked at 0. + pub fn new(mode: Mode, mtu: usize) -> Self { let q = Rc::new(Queues { inbound: RefCell::new(VecDeque::new()), - lost: RefCell::new(VecDeque::new()), - outbound: RefCell::new(Vec::new()), + attempted: RefCell::new(Vec::new()), }); Harness { node: Node::new( - me, + ME, 0, MemStore::new(), ScriptedTransport { mtu, q: q.clone() }, ), + mode, mtu, q, now: 0, - frames_seen: 0, - dropped_total: 0, + peers: BTreeMap::new(), + overflow_drops: 0, + stored: BTreeMap::new(), + whole: Vec::new(), + delivered: 0, } } - /// The local node id (constant across the run — a NodeId never changes). - pub fn me(&self) -> NodeId { - self.node.id() + /// The input mode this harness was built for. + pub fn mode(&self) -> Mode { + self.mode } /// Current simulated time in ms. @@ -277,303 +368,656 @@ impl Harness { self.now } - /// The backing store (public accessors let the twin assert on stored ids). + /// The backing store. 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 + /// `Delivered` events drained so far. + pub fn delivered(&self) -> u64 { + self.delivered } - /// 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 + /// Distinct broadcast ids the node has offered `addr` since `addr`'s + /// last summary. + pub fn offered_since_summary(&self, addr: Addr) -> usize { + self.peers.get(&addr).map_or(0, |p| p.offered.len()) } - /// 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> { - let out = self.q.outbound.borrow_mut().drain(..).collect::>(); - out + /// The identity a [`Who::Peer`] from `addr` resolves to. + pub fn identity(&self, addr: Addr) -> NodeId { + if let Some(first) = self.peers.get(&addr).and_then(|p| p.froms.first()) { + return *first; + } + let mut id = [0xA0u8; 16]; + id[..8].copy_from_slice(&addr.to_le_bytes()); + id } - /// 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); + fn who(&self, addr: Addr, who: Who) -> NodeId { + match who { + Who::Peer => self.identity(addr), + Who::Me => ME, + Who::Id(id) => id, } } - /// 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; - } - } + fn build_observation(&self, spec: &ObsSpec) -> Observation { + let observer = self.who(spec.addr, spec.observer); + let topic = &spec.topic[..spec.topic.len().min(MAX_TOPIC)]; + let payload = &spec.payload[..spec.payload.len().min(MAX_PAYLOAD)]; + let mut obs = match spec.dest { + None => Observation::broadcast(observer, spec.seq, topic, payload), + Some((to, lifetime_ms)) => Observation::addressed( + observer, + spec.seq, + topic, + payload, + self.who(spec.addr, to), + lifetime_ms, + ), + }; + obs.age_ms = spec.age_ms; + obs.observed_at = spec.observed_at; + obs.sig = spec.sig.as_ref().and_then(|s| { + let bytes = &s.bytes[..s.bytes.len().min(MAX_SIG)]; + conformance::signature(s.version, sig_algo(s.algo), bytes) + }); + obs } - /// 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()))); + /// Queue `bytes` from `addr` and record what the exchange must do with + /// it. Frames over `MAX_FRAME` can't be a transport event, so they are + /// skipped. + fn push(&mut self, addr: Addr, bytes: Vec) { + let Ok(buf) = FrameBuf::from_slice(&bytes) else { + return; + }; + let book = self.peers.entry(addr).or_default(); + book.frames += 1; + match conformance::decoded_sender(&bytes) { + None => book.expect_decode_drops += 1, + Some(from) => { + if book.froms.iter().any(|f| *f != from) { + book.identity_candidates += 1; + } + if !book.froms.contains(&from) { + book.froms.push(from); + } + if conformance::is_summary(&bytes) { + book.offered.clear(); } } - 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(); + } + self.q + .inbound + .borrow_mut() + .push_back(Event::Frame(addr, buf)); + } + + /// Queue one step's input (without polling the node). + pub fn step(&mut self, step: &Step) { + match step { + Step::Contact(addr) => self.q.inbound.borrow_mut().push_back(Event::Contact(*addr)), + Step::Lost(addr) => self.q.inbound.borrow_mut().push_back(Event::Lost(*addr)), + Step::Frame(addr, raw) => { + if let Some(obs) = conformance::obs_of(raw) { + self.whole.push((obs.id(), obs.sig.clone())); + } + self.push(*addr, raw.clone()); } - Step::OurselvesTo { to, lifetime_ms } => { - let _ = + Step::Obs(spec) => { + let from = self.who(spec.addr, spec.from); + let obs = self.build_observation(spec); + if let Some(bytes) = encode_with(|out| conformance::encode_obs(from, &obs, out)) { + if conformance::decoded_sender(&bytes).is_some() { + self.whole.push((obs.id(), obs.sig.clone())); + } + self.push(spec.addr, bytes); + } + } + Step::Frags(spec) => self.push_frags(spec), + Step::Summarize(spec) => { + let from = self.who(spec.addr, spec.from); + let bits = &spec.bits[..spec.bits.len().min(SUMMARY_BYTES)]; + if let Some(bytes) = + encode_with(|out| conformance::encode_summary(from, spec.salt, bits, out)) + { + for _ in 0..spec.repeat.max(1) { + self.push(spec.addr, bytes.clone()); + } + } + } + Step::Ack(spec) => { + let from = self.who(spec.addr, spec.from); + let sent = self + .peers + .get(&spec.addr) + .map(|p| p.sent.clone()) + .unwrap_or_default(); + let mut ids: Vec = Vec::new(); + if !sent.is_empty() { + ids.extend(spec.sent.iter().map(|i| sent[*i as usize % sent.len()])); + } + ids.extend(spec.raw.iter().copied()); + ids.truncate(ACK_IDS_CAP); + if let Some(bytes) = encode_with(|out| conformance::encode_ack(from, &ids, out)) { + self.push(spec.addr, bytes); + } + } + Step::Advance(delta) => { + self.now = if *delta >= 0 { + self.now.saturating_add(delta.unsigned_abs()) + } else { + self.now.saturating_sub(delta.unsigned_abs()) + }; + } + Step::Ourselves { size } => { + let payload = vec![0x5A; *size as usize % (MAX_PAYLOAD + 1)]; + if let Ok(id) = self.node.observe(b"self", &payload, None, self.now) { + self.whole.push((id, None)); + } + } + Step::OurselvesTo { + to, + lifetime_ms, + size, + } => { + let to = match to { + Who::Peer => [0xA0; 16], + Who::Me => ME, + Who::Id(id) => *id, + }; + let payload = vec![0xA5; *size as usize % (MAX_PAYLOAD + 1)]; + if let Ok(id) = self.node - .observe_to(*to, *lifetime_ms, &[0u8; 4], &[0u8; 8], None, self.now); - self.drain(); + .observe_to(to, *lifetime_ms, b"self/to", &payload, None, self.now) + { + self.whole.push((id, None)); + } } } } - /// 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 - ); + fn push_frags(&mut self, spec: &FragSpec) { + let from = self.who(spec.obs.addr, spec.obs.from); + let obs = self.build_observation(&spec.obs); + let mut buf = vec![0u8; MAX_FRAME * 2]; + let Some(len) = conformance::encode_observation(&obs, &mut buf) else { + return; + }; + let bytes = &buf[..len]; + let chunk = spec.chunk.clamp(1, MAX_FRAME); + let real_count = len.div_ceil(chunk).max(1); + let count = spec.count.unwrap_or(real_count.min(u8::MAX as usize) as u8); + let id = spec.id.unwrap_or_else(|| obs.id()); + let order: Vec = match &spec.order { + FragOrder::InOrder => (0..count).collect(), + FragOrder::Reverse => (0..count).rev().collect(), + FragOrder::Explicit(v) => v.clone(), + }; + for index in order { + let start = (index as usize).saturating_mul(chunk).min(len); + let end = start.saturating_add(chunk).min(len); + let data = &bytes[start..end]; + if let Some(frame) = + encode_with(|out| conformance::encode_frag(from, id, index, count, data, out)) + { + self.push(spec.obs.addr, frame); } } + } - // §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. + /// Poll the node until it stops sending (at most `POLLS_PER_STEP` + /// times), folding drained events into the counters and checking every + /// attempted send. + pub fn pump(&mut self) { + for _ in 0..POLLS_PER_STEP { + let before = self.q.attempted.borrow().len(); + let _ = self.node.poll(self.now); + self.drain_events(); + if self.q.attempted.borrow().len() == before && self.q.inbound.borrow().is_empty() { + break; + } + } + self.check_sends(); + } + + fn drain_events(&mut self) { + for ev in self.node.drain_events() { + match ev { + NodeEvent::Dropped { + addr: Some(addr), + reason, + count, + } => { + let book = self.peers.entry(addr).or_default(); + let count = u64::from(count); + match reason { + DropReason::VersionMismatch { .. } | DropReason::Malformed => { + book.seen_decode_drops += count + } + DropReason::IdentityChanged { .. } => book.seen_identity_drops += count, + _ => book.seen_other_drops += count, + } + } + NodeEvent::Dropped { + addr: None, count, .. + } => self.overflow_drops += u64::from(count), + NodeEvent::Delivered { .. } => self.delivered += 1, + _ => {} + } + } + } + + /// The MTU, Ack-only-for-stored, forwarded-lifetime and per-summary + /// output checks over every frame attempted since the last call. + fn check_sends(&mut self) { + let attempted: Vec<(Addr, Vec)> = self.q.attempted.borrow_mut().drain(..).collect(); + for (to, frame) in attempted { + assert!( + frame.len() <= self.mtu, + "sent a {}-byte frame over mtu {}", + frame.len(), + self.mtu + ); + let store = self.node.store(); + let mut acked_unstored = None; + if conformance::ack_of(&frame, &mut |id| { + if !store.contains(id) { + acked_unstored = Some(*id); + } + }) { + assert_eq!( + acked_unstored, None, + "acked an id the store does not hold (F1)" + ); + continue; + } + let offer = if let Some(obs) = conformance::obs_of(&frame) { + if let Some(dest) = &obs.dest { + assert!( + obs.age_ms <= dest.lifetime_ms, + "forwarded an addressed observation past its lifetime \ + (age {} > lifetime {})", + obs.age_ms, + dest.lifetime_ms + ); + } + Some((obs.id(), obs.dest.is_none())) + } else if let Some((id, index, _)) = conformance::frag_of(&frame) { + let broadcast = + index == 0 && store.get(&id, self.now).is_some_and(|o| o.dest.is_none()); + Some((id, broadcast)) + } else { + None + }; + if let Some((id, broadcast)) = offer { + let book = self.peers.entry(to).or_default(); + if !book.sent.contains(&id) { + book.sent.push(id); + } + if broadcast { + book.offered.insert(id); + assert!( + book.offered.len() <= conformance::SEND_QUEUE_CAP, + "offered {} broadcast ids to {to} on one summary (F9 bound {})", + book.offered.len(), + conformance::SEND_QUEUE_CAP + ); + } + } + } + } + + /// Every rejected frame is counted (§7), per address. + fn check_drops(&self) { + let mut expect_decode = 0u64; + let mut seen_decode = 0u64; + let mut seen_all = 0u64; + let mut frames = 0u64; + for (addr, b) in &self.peers { + assert!( + b.seen_decode_drops <= b.expect_decode_drops, + "{addr}: {} decode drops counted, only {} frames failed decode", + b.seen_decode_drops, + b.expect_decode_drops + ); + assert!( + b.seen_identity_drops <= b.identity_candidates, + "{addr}: {} IdentityChanged drops counted, only {} frames changed `from`", + b.seen_identity_drops, + b.identity_candidates + ); + let all = b.seen_decode_drops + b.seen_identity_drops + b.seen_other_drops; + assert!( + all <= b.frames, + "{addr}: {all} drops counted for {} frames", + b.frames + ); + if self.overflow_drops == 0 { + assert_eq!( + b.seen_decode_drops, b.expect_decode_drops, + "{addr}: a frame that failed decode was not counted" + ); + } + expect_decode += b.expect_decode_drops; + seen_decode += b.seen_decode_drops; + seen_all += all; + frames += b.frames; + } assert!( - self.dropped_total < 1_000_000, - "fuzz: drop bookkeeping overflowed (dropped_total = {})", - self.dropped_total + expect_decode <= seen_decode + self.overflow_drops, + "{expect_decode} frames failed decode, only {seen_decode} counted plus {} overflow", + self.overflow_drops + ); + assert!( + seen_all + self.overflow_drops <= frames, + "more drops counted than frames received" + ); + } + + /// Lifetime and signature checks over the store. + fn check_store(&mut self) { + let mut ids = Vec::new(); + self.node.store().for_each_id(&mut |id| ids.push(*id)); + let whole = std::mem::take(&mut self.whole); + for id in ids { + let obs = self + .node + .store() + .get(&id, self.now) + .expect("for_each_id yields stored ids"); + match self.stored.get(&id) { + Some(first) => { + assert_eq!(&obs.sig, first, "a stored observation's signature changed") + } + None => { + if let Some(dest) = &obs.dest { + assert!( + obs.age_ms <= dest.lifetime_ms, + "stored an addressed observation past its lifetime \ + (age {} > lifetime {})", + obs.age_ms, + dest.lifetime_ms + ); + } + if let Some((_, sig)) = whole.iter().find(|(w, _)| *w == id) { + assert_eq!( + &obs.sig, sig, + "a stored observation's signature did not round-trip" + ); + } + self.stored.insert(id, obs.sig.clone()); + } + } + } + } + + /// Run one step and every check. + pub fn run_step(&mut self, step: &Step) { + self.step(step); + self.pump(); + self.check_drops(); + self.check_store(); + } + + /// Run every step, checking after each. + pub fn run(&mut self, steps: &[Step]) { + for step in steps { + self.run_step(step); + } + } + + /// §10 peer-table recovery (F8): after `PEER_IDLE_MS` of silence from + /// every peer, a frame from a never-seen address still gets a slot, so + /// the node stores the observation it carries. + pub fn check_peer_table_recovers(&mut self) { + self.now = self.now.saturating_add(conformance::PEER_IDLE_MS + 1); + self.pump(); + let addr = Addr::MAX; + let obs = Observation::broadcast([0xEE; 16], u64::MAX, b"recovery", b"probe"); + let id = obs.id(); + let already = self.node.store().contains(&id); + let bytes = encode_with(|out| conformance::encode_obs([0xEE; 16], &obs, out)) + .expect("a small Obs frame fits MAX_FRAME"); + self.push(addr, bytes); + self.pump(); + assert!( + already || self.node.store().contains(&id), + "after {} ms of silence a new address still got no peer slot (F8)", + conformance::PEER_IDLE_MS + 1 ); } } -/// 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. +// --------------------------------------------------------------------------- +// Input decoding. +// --------------------------------------------------------------------------- + +/// Bounds-checked reader over fuzz input; every read returns `None` once the +/// input runs out, so decoding stops instead of panicking. 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(&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 { - let b = self.int::<8>()?; - Some(u64::from_le_bytes(b)) - } - - fn i64(&mut self) -> Option { - 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> { - 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(); + fn take(&mut self, n: usize) -> Option<&'a [u8]> { + let end = self.pos.checked_add(n)?; + let out = self.data.get(self.pos..end)?; self.pos = end; Some(out) } fn u8(&mut self) -> Option { - let b = *self.data.get(self.pos)?; - self.pos += 1; - Some(b) + self.take(1).map(|b| b[0]) + } + + fn u16(&mut self) -> Option { + self.take(2).map(|b| u16::from_le_bytes([b[0], b[1]])) + } + + fn u32(&mut self) -> Option { + let b = self.take(4)?; + Some(u32::from_le_bytes(b.try_into().ok()?)) + } + + fn u64(&mut self) -> Option { + let b = self.take(8)?; + Some(u64::from_le_bytes(b.try_into().ok()?)) + } + + fn id(&mut self) -> Option<[u8; 16]> { + let b = self.take(16)?; + b.try_into().ok() + } + + /// `Who`: 0 = the peer, 1 = the node under test, else 16 explicit bytes. + fn who(&mut self) -> Option { + Some(match self.u8()? % 3 { + 0 => Who::Peer, + 1 => Who::Me, + _ => Who::Id(self.id()?), + }) + } + + /// A `len`-byte run of one fill byte: cheap max-size fields. + fn filled(&mut self, len: usize) -> Option> { + let fill = self.u8()?; + Some(vec![fill; len]) } } -/// 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 { - 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 +/// How many addresses each mode draws from: basic stays small so contacts +/// interact; adversarial goes well past `MAX_PEERS`. +fn addr_of(mode: Mode, byte: u8) -> Addr { + match mode { + Mode::Basic => Addr::from(byte % 4), + Mode::Adversarial => Addr::from(byte) % (3 * conformance::MAX_PEERS as Addr), + } +} + +fn read_obs(c: &mut Cursor, mode: Mode) -> Option { + let addr = addr_of(mode, c.u8()?); + let from = c.who()?; + let observer = c.who()?; + let seq = c.u64()?; + let age_ms = u64::from(c.u32()?); + let observed_at = match c.u8()? % 2 { + 0 => None, + _ => Some(c.u64()?), + }; + let topic_len = usize::from(c.u8()?) % (MAX_TOPIC + 1); + let topic = c.filled(topic_len)?; + let payload_len = usize::from(c.u16()?) % (MAX_PAYLOAD + 1); + let payload = c.filled(payload_len)?; + let dest = match c.u8()? % 2 { + 0 => None, + _ => Some((c.who()?, u64::from(c.u32()?))), + }; + let sig = match c.u8()? % 2 { + 0 => None, + _ => { + let version = c.u8()?; + let algo = c.u8()?; + let len = usize::from(c.u8()?) % (MAX_SIG + 1); + Some(SigSpec { + version, + algo, + bytes: c.filled(len)?, + }) } - } - steps + }; + Some(ObsSpec { + addr, + from, + observer, + seq, + age_ms, + observed_at, + topic, + payload, + dest, + sig, + }) } -/// 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 { - 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) - 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)), - }) +fn read_step(c: &mut Cursor, mode: Mode) -> Option { + let common = 7u8; + let tags = match mode { + Mode::Basic => common, + Mode::Adversarial => common + 4, + }; + Some(match c.u8()? % tags { + 0 => Step::Contact(addr_of(mode, c.u8()?)), + 1 => Step::Lost(addr_of(mode, c.u8()?)), + // Small clock steps (±~524 s, in 16 ms units), either direction. + 2 => Step::Advance(i64::from(c.u16()? as i16) * 16), + 3 => Step::Advance(c.u64()? as i64), + 4 => Step::Ourselves { size: c.u16()? }, + 5 => Step::OurselvesTo { + to: c.who()?, + lifetime_ms: u64::from(c.u32()?), + size: c.u16()?, + }, + 6 => { + let addr = addr_of(mode, c.u8()?); + let len = usize::from(c.u16()?) % (MAX_FRAME + 1); + let len = len.min(c.data.len() - c.pos); + Step::Frame(addr, c.take(len)?.to_vec()) + } + 7 => Step::Obs(read_obs(c, mode)?), + 8 => { + let obs = read_obs(c, mode)?; + let chunk = usize::from(c.u16()?); + let order = match c.u8()? % 3 { + 0 => FragOrder::InOrder, + 1 => FragOrder::Reverse, + _ => { + let n = usize::from(c.u8()? % 16); + FragOrder::Explicit(c.take(n)?.to_vec()) } - _ => return None, - } + }; + let id = match c.u8()? % 4 { + 0 => Some(c.id()?), + _ => None, + }; + let count = match c.u8()? { + 0 => None, + n => Some(n), + }; + Step::Frags(FragSpec { + obs, + chunk, + order, + id, + count, + }) } - // 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(); + 9 => { + let addr = addr_of(mode, c.u8()?); + let from = c.who()?; + let salt = c.u32()?; + let len = usize::from(c.u16()?) % (SUMMARY_BYTES + 1); + let bits = match c.u8()? % 3 { + 0 => vec![0x00; len], + 1 => vec![0xFF; len], + _ => c.filled(len)?, + }; Step::Summarize(SpecSummary { addr, from, salt, bits, + repeat: c.u8()? % 4 + 1, }) } - // 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()?); - } + _ => { + let addr = addr_of(mode, c.u8()?); + let from = c.who()?; + let n = c.u8()? % 16; + let sent = (0..n).map(|_| c.u16()).collect::>>()?; + let r = c.u8()? % 4; + let raw = (0..r).map(|_| c.id()).collect::>>()?; Step::Ack(AckSpec { addr, - from: [0u8; 16], - ids, + from, + sent, + raw, }) } - // 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 } + }) +} + +/// Decode `data` into steps for `mode`, stopping at the first step the input +/// can't complete. Never panics. +pub fn decode_steps(mode: Mode, data: &[u8]) -> Vec { + let mut c = Cursor { data, pos: 0 }; + let mut steps = Vec::new(); + while c.pos < data.len() { + match read_step(&mut c, mode) { + Some(step) => steps.push(step), + None => break, } - _ => return None, // unknown tag: stop decoding gracefully + } + steps +} + +/// Decode a whole fuzz input (see the module docs for the format) and run it, +/// returning the harness for further checks. `None` if the input is too short +/// to name a mode and MTU. +pub fn run_input(data: &[u8]) -> Option { + let (&mode, rest) = data.split_first()?; + let (mtu, rest) = rest.split_first_chunk::<2>()?; + let mode = if mode & 1 == 0 { + Mode::Basic + } else { + Mode::Adversarial }; - *pos = c.pos; - Some(step) + let span = MAX_FRAME - conformance::MIN_MTU + 1; + let mtu = conformance::MIN_MTU + usize::from(u16::from_le_bytes(*mtu)) % span; + let mut h = Harness::new(mode, mtu); + h.run(&decode_steps(mode, rest)); + Some(h) } diff --git a/tines-core/tests/fuzz_twin.rs b/tines-core/tests/fuzz_twin.rs index 2cdb0d3c..8d526c2a 100644 --- a/tines-core/tests/fuzz_twin.rs +++ b/tines-core/tests/fuzz_twin.rs @@ -1,227 +1,299 @@ -//! Proptest twin of the `fuzz/` harness (Task 12). +//! Proptest twin of the `fuzz/` harness (spec §7 "Stable twin"). //! -//! The cargo-fuzz crate at `fuzz/` drives this very same module through a -//! scripted [`Node`]/[`MemStore`] transport, decoding arbitrary fuzzed bytes -//! into ordered steps and asserting no-panic / MTU-bounded output / drop-count -//! reconciliation (§7) plus the §10 adversarial-peer properties. The twin -//! reuses that *exact* harness via a `#[path]` include so any bug fixed in the -//! oracle is caught here too — "same harness", two entry points, not two copies. +//! The cargo-fuzz crate at `fuzz/` drives a scripted `Node` + `MemStore` +//! through decoded steps and checks the §7 properties after each one. The +//! twin compiles that *exact* harness in through a `#[path]` include, so it is +//! one oracle with two entry points, not two copies: //! -//! This is the bounded-case-count proptest half of Task 12: it runs under the -//! normal suite (`cargo test -p tines-core`) and needs no nightly toolchain. -//! When `tines-core` is built with the optional `conformance` feature, the -//! harness's own signature round-trip also runs (see its source); otherwise the -//! twin covers store lifecycle through the public `observe()` path instead. +//! * `fuzz_inputs` feeds random bytes through `fuzz::run_input`, the fuzz +//! target's own entry point, in both modes; +//! * `structured_*` feed generated steps straight to the harness, so every +//! case reaches well-formed frames; +//! * the rest pin specific §10 properties as scenarios. //! -//! The harness drives `MemStore`, which needs `alloc`; the twin is skipped in -//! `--no-default-features` builds (like `pair.rs`). -#![cfg(feature = "alloc")] +//! It runs on the pinned stable toolchain in the normal gate. It needs the +//! `conformance` feature (the harness builds frames through core's hidden +//! `conformance` helpers) and `alloc` (`MemStore`); Cargo.toml's +//! `required-features` skips it otherwise. #![allow(clippy::needless_range_loop)] #[path = "../../fuzz/src/lib.rs"] mod fuzz; +use fuzz::{AckSpec, FragOrder, FragSpec, Harness, Mode, ObsSpec, SpecSummary, Step, Who}; use proptest::prelude::*; -use tines_core::conformance::MAX_PEERS; +use tines_core::conformance::{MAX_PEERS, MIN_MTU, PEER_IDLE_MS, SUMMARY_INTERVAL_MS}; use tines_core::consts::{MAX_FRAME, MAX_PAYLOAD, MAX_TOPIC, SUMMARY_BYTES}; use tines_core::Store; -// A single bounded case count keeps the twin fast under `cargo test` while -// still exercising a meaningful surface of the harness. -const CASES: u32 = 32; +const CASES: u32 = 64; -fn node_id() -> impl Strategy { - any::<[u8; 16]>() +fn mtu() -> impl Strategy { + MIN_MTU..=MAX_FRAME } -fn topic() -> impl Strategy> { - proptest::collection::vec(any::(), 0..=MAX_TOPIC) +fn who() -> impl Strategy { + prop_oneof![ + 6 => Just(Who::Peer), + 1 => Just(Who::Me), + 1 => any::<[u8; 16]>().prop_map(Who::Id), + ] } -fn payload() -> impl Strategy> { - proptest::collection::vec(any::(), 0..MAX_PAYLOAD) +/// Addresses past `MAX_PEERS`, so the peer table overflows. +fn addr() -> impl Strategy { + 0..(3 * MAX_PEERS as u64) } -// A well-formed broadcast observation spec. `dest` stays broadcast (None): the -// addressed path shares storage/exchange logic and is exercised by the raw -// Frame step below, so broadcasting keeps the twin's invariants unambiguous. -fn obs_spec() -> impl Strategy { - ( - any::(), // addr - node_id(), // from - any::(), // seq - any::(), // age_ms - topic(), - payload(), +fn sig() -> impl Strategy> { + proptest::option::of( + ( + prop_oneof![Just(1u8), any::()], + prop_oneof![Just(1u8), any::()], + prop_oneof![Just(64usize), 0usize..=64], + ) + .prop_map(|(version, algo, len)| fuzz::SigSpec { + version, + algo, + bytes: vec![0x5A; len], + }), ) - .prop_map(|(addr, from, seq, age_ms, topic, payload)| fuzz::ObsSpec { - addr, - from, - seq, - age_ms: age_ms as u64, - topic, - payload, - dest: None, - }) } -// A well-formed summary spec with a chosen bits pattern and salt. -fn summary_spec() -> impl Strategy { +fn obs_spec() -> impl Strategy { ( - any::(), - node_id(), - any::(), - proptest::collection::vec(any::(), 0..SUMMARY_BYTES), + (addr(), who(), who(), 0u64..64, any::()), + ( + proptest::option::of(any::()), + prop_oneof![Just(MAX_TOPIC), 0..=MAX_TOPIC], + prop_oneof![Just(MAX_PAYLOAD), 0..=MAX_PAYLOAD], + proptest::option::of((who(), any::())), + sig(), + ), ) - .prop_map(|(addr, from, salt, bits)| fuzz::SpecSummary { - addr, - from, - salt, - bits, - }) + .prop_map( + |((addr, from, observer, seq, age), (observed_at, tl, pl, dest, sig))| ObsSpec { + addr, + from, + observer, + seq, + age_ms: u64::from(age), + observed_at, + topic: vec![b't'; tl], + payload: vec![b'p'; pl], + dest: dest.map(|(to, life)| (to, u64::from(life))), + sig, + }, + ) } -// An Ack spec for 0..=4 ids (sent or unsent). -fn ack_spec() -> impl Strategy { - (any::(), proptest::collection::vec(node_id(), 0..=8)).prop_map(|(addr, ids)| { - fuzz::AckSpec { - addr, - from: [0u8; 16], - ids, - } +fn structured_step() -> impl Strategy { + prop_oneof![ + obs_spec().prop_map(Step::Obs), + ( + obs_spec(), + 1usize..=MAX_FRAME, + prop_oneof![ + Just(FragOrder::InOrder), + Just(FragOrder::Reverse), + proptest::collection::vec(any::(), 0..8).prop_map(FragOrder::Explicit), + ], + proptest::option::of(any::<[u8; 16]>()), + proptest::option::of(any::()), + ) + .prop_map(|(obs, chunk, order, id, count)| Step::Frags(FragSpec { + obs, + chunk, + order, + id, + count, + })), + ( + addr(), + who(), + any::(), + 0..=SUMMARY_BYTES, + any::(), + 1u8..=3 + ) + .prop_map(|(addr, from, salt, len, fill, repeat)| Step::Summarize( + SpecSummary { + addr, + from, + salt, + bits: vec![fill; len], + repeat, + } + )), + ( + addr(), + who(), + proptest::collection::vec(any::(), 0..8), + proptest::collection::vec(any::<[u8; 16]>(), 0..2), + ) + .prop_map(|(addr, from, sent, raw)| Step::Ack(AckSpec { + addr, + from, + sent, + raw, + })), + addr().prop_map(Step::Contact), + addr().prop_map(Step::Lost), + prop_oneof![-5_000i64..=40_000, any::()].prop_map(Step::Advance), + any::().prop_map(|size| Step::Ourselves { size }), + (who(), any::(), any::()).prop_map(|(to, life, size)| Step::OurselvesTo { + to, + lifetime_ms: u64::from(life), + size, + }), + (addr(), proptest::collection::vec(any::(), 0..64)) + .prop_map(|(a, bytes)| Step::Frame(a, bytes)), + ] +} + +/// A broadcast observation from the peer at `addr`, distinct per `seq`. +fn peer_obs(addr: u64, seq: u64) -> ObsSpec { + ObsSpec { + addr, + from: Who::Peer, + observer: Who::Peer, + seq, + age_ms: 0, + observed_at: None, + topic: b"t".to_vec(), + payload: b"p".to_vec(), + dest: None, + sig: None, + } +} + +/// A summary from `addr` saying it holds nothing. +fn empty_summary(addr: u64) -> Step { + Step::Summarize(SpecSummary { + addr, + from: Who::Peer, + salt: 7, + bits: vec![0; SUMMARY_BYTES], + repeat: 1, }) } -// Any valid Step of any kind — drives the Basic mode with real, well-formed -// inputs rather than raw random bytes. -fn any_step() -> impl Strategy { - prop_oneof![ - // Addressed / broadcast observations (well-formed Obs frames). - obs_spec().prop_map(fuzz::Step::Obs), - // Summaries with chosen bits patterns and salts. - summary_spec().prop_map(fuzz::Step::Summarize), - // Acks for sent/unsent ids. - ack_spec().prop_map(fuzz::Step::Ack), - // Raw Frame: a peer (or our own) byte stream of up to MAX_FRAME bytes, - // including deliberately malformed content to exercise the drop path. - ( - any::(), - proptest::collection::vec(any::(), 0..MAX_FRAME), - ) - .prop_map(|(addr, bytes)| fuzz::Step::Frame(addr, bytes)), - // Contact / lost + clock advance (may go backwards). - any::().prop_map(fuzz::Step::Contact), - any::().prop_map(fuzz::Step::Lost), - any::().prop_map(fuzz::Step::Advance), - // Ourselves / ours-to (self observation). - (0u32..=0).prop_map(|_| fuzz::Step::Ourselves), - ] +/// Poll the harness `n` times at the current tick. +fn settle(h: &mut Harness, n: usize) { + for _ in 0..n { + h.pump(); + } } proptest! { #![proptest_config(ProptestConfig { cases: CASES, ..Default::default() })] - /// §7 core safety: an arbitrary sequence of well-formed steps run through a - /// randomised-MTU harness must never panic, must stay within the MTU on - /// every emitted frame, and must not overflow its drop counter. `run` - /// asserts all three; reaching it clean is the property. A single mixed - /// sequence may also exceed MAX_PEERS distinct addresses (addr is a full - /// u64), so this alone exercises the flood path without panic. + /// The fuzz target's own entry point on random bytes: both modes, a + /// random MTU, every §7 check after every step. #[test] - fn basic_run_no_panic_or_overflow( - steps in proptest::collection::vec(any_step(), 0..=128), - mtu in 32usize..=4096usize, + fn fuzz_inputs(data in proptest::collection::vec(any::(), 3..2048)) { + let _ = fuzz::run_input(&data); + } + + /// Generated well-formed adversarial steps from more than `MAX_PEERS` + /// addresses, every §7 check after every step. + #[test] + fn structured_adversarial( + steps in proptest::collection::vec(structured_step(), 1..=64), + mtu in mtu(), ) { - if !steps.is_empty() { - let mut h = fuzz::Harness::new(mtu); - h.run(&steps); - } + Harness::new(Mode::Adversarial, mtu).run(&steps); } - /// §10 idempotency: repeating an identical summary must not emit MORE - /// frames than the first time it was seen. + /// §10 / F9: a summary replayed within the interval re-sends nothing, + /// while the next on-time summary re-offers everything the peer still + /// lacks (lossy convergence is preserved). #[test] - fn summary_repeat_does_not_resend(spec in summary_spec(), mtu in 32usize..=4096) { - let mut h = fuzz::Harness::new(mtu); - - h.step(&fuzz::Step::Summarize(spec.clone())); - h.drain(); - let first = h.sent_frames().len(); - - h.step(&fuzz::Step::Summarize(spec)); - h.drain(); - let second = h.sent_frames().len(); - - prop_assert!(second <= first, "repeating a summary must not re-send ({} vs {} frames)", second, first); - } - - /// §10 flood recovery: feeding MORE than MAX_PEERS distinct-identity - /// summaries through the transport must never panic and output stays - /// bounded per received summary. - #[test] - fn flood_past_max_peers_does_not_panic(mtu in 32usize..=4096) { - let n = MAX_PEERS as u64 + 16; // deliberately beyond the cap - let mut steps = Vec::new(); - for i in 1..=n { - steps.push(fuzz::Step::Summarize(fuzz::SpecSummary { - addr: i, - from: [i as u8; 16], - salt: (i as u32), - bits: (0..SUMMARY_BYTES).map(|_| i as u8).collect(), - })); + fn summary_repeat_does_not_resend(n in 1usize..=12, mtu in 600usize..=MAX_FRAME) { + let mut h = Harness::new(Mode::Adversarial, mtu); + for _ in 0..n { + h.run_step(&Step::Ourselves { size: 16 }); } - let mut h = fuzz::Harness::new(mtu); - h.run(&steps); - } + prop_assert_eq!(h.store().len(), n); - /// §10 peer-table recovery after a beyond-MAX_PEERS flood: once the table - /// has been pushed past its cap, a fresh contact + our own observation must - /// still be stored (the node keeps working rather than wedging). - #[test] - fn peer_table_recovers_after_flood(mtu in 32usize..=4096) { - let mut h = fuzz::Harness::new(mtu); - for i in 1..=(MAX_PEERS as u64 + 16) { - let spec = fuzz::SpecSummary { - addr: i, - from: [i as u8; 16], - salt: (i as u32), - bits: (0..SUMMARY_BYTES).map(|_| i as u8).collect(), - }; - h.step(&fuzz::Step::Summarize(spec)); - h.drain(); - } - // A fresh contact + our own observation after the flood. - h.step(&fuzz::Step::Contact(1)); - h.step(&fuzz::Step::Ourselves); - h.drain(); - prop_assert!(!h.store().is_empty(), "node must still store an observation after a beyond-MAX_PEERS flood"); + // The peer says it holds nothing: every broadcast is offered once. + h.run_step(&empty_summary(1)); + settle(&mut h, n + 4); + prop_assert_eq!(h.offered_since_summary(1), n); + + // The same summary again, 1 ms later: ignored, nothing re-sent. + h.run_step(&Step::Advance(1)); + h.run_step(&empty_summary(1)); + settle(&mut h, n + 4); + prop_assert_eq!(h.offered_since_summary(1), 0, "a replayed summary re-sent ids"); + + // A full interval later the peer still lacks everything, so it gets + // everything again. + h.run_step(&Step::Advance(SUMMARY_INTERVAL_MS as i64)); + h.run_step(&empty_summary(1)); + settle(&mut h, n + 4); + prop_assert_eq!(h.offered_since_summary(1), n); } /// F1: acknowledging an unknown id must not evict a known, stored - /// observation. Seed a peer broadcast obs (guaranteed stored), then Ack an - /// id we did not observe and assert the seed is still present. + /// observation. #[test] - fn ack_unknown_does_not_evict_known(obs in obs_spec(), mtu in 32usize..=4096) { - let mut h = fuzz::Harness::new(mtu); - - // Seed a known broadcast observation from a peer. - h.step(&fuzz::Step::Obs(obs)); - h.drain(); - - let known: Vec<[u8; 16]> = { - let mut out = Vec::new(); - h.store().for_each_broadcast_id(&mut |id: &[u8; 16]| out.push(*id)); - out - }; - prop_assert!(!known.is_empty(), "expected a stored broadcast to ack against"); - - // An Ack referencing an id we never observed must not remove the seed. - let unknown = [0xFFu8; 16]; - h.step(&fuzz::Step::Ack(fuzz::AckSpec { addr: 1, from: [0u8; 16], ids: vec![unknown] })); - h.drain(); + fn ack_unknown_does_not_evict_known(obs in obs_spec(), mtu in mtu()) { + let mut h = Harness::new(Mode::Adversarial, mtu); + let obs = ObsSpec { dest: None, sig: None, observer: Who::Peer, from: Who::Peer, ..obs }; + let addr = obs.addr; + h.run_step(&Step::Obs(obs)); + let mut known = Vec::new(); + h.store().for_each_id(&mut |id| known.push(*id)); + prop_assert_eq!(known.len(), 1, "the peer's broadcast must be stored"); + h.run_step(&Step::Ack(AckSpec { addr, from: Who::Peer, sent: vec![], raw: vec![[0xFF; 16]] })); for id in &known { - prop_assert!(h.store().contains(id), "an Ack of an unknown id must not evict a known stored observation"); + prop_assert!(h.store().contains(id) && !h.store().is_released(id)); } } } + +/// §10 / F8 peer-table recovery: `MAX_PEERS + 16` addresses each send one +/// frame (filling the table), then everyone goes quiet. After +/// `PEER_IDLE_MS`, a never-seen address must get a slot. +#[test] +#[ignore = "needs S1 (idle sweep): silent peers are never evicted yet"] +fn peer_table_recovers_after_flood() { + let mut h = Harness::new(Mode::Adversarial, MAX_FRAME); + for addr in 0..(MAX_PEERS as u64 + 16) { + h.run_step(&Step::Obs(peer_obs(addr, addr))); + } + // Sanity: the flood really filled the table, so the probe tests eviction. + let before = h.store().len(); + h.run_step(&Step::Obs(peer_obs(1_000, 1_000))); + assert_eq!(h.store().len(), before, "the table should be full"); + + h.check_peer_table_recovers(); +} + +/// Same flood, but contact-only peers: `Contact` events with no frames must +/// age out too. +#[test] +#[ignore = "needs S1 (idle sweep): contact-only peers are never evicted yet"] +fn peer_table_recovers_after_contact_flood() { + let mut h = Harness::new(Mode::Adversarial, MAX_FRAME); + for addr in 0..(MAX_PEERS as u64) { + h.run_step(&Step::Contact(addr)); + } + h.check_peer_table_recovers(); +} + +/// Before the idle window ends the table stays full: the flood isn't +/// trivially recovered by any later frame. +#[test] +fn full_peer_table_refuses_a_newcomer_before_idle() { + let mut h = Harness::new(Mode::Adversarial, MAX_FRAME); + for addr in 0..(MAX_PEERS as u64) { + h.run_step(&Step::Obs(peer_obs(addr, addr))); + } + assert_eq!(h.store().len(), MAX_PEERS); + h.run_step(&Step::Advance(PEER_IDLE_MS as i64 / 2)); + h.run_step(&Step::Obs(peer_obs(500, 500))); + assert_eq!(h.store().len(), MAX_PEERS); +}