feat: operator doppelgänger protection with slot-based detection

anchor/client/src/cli.rs

@@ -484,6 +484,43 @@ pub struct Node {
     #[clap(long, help = "Disables gossipsub topic scoring.", hide = true)]
     pub disable_gossipsub_topic_scoring: bool,
 
+    // Operator Doppelgänger Protection
+    #[clap(
+        long,
+        help = "Enable operator doppelgänger protection. When enabled, the node will monitor \
+                for messages signed by its operator ID on startup and shut down if a twin \
+                (duplicate operator) is detected. Enabled by default.",
+        display_order = 0,
+        default_value_t = true,
+        help_heading = FLAG_HEADER,
+        action = ArgAction::Set
+    )]
+    pub operator_dg: bool,
+
+    #[clap(
+        long,
+        value_name = "EPOCHS",
+        help = "Number of epochs to wait in monitor mode before starting normal operation. \
+                During this period, the node listens for messages from its own operator ID \
+                to detect if another instance is running.",
+        display_order = 0,
+        default_value_t = 2,
+        requires = "operator_dg"
+    )]
+    pub operator_dg_wait_epochs: u64,
+
+    #[clap(
+        long,
+        value_name = "HEIGHTS",
+        help = "The freshness threshold for detecting operator twins. Only messages within \
+                this many consensus heights from the maximum observed height are considered \
+                fresh evidence of a twin. This prevents false positives from replayed old messages.",
+        display_order = 0,
+        default_value_t = 3,
+        requires = "operator_dg"
+    )]
+    pub operator_dg_fresh_k: u64,
+
     #[clap(flatten)]
     pub logging_flags: FileLoggingFlags,
 }

anchor/client/src/config.rs

@@ -72,6 +72,12 @@ pub struct Config {
     pub prefer_builder_proposals: bool,
     /// Controls whether the latency measurement service is enabled
     pub disable_latency_measurement_service: bool,
+    /// Enable operator doppelgänger protection
+    pub operator_dg: bool,
+    /// Number of epochs to wait in monitor mode
+    pub operator_dg_wait_epochs: u64,
+    /// Freshness threshold (K) for detecting operator twins
+    pub operator_dg_fresh_k: u64,
 }
 
 impl Config {
@@ -115,6 +121,9 @@ impl Config {
             prefer_builder_proposals: false,
             gas_limit: 36_000_000,
             disable_latency_measurement_service: false,
+            operator_dg: true,
+            operator_dg_wait_epochs: 2,
+            operator_dg_fresh_k: 3,
         }
     }
 }
@@ -243,6 +252,11 @@ pub fn from_cli(cli_args: &Node, global_config: GlobalConfig) -> Result<Config,
     config.impostor = cli_args.impostor.map(OperatorId);
     config.disable_latency_measurement_service = cli_args.disable_latency_measurement_service;
 
+    // Operator doppelgänger protection
+    config.operator_dg = cli_args.operator_dg;
+    config.operator_dg_wait_epochs = cli_args.operator_dg_wait_epochs;
+    config.operator_dg_fresh_k = cli_args.operator_dg_fresh_k;
+
     // Performance options
     if let Some(max_workers) = cli_args.max_workers {
         config.processor.max_workers = max_workers;

anchor/client/src/lib.rs

@@ -3,13 +3,14 @@ pub mod config;
 mod key;
 mod metrics;
 mod notifier;
+mod operator_doppelganger;
 
 use std::{
     fs::File,
     io::Read,
     net::SocketAddr,
     path::Path,
-    sync::Arc,
+    sync::{Arc, Mutex},
     time::{Duration, SystemTime, UNIX_EPOCH},
 };
 
@@ -31,7 +32,7 @@ use eth2::{
     BeaconNodeHttpClient, Timeouts,
     reqwest::{Certificate, ClientBuilder},
 };
-use message_receiver::NetworkMessageReceiver;
+use message_receiver::{DoppelgangerChecker, DoppelgangerConfig, NetworkMessageReceiver};
 use message_sender::{MessageSender, NetworkMessageSender, impostor::ImpostorMessageSender};
 use message_validator::Validator;
 use network::Network;
@@ -47,7 +48,7 @@ use task_executor::TaskExecutor;
 use tokio::{
     net::TcpListener,
     select,
-    sync::{mpsc, mpsc::unbounded_channel},
+    sync::{mpsc, mpsc::unbounded_channel, oneshot},
     time::{Instant, interval, sleep},
 };
 use tracing::{debug, error, info, warn};
@@ -63,7 +64,10 @@ use validator_services::{
     sync_committee_service::SyncCommitteeService,
 };
 
-use crate::{key::read_or_generate_private_key, notifier::spawn_notifier};
+use crate::{
+    key::read_or_generate_private_key, notifier::spawn_notifier,
+    operator_doppelganger::OperatorDoppelgangerService,
+};
 
 /// Specific timeout constants for HTTP requests involved in different validator duties.
 /// This can help ensure that proper endpoint fallback occurs.
@@ -466,6 +470,57 @@ impl Client {
 
         let (outcome_tx, outcome_rx) = mpsc::channel::<message_receiver::Outcome>(9000);
 
+        // Initialize operator doppelgänger protection if enabled
+        let doppelganger_config = if config.operator_dg && config.impostor.is_none() {
+            let own_operator_id = operator_id
+                .get()
+                .ok_or_else(|| "Operator ID not yet available".to_string())?;
+
+            let doppelganger_service = Arc::new(OperatorDoppelgangerService::<E, _>::new(
+                own_operator_id,
+                slot_clock.clone(),
+                config.operator_dg_wait_epochs,
+                config.operator_dg_fresh_k,
+                true, // enabled
+            ));
+
+            // Create shutdown channel
+            let (shutdown_tx, shutdown_rx) = oneshot::channel();
+
+            // Create checker callback
+            let service = doppelganger_service.clone();
+            let checker: DoppelgangerChecker =
+                Box::new(move |signed_msg, qbft_msg| service.check_message(signed_msg, qbft_msg));
+
+            // Spawn task to listen for shutdown signal
+            let executor_clone = executor.clone();
+            executor.spawn_without_exit(
+                async move {
+                    if shutdown_rx.await.is_ok() {
+                        error!(
+                            "Operator doppelgänger detected! Initiating fatal shutdown to prevent equivocation."
+                        );
+                        // Give time for the error log to be flushed
+                        tokio::time::sleep(Duration::from_millis(100)).await;
+                        // Trigger executor shutdown with failure reason
+                        let _ = executor_clone
+                            .shutdown_sender()
+                            .try_send(task_executor::ShutdownReason::Failure(
+                                "Operator doppelgänger detected",
+                            ));
+                    }
+                },
+                "doppelganger-shutdown",
+            );
+
+            Some(DoppelgangerConfig {
+                checker: Arc::new(checker),
+                shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))),
+            })
+        } else {
+            None
+        };
+
         let message_receiver = NetworkMessageReceiver::new(
             processor_senders.clone(),
             qbft_manager.clone(),
@@ -474,6 +529,7 @@ impl Client {
             is_synced.clone(),
             outcome_tx,
             message_validator,
+            doppelganger_config,
         );
 
         // Start the p2p network

anchor/client/src/operator_doppelganger/mod.rs

@@ -0,0 +1,4 @@
+mod service;
+mod state;
+
+pub use service::OperatorDoppelgangerService;

anchor/client/src/operator_doppelganger/service.rs

@@ -0,0 +1,156 @@
+use std::{marker::PhantomData, sync::Arc};
+
+use parking_lot::RwLock;
+use slot_clock::SlotClock;
+use ssv_types::{
+    OperatorId, consensus::QbftMessage, message::SignedSSVMessage, msgid::DutyExecutor,
+};
+use tracing::{error, info, warn};
+use types::EthSpec;
+
+use super::state::{DoppelgangerMode, DoppelgangerState};
+
+/// Service for detecting operator doppelgängers (duplicate instances)
+pub struct OperatorDoppelgangerService<E: EthSpec, S: SlotClock> {
+    /// Our operator ID to watch for
+    own_operator_id: OperatorId,
+    /// Current state
+    state: Arc<RwLock<DoppelgangerState>>,
+    /// Slot clock for epoch tracking
+    slot_clock: S,
+    /// Enabled flag
+    enabled: bool,
+    /// Phantom data for EthSpec
+    _phantom: PhantomData<E>,
+}
+
+impl<E: EthSpec, S: SlotClock> OperatorDoppelgangerService<E, S> {
+    /// Create a new operator doppelgänger service
+    pub fn new(
+        own_operator_id: OperatorId,
+        slot_clock: S,
+        wait_epochs: u64,
+        fresh_k: u64,
+        enabled: bool,
+    ) -> Self {
+        let current_epoch = slot_clock
+            .now()
+            .map(|slot| slot.epoch(E::slots_per_epoch()))
+            .unwrap_or_else(|| types::Epoch::new(0));
+
+        let state = Arc::new(RwLock::new(DoppelgangerState::new(
+            current_epoch,
+            wait_epochs,
+            fresh_k,
+        )));
+
+        if enabled {
+            info!(
+                operator_id = *own_operator_id,
+                current_epoch = current_epoch.as_u64(),
+                wait_epochs,
+                fresh_k,
+                "Operator doppelgänger protection enabled, entering monitor mode"
+            );
+        } else {
+            info!("Operator doppelgänger protection disabled");
+        }
+
+        Self {
+            own_operator_id,
+            state,
+            slot_clock,
+            enabled,
+            _phantom: PhantomData,
+        }
+    }
+
+    /// Check if a message indicates a potential doppelgänger
+    ///
+    /// Returns true if a twin is detected (should trigger shutdown)
+    pub fn check_message(
+        &self,
+        signed_message: &SignedSSVMessage,
+        qbft_message: &QbftMessage,
+    ) -> bool {
+        if !self.enabled {
+            return false;
+        }
+
+        // Update mode based on current epoch
+        if let Some(slot) = self.slot_clock.now() {
+            let current_epoch = slot.epoch(E::slots_per_epoch());
+            self.state.write().update_mode(current_epoch);
+        }
+
+        let state = self.state.read();
+
+        // Only check in monitor mode
+        if !state.is_monitoring() {
+            return false;
+        }
+
+        // Extract committee ID from message
+        let committee_id = match signed_message.ssv_message().msg_id().duty_executor() {
+            Some(DutyExecutor::Committee(committee_id)) => committee_id,
+            _ => return false, // Not a committee message
+        };
+
+        // Update the maximum height we've seen for this committee
+        drop(state);
+        self.state
+            .write()
+            .update_max_height(committee_id, qbft_message.height);
+        let state = self.state.read();
+
+        // Check if this is a single-signer message with our operator ID
+        let operator_ids = signed_message.operator_ids();
+        if operator_ids.len() != 1 {
+            // Not a single-signer message (could be aggregate/decided)
+            return false;
+        }
+
+        let signer = operator_ids[0];
+        if signer != self.own_operator_id {
+            // Not signed by us
+            return false;
+        }
+
+        // Check if the message is fresh
+        if !state.is_fresh(committee_id, qbft_message.height) {
+            // Stale message, likely a replay - not evidence of a twin
+            warn!(
+                operator_id = *self.own_operator_id,
+                committee = ?committee_id,
+                height = qbft_message.height,
+                "Received stale message with our operator ID (likely replay), ignoring"
+            );
+            return false;
+        }
+
+        // Fresh single-signer message with our operator ID = twin detected!
+        error!(
+            operator_id = *self.own_operator_id,
+            committee = ?committee_id,
+            height = qbft_message.height,
+            round = qbft_message.round,
+            message_type = ?qbft_message.qbft_message_type,
+            "OPERATOR DOPPELGÄNGER DETECTED: Received fresh message signed with our operator ID. \
+             Another instance of this operator is running. Shutting down to prevent equivocation."
+        );
+
+        true
+    }
+
+    /// Get the current mode
+    #[allow(dead_code)]
+    pub fn mode(&self) -> DoppelgangerMode {
+        self.state.read().mode()
+    }
+
+    /// Check if we're still in monitor mode
+    #[allow(dead_code)]
+    pub fn is_monitoring(&self) -> bool {
+        self.enabled && self.state.read().is_monitoring()
+    }
+}