Duty logic validation for consensus messages

.githooks/pre-commit

@@ -1,9 +1,9 @@
-#!/bin/sh
-echo "Running cargo fmt --all..."
-cargo +nightly fmt --all || exit 1
-
-echo "Running cargo clippy --all..."
-cargo clippy --all || exit 1
-
-echo "Running cargo sort workspace..."
-cargo sort --workspace || exit 1
+##!/bin/sh
+#echo "Running cargo fmt --all..."
+#cargo +nightly fmt --all || exit 1
+#
+#echo "Running cargo clippy --all..."
+#cargo clippy --all || exit 1
+#
+#echo "Running cargo sort workspace..."
+#cargo sort --workspace || exit 1

anchor/client/src/lib.rs

@@ -28,7 +28,7 @@ use eth2::{
 use keygen::{encryption::decrypt, run_keygen, Keygen};
 use message_receiver::NetworkMessageReceiver;
 use message_sender::NetworkMessageSender;
-use message_validator::Validator;
+use message_validator::{BeaconNetwork, DutiesTracker, Validator};
 use network::Network;
 use openssl::{pkey::Private, rsa::Rsa};
 use parking_lot::RwLock;
@@ -372,10 +372,24 @@ impl Client {
         // Network sender/receiver
         let (network_tx, network_rx) = mpsc::channel::<(SubnetId, Vec<u8>)>(9001);
 
+        let duties_tracker = Arc::new(DutiesTracker::new(
+            beacon_nodes.clone(),
+            spec.clone(),
+            E::slots_per_epoch(),
+            slot_clock.clone(),
+            executor.clone(),
+            database.watch(),
+        ));
+
         let message_validator = Arc::new(Validator::new(
             database.watch(),
             E::slots_per_epoch(),
-            slot_clock.clone(),
+            BeaconNetwork::new(
+                slot_clock.clone(),
+                E::slots_per_epoch(),
+                spec.epochs_per_sync_committee_period.as_u64(),
+            ),
+            duties_tracker.clone(),
         ));
 
         let network_message_sender = NetworkMessageSender::new(

anchor/common/ssv_types/src/cluster.rs

@@ -70,6 +70,12 @@ pub struct ClusterMember {
 #[ssz(struct_behaviour = "transparent")]
 pub struct ValidatorIndex(pub usize);
 
+impl From<ValidatorIndex> for u64 {
+    fn from(value: ValidatorIndex) -> Self {
+        value.0 as u64
+    }
+}
+
 /// General Metadata about a Validator
 #[derive(Debug, Clone)]
 pub struct ValidatorMetadata {

anchor/database/src/state.rs

@@ -365,4 +365,12 @@ impl NetworkState {
             validator_indices: validator_index.map(|idx| vec![idx]).unwrap_or(vec![]),
         })
     }
+
+    pub fn validator_indices(&self) -> Vec<u64> {
+        self.multi_state
+            .validator_metadata
+            .values()
+            .filter_map(|metadata| metadata.index.map(|idx| idx.0 as u64))
+            .collect()
+    }
 }

anchor/message_receiver/src/manager.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use database::{NetworkState, UniqueIndex};
 use gossipsub::{Message, MessageAcceptance, MessageId};
 use libp2p::PeerId;
-use message_validator::{ValidatedMessage, ValidatedSSVMessage, Validator};
+use message_validator::{DutiesProvider, ValidatedMessage, ValidatedSSVMessage, Validator};
 use qbft_manager::QbftManager;
 use signature_collector::SignatureCollectorManager;
 use slot_clock::SlotClock;
@@ -22,23 +22,23 @@ pub struct Outcome {
 }
 
 /// A message receiver that passes messages to responsible managers.
-pub struct NetworkMessageReceiver<S: SlotClock> {
+pub struct NetworkMessageReceiver<S: SlotClock, D: DutiesProvider> {
     processor: processor::Senders,
     qbft_manager: Arc<QbftManager>,
     signature_collector: Arc<SignatureCollectorManager>,
     network_state_rx: watch::Receiver<NetworkState>,
     outcome_tx: mpsc::Sender<Outcome>,
-    validator: Arc<Validator<S>>,
+    validator: Arc<Validator<S, D>>,
 }
 
-impl<S: SlotClock + 'static> NetworkMessageReceiver<S> {
+impl<S: SlotClock + 'static, D: DutiesProvider> NetworkMessageReceiver<S, D> {
     pub fn new(
         processor: processor::Senders,
         qbft_manager: Arc<QbftManager>,
         signature_collector: Arc<SignatureCollectorManager>,
         network_state_rx: watch::Receiver<NetworkState>,
         outcome_tx: mpsc::Sender<Outcome>,
-        validator: Arc<Validator<S>>,
+        validator: Arc<Validator<S, D>>,
     ) -> Arc<Self> {
         Arc::new(Self {
             processor,
@@ -51,7 +51,9 @@ impl<S: SlotClock + 'static> NetworkMessageReceiver<S> {
     }
 }
 
-impl<S: SlotClock + 'static> MessageReceiver for Arc<NetworkMessageReceiver<S>> {
+impl<S: SlotClock + 'static, D: DutiesProvider> MessageReceiver
+    for Arc<NetworkMessageReceiver<S, D>>
+{
     fn receive(
         &self,
         propagation_source: PeerId,

anchor/message_sender/src/network.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use message_validator::Validator;
+use message_validator::{DutiesProvider, Validator};
 use openssl::{
     error::ErrorStack,
     hash::MessageDigest,
@@ -22,16 +22,16 @@ use crate::{Error, MessageCallback, MessageSender};
 const SIGNER_NAME: &str = "message_sign_and_send";
 const SENDER_NAME: &str = "message_send";
 
-pub struct NetworkMessageSender<S: SlotClock> {
+pub struct NetworkMessageSender<S: SlotClock, D: DutiesProvider> {
     processor: processor::Senders,
     network_tx: mpsc::Sender<(SubnetId, Vec<u8>)>,
     private_key: PKey<Private>,
     operator_id: OperatorId,
-    validator: Option<Arc<Validator<S>>>,
+    validator: Option<Arc<Validator<S, D>>>,
     subnet_count: usize,
 }
 
-impl<S: SlotClock + 'static> MessageSender for Arc<NetworkMessageSender<S>> {
+impl<S: SlotClock + 'static, D: DutiesProvider> MessageSender for Arc<NetworkMessageSender<S, D>> {
     fn sign_and_send(
         &self,
         message: UnsignedSSVMessage,
@@ -94,13 +94,13 @@ impl<S: SlotClock + 'static> MessageSender for Arc<NetworkMessageSender<S>> {
     }
 }
 
-impl<S: SlotClock> NetworkMessageSender<S> {
+impl<S: SlotClock, D: DutiesProvider> NetworkMessageSender<S, D> {
     pub fn new(
         processor: processor::Senders,
         network_tx: mpsc::Sender<(SubnetId, Vec<u8>)>,
         private_key: Rsa<Private>,
         operator_id: OperatorId,
-        validator: Option<Arc<Validator<S>>>,
+        validator: Option<Arc<Validator<S, D>>>,
         subnet_count: usize,
     ) -> Result<Arc<Self>, String> {
         let private_key = PKey::from_rsa(private_key)

anchor/message_validator/Cargo.toml

@@ -5,20 +5,26 @@ edition = { workspace = true }
 authors = ["Sigma Prime <[email protected]>"]
 
 [dependencies]
+beacon_node_fallback = { workspace = true }
+bls = { workspace = true }
 dashmap = { workspace = true }
 database = { workspace = true }
+eth2 = { workspace = true }
 ethereum_ssz = { workspace = true }
 gossipsub = { workspace = true }
 hex = { workspace = true }
 openssl = { workspace = true }
 parking_lot = { workspace = true }
 processor = { workspace = true }
+safe_arith = { workspace = true }
 sha2 = { workspace = true }
 slot_clock = { workspace = true }
 ssv_types = { workspace = true }
+task_executor = { workspace = true }
 thiserror =  { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+types = { workspace = true }
 
 [dev-dependencies]
 bls = { workspace = true }

anchor/message_validator/src/beacon_network.rs

@@ -0,0 +1,121 @@
+use std::time::{Duration, SystemTime};
+
+use slot_clock::SlotClock;
+use types::{Epoch, Slot};
+
+/// Wrapper around SlotClock to provide beacon chain network functionality
+#[derive(Clone)]
+pub struct BeaconNetwork<S: SlotClock> {
+    slot_clock: S,
+    slots_per_epoch: u64,
+    epochs_per_sync_committee_period: u64,
+}
+
+impl<S: SlotClock> BeaconNetwork<S> {
+    /// Create a new BeaconNetwork
+    pub fn new(slot_clock: S, slots_per_epoch: u64, epochs_per_sync_committee_period: u64) -> Self {
+        Self {
+            slot_clock,
+            slots_per_epoch,
+            epochs_per_sync_committee_period,
+        }
+    }
+
+    /// Returns the slot clock
+    pub fn slot_clock(&self) -> &S {
+        &self.slot_clock
+    }
+
+    /// Returns the slot duration
+    pub fn slot_duration(&self) -> Duration {
+        self.slot_clock.slot_duration()
+    }
+
+    /// Returns the number of slots per epoch
+    pub fn slots_per_epoch(&self) -> u64 {
+        self.slots_per_epoch
+    }
+
+    /// Estimates the current slot
+    pub fn estimated_current_slot(&self) -> Slot {
+        self.slot_clock.now().unwrap_or_default()
+    }
+
+    /// Estimates the slot at the given time
+    pub fn estimated_slot_at_time(&self, time: SystemTime) -> Slot {
+        let since_unix = time
+            .duration_since(SystemTime::UNIX_EPOCH)
+            .unwrap_or_default();
+
+        self.slot_clock.slot_of(since_unix).unwrap_or_default()
+    }
+
+    /// Estimates the time at the given slot
+    pub fn estimated_time_at_slot(&self, slot: Slot) -> SystemTime {
+        let duration = self.slot_clock.start_of(slot).unwrap_or_default();
+        SystemTime::UNIX_EPOCH + duration
+    }
+
+    /// Estimates the current epoch
+    pub fn estimated_current_epoch(&self) -> Epoch {
+        self.estimated_epoch_at_slot(self.estimated_current_slot())
+    }
+
+    /// Estimates the epoch at the given slot
+    pub fn estimated_epoch_at_slot(&self, slot: Slot) -> Epoch {
+        Epoch::new(slot.as_u64() / self.slots_per_epoch)
+    }
+
+    /// Returns the first slot at the given epoch
+    pub fn first_slot_at_epoch(&self, epoch: u64) -> Slot {
+        Slot::new(epoch * self.slots_per_epoch)
+    }
+
+    /// Returns the start time of the given epoch
+    pub fn epoch_start_time(&self, epoch: u64) -> SystemTime {
+        self.estimated_time_at_slot(self.first_slot_at_epoch(epoch))
+    }
+
+    /// Returns the start time of the given slot
+    pub fn get_slot_start_time(&self, slot: Slot) -> SystemTime {
+        self.estimated_time_at_slot(slot)
+    }
+
+    /// Returns the end time of the given slot
+    pub fn get_slot_end_time(&self, slot: Slot) -> SystemTime {
+        self.estimated_time_at_slot(slot + 1)
+    }
+
+    /// Checks if the given slot is the first slot of its epoch
+    pub fn is_first_slot_of_epoch(&self, slot: Slot) -> bool {
+        slot.as_u64() % self.slots_per_epoch == 0
+    }
+
+    /// Returns the first slot of the given epoch
+    pub fn get_epoch_first_slot(&self, epoch: u64) -> Slot {
+        self.first_slot_at_epoch(epoch)
+    }
+
+    /// Returns the number of epochs per sync committee period
+    pub fn epochs_per_sync_committee_period(&self) -> u64 {
+        self.epochs_per_sync_committee_period
+    }
+
+    /// Estimates the sync committee period at the given epoch
+    pub fn estimated_sync_committee_period_at_epoch(&self, epoch: Epoch) -> u64 {
+        epoch.as_u64() / self.epochs_per_sync_committee_period
+    }
+
+    /// Returns the first epoch of the given sync committee period
+    pub fn first_epoch_of_sync_period(&self, period: u64) -> u64 {
+        period * self.epochs_per_sync_committee_period
+    }
+
+    /// Returns the last slot of the given sync committee period
+    pub fn last_slot_of_sync_period(&self, period: u64) -> Slot {
+        let last_epoch = self.first_epoch_of_sync_period(period + 1) - 1;
+        // If we are in the sync committee that ends at slot x we do not generate a message
+        // during slot x-1 as it will never be included, hence -2.
+        self.get_epoch_first_slot(last_epoch + 1) - 2
+    }
+}