feat: add validator id and verification to echo and ready messages

crates/topos-crypto/Cargo.toml

@@ -12,6 +12,13 @@ secp256k1.workspace = true
 byteorder.workspace = true
 hex.workspace = true
 thiserror.workspace = true
+ethers.workspace = true
 
 keccak-hash = "0.10.0"
 eth-keystore = "0.5.0"
+
+[dev-dependencies]
+rstest.workspace = true
+topos-tce-transport = { path = "../topos-tce-transport" }
+topos-core = { path = "../topos-core" }
+ethers.workspace = true

crates/topos-crypto/src/lib.rs

@@ -3,6 +3,7 @@ use thiserror::Error;
 pub mod hash;
 pub mod keys;
 pub mod keystore;
+pub mod messages;
 pub mod signatures;
 
 #[derive(Debug, Error)]

crates/topos-crypto/src/messages.rs

@@ -0,0 +1,50 @@
+use ethers::signers::Signer;
+use ethers::signers::{LocalWallet, WalletError};
+use ethers::types::{RecoveryMessage, SignatureError};
+use ethers::utils::hash_message;
+use std::sync::Arc;
+use thiserror::Error;
+
+pub use ethers::types::{Address, Signature, H160};
+
+#[derive(Error, Debug)]
+pub enum MessageSignerError {
+    #[error("Unable to parse private key")]
+    PrivateKeyParsing,
+}
+
+#[derive(Debug)]
+pub struct MessageSigner {
+    pub public_address: Address,
+    wallet: LocalWallet,
+}
+
+impl MessageSigner {
+    pub fn new(private_key: &str) -> Result<Arc<Self>, MessageSignerError> {
+        let wallet: LocalWallet = private_key
+            .parse()
+            .map_err(|_| MessageSignerError::PrivateKeyParsing)?;
+
+        Ok(Arc::new(Self {
+            public_address: wallet.address(),
+            wallet,
+        }))
+    }
+
+    pub fn sign_message(&self, payload: &[u8]) -> Result<Signature, WalletError> {
+        let hash = hash_message(payload);
+
+        LocalWallet::sign_hash(&self.wallet, hash)
+    }
+
+    pub fn verify_signature(
+        &self,
+        signature: Signature,
+        payload: &[u8],
+        public_key: Address,
+    ) -> Result<(), SignatureError> {
+        let message: RecoveryMessage = payload.into();
+
+        signature.verify(message, public_key)
+    }
+}

crates/topos-crypto/tests/messages.rs

@@ -0,0 +1,63 @@
+use rstest::*;
+use topos_core::uci::CertificateId;
+use topos_crypto::messages::MessageSigner;
+use topos_tce_transport::ValidatorId;
+
+#[rstest]
+pub fn test_signing_messages() {
+    let message_signer_sender =
+        MessageSigner::new("122f3ae6ade1fd136b292cea4f6243c7811160352c8821528547a1fe7c459daf")
+            .unwrap();
+    let validator_id_sender = ValidatorId::from(message_signer_sender.public_address);
+    let certificate_id = CertificateId::from_array([0u8; 32]);
+
+    let mut payload = Vec::new();
+    payload.extend_from_slice(certificate_id.as_array());
+    payload.extend_from_slice(validator_id_sender.as_bytes());
+
+    let signature = message_signer_sender
+        .sign_message(&payload)
+        .expect("Cannot create Signature");
+
+    let message_signer_receiver =
+        MessageSigner::new("a2e33a9bad88f7b7568228f51d5274c471a9217162d46f1533b6a290f0be1baf")
+            .unwrap();
+
+    let verify = message_signer_receiver.verify_signature(
+        signature,
+        &payload,
+        validator_id_sender.address(),
+    );
+
+    assert!(verify.is_ok());
+}
+
+#[rstest]
+pub fn fails_to_verify_with_own_public_address() {
+    let message_signer_sender =
+        MessageSigner::new("122f3ae6ade1fd136b292cea4f6243c7811160352c8821528547a1fe7c459daf")
+            .unwrap();
+    let validator_id_sender = ValidatorId::from(message_signer_sender.public_address);
+    let certificate_id = CertificateId::from_array([0u8; 32]);
+
+    let mut payload = Vec::new();
+    payload.extend_from_slice(certificate_id.as_array());
+    payload.extend_from_slice(validator_id_sender.as_bytes());
+
+    let signature = message_signer_sender
+        .sign_message(&payload)
+        .expect("Cannot create Signature");
+
+    let message_signer_receiver =
+        MessageSigner::new("a2e33a9bad88f7b7568228f51d5274c471a9217162d46f1533b6a290f0be1baf")
+            .unwrap();
+    let validator_id_receiver = ValidatorId::from(message_signer_receiver.public_address);
+
+    let verify = message_signer_receiver.verify_signature(
+        signature,
+        &payload,
+        validator_id_receiver.address(),
+    );
+
+    assert!(verify.is_err());
+}

crates/topos-sequencer-subnet-client/src/lib.rs

@@ -3,7 +3,7 @@ pub mod subnet_contract;
 use crate::subnet_contract::{create_topos_core_contract_from_json, get_block_events};
 use ethers::abi::ethabi::ethereum_types::{H160, U256};
 use ethers::core::k256::ecdsa::SigningKey;
-use ethers::prelude::Wallet;
+use ethers::signers::Wallet;
 use ethers::types::TransactionReceipt;
 use ethers::{
     abi::Token,

crates/topos-sequencer-subnet-client/src/subnet_contract.rs

@@ -1,7 +1,7 @@
 use crate::{Error, SubnetEvent};
 use ethers::abi::ethabi::ethereum_types::{H160, U64};
 use ethers::contract::ContractError;
-use ethers::prelude::LocalWallet;
+use ethers::signers::LocalWallet;
 use ethers::{
     prelude::abigen,
     providers::{Middleware, Provider, Ws},

crates/topos-sequencer-subnet-runtime/tests/subnet_contract.rs

@@ -3,8 +3,6 @@
 use dockertest::{
     Composition, DockerTest, Image, LogAction, LogOptions, LogPolicy, LogSource, PullPolicy, Source,
 };
-use ethers::prelude::Block;
-use ethers::types::H256;
 use ethers::{
     abi::{ethabi::ethereum_types::U256, Address},
     contract::abigen,
@@ -14,6 +12,7 @@ use ethers::{
     prelude::Wallet,
     providers::{Http, Middleware, Provider},
     signers::{LocalWallet, Signer},
+    types::{Block, H256},
 };
 use rstest::*;
 use serial_test::serial;

crates/topos-tce-api/tests/runtime.rs

@@ -29,7 +29,7 @@ use topos_test_sdk::storage::{create_fullnode_store, storage_client};
 use topos_test_sdk::tce::public_api::{broadcast_stream, create_public_api, PublicApiContext};
 
 #[rstest]
-#[timeout(Duration::from_secs(1))]
+#[timeout(Duration::from_secs(4))]
 #[test(tokio::test)]
 async fn runtime_can_dispatch_a_cert(
     #[future] create_public_api: (PublicApiContext, impl Stream<Item = RuntimeEvent>),
@@ -104,7 +104,7 @@ async fn runtime_can_dispatch_a_cert(
 }
 
 #[rstest]
-#[timeout(Duration::from_secs(2))]
+#[timeout(Duration::from_secs(4))]
 #[test(tokio::test)]
 async fn can_catchup_with_old_certs(
     #[with(SOURCE_SUBNET_ID_1, &[TARGET_SUBNET_ID_1], 15)]
@@ -191,7 +191,7 @@ async fn can_catchup_with_old_certs(
 }
 
 #[rstest]
-#[timeout(Duration::from_secs(2))]
+#[timeout(Duration::from_secs(4))]
 #[test(tokio::test)]
 async fn can_catchup_with_old_certs_with_position(
     broadcast_stream: broadcast::Receiver<CertificateDeliveredWithPositions>,
@@ -323,7 +323,7 @@ async fn can_catchup_with_old_certs_with_position(
 async fn can_listen_for_multiple_subnet_id() {}
 
 #[rstest]
-#[timeout(Duration::from_secs(2))]
+#[timeout(Duration::from_secs(4))]
 #[test(tokio::test)]
 async fn boots_healthy_graphql_server(
     broadcast_stream: broadcast::Receiver<CertificateDeliveredWithPositions>,
@@ -369,7 +369,7 @@ async fn boots_healthy_graphql_server(
 }
 
 #[rstest]
-#[timeout(Duration::from_secs(2))]
+#[timeout(Duration::from_secs(4))]
 #[test(tokio::test)]
 async fn graphql_server_enables_cors(
     broadcast_stream: broadcast::Receiver<CertificateDeliveredWithPositions>,
@@ -438,7 +438,7 @@ async fn graphql_server_enables_cors(
 }
 
 #[rstest]
-#[timeout(Duration::from_secs(2))]
+#[timeout(Duration::from_secs(4))]
 #[test(tokio::test)]
 async fn can_query_graphql_endpoint_for_certificates(
     broadcast_stream: broadcast::Receiver<CertificateDeliveredWithPositions>,

crates/topos-tce-broadcast/Cargo.toml

@@ -18,20 +18,20 @@ tokio = { workspace = true, features = ["full"] }
 tokio-stream = { workspace = true, features = ["sync"] }
 tracing-subscriber = { workspace = true, features = ["env-filter", "fmt"] }
 tracing.workspace = true
-
 tce_transport = { package = "topos-tce-transport", path = "../topos-tce-transport"}
 topos-core = { workspace = true, features = ["uci"] }
 topos-p2p = { path = "../topos-p2p/"}
 topos-metrics = { path = "../topos-metrics/" }
 topos-tce-storage = { path = "../topos-tce-storage/" }
+topos-crypto = { path = "../topos-crypto" }
 
 [dev-dependencies]
 criterion = { version = "0.5.1", features = ["async_futures", "async_tokio"] }
 test-log.workspace = true
 env_logger.workspace = true
 rstest.workspace = true
 rand.workspace = true
-
+hex.workspace = true
 topos-test-sdk = { path = "../topos-test-sdk/" }
 
 [features]

crates/topos-tce-broadcast/benches/task_manager.rs

@@ -1,14 +1,16 @@
 use std::collections::HashSet;
 use std::sync::Arc;
-use tce_transport::ReliableBroadcastParams;
+use tce_transport::{ReliableBroadcastParams, ValidatorId};
 use tokio::sync::{broadcast, mpsc, oneshot};
+use topos_crypto::messages::MessageSigner;
 use topos_tce_broadcast::double_echo::DoubleEcho;
 use topos_tce_broadcast::sampler::SubscriptionsView;
 use topos_tce_storage::validator::ValidatorStore;
 use topos_test_sdk::certificates::create_certificate_chain;
 use topos_test_sdk::constants::{SOURCE_SUBNET_ID_1, TARGET_SUBNET_ID_1};
 
 const CHANNEL_SIZE: usize = 256_000;
+const PRIVATE_KEY: &str = "47d361f6becb933a77d7e01dee7b1c1859b656adbd8428bf7bf9519503e5d5d6";
 
 struct TceParams {
     nb_peers: usize,
@@ -34,13 +36,21 @@ pub async fn processing_double_echo(n: u64, validator_store: Arc<ValidatorStore>
         },
     };
 
+    let message_signer: Arc<MessageSigner> = MessageSigner::new(PRIVATE_KEY).unwrap();
+
+    let mut validators = HashSet::new();
+    let validator_id = ValidatorId::from(message_signer.public_address);
+    validators.insert(validator_id);
+
     let mut double_echo = DoubleEcho::new(
         params.broadcast_params,
+        validator_id,
+        message_signer.clone(),
+        validators,
         task_manager_message_sender.clone(),
         cmd_receiver,
         event_sender,
         double_echo_shutdown_receiver,
-        0,
         validator_store,
         broadcast_sender,
     );
@@ -93,12 +103,24 @@ pub async fn processing_double_echo(n: u64, validator_store: Arc<ValidatorStore>
     }
 
     for cert in &certificates {
+        let mut payload = Vec::new();
+        payload.extend_from_slice(cert.certificate.id.as_array());
+        payload.extend_from_slice(validator_id.as_bytes());
+
         for p in &double_echo_selected_echo {
-            double_echo.handle_echo(*p, cert.certificate.id).await;
+            let signature = message_signer.sign_message(&payload).unwrap();
+
+            double_echo
+                .handle_echo(*p, cert.certificate.id, validator_id, signature)
+                .await;
         }
 
         for p in &double_echo_selected_ready {
-            double_echo.handle_ready(*p, cert.certificate.id).await;
+            let signature = message_signer.sign_message(&payload).unwrap();
+
+            double_echo
+                .handle_ready(*p, cert.certificate.id, validator_id, signature)
+                .await;
         }
     }