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.
@@ -83,6 +87,90 @@ const HTTP_DEFAULT_TIMEOUT_QUOTIENT: u32 = 4;
 
 pub struct Client {}
 
+/// Initialize operator doppelgänger protection if enabled
+///
+/// Returns `(DoppelgangerConfig, Option<watch::Receiver<bool>>)` where:
+/// - `DoppelgangerConfig` contains the checker callback and shutdown channel for message receiver
+/// - `watch::Receiver<bool>` broadcasts monitoring status (true = monitoring, false = active)
+fn initialize_operator_doppelganger<E: EthSpec>(
+    operator_dg: bool,
+    operator_dg_wait_epochs: u64,
+    operator_dg_fresh_k: u64,
+    operator_id: &OwnOperatorId,
+    slot_clock: &SystemTimeSlotClock,
+    slot_duration: Duration,
+    executor: &TaskExecutor,
+) -> Result<
+    (
+        Option<DoppelgangerConfig>,
+        Option<tokio::sync::watch::Receiver<bool>>,
+    ),
+    String,
+> {
+    if !operator_dg {
+        return Ok((None, None));
+    }
+
+    let own_operator_id = operator_id
+        .get()
+        .ok_or_else(|| "Operator ID not yet available".to_string())?;
+
+    let current_epoch = slot_clock
+        .now()
+        .ok_or_else(|| "Unable to read current slot".to_string())?
+        .epoch(E::slots_per_epoch());
+
+    let (service, is_monitoring_rx) = OperatorDoppelgangerService::<E, _>::new(
+        own_operator_id,
+        slot_clock.clone(),
+        current_epoch,
+        operator_dg_wait_epochs,
+        operator_dg_fresh_k,
+        slot_duration,
+    );
+    let doppelganger_service = Arc::new(service);
+
+    // Spawn background task to watch for monitoring period end
+    doppelganger_service.clone().spawn_monitor_task(executor);
+
+    // 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",
+    );
+
+    Ok((
+        Some(DoppelgangerConfig {
+            checker: Arc::new(checker),
+            shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))),
+        }),
+        Some(is_monitoring_rx),
+    ))
+}
+
 impl Client {
     /// Runs the Anchor Client
     pub async fn run<E: EthSpec>(executor: TaskExecutor, config: Config) -> Result<(), String> {
@@ -466,6 +554,17 @@ impl Client {
 
         let (outcome_tx, outcome_rx) = mpsc::channel::<message_receiver::Outcome>(9000);
 
+        // Initialize operator doppelgänger protection if enabled
+        let (doppelganger_config, is_monitoring) = initialize_operator_doppelganger::<E>(
+            config.operator_dg && config.impostor.is_none(),
+            config.operator_dg_wait_epochs,
+            config.operator_dg_fresh_k,
+            &operator_id,
+            &slot_clock,
+            Duration::from_secs(spec.seconds_per_slot),
+            &executor,
+        )?;
+
         let message_receiver = NetworkMessageReceiver::new(
             processor_senders.clone(),
             qbft_manager.clone(),
@@ -474,6 +573,7 @@ impl Client {
             is_synced.clone(),
             outcome_tx,
             message_validator,
+            doppelganger_config,
         );
 
         // Start the p2p network
@@ -573,6 +673,20 @@ impl Client {
             .map_err(|_| "Sync watch channel closed")?;
         info!("Sync complete, starting services...");
 
+        // Wait for operator doppelgänger monitoring to complete
+        if let Some(is_monitoring) = &is_monitoring {
+            info!(
+                wait_epochs = config.operator_dg_wait_epochs,
+                "Waiting for operator doppelgänger monitoring to complete before starting services..."
+            );
+            is_monitoring
+                .clone()
+                .wait_for(|&is_monitoring| !is_monitoring) // Wait until NOT monitoring
+                .await
+                .map_err(|_| "Monitoring watch channel closed")?;
+            info!("Operator doppelgänger monitoring complete, starting services...");
+        }
+
         let mut block_service_builder = BlockServiceBuilder::new()
             .slot_clock(slot_clock.clone())
             .validator_store(validator_store.clone())

anchor/client/src/operator_doppelganger/mod.rs

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