feat: operator doppelgänger protection with slot-based detection (#692)

Closes #627


  Implements **Operator Doppelgänger Protection** to detect when multiple instances of the same operator are running, preventing QBFT protocol violations.

### Problem

If two operator instances run simultaneously:
- Both participate in QBFT consensus
- Both send messages with the same operator ID
- QBFT receives conflicting votes from one operator ID
- Protocol correctness breaks (QBFT expects one vote per operator per round)

### Solution

**Slot-Based Detection**: Records the startup slot and monitors the network for messages with our operator ID that reference slots after startup.

| Phase      | Outgoing Messages | Detection Logic                                          |
|------------|-------------------|----------------------------------------------------------|
| Monitoring | ❌ BLOCKED         | Messages with slot > startup_slot → Twin detected        |
| Completed  | ✅ SENT            | No longer checking                                       |

**Why Slot Comparison Works:**
- All outgoing messages blocked during monitoring
- Messages with `slot ≤ startup_slot`: Ignored (our own old messages)
- Messages with `slot > startup_slot`: Twin detected (we didn't send them)
- No race conditions possible since we never compete with twins

**State Management:**
- Simple AtomicBool for lock-free monitoring state
- No grace period needed (slot comparison handles message filtering)
- Detection active immediately on startup

**Twin Detection:**
- Monitors single-signer QBFT and partial signature messages
- Twin detected → logs detailed context → graceful shutdown
- Handles edge cases: restart in same slot, network delays, clock skew

### Benefits Over Time-Based Approaches

- **Faster Startup**: No 411-second grace period wait
- **More Reliable**: Slot comparison is deterministic, unaffected by network delays
- **Simpler**: Fewer states, cleaner logic with lock-free operations

### Configuration

This feature is **opt-in** to avoid imposing a monitoring delay on all operators:

- `--operator-dg`: Enable protection (default: false)
- `--operator-dg-wait-epochs`: Monitoring duration in epochs (default: 2, ~768 seconds)

**Rationale for Opt-In:**
- QBFT's Byzantine tolerance already handles some operator duplication scenarios
- No clear path from operator duplication to validator slashing (operators don't directly sign beacon chain messages)
- Provides operational discipline rather than security guarantees
- ~12 minute startup delay may not be acceptable for all deployment scenarios
- Operators can enable when operational safety is prioritized

### Testing

8 comprehensive tests covering slot-based detection, state management, and edge cases. Manual testing verified all configuration scenarios work correctly.


Co-Authored-By: diego <[email protected]>

Author: diegomrsantos

CLAUDE.md

@@ -225,6 +225,16 @@ When contributing to Anchor, follow these Rust best practices:
 6. **Simplicity First**: Always choose the simplest solution that elegantly solves the problem, follows existing patterns, maintains performance, and uses basic constructs over complex data structures
 7. **Check Requirements First**: Before implementing or creating anything (PRs, commits, code), always read and follow existing templates, guidelines, and requirements in the codebase
 
+### Architecture and Design
+
+1. **Question Intermediaries**: If data flows A → B with no transformation, question why A → intermediate → B exists. Each layer should provide clear value (logging, transformation, validation, etc.). Ask: "What problem does this solve that direct communication doesn't?"
+
+2. **Separation Through Interfaces, Not Layers**: Clean boundaries come from well-defined APIs, not intermediary components. A component receiving a `Sender<Event>` achieves separation without needing forwarding tasks or wrapper channels.
+
+3. **Simplification is Always Valid**: Refactoring working code for simplicity is encouraged. Question architectural decisions even after tests pass. Fewer lines and fewer components often indicates better design.
+
+4. **Challenge Complexity**: Every abstraction should justify its existence. "We might need it later" or "it provides separation" aren't sufficient reasons. Complexity must solve specific, current problems.
+
 ### Specific Guidelines
 
 1. **Naming**:
@@ -409,9 +419,34 @@ When writing PR descriptions, follow these guidelines for maintainable and revie
 
 - **Keep "Proposed Changes" section high-level** - focus on what components were changed and why
 - **Avoid line-by-line documentation** - reviewers can see specific changes in the diff
-- **Use component-level summaries** rather than file-by-file breakdowns  
+- **Use component-level summaries** rather than file-by-file breakdowns
 - **Emphasize the principles** being applied and operational impact
 - **Be concise but complete** - provide context without overwhelming detail
+- **Don't mention implementation details** - avoid specifying exact files, line numbers, or function names
+- **Don't state the obvious** - don't mention that tests pass (CI will verify this)
+- **Avoid redundancy** - don't repeat information already in the title or commit message
+- **Focus on the "why"** - explain the motivation and impact, not the mechanics
+
+### Code Review Culture
+
+Effective code reviews question "why" architectural decisions exist:
+
+**Questions to Ask:**
+- "Why does this intermediary layer exist?"
+- "What problem does this abstraction solve?"
+- "Could components communicate directly?"
+- "Is this complexity providing clear value?"
+
+**Encourage Simplification:**
+- Working code can still be improved
+- Refactoring for clarity is valuable
+- Fewer components usually means better architecture
+- Test passing ≠ design complete
+
+**Balance:**
+- Question complexity, but respect existing patterns that solve real problems
+- Not every layer is unnecessary - some provide genuine value
+- Focus on "why" over "what"
 
 ## Development Tips
 

Cargo.lock

@@ -23,6 +23,7 @@ members = [
     "anchor/message_sender",
     "anchor/message_validator",
     "anchor/network",
+    "anchor/operator_doppelganger",
     "anchor/processor",
     "anchor/qbft_manager",
     "anchor/signature_collector",
@@ -54,6 +55,7 @@ message_receiver = { path = "anchor/message_receiver" }
 message_sender = { path = "anchor/message_sender" }
 message_validator = { path = "anchor/message_validator" }
 network = { path = "anchor/network" }
+operator_doppelganger = { path = "anchor/operator_doppelganger" }
 operator_key = { path = "anchor/common/operator_key" }
 processor = { path = "anchor/processor" }
 qbft = { path = "anchor/common/qbft" }

Cargo.toml

@@ -32,6 +32,7 @@ multiaddr = { workspace = true }
 network = { workspace = true }
 network_utils = { workspace = true }
 openssl = { workspace = true }
+operator_doppelganger = { workspace = true }
 operator_key = { workspace = true }
 parking_lot = { workspace = true }
 processor = { workspace = true }

anchor/client/Cargo.toml

@@ -491,6 +491,32 @@ 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 blocks all \
+                outgoing messages and monitors the network for messages signed with its operator ID \
+                that reference slots after startup. Shuts down if a twin operator is detected \
+                to prevent QBFT protocol violations.",
+        display_order = 0,
+        default_value_t = false,
+        help_heading = FLAG_HEADER,
+        action = ArgAction::Set
+    )]
+    pub operator_dg: bool,
+
+    #[clap(
+        long,
+        value_name = "EPOCHS",
+        help = "Number of epochs to monitor for twin operators using slot-based detection. \
+                During monitoring, outgoing messages remain blocked and the node checks incoming \
+                messages for slots after startup to detect duplicate operator instances.",
+        display_order = 0,
+        default_value_t = 2,
+        requires = "operator_dg"
+    )]
+    pub operator_dg_wait_epochs: u64,
+
     #[clap(flatten)]
     pub logging_flags: FileLoggingFlags,
 }

anchor/client/src/cli.rs

@@ -72,6 +72,10 @@ 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 (blocks messages and monitors for twins)
+    pub operator_dg: bool,
+    /// Number of epochs to monitor for twins after grace period
+    pub operator_dg_wait_epochs: u64,
 }
 
 impl Config {
@@ -115,6 +119,8 @@ impl Config {
             prefer_builder_proposals: false,
             gas_limit: 36_000_000,
             disable_latency_measurement_service: false,
+            operator_dg: false,
+            operator_dg_wait_epochs: 2,
         }
     }
 }
@@ -246,6 +252,10 @@ 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;
+
     // Performance options
     if let Some(max_workers) = cli_args.max_workers {
         config.processor.max_workers = max_workers;

anchor/client/src/config.rs

@@ -36,6 +36,7 @@ use message_sender::{MessageSender, NetworkMessageSender, impostor::ImpostorMess
 use message_validator::Validator;
 use network::Network;
 use openssl::rsa::Rsa;
+use operator_doppelganger::OperatorDoppelgangerService;
 use parking_lot::RwLock;
 use qbft_manager::QbftManager;
 use sensitive_url::SensitiveUrl;
@@ -394,6 +395,10 @@ impl Client {
             "syncer",
         );
 
+        // Create operator ID wrapper that watches the database for our operator ID.
+        // Follows the common pattern: pass OwnOperatorId to components, they call .get() only when
+        // needed. This allows initialization before sync completes (which populates the ID from
+        // chain).
         let operator_id = OwnOperatorId::new(database.watch());
 
         // Network sender/receiver
@@ -419,15 +424,34 @@ impl Client {
             &executor,
         );
 
+        // Create operator doppelgänger protection if enabled (will be started after sync)
+        let doppelganger_service = if config.operator_dg && config.impostor.is_none() {
+            // Get current slot for slot-based detection baseline
+            let startup_slot = slot_clock.now().ok_or_else(|| {
+                "Failed to get current slot for doppelgänger protection".to_string()
+            })?;
+
+            Some(Arc::new(OperatorDoppelgangerService::new(
+                operator_id.clone(),
+                startup_slot,
+                E::slots_per_epoch(),
+                Duration::from_secs(spec.seconds_per_slot),
+            )))
+        } else {
+            None
+        };
+
         let message_sender: Arc<dyn MessageSender> = if config.impostor.is_none() {
             Arc::new(NetworkMessageSender::new(
-                processor_senders.clone(),
-                network_tx.clone(),
-                key.clone(),
-                operator_id.clone(),
-                Some(message_validator.clone()),
-                SUBNET_COUNT,
-                is_synced.clone(),
+                message_sender::NetworkMessageSenderConfig {
+                    processor: processor_senders.clone(),
+                    network_tx: network_tx.clone(),
+                    private_key: key.clone(),
+                    operator_id: operator_id.clone(),
+                    validator: Some(message_validator.clone()),
+                    subnet_count: SUBNET_COUNT,
+                    is_synced: is_synced.clone(),
+                },
             )?)
         } else {
             Arc::new(ImpostorMessageSender::new(network_tx.clone(), SUBNET_COUNT))
@@ -474,6 +498,7 @@ impl Client {
             is_synced.clone(),
             outcome_tx,
             message_validator,
+            doppelganger_service.clone(),
         );
 
         // Start the p2p network
@@ -560,6 +585,7 @@ impl Client {
             duties_service.clone(),
             database.watch(),
             is_synced.clone(),
+            doppelganger_service.clone(),
             executor.clone(),
             &spec,
         );
@@ -573,6 +599,15 @@ impl Client {
             .map_err(|_| "Sync watch channel closed")?;
         info!("Sync complete, starting services...");
 
+        // Block client startup during operator doppelgänger monitoring period (now that sync
+        // is complete and operator ID available). During this period, incoming messages are
+        // checked for twins and duties services won't start until monitoring completes.
+        if let Some(service) = &doppelganger_service {
+            service
+                .monitor_blocking(config.operator_dg_wait_epochs)
+                .await?;
+        }
+
         let mut block_service_builder = BlockServiceBuilder::new()
             .slot_clock(slot_clock.clone())
             .validator_store(validator_store.clone())