feat: ciphernode fault attribution and slashing [skip-line-limit]

crates/aggregator/src/committee_finalizer.rs

@@ -6,8 +6,8 @@
 
 use actix::prelude::*;
 use e3_events::{
-    prelude::*, trap, BusHandle, CommitteeFinalizeRequested, CommitteeRequested, EType,
-    EnclaveEvent, EnclaveEventData, EventType, Shutdown, TypedEvent,
+    prelude::*, trap, BusHandle, CommitteeFinalizeRequested, CommitteeRequested, E3Failed, E3Stage,
+    E3StageChanged, EType, EnclaveEvent, EnclaveEventData, EventType, Shutdown, TypedEvent,
 };
 use e3_utils::{NotifySync, MAILBOX_LIMIT};
 use std::collections::HashMap;
@@ -33,7 +33,12 @@ impl CommitteeFinalizer {
         let addr = CommitteeFinalizer::new(bus).start();
 
         bus.subscribe_all(
-            &[EventType::CommitteeRequested, EventType::Shutdown],
+            &[
+                EventType::CommitteeRequested,
+                EventType::Shutdown,
+                EventType::E3Failed,
+                EventType::E3StageChanged,
+            ],
             addr.clone().recipient(),
         );
 
@@ -57,6 +62,10 @@ impl Handler<EnclaveEvent> for CommitteeFinalizer {
                 self.notify_sync(ctx, TypedEvent::new(data, ec))
             }
             EnclaveEventData::Shutdown(data) => self.notify_sync(ctx, data),
+            EnclaveEventData::E3Failed(data) => self.notify_sync(ctx, TypedEvent::new(data, ec)),
+            EnclaveEventData::E3StageChanged(data) => {
+                self.notify_sync(ctx, TypedEvent::new(data, ec))
+            }
             _ => (),
         }
     }
@@ -166,3 +175,38 @@ impl Handler<Shutdown> for CommitteeFinalizer {
         ctx.stop();
     }
 }
+
+impl Handler<TypedEvent<E3Failed>> for CommitteeFinalizer {
+    type Result = ();
+    fn handle(&mut self, msg: TypedEvent<E3Failed>, ctx: &mut Self::Context) -> Self::Result {
+        let e3_id_str = msg.e3_id.to_string();
+        if let Some(handle) = self.pending_committees.remove(&e3_id_str) {
+            info!(
+                e3_id = %msg.e3_id,
+                reason = ?msg.reason,
+                "E3 failed — cancelling pending committee finalization timer"
+            );
+            ctx.cancel_future(handle);
+        }
+    }
+}
+
+impl Handler<TypedEvent<E3StageChanged>> for CommitteeFinalizer {
+    type Result = ();
+    fn handle(&mut self, msg: TypedEvent<E3StageChanged>, ctx: &mut Self::Context) -> Self::Result {
+        match &msg.new_stage {
+            E3Stage::Complete | E3Stage::Failed => {
+                let e3_id_str = msg.e3_id.to_string();
+                if let Some(handle) = self.pending_committees.remove(&e3_id_str) {
+                    info!(
+                        e3_id = %msg.e3_id,
+                        stage = ?msg.new_stage,
+                        "E3 reached terminal stage — cancelling pending committee finalization timer"
+                    );
+                    ctx.cancel_future(handle);
+                }
+            }
+            _ => {}
+        }
+    }
+}

crates/aggregator/src/keyshare_created_filter_buffer.rs

@@ -6,13 +6,14 @@
 
 use actix::prelude::*;
 
-use e3_events::{prelude::*, EnclaveEvent, EnclaveEventData};
+use e3_events::{prelude::*, Die, EnclaveEvent, EnclaveEventData};
 use e3_utils::MAILBOX_LIMIT;
 use std::collections::HashSet;
+use tracing::info;
 
 use crate::PublicKeyAggregator;
 
-/// Buffer KeyshareCreated events until CommitteeFinalized has been published
+/// Buffers `KeyshareCreated` events until `CommitteeFinalized` arrives.
 pub struct KeyshareCreatedFilterBuffer {
     dest: Addr<PublicKeyAggregator>,
     committee: Option<HashSet<String>>,
@@ -65,14 +66,38 @@ impl Handler<EnclaveEvent> for KeyshareCreatedFilterBuffer {
                 _ => {}
             },
             EnclaveEventData::CommitteeFinalized(data) => {
-                self.dest.do_send(msg.clone()); // forward committee first
+                self.dest.do_send(msg.clone());
                 self.committee = Some(data.committee.iter().cloned().collect());
                 self.process_buffered_events();
             }
+            EnclaveEventData::CommitteeMemberExpelled(data) => {
+                // Only process raw events from chain (party_id not yet resolved).
+                if data.party_id.is_some() {
+                    return;
+                }
+
+                // Remove expelled node so we don't forward late KeyshareCreated events from them
+                if let Some(ref mut committee) = self.committee {
+                    let node_addr = data.node.to_string();
+                    info!(
+                        "KeyshareCreatedFilterBuffer: removing expelled node {} from committee filter (e3_id={})",
+                        node_addr, data.e3_id
+                    );
+                    committee.remove(&node_addr);
+                }
+                // Forward to PublicKeyAggregator for threshold_n adjustment
+                self.dest.do_send(msg);
+            }
             _ => {
-                // forward all other events
                 self.dest.do_send(msg);
             }
         }
     }
 }
+
+impl Handler<Die> for KeyshareCreatedFilterBuffer {
+    type Result = ();
+    fn handle(&mut self, _: Die, ctx: &mut Self::Context) -> Self::Result {
+        ctx.stop();
+    }
+}

crates/aggregator/src/publickey_aggregator.rs

@@ -137,6 +137,50 @@ impl PublicKeyAggregator {
             })
         })
     }
+
+    pub fn handle_member_expelled(
+        &mut self,
+        node: &str,
+        ec: &EventContext<Sequenced>,
+    ) -> Result<()> {
+        self.state.try_mutate(ec, |mut state| {
+            let PublicKeyAggregatorState::Collecting {
+                threshold_n,
+                keyshares,
+                nodes,
+                ..
+            } = &mut state
+            else {
+                return Ok(state);
+            };
+
+            // Remove the expelled node from the nodes set so it won't appear in
+            // PublicKeyAggregated.nodes (forwarded on-chain for reward distribution).
+            // Note: the corresponding keyshare cannot be removed because the
+            // keyshares OrderedSet is keyed by raw bytes with no node mapping.
+            // This is acceptable because BFV public key aggregation is additive
+            // and works correctly with any superset of valid keys.
+            nodes.remove(&node.to_string());
+
+            if *threshold_n > 0 {
+                *threshold_n -= 1;
+                info!(
+                    "PublicKeyAggregator: reduced threshold_n to {} after expelling {}",
+                    threshold_n, node
+                );
+            }
+
+            if keyshares.len() == *threshold_n && *threshold_n > 0 {
+                info!("PublicKeyAggregator: enough keyshares after expulsion, computing aggregate");
+                return Ok(PublicKeyAggregatorState::Computing {
+                    keyshares: std::mem::take(keyshares),
+                    nodes: std::mem::take(nodes),
+                });
+            }
+
+            Ok(state)
+        })
+    }
 }
 
 impl Actor for PublicKeyAggregator {
@@ -155,6 +199,50 @@ impl Handler<EnclaveEvent> for PublicKeyAggregator {
                 self.notify_sync(ctx, TypedEvent::new(data, ec))
             }
             EnclaveEventData::E3RequestComplete(_) => self.notify_sync(ctx, Die),
+            EnclaveEventData::CommitteeMemberExpelled(data) => {
+                // Only process raw events from chain (party_id not yet resolved).
+                if data.party_id.is_some() {
+                    return;
+                }
+
+                let node_addr = data.node.to_string();
+
+                if data.e3_id != self.e3_id {
+                    error!("Wrong e3_id sent to PublicKeyAggregator for expulsion. This should not happen.");
+                    return;
+                }
+
+                info!(
+                    "PublicKeyAggregator: committee member expelled: {} for e3_id={}",
+                    node_addr, data.e3_id
+                );
+                trap(EType::PublickeyAggregation, &self.bus.with_ec(&ec), || {
+                    let was_collecting = matches!(
+                        self.state.get(),
+                        Some(PublicKeyAggregatorState::Collecting { .. })
+                    );
+
+                    self.handle_member_expelled(&node_addr, &ec)?;
+
+                    if was_collecting {
+                        if let Some(PublicKeyAggregatorState::Computing { keyshares, .. }) =
+                            &self.state.get()
+                        {
+                            self.notify_sync(
+                                ctx,
+                                TypedEvent::new(
+                                    ComputeAggregate {
+                                        keyshares: keyshares.clone(),
+                                        e3_id: data.e3_id,
+                                    },
+                                    ec.clone(),
+                                ),
+                            );
+                        }
+                    }
+                    Ok(())
+                });
+            }
             _ => (),
         };
     }
@@ -217,7 +305,6 @@ impl Handler<TypedEvent<ComputeAggregate>> for PublicKeyAggregator {
                 self.fhe.params.moduli().to_vec(),
             )?;
 
-            // Update the local state
             self.set_pubkey(pubkey, &ec)?;
 
             if let Some(PublicKeyAggregatorState::Complete {

crates/ciphernode-builder/src/ciphernode_builder.rs

@@ -18,7 +18,9 @@ use e3_events::{
     AggregateConfig, AggregateId, BusHandle, EnclaveEvent, EventBus, EventBusConfig, EvmEventConfig,
 };
 use e3_evm::{BondingRegistrySolReader, CiphernodeRegistrySolReader, EnclaveSolWriter};
-use e3_evm::{CiphernodeRegistrySol, EnclaveSolReader};
+use e3_evm::{
+    CiphernodeRegistrySol, EnclaveSolReader, SlashingManagerSolReader, SlashingManagerSolWriter,
+};
 use e3_fhe::ext::FheExtension;
 use e3_fhe_params::BfvPreset;
 use e3_keyshare::ext::ThresholdKeyshareExtension;
@@ -95,6 +97,7 @@ pub struct ContractComponents {
     enclave: bool,
     ciphernode_registry: bool,
     bonding_registry: bool,
+    slashing_manager: bool,
 }
 
 #[derive(Clone, Debug)]
@@ -296,6 +299,13 @@ impl CiphernodeBuilder {
         self
     }
 
+    /// Setup a SlashingManager writer for submitting slash proposals on-chain.
+    /// Requires the `slashing_manager` contract address to be configured.
+    pub fn with_contract_slashing_manager(mut self) -> Self {
+        self.contract_components.slashing_manager = true;
+        self
+    }
+
     /// Setup net package components.
     pub fn with_net(mut self, peers: Vec<String>, quic_port: u16) -> Self {
         self.net_config = Some(NetConfig::new(peers, quic_port));
@@ -668,6 +678,41 @@ async fn setup_evm_system(
                     )
                 }
         }
+
+        if contract_components.slashing_manager {
+            if let Some(contract) = &chain.contracts.slashing_manager {
+                // Reader: read SlashExecuted events from chain
+                let contract_addr = contract.address()?;
+                system.with_contract(contract_addr, move |next| {
+                    SlashingManagerSolReader::setup(&next).recipient()
+                });
+
+                // Writer: submit proposeSlash transactions
+                match provider_cache.ensure_write_provider(&chain).await {
+                    Ok(write_provider) => {
+                        match SlashingManagerSolWriter::attach(
+                            &bus,
+                            write_provider.clone(),
+                            contract_addr,
+                        )
+                        .await
+                        {
+                            Ok(_) => {
+                                info!("SlashingManagerSolWriter attached for fault submission");
+                            }
+                            Err(e) => {
+                                error!("Failed to attach SlashingManagerSolWriter, skipping: {}", e)
+                            }
+                        }
+                    }
+                    Err(e) => error!(
+                        "Failed to create write provider for SlashingManager, skipping: {}",
+                        e
+                    ),
+                }
+            }
+        }
+
         system.build();
     }
 

crates/config/src/contract.rs

@@ -48,6 +48,7 @@ pub struct ContractAddresses {
     pub bonding_registry: Contract,
     pub e3_program: Option<Contract>,
     pub fee_token: Option<Contract>,
+    pub slashing_manager: Option<Contract>,
 }
 
 impl ContractAddresses {
@@ -58,6 +59,7 @@ impl ContractAddresses {
             Some(&self.bonding_registry),
             self.e3_program.as_ref(),
             self.fee_token.as_ref(),
+            self.slashing_manager.as_ref(),
         ]
         .into_iter()
         .flatten()

crates/entrypoint/src/config/setup.rs

@@ -59,6 +59,9 @@ chains:
       bonding_registry:
         address: "{}"
         deploy_block: {}
+      slashing_manager:
+        address: "{}"
+        deploy_block: {}
 "#,
         rpc_url,
         get_contract_info("Enclave")?.address,
@@ -67,6 +70,8 @@ chains:
         get_contract_info("CiphernodeRegistryOwnable")?.deploy_block,
         get_contract_info("BondingRegistry")?.address,
         get_contract_info("BondingRegistry")?.deploy_block,
+        get_contract_info("SlashingManager")?.address,
+        get_contract_info("SlashingManager")?.deploy_block,
     );
 
     fs::write(config_path.clone(), config_content)?;

crates/entrypoint/src/start/aggregator_start.rs

@@ -31,6 +31,7 @@ pub async fn execute(
         .with_contract_enclave_full()
         .with_contract_bonding_registry()
         .with_contract_ciphernode_registry()
+        .with_contract_slashing_manager()
         .with_max_threads()
         .with_pubkey_aggregation()
         .with_threshold_plaintext_aggregation()

crates/entrypoint/src/start/start.rs

@@ -35,6 +35,7 @@ pub async fn execute(config: &AppConfig) -> Result<CiphernodeHandle> {
         .with_contract_bonding_registry()
         .with_max_threads()
         .with_contract_ciphernode_registry()
+        .with_contract_slashing_manager()
         .with_trbfv()
         .with_zkproof(backend)
         .with_net(config.peers(), config.quic_port())