Merge branch 'maint'

CHANGELOG.md

@@ -139,6 +139,29 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   control socket interface is still pre-1.0 and not covered by
   stability guarantees
 
+### 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<f64>` and
+  returns `None` for saturated filters, preventing `f64::INFINITY`
+  from propagating into mesh-size estimates. The node-level
+  `estimated_mesh_size` field (already `Option<u64>`) propagates
+  `None` when any contributing filter is above cap.
+
 ### Fixed
 
 - Control socket path detection in fipsctl and fipstop now checks for

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<f64> {
         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()
     }
 }

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]

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.*`).

src/control/queries.rs

@@ -423,7 +423,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());
             }

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),

src/node/mod.rs

@@ -442,6 +442,13 @@ pub struct Node {
     /// Timestamp of last mesh size log emission.
     last_mesh_size_log: Option<std::time::Instant>,
 
+    // === 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<std::time::Instant>,
+
     // === Display Names ===
     /// Human-readable names for configured peers (alias or short npub).
     /// Populated at startup from peer config.
@@ -584,6 +591,7 @@ impl Node {
             last_congestion_log: None,
             estimated_mesh_size: None,
             last_mesh_size_log: None,
+            last_self_warn: None,
             peer_aliases: HashMap::new(),
             peer_acl,
             host_map,
@@ -704,6 +712,7 @@ impl Node {
             last_congestion_log: None,
             estimated_mesh_size: None,
             last_mesh_size_log: None,
+            last_self_warn: None,
             peer_aliases: HashMap::new(),
             peer_acl,
             host_map,
@@ -1071,17 +1080,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<u64> 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
@@ -1091,8 +1113,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;
+                        }
+                    }
                 }
             }
         }

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,

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}",

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<u8> {
+    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);
+}

src/node/tests/mod.rs

@@ -7,6 +7,7 @@ use std::time::Duration;
 #[cfg(target_os = "linux")]
 mod ble;
 mod bloom;
+mod bloom_poison;
 mod disconnect;
 mod discovery;
 #[cfg(unix)]