diff --git a/docs/docs/users/reference/env_variables.md b/docs/docs/users/reference/env_variables.md index 789bbb3708c5..fa8b322e8dd2 100644 --- a/docs/docs/users/reference/env_variables.md +++ b/docs/docs/users/reference/env_variables.md @@ -75,6 +75,7 @@ process. | `FOREST_MAX_CONCURRENT_INBOUND_CHAIN_EXCHANGE_REQUESTS` | positive integer | 32 | 32 | Maximum number of inbound chain exchange requests Forest will service concurrently. Excess requests are rejected with a `GoAway` response | | `FOREST_MAX_CONCURRENT_INBOUND_CHAIN_EXCHANGE_REQUESTS_PER_PEER` | positive integer | 4 | 4 | Per-peer cap on concurrent inbound chain exchange requests. Excess requests from a single peer are rejected with a `GoAway` response | | `FOREST_MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES` | positive integer | 16 | 16 | Bounds tipset fetches triggered by inbound `hello` requests that run concurrently; each chain-exchanges the peer's claimed head. Excess triggers are dropped, not queued. | +| `FOREST_MAX_CONCURRENT_DRAND_VERIFICATIONS` | positive integer | 4 | 4 | Bounds drand beacon entries from gossipsub verified concurrently; each costs a BLS pairing. Excess entries are dropped, not queued. | | `FOREST_MAX_OUTBOUND_CHAIN_EXCHANGE_RESPONSE_BYTES` | positive integer (bytes) | 10485760 (10 MiB) | 10485760 | Cap on the encoded byte size of a chain exchange response Forest serves to peers. Building stops as soon as the running encoded size would exceed this cap and the response is returned with `PartialResponse` status | | `FOREST_ETH_RPC_COMPUTE_STATE_ON_INDEX_MISS` | 1 or true | false | 1 | Allows Ethereum RPC methods to compute state trees on index miss | | `FOREST_ETH_RPC_COMPUTE_BLOOM_ON_MISS` | 1 or true | false | 1 | Allows `eth` block RPC methods to compute (and store) the block `logsBloom` when it is not already stored, otherwise such blocks report an all-ones bloom | diff --git a/proto/drand_pb.proto b/proto/drand_pb.proto new file mode 100644 index 000000000000..74319bd75a4b --- /dev/null +++ b/proto/drand_pb.proto @@ -0,0 +1,9 @@ +syntax = "proto3"; + +package drand_pb; + +// https://github.com/drand/drand/blob/v2.1.7/protobuf/drand/api.proto#L42-L53 +message PublicRandResponse { + uint64 round = 1; + bytes signature = 2; +} diff --git a/src/beacon/drand.rs b/src/beacon/drand.rs index bbf151b29366..1b1a3a9b7410 100644 --- a/src/beacon/drand.rs +++ b/src/beacon/drand.rs @@ -140,6 +140,13 @@ impl BeaconSchedule { } } + pub fn unchained_beacon(&self) -> Option<&BeaconImpl> { + self.0 + .iter() + .map(|point| &point.beacon) + .find(|beacon| beacon.network().is_unchained()) + } + pub fn beacon_for_epoch(&self, epoch: ChainEpoch) -> anyhow::Result<(ChainEpoch, &BeaconImpl)> { // Iterate over beacon schedule to find the latest randomness beacon to use. self.0 diff --git a/src/beacon/drand_pb.rs b/src/beacon/drand_pb.rs new file mode 100644 index 000000000000..3447b19f9cb6 --- /dev/null +++ b/src/beacon/drand_pb.rs @@ -0,0 +1,55 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT +// Automatically generated rust module for 'drand_pb.proto' file +// Command: `pb-rs -s -D proto/drand_pb.proto`, See + +#![allow(non_snake_case)] +#![allow(non_upper_case_globals)] +#![allow(non_camel_case_types)] +#![allow(unused_imports)] +#![allow(unknown_lints)] +#![allow(clippy::all)] +#![allow(clippy::assigning_clones)] +#![cfg_attr(rustfmt, rustfmt_skip)] + + +use quick_protobuf::{MessageInfo, MessageRead, MessageWrite, BytesReader, Writer, WriterBackend, Result}; +use quick_protobuf::sizeofs::*; +use super::*; + +#[allow(clippy::derive_partial_eq_without_eq)] +#[derive(Debug, Default, PartialEq, Clone)] +pub struct PublicRandResponse { + pub round: u64, + pub signature: Vec, +} + +impl<'a> MessageRead<'a> for PublicRandResponse { + fn from_reader(r: &mut BytesReader, bytes: &'a [u8]) -> Result { + let mut msg = Self::default(); + while !r.is_eof() { + match r.next_tag(bytes) { + Ok(8) => msg.round = r.read_uint64(bytes)?, + Ok(18) => msg.signature = r.read_bytes(bytes)?.to_owned(), + Ok(t) => { r.read_unknown(bytes, t)?; } + Err(e) => return Err(e), + } + } + Ok(msg) + } +} + +impl MessageWrite for PublicRandResponse { + fn get_size(&self) -> usize { + 0 + + if self.round == 0u64 { 0 } else { 1 + sizeof_varint(*(&self.round) as u64) } + + if self.signature.is_empty() { 0 } else { 1 + sizeof_len((&self.signature).len()) } + } + + fn write_message(&self, w: &mut Writer) -> Result<()> { + if self.round != 0u64 { w.write_with_tag(8, |w| w.write_uint64(*&self.round))?; } + if !self.signature.is_empty() { w.write_with_tag(18, |w| w.write_bytes(&**&self.signature))?; } + Ok(()) + } +} + diff --git a/src/beacon/metrics.rs b/src/beacon/metrics.rs index 0dc238ed58b5..9a3bd78f34dd 100644 --- a/src/beacon/metrics.rs +++ b/src/beacon/metrics.rs @@ -9,7 +9,7 @@ use std::sync::LazyLock; pub static DRAND_HTTP_FETCH_TOTAL: LazyLock = LazyLock::new(|| { let metric = Counter::default(); crate::metrics::default_registry().register( - "drand_http_fetch_total", + "drand_http_fetch", "Total number of drand rounds fetched over HTTP", metric.clone(), ); diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index e849804fb088..37fb5c73d722 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -3,10 +3,12 @@ pub mod beacon_entries; mod drand; +mod drand_pb; pub mod metrics; pub mod signatures; pub use beacon_entries::*; pub use drand::*; +pub use drand_pb::PublicRandResponse; #[cfg(test)] pub mod mock_beacon; @@ -15,4 +17,5 @@ pub mod tests { // `pub` so that helpers such as `drand::new_beacon_quicknet` can be shared with // tests in other modules. pub mod drand; + pub mod fake_drand; } diff --git a/src/beacon/signatures/mod.rs b/src/beacon/signatures/mod.rs index 19cb9b90d456..bd297cc86b96 100644 --- a/src/beacon/signatures/mod.rs +++ b/src/beacon/signatures/mod.rs @@ -13,7 +13,7 @@ use rayon::prelude::*; pub use bls_signatures::{PublicKey as PublicKeyOnG1, Signature as SignatureOnG2}; // See -const CSUITE_G1: &[u8] = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_"; +pub(crate) const CSUITE_G1: &[u8] = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_"; const CSUITE_G2: &[u8] = b"BLS_SIG_BLS12381G2_XMD:SHA-256_SSWU_RO_NUL_"; #[derive(Debug, Clone, Eq, PartialEq, derive_more::Deref)] diff --git a/src/beacon/tests/fake_drand.rs b/src/beacon/tests/fake_drand.rs new file mode 100644 index 000000000000..8240fc0f365d --- /dev/null +++ b/src/beacon/tests/fake_drand.rs @@ -0,0 +1,95 @@ +use crate::beacon::{Beacon, BeaconEntry, ChainInfo, DrandBeacon, DrandConfig, DrandNetwork}; +use blstrs::{G1Projective, G2Projective, Scalar}; +use group::{Curve, Group}; + +pub const FAKE_DRAND_GENESIS_TIME: i32 = 1_692_803_367; +pub const FAKE_DRAND_PERIOD: i32 = 3; + +pub const TEST_FIL_GENESIS_TIME: u64 = 1_598_306_400; +pub const TEST_FIL_BLOCK_DELAY: u64 = 30; + +pub struct FakeDrand { + secret: Scalar, + config: DrandConfig<'static>, +} + +impl FakeDrand { + pub fn new(servers: Vec, period: i32, genesis_time: i32) -> Self { + let secret = Scalar::from(0xC0FFEEu64); + let public = G2Projective::generator() * secret; + let public_key = hex::encode(public.to_affine().to_compressed()); + Self { + secret, + config: DrandConfig { + servers, + chain_info: ChainInfo { + public_key: public_key.into(), + period, + genesis_time, + hash: "0011".repeat(16).into(), + group_hash: "00".repeat(32).into(), + }, + network_type: DrandNetwork::Quicknet, // unchained + // The fixture builds beacons directly; registering a collector here + // would clash with the one the real quicknet config registers. + register_metrics: false, + }, + } + } + + // sign H(round) on G1, exactly what `verify_entries` checks for unchained. + pub fn entry(&self, round: u64) -> BeaconEntry { + let msg = BeaconEntry::message_unchained(round); + let point = + G1Projective::hash_to_curve(msg.as_ref(), crate::beacon::signatures::CSUITE_G1, &[]); + let point = point * self.secret; + BeaconEntry::new(round, point.to_affine().to_compressed().to_vec()) + } + + // encode PublicRandResponse to protobuf + pub fn to_protobuf(&self, round: u64) -> Vec { + let entry = self.entry(round); + let mut out = Vec::new(); + let mut w = quick_protobuf::Writer::new(&mut out); + quick_protobuf::MessageWrite::write_message( + &crate::beacon::drand_pb::PublicRandResponse { + round, + signature: entry.signature().to_vec(), + }, + &mut w, + ) + .unwrap(); + out + } + + pub fn to_json(&self, round: u64) -> serde_json::Value { + let entry = self.entry(round); + serde_json::json!({ + "round": round, + "randomness": "00".repeat(32), + "signature": hex::encode(entry.signature()), + "previous_signature": null, + }) + } + + pub fn beacon(&self, genesis_ts: u64, block_delay: u64) -> DrandBeacon { + DrandBeacon::new(genesis_ts, block_delay, &self.config) + } + + pub fn chain_info_hash(&self) -> String { + self.config.chain_info.hash.to_string() + } +} + +// just test the secret and public keys are correctly validating +#[test] +fn fake_drand_entries_verify() { + let d = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let beacon = d.beacon(TEST_FIL_GENESIS_TIME, TEST_FIL_BLOCK_DELAY); + let entries: Vec<_> = (1..=5).map(|r| d.entry(r)).collect(); + assert!( + beacon + .verify_entries(&entries, &BeaconEntry::default()) + .unwrap() + ); +} diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index e7d557a72bb0..697fb31acf20 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -18,6 +18,7 @@ use super::network_context::SyncNetworkContext; use crate::{ + beacon::{Beacon, BeaconEntry, BeaconSchedule}, blocks::{Block, FullTipset, Tipset, TipsetKey}, chain::{ChainStore, index::ResolveNullTipset}, chain_sync::{ @@ -41,6 +42,7 @@ use hashbrown::{HashMap, HashSet}; use libp2p::PeerId; use nonzero_ext::nonzero; use parking_lot::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; use std::{ borrow::Cow, sync::LazyLock, @@ -228,6 +230,10 @@ async fn chain_follower( let hello_fetch_limiter = Arc::new(Semaphore::new(*MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES)); + let drand_verify_limiter = Arc::new(Semaphore::new(*MAX_CONCURRENT_DRAND_VERIFICATIONS)); + + let last_drand_entry = Arc::new(AtomicU64::new(0)); + let mut set = JoinSet::new(); let cancellation_token = CancellationToken::new(); let _cancellation_token_drop_guard = cancellation_token.drop_guard_ref(); @@ -243,6 +249,8 @@ async fn chain_follower( let cancellation_token = cancellation_token.clone(); let hello_fetch_limiter = hello_fetch_limiter.shallow_clone(); let tipset_sender = tipset_sender.clone(); + let last_drand_entry = last_drand_entry.clone(); + let drand_verify_limiter = drand_verify_limiter.shallow_clone(); async move { while let Ok(event) = network_rx.recv_async().await { inc_gossipsub_event_metrics(&event); @@ -309,6 +317,12 @@ async fn chain_follower( debug!("Received invalid GossipSub message: {}", why); } } + PubsubMessage::DrandEntry(entry) => handle_drand_entry( + entry, + &drand_verify_limiter, + state_manager.beacon_schedule(), + &last_drand_entry, + ), }, _ => {} } @@ -316,6 +330,15 @@ async fn chain_follower( } }); + set.spawn({ + let state_manager = state_manager.shallow_clone(); + let last_drand_entry = last_drand_entry.clone(); + let cancellation_token = cancellation_token.clone(); + async move { + drand_gossip_watchdog(state_manager, last_drand_entry, cancellation_token).await; + } + }); + // Forward tipsets from miners into the state machine. set.spawn({ let state_changed = state_changed.clone(); @@ -473,6 +496,109 @@ async fn chain_follower( Ok(()) } +/// Validate and verify a `drand` beacon entry received over `gossipsub`, recording +/// the arrival time of verified entries for [`drand_gossip_watchdog`]. +fn handle_drand_entry( + entry: BeaconEntry, + drand_verify_limiter: &Arc, + beacon_schedule: &Arc, + last_drand_entry: &Arc, +) { + if entry.round() == 0 || entry.signature().is_empty() { + return; + } + let Ok(permit) = drand_verify_limiter.shallow_clone().try_acquire_owned() else { + debug!( + round = entry.round(), + "dropping drand entry: too many verifications in flight" + ); + return; + }; + let beacon_schedule = beacon_schedule.clone(); + let last_drand_entry = last_drand_entry.clone(); + tokio::task::spawn_blocking(move || { + let _permit = permit; + let Some(beacon) = beacon_schedule.unchained_beacon() else { + return; + }; + + if matches!( + beacon.verify_entries(std::slice::from_ref(&entry), &BeaconEntry::default()), + Ok(true) + ) { + info!(round = entry.round(), "verified drand entry from gossipsub"); + last_drand_entry.store(Utc::now().timestamp().max(0) as u64, Ordering::Relaxed); + } else { + debug!( + round = entry.round(), + "received invalid drand entry over gossipsub" + ); + } + }); +} + +/// `drand` `gossipsub` is stale when no entry has ever been +/// verified (`last_seen == 0`), or the last one is at least a deadline old. +fn drand_gossip_is_stale(last_seen: u64, now: u64, deadline_secs: u64) -> bool { + last_seen == 0 || now.saturating_sub(last_seen) >= deadline_secs +} + +/// Watch the `drand` `gossipsub` topic for staleness: if a `drand` beacon entry +/// is not received in half a chain epoch then we consider it stale for +/// that epoch and fall back to fetching the beacon over HTTP. +async fn drand_gossip_watchdog( + state_manager: StateManager, + last_drand_entry: Arc, + cancellation_token: CancellationToken, +) { + if state_manager.beacon_schedule().unchained_beacon().is_none() { + return; + } + + let deadline = + Duration::from_secs(u64::from(state_manager.chain_config().block_delay_secs).div_ceil(2)); + + let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + deadline, deadline); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + while cancellation_token + .run_until_cancelled(ticker.tick()) + .await + .is_some() + { + let Some(beacon) = state_manager.beacon_schedule().unchained_beacon() else { + continue; + }; + + let last_seen = last_drand_entry.load(Ordering::Relaxed); + let now = Utc::now().timestamp().max(0) as u64; + if !drand_gossip_is_stale(last_seen, now, deadline.as_secs()) { + continue; + } + + let epoch = state_manager.heaviest_tipset().epoch() + 1; + let network_version = state_manager.get_network_version(epoch); + let round = match beacon.max_beacon_round_for_epoch(network_version, epoch) { + Ok(round) => round, + Err(e) => { + debug!("no drand round for epoch {epoch}: {e:#}"); + continue; + } + }; + + // Inside the cancellation scope: `entry` retries with a 15s timeout across every + // configured server, so an in-flight fetch would otherwise hold up `join_all`. + match cancellation_token + .run_until_cancelled(beacon.entry(round)) + .await + { + None => return, + Some(Err(e)) => debug!("drand HTTP fallback for round {round} failed: {e:#}"), + Some(Ok(_)) => {} + } + } +} + // Increment the gossipsub event metrics. fn inc_gossipsub_event_metrics(event: &NetworkEvent) { let label = match event { @@ -485,6 +611,7 @@ fn inc_gossipsub_event_metrics(event: &NetworkEvent) { NetworkEvent::PubsubMessage { message } => match message { PubsubMessage::Block(_) => metrics::values::PUBSUB_BLOCK, PubsubMessage::Message(_) => metrics::values::PUBSUB_MESSAGE, + PubsubMessage::DrandEntry { .. } => metrics::values::PUBSUB_DRAND_ENTRY, }, NetworkEvent::ChainExchangeRequestOutbound => { metrics::values::CHAIN_EXCHANGE_REQUEST_OUTBOUND @@ -594,6 +721,16 @@ static MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES: LazyLock = LazyLock::new(| .min(Semaphore::MAX_PERMITS) }); +/// Concurrency cap for `drand` entries verified from `gossipsub`. Excess is dropped, not queued. +static MAX_CONCURRENT_DRAND_VERIFICATIONS: LazyLock = LazyLock::new(|| { + env_or_default_logged( + "FOREST_MAX_CONCURRENT_DRAND_VERIFICATIONS", + nonzero!(4_usize), + ) + .get() + .min(Semaphore::MAX_PERMITS) +}); + /// Fetches a tipset off the event loop, forwarding a success into `tipset_sender` /// (the same channel miner tipsets use). Any `permit` is held for the fetch. fn spawn_tipset_fetch( @@ -1543,4 +1680,120 @@ mod tests { (block_cid, BlockValidationOutcome::Applied) ); } + + #[test] + fn drand_gossip_staleness() { + let deadline_secs = 15; + + // No entry ever verified: stale from the very first tick. + assert!(drand_gossip_is_stale(0, 100, deadline_secs)); + + // A recent entry is fresh. + assert!(!drand_gossip_is_stale(135, 140, deadline_secs)); + + // An entry exactly `deadline` old counts as stale (inclusive bound). + assert!(drand_gossip_is_stale(135, 150, deadline_secs)); + assert!(!drand_gossip_is_stale(136, 150, deadline_secs)); + + // A clock that went backwards must not underflow into stale. + assert!(!drand_gossip_is_stale(150, 140, deadline_secs)); + } + + use crate::beacon::{ + BeaconPoint, BeaconSchedule, + tests::fake_drand::{ + FAKE_DRAND_GENESIS_TIME, FAKE_DRAND_PERIOD, FakeDrand, TEST_FIL_BLOCK_DELAY, + TEST_FIL_GENESIS_TIME, + }, + }; + + fn fake_drand_schedule() -> (FakeDrand, Arc) { + let drand = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let beacon = drand.beacon(TEST_FIL_GENESIS_TIME, TEST_FIL_BLOCK_DELAY); + ( + drand, + Arc::new(BeaconSchedule(vec![BeaconPoint::new(0, beacon)])), + ) + } + + async fn wait_until(mut cond: impl FnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(5), async { + while !cond() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("condition not reached in time"); + } + + #[tokio::test] + async fn drand_entry_verified_and_recorded() { + let (drand, schedule) = fake_drand_schedule(); + let limiter = Arc::new(Semaphore::new(1)); + let last = Arc::new(AtomicU64::new(0)); + + handle_drand_entry(drand.entry(7), &limiter, &schedule, &last); + + wait_until(|| last.load(Ordering::Relaxed) != 0).await; + wait_until(|| limiter.available_permits() == 1).await; + } + + #[tokio::test] + async fn drand_entry_with_invalid_signature_is_dropped() { + let (drand, schedule) = fake_drand_schedule(); + let limiter = Arc::new(Semaphore::new(1)); + let last = Arc::new(AtomicU64::new(0)); + + // Round 7 carrying round 8's signature: well-formed but fails verification. + let forged = BeaconEntry::new(7, drand.entry(8).signature().to_vec()); + handle_drand_entry(forged, &limiter, &schedule, &last); + + // The returned permit proves the verification task ran to completion. + wait_until(|| limiter.available_permits() == 1).await; + assert_eq!(last.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn drand_entry_rejected_before_verification() { + let (drand, schedule) = fake_drand_schedule(); + let limiter = Arc::new(Semaphore::new(1)); + let last = Arc::new(AtomicU64::new(0)); + + // Round zero and an empty signature are rejected synchronously: no + // permit is ever taken, so nothing can be in flight afterwards. + let round_zero = BeaconEntry::new(0, drand.entry(1).signature().to_vec()); + handle_drand_entry(round_zero, &limiter, &schedule, &last); + handle_drand_entry(BeaconEntry::new(1, vec![]), &limiter, &schedule, &last); + + assert_eq!(limiter.available_permits(), 1); + assert_eq!(last.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn drand_entry_dropped_when_verifications_saturated() { + let (drand, schedule) = fake_drand_schedule(); + let limiter = Arc::new(Semaphore::new(1)); + let last = Arc::new(AtomicU64::new(0)); + + let _held = limiter.clone().try_acquire_owned().unwrap(); + handle_drand_entry(drand.entry(7), &limiter, &schedule, &last); + + // Dropped synchronously: the held permit was not stolen and no + // verification was spawned. + assert_eq!(limiter.available_permits(), 0); + assert_eq!(last.load(Ordering::Relaxed), 0); + } + + #[tokio::test] + async fn drand_entry_dropped_without_unchained_beacon() { + let (drand, _) = fake_drand_schedule(); + let schedule = Arc::new(BeaconSchedule(vec![])); + let limiter = Arc::new(Semaphore::new(1)); + let last = Arc::new(AtomicU64::new(0)); + + handle_drand_entry(drand.entry(7), &limiter, &schedule, &last); + + wait_until(|| limiter.available_permits() == 1).await; + assert_eq!(last.load(Ordering::Relaxed), 0); + } } diff --git a/src/chain_sync/metrics.rs b/src/chain_sync/metrics.rs index 4f6c715aa7f3..78fdc4431de2 100644 --- a/src/chain_sync/metrics.rs +++ b/src/chain_sync/metrics.rs @@ -94,6 +94,8 @@ pub mod values { Libp2pMessageKindLabel::new("pubsub_message_block"); pub const PUBSUB_MESSAGE: Libp2pMessageKindLabel = Libp2pMessageKindLabel::new("pubsub_message_message"); + pub const PUBSUB_DRAND_ENTRY: Libp2pMessageKindLabel = + Libp2pMessageKindLabel::new("pubsub_message_drand_entry"); pub const CHAIN_EXCHANGE_REQUEST_OUTBOUND: Libp2pMessageKindLabel = Libp2pMessageKindLabel::new("chain_exchange_request_out"); pub const CHAIN_EXCHANGE_RESPONSE_INBOUND: Libp2pMessageKindLabel = diff --git a/src/libp2p/behaviour.rs b/src/libp2p/behaviour.rs index df9c1c1cda31..37cf9df89af6 100644 --- a/src/libp2p/behaviour.rs +++ b/src/libp2p/behaviour.rs @@ -10,8 +10,8 @@ use super::{ PeerManager, discovery::{DerivedDiscoveryBehaviourEvent, DiscoveryEvent, PeerInfo}, }; -use crate::libp2p_bitswap::BitswapBehaviour; use crate::utils::{encoding::blake2b_256, version::FOREST_VERSION_STRING}; +use crate::{libp2p::PubsubTopicCfg, libp2p_bitswap::BitswapBehaviour}; use crate::{ libp2p::{ chain_exchange::ChainExchangeBehaviour, @@ -79,11 +79,13 @@ const MAX_SUBSCRIPTIONS_PER_REQUEST: usize = 100; /// Filter accepting only Forest's topics, bounded in count and per request. pub(in crate::libp2p) fn build_subscription_filter( - network_name: &GenesisNetworkName, + cfg: PubsubTopicCfg<'_>, ) -> MaxCountSubscriptionFilter { - let allowed: Vec<_> = crate::libp2p::pubsub_topics(network_name) - .map(|t| t.hash()) + let allowed: Vec<_> = crate::libp2p::pubsub_topics(cfg) + .iter() + .map(|(_, t)| t.hash()) .collect(); + MaxCountSubscriptionFilter { // Whitelisted topics are the only ones counted, so their number is an // exact, self-maintaining bound. @@ -95,7 +97,7 @@ pub(in crate::libp2p) fn build_subscription_filter( pub(in crate::libp2p) fn build_gossipsub( local_key: &Keypair, - network_name: &GenesisNetworkName, + cfg: PubsubTopicCfg<'_>, ) -> anyhow::Result { let mut gs_config_builder = gossipsub::ConfigBuilder::default(); gs_config_builder.max_transmit_size(1 << 20); @@ -109,15 +111,12 @@ pub(in crate::libp2p) fn build_gossipsub( let mut gossipsub = Gossipsub::new_with_subscription_filter( MessageAuthenticity::Signed(local_key.clone()), gossipsub_config, - build_subscription_filter(network_name), + build_subscription_filter(cfg), ) .map_err(anyhow::Error::msg)?; gossipsub - .with_peer_score( - build_peer_score_params(network_name), - build_peer_score_threshold(), - ) + .with_peer_score(build_peer_score_params(cfg), build_peer_score_threshold()) .map_err(anyhow::Error::msg)?; Ok(gossipsub) @@ -128,6 +127,7 @@ impl ForestBehaviour { local_key: &Keypair, config: &Libp2pConfig, network_name: &GenesisNetworkName, + gossipsub: Gossipsub, peer_manager: Arc, ) -> anyhow::Result { const MAX_ESTABLISHED_PER_PEER: u32 = 4; @@ -146,8 +146,6 @@ impl ForestBehaviour { let max_concurrent_request_response_streams = (config.target_peer_count as usize) .saturating_mul(*MAX_CONCURRENT_REQUEST_RESPONSE_STREAMS_PER_PEER); - let gossipsub = build_gossipsub(local_key, network_name)?; - let bitswap = BitswapBehaviour::new( &[ "/chain/ipfs/bitswap/1.2.0", diff --git a/src/libp2p/gossip_params.rs b/src/libp2p/gossip_params.rs index 96a1ba54d1af..3da6dcc92d1b 100644 --- a/src/libp2p/gossip_params.rs +++ b/src/libp2p/gossip_params.rs @@ -7,9 +7,7 @@ use libp2p::gossipsub::{ PeerScoreParams, PeerScoreThresholds, TopicScoreParams, score_parameter_decay, }; -use strum::IntoEnumIterator as _; - -use crate::{libp2p::PubsubTopic, networks::GenesisNetworkName}; +use crate::libp2p::{PubsubTopic, PubsubTopicCfg, pubsub_topics}; // All these parameters are copied from what Lotus has set for their Topic // scores. They are currently unused because enabling them causes GossipSub @@ -80,18 +78,17 @@ fn build_block_topic_config() -> TopicScoreParams { } } -pub(in crate::libp2p) fn build_peer_score_params( - network_name: &GenesisNetworkName, -) -> PeerScoreParams { +pub(in crate::libp2p) fn build_peer_score_params(cfg: PubsubTopicCfg<'_>) -> PeerScoreParams { #[allow(clippy::disallowed_types)] let mut psp_topics = std::collections::HashMap::new(); - for topic in PubsubTopic::iter() { - let params = match topic { + for (variant, topic) in pubsub_topics(cfg) { + let params = match variant { PubsubTopic::Blocks => build_block_topic_config(), PubsubTopic::Messages => build_msg_topic_config(), + PubsubTopic::Drand => Default::default(), }; - psp_topics.insert(topic.ident(network_name).hash(), params); + psp_topics.insert(topic.hash(), params); } PeerScoreParams { diff --git a/src/libp2p/mod.rs b/src/libp2p/mod.rs index f9002fbd16c3..d7eb355c289e 100644 --- a/src/libp2p/mod.rs +++ b/src/libp2p/mod.rs @@ -25,5 +25,6 @@ pub use self::{config::*, peer_manager::*, service::*}; #[cfg(test)] mod tests { mod decode_test; + mod drand_gossip_tests; mod gossipsub_filter_test; } diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index f0ac4b257d07..c81581e745ec 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -3,6 +3,7 @@ use std::time::{Duration, UNIX_EPOCH}; +use crate::beacon::{BeaconEntry, PublicRandResponse}; use crate::prelude::*; use crate::{blocks::GossipBlock, rpc::net::NetInfoResult}; use crate::{chain::ChainStore, utils::encoding::from_slice_with_fallback}; @@ -15,6 +16,7 @@ use ahash::{HashMap, HashSet}; use anyhow::Context as _; use flume::Sender; use futures::{select, stream::StreamExt as _}; +use libp2p::gossipsub::TopicHash; pub use libp2p::gossipsub::{IdentTopic, Topic}; use libp2p::{ PeerId, Swarm, SwarmBuilder, @@ -30,11 +32,14 @@ use libp2p::{ tcp, yamux, }; use nonzero_ext::nonzero; +use quick_protobuf::{BytesReader, MessageRead as _}; + use tokio_stream::wrappers::IntervalStream; use tracing::{debug, error, info, trace, warn}; use super::{ ForestBehaviour, ForestBehaviourEvent, Libp2pConfig, + behaviour::build_gossipsub, chain_exchange::{ChainExchangeRequest, ChainExchangeResponse, make_chain_exchange_response}, discovery::{DerivedDiscoveryBehaviourEvent, PeerInfo}, }; @@ -78,31 +83,51 @@ crate::def_is_env_truthy!(libp2p_metrics_enabled, "FOREST_LIBP2P_METRICS_ENABLED pub const PUBSUB_BLOCK_STR: &str = "/fil/blocks"; /// `Gossipsub` Filecoin messages topic identifier. pub const PUBSUB_MSG_STR: &str = "/fil/msgs"; +/// `Gossipsub` `drand` randomness topic identifier. +pub const PUBSUB_DRAND_STR: &str = "/drand/pubsub/v0.0.0"; /// Gossipsub topics Forest uses. Subscription, the subscription-filter /// whitelist, and peer-score params all iterate the variants, so adding one is /// handled everywhere. -#[derive(Copy, Clone, Debug, strum::EnumIter, derive_more::Display)] +#[derive(Copy, Clone, Debug, strum::EnumIter, derive_more::Display, Eq, PartialEq)] pub enum PubsubTopic { #[display("{PUBSUB_BLOCK_STR}")] Blocks, #[display("{PUBSUB_MSG_STR}")] Messages, + #[display("{PUBSUB_DRAND_STR}")] + Drand, } -impl PubsubTopic { - /// Full topic on `network_name`, e.g. `/fil/blocks/`. - pub fn ident(self, network_name: impl std::fmt::Display) -> IdentTopic { - IdentTopic::new(format!("{self}/{network_name}")) - } +#[derive(Clone, Copy)] +pub struct PubsubTopicCfg<'a> { + pub network_name: &'a GenesisNetworkName, + pub drand_chain_hashes: &'a [String], } /// All gossipsub topics on `network_name`. -pub fn pubsub_topics( - network_name: impl std::fmt::Display + Copy, -) -> impl Iterator { +pub fn pubsub_topics(cfg: PubsubTopicCfg<'_>) -> Vec<(PubsubTopic, IdentTopic)> { use strum::IntoEnumIterator as _; - PubsubTopic::iter().map(move |t| t.ident(network_name)) + + let mut topics = Vec::new(); + for kind in PubsubTopic::iter() { + match kind { + PubsubTopic::Blocks | PubsubTopic::Messages => { + topics.push(( + kind, + IdentTopic::new(format!("{kind}/{}", cfg.network_name)), + )); + } + PubsubTopic::Drand => { + topics.extend( + cfg.drand_chain_hashes + .iter() + .map(move |h| (kind, IdentTopic::new(format!("{kind}/{h}")))), + ); + } + } + } + topics } pub const BITSWAP_TIMEOUT: Duration = Duration::from_secs(30); @@ -137,6 +162,8 @@ pub enum PubsubMessage { Block(GossipBlock), /// Messages that come over the message topic Message(SignedMessage), + /// Messages that come over the `drand` topic + DrandEntry(BeaconEntry), } /// Messages into the service to handle. @@ -191,8 +218,8 @@ pub struct Libp2pService { network_sender_in: Sender, network_receiver_out: flume::Receiver, network_sender_out: Sender, - network_name: String, genesis_cid: Cid, + pubsub_topic_kinds: HashMap, } impl Libp2pService { @@ -204,9 +231,19 @@ impl Libp2pService { network_name: GenesisNetworkName, genesis_cid: Cid, ) -> anyhow::Result { - let behaviour = - ForestBehaviour::new(&net_keypair, &config, &network_name, peer_manager.clone()) - .await?; + let pubsub_topic_cfg = PubsubTopicCfg { + network_name: &network_name, + drand_chain_hashes: &cs.chain_config().drand_gossip_chain_hashes(), + }; + let gossipsub = build_gossipsub(&net_keypair, pubsub_topic_cfg)?; + let behaviour = ForestBehaviour::new( + &net_keypair, + &config, + &network_name, + gossipsub, + peer_manager.clone(), + ) + .await?; let mut swarm = SwarmBuilder::with_existing_identity(net_keypair) .with_tokio() .with_tcp( @@ -227,11 +264,15 @@ impl Libp2pService { .build(); // Subscribe to gossipsub topics with the network name suffix - for topic in pubsub_topics(&network_name) { + // and for drand uses the current drand network hash + let mut pubsub_topic_kinds = HashMap::default(); + for (kind, topic) in pubsub_topics(pubsub_topic_cfg) { swarm .behaviour_mut() .subscribe(&topic) .with_context(|| format!("Failed to subscribe gossipsub topic {topic}"))?; + info!("Subscribed to gossipsub topic {topic} ({kind:?})"); + pubsub_topic_kinds.insert(topic.hash(), kind); } let (network_sender_in, network_receiver_in) = flume::unbounded(); @@ -282,8 +323,8 @@ impl Libp2pService { network_sender_in, network_receiver_out, network_sender_out, - network_name: network_name.into(), genesis_cid, + pubsub_topic_kinds, }) } @@ -302,8 +343,8 @@ impl Libp2pService { let mut network_stream = self.network_receiver_in.stream().fuse(); let mut interval = IntervalStream::new(tokio::time::interval(Duration::from_secs(15))).fuse(); - let pubsub_block_str = PubsubTopic::Blocks.ident(&self.network_name).to_string(); - let pubsub_msg_str = PubsubTopic::Messages.ident(&self.network_name).to_string(); + + let pubsub_topic_kinds = self.pubsub_topic_kinds; let (cx_response_tx, cx_response_rx) = flume::unbounded(); @@ -345,8 +386,7 @@ impl Libp2pService { &self.genesis_cid, &self.network_sender_out, cx_response_tx.clone(), - &pubsub_block_str, - &pubsub_msg_str,).await; + &pubsub_topic_kinds).await; }, None => { break; }, _ => { }, @@ -360,7 +400,8 @@ impl Libp2pService { bitswap_request_manager.shallow_clone(), message, &self.network_sender_out, - &self.peer_manager).await; + &self.peer_manager, + ).await; } None => { break; } }, @@ -651,11 +692,10 @@ async fn handle_discovery_event( } } -async fn handle_gossip_event( +pub(in crate::libp2p) async fn handle_gossip_event( e: gossipsub::Event, network_sender_out: &Sender, - pubsub_block_str: &str, - pubsub_msg_str: &str, + pubsub_topic_kinds: &HashMap, ) { if let gossipsub::Event::Message { propagation_source: source, @@ -663,11 +703,12 @@ async fn handle_gossip_event( .. } = e { - let topic = message.topic.as_str(); + let topic = message.topic; let message = message.data; trace!("Got a Gossip Message from {:?}", source); - if topic == pubsub_block_str { - match from_slice_with_fallback::(&message) { + + match pubsub_topic_kinds.get(&topic) { + Some(PubsubTopic::Blocks) => match from_slice_with_fallback::(&message) { Ok(b) => { emit_event( network_sender_out, @@ -680,24 +721,52 @@ async fn handle_gossip_event( Err(e) => { warn!("Gossip Block from peer {source:?} could not be deserialized: {e:#}",); } - } - } else if topic == pubsub_msg_str { - match from_slice_with_fallback::(&message) { - Ok(m) => { - emit_event( - network_sender_out, - NetworkEvent::PubsubMessage { - message: PubsubMessage::Message(m), - }, - ) - .await; + }, + Some(PubsubTopic::Messages) => { + match from_slice_with_fallback::(&message) { + Ok(m) => { + emit_event( + network_sender_out, + NetworkEvent::PubsubMessage { + message: PubsubMessage::Message(m), + }, + ) + .await; + } + Err(e) => { + warn!( + "Gossip Message from peer {source:?} could not be deserialized: {e:#}" + ); + } } - Err(e) => { - warn!("Gossip Message from peer {source:?} could not be deserialized: {e:#}"); + } + Some(PubsubTopic::Drand) => { + let mut reader = BytesReader::from_bytes(&message); + match PublicRandResponse::from_reader(&mut reader, &message) { + Ok(r) => { + info!( + "Received drand round {} from peer {source:?} on {topic}", + r.round + ); + emit_event( + network_sender_out, + NetworkEvent::PubsubMessage { + message: PubsubMessage::DrandEntry(BeaconEntry::new( + r.round, + r.signature, + )), + }, + ) + .await; + } + Err(e) => { + warn!( + "Gossip drand entry from peer {source:?} could not be decoded: {e:#}" + ); + } } } - } else { - warn!("Getting gossip messages from unknown topic: {topic}"); + None => warn!("Getting gossip messages from unknown topic: {topic}"), } } } @@ -923,8 +992,7 @@ async fn handle_forest_behaviour_event( request_response::ResponseChannel, ChainExchangeResponse, )>, - pubsub_block_str: &str, - pubsub_msg_str: &str, + pubsub_topic_kinds: &HashMap, ) { match event { ForestBehaviourEvent::Discovery(discovery_out) => { @@ -937,7 +1005,7 @@ async fn handle_forest_behaviour_event( .await } ForestBehaviourEvent::Gossipsub(e) => { - handle_gossip_event(e, network_sender_out, pubsub_block_str, pubsub_msg_str).await + handle_gossip_event(e, network_sender_out, pubsub_topic_kinds).await } ForestBehaviourEvent::Hello(rr_event) => { let behaviour_mut = swarm.behaviour_mut(); diff --git a/src/libp2p/tests/drand_gossip_tests.rs b/src/libp2p/tests/drand_gossip_tests.rs new file mode 100644 index 000000000000..546bafe7356b --- /dev/null +++ b/src/libp2p/tests/drand_gossip_tests.rs @@ -0,0 +1,266 @@ +use std::{sync::Arc, time::Duration}; + +use futures::StreamExt as _; +use libp2p::{ + Swarm, + gossipsub::{self, IdentTopic}, + swarm::SwarmEvent, +}; +use libp2p_swarm_test::SwarmExt as _; +use quick_protobuf::{BytesReader, MessageRead}; + +use crate::libp2p::{ + NetworkEvent, PUBSUB_DRAND_STR, PubsubMessage, PubsubTopic, build_gossipsub, + service::handle_gossip_event, +}; +use crate::networks::GenesisNetworkName; +use crate::{ + beacon::{ + Beacon, BeaconEntry, PublicRandResponse, + tests::fake_drand::{ + FAKE_DRAND_GENESIS_TIME, FAKE_DRAND_PERIOD, FakeDrand, TEST_FIL_BLOCK_DELAY, + TEST_FIL_GENESIS_TIME, + }, + }, + libp2p::{Gossipsub, PubsubTopicCfg}, +}; + +#[tokio::test] +async fn gossip_rounds_are_verified_and_cached() { + let drand = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + + let beacon = drand.beacon(TEST_FIL_GENESIS_TIME, TEST_FIL_BLOCK_DELAY); + let hash = drand.chain_info_hash(); + + let topic = IdentTopic::new(format!("{PUBSUB_DRAND_STR}/{hash}")); + + // `PubsubTopicCfg` borrows, so these have to outlive the swarm construction. + // The whitelist must carry the *fake* chain hash, otherwise the node refuses + // to subscribe to the topic the relay publishes on. + let network_name: GenesisNetworkName = "testdrandgossipsub".into(); + let drand_chain_hashes = vec![hash]; + let cfg = PubsubTopicCfg { + network_name: &network_name, + drand_chain_hashes: &drand_chain_hashes, + }; + + let mut node = Swarm::new_ephemeral_tokio(|id| build_gossipsub(&id, cfg).unwrap()); + + let mut relay = Swarm::new_ephemeral_tokio(|id| { + gossipsub::Behaviour::new( + gossipsub::MessageAuthenticity::Signed(id), + gossipsub::ConfigBuilder::default().build().unwrap(), + ) + .unwrap() + }); + + node.listen().with_memory_addr_external().await; + relay.connect(&mut node).await; + + relay.behaviour_mut().subscribe(&topic).unwrap(); + node.behaviour_mut().subscribe(&topic).unwrap(); + + wait_until_meshed(&mut node, &mut relay, &topic).await; + + let mut received = Vec::new(); + for round in 1..=5u64 { + relay + .behaviour_mut() + .publish(topic.clone(), drand.to_protobuf(round)) + .unwrap(); + let data = tokio::time::timeout(Duration::from_secs(5), async { + loop { + tokio::select! { + _ = relay.select_next_some() => {}, + ev = node.select_next_some() => { + if let SwarmEvent::Behaviour(gossipsub::Event::Message { message, .. }) = ev { + break message.data; + } + } + } + } + }).await.expect("no gossip message"); + + let mut reader = BytesReader::from_bytes(&data); + let decoded = PublicRandResponse::from_reader(&mut reader, &data).unwrap(); + received.push(BeaconEntry::new(decoded.round, decoded.signature)); + } + + assert_eq!(received.len(), 5); + assert!( + beacon + .verify_entries(&received, &BeaconEntry::default()) + .unwrap() + ); + + // verify every round is now served from cache. + for round in 1..=5u64 { + assert_eq!(beacon.entry(round).await.unwrap().round(), round); + } +} + +async fn wait_until_meshed( + node: &mut Swarm, + relay: &mut Swarm, + topic: &IdentTopic, +) { + let hash = topic.hash(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if relay.behaviour().mesh_peers(&hash).next().is_some() + && node.behaviour().mesh_peers(&hash).next().is_some() + { + return; + } + + tokio::select! { + _ = node.select_next_some() => {} + _ = relay.select_next_some() => {} + } + } + }) + .await + .expect("drand topic mesh never formed"); +} + +#[tokio::test] +async fn silence_past_deadline_fallback_to_http() { + use axum::{Json, Router, extract::Path, routing::get}; + use std::sync::atomic::{AtomicUsize, Ordering}; + + // mocks drand HTTP + let hits = Arc::new(AtomicUsize::new(0)); + let signer = Arc::new(FakeDrand::new( + vec![], + FAKE_DRAND_PERIOD, + FAKE_DRAND_GENESIS_TIME, + )); + + let app = { + let (hits, signer) = (hits.clone(), signer.clone()); + Router::new().route( + "/{hash}/public/{round}", + get(move |Path((_hash, round)): Path<(String, u64)>| { + let (hits, signer) = (hits.clone(), signer.clone()); + async move { + hits.fetch_add(1, Ordering::Relaxed); + Json(signer.to_json(round)) + } + }), + ) + }; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base: url::Url = format!("http://{}/", listener.local_addr().unwrap()) + .parse() + .unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let drand = FakeDrand::new(vec![base], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let beacon = drand.beacon(TEST_FIL_GENESIS_TIME, TEST_FIL_BLOCK_DELAY); + + // sanity check + assert_eq!(hits.load(Ordering::Relaxed), 0); + + // first fetch + let fetched = beacon.entry(42).await.unwrap(); + assert_eq!(fetched.round(), 42); + assert_eq!(hits.load(Ordering::Relaxed), 1, "expected one HTTP fetch"); + + // second call, same round, should fetch from cache + beacon.entry(42).await.unwrap(); + assert_eq!( + hits.load(Ordering::Relaxed), + 1, + "second call must not reach HTTP" + ); +} + +fn gossip_message_event(data: Vec, topic: gossipsub::TopicHash) -> gossipsub::Event { + gossipsub::Event::Message { + propagation_source: libp2p::PeerId::random(), + message_id: gossipsub::MessageId::new(b"test"), + message: gossipsub::Message { + source: None, + data, + sequence_number: None, + topic, + }, + } +} + +fn drand_topic_kinds( + drand: &FakeDrand, +) -> ( + IdentTopic, + ahash::HashMap, +) { + let topic = IdentTopic::new(format!("{PUBSUB_DRAND_STR}/{}", drand.chain_info_hash())); + let mut kinds = ahash::HashMap::default(); + kinds.insert(topic.hash(), PubsubTopic::Drand); + (topic, kinds) +} + +#[tokio::test] +async fn gossip_drand_message_is_decoded_and_emitted() { + let drand = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let (topic, kinds) = drand_topic_kinds(&drand); + let (tx, rx) = flume::unbounded(); + + // The payload is a bare `PublicRandResponse`, no length prefix (regression: + // decoding used to assume a prefix and reject every live relay message). + handle_gossip_event( + gossip_message_event(drand.to_protobuf(42), topic.hash()), + &tx, + &kinds, + ) + .await; + + match rx.try_recv().expect("no event emitted") { + NetworkEvent::PubsubMessage { + message: PubsubMessage::DrandEntry(entry), + } => { + assert_eq!(entry.round(), 42); + assert_eq!(entry.signature(), drand.entry(42).signature()); + } + other => panic!("unexpected event: {other:?}"), + } +} + +#[tokio::test] +async fn gossip_drand_malformed_payload_is_dropped() { + let drand = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let (topic, kinds) = drand_topic_kinds(&drand); + let (tx, rx) = flume::unbounded(); + + handle_gossip_event( + gossip_message_event(vec![0xff, 0xff, 0xff], topic.hash()), + &tx, + &kinds, + ) + .await; + + assert!( + rx.try_recv().is_err(), + "malformed payload must emit nothing" + ); +} + +#[tokio::test] +async fn gossip_message_on_unknown_topic_is_dropped() { + let drand = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let (_, kinds) = drand_topic_kinds(&drand); + let (tx, rx) = flume::unbounded(); + + handle_gossip_event( + gossip_message_event( + drand.to_protobuf(1), + gossipsub::TopicHash::from_raw("/unknown/topic"), + ), + &tx, + &kinds, + ) + .await; + + assert!(rx.try_recv().is_err(), "unknown topic must emit nothing"); +} diff --git a/src/libp2p/tests/gossipsub_filter_test.rs b/src/libp2p/tests/gossipsub_filter_test.rs index 8b0d05ed0eec..774bbdde6480 100644 --- a/src/libp2p/tests/gossipsub_filter_test.rs +++ b/src/libp2p/tests/gossipsub_filter_test.rs @@ -14,14 +14,51 @@ use libp2p::{ }; use libp2p_swarm_test::SwarmExt as _; -use crate::libp2p::{Gossipsub, build_gossipsub, build_subscription_filter, pubsub_topics}; +use crate::libp2p::{ + Gossipsub, PubsubTopicCfg, build_gossipsub, build_subscription_filter, pubsub_topics, +}; +use crate::networks::GenesisNetworkName; const NETWORK: &str = "testnetname"; +/// quicknet, the one unchained drand network Forest subscribes to. +const DRAND_HASH: &str = "52db9ba70e0cc0f6eaf7803dd07447a1f5477735fd3f661792ba94600c84e971"; + +/// Owns what [`PubsubTopicCfg`] borrows. +pub(in crate::libp2p) struct TopicCfgOwner { + network_name: GenesisNetworkName, + drand_chain_hashes: Vec, +} + +impl TopicCfgOwner { + pub(in crate::libp2p) fn new() -> Self { + Self { + network_name: NETWORK.into(), + drand_chain_hashes: vec![DRAND_HASH.to_string()], + } + } + + pub(in crate::libp2p) fn cfg(&self) -> PubsubTopicCfg<'_> { + PubsubTopicCfg { + network_name: &self.network_name, + drand_chain_hashes: &self.drand_chain_hashes, + } + } +} + +/// Every topic the node should accept, drand included. +fn allowed_topics() -> Vec { + let owner = TopicCfgOwner::new(); + pubsub_topics(owner.cfg()) + .into_iter() + .map(|(_, topic)| topic) + .collect() +} /// Swarm using Forest's subscription filter (the code under test). fn filtered_swarm() -> Swarm { + let owner = TopicCfgOwner::new(); Swarm::new_ephemeral_tokio(|identity| { - build_gossipsub(&identity, &NETWORK.into()).expect("failed to build gossipsub") + build_gossipsub(&identity, owner.cfg()).expect("failed to build gossipsub") }) } @@ -53,7 +90,7 @@ async fn only_whitelisted_topics_are_tracked() { let unlisted = IdentTopic::new(format!("/other/topic/{i}")); peer.behaviour_mut().subscribe(&unlisted).unwrap(); } - let allowed: Vec = pubsub_topics(NETWORK).collect(); + let allowed = allowed_topics(); for topic in &allowed { peer.behaviour_mut().subscribe(topic).unwrap(); } @@ -84,19 +121,22 @@ async fn only_whitelisted_topics_are_tracked() { #[test] fn filter_allows_only_whitelisted_topics() { - let mut filter = build_subscription_filter(&NETWORK.into()); - for topic in pubsub_topics(NETWORK) { + let owner = TopicCfgOwner::new(); + let mut filter = build_subscription_filter(owner.cfg()); + for topic in allowed_topics() { assert!(filter.can_subscribe(&topic.hash())); } assert!(!filter.can_subscribe(&IdentTopic::new("/cth/ulhu").hash())); assert!(!filter.can_subscribe(&TopicHash::from_raw("x".repeat(1 << 20)))); // Wrong network suffix must not match. assert!(!filter.can_subscribe(&IdentTopic::new("/fil/blocks/lovecraftnet").hash())); + assert!(!filter.can_subscribe(&IdentTopic::new("/drand/pubsub/v0.0.0/deadbeef").hash())); } #[test] fn filter_caps_are_set() { - let filter = build_subscription_filter(&NETWORK.into()); - assert_eq!(filter.max_subscribed_topics, pubsub_topics(NETWORK).count()); + let owner = TopicCfgOwner::new(); + let filter = build_subscription_filter(owner.cfg()); + assert_eq!(filter.max_subscribed_topics, allowed_topics().len()); assert_eq!(filter.max_subscriptions_per_request, 100); } diff --git a/src/networks/mod.rs b/src/networks/mod.rs index 75ee05bb5d6d..95a1f5d8779a 100644 --- a/src/networks/mod.rs +++ b/src/networks/mod.rs @@ -1,6 +1,7 @@ // Copyright 2019-2026 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT +use std::slice::Iter; use std::str::FromStr; use std::sync::LazyLock; @@ -486,16 +487,25 @@ impl ChainConfig { 0 } - pub fn get_beacon_schedule(&self, genesis_ts: u64) -> BeaconSchedule { - let ds_iter = match self.network { + fn drand_points(&self) -> Iter<'_, DrandPoint<'static>> { + match self.network { NetworkChain::Mainnet => mainnet::DRAND_SCHEDULE.iter(), NetworkChain::Calibnet => calibnet::DRAND_SCHEDULE.iter(), NetworkChain::Butterflynet => butterflynet::DRAND_SCHEDULE.iter(), NetworkChain::Devnet(_) => devnet::DRAND_SCHEDULE.iter(), - }; + } + } + pub fn drand_gossip_chain_hashes(&self) -> Vec { + self.drand_points() + .filter(|p| p.config.network_type.is_unchained()) + .map(|p| p.config.chain_info.hash.to_string()) + .collect() + } + + pub fn get_beacon_schedule(&self, genesis_ts: u64) -> BeaconSchedule { BeaconSchedule( - ds_iter + self.drand_points() .map(|dc| { BeaconPoint::new( dc.height,