From c83e14ac976bfa9178a7f122574aae06ffa352d0 Mon Sep 17 00:00:00 2001 From: AndrewH <36428679+andrewheadricke@users.noreply.github.com> Date: Wed, 22 Apr 2026 01:06:39 +1000 Subject: [PATCH 1/2] Switch std::atomic to portable_atomic for mips support (#62) Co-authored-by: andrewheadricke --- Cargo.lock | 1 + Cargo.toml | 1 + src/transport/ble/stats.rs | 2 +- src/transport/ethernet/stats.rs | 2 +- src/transport/tcp/stats.rs | 2 +- src/transport/tor/stats.rs | 2 +- src/transport/udp/stats.rs | 2 +- 7 files changed, 7 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 1b045dc..2f5f2b3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -985,6 +985,7 @@ dependencies = [ "hex", "hkdf", "libc", + "portable-atomic", "rand 0.10.0", "ratatui", "rtnetlink", diff --git a/Cargo.toml b/Cargo.toml index 121157c..accca49 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ futures = "0.3" simple-dns = "0.11.2" socket2 = { version = "0.6.2", features = ["all"] } tokio-socks = "0.5" +portable-atomic = { version = "1", features = ["std"] } rustables = { version = "0.8.7", optional = true } diff --git a/src/transport/ble/stats.rs b/src/transport/ble/stats.rs index bae7461..f4fbccf 100644 --- a/src/transport/ble/stats.rs +++ b/src/transport/ble/stats.rs @@ -1,6 +1,6 @@ //! BLE transport statistics. -use std::sync::atomic::{AtomicU64, Ordering}; +use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; diff --git a/src/transport/ethernet/stats.rs b/src/transport/ethernet/stats.rs index 67e542d..c20492f 100644 --- a/src/transport/ethernet/stats.rs +++ b/src/transport/ethernet/stats.rs @@ -1,6 +1,6 @@ //! Ethernet transport statistics. -use std::sync::atomic::{AtomicU64, Ordering}; +use portable_atomic::{AtomicU64, Ordering}; /// Statistics for an Ethernet transport instance. /// diff --git a/src/transport/tcp/stats.rs b/src/transport/tcp/stats.rs index badcaa9..0434038 100644 --- a/src/transport/tcp/stats.rs +++ b/src/transport/tcp/stats.rs @@ -1,6 +1,6 @@ //! TCP transport statistics. -use std::sync::atomic::{AtomicU64, Ordering}; +use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; diff --git a/src/transport/tor/stats.rs b/src/transport/tor/stats.rs index f37dd77..4a6c8d9 100644 --- a/src/transport/tor/stats.rs +++ b/src/transport/tor/stats.rs @@ -1,6 +1,6 @@ //! Tor transport statistics. -use std::sync::atomic::{AtomicU64, Ordering}; +use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; diff --git a/src/transport/udp/stats.rs b/src/transport/udp/stats.rs index a85870c..a0bb79c 100644 --- a/src/transport/udp/stats.rs +++ b/src/transport/udp/stats.rs @@ -1,6 +1,6 @@ //! UDP transport statistics. -use std::sync::atomic::{AtomicU64, Ordering}; +use portable_atomic::{AtomicU64, Ordering}; use serde::Serialize; From db4b32110cf2d2cc531c57613e967c061ad8b9ec Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Tue, 21 Apr 2026 19:18:44 +0000 Subject: [PATCH 2/2] Validate bloom filter fill ratio on FilterAnnounce ingress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A malformed FilterAnnounce whose fill ratio produces an implausibly high false-positive rate is mostly useless for routing and, once merged into our outgoing filter via bitwise OR, propagates the saturated state to tree peers one hop per announce tick. A saturated filter also made estimated_count() return f64::INFINITY, which compute_mesh_size summed into its cached estimate. handle_filter_announce now rejects inbound FilterAnnounce whose derived FPR exceeds `node.bloom.max_inbound_fpr` (new config field, default 0.05 ≈ fill 0.549 at k=5). Rejection is silent on the wire, logs at WARN, and increments a new `bloom.fill_exceeded` counter. The peer's prior stored filter and filter_sequence are left unchanged so a single rejected announce does not wipe the peer's existing contribution to aggregation. After a successful outgoing FilterAnnounce send, a rate-limited WARN fires if our own filter's FPR exceeds the same cap, surfacing aggregation drift. Limited to once per 60 seconds via a new Node.last_self_warn field. BloomFilter::estimated_count() now takes max_fpr and returns Option. Returns None for saturated filters (regardless of cap) or when the filter's FPR exceeds max_fpr. Callers updated: debug logs render None as "—", the Debug impl uses f64::INFINITY as "no cap" and prints "saturated" instead of inf, control-socket JSON emits null, and compute_mesh_size propagates None into the already- Option estimated_mesh_size field. --- CHANGELOG.md | 23 +++++ src/bloom/filter.rs | 30 +++++- src/bloom/tests.rs | 41 ++++++++- src/config/node.rs | 11 +++ src/control/queries.rs | 3 +- src/node/bloom.rs | 61 +++++++++++- src/node/mod.rs | 40 +++++++- src/node/stats.rs | 3 + src/node/tests/bloom.rs | 15 ++- src/node/tests/bloom_poison.rs | 164 +++++++++++++++++++++++++++++++++ src/node/tests/mod.rs | 1 + 11 files changed, 372 insertions(+), 20 deletions(-) create mode 100644 src/node/tests/bloom_poison.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 93bf41b..3520d99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 and `.deb` packages on `v*` tag push, with SHA-256 checksums - AUR publish workflow for tagged stable releases +### Security + +- Bloom filter poisoning defense. Reject inbound FilterAnnounce + messages whose false-positive rate exceeds a configurable cap + (`node.bloom.max_inbound_fpr`, default 0.05). Previously a peer + running a modified build could send an all-ones filter, causing + (1) lookup attraction / black-hole routing for unknown targets, + (2) aggregation contamination as the poisoned bits propagated one + hop per announce tick via strict-OR merging, and (3) mesh-size + estimate blowup to `f64::INFINITY`. Rejection is silent on the + wire; rejected announces log at WARN and increment a new + `bloom.fill_exceeded` counter. The peer's prior accepted filter + and sequence number are preserved on rejection so a single bad + announce cannot wipe a peer's contribution to aggregation. + An independent self-plausibility WARN fires (rate-limited to once + per 60s) if our own outgoing filter ever exceeds the cap, + surfacing aggregation drift or ingress-check bypasses. + `BloomFilter::estimated_count` now returns `Option` and + returns `None` for saturated filters, preventing `f64::INFINITY` + from propagating into mesh-size estimates. The node-level + `estimated_mesh_size` field (already `Option`) propagates + `None` when any contributing filter is above cap. + ### Fixed - Control socket path detection in fipsctl and fipstop now checks for diff --git a/src/bloom/filter.rs b/src/bloom/filter.rs index e28b0a3..3566862 100644 --- a/src/bloom/filter.rs +++ b/src/bloom/filter.rs @@ -2,6 +2,8 @@ use std::fmt; +use tracing::trace; + use super::{BloomError, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; use crate::NodeAddr; @@ -143,16 +145,30 @@ impl BloomFilter { /// /// Uses the formula: n = -(m/k) * ln(1 - X/m) /// where m = num_bits, k = hash_count, X = count_ones - pub fn estimated_count(&self) -> f64 { + /// + /// Returns `None` when the filter's FPR exceeds `max_fpr` (antipoison + /// cap) or the filter is saturated (`count_ones() >= num_bits`). Pass + /// `f64::INFINITY` for `max_fpr` to disable the cap — useful in + /// Debug/log contexts where no policy is in scope. The saturated + /// branch is always honored regardless of `max_fpr`, preventing the + /// `f64::INFINITY` return that the previous signature produced. + pub fn estimated_count(&self, max_fpr: f64) -> Option { let m = self.num_bits as f64; let k = self.hash_count as f64; let x = self.count_ones() as f64; if x >= m { - return f64::INFINITY; + return None; } - -(m / k) * (1.0 - x / m).ln() + let fill = x / m; + let fpr = fill.powi(self.hash_count as i32); + if fpr > max_fpr { + trace!(fill, fpr, max_fpr, "estimated_count: filter above cap"); + return None; + } + + Some(-(m / k) * (1.0 - fill).ln()) } /// Check if the filter is empty. @@ -234,7 +250,13 @@ impl fmt::Debug for BloomFilter { .field("bits", &self.num_bits) .field("hash_count", &self.hash_count) .field("fill_ratio", &format!("{:.2}%", self.fill_ratio() * 100.0)) - .field("est_count", &format!("{:.0}", self.estimated_count())) + .field( + "est_count", + &match self.estimated_count(f64::INFINITY) { + Some(n) => format!("{:.0}", n), + None => "saturated".to_string(), + }, + ) .finish() } } diff --git a/src/bloom/tests.rs b/src/bloom/tests.rs index 41df06c..df23c8e 100644 --- a/src/bloom/tests.rs +++ b/src/bloom/tests.rs @@ -159,7 +159,7 @@ fn test_bloom_filter_estimated_count() { let mut filter = BloomFilter::new(); // Empty filter - assert_eq!(filter.estimated_count(), 0.0); + assert_eq!(filter.estimated_count(f64::INFINITY), Some(0.0)); // Insert some items for i in 0..50 { @@ -167,7 +167,7 @@ fn test_bloom_filter_estimated_count() { } // Estimate should be reasonably close to 50 - let estimate = filter.estimated_count(); + let estimate = filter.estimated_count(f64::INFINITY).unwrap(); assert!( estimate > 30.0 && estimate < 100.0, "Unexpected estimate: {}", @@ -234,7 +234,42 @@ fn test_bloom_filter_estimated_count_saturated() { let bytes = vec![0xFF; 8]; // all bits set let filter = BloomFilter::from_bytes(bytes, 3).unwrap(); - assert!(filter.estimated_count().is_infinite()); + // Saturated filter returns None regardless of cap (defense in depth). + // Previously returned f64::INFINITY. + assert_eq!(filter.estimated_count(f64::INFINITY), None); + assert_eq!(filter.estimated_count(0.05), None); +} + +#[test] +fn test_bloom_filter_estimated_count_fpr_cap_boundary() { + // Cap boundary: FPR = fill^k = 0.05 at k=5 ⇒ fill ≈ 0.5493 + // 1KB filter (8192 bits). 560 bytes of 0xFF = 4480 bits set = + // fill 0.5469, FPR ≈ 0.04877 — just below cap. + // 564 bytes of 0xFF = 4512 bits set = fill 0.5508, FPR ≈ 0.05060 — + // just above cap. + + let mut below = vec![0x00u8; 1024]; + below[..560].fill(0xFF); + let below_filter = BloomFilter::from_bytes(below, DEFAULT_HASH_COUNT).unwrap(); + assert!( + below_filter.estimated_count(0.05).is_some(), + "fill 0.5469 (FPR ≈ 0.049) must be accepted by cap 0.05" + ); + + let mut above = vec![0x00u8; 1024]; + above[..564].fill(0xFF); + let above_filter = BloomFilter::from_bytes(above, DEFAULT_HASH_COUNT).unwrap(); + assert_eq!( + above_filter.estimated_count(0.05), + None, + "fill 0.5508 (FPR ≈ 0.051) must be rejected by cap 0.05" + ); + + // Same above-cap filter with a looser cap is accepted. + assert!( + above_filter.estimated_count(0.10).is_some(), + "fill 0.5508 (FPR ≈ 0.051) must be accepted by cap 0.10" + ); } #[test] diff --git a/src/config/node.rs b/src/config/node.rs index 413cfdc..a4ba7e0 100644 --- a/src/config/node.rs +++ b/src/config/node.rs @@ -344,12 +344,20 @@ pub struct BloomConfig { /// Debounce interval for filter updates in ms (`node.bloom.update_debounce_ms`). #[serde(default = "BloomConfig::default_update_debounce_ms")] pub update_debounce_ms: u64, + /// Antipoison cap: reject inbound FilterAnnounce whose FPR exceeds + /// this value (`node.bloom.max_inbound_fpr`). Valid range `(0.0, 1.0)`. + /// Default `0.05` ≈ fill 0.549 at k=5 ≈ ~3,200 entries on the 1KB + /// filter. Conceptually distinct from future autoscaling hysteresis + /// setpoints — same unit, different knobs. + #[serde(default = "BloomConfig::default_max_inbound_fpr")] + pub max_inbound_fpr: f64, } impl Default for BloomConfig { fn default() -> Self { Self { update_debounce_ms: 500, + max_inbound_fpr: 0.05, } } } @@ -358,6 +366,9 @@ impl BloomConfig { fn default_update_debounce_ms() -> u64 { 500 } + fn default_max_inbound_fpr() -> f64 { + 0.05 + } } /// Session/data plane (`node.session.*`). diff --git a/src/control/queries.rs b/src/control/queries.rs index 0ec2475..ac68177 100644 --- a/src/control/queries.rs +++ b/src/control/queries.rs @@ -340,7 +340,8 @@ pub fn show_bloom(node: &Node) -> Value { "filter_sequence": peer.filter_sequence(), }); if let Some(filter) = peer.inbound_filter() { - pf["estimated_count"] = json!(filter.estimated_count()); + let max_fpr = node.config().node.bloom.max_inbound_fpr; + pf["estimated_count"] = json!(filter.estimated_count(max_fpr)); pf["set_bits"] = json!(filter.count_ones()); pf["fill_ratio"] = json!(filter.fill_ratio()); } diff --git a/src/node/bloom.rs b/src/node/bloom.rs index 6103ee3..69c0289 100644 --- a/src/node/bloom.rs +++ b/src/node/bloom.rs @@ -9,7 +9,7 @@ use crate::protocol::FilterAnnounce; use super::{Node, NodeError}; use std::collections::HashMap; -use tracing::debug; +use tracing::{debug, warn}; impl Node { /// Collect inbound filters from all peers for outgoing filter computation. @@ -78,11 +78,41 @@ impl Node { self.stats_mut().bloom.sent += 1; + // Self-plausibility check: WARN if our own outgoing filter is + // above the antipoison cap. Independent detection signal if + // aggregation drift or an ingress-check bypass pushes us over + // despite M1. Rate-limited to once per 60s globally — outgoing + // cadence can be per-tick during churn, and we want the + // operator to see one clear message, not spam. + let max_fpr = self.config.node.bloom.max_inbound_fpr; + let out_fill = sent_filter.fill_ratio(); + let out_fpr = out_fill.powi(sent_filter.hash_count() as i32); + if out_fpr > max_fpr { + let now = std::time::Instant::now(); + let should_warn = self + .last_self_warn + .map(|t| now.duration_since(t) >= std::time::Duration::from_secs(60)) + .unwrap_or(true); + if should_warn { + self.last_self_warn = Some(now); + warn!( + to = %self.peer_display_name(peer_addr), + fill = format_args!("{:.3}", out_fill), + fpr = format_args!("{:.4}", out_fpr), + cap = format_args!("{:.4}", max_fpr), + "Outgoing filter above FPR cap — aggregation drift or missed ingress?" + ); + } + } + // Record send and store the filter for change detection debug!( peer = %self.peer_display_name(peer_addr), seq = announce.sequence, - est_entries = format_args!("{:.0}", sent_filter.estimated_count()), + est_entries = match sent_filter.estimated_count(max_fpr) { + Some(n) => format!("{:.0}", n), + None => "—".to_string(), + }, set_bits = sent_filter.count_ones(), fill = format_args!("{:.1}%", sent_filter.fill_ratio() * 100.0), tree_peer = self.is_tree_peer(peer_addr), @@ -174,6 +204,28 @@ impl Node { return; } + // Antipoison FPR cap. Reject announces whose FPR exceeds + // node.bloom.max_inbound_fpr. Silent on the wire (no NACK) — + // the peer's prior accepted filter and filter_sequence stay + // untouched so the peer is not permanently silenced and an + // on-path attacker cannot weaponize a single corrupted frame + // to wipe a victim's contribution to aggregation. + let max_fpr = self.config.node.bloom.max_inbound_fpr; + let fill = announce.filter.fill_ratio(); + let fpr = fill.powi(announce.filter.hash_count() as i32); + if fpr > max_fpr { + self.stats_mut().bloom.fill_exceeded += 1; + warn!( + from = %self.peer_display_name(from), + seq = announce.sequence, + fill = format_args!("{:.3}", fill), + fpr = format_args!("{:.4}", fpr), + cap = format_args!("{:.4}", max_fpr), + "FilterAnnounce above FPR cap — rejected" + ); + return; + } + self.stats_mut().bloom.accepted += 1; let now_ms = std::time::SystemTime::now() @@ -184,7 +236,10 @@ impl Node { debug!( from = %self.peer_display_name(from), seq = announce.sequence, - est_entries = format_args!("{:.0}", announce.filter.estimated_count()), + est_entries = match announce.filter.estimated_count(max_fpr) { + Some(n) => format!("{:.0}", n), + None => "—".to_string(), + }, set_bits = announce.filter.count_ones(), fill = format_args!("{:.1}%", announce.filter.fill_ratio() * 100.0), tree_peer = self.is_tree_peer(from), diff --git a/src/node/mod.rs b/src/node/mod.rs index a505b93..26da467 100644 --- a/src/node/mod.rs +++ b/src/node/mod.rs @@ -430,6 +430,13 @@ pub struct Node { /// Timestamp of last mesh size log emission. last_mesh_size_log: Option, + // === Bloom Self-Plausibility === + /// Rate-limit state for the self-plausibility WARN. Fires at most + /// once per 60s globally when our own outgoing FilterAnnounce has + /// an FPR above `node.bloom.max_inbound_fpr`, signalling either + /// aggregation drift or an ingress bypass. + last_self_warn: Option, + // === Display Names === /// Human-readable names for configured peers (alias or short npub). /// Populated at startup from peer config. @@ -558,6 +565,7 @@ impl Node { last_congestion_log: None, estimated_mesh_size: None, last_mesh_size_log: None, + last_self_warn: None, peer_aliases: HashMap::new(), host_map, }) @@ -663,6 +671,7 @@ impl Node { last_congestion_log: None, estimated_mesh_size: None, last_mesh_size_log: None, + last_self_warn: None, peer_aliases: HashMap::new(), host_map, } @@ -952,17 +961,30 @@ impl Node { let parent_id = *self.tree_state.my_declaration().parent_id(); let is_root = self.tree_state.is_root(); + let max_fpr = self.config.node.bloom.max_inbound_fpr; let mut total: f64 = 1.0; // count self let mut child_count: u32 = 0; let mut has_data = false; - // Parent's filter: nodes reachable upward through the tree + // Parent's filter: nodes reachable upward through the tree. + // If any contributing filter is above the FPR cap, we refuse to + // estimate rather than substitute a partial/biased aggregate — + // Node.estimated_mesh_size is already Option and consumers + // (control socket, fipstop, periodic debug log) handle None. if !is_root && let Some(parent) = self.peers.get(&parent_id) && let Some(filter) = parent.inbound_filter() { - total += filter.estimated_count(); - has_data = true; + match filter.estimated_count(max_fpr) { + Some(n) => { + total += n; + has_data = true; + } + None => { + self.estimated_mesh_size = None; + return; + } + } } // Children's filters: each child's subtree is disjoint @@ -972,8 +994,16 @@ impl Node { { child_count += 1; if let Some(filter) = peer.inbound_filter() { - total += filter.estimated_count(); - has_data = true; + match filter.estimated_count(max_fpr) { + Some(n) => { + total += n; + has_data = true; + } + None => { + self.estimated_mesh_size = None; + return; + } + } } } } diff --git a/src/node/stats.rs b/src/node/stats.rs index d4e2008..4bd0b24 100644 --- a/src/node/stats.rs +++ b/src/node/stats.rs @@ -211,6 +211,7 @@ pub struct BloomStats { pub non_v1: u64, pub unknown_peer: u64, pub stale: u64, + pub fill_exceeded: u64, pub accepted: u64, // Outbound announce sending pub sent: u64, @@ -227,6 +228,7 @@ impl BloomStats { non_v1: self.non_v1, unknown_peer: self.unknown_peer, stale: self.stale, + fill_exceeded: self.fill_exceeded, accepted: self.accepted, sent: self.sent, debounce_suppressed: self.debounce_suppressed, @@ -397,6 +399,7 @@ pub struct BloomStatsSnapshot { pub non_v1: u64, pub unknown_peer: u64, pub stale: u64, + pub fill_exceeded: u64, pub accepted: u64, pub sent: u64, pub debounce_suppressed: u64, diff --git a/src/node/tests/bloom.rs b/src/node/tests/bloom.rs index f2df5ac..edc4707 100644 --- a/src/node/tests/bloom.rs +++ b/src/node/tests/bloom.rs @@ -270,10 +270,13 @@ fn print_filter_cardinality(nodes: &[TestNode]) { { let is_tree = tn.node.is_tree_peer(&addr); println!( - " n{} <- n{}: est={:.1} set_bits={} fill={:.1}% tree={}", + " n{} <- n{}: est={} set_bits={} fill={:.1}% tree={}", i, j, - filter.estimated_count(), + match filter.estimated_count(f64::INFINITY) { + Some(n) => format!("{:.1}", n), + None => "saturated".to_string(), + }, filter.count_ones(), filter.fill_ratio() * 100.0, is_tree, @@ -369,7 +372,9 @@ async fn test_bloom_filter_split_horizon() { } // Cardinality should match subtree size - let up_est = filter_up.estimated_count(); + let up_est = filter_up + .estimated_count(f64::INFINITY) + .expect("upward filter should not be saturated in tree convergence test"); assert!( (up_est - child_subtree.len() as f64).abs() < 1.5, "Upward filter (n{}→n{}): expected ~{} entries, got {:.1}", @@ -414,7 +419,9 @@ async fn test_bloom_filter_split_horizon() { } // Cardinality should match complement size - let down_est = filter_down.estimated_count(); + let down_est = filter_down + .estimated_count(f64::INFINITY) + .expect("downward filter should not be saturated in tree convergence test"); assert!( (down_est - complement.len() as f64).abs() < 1.5, "Downward filter (n{}→n{}): expected ~{} entries, got {:.1}", diff --git a/src/node/tests/bloom_poison.rs b/src/node/tests/bloom_poison.rs new file mode 100644 index 0000000..17f5ace --- /dev/null +++ b/src/node/tests/bloom_poison.rs @@ -0,0 +1,164 @@ +//! Direct tests for the M1 antipoison FPR cap in handle_filter_announce. +//! +//! These tests construct a minimal Node with a single synthetic peer, +//! then call handle_filter_announce directly with crafted FilterAnnounce +//! payloads. Focused on the ingress check semantics; broader +//! filter-exchange behavior is covered by the multi-node tests in +//! bloom.rs. + +use super::*; +use crate::bloom::{BloomFilter, DEFAULT_FILTER_SIZE_BITS, DEFAULT_HASH_COUNT}; +use crate::peer::ActivePeer; +use crate::protocol::FilterAnnounce; + +/// Inject a synthetic active peer into the node with a known NodeAddr. +/// Returns the peer's NodeAddr. +fn inject_peer(node: &mut Node) -> NodeAddr { + let peer_identity = make_peer_identity(); + let peer_addr = *peer_identity.node_addr(); + let peer = ActivePeer::new(peer_identity, LinkId::new(1), 0); + node.peers.insert(peer_addr, peer); + peer_addr +} + +/// Encode a FilterAnnounce to the payload format handle_filter_announce +/// expects (msg_type byte stripped). +fn encode_payload(announce: &FilterAnnounce) -> Vec { + let mut full = announce.encode().unwrap(); + full.remove(0); // strip msg_type byte + full +} + +#[tokio::test] +async fn test_m1_rejects_all_ones_filter_announce() { + let mut node = make_node(); + let peer_addr = inject_peer(&mut node); + + // Craft an all-ones FilterAnnounce (the observed-in-the-wild attack). + let all_ones = BloomFilter::from_bytes( + vec![0xFFu8; DEFAULT_FILTER_SIZE_BITS / 8], + DEFAULT_HASH_COUNT, + ) + .unwrap(); + let announce = FilterAnnounce::new(all_ones, 1); + let payload = encode_payload(&announce); + + let before_fill_exceeded = node.stats().bloom.fill_exceeded; + let before_accepted = node.stats().bloom.accepted; + + node.handle_filter_announce(&peer_addr, &payload).await; + + let after = &node.stats().bloom; + assert_eq!( + after.fill_exceeded, + before_fill_exceeded + 1, + "fill_exceeded counter must increment on all-ones rejection" + ); + assert_eq!( + after.accepted, before_accepted, + "accepted counter must NOT increment on rejection" + ); + + // Peer state unchanged: no filter stored, sequence not advanced. + let peer = node.get_peer(&peer_addr).expect("peer still present"); + assert!( + peer.inbound_filter().is_none(), + "peer must NOT have a stored filter after rejection" + ); + assert_eq!( + peer.filter_sequence(), + 0, + "peer filter_sequence must NOT advance on rejection" + ); +} + +#[tokio::test] +async fn test_m1_accepts_sub_cap_filter() { + let mut node = make_node(); + let peer_addr = inject_peer(&mut node); + + // A legitimate filter with 50 entries — fill ~0.03, FPR ~2e-8, + // far below the 0.05 cap. Represents normal mesh traffic. + let mut filter = BloomFilter::new(); + for i in 0..50u8 { + let mut bytes = [0u8; 16]; + bytes[0] = i; + filter.insert(&NodeAddr::from_bytes(bytes)); + } + let announce = FilterAnnounce::new(filter, 1); + let payload = encode_payload(&announce); + + let before_fill_exceeded = node.stats().bloom.fill_exceeded; + let before_accepted = node.stats().bloom.accepted; + + node.handle_filter_announce(&peer_addr, &payload).await; + + let after = &node.stats().bloom; + assert_eq!( + after.fill_exceeded, before_fill_exceeded, + "fill_exceeded must NOT increment on legitimate sub-cap filter" + ); + assert_eq!( + after.accepted, + before_accepted + 1, + "accepted must increment on legitimate filter" + ); + + // Peer state updated: filter stored, sequence advanced. + let peer = node.get_peer(&peer_addr).expect("peer still present"); + assert!( + peer.inbound_filter().is_some(), + "peer must have a stored filter after acceptance" + ); + assert_eq!( + peer.filter_sequence(), + 1, + "peer filter_sequence must advance to announce's sequence" + ); +} + +#[tokio::test] +async fn test_m1_sequence_not_advanced_allows_recovery() { + // Confirms the "keep prior filter, don't advance seq" rejection + // semantics: a compliant announce after a rejected one still + // succeeds at seq=1, because the rejected announce (also seq=1) + // did not advance the peer's recorded sequence. + let mut node = make_node(); + let peer_addr = inject_peer(&mut node); + + // First announce: all-ones, rejected. + let bad = BloomFilter::from_bytes( + vec![0xFFu8; DEFAULT_FILTER_SIZE_BITS / 8], + DEFAULT_HASH_COUNT, + ) + .unwrap(); + let bad_announce = FilterAnnounce::new(bad, 1); + node.handle_filter_announce(&peer_addr, &encode_payload(&bad_announce)) + .await; + assert_eq!( + node.get_peer(&peer_addr).unwrap().filter_sequence(), + 0, + "rejected announce must not advance sequence" + ); + + // Second announce: legitimate, seq=1 (would be stale if rejection + // had advanced the recorded sequence). Must be accepted. + let mut good = BloomFilter::new(); + for i in 0..10u8 { + let mut bytes = [0u8; 16]; + bytes[0] = i; + good.insert(&NodeAddr::from_bytes(bytes)); + } + let good_announce = FilterAnnounce::new(good, 1); + node.handle_filter_announce(&peer_addr, &encode_payload(&good_announce)) + .await; + + let peer = node.get_peer(&peer_addr).unwrap(); + assert!( + peer.inbound_filter().is_some(), + "compliant announce at same seq must be accepted after rejection" + ); + assert_eq!(peer.filter_sequence(), 1); + assert_eq!(node.stats().bloom.fill_exceeded, 1); + assert_eq!(node.stats().bloom.accepted, 1); +} diff --git a/src/node/tests/mod.rs b/src/node/tests/mod.rs index cab6b51..cd69e62 100644 --- a/src/node/tests/mod.rs +++ b/src/node/tests/mod.rs @@ -5,6 +5,7 @@ use crate::utils::index::SessionIndex; use std::time::Duration; mod bloom; +mod bloom_poison; mod disconnect; mod discovery; #[cfg(target_os = "linux")]