Skip to content

Commit 68d3c4f

Browse files
perfogicperfogic
committed
fix(gossip): report validation for beacon messages
1 parent 003c550 commit 68d3c4f

6 files changed

Lines changed: 142 additions & 58 deletions

File tree

‎crates/networking/manager/src/gossipsub/handle.rs‎

Lines changed: 60 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
use std::time::{SystemTime, UNIX_EPOCH};
22

3-
use libp2p::gossipsub::Message;
3+
use libp2p::gossipsub::{Message, MessageAcceptance};
44
use ream_chain_beacon::beacon_chain::BeaconChain;
55
use ream_consensus_beacon::{
66
blob_sidecar::BlobIdentifier,
@@ -134,10 +134,13 @@ async fn import_gossip_attestation(
134134
let (attestation, should_process_attestation) = {
135135
let store = beacon_chain.store.lock().await;
136136
let head_root = store.get_head()?;
137-
let state =
137+
let mut state =
138138
store.db.state_provider().get(head_root)?.ok_or_else(|| {
139139
anyhow::anyhow!("No beacon state found for head root: {head_root}")
140140
})?;
141+
if state.slot < single_attestation.data.slot {
142+
state.process_slots(single_attestation.data.slot)?;
143+
}
141144
let attestation = single_attestation_to_attestation(single_attestation, &state)?;
142145

143146
store
@@ -165,13 +168,21 @@ fn forward_gossip_message(message: &Message, p2p_sender: &P2PSender, data: Vec<u
165168
});
166169
}
167170

171+
fn message_acceptance(validation_result: &ValidationResult) -> MessageAcceptance {
172+
match validation_result {
173+
ValidationResult::Accept => MessageAcceptance::Accept,
174+
ValidationResult::Reject(_) => MessageAcceptance::Reject,
175+
ValidationResult::Ignore(_) => MessageAcceptance::Ignore,
176+
}
177+
}
178+
168179
/// Dispatches a gossipsub message to its appropriate handler.
169180
pub async fn handle_gossipsub_message(
170181
message: Message,
171182
beacon_chain: &BeaconChain,
172183
cached_db: &BeaconCacheDB,
173184
p2p_sender: &P2PSender,
174-
) {
185+
) -> MessageAcceptance {
175186
match GossipsubMessage::decode(&message.topic, &message.data) {
176187
Ok(gossip_message) => match gossip_message {
177188
GossipsubMessage::BeaconBlock(signed_block) => {
@@ -189,7 +200,7 @@ pub async fn handle_gossipsub_message(
189200
};
190201
if let Err(err) = beacon_chain.process_tick(tick_time).await {
191202
warn!("Failed to process gossipsub tick before block validation: {err}");
192-
return;
203+
return MessageAcceptance::Ignore;
193204
}
194205

195206
let validation_result = match validate_gossip_beacon_block(
@@ -202,10 +213,11 @@ pub async fn handle_gossipsub_message(
202213
Ok(result) => result,
203214
Err(err) => {
204215
warn!("Failed to validate gossipsub beacon block: {err}");
205-
return;
216+
return MessageAcceptance::Ignore;
206217
}
207218
};
208219

220+
let acceptance = message_acceptance(&validation_result);
209221
match validation_result {
210222
ValidationResult::Accept => {
211223
let signed_block_bytes = signed_block.as_ssz_bytes();
@@ -221,45 +233,51 @@ pub async fn handle_gossipsub_message(
221233
warn!("Rejecting gossipsub beacon block: {reason}");
222234
}
223235
}
236+
acceptance
224237
}
225238
GossipsubMessage::BeaconAttestation((single_attestation, subnet_id)) => {
226239
trace!(
227240
"Beacon Attestation received over gossipsub: root: {}",
228241
single_attestation.tree_hash_root()
229242
);
230243

231-
match validate_beacon_attestation(
244+
let validation_result = match validate_beacon_attestation(
232245
&single_attestation,
233246
beacon_chain,
234247
subnet_id,
235248
cached_db,
236249
)
237250
.await
238251
{
239-
Ok(validation_result) => match validation_result {
240-
ValidationResult::Accept => {
241-
if let Err(err) =
242-
import_gossip_attestation(beacon_chain, &single_attestation).await
243-
{
244-
warn!("Failed to import gossipsub beacon attestation: {err}");
245-
}
246-
forward_gossip_message(
247-
&message,
248-
p2p_sender,
249-
single_attestation.as_ssz_bytes(),
250-
);
251-
}
252-
ValidationResult::Reject(reason) => {
253-
info!("Attestation rejected: {reason}");
254-
}
255-
ValidationResult::Ignore(reason) => {
256-
info!("Attestation ignored: {reason}");
257-
}
258-
},
252+
Ok(validation_result) => validation_result,
259253
Err(err) => {
260254
trace!("Could not validate attestation: {err}");
255+
return MessageAcceptance::Ignore;
256+
}
257+
};
258+
259+
let acceptance = message_acceptance(&validation_result);
260+
match validation_result {
261+
ValidationResult::Accept => {
262+
if let Err(err) =
263+
import_gossip_attestation(beacon_chain, &single_attestation).await
264+
{
265+
warn!("Failed to import gossipsub beacon attestation: {err}");
266+
}
267+
forward_gossip_message(
268+
&message,
269+
p2p_sender,
270+
single_attestation.as_ssz_bytes(),
271+
);
272+
}
273+
ValidationResult::Reject(reason) => {
274+
info!("Attestation rejected: {reason}");
275+
}
276+
ValidationResult::Ignore(reason) => {
277+
info!("Attestation ignored: {reason}");
261278
}
262279
}
280+
acceptance
263281
}
264282
GossipsubMessage::BlsToExecutionChange(signed_bls_to_execution_change) => {
265283
info!(
@@ -293,6 +311,7 @@ pub async fn handle_gossipsub_message(
293311
error!("Could not validate BLS to Execution Change: {err}");
294312
}
295313
}
314+
MessageAcceptance::Ignore
296315
}
297316
GossipsubMessage::AggregateAndProof(aggregate_and_proof) => {
298317
info!(
@@ -322,6 +341,7 @@ pub async fn handle_gossipsub_message(
322341
error!("Could not validate aggregate and proof: {err}");
323342
}
324343
}
344+
MessageAcceptance::Ignore
325345
}
326346
GossipsubMessage::SyncCommittee((sync_committee, subnet_id)) => {
327347
trace!(
@@ -351,6 +371,7 @@ pub async fn handle_gossipsub_message(
351371
error!("Could not validate sync committee message: {err}");
352372
}
353373
}
374+
MessageAcceptance::Ignore
354375
}
355376
GossipsubMessage::SyncCommitteeContributionAndProof(signed_contribution_and_proof) => {
356377
info!(
@@ -385,6 +406,7 @@ pub async fn handle_gossipsub_message(
385406
error!("Could not validate sync committee contribution and proof: {err}");
386407
}
387408
}
409+
MessageAcceptance::Ignore
388410
}
389411
GossipsubMessage::AttesterSlashing(attester_slashing) => {
390412
info!(
@@ -416,6 +438,7 @@ pub async fn handle_gossipsub_message(
416438
error!("Could not validate attester slashing: {err}");
417439
}
418440
}
441+
MessageAcceptance::Ignore
419442
}
420443
GossipsubMessage::ProposerSlashing(proposer_slashing) => {
421444
info!(
@@ -444,6 +467,7 @@ pub async fn handle_gossipsub_message(
444467
error!("Could not validate proposer slashing: {err}");
445468
}
446469
}
470+
MessageAcceptance::Ignore
447471
}
448472
GossipsubMessage::BlobSidecar(blob_sidecar) => {
449473
info!(
@@ -494,6 +518,7 @@ pub async fn handle_gossipsub_message(
494518
error!("Could not validate blob_sidecar: {err}");
495519
}
496520
}
521+
MessageAcceptance::Ignore
497522
}
498523
GossipsubMessage::DataColumnSidecar(data_column_sidecar) => {
499524
info!(
@@ -511,12 +536,12 @@ pub async fn handle_gossipsub_message(
511536
GossipTopicKind::DataColumnSidecar(id) => id,
512537
_ => {
513538
error!("Unexpected topic kind for data column sidecar");
514-
return;
539+
return MessageAcceptance::Ignore;
515540
}
516541
},
517542
Err(err) => {
518543
error!("Failed to parse topic for data column sidecar: {err}");
519-
return;
544+
return MessageAcceptance::Ignore;
520545
}
521546
};
522547

@@ -531,7 +556,7 @@ pub async fn handle_gossipsub_message(
531556
Ok(validation_result) => validation_result,
532557
Err(err) => {
533558
error!("Could not validate data_column_sidecar: {err}");
534-
return;
559+
return MessageAcceptance::Ignore;
535560
}
536561
};
537562

@@ -567,6 +592,7 @@ pub async fn handle_gossipsub_message(
567592
info!("Data column sidecar ignored: {reason}");
568593
}
569594
}
595+
MessageAcceptance::Ignore
570596
}
571597
GossipsubMessage::LightClientFinalityUpdate(light_client_finality_update) => {
572598
info!(
@@ -599,6 +625,7 @@ pub async fn handle_gossipsub_message(
599625
error!("Could not validate light client finality update: {err}");
600626
}
601627
}
628+
MessageAcceptance::Ignore
602629
}
603630
GossipsubMessage::LightClientOptimisticUpdate(light_client_optimistic_update) => {
604631
info!(
@@ -635,6 +662,7 @@ pub async fn handle_gossipsub_message(
635662
error!("Could not validate light client optimistic update: {err}");
636663
}
637664
}
665+
MessageAcceptance::Ignore
638666
}
639667
GossipsubMessage::VoluntaryExit(voluntary_exit) => {
640668
info!(
@@ -662,10 +690,12 @@ pub async fn handle_gossipsub_message(
662690
error!("Could not validate voluntary_exit: {err}");
663691
}
664692
}
693+
MessageAcceptance::Ignore
665694
}
666695
},
667696
Err(err) => {
668697
trace!("Failed to decode gossip message: {err:?}");
698+
MessageAcceptance::Reject
669699
}
670-
};
700+
}
671701
}

‎crates/networking/manager/src/gossipsub/validate/beacon_attestation.rs‎

Lines changed: 17 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -25,12 +25,26 @@ pub async fn validate_beacon_attestation(
2525
let store = beacon_chain.store.lock().await;
2626

2727
let head_root = store.get_head()?;
28-
let state: BeaconState = store
28+
let mut state: BeaconState = store
2929
.db
3030
.state_provider()
3131
.get(head_root)?
3232
.ok_or_else(|| anyhow!("No beacon state found for head root: {head_root}"))?;
3333

34+
let current_slot = store.get_current_slot()?;
35+
36+
// [IGNORE] attestation.data.slot is equal to or earlier than the current_slot (with a
37+
// MAXIMUM_GOSSIP_CLOCK_DISPARITY allowance)
38+
if attestation.data.slot > current_slot {
39+
return Ok(ValidationResult::Ignore(
40+
"Attestation is from a future slot".to_string(),
41+
));
42+
}
43+
44+
if state.slot < attestation.data.slot {
45+
state.process_slots(attestation.data.slot)?;
46+
}
47+
3448
let committee_index = attestation.committee_index;
3549
let committees_per_slot = state.get_committee_count_per_slot(attestation.data.target.epoch);
3650

@@ -58,27 +72,11 @@ pub async fn validate_beacon_attestation(
5872
));
5973
}
6074

61-
let block = store
62-
.db
63-
.block_provider()
64-
.get(head_root)?
65-
.ok_or_else(|| anyhow!("Could not get block for head root: {head_root}"))?;
66-
67-
let current_slot = block.message.slot;
68-
69-
// [IGNORE] attestation.data.slot is equal to or earlier than the current_slot (with a
70-
// MAXIMUM_GOSSIP_CLOCK_DISPARITY allowance)
71-
if attestation.data.slot > current_slot {
72-
return Ok(ValidationResult::Ignore(
73-
"Attestation is from a future slot".to_string(),
74-
));
75-
}
76-
7775
// [IGNORE] the epoch of attestation.data.slot is either the current or previous epoch (with a
7876
// MAXIMUM_GOSSIP_CLOCK_DISPARITY allowance)
7977
let attestation_epoch = compute_epoch_at_slot(attestation.data.slot);
80-
let current_epoch = state.get_current_epoch();
81-
let previous_epoch = state.get_previous_epoch();
78+
let current_epoch = compute_epoch_at_slot(current_slot);
79+
let previous_epoch = current_epoch.saturating_sub(1);
8280

8381
if attestation_epoch != current_epoch && attestation_epoch != previous_epoch {
8482
return Ok(ValidationResult::Ignore(

‎crates/networking/manager/src/p2p_sender.rs‎

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
11
use anyhow::anyhow;
2-
use libp2p::{PeerId, swarm::ConnectionId};
2+
use libp2p::{
3+
PeerId,
4+
gossipsub::{MessageAcceptance, MessageId},
5+
swarm::ConnectionId,
6+
};
37
use ream_p2p::network::beacon::channel::{GossipMessage, P2PMessage, P2PResponse};
48
use ream_req_resp::{
59
beacon::messages::BeaconResponseMessage, error::ReqRespError, handler::RespMessage,
@@ -18,6 +22,21 @@ impl P2PSender {
1822
}
1923
}
2024

25+
pub fn report_gossip_validation(
26+
&self,
27+
message_id: MessageId,
28+
propagation_source: PeerId,
29+
acceptance: MessageAcceptance,
30+
) {
31+
if let Err(err) = self.0.send(P2PMessage::ReportGossipValidation {
32+
message_id,
33+
propagation_source,
34+
acceptance,
35+
}) {
36+
warn!("Failed to send gossip validation report: {err}");
37+
}
38+
}
39+
2140
pub fn send_response(
2241
&self,
2342
peer_id: PeerId,

‎crates/networking/manager/src/service.rs‎

Lines changed: 9 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -191,13 +191,19 @@ impl NetworkManagerService {
191191
Some(event) = manager_receiver.recv() => {
192192
match event {
193193
// Handles Gossipsub messages from other peers.
194-
ReamNetworkEvent::GossipsubMessage { message } =>
195-
handle_gossipsub_message(
194+
ReamNetworkEvent::GossipsubMessage { propagation_source, message_id, message } => {
195+
let acceptance = handle_gossipsub_message(
196196
message,
197197
&beacon_chain,
198198
&cached_db,
199199
&p2p_sender,
200-
).await,
200+
).await;
201+
p2p_sender.report_gossip_validation(
202+
message_id,
203+
propagation_source,
204+
acceptance,
205+
);
206+
}
201207
// Handles Req/Resp messages from other peers.
202208
ReamNetworkEvent::RequestMessage { peer_id, stream_id, connection_id, message } =>
203209
handle_req_resp_message(peer_id, stream_id, connection_id, message, &p2p_sender, &ream_db, network_state.clone()).await,

‎crates/networking/p2p/src/network/beacon/channel.rs‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,11 @@
11
use std::sync::Arc;
22

33
use alloy_primitives::B256;
4-
use libp2p::{PeerId, swarm::ConnectionId};
4+
use libp2p::{
5+
PeerId,
6+
gossipsub::{MessageAcceptance, MessageId},
7+
swarm::ConnectionId,
8+
};
59
use ream_consensus_beacon::blob_sidecar::BlobIdentifier;
610
use ream_req_resp::{
711
beacon::messages::{BeaconResponseMessage, status::Status},
@@ -22,6 +26,11 @@ pub enum P2PMessage {
2226
Request(P2PRequest),
2327
Response(P2PResponse),
2428
Gossip(GossipMessage),
29+
ReportGossipValidation {
30+
message_id: MessageId,
31+
propagation_source: PeerId,
32+
acceptance: MessageAcceptance,
33+
},
2534
}
2635

2736
pub enum P2PRequest {

0 commit comments

Comments
 (0)