Merge branch 'unstable' into attestation-rewards-electra

Author: michaelsproul

Cargo.lock

@@ -250,7 +250,7 @@ install-audit:
 	cargo install --force cargo-audit
 
 audit-CI:
-	cargo audit --ignore RUSTSEC-2024-0421
+	cargo audit
 
 # Runs `cargo vendor` to make sure dependencies can be vendored for packaging, reproducibility and archival purpose.
 vendor:

Makefile

@@ -2,11 +2,8 @@ mod common;
 pub mod validator;
 pub mod wallet;
 
-use clap::Arg;
-use clap::ArgAction;
 use clap::ArgMatches;
 use clap::Command;
-use clap_utils::FLAG_HEADER;
 use environment::Environment;
 use types::EthSpec;
 
@@ -21,15 +18,6 @@ pub fn cli_app() -> Command {
         .visible_aliases(["a", "am", "account"])
         .about("Utilities for generating and managing Ethereum 2.0 accounts.")
         .display_order(0)
-        .arg(
-            Arg::new("help")
-                .long("help")
-                .short('h')
-                .help("Prints help information")
-                .action(ArgAction::HelpLong)
-                .display_order(0)
-                .help_heading(FLAG_HEADER),
-        )
         .subcommand(wallet::cli_app())
         .subcommand(validator::cli_app())
 }

account_manager/src/lib.rs

@@ -8,7 +8,6 @@ pub mod slashing_protection;
 
 use crate::{VALIDATOR_DIR_FLAG, VALIDATOR_DIR_FLAG_ALIAS};
 use clap::{Arg, ArgAction, ArgMatches, Command};
-use clap_utils::FLAG_HEADER;
 use directory::{parse_path_or_default_with_flag, DEFAULT_VALIDATOR_DIR};
 use environment::Environment;
 use std::path::PathBuf;
@@ -20,16 +19,6 @@ pub fn cli_app() -> Command {
     Command::new(CMD)
         .display_order(0)
         .about("Provides commands for managing Eth2 validators.")
-        .arg(
-            Arg::new("help")
-                .long("help")
-                .short('h')
-                .help("Prints help information")
-                .action(ArgAction::HelpLong)
-                .display_order(0)
-                .help_heading(FLAG_HEADER)
-                .global(true),
-        )
         .arg(
             Arg::new(VALIDATOR_DIR_FLAG)
                 .long(VALIDATOR_DIR_FLAG)

account_manager/src/validator/mod.rs

@@ -4,7 +4,6 @@ pub mod recover;
 
 use crate::WALLETS_DIR_FLAG;
 use clap::{Arg, ArgAction, ArgMatches, Command};
-use clap_utils::FLAG_HEADER;
 use directory::{parse_path_or_default_with_flag, DEFAULT_WALLET_DIR};
 use std::fs::create_dir_all;
 use std::path::PathBuf;
@@ -15,16 +14,6 @@ pub fn cli_app() -> Command {
     Command::new(CMD)
         .about("Manage wallets, from which validator keys can be derived.")
         .display_order(0)
-        .arg(
-            Arg::new("help")
-            .long("help")
-            .short('h')
-            .help("Prints help information")
-            .action(ArgAction::HelpLong)
-            .display_order(0)
-            .help_heading(FLAG_HEADER)
-            .global(true)
-        )
         .arg(
             Arg::new(WALLETS_DIR_FLAG)
                 .long(WALLETS_DIR_FLAG)

account_manager/src/wallet/mod.rs

@@ -9,6 +9,7 @@ use crate::fork_choice_signal::ForkChoiceSignalTx;
 use crate::fork_revert::{reset_fork_choice_to_finalization, revert_to_fork_boundary};
 use crate::graffiti_calculator::{GraffitiCalculator, GraffitiOrigin};
 use crate::head_tracker::HeadTracker;
+use crate::kzg_utils::blobs_to_data_column_sidecars;
 use crate::light_client_server_cache::LightClientServerCache;
 use crate::migrate::{BackgroundMigrator, MigratorConfig};
 use crate::observed_data_sidecars::ObservedDataSidecars;
@@ -562,9 +563,30 @@ where
             .put_block(&weak_subj_block_root, weak_subj_block.clone())
             .map_err(|e| format!("Failed to store weak subjectivity block: {e:?}"))?;
         if let Some(blobs) = weak_subj_blobs {
-            store
-                .put_blobs(&weak_subj_block_root, blobs)
-                .map_err(|e| format!("Failed to store weak subjectivity blobs: {e:?}"))?;
+            if self
+                .spec
+                .is_peer_das_enabled_for_epoch(weak_subj_block.epoch())
+            {
+                // After PeerDAS recompute columns from blobs to not force the checkpointz server
+                // into exposing another route.
+                let blobs = blobs
+                    .iter()
+                    .map(|blob_sidecar| &blob_sidecar.blob)
+                    .collect::<Vec<_>>();
+                let data_columns =
+                    blobs_to_data_column_sidecars(&blobs, &weak_subj_block, &self.kzg, &self.spec)
+                        .map_err(|e| {
+                            format!("Failed to compute weak subjectivity data_columns: {e:?}")
+                        })?;
+                // TODO(das): only persist the columns under custody
+                store
+                    .put_data_columns(&weak_subj_block_root, data_columns)
+                    .map_err(|e| format!("Failed to store weak subjectivity data_column: {e:?}"))?;
+            } else {
+                store
+                    .put_blobs(&weak_subj_block_root, blobs)
+                    .map_err(|e| format!("Failed to store weak subjectivity blobs: {e:?}"))?;
+            }
         }
 
         // Stage the database's metadata fields for atomic storage when `build` is called.

beacon_node/beacon_chain/src/builder.rs

@@ -27,8 +27,8 @@ mod overflow_lru_cache;
 mod state_lru_cache;
 
 use crate::data_column_verification::{
-    verify_kzg_for_data_column, verify_kzg_for_data_column_list, CustodyDataColumn,
-    GossipVerifiedDataColumn, KzgVerifiedCustodyDataColumn, KzgVerifiedDataColumn,
+    verify_kzg_for_data_column_list_with_scoring, CustodyDataColumn, GossipVerifiedDataColumn,
+    KzgVerifiedCustodyDataColumn, KzgVerifiedDataColumn,
 };
 use crate::metrics::{
     KZG_DATA_COLUMN_RECONSTRUCTION_ATTEMPTS, KZG_DATA_COLUMN_RECONSTRUCTION_FAILURES,
@@ -230,19 +230,14 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
         block_root: Hash256,
         custody_columns: DataColumnSidecarList<T::EthSpec>,
     ) -> Result<Availability<T::EthSpec>, AvailabilityCheckError> {
-        // TODO(das): report which column is invalid for proper peer scoring
-        // TODO(das): batch KZG verification here, but fallback into checking each column
-        // individually to report which column(s) are invalid.
-        let verified_custody_columns = custody_columns
+        // Attributes fault to the specific peer that sent an invalid column
+        let kzg_verified_columns = KzgVerifiedDataColumn::from_batch(custody_columns, &self.kzg)
+            .map_err(AvailabilityCheckError::InvalidColumn)?;
+
+        let verified_custody_columns = kzg_verified_columns
             .into_iter()
-            .map(|column| {
-                let index = column.index;
-                Ok(KzgVerifiedCustodyDataColumn::from_asserted_custody(
-                    KzgVerifiedDataColumn::new(column, &self.kzg)
-                        .map_err(|e| AvailabilityCheckError::InvalidColumn(index, e))?,
-                ))
-            })
-            .collect::<Result<Vec<_>, AvailabilityCheckError>>()?;
+            .map(KzgVerifiedCustodyDataColumn::from_asserted_custody)
+            .collect::<Vec<_>>();
 
         self.availability_cache.put_kzg_verified_data_columns(
             block_root,
@@ -365,7 +360,8 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
                         .iter()
                         .map(|custody_column| custody_column.as_data_column()),
                     &self.kzg,
-                )?;
+                )
+                .map_err(AvailabilityCheckError::InvalidColumn)?;
                 Ok(MaybeAvailableBlock::Available(AvailableBlock {
                     block_root,
                     block,
@@ -432,8 +428,9 @@ impl<T: BeaconChainTypes> DataAvailabilityChecker<T> {
 
         // verify kzg for all data columns at once
         if !all_data_columns.is_empty() {
-            // TODO: Need to also attribute which specific block is faulty
-            verify_kzg_for_data_column_list_with_scoring(all_data_columns.iter(), &self.kzg)?;
+            // Attributes fault to the specific peer that sent an invalid column
+            verify_kzg_for_data_column_list_with_scoring(all_data_columns.iter(), &self.kzg)
+                .map_err(AvailabilityCheckError::InvalidColumn)?;
         }
 
         for block in blocks {
@@ -716,32 +713,6 @@ async fn availability_cache_maintenance_service<T: BeaconChainTypes>(
     }
 }
 
-fn verify_kzg_for_data_column_list_with_scoring<'a, E: EthSpec, I>(
-    data_column_iter: I,
-    kzg: &'a Kzg,
-) -> Result<(), AvailabilityCheckError>
-where
-    I: Iterator<Item = &'a Arc<DataColumnSidecar<E>>> + Clone,
-{
-    let Err(batch_err) = verify_kzg_for_data_column_list(data_column_iter.clone(), kzg) else {
-        return Ok(());
-    };
-
-    let data_columns = data_column_iter.collect::<Vec<_>>();
-    // Find which column is invalid. If len is 1 or 0 continue to default case below.
-    // If len > 1 at least one column MUST fail.
-    if data_columns.len() > 1 {
-        for data_column in data_columns {
-            if let Err(e) = verify_kzg_for_data_column(data_column.clone(), kzg) {
-                return Err(AvailabilityCheckError::InvalidColumn(data_column.index, e));
-            }
-        }
-    }
-
-    // len 0 should never happen
-    Err(AvailabilityCheckError::InvalidColumn(0, batch_err))
-}
-
 /// A fully available block that is ready to be imported into fork choice.
 #[derive(Clone, Debug, PartialEq)]
 pub struct AvailableBlock<E: EthSpec> {

beacon_node/beacon_chain/src/data_availability_checker.rs

@@ -4,7 +4,7 @@ use types::{BeaconStateError, ColumnIndex, Hash256};
 #[derive(Debug)]
 pub enum Error {
     InvalidBlobs(KzgError),
-    InvalidColumn(ColumnIndex, KzgError),
+    InvalidColumn(Vec<(ColumnIndex, KzgError)>),
     ReconstructColumnsError(KzgError),
     KzgCommitmentMismatch {
         blob_commitment: KzgCommitment,

beacon_node/beacon_chain/src/data_availability_checker/error.rs

@@ -239,6 +239,18 @@ impl<E: EthSpec> KzgVerifiedDataColumn<E> {
     pub fn new(data_column: Arc<DataColumnSidecar<E>>, kzg: &Kzg) -> Result<Self, KzgError> {
         verify_kzg_for_data_column(data_column, kzg)
     }
+
+    pub fn from_batch(
+        data_columns: Vec<Arc<DataColumnSidecar<E>>>,
+        kzg: &Kzg,
+    ) -> Result<Vec<Self>, Vec<(ColumnIndex, KzgError)>> {
+        verify_kzg_for_data_column_list_with_scoring(data_columns.iter(), kzg)?;
+        Ok(data_columns
+            .into_iter()
+            .map(|column| Self { data: column })
+            .collect())
+    }
+
     pub fn to_data_column(self) -> Arc<DataColumnSidecar<E>> {
         self.data
     }
@@ -378,6 +390,38 @@ where
     Ok(())
 }
 
+/// Complete kzg verification for a list of `DataColumnSidecar`s.
+///
+/// If there's at least one invalid column, it re-verifies all columns individually to identify the
+/// first column that is invalid. This is necessary to attribute fault to the specific peer that
+/// sent bad data. The re-verification cost should not be significant. If a peer sends invalid data it
+/// will be quickly banned.
+pub fn verify_kzg_for_data_column_list_with_scoring<'a, E: EthSpec, I>(
+    data_column_iter: I,
+    kzg: &'a Kzg,
+) -> Result<(), Vec<(ColumnIndex, KzgError)>>
+where
+    I: Iterator<Item = &'a Arc<DataColumnSidecar<E>>> + Clone,
+{
+    if verify_kzg_for_data_column_list(data_column_iter.clone(), kzg).is_ok() {
+        return Ok(());
+    };
+
+    // Find all columns that are invalid and identify by index. If we hit this condition there
+    // should be at least one invalid column
+    let errors = data_column_iter
+        .filter_map(|data_column| {
+            if let Err(e) = verify_kzg_for_data_column(data_column.clone(), kzg) {
+                Some((data_column.index, e))
+            } else {
+                None
+            }
+        })
+        .collect::<Vec<_>>();
+
+    Err(errors)
+}
+
 pub fn validate_data_column_sidecar_for_gossip<T: BeaconChainTypes, O: ObservationStrategy>(
     data_column: Arc<DataColumnSidecar<T::EthSpec>>,
     subnet: u64,

beacon_node/beacon_chain/src/data_column_verification.rs

@@ -6,7 +6,9 @@ authors = ["Sean Anderson <[email protected]>"]
 
 [dependencies]
 eth2 = { workspace = true }
+ethereum_ssz = { workspace = true }
 lighthouse_version = { workspace = true }
 reqwest = { workspace = true }
 sensitive_url = { workspace = true }
 serde = { workspace = true }
+serde_json = { workspace = true }